jesse-uxandGrok (agent loop)
2e9ca2d73a
产品负责人 2026-11-08：用户看到的报告文字不出现「联网 / 外部资料 / 外部核验 / MEVG / 真实案例 / Real Case / Global Web」，后台记录保留。 - 个人报告写作提示词：依据与边界改为「没有用户经历可对照」，置信度封顶 B 不变 - 年报：技法执行情况跳过 MEVG / Real Case 两行；没有纳入的删掉对应条目 - 合盘：技法执行情况删掉「外部资料与真实案例核对：未进行」整行 - 产品导出审计章节：这两行不再渲染，证据边界同类一句删除，packet 原样保留 - 质量门：只要求 Functional Benefic/Malefic 渲染；新增后台检查，要求 packet 仍带这两行且它们没有出现在 markdown 里 Tests: 新增 frontend/tests/report-no-web-wording-20261109.test.ts（4）与 tests/test_report_no_web_wording_20261109.py（3）；更新年报、合盘、审计章节、 质量门既有断言（原值/新值/原因已注明）。 Co-Authored-By: Grok (agent loop)
651 lines
30 KiB
Python
Executable File
651 lines
30 KiB
Python
Executable File
#!/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
|
||
|
||
try:
|
||
from scripts.calculation_profile_contract import (
|
||
ALLOWED_BINDING_EXCLUDED_PATHS,
|
||
ALLOWED_BINDING_EXCLUDED_TOP_KEYS,
|
||
hash_bound_result,
|
||
)
|
||
except ImportError: # pragma: no cover - direct-script execution
|
||
from calculation_profile_contract import (
|
||
ALLOWED_BINDING_EXCLUDED_PATHS,
|
||
ALLOWED_BINDING_EXCLUDED_TOP_KEYS,
|
||
hash_bound_result,
|
||
)
|
||
|
||
|
||
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",
|
||
)
|
||
TECHNIQUE_AUDIT_MARKERS = ("## 完整 Technique Audit Table", "Full Technique Audit Table")
|
||
REQUIRED_TECHNIQUE_AUDIT_ROWS = (
|
||
"Functional Benefic/Malefic",
|
||
)
|
||
# 原值：这三行都必须渲染在审计表里
|
||
# 新值：后两行只在后台 packet 里检查，渲染里不再出现
|
||
# 原因：产品负责人 2026-11-08：用户可见文字不出现联网/外部核验字样，后台保留
|
||
BACKSTAGE_TECHNIQUE_AUDIT_ROWS = (
|
||
"MEVG / Global Web Evidence",
|
||
"Real Case Calibration",
|
||
)
|
||
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_evidence_claims(packet: dict[str, Any]) -> dict[str, Any]:
|
||
"""Validate explicit claim bindings, not arbitrary free-text astrological prose.
|
||
|
||
Each claim has a path (list of dictionary keys/list indices) and a value.
|
||
Model-declared 'used' or a repeated answer never counts as a receipt.
|
||
"""
|
||
claims = packet.get("evidence_claims")
|
||
if claims is None or claims == []:
|
||
return {"status": "not_supplied", "checked": 0, "errors": [],
|
||
"boundary": "free_text_not_semantically_verified"}
|
||
if not isinstance(claims, list):
|
||
return {"status": "blocked", "checked": 0, "errors": ["claims_not_list"]}
|
||
errors = []
|
||
for i, claim in enumerate(claims):
|
||
if not isinstance(claim, dict) or "value" not in claim:
|
||
errors.append(f"claim_{i}:invalid_binding")
|
||
continue
|
||
path = claim.get("path")
|
||
if not isinstance(path, list) or not path or path[0] == "evidence_claims":
|
||
errors.append(f"claim_{i}:invalid_path")
|
||
continue
|
||
value = packet
|
||
try:
|
||
for key in path:
|
||
if isinstance(value, dict) and isinstance(key, str):
|
||
value = value[key]
|
||
elif isinstance(value, list) and type(key) is int and key >= 0:
|
||
value = value[key]
|
||
else:
|
||
raise KeyError("invalid_path_component")
|
||
except (KeyError, IndexError, TypeError):
|
||
errors.append(f"claim_{i}:source_missing")
|
||
continue
|
||
if type(value) is not type(claim["value"]) or value != claim["value"]:
|
||
errors.append(f"claim_{i}:value_mismatch")
|
||
return {"status": "blocked" if errors else "bindings_match", "checked": len(claims),
|
||
"errors": errors, "boundary": "bound_values_only_not_formula_or_prediction_validation"}
|
||
|
||
|
||
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] = []
|
||
claim_validation = evaluate_evidence_claims(packet)
|
||
if claim_validation["status"] == "blocked":
|
||
blocking_reasons.append("evidence_claim_bindings_invalid")
|
||
|
||
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"))
|
||
|
||
scope = _as_dict(binding.get("binding_scope"))
|
||
if "excluded_top_keys" not in scope or "excluded_paths" not in scope:
|
||
blocking_reasons.append("provenance_binding_scope_absent")
|
||
checks.append(_check(
|
||
"provenance:result_binding_scope",
|
||
"blocked",
|
||
"provenance_binding_scope_absent",
|
||
))
|
||
else:
|
||
extra_top = {
|
||
str(item) for item in (scope.get("excluded_top_keys") or ())
|
||
} - set(ALLOWED_BINDING_EXCLUDED_TOP_KEYS)
|
||
extra_paths = {
|
||
str(item) for item in (scope.get("excluded_paths") or ())
|
||
} - set(ALLOWED_BINDING_EXCLUDED_PATHS)
|
||
if extra_top or extra_paths:
|
||
blocking_reasons.append("provenance_binding_scope_unexpected_exclusion")
|
||
checks.append(_check(
|
||
"provenance:result_binding_scope",
|
||
"blocked",
|
||
"provenance_binding_scope_unexpected_exclusion",
|
||
))
|
||
elif (
|
||
isinstance(expected_input_hash, str)
|
||
and hash_bound_result(packet, expected_input_hash, scope) != expected_result_hash
|
||
):
|
||
blocking_reasons.append("provenance_binding_scope_mismatch")
|
||
checks.append(_check(
|
||
"provenance:result_binding_scope",
|
||
"blocked",
|
||
"provenance_binding_scope_mismatch",
|
||
))
|
||
else:
|
||
checks.append(_check(
|
||
"provenance:result_binding_scope",
|
||
"passed",
|
||
"clipped payload matches result_hash",
|
||
))
|
||
|
||
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": ("D1–D60 完整原始分盘账本", "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"))
|
||
|
||
# Upstream check restored (产品 2026-11-07 占星师方案：导出保留审计). The product
|
||
# export renders the table as 「### 完整 Technique Audit Table」 / 「Full Technique Audit Table」.
|
||
technique_audit_table = packet.get("technique_audit_table")
|
||
if not isinstance(technique_audit_table, list) or not technique_audit_table:
|
||
review_reasons.append("technique_audit_table_missing")
|
||
checks.append(_check("technique_audit_table", "review_required", "missing"))
|
||
elif rendered_markdown is not None and not _marker_present(rendered_markdown, TECHNIQUE_AUDIT_MARKERS):
|
||
review_reasons.append("technique_audit_table_not_rendered")
|
||
checks.append(_check("technique_audit_table", "review_required", "not rendered"))
|
||
else:
|
||
checks.append(_check("technique_audit_table", "passed", f"{len(technique_audit_table)} rows"))
|
||
if isinstance(technique_audit_table, list) and technique_audit_table:
|
||
named = {str(row.get("technique")) for row in technique_audit_table if isinstance(row, dict)}
|
||
missing_rows = [name for name in REQUIRED_TECHNIQUE_AUDIT_ROWS if name not in named]
|
||
if missing_rows:
|
||
review_reasons.append("technique_audit_required_rows_missing")
|
||
checks.append(_check("technique_audit_table:required_rows", "review_required", ", ".join(missing_rows)))
|
||
else:
|
||
checks.append(_check("technique_audit_table:required_rows", "passed", ", ".join(REQUIRED_TECHNIQUE_AUDIT_ROWS)))
|
||
# 原值：这两行必须出现在渲染出的审计表里
|
||
# 新值：渲染里不再出现（产品负责人 2026-11-08：用户可见文字不出现联网/外部核验字样），
|
||
# 改为检查后台 packet 的 technique_audit_table 仍然带着它们
|
||
# 原因：产品负责人 2026-11-08：用户可见文字不出现联网/外部核验字样，后台保留
|
||
missing_backstage = [name for name in BACKSTAGE_TECHNIQUE_AUDIT_ROWS if name not in named]
|
||
if missing_backstage:
|
||
review_reasons.append("technique_audit_backstage_rows_missing")
|
||
checks.append(_check("technique_audit_table:backstage_rows", "review_required", ", ".join(missing_backstage)))
|
||
else:
|
||
checks.append(_check("technique_audit_table:backstage_rows", "passed", ", ".join(BACKSTAGE_TECHNIQUE_AUDIT_ROWS)))
|
||
leaked = [name for name in BACKSTAGE_TECHNIQUE_AUDIT_ROWS
|
||
if rendered_markdown is not None and name in rendered_markdown]
|
||
if leaked:
|
||
review_reasons.append("technique_audit_backstage_rows_rendered")
|
||
checks.append(_check("technique_audit_table:backstage_rows_hidden", "review_required", ", ".join(leaked)))
|
||
else:
|
||
checks.append(_check("technique_audit_table:backstage_rows_hidden", "passed", "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")),
|
||
("D2（Hora 财富分盘）",),
|
||
),
|
||
(
|
||
"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")),
|
||
("D11（Rudramsa 收益分盘）",),
|
||
),
|
||
(
|
||
"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,
|
||
"evidence_claim_validation": claim_validation,
|
||
"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())
|