Files
Jyotisha/scripts/full_report_quality_gate.py
T
Jesse_Chen bab0718700
Independent Staging Quality Gate / validate (push) Failing after 9m41s
Independent Staging Quality Gate / publish (push) Has been skipped
feat(report): ship full-mode longform appendix beside the five-chapter report
Web export now calls the same full pack as the long skill report and caches an owner-only Markdown download. Appendix failure stays unavailable and does not change the main report status.

Co-authored-by: Cursor <cursoragent@cursor.com>
2026-09-05 11:10:44 +08:00

508 lines
23 KiB
Python
Executable File
Raw Blame History

This file contains ambiguous Unicode characters
This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.
#!/usr/bin/env python3
"""Read-only quality checks for PL9-grade full personal report packets."""
from __future__ import annotations
import argparse
import json
import re
from pathlib import Path
from typing import Any
SCHEMA_VERSION = "jyotish.full_report_quality_gate.v1"
REPORT_SCHEMA = "pl9_style_professional_export_v1"
REQUIRED_PACK_SECTIONS = (
"base",
"strength",
"dasha",
"annual",
"transit",
"d1_d60_ledger",
"professional_support",
"audit_appendix",
)
RESTRICTED_MATERIAL_IDS = (
"jaimini_special_points",
"tajika_named_yoga",
"annual_sahams",
"kranti",
"alternate_ashtottari",
)
def _as_dict(value: Any) -> dict[str, Any]:
return value if isinstance(value, dict) else {}
def _present(value: Any) -> bool:
return bool(value) if isinstance(value, (dict, list, tuple, set, str)) else value is not None
def _status(value: Any) -> str:
item = _as_dict(value)
return str(item.get("status") or item.get("execution_status") or "available")
def _marker_present(markdown: str | None, markers: tuple[str, ...]) -> bool | None:
if markdown is None:
return None
return any(marker in markdown for marker in markers)
def _check(name: str, status: str, detail: str) -> dict[str, str]:
return {"name": name, "status": status, "detail": detail}
def _coverage_entry(
material_id: str,
tier: str,
source_reference: str,
upstream_value: Any,
markdown: str | None,
markers: tuple[str, ...],
) -> tuple[dict[str, Any], str | None]:
upstream_present = _present(upstream_value)
rendered = _marker_present(markdown, markers)
status = _status(upstream_value) if upstream_present else "not_available"
entry = {
"material_id": material_id,
"admission_tier": tier,
"report_role": "professional_support",
"source_reference": source_reference,
"upstream_present": upstream_present,
"status": status,
"source_status": status,
"surface_location": (
"professional_support_cross_reference"
if material_id in {
"special_lagnas",
"patyayini_annual_support",
"narayana_alignment",
}
else "thematic_or_operator_appendix"
),
"rendered": rendered,
"limitation_reference": None if status not in {"blocked", "partial", "parameter_sensitive", "conflict"} else status,
}
if upstream_present and rendered is False:
return entry, f"required_support_not_rendered:{material_id}"
return entry, None
def _patyayini_normalized_rows_present(packet: dict[str, Any]) -> bool | None:
annual_section = _as_dict(_as_dict(_as_dict(packet.get("full_report_pack")).get("sections")).get("annual"))
annual_data = _as_dict(packet.get("worksheets"))
timing = _as_dict(annual_data.get("timing_and_predictive_systems"))
annual_pack = _as_dict(timing.get("annual_tajika_pack"))
candidates = (
_as_dict(_as_dict(annual_pack.get("external_engine_comparison")).get("pyjhora")).get("patyayini_dasha"),
_as_dict(_as_dict(annual_section.get("external_engine_comparison")).get("pyjhora")).get("patyayini_dasha"),
)
saw_patyayini = False
for candidate in candidates:
if candidate:
saw_patyayini = True
rows = candidate.get("normalized_rows")
if isinstance(rows, list) and rows:
return True
if saw_patyayini:
return False
return None
def _promotion_status(blocking_reasons: list[str], review_reasons: list[str], warning_reasons: list[str]) -> str:
if blocking_reasons:
return "blocked"
if review_reasons:
return "review_required"
if warning_reasons:
return "passed_with_limitations"
return "passed"
def evaluate_full_report(packet: dict[str, Any], rendered_markdown: str | None = None) -> dict[str, Any]:
"""Evaluate an assembled report without mutating calculation or report content."""
packet = _as_dict(packet)
checks: list[dict[str, str]] = []
blocking_reasons: list[str] = []
review_reasons: list[str] = []
warning_reasons: list[str] = []
source_schema = str(packet.get("schema") or "")
if source_schema != REPORT_SCHEMA:
blocking_reasons.append("report_schema_invalid")
checks.append(_check("report_schema", "blocked", f"expected {REPORT_SCHEMA}, got {source_schema or 'missing'}"))
else:
checks.append(_check("report_schema", "passed", source_schema))
profile = _as_dict(packet.get("calculation_profile"))
for key, value in (
("calculation_profile_id", packet.get("calculation_profile_id") or profile.get("profile_id")),
("result_hash", packet.get("result_hash")),
("ayanamsa", profile.get("ayanamsa") or _as_dict(profile.get("effective_settings")).get("ayanamsa")),
("node_mode", profile.get("node_mode") or _as_dict(profile.get("effective_settings")).get("node_mode")),
):
if value in (None, ""):
blocking_reasons.append(f"provenance_missing:{key}")
checks.append(_check(f"provenance:{key}", "blocked", "missing"))
else:
checks.append(_check(f"provenance:{key}", "passed", "present"))
for key, value in (
("input_hash", profile.get("input_hash")),
("calculation_source", profile.get("algorithm")),
("profile_version", profile.get("profile_version")),
("engine_metadata", profile.get("engine")),
):
if not _present(value):
blocking_reasons.append(f"provenance_missing:{key}")
checks.append(_check(f"provenance:{key}", "blocked", "missing"))
else:
checks.append(_check(f"provenance:{key}", "passed", "present"))
binding = _as_dict(packet.get("result_binding"))
expected_input_hash = profile.get("input_hash")
expected_result_hash = packet.get("result_hash")
if binding.get("input_hash") != expected_input_hash or binding.get("result_hash") != expected_result_hash:
blocking_reasons.append("provenance_lineage_binding_invalid")
checks.append(_check("provenance:result_binding", "blocked", "does not match profile/result hashes"))
else:
checks.append(_check("provenance:result_binding", "passed", "input/result hashes match"))
chart_identity = _as_dict(packet.get("chart_identity"))
for key in (
"chart_profile_id",
"birth_data_status",
"rectification_status",
"approval_status",
):
value = chart_identity.get(key)
if value in (None, ""):
blocking_reasons.append(f"chart_identity_missing:{key}")
checks.append(_check(f"chart_identity:{key}", "blocked", "missing"))
else:
checks.append(_check(f"chart_identity:{key}", "passed", str(value)))
full_report_pack = _as_dict(packet.get("full_report_pack"))
sections = _as_dict(full_report_pack.get("sections"))
if not full_report_pack:
blocking_reasons.append("full_report_pack_missing")
checks.append(_check("full_report_pack", "blocked", "missing"))
else:
checks.append(_check("full_report_pack", "passed", str(full_report_pack.get("schema") or "present")))
for key in REQUIRED_PACK_SECTIONS:
section = _as_dict(sections.get(key))
if not section:
blocking_reasons.append(f"required_pack_section_missing:{key}")
checks.append(_check(f"pack_section:{key}", "blocked", "missing"))
else:
section_status = _status(section)
checks.append(_check(f"pack_section:{key}", "passed" if section_status in {"verified", "available"} else "warning", section_status))
if section_status in {"blocked", "partial", "partial_verified", "parameter_sensitive", "conflict"}:
warning_reasons.append(f"pack_section_limited:{key}:{section_status}")
d1_d60_ledger_section = _as_dict(sections.get("d1_d60_ledger"))
d1_d60_ledger = _as_dict(d1_d60_ledger_section.get("d1_to_d60"))
d1_d60_summary = _as_dict(d1_d60_ledger_section.get("summary"))
if d1_d60_ledger_section:
if len(d1_d60_ledger) != 60:
blocking_reasons.append("d1_d60_ledger_incomplete")
checks.append(_check("d1_d60_ledger:row_count", "blocked", f"expected 60, got {len(d1_d60_ledger)}"))
else:
checks.append(_check("d1_d60_ledger:row_count", "passed", "60 rows"))
expected_summary = {
"formal_traditional_division_count": 20,
"research_generic_dn_division_count": 40,
}
if {key: d1_d60_summary.get(key) for key in expected_summary} != expected_summary:
review_reasons.append("d1_d60_ledger_classification_invalid")
checks.append(_check("d1_d60_ledger:classification", "review_required", str(d1_d60_summary)))
else:
checks.append(_check("d1_d60_ledger:classification", "passed", "20 formal + 40 research"))
audit_appendix = _as_dict(sections.get("audit_appendix"))
if not audit_appendix:
blocking_reasons.append("audit_appendix_missing")
elif _status(audit_appendix) in {"blocked", "partial", "partial_verified", "parameter_sensitive", "conflict"}:
warning_reasons.append(f"audit_appendix_limited:{_status(audit_appendix)}")
markdown_checks = {
"chart_identity": ("基础资料表", "Birth Particulars"),
"d1": ("D1 — Rashi Chart", "D1 / Base Chart"),
"varga": ("Vargas I", "Varga Analysis"),
"d1_d60_ledger": ("D1D60 完整原始分盘账本", "D1-D60 Complete Raw Divisional Ledger"),
"timing": ("大运与时间主线", "Dasha Analysis"),
"kp": ("KP 三年流月支持", "KP and Transit Timing Layer"),
"annual": ("年度重点", "Annual / Tajika / Yearly Focus"),
"professional_support": ("专业支持专题:第二证据轴",),
"restricted_materials": ("未闭环专业层:可见但不入判断",),
"limitations": ("Blocked / Audit Appendix", "Conflict and Limitation"),
"audit_appendix": ("Audit Appendix", "Operator Appendix"),
}
if rendered_markdown is None:
warning_reasons.append("rendered_markdown_not_supplied")
checks.append(_check("reader_surface", "not_evaluated", "rendered Markdown was not supplied"))
else:
for name, markers in markdown_checks.items():
if _marker_present(rendered_markdown, markers):
checks.append(_check(f"reader_section:{name}", "passed", "rendered"))
else:
review_reasons.append(f"reader_section_missing:{name}")
checks.append(_check(f"reader_section:{name}", "review_required", "not rendered"))
worksheets = _as_dict(packet.get("worksheets"))
divisional = _as_dict(worksheets.get("divisional_and_special_charts"))
strengths = _as_dict(worksheets.get("strengths_and_scores"))
timing = _as_dict(worksheets.get("timing_and_predictive_systems"))
annual = _as_dict(timing.get("annual_tajika_pack"))
dasha_master = _as_dict(timing.get("dasha_master_pack"))
dasha_families = _as_dict(dasha_master.get("families"))
narayana = timing.get("narayana_dasha") or dasha_families.get("narayana")
patyayini = annual.get("patyayini_dasha") or annual.get("patyayini")
auxiliary_dasha_support = {
key: value
for key, value in dasha_families.items()
if key in {"yogini", "ashtottari", "kala_chakra"} and _present(value)
}
advanced = _as_dict(worksheets.get("advanced_systems"))
kp_monthly_report = _as_dict(advanced.get("kp_monthly_report"))
annual_audit = _as_dict(annual.get("audit"))
annual_sections = _as_dict(annual.get("report_sections"))
if not annual_audit or not annual_sections:
review_reasons.append("annual_evidence_status_missing")
checks.append(_check("annual_evidence_status", "review_required", "annual audit or report sections missing"))
else:
checks.append(_check("annual_evidence_status", "passed", "audit and report sections present"))
if rendered_markdown is not None and "年度专题证据状态" not in rendered_markdown:
review_reasons.append("annual_evidence_status_not_rendered")
checks.append(_check("annual_evidence_status:rendered", "review_required", "not rendered"))
if not kp_monthly_report:
review_reasons.append("kp_evidence_status_missing")
checks.append(_check("kp_evidence_status", "review_required", "monthly report missing"))
else:
kp_maturity = _as_dict(kp_monthly_report.get("maturity_profile"))
kp_restrictions = kp_monthly_report.get("must_not_claim")
if not kp_maturity.get("claim_status") or not isinstance(kp_restrictions, list) or not kp_restrictions:
review_reasons.append("kp_evidence_status_missing")
checks.append(_check("kp_evidence_status", "review_required", "maturity profile or must_not_claim missing"))
else:
checks.append(_check("kp_evidence_status", "passed", str(kp_maturity.get("claim_status"))))
if rendered_markdown is not None and "KP 使用边界" not in rendered_markdown:
review_reasons.append("kp_evidence_status_not_rendered")
checks.append(_check("kp_evidence_status:rendered", "review_required", "not rendered"))
if rendered_markdown is None:
checks.append(_check("three_year_kp_monthly", "not_evaluated", "rendered Markdown was not supplied"))
else:
rendered_years = sorted(
{int(year) for year in re.findall(r"^###\s+(\d{4})\s+KP 月度支持\s*$", rendered_markdown, re.MULTILINE)}
)
consecutive = len(rendered_years) >= 3 and all(
year == rendered_years[0] + offset for offset, year in enumerate(rendered_years)
)
if not consecutive:
review_reasons.append("three_year_kp_monthly_incomplete")
checks.append(_check("three_year_kp_monthly", "review_required", str(rendered_years)))
else:
checks.append(_check("three_year_kp_monthly", "passed", f"{rendered_years[0]}-{rendered_years[-1]}"))
source_years = sorted(
{
int(row["year"])
for row in kp_monthly_report.get("yearly_highlights", [])
if isinstance(row, dict) and isinstance(row.get("year"), int)
}
)
if source_years and source_years != rendered_years:
review_reasons.append("three_year_kp_monthly_source_render_mismatch")
checks.append(_check("three_year_kp_monthly:source_alignment", "review_required", f"source={source_years}; rendered={rendered_years}"))
elif source_years:
checks.append(_check("three_year_kp_monthly:source_alignment", "passed", str(source_years)))
natal_special_factor_support = {
"sahams": _as_dict(advanced.get("sahams")),
"avasthas": _as_dict(advanced.get("avasthas")),
"upagrahas": _as_dict(divisional.get("upagrahas")),
}
natal_special_factor_support = {
key: value for key, value in natal_special_factor_support.items() if _present(value)
}
special_lagnas = _as_dict(divisional.get("special_lagnas"))
required_special_lagnas = {
"Bhava_Lagna": "Bhava Lagna",
"Hora_Lagna": "Hora Lagna",
"Ghati_Lagna": "Ghati Lagna",
"ViGhati_Lagna": "ViGhati Lagna",
"Sree_Lagna": "Sree Lagna",
"Indu_Lagna": "Indu Lagna",
}
normalized_special_lagnas = {
str(key).strip().lower().replace(" ", "_"): _as_dict(value)
for key, value in special_lagnas.items()
}
missing_special_lagnas: list[str] = []
for key, display_name in required_special_lagnas.items():
row = normalized_special_lagnas.get(key.lower(), {})
explicit_status = _status(row)
has_position = any(row.get(field) not in (None, "") for field in ("sign", "degree", "longitude"))
explicitly_blocked = explicit_status in {"blocked", "not_applicable"}
if not has_position and not explicitly_blocked:
missing_special_lagnas.append(key)
checks.append(_check(f"special_lagna:{key}", "review_required", "missing value or blocked status"))
else:
checks.append(_check(f"special_lagna:{key}", "passed", explicit_status if explicitly_blocked else "position present"))
if rendered_markdown is not None and (has_position or explicitly_blocked) and display_name not in rendered_markdown:
review_reasons.append(f"special_lagna_not_rendered:{key}")
checks.append(_check(f"special_lagna:{key}:rendered", "review_required", "not rendered"))
if len(missing_special_lagnas) >= 2:
blocking_reasons.append("special_lagnas_incomplete")
elif missing_special_lagnas:
review_reasons.extend(f"special_lagna_missing:{key}" for key in missing_special_lagnas)
wealth_vargas = _as_dict(divisional.get("varga_full"))
for key, label in (("D2_Hora", "D2"), ("D11_Rudramsa", "D11")):
chart = _as_dict(wealth_vargas.get(key))
ascendant = _as_dict(chart.get("Ascendant") or chart.get("ascendant"))
has_ascendant = any(ascendant.get(field) not in (None, "") for field in ("sign", "degree_in_sign", "longitude"))
if not has_ascendant:
blocking_reasons.append(f"wealth_varga_structure_missing:{label}")
checks.append(_check(f"wealth_varga:{label}", "blocked", "missing ascendant structure"))
else:
checks.append(_check(f"wealth_varga:{label}", "passed", "ascendant structure present"))
coverage_inputs = (
(
"wealth_d2_chart",
"required_support",
"worksheets.divisional_and_special_charts.varga_full.D2_Hora",
_as_dict(_as_dict(divisional.get("varga_full")).get("D2_Hora")),
("D2Hora 财富分盘)",),
),
(
"wealth_d11_gains_chart",
"required_support",
"worksheets.divisional_and_special_charts.varga_full.D11_Rudramsa",
_as_dict(_as_dict(divisional.get("varga_full")).get("D11_Rudramsa")),
("D11Rudramsa 收益分盘)",),
),
(
"special_lagnas",
"required_support",
"worksheets.divisional_and_special_charts.special_lagnas",
divisional.get("special_lagnas"),
("Hora Lagna", "Bhava Lagna", "Standard Lagnas", "Special Lagnas"),
),
(
"patyayini_annual_support",
"required_support",
"worksheets.timing_and_predictive_systems.annual_tajika_pack.patyayini_dasha",
patyayini,
("Patyayini Dasha", "Patyayini"),
),
(
"narayana_alignment",
"required_support",
"worksheets.timing_and_predictive_systems.narayana_dasha",
narayana,
("Narayana Rashi Dasha", "Narayana 参考对齐", "Narayana"),
),
(
"functional_benefic_malefic",
"required_support",
"worksheets.strengths_and_scores.functional_benefic_malefic",
strengths.get("functional_benefic_malefic"),
("功能性吉凶", "Functional Benefic/Malefic"),
),
(
"auxiliary_dasha_support",
"required_support",
"worksheets.timing_and_predictive_systems.dasha_master_pack.families",
auxiliary_dasha_support,
("辅助大运族:Yogini / Ashtottari / Kala Chakra",),
),
(
"natal_special_factor_support",
"required_support",
"worksheets.advanced_systems.sahams/avasthas and worksheets.divisional_and_special_charts.upagrahas",
natal_special_factor_support,
("本命补充因子:Sahams / Avasthas / Upagrahas",),
),
)
professional_coverage_manifest: list[dict[str, Any]] = []
for material_id, tier, source_reference, upstream_value, markers in coverage_inputs:
entry, review_reason = _coverage_entry(
material_id,
tier,
source_reference,
upstream_value,
rendered_markdown,
markers,
)
professional_coverage_manifest.append(entry)
if review_reason:
review_reasons.append(review_reason)
patyayini_rows_present = _patyayini_normalized_rows_present(packet)
if patyayini_rows_present is True:
checks.append(_check("patyayini:normalized_rows", "passed", "present"))
elif patyayini_rows_present is False:
review_reasons.append("patyayini_normalized_rows_missing")
checks.append(_check("patyayini:normalized_rows", "review_required", "missing or empty"))
overrides = _as_dict(packet.get("professional_coverage_overrides"))
for material_id in RESTRICTED_MATERIAL_IDS:
override = _as_dict(overrides.get(material_id))
override_status = str(override.get("status") or "restricted_or_unclosed")
professional_coverage_manifest.append(
{
"material_id": material_id,
"admission_tier": "restricted_or_unclosed",
"report_role": "restricted_research_material",
"source_reference": "professional_coverage_overrides",
"upstream_present": bool(override),
"status": override_status,
"source_status": override_status,
"surface_location": "blocked_or_research_appendix",
"rendered": None,
"limitation_reference": "must_not_generate_result_claim",
}
)
if override_status in {"confirmed", "verified", "approved", "promoted"}:
blocking_reasons.append(f"restricted_material_promoted:{material_id}")
checks.append(_check(f"restricted:{material_id}", "blocked", override_status))
status = _promotion_status(blocking_reasons, review_reasons, warning_reasons)
return {
"schema_version": SCHEMA_VERSION,
"status": status,
"checks": checks,
"blocking_reasons": blocking_reasons,
"review_reasons": review_reasons,
"warning_reasons": warning_reasons,
"professional_coverage_manifest": professional_coverage_manifest,
"audit_reference": {
"source_schema": source_schema or None,
"calculation_profile_id": packet.get("calculation_profile_id") or profile.get("profile_id"),
"result_hash": packet.get("result_hash"),
"read_only": True,
},
}
def main() -> int:
parser = argparse.ArgumentParser(description="Evaluate an assembled PL9 full-report packet without modifying it")
parser.add_argument("--packet", required=True, help="PL9 export JSON packet")
parser.add_argument("--markdown", help="Optional rendered Markdown to validate reader-facing coverage")
args = parser.parse_args()
packet = json.loads(Path(args.packet).read_text(encoding="utf-8"))
markdown = Path(args.markdown).read_text(encoding="utf-8") if args.markdown else None
print(json.dumps(evaluate_full_report(packet, markdown), ensure_ascii=False, indent=2, sort_keys=True))
return 0
if __name__ == "__main__": # pragma: no cover
raise SystemExit(main())