diff --git a/scripts/career_vedastro_radar.py b/scripts/career_vedastro_radar.py new file mode 100644 index 00000000..bdc4d770 --- /dev/null +++ b/scripts/career_vedastro_radar.py @@ -0,0 +1,78 @@ +#!/usr/bin/env python3 +"""VedAstro-assisted career timing radar. + +External VedAstro signals are secondary evidence only. They do not change +local scores, dominant labels, or final career/prashna adjudication by +themselves. +""" + +from __future__ import annotations + +import argparse +import json +from pathlib import Path +from typing import Any + +import vedastro_service_adapter + + +def build_career_radar_packet(case: dict[str, Any], *, start_date: str, end_date: str, case_id: str = "user_chart") -> dict[str, Any]: + result = vedastro_service_adapter.run_range_scan_for_case( + case, + "career", + start_date, + end_date, + case_id=case_id, + ) + policy = result.get("adjudicator_policy") if isinstance(result.get("adjudicator_policy"), dict) else {} + can_change_score = bool(policy.get("can_change_score", False)) + status = result.get("status", "blocked") + return { + "scope": "career_vedastro_radar", + "status": "ok" if status == "ok" else "blocked", + "blocked_reason": None if status == "ok" else result.get("reason") or status, + "domain": "career", + "adjudicator_use": "secondary_evidence_only", + "can_change_score": can_change_score, + "can_set_final_verdict": False, + "start_date": start_date, + "end_date": end_date, + "vedastro_range_scan_result": result, + "technique_audit_row": { + "technique": "VedAstro Career Range Scan", + "used": status == "ok", + "status": status, + "role": "external_secondary_evidence", + "confidence_effect": "raises_attention_only_not_final_score" if status == "ok" else "blocked_no_effect", + }, + } + + +def _load_case(path: Path) -> dict[str, Any]: + return json.loads(path.read_text(encoding="utf-8")) + + +def main() -> int: + parser = argparse.ArgumentParser(description=__doc__) + parser.add_argument("--case-json", type=Path, required=True) + parser.add_argument("--start-date", required=True) + parser.add_argument("--end-date", required=True) + parser.add_argument("--case-id", default="user_chart") + parser.add_argument("--output", type=Path) + args = parser.parse_args() + packet = build_career_radar_packet( + _load_case(args.case_json), + start_date=args.start_date, + end_date=args.end_date, + case_id=args.case_id, + ) + text = json.dumps(packet, ensure_ascii=False, indent=2, sort_keys=True) + if args.output: + args.output.parent.mkdir(parents=True, exist_ok=True) + args.output.write_text(text + "\n", encoding="utf-8") + print(text) + return 0 if packet["status"] == "ok" else 1 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/tests/test_career_vedastro_radar.py b/tests/test_career_vedastro_radar.py new file mode 100644 index 00000000..61a9ae2d --- /dev/null +++ b/tests/test_career_vedastro_radar.py @@ -0,0 +1,54 @@ +from __future__ import annotations + +import sys +from pathlib import Path + + +ROOT = Path(__file__).resolve().parents[1] +SCRIPTS = ROOT / "scripts" +if str(SCRIPTS) not in sys.path: + sys.path.insert(0, str(SCRIPTS)) + +import career_vedastro_radar # noqa: E402 + + +def test_career_radar_wraps_vedastro_range_scan_as_secondary_evidence(monkeypatch) -> None: + def fake_run(case, domain, start_date, end_date, case_id="user_chart"): + return { + "status": "ok", + "domain": domain, + "operation": "range_scan", + "evidence_ledger": [{"event_id": "CareerExpansionWindow", "date": "2026-09-01"}], + "adjudicator_policy": {"can_change_score": False}, + } + + monkeypatch.setattr(career_vedastro_radar.vedastro_service_adapter, "run_range_scan_for_case", fake_run) + + packet = career_vedastro_radar.build_career_radar_packet( + {"year": 1990, "month": 1, "day": 1, "hour": 12, "minute": 0, "lat": 36.4, "lon": 114.2, "tz": 8}, + start_date="2026-07-16", + end_date="2026-12-31", + ) + + assert packet["status"] == "ok" + assert packet["domain"] == "career" + assert packet["adjudicator_use"] == "secondary_evidence_only" + assert packet["can_change_score"] is False + assert packet["vedastro_range_scan_result"]["evidence_ledger"][0]["event_id"] == "CareerExpansionWindow" + + +def test_career_radar_preserves_blocked_boundary(monkeypatch) -> None: + def fake_run(case, domain, start_date, end_date, case_id="user_chart"): + return {"status": "blocked", "reason": "official_endpoint_not_configured", "evidence_ledger": []} + + monkeypatch.setattr(career_vedastro_radar.vedastro_service_adapter, "run_range_scan_for_case", fake_run) + + packet = career_vedastro_radar.build_career_radar_packet( + {"year": 1990, "month": 1, "day": 1, "hour": 12, "minute": 0, "lat": 36.4, "lon": 114.2, "tz": 8}, + start_date="2026-07-16", + end_date="2026-12-31", + ) + + assert packet["status"] == "blocked" + assert packet["can_change_score"] is False + assert packet["blocked_reason"] == "official_endpoint_not_configured"