From e2595bec13e5bcd64a713e16465bdb27cdf9e4dc Mon Sep 17 00:00:00 2001 From: 732642856 <732642856@qq.com> Date: Mon, 29 Jun 2026 15:40:34 +0800 Subject: [PATCH] Wire VedAstro official snapshot priority --- SKILL.md | 1 + jyotish-app/main.js | 2 +- mcp_server.py | 53 +- scripts/jyotish_api_server.py | 53 +- scripts/jyotish_engine.py | 94 +++ scripts/vedastro_evidence_orchestrator.py | 16 +- scripts/vedastro_priority.py | 167 +++++ scripts/vedastro_service_adapter.py | 590 +++++++++++++++++- ...st_vedastro_external_technique_evidence.py | 39 ++ tests/test_vedastro_official_full_snapshot.py | 442 +++++++++++++ 10 files changed, 1445 insertions(+), 12 deletions(-) create mode 100644 scripts/vedastro_priority.py create mode 100644 tests/test_vedastro_official_full_snapshot.py diff --git a/SKILL.md b/SKILL.md index 93bd9031..625e8617 100644 --- a/SKILL.md +++ b/SKILL.md @@ -64,6 +64,7 @@ description: 印度占星（Jyotish）专业解盘与推运系统。核心能力 2. **阶段一**（仅B）：PDF/图片提取 + Quality Gate 3. **阶段二**：意图识别 → 路由目标宫位（无明确意图→Level 2综合解盘） 4. **阶段二点五**：若 `full-reading` 或网页/API 返回 `ai_prompt_pack`，必须优先读取 `prompt_zh`、`evidence_snapshot`、`retrieval_plan` 作为 AI/RAG 主上下文；若没有该字段，再退回传统 JSON 摘要。 +4.1 **VedAstro 官方优先级**：用户给出生信息后，网页、Skill、MCP 都必须默认走同一条数据优先级：`VedAstro official snapshot -> local supplemental modules -> local fallback only when official blocked`。用户不需要主动要求“调用 VedAstro”。若 `evidence_snapshot.vedastro_official_full_snapshot.status` 为 `ok/partial` 且官方 chart 可用，D1/分盘/官方返回的原始字段以 VedAstro 为主；本地引擎只做补充、交叉检查或官方 blocked 时 fallback。 4. **阶段三**：静态分析10步（宫位→承诺→Yoga→Argala→逆行→NK→Shadbala→AV→Ketu→分盘） 5. **阶段四**：动态推运7步（Dasha→五系统Convergence→Transit→Double Transit→Jaimini→KP→Varshaphala） 6. **阶段五**：应期输出（五层验证→时间窗口→Actionable Output+案例检索） diff --git a/jyotish-app/main.js b/jyotish-app/main.js index bd1f1555..7d990b48 100644 --- a/jyotish-app/main.js +++ b/jyotish-app/main.js @@ -2439,7 +2439,7 @@ function renderVedAstroRangeScanResult(state = getVedAstroScanState()) { VedAstro Range Scan：${escapeHtml(status)} · ${escapeHtml(String(eventCount))} events · ${escapeHtml(state.start_date || '-')}/${escapeHtml(state.end_date || '-')}
- ${escapeHtml(result.reason || state.result?.boundary || '外部雷达结果已挂到 chartData.modules.vedastro_range_scan_result；本地 Jyotish gates 仍为主判断。')} + ${escapeHtml(result.reason || state.result?.boundary || 'VedAstro official snapshot 为主证据；range scan 是外部事件雷达，本地 Jyotish gates 作为补充与官方 blocked 时的 fallback。')} ${metadata.artifact_path ? ` artifact: ${escapeHtml(metadata.artifact_path)}` : ''}
${events.length ? ` diff --git a/mcp_server.py b/mcp_server.py index 10b4e20a..10203dd5 100644 --- a/mcp_server.py +++ b/mcp_server.py @@ -39,6 +39,7 @@ from local_env import load_local_env from mcp.server.fastmcp import FastMCP from functional_benefics import derive_functional_benefic_malefic +from vedastro_priority import official_snapshot_evidence load_local_env(SCRIPT_DIR) @@ -514,6 +515,29 @@ def _external_activation_audit(external_activation: Any) -> List[Dict[str, Any]] return [] +def _official_snapshot_audit(official_snapshot: Any) -> List[Dict[str, Any]]: + if not isinstance(official_snapshot, dict): + return [] + if official_snapshot.get("level") == "primary": + return [ + { + "technique": "VedAstro Official Full Snapshot", + "status": "used", + "role": "primary_raw_evidence", + "effect": "chart_and_varga_values_take_priority_over_local_engine", + } + ] + return [ + { + "technique": "VedAstro Official Full Snapshot", + "status": "blocked", + "role": "primary_raw_evidence", + "effect": "local_engine_fallback_only_with_boundary", + "reason": official_snapshot.get("reason") or official_snapshot.get("status"), + } + ] + + def _derive_external_technique_evidence(modules: Dict[str, Any], domain: str) -> Dict[str, Any]: ledger = _safe_get(modules, "external_technique_evidence", "evidence_ledger") if not isinstance(ledger, list): @@ -1581,9 +1605,11 @@ def _collect_strict_evidence(route: str, result: Dict[str, Any]) -> Dict[str, An present["argala_support"] = _derive_argala_support(modules, 10) present["external_activation"] = _derive_external_activation_support(modules, "career") present["external_technique_evidence"] = _derive_external_technique_evidence(modules, "career") + present["vedastro_official_snapshot"] = official_snapshot_evidence(modules) + present["source_priority"] = modules.get("source_priority") if isinstance(modules.get("source_priority"), dict) else {} present["functional_benefic_malefic"] = _derive_functional_benefic_malefic(modules) missing = [key for key, value in present.items() if key not in { - "external_activation", "external_technique_evidence", "argala_support", "shadbala", "shadbala_component_audit", "kakshya_career_support", "functional_benefic_malefic" + "external_activation", "external_technique_evidence", "vedastro_official_snapshot", "source_priority", "argala_support", "shadbala", "shadbala_component_audit", "kakshya_career_support", "functional_benefic_malefic" } and value in (None, {}, [], "")] convergence = present["career_convergence"] or {} confidence_cap = "medium" @@ -1612,7 +1638,8 @@ def _collect_strict_evidence(route: str, result: Dict[str, Any]) -> Dict[str, An ), } audit = ( - _external_activation_audit(present.get("external_activation")) + _official_snapshot_audit(present.get("vedastro_official_snapshot")) + + _external_activation_audit(present.get("external_activation")) + _external_technique_audit(present.get("external_technique_evidence")) ) if audit: @@ -1647,11 +1674,13 @@ def _collect_strict_evidence(route: str, result: Dict[str, Any]) -> Dict[str, An present["argala_support"] = _derive_argala_support(modules, 7) present["external_activation"] = _derive_external_activation_support(modules, "marriage") present["external_technique_evidence"] = _derive_external_technique_evidence(modules, "marriage") + present["vedastro_official_snapshot"] = official_snapshot_evidence(modules) + present["source_priority"] = modules.get("source_priority") if isinstance(modules.get("source_priority"), dict) else {} present["dignity_guardrail"] = _derive_dignity_guardrail(route, present) present["functional_benefic_malefic"] = _derive_functional_benefic_malefic(modules) missing = [ key for key, value in present.items() - if key not in {"chart", "external_activation", "external_technique_evidence", "dignity_guardrail", "jaimini_marriage_support", "jaimini_timing_support", "synastry_relationship_support", "argala_support", "shadbala", "shadbala_component_audit", "functional_benefic_malefic"} + if key not in {"chart", "external_activation", "external_technique_evidence", "vedastro_official_snapshot", "source_priority", "dignity_guardrail", "jaimini_marriage_support", "jaimini_timing_support", "synastry_relationship_support", "argala_support", "shadbala", "shadbala_component_audit", "functional_benefic_malefic"} and value in (None, {}, [], "") ] convergence = present["marriage_convergence"] or {} @@ -1683,7 +1712,8 @@ def _collect_strict_evidence(route: str, result: Dict[str, Any]) -> Dict[str, An ), } audit = ( - _external_activation_audit(present.get("external_activation")) + _official_snapshot_audit(present.get("vedastro_official_snapshot")) + + _external_activation_audit(present.get("external_activation")) + _external_technique_audit(present.get("external_technique_evidence")) ) if audit: @@ -1728,10 +1758,12 @@ def _collect_strict_evidence(route: str, result: Dict[str, Any]) -> Dict[str, An present["kakshya_finance_support"] = _derive_kakshya_finance_support(_safe_get(modules, "kakshya")) present["external_activation"] = _derive_external_activation_support(modules, "wealth") present["external_technique_evidence"] = _derive_external_technique_evidence(modules, "wealth") + present["vedastro_official_snapshot"] = official_snapshot_evidence(modules) + present["source_priority"] = modules.get("source_priority") if isinstance(modules.get("source_priority"), dict) else {} present["dignity_guardrail"] = _derive_dignity_guardrail(route, present) present["functional_benefic_malefic"] = _derive_functional_benefic_malefic(modules) missing = [key for key, value in present.items() if key not in { - "chart", "external_activation", "external_technique_evidence", "dignity_guardrail", "gains_convergence", "career_convergence", "avayogi_risk", "ashtakavarga_finance_support", "shadbala_component_audit", "asc_sign", "pav_finance_support", "sodhita_finance_support", "kakshya_finance_support", "functional_benefic_malefic" + "chart", "external_activation", "external_technique_evidence", "vedastro_official_snapshot", "source_priority", "dignity_guardrail", "gains_convergence", "career_convergence", "avayogi_risk", "ashtakavarga_finance_support", "shadbala_component_audit", "asc_sign", "pav_finance_support", "sodhita_finance_support", "kakshya_finance_support", "functional_benefic_malefic" } and value in (None, {}, [], "")] convergence_hits: List[Dict[str, Any]] = [ item for item in [ @@ -1772,7 +1804,8 @@ def _collect_strict_evidence(route: str, result: Dict[str, Any]) -> Dict[str, An ), } audit = ( - _external_activation_audit(present.get("external_activation")) + _official_snapshot_audit(present.get("vedastro_official_snapshot")) + + _external_activation_audit(present.get("external_activation")) + _external_technique_audit(present.get("external_technique_evidence")) ) if audit: @@ -1876,6 +1909,14 @@ def _maybe_attach_vedastro_evidence( enriched = dict(result) enriched["modules"] = dict(modules) enriched["modules"]["vedastro_range_scan_result"] = attached_scan + official_snapshot = attached_scan.get("official_full_snapshot") + if isinstance(official_snapshot, dict): + try: + from vedastro_priority import apply_vedastro_source_priority + + apply_vedastro_source_priority(enriched, official_snapshot=official_snapshot) + except Exception: + enriched["modules"]["vedastro_official_full_snapshot"] = official_snapshot return enriched diff --git a/scripts/jyotish_api_server.py b/scripts/jyotish_api_server.py index c0ee35a1..5f3740c7 100644 --- a/scripts/jyotish_api_server.py +++ b/scripts/jyotish_api_server.py @@ -57,6 +57,7 @@ def _attach_vedastro_main_entry_overview(chart_result, birth_payload): try: orchestrator = _load_local_module('vedastro_evidence_orchestrator') + priority = _load_local_module('vedastro_priority') except Exception: return chart_result @@ -65,7 +66,7 @@ def _attach_vedastro_main_entry_overview(chart_result, birth_payload): or birth_payload.get('today') or datetime.utcnow().strftime('%Y-%m-%d') )[:10] - modules['vedastro_range_scan_result'] = orchestrator.orchestrate_vedastro_evidence({ + vedastro_evidence = orchestrator.orchestrate_vedastro_evidence({ 'year': birth_payload.get('year'), 'month': birth_payload.get('month'), 'day': birth_payload.get('day'), @@ -78,6 +79,17 @@ def _attach_vedastro_main_entry_overview(chart_result, birth_payload): 'ayanamsa_policy': birth_payload.get('ayanamsa') or 'lahiri', 'node_policy': birth_payload.get('node_mode') or birth_payload.get('nodeMode') or 'mean', }, route='overview', reference_date=reference_date, case_id='api_chart') + if isinstance(vedastro_evidence, dict): + metadata = vedastro_evidence.get('source_metadata') + if not isinstance(metadata, dict): + metadata = {} + metadata.setdefault('ingestion_profile', 'main_entry_overview') + metadata.setdefault('reference_date', reference_date) + vedastro_evidence['source_metadata'] = metadata + modules['vedastro_range_scan_result'] = vedastro_evidence + official_snapshot = vedastro_evidence.get('official_full_snapshot') if isinstance(vedastro_evidence, dict) else None + if isinstance(official_snapshot, dict): + priority.apply_vedastro_source_priority(chart_result, official_snapshot=official_snapshot) return chart_result @@ -134,6 +146,39 @@ def _build_vedastro_overview_payload_from_chart(chart): 'visibility': 'user_visible_overview_only', } + +def _build_vedastro_official_full_snapshot_payload_from_chart(chart): + modules = chart.get('modules') if isinstance(chart, dict) else {} + snapshot = modules.get('vedastro_official_full_snapshot') if isinstance(modules, dict) else {} + if not isinstance(snapshot, dict) or not snapshot: + return { + 'status': 'blocked', + 'available': False, + 'operation': 'official_full_snapshot', + 'primary_source': 'vedastro_official', + 'boundary_note': 'VedAstro official full snapshot is not attached.', + } + manifest = snapshot.get('request_manifest') if isinstance(snapshot.get('request_manifest'), dict) else {} + requests = manifest.get('requests') if isinstance(manifest.get('requests'), list) else [] + sections = snapshot.get('snapshot_sections') if isinstance(snapshot.get('snapshot_sections'), dict) else {} + return { + 'status': snapshot.get('status') or 'blocked', + 'available': bool(snapshot.get('available')), + 'operation': snapshot.get('operation') or 'official_full_snapshot', + 'primary_source': snapshot.get('primary_source') or 'vedastro_official', + 'section_statuses': snapshot.get('section_statuses') or {}, + 'snapshot_section_keys': sorted(sections.keys()), + 'request_section_count': len(requests), + 'request_sections': [item.get('section') for item in requests if isinstance(item, dict)], + 'method_catalog': manifest.get('method_catalog') or {}, + 'user_visibility': snapshot.get('user_visibility') or 'backend_raw_evidence_not_direct_user_report', + 'source_metadata': snapshot.get('source_metadata') or {}, + 'boundary_note': ( + snapshot.get('reason') + or 'VedAstro official full snapshot is the primary raw evidence layer; user reports consume selected slices only.' + ), + } + SIGNS = ['Aries','Taurus','Gemini','Cancer','Leo','Virgo', 'Libra','Scorpio','Sagittarius','Capricorn','Aquarius','Pisces'] @@ -2300,6 +2345,7 @@ class JyotishAPIHandler(BaseHTTPRequestHandler): shadbala = chart.get('shadbala') or {} functional_layer = self._functional_benefic_malefic_snapshot(planets, ascendant) vedastro_overview = _build_vedastro_overview_payload_from_chart(chart) + vedastro_official_full_snapshot = _build_vedastro_official_full_snapshot_payload_from_chart(chart) _attach_guided_topics(chart) modules = chart.get('modules') if isinstance(chart.get('modules'), dict) else {} guided_topics = modules.get('guided_topics') if isinstance(modules.get('guided_topics'), list) else [] @@ -2328,10 +2374,13 @@ class JyotishAPIHandler(BaseHTTPRequestHandler): )[:7] core_planets = { planet: { + 'source': pdata.get('source'), 'sign': pdata.get('sign'), 'degree': pdata.get('degree'), + 'degree_in_sign': pdata.get('degree_in_sign'), 'house': pdata.get('house'), 'lon': pdata.get('lon'), + 'vargas': pdata.get('vargas'), } for planet, pdata in planets.items() if planet in {'Sun', 'Moon', 'Mars', 'Mercury', 'Jupiter', 'Venus', 'Saturn', 'Rahu', 'Ketu'} @@ -2345,6 +2394,7 @@ class JyotishAPIHandler(BaseHTTPRequestHandler): f'本盘使用 {ayanamsa_display} ayanamsa，节点口径为 {node_mode}。', '不要仅凭单一配置下结论；核心判断至少交叉 D1、D9、Dasha、Shadbala/Ashtakavarga 或 Transit 中的两个证据层。', '必须显式标注置信度和边界：Dasha/PDF 起点差异、Shadbala 外部绝对值 oracle 尚未完成时，不得声称已经完全校准。', + 'VedAstro 官方全量快照是第一原始证据层；若该层 blocked，必须把本地结果标记为 fallback。', ] oracle_progress = { 'scope': 'external_oracle_evidence_validation', @@ -2382,6 +2432,7 @@ class JyotishAPIHandler(BaseHTTPRequestHandler): 'shadbala_ranking': top_strength, }, 'functional_benefic_malefic': functional_layer, + 'vedastro_official_full_snapshot': vedastro_official_full_snapshot, 'vedastro_overview': vedastro_overview, 'guided_topics': guided_topics, 'capability_evidence_pool': capability_evidence_pool, diff --git a/scripts/jyotish_engine.py b/scripts/jyotish_engine.py index e4896441..1699f892 100644 --- a/scripts/jyotish_engine.py +++ b/scripts/jyotish_engine.py @@ -822,10 +822,14 @@ def _planet_snapshot(planets, planet_name): if not isinstance(pdata, dict): return {} return { + 'source': pdata.get('source'), 'sign': pdata.get('sign'), 'sign_cn': pdata.get('sign_cn'), 'house': pdata.get('house'), + 'degree': pdata.get('degree'), + 'lon': pdata.get('lon'), 'degree_in_sign': pdata.get('degree_in_sign'), + 'vargas': pdata.get('vargas'), 'nakshatra': pdata.get('nakshatra'), 'nakshatra_pada': pdata.get('nakshatra_pada'), 'status': pdata.get('status'), @@ -1146,6 +1150,38 @@ def _build_vedastro_overview_payload(modules): } +def _build_vedastro_official_full_snapshot_payload(modules): + snapshot = modules.get('vedastro_official_full_snapshot') if isinstance(modules, dict) else {} + if not isinstance(snapshot, dict) or not snapshot: + return { + 'status': 'blocked', + 'available': False, + 'operation': 'official_full_snapshot', + 'primary_source': 'vedastro_official', + 'boundary_note': 'VedAstro official full snapshot is not attached.', + } + manifest = snapshot.get('request_manifest') if isinstance(snapshot.get('request_manifest'), dict) else {} + requests = manifest.get('requests') if isinstance(manifest.get('requests'), list) else [] + snapshot_sections = snapshot.get('snapshot_sections') if isinstance(snapshot.get('snapshot_sections'), dict) else {} + return { + 'status': snapshot.get('status') or 'blocked', + 'available': bool(snapshot.get('available')), + 'operation': snapshot.get('operation') or 'official_full_snapshot', + 'primary_source': snapshot.get('primary_source') or 'vedastro_official', + 'section_statuses': snapshot.get('section_statuses') or {}, + 'snapshot_section_keys': sorted(snapshot_sections.keys()), + 'request_section_count': len(requests), + 'request_sections': [item.get('section') for item in requests if isinstance(item, dict)], + 'method_catalog': manifest.get('method_catalog') or {}, + 'user_visibility': snapshot.get('user_visibility') or 'backend_raw_evidence_not_direct_user_report', + 'source_metadata': snapshot.get('source_metadata') or {}, + 'boundary_note': ( + snapshot.get('reason') + or 'VedAstro official full snapshot is the primary raw evidence layer; user reports consume selected slices only.' + ), + } + + def _build_ai_prompt_pack(report): """Build a compact, evidence-first prompt pack for downstream AI/RAG reading.""" modules = report.get('modules', {}) if isinstance(report, dict) else {} @@ -1167,6 +1203,7 @@ def _build_ai_prompt_pack(report): relationship_narrative = _build_relationship_narrative_payload(modules.get('relationship_strict_evidence')) vimsopaka_semantic_summary = _build_vimsopaka_semantic_summary(modules.get('vimsopaka')) vedastro_overview = _build_vedastro_overview_payload(modules) + vedastro_official_full_snapshot = _build_vedastro_official_full_snapshot_payload(modules) guided_topics = modules.get('guided_topics') if isinstance(modules.get('guided_topics'), list) else build_guided_topics(report) capability_evidence_pool = build_capability_evidence_pool_summary() @@ -1239,6 +1276,7 @@ def _build_ai_prompt_pack(report): }, 'oracle_progress': oracle_progress, 'functional_benefic_malefic': functional_layer, + 'vedastro_official_full_snapshot': vedastro_official_full_snapshot, 'vedastro_overview': vedastro_overview, 'guided_topics': guided_topics, 'capability_evidence_pool': capability_evidence_pool, @@ -1256,6 +1294,7 @@ def _build_ai_prompt_pack(report): "输出结构建议：参数声明、核心星盘、关系/事业/财富/健康分主题、当前时机、证据表、风险边界、可行动建议。", "若引用经典法则，请优先检索 retrieval_plan.local_reference_docs；需要外部断语时再做 web/source verification。", "若 evidence_snapshot.vedastro_overview.status 为 ok，请把它作为用户可见外部概览证据明确写出，但不要把 overview-only 结果误当作长周期精扫结论。", + "VedAstro 官方全量快照是第一原始证据层；若 evidence_snapshot.vedastro_official_full_snapshot.status 不是 ok/partial，必须说明官方全量资料 blocked，并把本地结果标记为 fallback。", "若 evidence_snapshot.capability_evidence_pool 存在，请把 89 项视为后台备选证据池；不要把所有能力条目平铺成结论，也不要让 audit_only/alias 条目影响占星判断。", ] @@ -1327,6 +1366,11 @@ def _attach_vedastro_main_entry_overview(report, args): 'tz': getattr(args, 'tz', None), 'ayanamsa_policy': getattr(args, 'ayanamsa', None) or _current_ayanamsa_name(args), 'node_policy': getattr(args, 'node_mode', 'mean'), + 'reference_date': ( + getattr(args, 'transit_date', None) + or getattr(args, 'today', None) + or datetime.now().strftime('%Y-%m-%d') + ), } def _scan_domain(domain: str): @@ -1411,6 +1455,51 @@ def _attach_vedastro_main_entry_overview(report, args): return report +def _attach_vedastro_official_full_snapshot(report, args): + if not isinstance(report, dict): + return report + modules = report.setdefault('modules', {}) + if not isinstance(modules, dict): + return report + if modules.get('vedastro_official_full_snapshot'): + return report + + try: + from vedastro_service_adapter import run_official_full_snapshot_for_case + from vedastro_priority import apply_vedastro_source_priority + except Exception as exc: # pragma: no cover - import guard + report.setdefault('warnings', []).append(f"vedastro-official-full-snapshot-import: {exc}") + return report + + case = { + 'year': getattr(args, 'year', None), + 'month': getattr(args, 'month', None), + 'day': getattr(args, 'day', None), + 'hour': getattr(args, 'hour', None), + 'minute': getattr(args, 'minute', None), + 'second': _arg_second(args), + 'lat': getattr(args, 'lat', None), + 'lon': getattr(args, 'lon', None), + 'tz': getattr(args, 'tz', None), + 'ayanamsa_policy': getattr(args, 'ayanamsa', None) or _current_ayanamsa_name(args), + 'node_policy': getattr(args, 'node_mode', 'mean'), + 'reference_date': ( + getattr(args, 'transit_date', None) + or getattr(args, 'today', None) + or datetime.now().strftime('%Y-%m-%d') + ), + } + modules['vedastro_official_full_snapshot'] = run_official_full_snapshot_for_case( + case, + case_id='full_reading_official_primary', + ) + apply_vedastro_source_priority( + report, + official_snapshot=modules['vedastro_official_full_snapshot'], + ) + return report + + def _load_relationship_strict_collector(): try: from mcp_server import _collect_strict_evidence as collector @@ -5132,6 +5221,11 @@ def cmd_full_reading(args): except Exception as e: report['errors'].append(f"relationship-strict-evidence: {e}") + try: + _attach_vedastro_official_full_snapshot(report, args) + except Exception as e: + report['warnings'].append(f"vedastro-official-full-snapshot: {e}") + try: _attach_vedastro_main_entry_overview(report, args) except Exception as e: diff --git a/scripts/vedastro_evidence_orchestrator.py b/scripts/vedastro_evidence_orchestrator.py index 6cebe9a0..60e944e0 100644 --- a/scripts/vedastro_evidence_orchestrator.py +++ b/scripts/vedastro_evidence_orchestrator.py @@ -14,10 +14,15 @@ from typing import Any try: from scripts.vedastro_service_adapter import ( VEDASTRO_CALCULATION_COVERAGE, + run_official_full_snapshot_for_case, run_range_scan_for_case, ) except ModuleNotFoundError: # pragma: no cover - script execution path - from vedastro_service_adapter import VEDASTRO_CALCULATION_COVERAGE, run_range_scan_for_case + from vedastro_service_adapter import ( + VEDASTRO_CALCULATION_COVERAGE, + run_official_full_snapshot_for_case, + run_range_scan_for_case, + ) ROUTE_DOMAIN_MAP = { @@ -75,6 +80,7 @@ def orchestrate_vedastro_evidence( domains = ROUTE_DOMAIN_MAP.get(route, ROUTE_DOMAIN_MAP["general"]) window_start, window_end = (start_date, end_date) if start_date and end_date else _default_window(reference_date) case = _normalize_case(birth_payload) + case["reference_date"] = str(reference_date or window_start)[:10] domain_reports: dict[str, Any] = {} evidence_ledger: list[dict[str, Any]] = [] top_events_by_domain: dict[str, Any] = {} @@ -82,6 +88,10 @@ def orchestrate_vedastro_evidence( domain_event_counts: dict[str, int] = {} available = False first_reason = None + official_full_snapshot = run_official_full_snapshot_for_case( + case, + case_id=f"{case_id}_official_full_snapshot", + ) for domain in domains: report = run_range_scan_for_case( @@ -115,11 +125,13 @@ def orchestrate_vedastro_evidence( "top_events_by_domain": top_events_by_domain, "evidence_ledger": evidence_ledger, "reason": None if status == "ok" else first_reason, + "official_full_snapshot": official_full_snapshot, "domain_reports": domain_reports, "source_metadata": { "auto_ingested_by": "VedAstroEvidenceOrchestrator", - "strategy": "minimal_route_scoped_orchestration", + "strategy": "official_full_snapshot_first_then_route_scoped_range_scan", "node_coverage": { + "official_full_snapshot_first": True, "strategy": "domain_scoped_range_scan", "official_calculation_coverage": VEDASTRO_CALCULATION_COVERAGE, "selected_domains": domains, diff --git a/scripts/vedastro_priority.py b/scripts/vedastro_priority.py new file mode 100644 index 00000000..de15b09b --- /dev/null +++ b/scripts/vedastro_priority.py @@ -0,0 +1,167 @@ +"""Shared VedAstro-first source priority helpers. + +This module keeps the user-facing data order identical across the CLI engine, +API server, and MCP strict workflow: + +1. VedAstro official full snapshot when it contains an official chart. +2. Local modules as supplemental evidence and cross-checks. +3. Local chart as fallback only when the official snapshot is blocked. +""" + +from __future__ import annotations + +from copy import deepcopy +from typing import Any + + +SOURCE_PRIORITY = [ + "vedastro_official_snapshot", + "local_supplemental_modules", + "local_engine_fallback_when_official_blocked", +] + + +def official_chart_available(official_snapshot: dict[str, Any] | None) -> bool: + if not isinstance(official_snapshot, dict): + return False + official_chart = official_snapshot.get("official_chart") + if not isinstance(official_chart, dict): + return False + return ( + isinstance(official_chart.get("planets"), dict) + and bool(official_chart.get("planets")) + and isinstance(official_chart.get("ascendant"), dict) + and bool(official_chart.get("ascendant")) + ) + + +def _local_chart_from(report: dict[str, Any], modules: dict[str, Any]) -> dict[str, Any]: + chart = report.get("chart") + if isinstance(chart, dict) and chart: + return chart + chart = modules.get("chart") + return chart if isinstance(chart, dict) else {} + + +def _blocked_reason(official_snapshot: dict[str, Any] | None) -> str: + status = "missing" + if isinstance(official_snapshot, dict): + status = str(official_snapshot.get("status") or "blocked") + return f"VedAstro official snapshot blocked: {status}" + + +def build_source_priority_metadata( + official_snapshot: dict[str, Any] | None, + *, + official_primary: bool, +) -> dict[str, Any]: + status = official_snapshot.get("status") if isinstance(official_snapshot, dict) else "missing" + return { + "mode": "vedastro_official_primary" if official_primary else "local_fallback_official_blocked", + "priority": list(SOURCE_PRIORITY), + "official_snapshot_first": True, + "official_snapshot_status": status or "blocked", + "local_engine_role": ( + "supplemental_crosscheck_or_fallback" + if official_primary + else "fallback_only_because_official_blocked" + ), + "user_visible_policy": ( + "show_vedastro_verified_when_official_chart_available" + if official_primary + else "show_local_fallback_with_official_blocked_boundary" + ), + } + + +def apply_vedastro_source_priority( + report: dict[str, Any], + *, + official_snapshot: dict[str, Any] | None, +) -> dict[str, Any]: + if not isinstance(report, dict): + return report + modules = report.setdefault("modules", {}) + if not isinstance(modules, dict): + modules = {} + report["modules"] = modules + + if isinstance(official_snapshot, dict): + modules["vedastro_official_full_snapshot"] = official_snapshot + + local_chart = _local_chart_from(report, modules) + + if official_chart_available(official_snapshot): + official_chart = official_snapshot.get("official_chart") # type: ignore[union-attr] + modules["local_engine_chart_fallback"] = deepcopy(local_chart) + public_chart = { + **deepcopy(local_chart), + "source": "vedastro_official_primary", + "primary_source": "vedastro_official", + "fallback_source": "local_engine", + "local_engine_role": "supplemental_crosscheck_or_fallback", + "local_crosscheck_status": "pending", + "source_priority": list(SOURCE_PRIORITY), + "planets": official_chart.get("planets", {}), + "ascendant": official_chart.get("ascendant", {}), + "houses": official_chart.get("houses", {}), + "birth_info": local_chart.get("birth_info", report.get("birth_info", report.get("birth", {}))), + "official_coverage": official_chart.get("coverage", {}), + } + report["chart"] = public_chart + modules["chart"] = public_chart + for key in ("source", "primary_source", "fallback_source", "planets", "ascendant", "houses"): + if key in public_chart: + report[key] = public_chart[key] + modules["source_priority"] = build_source_priority_metadata( + official_snapshot, + official_primary=True, + ) + return report + + if local_chart: + fallback_chart = { + **deepcopy(local_chart), + "source": "local_engine_fallback", + "primary_source": "local_engine", + "fallback_reason": _blocked_reason(official_snapshot), + "local_engine_role": "fallback_only_because_official_blocked", + "source_priority": list(SOURCE_PRIORITY), + } + report["chart"] = fallback_chart + modules["chart"] = fallback_chart + for key in ("source", "primary_source", "planets", "ascendant", "houses"): + if key in fallback_chart: + report[key] = fallback_chart[key] + + modules["source_priority"] = build_source_priority_metadata( + official_snapshot, + official_primary=False, + ) + return report + + +def official_snapshot_evidence(modules: dict[str, Any]) -> dict[str, Any]: + snapshot = modules.get("vedastro_official_full_snapshot") if isinstance(modules, dict) else {} + source_priority = modules.get("source_priority") if isinstance(modules, dict) else {} + if not isinstance(snapshot, dict) or not snapshot: + return { + "level": "blocked", + "source": "vedastro_official", + "status": "missing", + "operation": "official_full_snapshot", + "source_priority": source_priority if isinstance(source_priority, dict) else {}, + "reason": "VedAstro official full snapshot is not attached.", + } + level = "primary" if official_chart_available(snapshot) else "blocked" + return { + "level": level, + "source": "vedastro_official", + "status": snapshot.get("status") or "blocked", + "available": bool(snapshot.get("available")), + "operation": snapshot.get("operation") or "official_full_snapshot", + "source_priority": source_priority if isinstance(source_priority, dict) else {}, + "section_statuses": snapshot.get("section_statuses") or {}, + "chart_available": official_chart_available(snapshot), + "reason": snapshot.get("reason"), + } diff --git a/scripts/vedastro_service_adapter.py b/scripts/vedastro_service_adapter.py index 812c8144..d69669d7 100644 --- a/scripts/vedastro_service_adapter.py +++ b/scripts/vedastro_service_adapter.py @@ -15,6 +15,7 @@ import json import os import socket import time +from datetime import datetime from pathlib import Path from typing import Any from urllib import request, error @@ -86,6 +87,62 @@ SUPPORTED_EXTERNAL_TECHNIQUE_DOMAINS = {"marriage", "wealth", "career", "general OFFICIAL_SEARCH_EVENTS_ENDPOINT_PATH = "/Calculate/SearchEvents" OFFICIAL_SEARCH_EVENTS_METHOD = "POST" OFFICIAL_SEARCH_EVENTS_PROFILE_VERSION = "official_builder_search_events_v1" +OFFICIAL_FULL_SNAPSHOT_PROFILE_VERSION = "official_full_snapshot_v1" +OFFICIAL_METHOD_CATALOG_URL = "https://vedastro.org/Complete-List-VedAstro-API-Methods-Calculators.html" +OFFICIAL_FULL_SNAPSHOT_METHODS = [ + { + "section": "chart_core", + "endpoint_path": "/Calculate/AllPlanetData", + "calculator_name": "AllPlanetData", + "role": "core_chart_raw_evidence", + "description": "Core planet, ascendant, house, nakshatra, ayanamsa and node-mode evidence when supported by the official service.", + "fanout": "planetName", + }, + { + "section": "house_core", + "endpoint_path": "/Calculate/AllHouseData", + "calculator_name": "AllHouseData", + "role": "core_house_raw_evidence", + "description": "Official house data snapshot when supported by the official service.", + "fanout": "houseName", + }, + { + "section": "dasha_all", + "endpoint_path": "/Calculate/DasaAtRange", + "calculator_name": "DasaAtRange", + "role": "all_dasha_raw_evidence", + "description": "Official dasha timeline snapshot where available.", + }, + { + "section": "events_overview", + "endpoint_path": OFFICIAL_SEARCH_EVENTS_ENDPOINT_PATH, + "calculator_name": "SearchEvents", + "role": "life_event_raw_evidence", + "description": "Official event radar using SearchEvents for career, marriage and wealth tags.", + }, +] +OFFICIAL_FULL_SNAPSHOT_BACKLOG_SECTIONS = [ + { + "section": "varga_all", + "role": "all_varga_raw_evidence", + "status": "catalog_pending", + "description": "Awaiting official method mapping for all divisional charts; local varga remains fallback until mapped.", + }, + { + "section": "shadbala", + "role": "strength_raw_evidence", + "status": "catalog_pending", + "description": "Awaiting official method mapping for Shadbala; local Shadbala remains fallback until mapped.", + }, + { + "section": "ashtakavarga", + "role": "ashtakavarga_raw_evidence", + "status": "catalog_pending", + "description": "Awaiting official method mapping for Ashtakavarga; local Ashtakavarga remains fallback until mapped.", + }, +] +OFFICIAL_SNAPSHOT_PLANETS = ["Sun", "Moon", "Mars", "Mercury", "Jupiter", "Venus", "Saturn", "Rahu", "Ketu", "Ascendant"] +OFFICIAL_SNAPSHOT_HOUSES = [f"House{i}" for i in range(1, 13)] OFFICIAL_RANGE_SCAN_EVENT_TAGS = { "marriage": ["Marriage", "Personal", "General"], "wealth": ["LendingMoney", "BorrowingMoney", "BuyingSelling", "General"], @@ -418,7 +475,40 @@ def schema() -> dict[str, Any]: "AtTime | StartTime + EndTime + PrecisionHours", ], }, + "official_full_snapshot_request_contract": { + "profile_version": OFFICIAL_FULL_SNAPSHOT_PROFILE_VERSION, + "primary_source": "vedastro_official", + "method_catalog_url": OFFICIAL_METHOD_CATALOG_URL, + "strategy": "fetch_official_raw_sections_first_then_local_crosscheck", + "common_body_fields": [ + "BirthTime", + "Ayanamsa", + "NodeMode", + "CalculationPreferences", + ], + "request_sections": [ + { + "section": item["section"], + "endpoint_path": item["endpoint_path"], + "calculator_name": item.get("calculator_name"), + "role": item["role"], + } + for item in OFFICIAL_FULL_SNAPSHOT_METHODS + ], + "backlog_sections": OFFICIAL_FULL_SNAPSHOT_BACKLOG_SECTIONS, + "user_visibility": "backend_raw_evidence_not_direct_user_report", + }, "vedastro_calculation_coverage": VEDASTRO_CALCULATION_COVERAGE, + "official_full_snapshot_response_contract": [ + "backend", + "available", + "status", + "operation", + "primary_source", + "snapshot_sections", + "request_manifest", + "source_metadata", + ], "external_technique_request_contract": [ "operation", "role", @@ -515,9 +605,20 @@ def _format_std_time(date_text: str, hour: Any, minute: Any, tz: Any) -> str: return f"{hour_int:02d}:{minute_int:02d} {day}/{month}/{year} {tz}" -def _time_json_from_case(case: dict[str, Any], date_text: str) -> dict[str, Any]: +def _time_json_from_case( + case: dict[str, Any], + date_text: str, + *, + hour: Any | None = None, + minute: Any | None = None, +) -> dict[str, Any]: return { - "StdTime": _format_std_time(date_text, case.get("hour", 0), case.get("minute", 0), case.get("tz", "+00:00")), + "StdTime": _format_std_time( + date_text, + case.get("hour", 0) if hour is None else hour, + case.get("minute", 0) if minute is None else minute, + case.get("tz", "+00:00"), + ), "Location": { "Name": case.get("case_id") or "UserLocation", "Latitude": case.get("lat"), @@ -590,6 +691,108 @@ def _build_live_sampling_search_events_profile(request_preview: dict[str, Any]) } +def _official_common_body(case: dict[str, Any]) -> dict[str, Any]: + normalized = dict(case) + normalized["tz"] = _normalize_tz(normalized) + return { + "time": _time_json_from_case( + normalized, + f"{int(normalized['year']):04d}-{int(normalized['month']):02d}-{int(normalized['day']):02d}", + ), + "Ayanamsa": str(normalized.get("ayanamsa_policy") or "lahiri"), + "NodeMode": str(normalized.get("node_policy") or "mean"), + "CalculationPreferences": { + "scope": "all_supported_official_calculations", + "user_visibility": "backend_raw_evidence_not_direct_user_report", + }, + } + + +def _official_snapshot_reference_date(case: dict[str, Any]) -> str: + for key in ("reference_date", "today", "transit_date", "current_date"): + value = case.get(key) + if not value: + continue + raw = str(value)[:10] + try: + datetime.strptime(raw, "%Y-%m-%d") + return raw + except ValueError: + continue + return datetime.utcnow().strftime("%Y-%m-%d") + + +def _official_dasha_range_body(case: dict[str, Any], common_body: dict[str, Any]) -> dict[str, Any]: + normalized = dict(case) + normalized["tz"] = _normalize_tz(normalized) + reference = datetime.strptime(_official_snapshot_reference_date(normalized), "%Y-%m-%d").date() + start_date = reference.replace(month=1, day=1) + end_date = reference.replace(month=12, day=31) + return { + "birthTime": common_body["time"], + "startTime": _time_json_from_case(normalized, start_date.isoformat(), hour=0, minute=0), + "endTime": _time_json_from_case(normalized, end_date.isoformat(), hour=23, minute=59), + "levels": int(normalized.get("dasha_levels") or 3), + "precisionHours": int(normalized.get("dasha_precision_hours") or 100), + "Ayanamsa": common_body["Ayanamsa"], + } + + +def _official_full_snapshot_manifest(case: dict[str, Any], case_id: str = "user_chart") -> dict[str, Any]: + common_body = _official_common_body(case) + reference_date = _official_snapshot_reference_date(case) + headers: dict[str, str] = {"Content-Type": "application/json"} + api_key = os.environ.get("VEDASTRO_API_KEY", "").strip() + if api_key: + headers["x-api-key"] = api_key + requests = [] + for item in OFFICIAL_FULL_SNAPSHOT_METHODS: + body = dict(common_body) + if item["section"] == "events_overview": + body = { + "BirthTime": common_body["time"], + "Ayanamsa": common_body["Ayanamsa"], + "EventTagList": sorted({tag for tags in OFFICIAL_RANGE_SCAN_EVENT_TAGS.values() for tag in tags}), + "AtTime": common_body["time"], + } + if item["section"] == "dasha_all": + body = _official_dasha_range_body(case, common_body) + fanout_values = [] + if item.get("fanout") == "planetName": + fanout_values = OFFICIAL_SNAPSHOT_PLANETS + elif item.get("fanout") == "houseName": + fanout_values = OFFICIAL_SNAPSHOT_HOUSES + requests.append( + { + "section": item["section"], + "role": item["role"], + "calculator_name": item.get("calculator_name"), + "endpoint_path": item["endpoint_path"], + "method": "POST", + "headers": headers, + "body": body, + "fanout_parameter": item.get("fanout"), + "fanout_values": fanout_values, + "description": item["description"], + } + ) + return { + "operation": "official_full_snapshot", + "profile_version": OFFICIAL_FULL_SNAPSHOT_PROFILE_VERSION, + "source_role": "primary_official_raw_evidence", + "primary_source": "vedastro_official", + "case_id": case_id, + "reference_date": reference_date, + "method_catalog": { + "url": OFFICIAL_METHOD_CATALOG_URL, + "declared_coverage": VEDASTRO_CALCULATION_COVERAGE, + "catalog_role": "all_supported_method_reference_not_user_visible_output", + "backlog_sections": OFFICIAL_FULL_SNAPSHOT_BACKLOG_SECTIONS, + }, + "requests": requests, + } + + def _external_technique_preview( case: dict[str, Any], domain: str, @@ -1114,6 +1317,385 @@ def run_case(case_id: str) -> dict[str, Any]: return _normalize_success(payload, endpoint, request_preview) +def _official_full_snapshot_metadata(endpoint: str | None, manifest: dict[str, Any]) -> dict[str, Any]: + metadata = { + "transport": "http_json_service_boundary", + "operation": "official_full_snapshot", + "primary_source": "vedastro_official", + "provenance_mode": "vedastro_official_primary_candidate", + "timeout_seconds": _timeout_seconds(), + "retry_policy": {**RETRY_POLICY, "backoff_seconds": _backoff_seconds()}, + "network_execution_env": ALLOW_NETWORK_ENV, + "method_catalog_url": OFFICIAL_METHOD_CATALOG_URL, + "reference_date": manifest.get("reference_date"), + "request_hash": _hash_payload(manifest), + } + if endpoint: + metadata["endpoint"] = endpoint + metadata["endpoint_host"] = _endpoint_host(endpoint) + return metadata + + +def _payload_status(payload: dict[str, Any]) -> str: + if not isinstance(payload, dict): + return "invalid" + if str(payload.get("Status") or "").lower() == "fail": + failure_text = json.dumps(payload.get("Payload"), ensure_ascii=False).lower() + if "rate limit" in failure_text or "calls/minute" in failure_text or "too many requests" in failure_text: + return "rate_limited" + return "ok" if payload.get("Status") == "Pass" else "fail" + + +def _aggregate_section_status(statuses: list[str]) -> str: + if statuses and all(status == "ok" for status in statuses): + return "ok" + if any(status == "rate_limited" for status in statuses): + return "rate_limited" + return "partial" + + +def _degrees_from_sign_payload(value: Any) -> float | None: + if not isinstance(value, dict): + return None + degrees = value.get("DegreesIn") if isinstance(value.get("DegreesIn"), dict) else {} + raw = degrees.get("TotalDegrees") + try: + return float(raw) + except (TypeError, ValueError): + return None + + +def _sign_position(value: Any) -> dict[str, Any] | None: + if not isinstance(value, dict): + return None + sign = value.get("Name") + degree = _degrees_from_sign_payload(value) + if not sign: + return None + return { + "sign": sign, + "degree_in_sign": degree, + } + + +def _extract_all_planet_data(payload: dict[str, Any]) -> dict[str, Any]: + if not isinstance(payload, dict): + return {} + body = payload.get("Payload") if isinstance(payload.get("Payload"), dict) else {} + data = body.get("AllPlanetData") if isinstance(body.get("AllPlanetData"), dict) else {} + return data if isinstance(data, dict) else {} + + +def _extract_all_house_data(payload: dict[str, Any]) -> dict[str, Any]: + if not isinstance(payload, dict): + return {} + body = payload.get("Payload") if isinstance(payload.get("Payload"), dict) else {} + data = body.get("AllHouseData") if isinstance(body.get("AllHouseData"), dict) else {} + return data if isinstance(data, dict) else {} + + +def _official_planet_snapshot(planet_name: str, data: dict[str, Any]) -> dict[str, Any]: + d1 = _sign_position(data.get("PlanetRasiD1Sign")) or {} + raw_lon = None + nirayana = data.get("PlanetNirayanaLongitude") + if isinstance(nirayana, dict): + try: + raw_lon = float(nirayana.get("TotalDegrees")) + except (TypeError, ValueError): + raw_lon = None + house_text = data.get("HousePlanetOccupiesBasedOnSign") or data.get("HousePlanetOccupiesBasedOnLongitudes") + house = None + if isinstance(house_text, str) and house_text.lower().startswith("house"): + try: + house = int("".join(ch for ch in house_text if ch.isdigit())) + except ValueError: + house = None + return { + "source": "vedastro_official", + "name": planet_name, + "sign": d1.get("sign"), + "degree_in_sign": d1.get("degree_in_sign"), + "degree": raw_lon, + "lon": raw_lon, + "house": house, + "vargas": { + "D1": d1, + "D2": _sign_position(data.get("PlanetHoraD2Signs")), + "D3": _sign_position(data.get("PlanetDrekkanaD3Sign")), + "D4": _sign_position(data.get("PlanetChaturthamshaD4Sign")), + "D7": _sign_position(data.get("PlanetSaptamshaD7Sign")), + "D9": _sign_position(data.get("PlanetNavamshaD9Sign")), + "D10": _sign_position(data.get("PlanetDashamamshaD10Sign")), + "D12": _sign_position(data.get("PlanetDwadashamshaD12Sign")), + "D16": _sign_position(data.get("PlanetShodashamshaD16Sign")), + "D20": _sign_position(data.get("PlanetVimshamshaD20Sign")), + "D24": _sign_position(data.get("PlanetChaturvimshamshaD24Sign")), + "D27": _sign_position(data.get("PlanetBhamshaD27Sign")), + "D30": _sign_position(data.get("PlanetTrimshamshaD30Sign")), + "D40": _sign_position(data.get("PlanetKhavedamshaD40Sign")), + "D45": _sign_position(data.get("PlanetAkshavedamshaD45Sign")), + "D60": _sign_position(data.get("PlanetShashtyamshaD60Sign")), + }, + "nakshatra": data.get("PlanetConstellation"), + "raw_source_keys": sorted(data.keys()), + } + + +def _official_house_snapshot(house_name: str, data: dict[str, Any]) -> dict[str, Any]: + d1 = _sign_position(data.get("HouseRasiD1Sign") or data.get("HouseBhavaChalitSign")) or {} + return { + "source": "vedastro_official", + "name": house_name, + "sign": d1.get("sign"), + "degree_in_sign": d1.get("degree_in_sign"), + "vargas": { + "D1": d1, + "D2": _sign_position(data.get("HouseHoraD2Sign") or data.get("HouseHoraD2Signs")), + "D3": _sign_position(data.get("HouseDrekkanaD3Sign")), + "D4": _sign_position(data.get("HouseChaturthamshaD4Sign")), + "D7": _sign_position(data.get("HouseSaptamshaD7Sign")), + "D9": _sign_position(data.get("HouseNavamshaD9Sign") or data.get("HouseNavamsaD9Sign")), + "D10": _sign_position(data.get("HouseDashamamshaD10Sign")), + "D12": _sign_position(data.get("HouseDwadashamshaD12Sign")), + "D16": _sign_position(data.get("HouseShodashamshaD16Sign")), + "D20": _sign_position(data.get("HouseVimshamshaD20Sign")), + "D24": _sign_position(data.get("HouseChaturvimshamshaD24Sign")), + "D27": _sign_position(data.get("HouseBhamshaD27Sign")), + "D30": _sign_position(data.get("HouseTrimshamshaD30Sign")), + "D40": _sign_position(data.get("HouseKhavedamshaD40Sign")), + "D45": _sign_position(data.get("HouseAkshavedamshaD45Sign")), + "D60": _sign_position(data.get("HouseShashtyamshaD60Sign")), + }, + "nakshatra": data.get("HouseConstellation"), + "raw_source_keys": sorted(data.keys()), + } + + +def _build_official_chart_from_snapshot(sections: dict[str, Any]) -> dict[str, Any]: + chart_core = sections.get("chart_core") if isinstance(sections.get("chart_core"), dict) else {} + house_core = sections.get("house_core") if isinstance(sections.get("house_core"), dict) else {} + planets: dict[str, Any] = {} + for planet_name, payload in chart_core.items(): + data = _extract_all_planet_data(payload) + if data: + planets[planet_name] = _official_planet_snapshot(planet_name, data) + houses: dict[str, Any] = {} + for house_name, payload in house_core.items(): + data = _extract_all_house_data(payload) + if data: + houses[house_name] = _official_house_snapshot(house_name, data) + ascendant = houses.get("House1") or {} + return { + "source": "vedastro_official", + "primary_source": "vedastro_official", + "planets": planets, + "houses": houses, + "ascendant": ascendant, + "coverage": { + "planet_count": len(planets), + "house_count": len(houses), + "varga_keys": ["D1", "D2", "D3", "D4", "D7", "D9", "D10", "D12", "D16", "D20", "D24", "D27", "D30", "D40", "D45", "D60"], + }, + } + + +def _post_official_snapshot_section(endpoint: str, request_item: dict[str, Any]) -> tuple[dict[str, Any], int, list[int]]: + body = dict(request_item["body"]) + fanout_parameter = request_item.get("fanout_parameter") + fanout_value = request_item.get("fanout_value") + if fanout_parameter and fanout_value is not None: + if fanout_parameter == "planetName": + body[fanout_parameter] = {"Name": str(fanout_value)} + else: + body[fanout_parameter] = str(fanout_value) + section_preview = { + "operation": "official_full_snapshot", + "section": request_item["section"], + "official_request_profile": { + "profile_version": OFFICIAL_FULL_SNAPSHOT_PROFILE_VERSION, + "endpoint_path": request_item["endpoint_path"], + "method": request_item["method"], + "headers": request_item["headers"], + "body": body, + }, + } + return _post_json_with_retry(endpoint, section_preview) + + +def _normalize_official_full_snapshot_success( + endpoint: str, + manifest: dict[str, Any], + sections: dict[str, Any], + section_statuses: dict[str, str], + attempt_count: int, + retry_error_codes: list[int], +) -> dict[str, Any]: + primary_sections = [item["section"] for item in manifest["requests"]] + ok_count = sum(1 for section in primary_sections if section_statuses.get(section) == "ok") + status = "ok" if ok_count == len(primary_sections) else "partial" + rate_limited_sections = [ + section + for section in primary_sections + if section_statuses.get(section) == "rate_limited" + ] + metadata = { + **_official_full_snapshot_metadata(endpoint, manifest), + "called_at": _utc_timestamp(), + "section_statuses": section_statuses, + "section_count": len(primary_sections), + "section_ok_count": ok_count, + "rate_limited_sections": rate_limited_sections, + "attempt_count": attempt_count, + "retry_error_codes": retry_error_codes, + "response_hash": _hash_payload({"sections": sections, "section_statuses": section_statuses}), + } + if rate_limited_sections: + metadata["production_hint"] = "configure_vedastro_api_key_or_self_host_official_api" + result = { + "backend": "vedastro_service_adapter_candidate", + "available": ok_count > 0, + "status": status, + "operation": "official_full_snapshot", + "primary_source": "vedastro_official", + "snapshot_sections": sections, + "official_chart": _build_official_chart_from_snapshot(sections), + "section_statuses": section_statuses, + "request_manifest": manifest, + "user_visibility": "backend_raw_evidence_not_direct_user_report", + "source_metadata": metadata, + } + result["source_metadata"]["artifact_path"] = _write_artifact(result) + return result + + +def _run_official_full_snapshot_case(case: dict[str, Any], case_id: str = "user_chart") -> dict[str, Any]: + user_case = { + "case_id": case_id, + "year": case.get("year"), + "month": case.get("month"), + "day": case.get("day"), + "hour": case.get("hour"), + "minute": case.get("minute"), + "second": case.get("second", 0), + "lat": case.get("lat"), + "lon": case.get("lon"), + "tz": case.get("tz"), + "ayanamsa_policy": case.get("ayanamsa_policy") or case.get("ayanamsa") or "lahiri", + "node_policy": case.get("node_policy") or case.get("node_mode") or "mean", + "reference_date": case.get("reference_date") or case.get("today") or case.get("transit_date") or case.get("current_date"), + "dasha_levels": case.get("dasha_levels"), + "dasha_precision_hours": case.get("dasha_precision_hours"), + } + manifest = _official_full_snapshot_manifest(user_case, case_id) + endpoint = os.environ.get("VEDASTRO_API_ENDPOINT", "").strip() + if not endpoint: + return { + "backend": "vedastro_service_adapter_candidate", + "available": False, + "status": "service_endpoint_not_configured", + "operation": "official_full_snapshot", + "primary_source": "vedastro_official", + "reason": "VEDASTRO_API_ENDPOINT is not configured; official full snapshot stops before network access.", + "snapshot_sections": {}, + "request_manifest": manifest, + "user_visibility": "backend_raw_evidence_not_direct_user_report", + "source_metadata": _official_full_snapshot_metadata(None, manifest), + } + + if os.environ.get(ALLOW_NETWORK_ENV, "").strip().lower() not in {"1", "true", "yes"}: + return { + "backend": "vedastro_service_adapter_candidate", + "available": False, + "status": "network_execution_disabled", + "operation": "official_full_snapshot", + "primary_source": "vedastro_official", + "reason": f"{ALLOW_NETWORK_ENV} is not enabled; official full snapshot stops after building request manifest.", + "snapshot_sections": {}, + "request_manifest": manifest, + "user_visibility": "backend_raw_evidence_not_direct_user_report", + "source_metadata": _official_full_snapshot_metadata(endpoint, manifest), + } + + sections: dict[str, Any] = {} + section_statuses: dict[str, str] = {} + attempt_count = 0 + retry_error_codes: list[int] = [] + for request_item in manifest["requests"]: + section = request_item["section"] + fanout_values = request_item.get("fanout_values") if isinstance(request_item.get("fanout_values"), list) else [] + if fanout_values: + section_payloads: dict[str, Any] = {} + fanout_statuses: dict[str, str] = {} + for value in fanout_values: + fanout_request = {**request_item, "fanout_value": value} + try: + payload, attempts, retries = _post_official_snapshot_section(endpoint, fanout_request) + section_payloads[str(value)] = payload + fanout_statuses[str(value)] = _payload_status(payload) + attempt_count += attempts + retry_error_codes.extend(retries) + except error.HTTPError as exc: + fanout_statuses[str(value)] = f"http_error:{exc.code}" + except (error.URLError, http.client.RemoteDisconnected) as exc: + fanout_statuses[str(value)] = f"network_error:{getattr(exc, 'reason', str(exc))}" + except (TimeoutError, socket.timeout): + fanout_statuses[str(value)] = "timeout" + except json.JSONDecodeError: + fanout_statuses[str(value)] = "invalid_json" + sections[section] = section_payloads + section_statuses[section] = _aggregate_section_status(list(fanout_statuses.values())) + section_statuses[f"{section}_fanout"] = fanout_statuses + continue + + try: + payload, attempts, retries = _post_official_snapshot_section(endpoint, request_item) + sections[section] = payload + section_statuses[section] = _payload_status(payload) + attempt_count += attempts + retry_error_codes.extend(retries) + except error.HTTPError as exc: + section_statuses[section] = f"http_error:{exc.code}" + except (error.URLError, http.client.RemoteDisconnected) as exc: + section_statuses[section] = f"network_error:{getattr(exc, 'reason', str(exc))}" + except (TimeoutError, socket.timeout): + section_statuses[section] = "timeout" + except json.JSONDecodeError: + section_statuses[section] = "invalid_json" + + return _normalize_official_full_snapshot_success( + endpoint, + manifest, + sections, + section_statuses, + attempt_count or 1, + retry_error_codes, + ) + + +def run_official_full_snapshot(case_id: str, reference_date: str | None = None) -> dict[str, Any]: + if case_id not in PARITY_CASES: + return { + "backend": "vedastro_service_adapter_candidate", + "available": False, + "status": "unknown_case_id", + "operation": "official_full_snapshot", + "primary_source": "vedastro_official", + "reason": f"Unknown parity case: {case_id}", + } + case = dict(PARITY_CASES[case_id]) + if reference_date: + case["reference_date"] = reference_date + return _run_official_full_snapshot_case(case, case_id=case_id) + + +def run_official_full_snapshot_for_case( + case: dict[str, Any], + *, + case_id: str = "user_chart", +) -> dict[str, Any]: + return _run_official_full_snapshot_case(case, case_id=case_id) + + def _run_range_scan_case(case: dict[str, Any], domain: str, start_date: str, end_date: str) -> dict[str, Any]: if domain not in SUPPORTED_RANGE_SCAN_DOMAINS: return { @@ -1330,9 +1912,11 @@ def main() -> int: parser.add_argument("--print-schema", action="store_true") parser.add_argument("--case", default="beijing_first_use_demo") parser.add_argument("--range-scan", action="store_true") + parser.add_argument("--official-full-snapshot", action="store_true") parser.add_argument("--domain", choices=sorted(SUPPORTED_EXTERNAL_TECHNIQUE_DOMAINS), default="marriage") parser.add_argument("--start-date", default="2026-01-01") parser.add_argument("--end-date", default="2031-01-01") + parser.add_argument("--reference-date", default=None) parser.add_argument("--external-technique", action="store_true") parser.add_argument("--method", default="") parser.add_argument("--api-endpoint", default="") @@ -1340,6 +1924,8 @@ def main() -> int: if args.print_schema: result = schema() + elif args.official_full_snapshot: + result = run_official_full_snapshot(args.case, reference_date=args.reference_date) elif args.external_technique: result = run_external_technique(args.case, args.domain, args.method, args.api_endpoint) elif args.range_scan: diff --git a/tests/test_vedastro_external_technique_evidence.py b/tests/test_vedastro_external_technique_evidence.py index 9d127d87..3335d83f 100644 --- a/tests/test_vedastro_external_technique_evidence.py +++ b/tests/test_vedastro_external_technique_evidence.py @@ -198,6 +198,45 @@ def test_strict_workflow_requires_vedastro_range_scan_boundary_for_timing_routes } in strict["technique_audit"] +def test_strict_workflow_exposes_official_snapshot_as_primary_evidence_layer() -> None: + modules = _finance_modules() + modules["vedastro_official_full_snapshot"] = { + "status": "partial", + "available": True, + "operation": "official_full_snapshot", + "primary_source": "vedastro_official", + "official_chart": { + "source": "vedastro_official", + "planets": {"Sun": {"source": "vedastro_official", "sign": "Aries"}}, + "ascendant": {"source": "vedastro_official", "sign": "Leo"}, + }, + } + modules["source_priority"] = { + "mode": "vedastro_official_primary", + "priority": [ + "vedastro_official_snapshot", + "local_supplemental_modules", + "local_engine_fallback_when_official_blocked", + ], + "local_engine_role": "supplemental_crosscheck_or_fallback", + "official_snapshot_status": "partial", + } + + strict = _collect_strict_evidence("finance", {"modules": modules}) + + official = strict["present_evidence"]["vedastro_official_snapshot"] + assert official["level"] == "primary" + assert official["source"] == "vedastro_official" + assert official["status"] == "partial" + assert strict["present_evidence"]["source_priority"]["mode"] == "vedastro_official_primary" + assert any( + row["technique"] == "VedAstro Official Full Snapshot" + and row["status"] == "used" + and row["role"] == "primary_raw_evidence" + for row in strict["technique_audit"] + ) + + def test_strict_workflow_marks_vedastro_range_scan_used_without_score_or_label_override() -> None: base_result = {"modules": _finance_modules()} with_range_scan = {"modules": deepcopy(base_result["modules"])} diff --git a/tests/test_vedastro_official_full_snapshot.py b/tests/test_vedastro_official_full_snapshot.py new file mode 100644 index 00000000..0f9f55de --- /dev/null +++ b/tests/test_vedastro_official_full_snapshot.py @@ -0,0 +1,442 @@ +from __future__ import annotations + +import json +import os +import subprocess +import sys +import types +from pathlib import Path + + +ROOT = Path(__file__).resolve().parents[1] + + +def _run_adapter(*args: str, env: dict[str, str] | None = None) -> dict: + completed = subprocess.run( + [sys.executable, "scripts/vedastro_service_adapter.py", *args], + cwd=ROOT, + text=True, + capture_output=True, + timeout=120, + check=False, + env=env, + ) + assert completed.returncode == 0, completed.stderr or completed.stdout + return json.loads(completed.stdout) + + +def test_schema_declares_official_full_snapshot_contract() -> None: + report = _run_adapter("--print-schema") + + assert "official_full_snapshot_request_contract" in report + assert "BirthTime" in report["official_full_snapshot_request_contract"]["common_body_fields"] + assert report["official_full_snapshot_request_contract"]["primary_source"] == "vedastro_official" + assert "official_full_snapshot_response_contract" in report + assert "snapshot_sections" in report["official_full_snapshot_response_contract"] + assert report["vedastro_calculation_coverage"]["official_api_builder_calculators"] == "600+" + + +def test_official_full_snapshot_unconfigured_builds_full_request_manifest() -> None: + env = os.environ.copy() + env.pop("VEDASTRO_API_ENDPOINT", None) + env.pop("VEDASTRO_ENABLE_NETWORK", None) + env["JYOTISH_SKIP_LOCAL_ENV"] = "1" + + report = _run_adapter("--official-full-snapshot", "--case", "user_REDACTED_YEAR_test", env=env) + + assert report["backend"] == "vedastro_service_adapter_candidate" + assert report["operation"] == "official_full_snapshot" + assert report["primary_source"] == "vedastro_official" + assert report["status"] == "service_endpoint_not_configured" + assert report["available"] is False + assert report["snapshot_sections"] == {} + assert report["request_manifest"]["source_role"] == "primary_official_raw_evidence" + section_names = {item["section"] for item in report["request_manifest"]["requests"]} + assert {"chart_core", "house_core", "dasha_all", "events_overview"}.issubset(section_names) + backlog_names = { + item["section"] + for item in report["request_manifest"]["method_catalog"]["backlog_sections"] + } + assert {"varga_all", "shadbala", "ashtakavarga"}.issubset(backlog_names) + assert report["user_visibility"] == "backend_raw_evidence_not_direct_user_report" + assert report["source_metadata"]["provenance_mode"] == "vedastro_official_primary_candidate" + + +def test_official_full_snapshot_preview_when_network_disabled() -> None: + env = os.environ.copy() + env["VEDASTRO_API_ENDPOINT"] = "https://example.invalid/api" + env.pop("VEDASTRO_ENABLE_NETWORK", None) + env["JYOTISH_SKIP_LOCAL_ENV"] = "1" + + report = _run_adapter("--official-full-snapshot", "--case", "beijing_first_use_demo", env=env) + + assert report["status"] == "network_execution_disabled" + assert report["request_manifest"]["requests"] + assert report["source_metadata"]["endpoint"] == "https://example.invalid/api" + assert report["source_metadata"]["provenance_mode"] == "vedastro_official_primary_candidate" + + +def test_official_full_snapshot_dasha_request_uses_official_range_contract(monkeypatch) -> None: + from scripts import vedastro_service_adapter as adapter + + monkeypatch.delenv("VEDASTRO_API_ENDPOINT", raising=False) + monkeypatch.delenv("VEDASTRO_ENABLE_NETWORK", raising=False) + + report = adapter.run_official_full_snapshot_for_case( + { + "year": REDACTED_YEAR, + "month": 4, + "day": 17, + "hour": 14, + "minute": 49, + "lat": 36.42, + "lon": 114.2, + "tz": 8, + "reference_date": "2026-06-29", + }, + case_id="unit_dasha_contract", + ) + + dasha_request = next( + item for item in report["request_manifest"]["requests"] if item["section"] == "dasha_all" + ) + body = dasha_request["body"] + + assert report["request_manifest"]["reference_date"] == "2026-06-29" + assert report["source_metadata"]["reference_date"] == "2026-06-29" + assert dasha_request["calculator_name"] == "DasaAtRange" + assert body["birthTime"]["StdTime"] == "REDACTED_TIME 17/04/REDACTED_YEAR +08:00" + assert body["startTime"]["StdTime"] == "00:00 01/01/2026 +08:00" + assert body["endTime"]["StdTime"] == "23:59 31/12/2026 +08:00" + assert body["levels"] == 3 + assert body["precisionHours"] == 100 + assert body["Ayanamsa"] == "lahiri" + assert "StartTime" not in body + assert "EndTime" not in body + assert "time" not in body + + +def test_official_full_snapshot_marks_semantic_rate_limit_payloads(monkeypatch) -> None: + from scripts import vedastro_service_adapter as adapter + + calls = [] + + def fake_post(endpoint, request_item): + calls.append((request_item["section"], request_item.get("fanout_value"))) + if request_item["section"] == "chart_core" and request_item.get("fanout_value") == "Mars": + return { + "Status": "Fail", + "Payload": "Free tier rate limit exceeded (5 calls/minute).", + }, 1, [] + return { + "Status": "Pass", + "Payload": { + "AllPlanetData": { + "PlanetRasiD1Sign": {"Name": "Aries", "DegreesIn": {"TotalDegrees": "3.5"}}, + "PlanetNirayanaLongitude": {"TotalDegrees": "3.5"}, + }, + "AllHouseData": { + "HouseRasiD1Sign": {"Name": "Leo", "DegreesIn": {"TotalDegrees": "13.0"}}, + }, + "DasaAtRange": {}, + "SearchEvents": [], + }, + }, 1, [] + + monkeypatch.setenv("VEDASTRO_API_ENDPOINT", "https://example.invalid/api") + monkeypatch.setenv("VEDASTRO_ENABLE_NETWORK", "1") + monkeypatch.setattr(adapter, "_post_official_snapshot_section", fake_post) + + result = adapter.run_official_full_snapshot_for_case( + { + "year": REDACTED_YEAR, + "month": 4, + "day": 17, + "hour": 14, + "minute": 49, + "lat": 36.42, + "lon": 114.2, + "tz": 8, + "reference_date": "2026-06-29", + }, + case_id="unit_rate_limit", + ) + + assert ("chart_core", "Mars") in calls + assert result["section_statuses"]["chart_core"] == "rate_limited" + assert result["section_statuses"]["chart_core_fanout"]["Mars"] == "rate_limited" + assert result["source_metadata"]["rate_limited_sections"] == ["chart_core"] + assert result["source_metadata"]["production_hint"] == "configure_vedastro_api_key_or_self_host_official_api" + + +def test_orchestrator_attaches_official_full_snapshot_before_range_scan(monkeypatch) -> None: + from scripts import vedastro_evidence_orchestrator as orchestrator + + calls: list[str] = [] + + def fake_snapshot(birth_payload, *, case_id="user_chart"): + calls.append("snapshot") + assert birth_payload["reference_date"] == "2026-06-29" + return { + "backend": "vedastro_service_adapter_candidate", + "operation": "official_full_snapshot", + "primary_source": "vedastro_official", + "available": False, + "status": "network_execution_disabled", + "snapshot_sections": {}, + "source_metadata": {"provenance_mode": "vedastro_official_primary_candidate"}, + } + + def fake_scan(case, domain, start_date, end_date, case_id): + calls.append(f"range:{domain}") + return { + "backend": "vedastro_service_adapter_candidate", + "available": False, + "status": "network_execution_disabled", + "operation": "range_scan", + "domain": domain, + "event_count": 0, + "evidence_ledger": [], + } + + monkeypatch.setattr(orchestrator, "run_official_full_snapshot_for_case", fake_snapshot) + monkeypatch.setattr(orchestrator, "run_range_scan_for_case", fake_scan) + + result = orchestrator.orchestrate_vedastro_evidence( + { + "year": REDACTED_YEAR, + "month": 4, + "day": 17, + "hour": 14, + "minute": 49, + "lat": 36.42, + "lon": 114.2, + "tz": 8.0, + }, + route="overview", + reference_date="2026-06-29", + ) + + assert calls[0] == "snapshot" + assert result["official_full_snapshot"]["primary_source"] == "vedastro_official" + assert result["source_metadata"]["node_coverage"]["official_full_snapshot_first"] is True + + +def test_shared_priority_promotes_official_chart_and_keeps_local_as_supplemental() -> None: + from scripts.vedastro_priority import apply_vedastro_source_priority + + local_chart = { + "source": "local_engine", + "planets": {"Sun": {"sign": "Cancer"}}, + "ascendant": {"sign": "Virgo"}, + "houses": {"House1": {"sign": "Virgo"}}, + "birth_info": {"date": "REDACTED_DATE"}, + } + report = {"chart": local_chart, "modules": {"chart": local_chart}} + official_snapshot = { + "status": "partial", + "available": True, + "operation": "official_full_snapshot", + "primary_source": "vedastro_official", + "official_chart": { + "source": "vedastro_official", + "primary_source": "vedastro_official", + "planets": {"Sun": {"source": "vedastro_official", "sign": "Aries"}}, + "ascendant": {"source": "vedastro_official", "sign": "Leo"}, + "houses": {"House1": {"source": "vedastro_official", "sign": "Leo"}}, + "coverage": {"planet_count": 1, "house_count": 1}, + }, + } + + apply_vedastro_source_priority(report, official_snapshot=official_snapshot) + + assert report["chart"]["source"] == "vedastro_official_primary" + assert report["chart"]["primary_source"] == "vedastro_official" + assert report["chart"]["planets"]["Sun"]["sign"] == "Aries" + assert report["chart"]["source_priority"][0] == "vedastro_official_snapshot" + assert report["chart"]["local_engine_role"] == "supplemental_crosscheck_or_fallback" + assert report["modules"]["local_engine_chart_fallback"]["planets"]["Sun"]["sign"] == "Cancer" + assert report["modules"]["source_priority"]["mode"] == "vedastro_official_primary" + assert report["modules"]["source_priority"]["local_engine_role"] == "supplemental_crosscheck_or_fallback" + + +def test_shared_priority_marks_local_fallback_only_when_official_blocked() -> None: + from scripts.vedastro_priority import apply_vedastro_source_priority + + local_chart = { + "source": "local_engine", + "planets": {"Sun": {"sign": "Cancer"}}, + "ascendant": {"sign": "Virgo"}, + "houses": {"House1": {"sign": "Virgo"}}, + } + report = {"chart": local_chart, "modules": {"chart": local_chart}} + official_snapshot = { + "status": "network_execution_disabled", + "available": False, + "operation": "official_full_snapshot", + "primary_source": "vedastro_official", + "reason": "network disabled", + } + + apply_vedastro_source_priority(report, official_snapshot=official_snapshot) + + assert report["chart"]["source"] == "local_engine_fallback" + assert report["chart"]["primary_source"] == "local_engine" + assert report["chart"]["fallback_reason"] == "VedAstro official snapshot blocked: network_execution_disabled" + assert report["modules"]["source_priority"]["mode"] == "local_fallback_official_blocked" + assert report["modules"]["source_priority"]["official_snapshot_status"] == "network_execution_disabled" + + +def test_full_reading_prompt_pack_exposes_vedastro_official_snapshot_boundary() -> None: + result = subprocess.run( + [ + sys.executable, + "scripts/jyotish_engine.py", + "full-reading", + "--year", + "REDACTED_YEAR", + "--month", + "4", + "--day", + "17", + "--hour", + "14", + "--minute", + "49", + "--lat", + "36.42", + "--lon", + "114.2", + "--tz", + "8", + "--today", + "2026-06-29", + "--transit-date", + "2026-06-29", + ], + cwd=ROOT, + text=True, + capture_output=True, + timeout=180, + check=False, + ) + assert result.returncode == 0, result.stderr or result.stdout + report = json.loads(result.stdout) + + snapshot = report["modules"]["vedastro_official_full_snapshot"] + assert snapshot["operation"] == "official_full_snapshot" + assert snapshot["primary_source"] == "vedastro_official" + prompt_snapshot = report["ai_prompt_pack"]["evidence_snapshot"]["vedastro_official_full_snapshot"] + assert prompt_snapshot["primary_source"] == "vedastro_official" + assert prompt_snapshot["status"] in {"ok", "partial", "service_endpoint_not_configured", "network_execution_disabled", "blocked"} + + +def test_full_reading_official_snapshot_uses_requested_reference_date(monkeypatch) -> None: + from scripts import jyotish_engine + + captured: dict[str, str] = {} + + def fake_snapshot(case, *, case_id="user_chart"): + captured["reference_date"] = case.get("reference_date") + return { + "status": "network_execution_disabled", + "available": False, + "operation": "official_full_snapshot", + "primary_source": "vedastro_official", + "reason": "stub", + "snapshot_sections": {}, + "source_metadata": {}, + } + + fake_adapter = types.SimpleNamespace(run_official_full_snapshot_for_case=fake_snapshot) + fake_priority = types.SimpleNamespace(apply_vedastro_source_priority=lambda report, official_snapshot: report) + monkeypatch.setitem(sys.modules, "vedastro_service_adapter", fake_adapter) + monkeypatch.setitem(sys.modules, "vedastro_priority", fake_priority) + + class Args: + year = REDACTED_YEAR + month = 4 + day = 17 + hour = 14 + minute = 49 + lat = 36.42 + lon = 114.2 + tz = 8 + ayanamsa = "lahiri" + node_mode = "mean" + today = "2026-06-09" + transit_date = "2026-06-09" + + report = {"modules": {}, "warnings": []} + jyotish_engine._attach_vedastro_official_full_snapshot(report, Args()) + + assert captured["reference_date"] == "2026-06-09" + assert report["modules"]["vedastro_official_full_snapshot"]["operation"] == "official_full_snapshot" + + +def test_official_full_snapshot_extracts_official_chart_and_varga_from_pass_payload(monkeypatch) -> None: + from scripts import vedastro_service_adapter as adapter + + calls = [] + + def fake_post(endpoint, request_item): + calls.append((request_item["section"], request_item.get("fanout_value"))) + section = request_item["section"] + if section == "chart_core": + planet = request_item["fanout_value"] + return { + "Status": "Pass", + "Payload": { + "AllPlanetData": { + "PlanetRasiD1Sign": {"Name": "Aries", "DegreesIn": {"TotalDegrees": "3.5"}}, + "PlanetHoraD2Signs": {"Name": "Leo", "DegreesIn": {"TotalDegrees": "7.0"}}, + "PlanetNavamshaD9Sign": {"Name": "Taurus", "DegreesIn": {"TotalDegrees": "1.5"}}, + "PlanetDashamamshaD10Sign": {"Name": "Taurus", "DegreesIn": {"TotalDegrees": "5.0"}}, + "PlanetNirayanaLongitude": {"TotalDegrees": "3.5"}, + "HousePlanetOccupiesBasedOnSign": "House9", + } + }, + }, 1, [] + if section == "house_core": + return { + "Status": "Pass", + "Payload": { + "AllHouseData": { + "HouseRasiD1Sign": {"Name": "Leo", "DegreesIn": {"TotalDegrees": "13.0"}}, + "HouseHoraD2Sign": {"Name": "Leo", "DegreesIn": {"TotalDegrees": "26.0"}}, + "HouseNavamshaD9Sign": {"Name": "Cancer", "DegreesIn": {"TotalDegrees": "27.7"}}, + "HouseDashamamshaD10Sign": {"Name": "Sagittarius", "DegreesIn": {"TotalDegrees": "10.8"}}, + } + }, + }, 1, [] + return {"Status": "Fail", "Payload": "not mapped"}, 1, [] + + monkeypatch.setenv("VEDASTRO_API_ENDPOINT", "https://example.invalid/api") + monkeypatch.setenv("VEDASTRO_ENABLE_NETWORK", "1") + monkeypatch.setattr(adapter, "_post_official_snapshot_section", fake_post) + + result = adapter.run_official_full_snapshot_for_case( + { + "year": REDACTED_YEAR, + "month": 4, + "day": 17, + "hour": 14, + "minute": 49, + "lat": 36.42, + "lon": 114.2, + "tz": 8, + }, + case_id="unit", + ) + + assert ("chart_core", "Sun") in calls + assert ("house_core", "House1") in calls + assert result["status"] == "partial" + assert result["section_statuses"]["events_overview"] == "fail" + official_chart = result["official_chart"] + assert official_chart["source"] == "vedastro_official" + assert official_chart["planets"]["Sun"]["sign"] == "Aries" + assert official_chart["planets"]["Sun"]["source"] == "vedastro_official" + assert official_chart["planets"]["Sun"]["vargas"]["D9"]["sign"] == "Taurus" + assert official_chart["ascendant"]["sign"] == "Leo" + assert official_chart["ascendant"]["vargas"]["D10"]["sign"] == "Sagittarius"