Jesse_Chen
f224146348
Three-way-merge calculation modules and pl9-export into the product fork while keeping commercial API routes, Raman ayanamsa, and the consultation contract as a keypath superset. Co-authored-by: Cursor <cursoragent@cursor.com>
255 lines
9.9 KiB
Python
255 lines
9.9 KiB
Python
"""Shared report-pack contract normalizer.
|
||
|
||
This module intentionally stays thin: it does not compute astrology results and
|
||
does not adjudicate pack truth. It converts existing pack-shaped dictionaries
|
||
into one stable envelope that the final PL9-style renderer can consume.
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
from collections.abc import Mapping
|
||
from typing import Any
|
||
|
||
try:
|
||
from canonical_jyotish_profile import (
|
||
build_canonical_jyotish_profile,
|
||
build_disputed_method_policy,
|
||
build_reader_engine_boundary_notice,
|
||
)
|
||
except Exception: # pragma: no cover - research helpers are not vendored here
|
||
try:
|
||
from scripts.canonical_jyotish_profile import (
|
||
build_canonical_jyotish_profile,
|
||
build_disputed_method_policy,
|
||
build_reader_engine_boundary_notice,
|
||
)
|
||
except Exception:
|
||
def build_canonical_jyotish_profile():
|
||
return {"profile_id": "product_canonical", "status": "local_fallback"}
|
||
|
||
def build_disputed_method_policy():
|
||
return {}
|
||
|
||
def build_reader_engine_boundary_notice():
|
||
return {"status": "local_fallback"}
|
||
try:
|
||
from profile_aware_benchmark_boundary_dashboard import load_dashboard_payload
|
||
except Exception: # pragma: no cover - research helpers are not vendored here
|
||
try:
|
||
from scripts.profile_aware_benchmark_boundary_dashboard import load_dashboard_payload
|
||
except Exception:
|
||
def load_dashboard_payload():
|
||
return {"status": "blocked", "reason": "profile_aware_dashboard_absent"}
|
||
|
||
|
||
SCHEMA = "pl9.unified_report_pack_contract.v1"
|
||
|
||
|
||
def normalize_report_pack_contract(pack: Mapping[str, Any], *, pack_id: str) -> dict[str, Any]:
|
||
"""Return a uniform report-ready envelope for an existing pack."""
|
||
|
||
report_sections = _normalize_report_sections(pack.get("report_sections"))
|
||
audit = _normalize_audit(pack)
|
||
pl9_pages = _as_list(pack.get("pl9_pages") or audit.get("pl9_pages"))
|
||
blocked_reasons = _blocked_reasons(pack, report_sections)
|
||
|
||
contract = {
|
||
"schema": SCHEMA,
|
||
"pack_id": pack_id,
|
||
"source_schema": pack.get("schema"),
|
||
"status": _status(pack, audit),
|
||
"profile": dict(pack.get("profile") or {}),
|
||
"pl9_pages": pl9_pages,
|
||
"summary": dict(pack.get("summary") or {}),
|
||
"canonical_jyotish_profile": dict(pack.get("canonical_jyotish_profile") or build_canonical_jyotish_profile()),
|
||
"reader_engine_boundary_notice": dict(
|
||
pack.get("reader_engine_boundary_notice") or build_reader_engine_boundary_notice()
|
||
),
|
||
"multi_engine_difference_notice": _multi_engine_difference_notice(pack),
|
||
"profile_aware_dashboard": dict(pack.get("profile_aware_dashboard") or load_dashboard_payload()),
|
||
"raw_data": _raw_data(pack),
|
||
"normalized_data": _normalized_data(pack),
|
||
"report_sections": report_sections,
|
||
"exports": _normalize_exports(pack.get("exports")),
|
||
"audit": audit,
|
||
"blocked_reasons": blocked_reasons,
|
||
"contract_audit": {
|
||
"missing_fields": _missing_fields(report_sections),
|
||
"source_pack_keys": list(pack.keys()),
|
||
"normalization_boundary": "contract_only_no_astrological_recalculation",
|
||
},
|
||
}
|
||
return contract
|
||
|
||
|
||
def _status(pack: Mapping[str, Any], audit: Mapping[str, Any]) -> str:
|
||
explicit = pack.get("status") or audit.get("status")
|
||
if explicit:
|
||
return str(explicit)
|
||
|
||
statuses: list[str] = []
|
||
report_sections = pack.get("report_sections") if isinstance(pack.get("report_sections"), Mapping) else {}
|
||
for key in ("executive_summary", "thematic_narrative", "evidence_appendix", "pdf_sections"):
|
||
section = report_sections.get(key)
|
||
if isinstance(section, Mapping) and section.get("status"):
|
||
statuses.append(str(section.get("status")))
|
||
for key, value in pack.items():
|
||
if key in {"schema", "profile", "pl9_pages", "summary", "report_sections", "exports", "audit"}:
|
||
continue
|
||
if isinstance(value, Mapping) and value.get("status"):
|
||
statuses.append(str(value.get("status")))
|
||
|
||
if not statuses:
|
||
return "blocked"
|
||
if any(status in {"conflict", "parameter_sensitive"} for status in statuses):
|
||
return "parameter_sensitive"
|
||
if any(status in {"partial_verified", "verified"} for status in statuses):
|
||
return "partial_verified"
|
||
if any(status == "not_applicable" for status in statuses):
|
||
return "not_applicable"
|
||
return "blocked"
|
||
|
||
|
||
def _normalize_report_sections(value: Any) -> dict[str, Any]:
|
||
sections = dict(value or {}) if isinstance(value, Mapping) else {}
|
||
executive = sections.get("executive_summary")
|
||
thematic = sections.get("thematic_narrative")
|
||
evidence = sections.get("evidence_appendix")
|
||
pdf_sections = sections.get("pdf_sections")
|
||
|
||
if isinstance(executive, Mapping):
|
||
executive = [str(executive.get("headline") or executive.get("status") or "")]
|
||
elif isinstance(executive, str):
|
||
executive = [executive]
|
||
elif executive is None:
|
||
executive = []
|
||
else:
|
||
executive = list(executive)
|
||
|
||
if thematic is None:
|
||
thematic = sections.get("trigger_seed_narrative") or []
|
||
if evidence is None:
|
||
evidence = []
|
||
for key in ("visual_chart_audit", "audit_appendix"):
|
||
if key in sections:
|
||
evidence.append(sections[key])
|
||
|
||
return {
|
||
"executive_summary": executive,
|
||
"thematic_narrative": list(thematic) if isinstance(thematic, list) else _as_list(thematic),
|
||
"evidence_appendix": list(evidence) if isinstance(evidence, list) else _as_list(evidence),
|
||
"pdf_sections": list(pdf_sections) if isinstance(pdf_sections, list) else _as_list(pdf_sections),
|
||
}
|
||
|
||
|
||
def _normalize_exports(value: Any) -> dict[str, Any]:
|
||
exports = dict(value or {}) if isinstance(value, Mapping) else {}
|
||
return {
|
||
"json": exports.get("json"),
|
||
"markdown": exports.get("markdown"),
|
||
"pdf_sections": exports.get("pdf_sections"),
|
||
"ai_evidence_bundle": dict(exports.get("ai_evidence_bundle") or {}),
|
||
}
|
||
|
||
|
||
def _multi_engine_difference_notice(pack: Mapping[str, Any]) -> dict[str, Any]:
|
||
if isinstance(pack.get("multi_engine_difference_notice"), Mapping):
|
||
return dict(pack["multi_engine_difference_notice"])
|
||
policy = build_disputed_method_policy()
|
||
lanes = []
|
||
for lane_id, lane in policy.items():
|
||
lanes.append(
|
||
{
|
||
"lane_id": lane_id,
|
||
"canonical_standard": lane["canonical_standard"],
|
||
"external_difference_display": "external_observation_conflict",
|
||
"report_assertion_ceiling": lane["report_assertion_ceiling"],
|
||
"reader_text_zh": (
|
||
f"{lane_id} 主口径按 {lane['canonical_standard']}；"
|
||
"其他引擎若不同，会显示为外部观察冲突，不直接改写主结论。"
|
||
),
|
||
}
|
||
)
|
||
return {
|
||
"status": "active",
|
||
"purpose": "reader_safe_multi_engine_difference_display",
|
||
"canonical_profile_id": build_canonical_jyotish_profile()["profile_id"],
|
||
"lanes": lanes,
|
||
"must_not_claim": [
|
||
"external_engine_conflict_overrides_canonical_result",
|
||
"internal_consistency_is_global_truth_closure",
|
||
],
|
||
}
|
||
|
||
|
||
def _normalize_audit(pack: Mapping[str, Any]) -> dict[str, Any]:
|
||
audit = dict(pack.get("audit") or {})
|
||
must_not_claim = list(audit.get("must_not_claim") or [])
|
||
sections = pack.get("report_sections") if isinstance(pack.get("report_sections"), Mapping) else {}
|
||
for key in ("visual_chart_audit", "evidence_audit"):
|
||
section = sections.get(key) if isinstance(sections, Mapping) else None
|
||
if isinstance(section, Mapping):
|
||
must_not_claim.extend(item for item in section.get("must_not_claim") or [] if item not in must_not_claim)
|
||
if must_not_claim:
|
||
audit["must_not_claim"] = must_not_claim
|
||
return audit
|
||
|
||
|
||
def _blocked_reasons(pack: Mapping[str, Any], report_sections: Mapping[str, Any]) -> list[str]:
|
||
reasons: list[str] = []
|
||
for key in ("blocked_reasons", "blocked_fields"):
|
||
reasons.extend(str(item) for item in pack.get(key) or [])
|
||
original_sections = pack.get("report_sections") if isinstance(pack.get("report_sections"), Mapping) else {}
|
||
executive = original_sections.get("executive_summary") if isinstance(original_sections, Mapping) else None
|
||
if isinstance(executive, Mapping):
|
||
reasons.extend(str(item) for item in executive.get("blocked_fields") or [])
|
||
if not reasons and _status(pack, pack.get("audit") or {}) == "blocked":
|
||
for item in report_sections.get("evidence_appendix") or []:
|
||
if isinstance(item, str):
|
||
reasons.append(item)
|
||
return list(dict.fromkeys(reasons))
|
||
|
||
|
||
def _raw_data(pack: Mapping[str, Any]) -> dict[str, Any]:
|
||
return {
|
||
key: value
|
||
for key, value in pack.items()
|
||
if key
|
||
not in {
|
||
"schema",
|
||
"status",
|
||
"profile",
|
||
"pl9_pages",
|
||
"summary",
|
||
"report_sections",
|
||
"exports",
|
||
"audit",
|
||
"profile_aware_dashboard",
|
||
}
|
||
}
|
||
|
||
|
||
def _normalized_data(pack: Mapping[str, Any]) -> dict[str, Any]:
|
||
return {
|
||
"source_schema": pack.get("schema"),
|
||
"status": pack.get("status") or (pack.get("audit") or {}).get("status") if isinstance(pack.get("audit"), Mapping) else pack.get("status"),
|
||
}
|
||
|
||
|
||
def _missing_fields(report_sections: Mapping[str, Any]) -> list[str]:
|
||
missing = []
|
||
for key in ("executive_summary", "thematic_narrative", "evidence_appendix"):
|
||
if not report_sections.get(key):
|
||
missing.append(key)
|
||
return missing
|
||
|
||
|
||
def _as_list(value: Any) -> list[Any]:
|
||
if value is None:
|
||
return []
|
||
if isinstance(value, list):
|
||
return value
|
||
if isinstance(value, tuple):
|
||
return list(value)
|
||
return [value]
|