Jyotisha/scripts/research/capture_consult_evidence_card_golden.py

"""Capture real engine consultation responses for the evidence-card tests.

Three public AA charts (Steve Jobs, Barack Obama, Elizabeth Taylor), the
research reference date, raman ayanamsa, mean nodes. Each chart carries the
family route (`workflow`) and, for evidence card v2, the annual
(`annual_workflow`) and timing (`timing_workflow`) routes with the frontend's
default twelve-month horizon. External VedAstro is not called (same stand-in
as the research runner). Each response is trimmed by key only: every kept
value is the engine's own value, unchanged.

    PYTHONHASHSEED=0 python scripts/research/capture_consult_evidence_card_golden.py
"""

from __future__ import annotations

import json
import os
import sys
from pathlib import Path
from typing import Any

ROOT = Path(__file__).resolve().parents[2]
sys.path[:0] = [str(ROOT), str(ROOT / "scripts"), str(ROOT / "scripts" / "research")]

from consult_evidence_card_lib import (  # noqa: E402
    AYANAMSA,
    NODE_MODE,
    QUESTIONS,
    REFERENCE_DATE,
    load_public_charts,
)
import consult_evidence_card_run as runner  # noqa: E402

OUT = ROOT / "frontend" / "tests" / "fixtures" / "consult-evidence-card-golden.json"

TOP_KEEP = (
    "success", "question", "routing", "consumer_context", "thematic_report",
    "birth_time_sensitivity", "reference_transparency", "candidate_range",
    "range_boundary_contexts",
)
RECTIFICATION_KEEP = ("effective_accuracy", "lagna_boundary", "summary", "enabled_vargas")
CHART_KEEP = (
    "success", "birth", "ascendant", "planets", "houses", "shadbala", "dasha", "yogas",
    # Evidence card v2: native technique layers (scripts/consultation_native_layers.py).
    "consultation_native_layers",
)
EXTRA_ROUTES = (
    ("annual_workflow", {"id": "annual", "domain": "annual", "question": "未来一年重点是什么？"}),
    ("timing_workflow", {"id": "timing", "domain": "timing", "question": "接下来一年哪些时间点重要？"}),
)
MODULE_KEEP = (
    "varga_spectrum", "shadbala", "arudha_padas", "jaimini", "narayana_dasha",
    "ashtakavarga", "dasha_sub_periods", "kp_cusps", "gulika",
    "functional_benefic_malefic", "kakshya", "yogas", "chara_dasha", "transits",
)
NARAYANA_KEEP = ("lagna_sign", "current_dasha", "current_year", "current_age", "mahadasha_sequence")


def _pick(value: dict[str, Any], keys: tuple[str, ...]) -> dict[str, Any]:
    return {key: value[key] for key in keys if key in value}


def trim(workflow: dict[str, Any]) -> dict[str, Any]:
    out = _pick(workflow, TOP_KEEP)
    out["rectification"] = _pick(workflow.get("rectification") or {}, RECTIFICATION_KEEP)
    chart = workflow.get("chart") or {}
    kept_chart = _pick(chart, CHART_KEEP)
    modules = _pick(chart.get("modules") or {}, MODULE_KEEP)
    if isinstance(modules.get("narayana_dasha"), dict):
        modules["narayana_dasha"] = _pick(modules["narayana_dasha"], NARAYANA_KEEP)
    kept_chart["modules"] = modules
    out["chart"] = kept_chart
    return out


def main() -> int:
    if os.environ.get("PYTHONHASHSEED") != "0":
        raise SystemExit("Set PYTHONHASHSEED=0 before starting this process (ERR-111).")
    runner._block_external_vedastro()
    question = next(item for item in QUESTIONS if item["id"] == "family")
    charts = []
    for chart in load_public_charts(ROOT):
        workflow = runner._run_workflow(runner._workflow_body(chart, question))
        entry = {
            "id": chart["id"],
            "label": chart["label"],
            "source": chart["source"],
            "case_id": chart["case_id"],
            "rodden_rating": chart["rodden_rating"],
            "workflow": trim(workflow),
        }
        for key, extra in EXTRA_ROUTES:
            entry[key] = trim(runner._run_workflow(runner._workflow_body(chart, extra)))
        charts.append(entry)
    payload = {
        "source": "scripts/research/capture_consult_evidence_card_golden.py",
        "note": "Real engine consultation_workflow responses for three public AA charts, trimmed by key only.",
        "reference_date": REFERENCE_DATE,
        "ayanamsa": AYANAMSA,
        "node_mode": NODE_MODE,
        "route": question["domain"],
        "question": question["question"],
        "extra_routes": {key: extra for key, extra in EXTRA_ROUTES},
        "external_vedastro": "not_called",
        "charts": charts,
    }
    OUT.write_text(json.dumps(payload, ensure_ascii=False, separators=(",", ":")) + "\n", encoding="utf-8")
    print(f"wrote {OUT} ({OUT.stat().st_size} bytes)")
    return 0


if __name__ == "__main__":
    raise SystemExit(main())