Jyotisha/scripts/orchestrator_bridge.py

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
orchestrator_bridge.py — 编排器桥接层
=====================================
打通 reading_orchestrator (技法执行层) 与 report_orchestrator (叙事生成层)。

职责：
1. 将 ReadingChapter 转换为 report_orchestrator 可消费的 TechniqueResult
2. 映射主题（ReadingTheme ↔ ThemeName）
3. 统一冲突裁决的表达方式
4. 提供端到端的一体化报告生成入口

版本: v1.0 | 2026-06-07
"""

from typing import Dict, List, Any, Optional
from datetime import datetime
import json

# Import dasha systems (handle relative imports)
try:
    from ashtottari_dasha import calculate_ashtottari_dasha
    from yogini_dasha import calculate_yogini_dasha
    from kalachakra_dasha import calculate_kalachakra_dasha
except ImportError:
    from scripts.ashtottari_dasha import calculate_ashtottari_dasha
    from scripts.yogini_dasha import calculate_yogini_dasha
    from scripts.kalachakra_dasha import calculate_kalachakra_dasha

# Import both orchestrators (handle relative imports)
try:
    from reading_orchestrator import (
        ReadingOrchestrator, ReadingTheme, ReadingChapter,
        TechniqueResult as ReadingTechniqueResult,
    )
except ImportError:
    from scripts.reading_orchestrator import (
        ReadingOrchestrator, ReadingTheme, ReadingChapter,
        TechniqueResult as ReadingTechniqueResult,
    )

try:
    from report_orchestrator import (
        ThematicReportOrchestrator as ReportOrchestrator,
        ThemeName, ThemeReport,
        TechniqueResult as ReportTechniqueResult,
        StrengthLevel, TimingAnchor,
    )
except ImportError:
    from scripts.report_orchestrator import (
        ThematicReportOrchestrator as ReportOrchestrator,
        ThemeName, ThemeReport,
        TechniqueResult as ReportTechniqueResult,
        StrengthLevel, TimingAnchor,
    )


# ═══════════════════════════════════════════════════════════════
# 主题映射
# ═══════════════════════════════════════════════════════════════

THEME_MAPPING: Dict[ReadingTheme, ThemeName] = {
    ReadingTheme.MARRIAGE: ThemeName.MARRIAGE,
    ReadingTheme.CAREER: ThemeName.CAREER,
    ReadingTheme.WEALTH: ThemeName.WEALTH,
    ReadingTheme.HEALTH: ThemeName.HEALTH,
    ReadingTheme.SPIRITUAL: ThemeName.SPIRITUALITY,
}

REVERSE_THEME_MAPPING: Dict[ThemeName, ReadingTheme] = {
    v: k for k, v in THEME_MAPPING.items()
}


# ═══════════════════════════════════════════════════════════════
# 桥接转换器
# ═══════════════════════════════════════════════════════════════

class OrchestratorBridge:
    """
    编排器桥接器

    将 reading_orchestrator 的输出（ReadingChapter）转换为
    report_orchestrator 的输入（TechniqueResult + ThemeReport）。
    """

    def __init__(
        self,
        reading_orchestrator: ReadingOrchestrator,
        report_orchestrator: ReportOrchestrator,
    ):
        self.ro = reading_orchestrator
        self.rpo = report_orchestrator

    # ── ReadingChapter → ReportTechniqueResult ──

    @staticmethod
    def chapter_to_technique_results(chapter: ReadingChapter) -> List[ReportTechniqueResult]:
        """
        将 ReadingChapter 拆解为 report_orchestrator 的 TechniqueResult 列表。

        策略：
        - findings → 每个 finding 对应一个 TechniqueResult
        - conflicts → 每个 conflict 也作为一个 TechniqueResult（sentiment=neutral，标注矛盾）
        - timing → 如果时间锚定存在，也生成一个 TechniqueResult
        """
        results: List[ReportTechniqueResult] = []

        # 1. Findings → positive/negative TechniqueResult
        for finding in chapter.findings:
            sentiment = _infer_sentiment(finding)
            strength = _infer_strength(finding, chapter.conflicts)
            results.append(ReportTechniqueResult(
                technique=f"finding_{len(results)}",
                chart="D1",  # default; can be refined
                conclusion=finding,
                sentiment=sentiment,
                strength=strength,
            ))

        # 2. Conflicts → neutral TechniqueResult with details
        for conflict in chapter.conflicts:
            results.append(ReportTechniqueResult(
                technique="conflict_resolution",
                chart="D1",
                conclusion=conflict.narrative,
                sentiment="neutral",
                strength=StrengthLevel.MODERATE,
                details={
                    "dimension": conflict.dimension,
                    "d1_finding": conflict.d1_finding,
                    "d9_finding": conflict.d9_finding,
                    "resolution": conflict.resolution,
                },
            ))

        # 3. Timing → TechniqueResult
        if chapter.timing:
            results.append(ReportTechniqueResult(
                technique="timing_anchor",
                chart="Dasha",
                conclusion=chapter.timing,
                sentiment="neutral",
                strength=StrengthLevel.MODERATE,
            ))

        return results

    # ── 推运系统注入 ──

    def inject_dasha_results(self, chart_data: Dict[str, Any]) -> Dict[str, Any]:
        """
        计算并注入 Ashtottari / Yogini / Kalachakra Dasha 结果。

        从 chart_data 提取 birth_info，调用三个推运模块，
        将当前推运周期转换为 TimingAnchor 注入 report_orchestrator。

        Returns:
            三个推运系统的原始计算结果，供后续使用。
        """
        # 从 chart_data 提取 birth_info
        birth_info = self._extract_birth_info(chart_data)
        if not birth_info:
            return {}

        results: Dict[str, Any] = {}

        # 1. Ashtottari Dasha (条件性推运)
        try:
            ash = calculate_ashtottari_dasha(birth_info)
            results["ashtottari"] = ash
            if ash.get("applicable") and ash.get("current"):
                self._add_dasha_timing_anchor(
                    system="Ashtottari",
                    current=ash["current"],
                    total_cycle=ash.get("total_cycle", 108),
                )
        except Exception as e:
            import logging
            logging.warning(f"[orchestrator] ashtottari dasha calculation failed: {e}")

        # 2. Yogini Dasha (普遍适用)
        try:
            yog = calculate_yogini_dasha(birth_info)
            results["yogini"] = yog
            if yog.get("current"):
                self._add_dasha_timing_anchor(
                    system="Yogini",
                    current=yog["current"],
                    total_cycle=yog.get("total_cycle", 36),
                )
        except Exception as e:
            import logging
            logging.warning(f"[orchestrator] yogini dasha calculation failed: {e}")

        # 3. Kalachakra Dasha (条件性推运)
        try:
            kal = calculate_kalachakra_dasha(birth_info)
            results["kalachakra"] = kal
            if kal.get("current"):
                self._add_dasha_timing_anchor(
                    system="Kalachakra",
                    current=kal["current"],
                    total_cycle=kal.get("total_cycle", 0),
                )
        except Exception as e:
            import logging
            logging.warning(f"[orchestrator] kalachakra dasha calculation failed: {e}")

        # 4. 将推运结果也作为 TechniqueResult 注入所有主题
        self._inject_dasha_technique_results(results)

        return results

    def _extract_birth_info(self, chart_data: Any) -> Optional[Dict[str, Any]]:
        """从 chart_data 提取推运计算所需的 birth_info。

        支持 dict 和 dataclass 对象（如 BirthChartData）。
        如果数据不足，返回 None（推运注入将静默跳过）。
        """
        import dataclasses

        # 统一转为 dict
        if hasattr(chart_data, "__dataclass_fields__"):
            data = dataclasses.asdict(chart_data)
        elif isinstance(chart_data, dict):
            data = chart_data
        else:
            return None

        info: Dict[str, Any] = {}

        # birth_datetime
        birth_dt = data.get("birth_datetime")
        if birth_dt is None:
            chart = data.get("chart") or data.get("natal_chart")
            if chart:
                birth_dt = chart.get("birth_datetime")
        if isinstance(birth_dt, str):
            birth_dt = datetime.fromisoformat(birth_dt.replace("Z", "+00:00"))
        info["birth_datetime"] = birth_dt

        # moon_nakshatra_index
        moon_nak = data.get("moon_nakshatra_index")
        if moon_nak is None:
            chart = data.get("chart") or data.get("natal_chart")
            if chart:
                nakshatras = chart.get("nakshatras") or chart.get("nakshatra")
                if nakshatras and "Moon" in nakshatras:
                    moon_nak = nakshatras["Moon"].get("index")
        if moon_nak is not None:
            info["moon_nakshatra_index"] = int(moon_nak)

        # is_shukla_paksha (从Tithi推断)
        tithi = data.get("tithi")
        if tithi is None:
            chart = data.get("chart") or data.get("natal_chart")
            if chart:
                tithi = chart.get("tithi")
        if tithi is not None:
            tithi_num = int(tithi) if isinstance(tithi, (int, float, str)) else 15
            info["is_shukla_paksha"] = 1 <= tithi_num <= 15

        # lagna_rashi_index
        lagna = data.get("ascendant") or data.get("lagna")
        if lagna is None:
            chart = data.get("chart") or data.get("natal_chart")
            if chart:
                lagna = chart.get("ascendant") or chart.get("lagna")
        if lagna is not None:
            if isinstance(lagna, str):
                sign_names = [
                    "Aries", "Taurus", "Gemini", "Cancer", "Leo", "Virgo",
                    "Libra", "Scorpio", "Sagittarius", "Capricorn", "Aquarius", "Pisces",
                ]
                if lagna in sign_names:
                    info["lagna_rashi_index"] = sign_names.index(lagna)
            elif isinstance(lagna, dict):
                sign = lagna.get("sign") or lagna.get("rashi")
                if sign:
                    sign_names = [
                        "Aries", "Taurus", "Gemini", "Cancer", "Leo", "Virgo",
                        "Libra", "Scorpio", "Sagittarius", "Capricorn", "Aquarius", "Pisces",
                    ]
                    if sign in sign_names:
                        info["lagna_rashi_index"] = sign_names.index(sign)
                degree = lagna.get("degree")
                if degree is not None:
                    info["lagna_rashi_index"] = int(degree / 30) % 12

        # moon_pada (for Kalachakra)
        moon_pada = data.get("moon_pada")
        if moon_pada is None:
            chart = data.get("chart") or data.get("natal_chart")
            if chart:
                nakshatras = chart.get("nakshatras") or chart.get("nakshatra")
                if nakshatras and "Moon" in nakshatras:
                    moon_pada = nakshatras["Moon"].get("pada")
        if moon_pada is not None:
            info["moon_pada"] = int(moon_pada)
        else:
            info["moon_pada"] = 1

        return info if info.get("birth_datetime") else None

    def _add_dasha_timing_anchor(
        self,
        system: str,
        current: Dict[str, Any],
        total_cycle: int,
    ) -> None:
        """将单个推运当前周期转换为 TimingAnchor 并注入 report_orchestrator。"""
        start_date = current.get("start_date", "")
        end_date = current.get("end_date", "")

        # 提取年份
        try:
            start_year = int(start_date[:4]) if isinstance(start_date, str) else datetime.now().year
            end_year = int(end_date[:4]) if isinstance(end_date, str) else (start_year + 5)
        except (ValueError, TypeError):
            start_year = datetime.now().year
            end_year = start_year + 5

        # 确定行星名称
        planet = current.get("planet") or current.get("yogini") or current.get("lord") or "Unknown"

        # 构建激活描述
        activation = f"{system}推运中，{planet}主运带来人生阶段的转换与业力展现"
        if system == "Yogini":
            activation = f"Yogini推运中，{planet}女神主导当前周期，影响心理与事件层面"
        elif system == "Kalachakra":
            rashi = current.get("rashi", "")
            mode = current.get("mode", "")
            activation = f"Kalachakra {mode}模式推运中，{planet}主宰{rashi}宫阶段"

        anchor = TimingAnchor(
            dasha_period=f"{system}-{planet}",
            start_year=start_year,
            end_year=end_year,
            activation_description=activation,
            is_current=True,
        )

        # 注入到所有5个主题
        for tn in ThemeName:
            self.rpo.add_timing(tn, anchor)

    def _inject_dasha_technique_results(self, dasha_results: Dict[str, Any]) -> None:
        """将推运结果作为 TechniqueResult 注入所有主题。"""
        for system, result in dasha_results.items():
            if not result:
                continue
            current = result.get("current")
            if not current:
                continue

            planet = current.get("planet") or current.get("yogini") or current.get("lord") or "Unknown"
            system_zh = {"ashtottari": "Ashtottari推运", "yogini": "Yogini推运", "kalachakra": "Kalachakra推运"}.get(system, system)

            conclusion = f"{system_zh}：当前处于{planet}主运周期"
            if system == "ashtottari" and result.get("applicable") is False:
                conclusion = f"{system_zh}：不适用（出生条件不符合）"

            tech_result = ReportTechniqueResult(
                technique=f"{system}_dasha",
                chart="Dasha",
                conclusion=conclusion,
                sentiment="neutral",
                strength=StrengthLevel.MODERATE,
                details={
                    "system": system,
                    "current_period": current,
                    "total_cycle": result.get("total_cycle"),
                    "applicable": result.get("applicable", True),
                },
            )

            for tn in ThemeName:
                self.rpo.add_technique(tn, tech_result)

    # ── full-reading.modules → ReportTechniqueResult ──

    def inject_full_reading_modules(self, chart_data: Dict[str, Any]) -> Dict[str, Any]:
        """
        将 `jyotish_engine.py full-reading` 的真实模块输出注入主题化报告。

        v6.1.8 之前，桥接层主要依赖 reading_orchestrator 的 registry；当 registry
        未注册真实执行函数时，主题报告容易退化为少量模拟/占位信息。本方法直接消费
        full-reading.modules 中已经计算完成的模块，把 Yoga、分盘、Ashtakavarga、
        Shadbala、特殊上升点、Vivah Saham、Transit、多系统 Dasha Convergence 等
        转成 report_orchestrator 可叙事的 TechniqueResult。
        """
        modules = _extract_modules(chart_data)
        if not modules:
            return {"injected": False, "reason": "no full-reading modules found", "counts": {}}

        counts = {theme.value: 0 for theme in ThemeName}

        def add(theme: ThemeName, result: ReportTechniqueResult) -> None:
            self.rpo.add_technique(theme, result)
            counts[theme.value] += 1

        self._inject_yoga_module(modules, add)
        self._inject_varga_modules(modules, add)
        self._inject_strength_modules(modules, add)
        self._inject_jaimini_relationship_modules(modules, add)
        self._inject_special_points(modules, add)
        self._inject_dasa_convergence_module(modules, add)
        self._inject_transit_module(modules, add)

        return {
            "injected": True,
            "source": "full-reading.modules",
            "counts": counts,
            "source_modules": sorted(k for k in modules.keys() if k in {
                "yoga", "varga_full", "d9_navamsa_expanded", "ashtakavarga", "shadbala",
                "jaimini", "rashi_tulya_navamsa", "special_lagnas", "vivah_saham",
                "dasa_convergence", "transit_multi_reference",
                "dasha", "yogini_dasha", "ashtottari_dasha", "kalachakra_dasha",
            }),
        }

    def _inject_yoga_module(self, modules: Dict[str, Any], add) -> None:
        yoga = modules.get("yoga") or {}
        yogas = yoga.get("yogas") or yoga.get("detected_yogas") or []
        if not isinstance(yogas, list):
            return

        added_per_theme = {theme: 0 for theme in ThemeName}
        for y in yogas:
            if not isinstance(y, dict):
                continue
            themes = _themes_for_yoga(y)
            if not themes:
                continue
            strength = _strength_from_text(y.get("strength"), default=StrengthLevel.MODERATE)
            sentiment = _sentiment_from_yoga(y)
            effects = y.get("effects") or []
            if isinstance(effects, list):
                effect_text = "、".join(str(e) for e in effects[:4])
            else:
                effect_text = str(effects)
            conclusion = f"{y.get('name_cn') or y.get('name') or 'Yoga'}：{y.get('combination', '检测到相关组合')}"
            if effect_text:
                conclusion += f"；传统效应：{effect_text}"

            for theme in themes:
                if added_per_theme[theme] >= 8:
                    continue
                add(theme, ReportTechniqueResult(
                    technique=f"yoga:{y.get('rule_id') or y.get('name') or 'unknown'}",
                    chart="D1",
                    conclusion=conclusion,
                    sentiment=sentiment,
                    strength=strength,
                    details={"source_module": "yoga", "raw": y},
                ))
                added_per_theme[theme] += 1

    def _inject_varga_modules(self, modules: Dict[str, Any], add) -> None:
        d9 = modules.get("d9_navamsa_expanded") or {}
        if isinstance(d9, dict) and d9:
            asc = d9.get("Ascendant") or {}
            venus = d9.get("Venus") or {}
            jupiter = d9.get("Jupiter") or {}
            parts = []
            if asc:
                parts.append(f"D9上升落{asc.get('sign_cn') or asc.get('sign')}，主星{asc.get('lord', '未知')}")
            if venus:
                parts.append(f"Venus在D9落{venus.get('sign_cn') or venus.get('sign')}，尊严={venus.get('dignity', '未知')}")
            if jupiter:
                parts.append(f"Jupiter在D9落{jupiter.get('sign_cn') or jupiter.get('sign')}，尊严={jupiter.get('dignity', '未知')}")
            add(ThemeName.MARRIAGE, ReportTechniqueResult(
                technique="d9_navamsa_expanded",
                chart="D9",
                conclusion="；".join(parts[:3]) or "D9 Navamsa 已完成展开",
                sentiment=_sentiment_from_dignities([venus.get("dignity"), jupiter.get("dignity")]),
                strength=StrengthLevel.STRONG,
                details={"source_module": "d9_navamsa_expanded", "ascendant": asc, "venus": venus, "jupiter": jupiter},
            ))

        varga = modules.get("varga_full") or {}
        if not isinstance(varga, dict):
            return
        mapping = [
            ("D2_Hora", ThemeName.WEALTH, "D2", "Hora财富分盘"),
            ("D10_Dasamsa", ThemeName.CAREER, "D10", "Dashamsha事业分盘"),
            ("D20_Vimsamsa", ThemeName.SPIRITUALITY, "D20", "Vimsamsa灵性分盘"),
            ("D30_Trimsamsa", ThemeName.HEALTH, "D30", "Trimsamsa健康/灾厄分盘"),
            ("D60_Shashtyamsa", ThemeName.SPIRITUALITY, "D60", "Shashtyamsha业力分盘"),
        ]
        for key, theme, chart, label in mapping:
            data = varga.get(key) or {}
            if not isinstance(data, dict) or not data:
                continue
            asc = data.get("Ascendant") or {}
            dignity = data.get("_dignity") or {}
            strong = [p for p, d in dignity.items() if str(d).lower() in {"exalted", "own sign", "moolatrikona"}]
            weak = [p for p, d in dignity.items() if "debil" in str(d).lower()]
            conclusion = f"{label}：上升落{asc.get('sign', '未知')}；强势行星{strong[:3] or '未突出'}；需留意行星{weak[:3] or '无明显落陷'}"
            add(theme, ReportTechniqueResult(
                technique=f"varga_full:{key}",
                chart=chart,
                conclusion=conclusion,
                sentiment="positive" if len(strong) > len(weak) else ("negative" if weak else "neutral"),
                strength=StrengthLevel.STRONG,
                details={"source_module": "varga_full", "varga": key, "ascendant": asc, "dignity": dignity},
            ))

    def _inject_strength_modules(self, modules: Dict[str, Any], add) -> None:
        shadbala = modules.get("shadbala") or {}
        if isinstance(shadbala, dict) and shadbala:
            strongest = shadbala.get("strongest") or []
            weakest = shadbala.get("weakest") or []
            conclusion = f"Shadbala相对力量：最强={strongest}；最弱={weakest}。该模块当前用于相对强弱排序。"
            result = ReportTechniqueResult(
                technique="shadbala_relative_strength",
                chart="D1",
                conclusion=conclusion,
                sentiment="neutral",
                strength=StrengthLevel.MODERATE,
                details={"source_module": "shadbala", "strongest": strongest, "weakest": weakest, "status": shadbala.get("status")},
            )
            for theme in [ThemeName.CAREER, ThemeName.HEALTH, ThemeName.SPIRITUALITY]:
                add(theme, result)

        av = modules.get("ashtakavarga") or {}
        if isinstance(av, dict) and av:
            house_scores = av.get("house_scores") or av.get("house_scores_full") or {}
            wealth_scores = _pick_house_scores(house_scores, [2, 11])
            health_scores = _pick_house_scores(house_scores, [6, 8, 12])
            career_scores = _pick_house_scores(house_scores, [10])
            for theme, houses, label in [
                (ThemeName.WEALTH, wealth_scores, "2/11宫财富"),
                (ThemeName.HEALTH, health_scores, "6/8/12宫健康压力"),
                (ThemeName.CAREER, career_scores, "10宫事业"),
            ]:
                if not houses:
                    continue
                avg = sum(houses.values()) / len(houses)
                add(theme, ReportTechniqueResult(
                    technique=f"ashtakavarga:{label}",
                    chart="AV",
                    conclusion=f"Ashtakavarga {label}分值：{houses}，均值约{avg:.1f}",
                    sentiment="positive" if avg >= 28 else ("negative" if avg <= 24 else "neutral"),
                    strength=StrengthLevel.MODERATE,
                    details={"source_module": "ashtakavarga", "scores": houses},
                ))

    def _inject_jaimini_relationship_modules(self, modules: Dict[str, Any], add) -> None:
        """Inject relationship evidence from Jaimini DK and RTN modules.

        v6.1.10 connects two previously underused P0 relationship layers:
        `modules.jaimini.darakaraka` and `modules.rashi_tulya_navamsa`.
        Both are deterministic full-reading outputs; the bridge translates them into
        marriage/health/spirituality evidence so thematic reports can narrate them.
        """
        jaimini = modules.get("jaimini") or {}
        darakaraka = jaimini.get("darakaraka") if isinstance(jaimini, dict) else None
        if isinstance(darakaraka, dict) and darakaraka and not darakaraka.get("error"):
            score = darakaraka.get("marriage_quality_score")
            score_text = f"，质量评分约{score:.0%}" if isinstance(score, (int, float)) else ""
            conclusion = (
                f"Darakaraka：{darakaraka.get('dk_planet', '未知')}落{darakaraka.get('dk_sign', '未知')}"
                f"第{darakaraka.get('dk_house', '?')}宫，伴侣原型={darakaraka.get('core_profile', '未标注')}{score_text}。"
            )
            if darakaraka.get("d9_sign"):
                conclusion += f" D9中落{darakaraka.get('d9_sign')}，尊严={darakaraka.get('d9_dignity', '未知')}。"
            add(ThemeName.MARRIAGE, ReportTechniqueResult(
                technique="jaimini:darakaraka_deep_reader",
                chart="D1/D9",
                conclusion=conclusion,
                sentiment="positive" if isinstance(score, (int, float)) and score >= 0.6 else ("negative" if isinstance(score, (int, float)) and score < 0.4 else "neutral"),
                strength=StrengthLevel.STRONG,
                details={"source_module": "jaimini.darakaraka", "raw": darakaraka},
            ))

        rtn_module = modules.get("rashi_tulya_navamsa") or {}
        rtn = rtn_module.get("analysis") if isinstance(rtn_module, dict) and isinstance(rtn_module.get("analysis"), dict) else rtn_module
        if isinstance(rtn, dict) and rtn and not rtn.get("error"):
            house_mapping = rtn.get("house_mapping") or {}
            curse_yogas = rtn.get("curse_yogas") or []
            strength_score = rtn.get("strength_score")
            weakness_score = rtn.get("weakness_score")
            marriage_planets = house_mapping.get(7) or house_mapping.get("7") or []
            summary_parts = []
            if marriage_planets:
                summary_parts.append(f"RTN第7宫映射行星={marriage_planets}")
            if curse_yogas:
                summary_parts.append(f"检测到{len(curse_yogas)}个凶星合相命名")
            if isinstance(strength_score, (int, float)) and isinstance(weakness_score, (int, float)):
                summary_parts.append(f"隐藏力量={strength_score:.0%}，弱点={weakness_score:.0%}")
            conclusion = "；".join(summary_parts) or "Rashi Tulya Navamsa 已完成 D1/D9 映射分析。"
            sentiment = "negative" if curse_yogas or (isinstance(weakness_score, (int, float)) and weakness_score > strength_score) else "positive"
            add(ThemeName.MARRIAGE, ReportTechniqueResult(
                technique="rashi_tulya_navamsa:relationship_mapping",
                chart="D1/D9",
                conclusion=conclusion,
                sentiment=sentiment,
                strength=StrengthLevel.MODERATE,
                details={"source_module": "rashi_tulya_navamsa", "raw": rtn},
            ))
            if curse_yogas:
                add(ThemeName.HEALTH, ReportTechniqueResult(
                    technique="rashi_tulya_navamsa:curse_yogas",
                    chart="D1/D9",
                    conclusion=f"RTN凶星合相提示健康/危机压力：{[c.get('name') for c in curse_yogas[:3]]}",
                    sentiment="negative",
                    strength=StrengthLevel.MODERATE,
                    details={"source_module": "rashi_tulya_navamsa", "curse_yogas": curse_yogas},
                ))
            guna = rtn.get("guna_narrative")
            if guna:
                add(ThemeName.SPIRITUALITY, ReportTechniqueResult(
                    technique="rashi_tulya_navamsa:guna_balance",
                    chart="D1/D9",
                    conclusion=f"RTN Gunas：{guna}",
                    sentiment="neutral",
                    strength=StrengthLevel.MODERATE,
                    details={"source_module": "rashi_tulya_navamsa", "guna_balance": rtn.get("guna_balance")},
                ))

    def _inject_special_points(self, modules: Dict[str, Any], add) -> None:
        special = modules.get("special_lagnas") or {}
        if isinstance(special, dict):
            upapada = special.get("Upapada_Lagna") or {}
            a10 = special.get("A10_Karma_Pada") or {}
            arudha = special.get("Arudha_Lagna") or {}
            if upapada:
                add(ThemeName.MARRIAGE, ReportTechniqueResult(
                    technique="special_lagnas:upapada_lagna",
                    chart="D1",
                    conclusion=f"Upapada Lagna落{upapada.get('sign', '未知')}第{upapada.get('house', '?')}宫，提示婚姻的社会呈现与伴侣形象。",
                    sentiment="neutral",
                    strength=StrengthLevel.STRONG,
                    details={"source_module": "special_lagnas", "upapada_lagna": upapada},
                ))
            if a10:
                add(ThemeName.CAREER, ReportTechniqueResult(
                    technique="special_lagnas:a10_karma_pada",
                    chart="D1",
                    conclusion=f"A10/Karma Pada落{a10.get('sign', '未知')}第{a10.get('house', '?')}宫，用于观察事业名声与业力方向。",
                    sentiment="neutral",
                    strength=StrengthLevel.STRONG,
                    details={"source_module": "special_lagnas", "a10_karma_pada": a10},
                ))
            if arudha:
                add(ThemeName.CAREER, ReportTechniqueResult(
                    technique="special_lagnas:arudha_lagna",
                    chart="D1",
                    conclusion=f"Arudha Lagna落{arudha.get('sign', '未知')}第{arudha.get('house', '?')}宫，显示公众形象与外界评价。",
                    sentiment="neutral",
                    strength=StrengthLevel.MODERATE,
                    details={"source_module": "special_lagnas", "arudha_lagna": arudha},
                ))

        vivah = modules.get("vivah_saham") or {}
        if isinstance(vivah, dict) and vivah:
            point = vivah.get("vivah_saham") or vivah
            add(ThemeName.MARRIAGE, ReportTechniqueResult(
                technique="vivah_saham",
                chart="Tajika",
                conclusion=f"Vivah Saham婚姻敏感点落{point.get('sign', vivah.get('saham_sign', '未知'))}第{point.get('house', vivah.get('saham_house', '?'))}宫；适合与Transit触发联合判断婚恋窗口。",
                sentiment="neutral",
                strength=StrengthLevel.MODERATE,
                details={"source_module": "vivah_saham", "raw": vivah},
            ))

    def _inject_dasa_convergence_module(self, modules: Dict[str, Any], add) -> None:
        convergence = modules.get("dasa_convergence") or {}
        if not isinstance(convergence, dict) or not convergence:
            return

        # 从 full-reading 的 Vimshottari current antar 注入真实时间锚，避免报告退回模拟 TimingAnchorBuilder。
        systems_summary = convergence.get("systems_summary") or {}
        vim = systems_summary.get("vimshottari") or {}
        antar = vim.get("antar") or {}
        if isinstance(antar, dict):
            start_year = _year_from_date(antar.get("start"), datetime.now().year)
            end_year = _year_from_date(antar.get("end"), start_year + 3)
            maha = vim.get("maha") or "Unknown"
            antar_lord = antar.get("lord") or "Unknown"
            anchor = TimingAnchor(
                dasha_period=f"Vimshottari-{maha}-{antar_lord}",
                start_year=start_year,
                end_year=end_year,
                activation_description="full-reading 五系统 Dasa Convergence 已完成，当前 Vimshottari 主/副运作为主题报告时间锚。",
                is_current=bool(antar.get("is_current", True)),
            )
            for tn in ThemeName:
                self.rpo.add_timing(tn, anchor)

        domain_map = {
            "marriage_partnership": ThemeName.MARRIAGE,
            "career_status": ThemeName.CAREER,
            "wealth_family": ThemeName.WEALTH,
            "health_service": ThemeName.HEALTH,
            "fortune_dharma": ThemeName.SPIRITUALITY,
            "transformation": ThemeName.SPIRITUALITY,
        }
        domains = convergence.get("domain_activations") or {}
        for domain, data in domains.items():
            theme = domain_map.get(domain)
            if not theme or not isinstance(data, dict):
                continue
            activations = data.get("activations") or []
            systems = sorted({a.get("system") for a in activations if isinstance(a, dict) and a.get("system")})
            conclusion = f"Dasa Convergence：{domain} 获得{data.get('system_count', len(systems))}个系统激活（{', '.join(systems) or '未列出'}），收敛等级{data.get('convergence_level', '未知')}，概率{data.get('probability', '未标注')}。"
            add(theme, ReportTechniqueResult(
                technique=f"dasa_convergence:{domain}",
                chart="Dasha",
                conclusion=conclusion,
                sentiment="positive" if data.get("system_count", 0) >= 2 else "neutral",
                strength=StrengthLevel.STRONG if data.get("system_count", 0) >= 3 else StrengthLevel.MODERATE,
                details={"source_module": "dasa_convergence", "domain": domain, "raw": data},
            ))

    def _inject_transit_module(self, modules: Dict[str, Any], add) -> None:
        transit = modules.get("transit_multi_reference") or {}
        if not isinstance(transit, dict) or not transit:
            return
        target_date = transit.get("target_date")
        analysis = transit.get("transit_analysis") or {}
        for planet in ["Jupiter", "Saturn", "Rahu", "Ketu"]:
            pdata = analysis.get(planet)
            if not isinstance(pdata, dict):
                continue
            house_map = pdata.get("house_from_ref") or {}
            lagna_house = (house_map.get("Lagna") or {}).get("house")
            if not lagna_house:
                continue
            themes = _themes_for_transit_house(int(lagna_house))
            for theme in themes:
                add(theme, ReportTechniqueResult(
                    technique=f"transit_multi_reference:{planet}",
                    chart="Transit",
                    conclusion=f"{target_date or '当前'}真实过境：{planet}从Lagna参考点落第{lagna_house}宫，并需结合Chandra/Arudha/Navamsa多参考点校验。",
                    sentiment="positive" if planet == "Jupiter" and lagna_house in [2, 5, 7, 9, 10, 11] else ("negative" if planet in ["Saturn", "Rahu", "Ketu"] and lagna_house in [6, 8, 12] else "neutral"),
                    strength=StrengthLevel.MODERATE,
                    details={"source_module": "transit_multi_reference", "planet": planet, "raw": pdata},
                ))

    # ── 端到端报告生成 ──

    def generate_full_report(
        self,
        chart_data: Dict[str, Any],
        themes: Optional[List[ReadingTheme]] = None,
    ) -> Dict[str, Any]:
        """
        端到端报告生成：从星盘数据 → ReadingChapter → ThemeReport。

        Args:
            chart_data: 星盘完整数据
            themes: 要分析的主题列表，None=全部5大主题

        Returns:
            {
                "reading_chapters": Dict[ReadingTheme, List[ReadingChapter]],
                "theme_reports": Dict[ThemeName, ThemeReport],
                "unified_narrative": str,
                "dasha_results": Dict[str, Any],
            }
        """
        if themes is None:
            themes = [
                ReadingTheme.MARRIAGE,
                ReadingTheme.CAREER,
                ReadingTheme.WEALTH,
                ReadingTheme.HEALTH,
                ReadingTheme.SPIRITUAL,
            ]

        # Step 0: 注入推运结果与 full-reading 真实模块（在所有主题分析之前）
        dasha_results = self.inject_dasha_results(chart_data)
        full_reading_injection = self.inject_full_reading_modules(chart_data)

        reading_chapters: Dict[str, List[ReadingChapter]] = {}
        theme_reports: Dict[str, Any] = {}

        for rt in themes:
            # Step 1: reading_orchestrator 执行技法
            chapters = self.ro.analyze(chart_data, rt)
            reading_chapters[rt.value] = chapters

            # Step 2: 映射到 ThemeName
            tn = THEME_MAPPING.get(rt)
            if not tn:
                continue

            # Step 3: 将 chapters 转换为 TechniqueResults 并喂给 report_orchestrator
            for ch in chapters:
                tech_results = self.chapter_to_technique_results(ch)
                for tr in tech_results:
                    self.rpo.add_technique(tn, tr)

            # Step 4: 生成 ThemeReport
            report = self.rpo.generate_report(tn)
            theme_reports[tn.value] = report.to_dict()

        # Step 5: 生成统一叙事
        unified = self._build_unified_narrative(theme_reports)

        return {
            "reading_chapters": {
                k: [self._chapter_to_dict(ch) for ch in v]
                for k, v in reading_chapters.items()
            },
            "theme_reports": theme_reports,
            "unified_narrative": unified,
            "dasha_results": dasha_results,
            "full_reading_injection": full_reading_injection,
        }

    # ── 内部辅助 ──

    @staticmethod
    def _build_unified_narrative(theme_reports: Dict[str, Any]) -> str:
        """将多个 ThemeReport 拼接为统一的人生叙事。"""
        parts = []
        order = ["marriage", "career", "wealth", "health", "spirituality"]
        for theme_key in order:
            if theme_key in theme_reports:
                r = theme_reports[theme_key]
                parts.append(f"\n## {r.get('summary', theme_key)}")
                parts.append(r.get("narrative", ""))
                if r.get("recommendations"):
                    parts.append("\n**建议：**")
                    for rec in r["recommendations"]:
                        parts.append(f"- {rec}")
        return "\n".join(parts)

    @staticmethod
    def _chapter_to_dict(ch: ReadingChapter) -> Dict[str, Any]:
        return {
            "title": ch.title,
            "subtitle": ch.subtitle,
            "priority": ch.priority,
            "techniques_used": ch.techniques_used,
            "findings": ch.findings,
            "conflicts": [
                {
                    "dimension": c.dimension,
                    "d1_finding": c.d1_finding,
                    "d9_finding": c.d9_finding,
                    "resolution": c.resolution,
                    "narrative": c.narrative,
                }
                for c in ch.conflicts
            ],
            "narrative": ch.narrative,
            "actionable": ch.actionable,
            "timing": ch.timing,
        }


# ═══════════════════════════════════════════════════════════════
# 辅助函数
# ═══════════════════════════════════════════════════════════════

def _extract_modules(chart_data: Any) -> Dict[str, Any]:
    """Return full-reading modules from either a full report dict or a modules dict."""
    if not isinstance(chart_data, dict):
        return {}
    modules = chart_data.get("modules")
    if isinstance(modules, dict):
        return modules
    # Allow passing modules directly in tests.
    known = {
        "yoga", "varga_full", "d9_navamsa_expanded", "ashtakavarga", "shadbala",
        "jaimini", "rashi_tulya_navamsa", "special_lagnas", "vivah_saham",
        "dasa_convergence", "transit_multi_reference",
        "dasha", "yogini_dasha", "ashtottari_dasha", "kalachakra_dasha",
    }
    if any(k in chart_data for k in known):
        return chart_data
    return {}


def _strength_from_text(value: Any, default: StrengthLevel = StrengthLevel.MODERATE) -> StrengthLevel:
    text = str(value or "").lower()
    if any(x in text for x in ["强", "strong", "high", "极"]):
        return StrengthLevel.STRONG
    if any(x in text for x in ["弱", "weak", "low"]):
        return StrengthLevel.WEAK
    return default


def _sentiment_from_yoga(yoga: Dict[str, Any]) -> str:
    category = str(yoga.get("category") or yoga.get("name") or yoga.get("name_cn") or "").lower()
    negative_tokens = ["darid", "dharidhra", "kapata", "arista", "dosha", "curse", "poverty", "roga", "mrityu", "凶", "贫", "病", "灾", "欺"]
    positive_tokens = ["raja", "dhana", "lakshmi", "gaja", "kesari", "amala", "yoga", "财富", "王", "吉", "成就"]
    if any(t in category for t in negative_tokens):
        return "negative"
    if any(t in category for t in positive_tokens):
        return "positive"
    return "neutral"


def _themes_for_yoga(yoga: Dict[str, Any]) -> List[ThemeName]:
    text = " ".join(str(yoga.get(k, "")) for k in ["name", "name_cn", "category", "rule_id", "combination"])
    effects = yoga.get("effects") or []
    if isinstance(effects, list):
        text += " " + " ".join(str(e) for e in effects)
    lower = text.lower()
    themes: List[ThemeName] = []
    keyword_map = [
        (ThemeName.WEALTH, ["dhana", "lakshmi", "wealth", "money", "income", "财富", "财", "收入", "资源"]),
        (ThemeName.CAREER, ["raja", "career", "profession", "status", "authority", "power", "事业", "权力", "地位", "名声", "成就"]),
        (ThemeName.MARRIAGE, ["marriage", "spouse", "venus", "vivah", "partner", "婚", "配偶", "伴侣", "金星"]),
        (ThemeName.HEALTH, ["arista", "roga", "disease", "health", "illness", "saturn-mars", "病", "健康", "灾", "凶"]),
        (ThemeName.SPIRITUALITY, ["moksha", "jupiter", "ketu", "spiritual", "dharma", "智慧", "灵性", "木星", "解脱"]),
    ]
    for theme, keywords in keyword_map:
        if any(k in lower or k in text for k in keywords):
            themes.append(theme)
    if not themes and yoga.get("category") in ["raja", "bvr"]:
        themes.append(ThemeName.CAREER)
    return themes[:3]


def _sentiment_from_dignities(values: List[Any]) -> str:
    text = " ".join(str(v or "").lower() for v in values)
    if any(k in text for k in ["exalted", "own", "moola", "friend"]):
        return "positive"
    if any(k in text for k in ["debil", "enemy"]):
        return "negative"
    return "neutral"


def _pick_house_scores(house_scores: Any, houses: List[int]) -> Dict[int, float]:
    if not isinstance(house_scores, dict):
        return {}
    picked: Dict[int, float] = {}
    for house in houses:
        value = None
        for key in [house, str(house), f"house_{house}", f"H{house}"]:
            if key in house_scores:
                value = house_scores[key]
                break
        if isinstance(value, dict):
            value = value.get("score") or value.get("bindus") or value.get("sav")
        if isinstance(value, (int, float)):
            picked[house] = float(value)
    return picked


def _year_from_date(value: Any, default: int) -> int:
    if isinstance(value, int):
        return value
    if isinstance(value, str) and len(value) >= 4:
        try:
            return int(value[:4])
        except ValueError:
            return default
    return default


def _themes_for_transit_house(house: int) -> List[ThemeName]:
    mapping = {
        1: [ThemeName.HEALTH],
        2: [ThemeName.WEALTH],
        5: [ThemeName.WEALTH, ThemeName.SPIRITUALITY],
        6: [ThemeName.HEALTH, ThemeName.CAREER],
        7: [ThemeName.MARRIAGE],
        8: [ThemeName.HEALTH, ThemeName.SPIRITUALITY],
        9: [ThemeName.SPIRITUALITY, ThemeName.CAREER],
        10: [ThemeName.CAREER],
        11: [ThemeName.WEALTH, ThemeName.CAREER],
        12: [ThemeName.HEALTH, ThemeName.SPIRITUALITY],
    }
    return mapping.get(house, [])


def _infer_sentiment(text: str) -> str:
    """从 finding 文本推断 sentiment。"""
    negative_keywords = [
        "阻碍", "困难", "挑战", "风险", "不利", "凶", "克", "弱", "陷",
        "冲突", "矛盾", "破坏", "损失", "障碍", "延迟", "问题",
    ]
    positive_keywords = [
        "有利", "吉", "旺", "强", "助力", "支持", "机遇", "突破",
        "成就", "成功", "和谐", "稳定", "增益", "提升",
    ]
    text_lower = text.lower()
    neg_score = sum(1 for w in negative_keywords if w in text_lower)
    pos_score = sum(1 for w in positive_keywords if w in text_lower)
    if neg_score > pos_score:
        return "negative"
    if pos_score > neg_score:
        return "positive"
    return "neutral"


def _infer_strength(text: str, conflicts: List[Any]) -> StrengthLevel:
    """从 finding 文本和冲突情况推断 strength。"""
    strong_keywords = ["非常", "极强", "显著", "明确", "主导", "绝对"]
    weak_keywords = ["轻微", "略有", "潜在", "可能", "模糊", "微弱"]
    text_lower = text.lower()
    if any(w in text_lower for w in strong_keywords):
        return StrengthLevel.STRONG
    if any(w in text_lower for w in weak_keywords):
        return StrengthLevel.WEAK
    if conflicts:
        return StrengthLevel.MODERATE
    return StrengthLevel.MODERATE


# ═══════════════════════════════════════════════════════════════
# CLI / 测试
# ═══════════════════════════════════════════════════════════════

def demo():
    """演示桥接功能。

    默认使用 MockDataFactory；如果传入一个 full-reading JSON 路径，则直接消费
    `modules` 真实结果，验证 v6.1.8 的真实模块接线。
    """
    import sys
    from report_orchestrator import MockDataFactory

    # 创建星盘数据：优先使用 full-reading JSON，否则使用模拟数据
    if len(sys.argv) > 1:
        with open(sys.argv[1], "r", encoding="utf-8") as f:
            chart_data = json.load(f)
    else:
        chart_data = MockDataFactory.create_sample_chart()

    # 创建 orchestrators
    # NOTE: reading_orchestrator 需要 registry，这里用空 registry 演示；真实证据由 full-reading 注入层提供
    class DummyRegistry:
        def get(self, name):
            return None

    ro = ReadingOrchestrator(DummyRegistry())
    rpo = ReportOrchestrator(chart_data)

    # 桥接
    bridge = OrchestratorBridge(ro, rpo)
    result = bridge.generate_full_report(chart_data)

    print("=" * 60)
    print("Orchestrator Bridge Demo")
    print("=" * 60)
    print(f"\n生成主题报告数: {len(result['theme_reports'])}")
    for theme, report in result["theme_reports"].items():
        print(f"\n  [{theme}]")
        print(f"    总结: {report.get('summary', 'N/A')}")
        print(f"    强度: {report.get('strength', 'N/A')}")
        print(f"    证据数: {len(report.get('evidence', []))}")

    print("\n" + "=" * 60)
    print("统一叙事（前500字）:")
    print("=" * 60)
    print(result["unified_narrative"][:500] + "...")


if __name__ == "__main__":
    demo()