feat(skill): merge upstream reader-report contract

2026-08-06 11:31:51 +08:00
parent 685ed00e2f
commit 03d3b58ff7
8 changed files with 1163 additions and 2 deletions
@@ -1024,6 +1024,42 @@ class MockDataFactory:
        ]


+# ═══════════════════════════════════════════════════════════════
+# Reader report output contract
+# ═══════════════════════════════════════════════════════════════
+
+def render_reader_report(reader_report: Dict[str, Any]) -> Dict[str, Any]:
+    """Return the stable reader order without recalculating report facts.
+
+    The evidence appendix is collapsed for normal users and expanded by default
+    only in explicit research mode. Unknown commercial extensions are retained
+    after the three reader-facing sections for backward compatibility.
+    """
+    executive_summary = reader_report.get("executive_summary")
+    thematic_narrative = reader_report.get("thematic_narrative")
+    supplied_appendix = reader_report.get("evidence_appendix")
+    evidence_appendix = dict(supplied_appendix) if isinstance(supplied_appendix, dict) else {}
+    presentation_mode = reader_report.get("presentation_mode")
+    if not presentation_mode and isinstance(executive_summary, dict):
+        presentation_mode = executive_summary.get("presentation_mode")
+    presentation_mode = presentation_mode or "default"
+    evidence_appendix["expanded"] = presentation_mode == "research"
+
+    extensions = {
+        key: value
+        for key, value in reader_report.items()
+        if key not in {"presentation_mode", "executive_summary", "thematic_narrative", "evidence_appendix"}
+    }
+    rendered = {
+        "presentation_mode": presentation_mode,
+        "executive_summary": executive_summary,
+        "thematic_narrative": thematic_narrative,
+        "evidence_appendix": evidence_appendix,
+    }
+    rendered.update(extensions)
+    return rendered
+
+
 # ═══════════════════════════════════════════════════════════════
 # 演示与测试入口
 # ═══════════════════════════════════════════════════════════════
@@ -1,14 +1,23 @@
 #!/usr/bin/env python3
-# -*- coding: utf-8 -*-
 """Shared orchestration contract for skill/MCP and web/API surfaces."""

 from __future__ import annotations

 import json
+import re
 from dataclasses import dataclass
 from pathlib import Path
 from typing import Any

+
+def _formal_divisions() -> tuple[int, ...]:
+    registry = Path(__file__).resolve().parents[1] / "references/oracle/d1_d60_varga_mapping_registry_2026_07_19.json"
+    rows = json.loads(registry.read_text(encoding="utf-8"))["rows"]
+    return tuple(int(row["number"]) for row in rows if row.get("formal_name_present"))
+
+
+FORMAL_DIVISIONS = _formal_divisions()
+
 try:
    from diagnose_pyjhora_adapter import build_report as build_pyjhora_adapter_report
 except Exception:  # pragma: no cover - import path varies in tests/CLI
@@ -204,6 +213,55 @@ class UnifiedConsultationOrchestrator:
        "official_event_radar_expansion",
        "extended_prompt_pack_refresh",
    ]
+    _DOMAIN_PROFILE_SECTIONS = {
+        "relationship": ["core_partner_profile", "temperament_and_compatibility", "timing_windows", "red_flags", "verification_questions"],
+        "career": ["career_direction", "role_and_responsibility", "income_and_recognition", "opportunity_windows", "risks_and_verification_questions"],
+        "finance": ["wealth_path", "income_structure", "asset_and_cashflow_pattern", "opportunity_windows", "verification_questions"],
+        "health": ["non_medical_pattern", "pressure_factors", "protective_factors", "verification_questions"],
+        "migration": ["relocation_pattern", "foreign_link", "candidate_windows", "verification_questions"],
+        "family": ["family_structure", "home_and_care", "children_boundary", "verification_questions"],
+        "education": ["study_vs_work_fit", "exam_and_degree_path", "candidate_windows", "verification_questions"],
+        "annual": ["annual_themes", "candidate_windows", "claim_boundaries", "verification_questions"],
+        "timing": ["active_themes", "candidate_windows", "triggering_techniques", "verification_questions"],
+        "general": ["life_themes", "strengths_and_pressures", "candidate_windows", "verification_questions"],
+    }
+    _THEME_VARGA_DISPATCH = {
+        "career": ("D1", "D10", "D24"),
+        "marriage": ("D1", "D9", "D7", "D12"),
+        "wealth": ("D1", "D2", "D11", "D4"),
+        "health": ("D1", "D6", "D8", "D30"),
+        "migration": ("D1", "D4", "D12"),
+        "family": ("D1", "D7", "D12"),
+        "education": ("D1", "D5", "D24"),
+        "annual": ("D1", "D9", "D10"),
+        "spirituality": ("D1", "D20", "D24", "D60"),
+    }
+    _THEME_TECHNIQUE_IDENTIFIERS = {
+        "career": {"D10", "A10"},
+        "marriage": {"D9", "UL", "UPAPADA", "VIVAH"},
+        "wealth": {"D2", "D11"},
+        "health": {"D6", "D8", "D30"},
+        "migration": {"D4", "D12"},
+        "family": {"D7", "D12"},
+        "education": {"D5", "D24"},
+        "annual": {"DASHA", "TRANSIT", "TAJIKA"},
+        "spirituality": {"D20", "D60"},
+    }
+
+    @classmethod
+    def route_profile_contract(cls, route_name: str) -> dict[str, Any]:
+        """Expose reader sections without replacing runtime technique audit."""
+        route = route_name if route_name in cls._DOMAIN_PROFILE_SECTIONS else "general"
+        return {
+            "version": "domain_profile_v1",
+            "route": route,
+            "sections": list(cls._DOMAIN_PROFILE_SECTIONS[route]),
+            "assertion_levels": [
+                "multi_system_consensus", "single_system_inference", "parameter_sensitive",
+                "unclosed_divisional_chart", "user_history_verification_required", "blocked",
+            ],
+            "execution_boundary": "Profile labels do not replace Technique Audit Table execution status.",
+        }

    def normalize_themes(self, raw: Any) -> list[str]:
        if raw in (None, "", "all"):
@@ -281,6 +339,140 @@ class UnifiedConsultationOrchestrator:
            "display_label": route.display_label,
        }

+    def route_profile(self, question: str, themes: list[str] | None = None) -> dict[str, Any]:
+        """Select presentation depth and on-demand Vargas without changing routing."""
+        normalized_themes = self.normalize_themes(themes)
+        request = (question or "").lower()
+        is_research = any(token in request for token in ("研究模式", "research_mode", "原始数据", "全量数据", "raw_data"))
+        selected: list[str] = []
+        for theme in normalized_themes:
+            for code in self._THEME_VARGA_DISPATCH.get(theme, ("D1",)):
+                if code not in selected:
+                    selected.append(code)
+        formal = [f"D{division}" for division in FORMAL_DIVISIONS]
+        return {
+            "question": question or "",
+            "themes": normalized_themes,
+            "presentation_mode": "research" if is_research else "default",
+            "appendix_expanded": is_research,
+            "varga_dispatch": {
+                "mode": "on_demand",
+                "selected_theme_vargas": selected,
+                "all_formal_vargas": formal,
+                "deferred_vargas": [code for code in formal if code not in selected],
+                "rule": "先调用与主题直接相关的分盘；其余正式分盘只在追问或冲突时展开。",
+            },
+        }
+
+    @staticmethod
+    def _deduplicate_sentences(text: str) -> str:
+        parts = re.split(r"(?<=[.!?。！？])", text)
+        seen: set[str] = set()
+        kept: list[str] = []
+        for part in parts:
+            key = part.strip()
+            if key and key not in seen:
+                seen.add(key)
+                kept.append(part)
+        return "".join(kept)
+
+    @classmethod
+    def _suppress_definitive_claims(cls, text: str) -> str:
+        conditional = text.replace("确定", "尚无法确认").replace("必然", "未必").replace("一定", "尚无法确认")
+        conditional = re.sub(r"\bwill\s+(?:definitely|certainly|inevitably)\b", "may", conditional, flags=re.IGNORECASE)
+        return re.sub(r"\b(?:definitely|certainly|certain|inevitably|guaranteed)(?:\s+(?:definitely|certainly|certain|inevitably|guaranteed))*\b", "not yet verified", conditional, flags=re.IGNORECASE)
+
+    def _audit_applies_to_theme(self, row: dict[str, Any], theme: str) -> bool:
+        for field in ("theme", "domain"):
+            if str(row.get(field) or "").lower() == theme:
+                return True
+        for field in ("themes", "domains", "applicable_themes"):
+            values = row.get(field)
+            if isinstance(values, str) and values.lower() == theme:
+                return True
+            if isinstance(values, list) and theme in {str(value).lower() for value in values}:
+                return True
+        technique = str(row.get("technique") or row.get("name") or "").upper()
+        identifiers = set(re.findall(r"\b[A-Z]+\d*\b", technique))
+        return bool(identifiers & self._THEME_TECHNIQUE_IDENTIFIERS.get(theme, set()))
+
+    @staticmethod
+    def _infer_technique_system(row: dict[str, Any]) -> str:
+        technique = str(row.get("technique") or row.get("name") or "").lower()
+        if "cross-system" in technique or "cross system" in technique:
+            return "cross_system"
+        if any(token in technique for token in ("western", "solar return", "secondary progression", "solar arc", "midpoint")):
+            return "western"
+        return "jyotish"
+
+    def _normalize_audit_row(self, row: dict[str, Any]) -> dict[str, Any]:
+        normalized = dict(row)
+        status = str(normalized.get("status") or "unknown").lower()
+        normalized["system"] = str(normalized.get("system") or self._infer_technique_system(normalized))
+        normalized["confidence_label"] = str(normalized.get("confidence_label") or {
+            "executed": "multi_system_consensus", "used": "multi_system_consensus", "complete": "multi_system_consensus",
+            "partial": "parameter_sensitive", "research_only": "parameter_sensitive", "blocked": "blocked",
+        }.get(status, "single_system_inference"))
+        normalized["user_visible_summary"] = str(normalized.get("user_visible_summary") or f"{normalized.get('technique') or normalized.get('name') or 'Technique'} · {normalized['system']} · {status}")
+        return normalized
+
+    @staticmethod
+    def _audit_overview(rows: list[dict[str, Any]]) -> dict[str, Any]:
+        statuses: dict[str, int] = {}
+        systems: dict[str, int] = {}
+        for row in rows:
+            status, system = str(row.get("status") or "unknown").lower(), str(row.get("system") or "unknown").lower()
+            statuses[status] = statuses.get(status, 0) + 1
+            systems[system] = systems.get(system, 0) + 1
+        return {"status_counts": statuses, "system_counts": systems, "blocked_count": statuses.get("blocked", 0)}
+
+    def build_reader_report(
+        self,
+        route_profile: dict[str, Any],
+        theme_reports: dict[str, Any],
+        raw_data: Any = None,
+        technique_audit: list[dict[str, Any]] | None = None,
+        conflicts: list[Any] | None = None,
+    ) -> dict[str, Any]:
+        """Build summary -> narrative -> appendix while preserving blocked truth."""
+        audit_rows = [self._normalize_audit_row(row) for row in list(technique_audit or [])]
+        narrative: dict[str, Any] = {}
+        for theme, report in theme_reports.items():
+            item = dict(report) if isinstance(report, dict) else {"summary": str(report)}
+            blocked = str(item.get("status") or "").lower() == "blocked" or any(
+                str(row.get("status") or "").lower() == "blocked" and self._audit_applies_to_theme(row, theme)
+                for row in audit_rows
+            )
+            for field, value in list(item.items()):
+                if isinstance(value, str):
+                    item[field] = self._deduplicate_sentences(self._suppress_definitive_claims(value) if blocked else value)
+            views = {
+                "jyotish": item.get("jyotish_summary") or item.get("vedic_summary"),
+                "western": item.get("western_summary"),
+                "consensus": item.get("consensus_summary") or item.get("cross_system_summary"),
+            }
+            if any(views.values()):
+                item["system_views"] = {key: value for key, value in views.items() if value}
+            narrative[theme] = item
+        dispatch = route_profile.get("varga_dispatch") if isinstance(route_profile.get("varga_dispatch"), dict) else {}
+        return {
+            "executive_summary": {
+                "themes": list(route_profile.get("themes") or []),
+                "presentation_mode": route_profile.get("presentation_mode") or "default",
+                "selected_theme_vargas": list(dispatch.get("selected_theme_vargas") or []),
+            },
+            "thematic_narrative": narrative,
+            "evidence_appendix": {
+                "expanded": bool(route_profile.get("appendix_expanded")),
+                "raw_data": raw_data,
+                "technique_audit": audit_rows,
+                "audit_overview": self._audit_overview(audit_rows),
+                "blocked_techniques": [str(row.get("technique") or row.get("name") or "unknown") for row in audit_rows if str(row.get("status") or "").lower() == "blocked"],
+                "conflicts": list(conflicts or []),
+                "varga_dispatch": dispatch,
+            },
+        }
+
    def shared_contract(
        self,
        *,
@@ -297,6 +489,7 @@ class UnifiedConsultationOrchestrator:
            "question": question or "",
            "themes": list(themes),
            "route": dict(route_packet),
+            "route_profile_contract": self.route_profile_contract(str(route_packet.get("question_type") or "general")),
            "source_priority": {
                "mode": self.SOURCE_PRIORITY["mode"],
                "priority": list(self.SOURCE_PRIORITY["priority"]),