feat(report): add dense personal report appendix

Co-Authored-By: Claude Code <noreply@anthropic.com>
This commit is contained in:
jesse-ux
2026-09-23 10:42:17 +08:00
co-authored by Claude Code
parent a7f82dc2a6
commit bbd96d3b6b
44 changed files with 48833 additions and 50 deletions
+9 -2
View File
@@ -43,7 +43,7 @@ TECHNIQUE_STATUSES = ("verified", "partial", "blocked")
CONFLICT_STATUSES = ("unresolved", "partial", "resolved")
REPORT_ACTION_PRIORITIES = ("now", "next", "watch")
V1_CHART_IDS = ("D1", "D9", "D10")
CHART_IDS = ("D1", "D2", "D6", "D8", "D9", "D10", "D11", "D24", "D30")
CHART_IDS = ("D1", "D2", "D3", "D4", "D7", "D9", "D10", "D11", "D12", "D16", "D20", "D24", "D27", "D30", "D40", "D60")
HOUSE_NUMBERS = tuple(range(1, 13))
FAILURE_CODES = (
@@ -778,7 +778,14 @@ def validate_report_document(document: Any) -> ValidationResult:
"thematicNarrative", "blockedConflictDisclosure", "evidenceAppendix", "disclaimer",
)
required = v1_required if version == V1_SCHEMA_VERSION else v2_required
_check_keys(result, "(root)", document, required)
_check_keys(result, "(root)", document, required,
(*required, "factTables") if version == V2_SCHEMA_VERSION else required)
if version == V2_SCHEMA_VERSION and "factTables" in document:
try:
from scripts.report_fact_table_contract import validate_fact_tables
except ModuleNotFoundError:
from report_fact_table_contract import validate_fact_tables
validate_fact_tables(document["factTables"], result.add)
_check_pattern(result, "reportId", document.get("reportId"), UUID_PATTERN, "uuid")
_check_enum(result, "reportType", document.get("reportType"), REPORT_TYPES)
_check_enum(result, "presentationMode", document.get("presentationMode"), PRESENTATION_MODES)
+12
View File
@@ -12,6 +12,15 @@ from importlib import import_module
from types import SimpleNamespace
from typing import Any
try:
from scripts.reader_appendix_language import clean_reader_appendix_markdown
from scripts.reader_dasha_applicability import render_dasha_applicability
from scripts.report_density_packet import report_density_packet
except ModuleNotFoundError: # pragma: no cover - direct scripts/ execution path
from reader_appendix_language import clean_reader_appendix_markdown
from reader_dasha_applicability import render_dasha_applicability
from report_density_packet import report_density_packet
class ProfessionalReportReferenceInputError(ValueError):
"""The professional-reference request is outside the public contract."""
@@ -109,6 +118,9 @@ def build_professional_report_reference(handler, body: dict[str, Any], *, engine
return {
"format": "markdown",
"markdown": resolved_engine.render_pl9_markdown(packet),
**({"fact_table_packet": report_density_packet(packet),
"reader_dasha_applicability": clean_reader_appendix_markdown(render_dasha_applicability(packet))}
if body.get("include_fact_tables") is True else {}),
}
return {
"format": "json",
+45
View File
@@ -0,0 +1,45 @@
"""Reader-only vocabulary projection; calculation packets and CLI stay untouched."""
from __future__ import annotations
import re
STATUS_LABELS = {
"parameter_sensitive": "参数敏感",
"pyjhora_behavior_only": "仅单一外部参照,未做多引擎核对",
"not_multiengine_parity": "仅单一外部参照,未做多引擎核对",
"unresolved_external_tuple_boundary": "外部边界未对齐",
"partial_verified": "部分核验",
"raw_appendix_only": "仅原始附录可见",
"missing_in_local": "本地暂无",
"internal_reference_omitted": "内部参照已省略",
}
CELL_LABELS = {"blocked": "暂不可用", "executed": "已执行", "available": "可用", "computed": "已计算"}
_TOKEN = re.compile(r"\b(" + "|".join(STATUS_LABELS) + r")\b")
_PRODUCT = re.compile(
r"\bPL9(?:\.pdf)?(?:[ \t]*(?:第[ \t]*\d+(?:[ \t]*[–—/-][ \t]*\d+)*[ \t]*页|"
r"(?:pages?|p)[ \t-]*\d+(?:[ \t]*[–—/-][ \t]*\d+)*))?\b|"
r"\bPL9[ \t]*第[ \t]*\d+(?:[ \t]*[–—/-][ \t]*\d+)*[ \t]*页",
re.IGNORECASE,
)
_FUNCTION = re.compile(r"\b(?:jyotish_engine\.)?(?:cmd_[a-z0-9_]+|render_pl9_markdown|build_professional_report_reference_packet)\b")
_CELL = re.compile(r"(?<=\|)([ \t]*)(`?)(blocked|executed|available|computed)\2([ \t]*)(?=\|)")
def clean_reader_appendix_markdown(markdown: str) -> str:
"""Translate copy without dropping lines, field names, or numeric cells.
Ambiguous English words are changed only as complete pipe-table cells.
Chart fences are immutable: their JSON is a separate renderer contract.
"""
parts = re.split(r"(^[ \t]*```jyotish-chart[^\n]*\n[\s\S]*?^[ \t]*```[^\n]*(?:\n|$))", markdown, flags=re.MULTILINE)
for index in range(0, len(parts), 2):
value = _TOKEN.sub(lambda match: STATUS_LABELS[match[0]], parts[index])
value = _FUNCTION.sub("本地计算", value)
value = re.sub(r"\b(?:[a-z0-9_]+pl9[a-z0-9_]*|pl9_[a-z0-9_]+)(?:\.v\d+)?\b", "外部参照资料", value, flags=re.IGNORECASE)
value = _PRODUCT.sub("外部参照资料", value)
value = re.sub(r"\b(?:PyJHora|JHora)\b", "外部参照引擎", value, flags=re.IGNORECASE)
value = _CELL.sub(lambda match: f"{match[1]}{match[2]}{CELL_LABELS[match[3]]}{match[2]}{match[4]}", value)
parts[index] = value
return "".join(parts)
+73
View File
@@ -0,0 +1,73 @@
"""Reader-only applicability notes; never changes dasha calculations or periods."""
BOUNDARY = "第二时间轴,仅供交叉核对,不单独生成应期"
MISSING = "缺少本次计算的结构化适用性证据,无法确认各辅助时间轴是否适用;不补算、不补写日期。"
REASON_LABELS = {
"patyayini_dasha_missing": "本次计算未提供帕特雅伊尼年运周期,暂不能用于时间核对。",
"The Shodashottari hora-plus-paksha applicability gate is not met.": "未满足该大运对时辰分区与月相组合的适用条件。",
"This profile requires Navamsa Lagna in Taurus or Libra.": "该大运要求九分盘上升落在金牛座或天秤座,本次未满足。",
"Dwisaptatisama requires Lagna lord in the 1st or 7th house.": "该大运要求上升主星位于第一宫或第七宫,本次未满足。",
"Panchottari requires a Cancer Ascendant that is also Cancer in Dwadasamsa.": "该大运要求本命及十二分盘上升均为巨蟹座,本次未满足。",
"Satabdika requires a vargottama Ascendant (same Rasi and Navamsa sign).": "该大运要求本命及九分盘上升同星座,本次未满足。",
"Chaturshitisama requires the 10th lord to occupy the 10th house.": "该大运要求第十宫主星位于第十宫,本次未满足。",
"Shastihayani requires Sun in the Lagna sign.": "该大运要求太阳位于上升星座,本次未满足。",
}
def _reader_reason(value):
if value in REASON_LABELS:
return REASON_LABELS[value]
if value.startswith("Moon Nakshatra '") and "does not qualify for Ashtottari Dasha under " in value:
return "月宿与月相组合未满足该大运的适用条件,不绕过判定。"
return "计算返回了限制原因,但暂无可核对的详细说明;不据此补写适用条件或日期。"
def _has_periods(key, raw, source):
fields = ("periods", "major") + (("mahadasha_sequence", "dense_boundaries") if key == "narayana" else ())
return any(isinstance(data.get(field), (list, dict)) and bool(data[field]) for data in (raw, source) for field in fields)
def _cell(value):
return str(value).replace("|", "/").replace("\r", " ").replace("\n", " ").replace("<", "<").replace(">", ">")
def render_dasha_applicability(packet):
timing = packet.get("worksheets", {}).get("timing_and_predictive_systems", {})
master = timing.get("dasha_master_pack", {})
families = master.get("families", {})
lines = ["", "## 辅助时间轴适用性(参数敏感)", ""]
if not isinstance(families, dict) or not families:
return "\n".join(lines + [MISSING, ""])
families = dict(families)
annual = timing.get("annual_tajika_pack", {})
for key, value in annual.items():
if key.endswith("_dasha"):
families.setdefault(key.removesuffix("_dasha"), value)
lines += ["| 大运族 | 是否适用 / 原因 | 使用边界 |", "| --- | --- | --- |"]
for key in sorted(families):
if key == "vimshottari":
continue
raw = families[key] if isinstance(families[key], dict) else {}
source_key = "kalachakra_dasha" if key == "kala_chakra" else f"{key}_dasha"
source = timing.get(source_key, {})
source = source if isinstance(source, dict) else {}
status = raw.get("status", raw.get("execution_status"))
applicable = source.get("applicable", raw.get("applicable"))
if applicable is False or status == "not_applicable":
reason = "本次不适用,不绕过原适用条件,不补写日期。"
elif status in {"blocked", "missing_in_local", "unavailable"}:
reason = "本次不可用:本地计算尚未提供可用周期,不能据此判断时间。"
elif _has_periods(key, raw, source):
reason = "本次已返回周期,可作交叉核对;这不代表已完成多引擎验证。"
else:
reason = "本次适用性尚未核实,缺少可核对周期,不补写日期。"
if applicable is True:
reason += " 原适用性判定:适用。"
reasons = list(dict.fromkeys(_reader_reason(str(value)) for value in (raw.get("reason"), source.get("reason")) if value))
if reasons:
reason += " 原因:" + ";".join(reasons)
name = "Kala Chakra" if key == "kala_chakra" else key.replace("_", "-").title()
lines.append(f"| {_cell(name)} | {_cell(reason)} | {BOUNDARY} |")
return "\n".join(lines) + "\n"
+19
View File
@@ -0,0 +1,19 @@
"""Select calculation groups for server-owned report assembly, not writer input."""
GROUPS = {
"strengths_and_scores": ("ashtakavarga", "shadbala", "functional_benefic_malefic"),
"timing_and_predictive_systems": ("dasha", "annual_tajika_pack"),
"advanced_systems": ("avasthas", "yogas_doshas", "sahams"),
"divisional_and_special_charts": ("upagrahas", "special_lagnas", "sensitive_points", "varga_full"),
}
def report_density_packet(packet):
worksheets = packet.get("worksheets", {})
return {
"core_chart": packet.get("core_chart", {}),
"worksheets": {
group: {key: worksheets.get(group, {})[key] for key in keys if key in worksheets.get(group, {})}
for group, keys in GROUPS.items()
},
}
+61
View File
@@ -0,0 +1,61 @@
"""Optional v2 server-owned fact tables; no writer output accepts these fields."""
import math
import re
GROUPS = {"vimshottari", "ashtakavarga", "shadbala", "functional_roles", "avasthas", "special_points", "sade_sati", "annual"}
PATH = re.compile(r"[A-Za-z0-9_.\[\]-]{1,400}")
STATUSES = {"multi_system_consensus", "single_system_inference", "parameter_sensitive", "unclosed_divisional_chart", "user_history_verification_required", "blocked"}
def validate_fact_tables(value, add):
def text(item, maximum, minimum=0):
return isinstance(item, str) and minimum <= len(item) <= maximum
if not isinstance(value, list) or len(value) > 8:
add("factTables", "must be array with at most eight groups")
return
seen = set()
for index, table in enumerate(value):
path = f"factTables[{index}]"
if not isinstance(table, dict) or set(table) != {"id", "title", "claimStatus", "sourcePath", "note", "columns", "rows"}:
add(path, "invalid table keys")
continue
group = table["id"]
if not isinstance(group, str) or group not in GROUPS or group in seen:
add(path, "invalid or duplicate group")
else:
seen.add(group)
source = table["sourcePath"]
if not isinstance(source, str) or not PATH.fullmatch(source):
add(path, "invalid source path")
continue
if not text(table["title"], 120, 1) or not text(table["note"], 500):
add(path, "invalid text")
if table["claimStatus"] not in STATUSES:
add(path, "invalid claim status")
columns = table["columns"]
if not isinstance(columns, list) or not 1 <= len(columns) <= 16 or not all(text(c, 80, 1) for c in columns):
add(path, "invalid columns")
continue
rows = table["rows"]
if not isinstance(rows, list) or len(rows) > 2000:
add(path, "invalid rows")
continue
for row in rows:
if not isinstance(row, dict) or set(row) != {"sourcePath", "cells"}:
add(path, "invalid row keys")
continue
origin = row["sourcePath"]
if not isinstance(origin, str) or not PATH.fullmatch(origin) or not (origin == source or origin.startswith(source + ".") or origin.startswith(source + "[")):
add(path, "fact_table_source_mismatch")
cells = row["cells"]
if not isinstance(cells, list) or len(cells) != len(columns):
add(path, "fact_table_cell_count")
continue
for cell in cells:
if cell is None or isinstance(cell, bool) or text(cell, 300):
continue
if isinstance(cell, (int, float)) and math.isfinite(cell):
continue
add(path, "invalid cell")