root/Jyotisha
1
0
Fork 0
Code Issues 25 Pull Requests 1 Actions 1 Packages Projects Releases Wiki Activity

Wire VedAstro official snapshot priority

Browse Source
This commit is contained in:
732642856
2026-06-29 15:40:34 +08:00
parent 1c08af34fa
commit e2595bec13
10 changed files with 1445 additions and 12 deletions
Download Patch File Download Diff File Expand all files Collapse all files
SKILL.md
+1
View File
@@ -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+案例检索）
jyotish-app/main.js
+1 -1
View File
@@ -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 || '-')}
</div>
<div class="calculation-settings-note">
${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)}` : ''}
</div>
${events.length ? `
mcp_server.py
+47 -6
View File
@@ -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
scripts/jyotish_api_server.py
+52 -1
View File
@@ -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,
scripts/jyotish_engine.py
+94
View File
@@ -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:
scripts/vedastro_evidence_orchestrator.py
+14 -2
View File
@@ -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,
scripts/vedastro_priority.py
+167
View File
@@ -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"),
}
scripts/vedastro_service_adapter.py
+588 -2
View File
@@ -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:
tests/test_vedastro_external_technique_evidence.py
+39
View File
@@ -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"])}
tests/test_vedastro_official_full_snapshot.py
+442
View File
@@ -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"