Jesse_Chen
7faf8555a5
The extractor only read an obsolete varga object shape, never issued Transit receipts, and the consultation response omitted Chara Karakas, so career/marriage/timing/wealth all blocked. Supply the live engine shape, fail-closed receipts, and jaimini.chara_karakas without relaxing the evidence plan. Co-authored-by: Cursor <cursoragent@cursor.com>
258 lines
9.8 KiB
Python
258 lines
9.8 KiB
Python
#!/usr/bin/env python3
|
|
"""Capture trimmed consultation_workflow responses for report blocked-repairs fixtures.
|
|
|
|
Smoke birth only: 1993-06-15 10:30, lat 36.42 / lon 114.21 / tz 8.
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
import sys
|
|
from pathlib import Path
|
|
|
|
ROOT = Path(__file__).resolve().parents[1]
|
|
sys.path.insert(0, str(ROOT / "scripts"))
|
|
|
|
from consultation_plan_contract import PLAN_VERSION, _ROUTE_CONTRACTS # noqa: E402
|
|
from jyotish_api_server import JyotishAPIHandler # noqa: E402
|
|
|
|
BIRTH = {
|
|
"year": 1993,
|
|
"month": 6,
|
|
"day": 15,
|
|
"hour": 10,
|
|
"minute": 30,
|
|
"lat": 36.42,
|
|
"lon": 114.21,
|
|
"tz": 8,
|
|
"city": "Handan",
|
|
}
|
|
THEMES = ("career", "marriage", "timing", "wealth")
|
|
VARGA_KEEP = (
|
|
"D2_Hora", "D2",
|
|
"D9_Navamsa", "D9",
|
|
"D10_Dasamsa", "D10",
|
|
"D11_Rudramsa", "D11",
|
|
"D24_Chaturvimsamsa", "D24_Siddhamsa", "D24",
|
|
)
|
|
MODULE_KEEP = (
|
|
"varga_full",
|
|
"transits",
|
|
"jaimini",
|
|
"narayana_dasha",
|
|
"arudha_padas",
|
|
"yogas",
|
|
"ashtakavarga",
|
|
"dasha_sub_periods",
|
|
)
|
|
CONSUMER_KEEP = (
|
|
"route",
|
|
"core_status",
|
|
"available_layers",
|
|
"missing_route_layers",
|
|
"hard_blockers",
|
|
"answer_policy",
|
|
)
|
|
CHART_KEEP = ("success", "ascendant", "planets", "houses", "dasha")
|
|
|
|
|
|
def _handler() -> JyotishAPIHandler:
|
|
return JyotishAPIHandler.__new__(JyotishAPIHandler)
|
|
|
|
|
|
def _body(theme: str) -> dict:
|
|
contract = _ROUTE_CONTRACTS[theme]
|
|
return {
|
|
**BIRTH,
|
|
"question": f"请为个人报告计算 {theme} 主题证据",
|
|
"question_text": f"请为个人报告计算 {theme} 主题证据",
|
|
"theme": list(contract.themes),
|
|
"entry_mode": "direct_chart",
|
|
"plan_version": PLAN_VERSION,
|
|
"strict_workflow_route": contract.resolved_routes[0],
|
|
"required_layers": list(contract.required_layers),
|
|
"claim_boundary": contract.claim_boundary,
|
|
"plan_depth": "standard",
|
|
"requested_domains": list(contract.requested_domains),
|
|
"timing_horizon": "next_12_months" if theme == "timing" else None,
|
|
"precision_boundary": "server_evidence_required",
|
|
"required_evidence_categories": list(contract.required_evidence_categories),
|
|
"defer_optional_external_evidence": True,
|
|
}
|
|
|
|
|
|
def _trim_varga(varga_full: dict) -> dict:
|
|
trimmed = {}
|
|
for key in VARGA_KEEP:
|
|
value = varga_full.get(key)
|
|
if isinstance(value, dict):
|
|
trimmed[key] = value
|
|
if trimmed:
|
|
return trimmed
|
|
# Fall back to prefix match if the live keys drifted.
|
|
for key, value in varga_full.items():
|
|
if not isinstance(key, str) or not isinstance(value, dict):
|
|
continue
|
|
if any(key == keep or key.startswith(f"{keep.split('_')[0]}_") for keep in ("D2", "D9", "D10", "D11", "D24")):
|
|
if key.startswith(("D2", "D9", "D10", "D11", "D24")):
|
|
trimmed[key] = value
|
|
return trimmed
|
|
|
|
|
|
def _trim_workflow(raw: dict) -> dict:
|
|
chart = raw.get("chart") if isinstance(raw.get("chart"), dict) else {}
|
|
modules = chart.get("modules") if isinstance(chart.get("modules"), dict) else {}
|
|
kept_modules = {}
|
|
for name in MODULE_KEEP:
|
|
value = modules.get(name)
|
|
if name == "varga_full" and isinstance(value, dict):
|
|
kept_modules[name] = _trim_varga(value)
|
|
elif value is not None:
|
|
kept_modules[name] = value
|
|
kept_chart = {key: chart[key] for key in CHART_KEEP if key in chart}
|
|
if chart.get("success") is not None:
|
|
kept_chart["success"] = chart.get("success")
|
|
kept_chart["modules"] = kept_modules
|
|
consumer = raw.get("consumer_context") if isinstance(raw.get("consumer_context"), dict) else {}
|
|
machine = raw.get("machine_evidence_packet") if isinstance(raw.get("machine_evidence_packet"), dict) else {}
|
|
workflow = {
|
|
"success": bool(raw.get("success")),
|
|
"chart": kept_chart,
|
|
"consumer_context": {key: consumer[key] for key in CONSUMER_KEEP if key in consumer},
|
|
"machine_evidence_packet": {
|
|
"sections": machine.get("sections") if isinstance(machine.get("sections"), dict) else {},
|
|
},
|
|
}
|
|
return _compact_workflow(workflow)
|
|
|
|
|
|
def _pick(row: dict, keys: tuple[str, ...]) -> dict:
|
|
return {key: row[key] for key in keys if key in row}
|
|
|
|
|
|
def _compact_workflow(workflow: dict) -> dict:
|
|
"""Drop bulky unused fields after the extraction-layer keep list."""
|
|
chart = workflow["chart"]
|
|
planets = chart.get("planets")
|
|
if isinstance(planets, dict):
|
|
chart["planets"] = {
|
|
name: _pick(row, ("sign", "sign_idx", "sign_index", "lon", "degree", "degree_in_sign", "house", "retrograde"))
|
|
if isinstance(row, dict) else row
|
|
for name, row in planets.items()
|
|
}
|
|
houses = chart.get("houses")
|
|
if isinstance(houses, dict):
|
|
chart["houses"] = {
|
|
key: _pick(row, ("cusp_sign", "sign", "sign_idx", "lord")) if isinstance(row, dict) else row
|
|
for key, row in houses.items()
|
|
}
|
|
dasha = chart.get("dasha")
|
|
if isinstance(dasha, dict):
|
|
periods = [
|
|
_pick(item, ("lord", "start", "end"))
|
|
for item in (dasha.get("periods") or [])
|
|
if isinstance(item, dict)
|
|
]
|
|
compact_dasha = {}
|
|
if periods:
|
|
compact_dasha["periods"] = periods
|
|
compact_dasha["mahadashas"] = periods
|
|
for key in ("current_md", "start_date"):
|
|
if key in dasha:
|
|
compact_dasha[key] = dasha[key]
|
|
chart["dasha"] = compact_dasha
|
|
modules = chart["modules"]
|
|
transits = modules.get("transits")
|
|
if isinstance(transits, dict):
|
|
modules["transits"] = _pick(transits, ("status", "sade_sati", "search_period", "trigger_count", "boundary"))
|
|
sav = (modules.get("ashtakavarga") or {}).get("sav") if isinstance(modules.get("ashtakavarga"), dict) else None
|
|
if isinstance(sav, dict):
|
|
modules["ashtakavarga"] = {"sav": _pick(sav, ("scores", "total"))}
|
|
yogas = modules.get("yogas")
|
|
if isinstance(yogas, dict):
|
|
rows = [
|
|
_pick(item, ("name", "category", "hit", "planets"))
|
|
for item in (yogas.get("yogas") or [])
|
|
if isinstance(item, dict)
|
|
]
|
|
modules["yogas"] = {"status": yogas.get("status"), "count": yogas.get("count"), "yogas": rows}
|
|
nara = modules.get("narayana_dasha")
|
|
if isinstance(nara, dict):
|
|
modules["narayana_dasha"] = _pick(nara, ("current_dasha", "lagna_sign"))
|
|
arudha = modules.get("arudha_padas")
|
|
if isinstance(arudha, dict):
|
|
padas = arudha.get("padas") if isinstance(arudha.get("padas"), dict) else {}
|
|
modules["arudha_padas"] = {"padas": {key: padas[key] for key in ("A7", "A10", "UL") if key in padas}}
|
|
jaimini = modules.get("jaimini")
|
|
if isinstance(jaimini, dict):
|
|
compact_jaimini = {}
|
|
if isinstance(jaimini.get("arudha_padas"), dict):
|
|
padas = jaimini["arudha_padas"].get("padas") if isinstance(jaimini["arudha_padas"].get("padas"), dict) else {}
|
|
compact_jaimini["arudha_padas"] = {"padas": {key: padas[key] for key in ("A7", "A10", "UL") if key in padas}}
|
|
karakas = jaimini.get("chara_karakas")
|
|
if isinstance(karakas, dict):
|
|
compact_jaimini["chara_karakas"] = {
|
|
key: _pick(row, ("planet", "degree_in_sign", "rank", "domain", "cn_name")) if isinstance(row, dict) else row
|
|
for key, row in karakas.items()
|
|
}
|
|
modules["jaimini"] = compact_jaimini
|
|
sub = modules.get("dasha_sub_periods")
|
|
if isinstance(sub, dict) and isinstance(sub.get("current"), dict):
|
|
modules["dasha_sub_periods"] = {"current": sub["current"]}
|
|
sections = workflow["machine_evidence_packet"].get("sections")
|
|
if isinstance(sections, dict):
|
|
keep = {
|
|
"D1", "D2", "D4", "D6", "D7", "D9", "D10", "D11", "D12", "D24", "D30",
|
|
"A7", "A10", "UL", "dasha_boundaries", "narayana_dasha", "yogas", "ashtakavarga",
|
|
"planet_degrees", "house_degrees", "dasha_sub_periods",
|
|
}
|
|
workflow["machine_evidence_packet"]["sections"] = {key: value for key, value in sections.items() if key in keep}
|
|
return workflow
|
|
|
|
|
|
def main() -> int:
|
|
handler = _handler()
|
|
out_dir = ROOT / "frontend/tests/fixtures"
|
|
out_dir.mkdir(parents=True, exist_ok=True)
|
|
payload = {
|
|
"source": "scripts/jyotish_api_server.py _compute_consultation_workflow",
|
|
"smoke_birth": {**BIRTH, "note": "fictional capture identity; not a real person"},
|
|
"trim": {
|
|
"keep_varga_keys": list(VARGA_KEEP),
|
|
"keep_modules": list(MODULE_KEEP),
|
|
"drop": [
|
|
"ai_prompt_pack",
|
|
"vedastro",
|
|
"varga_spectrum",
|
|
"guided_topics",
|
|
"western_spectrum",
|
|
"technique_audit_table",
|
|
],
|
|
},
|
|
"themes": {},
|
|
}
|
|
for theme in THEMES:
|
|
raw = handler._compute_consultation_workflow(_body(theme))
|
|
payload["themes"][theme] = _trim_workflow(raw)
|
|
varga = payload["themes"][theme]["chart"]["modules"].get("varga_full") or {}
|
|
transits = payload["themes"][theme]["chart"]["modules"].get("transits") or {}
|
|
jaimini = payload["themes"][theme]["chart"]["modules"].get("jaimini") or {}
|
|
print(
|
|
theme,
|
|
"success=",
|
|
payload["themes"][theme]["success"],
|
|
"varga_keys=",
|
|
sorted(varga),
|
|
"transits=",
|
|
transits.get("status"),
|
|
"karakas=",
|
|
sorted((jaimini.get("chara_karakas") or {})),
|
|
)
|
|
dest = out_dir / "consultation-workflow-report-blocked-repairs-golden.json"
|
|
dest.write_text(json.dumps(payload, ensure_ascii=False, indent=2) + "\n", encoding="utf-8")
|
|
print("wrote", dest, "bytes", dest.stat().st_size)
|
|
return 0
|
|
|
|
|
|
if __name__ == "__main__":
|
|
raise SystemExit(main())
|