From 067b400625a8cd73c4c5a3080fea6fa5f3a60cb7 Mon Sep 17 00:00:00 2001 From: 732642856 <732642856@qq.com> Date: Sun, 7 Jun 2026 15:17:09 +0800 Subject: [PATCH] v6.1.5: Integrate Ashtottari/Yogini/Kalachakra Dasha into report pipeline - Register 3 dasha systems in technique_registry.json - Extend orchestrator_bridge.py with inject_dasha_results() - Add add_timing() to ThematicReportOrchestrator - Dashas auto-convert to TimingAnchor and inject into all 5 themes - Verified: Yogini + Kalachakra timing anchors flow into ThemeReport --- references/technique_registry.json | 60 +++++++ scripts/orchestrator_bridge.py | 245 +++++++++++++++++++++++++++++ scripts/report_orchestrator.py | 15 +- 3 files changed, 318 insertions(+), 2 deletions(-) diff --git a/references/technique_registry.json b/references/technique_registry.json index b4a74488..8d574f02 100644 --- a/references/technique_registry.json +++ b/references/technique_registry.json @@ -111,6 +111,66 @@ "audit_label": "Vimshottari", "missing_impact": "Primary timing layer unavailable." }, + "ashtottari_dasha": { + "name": "Ashtottari Dasha / 108年周期", + "domains": [ + "timing", + "event", + "nakshatra" + ], + "status": "covered", + "knowledge_refs": [ + "references/alternative-dasha-systems.md" + ], + "commands": [ + "full-reading" + ], + "output_paths": [ + "modules.ashtottari_dasha" + ], + "audit_label": "Ashtottari", + "missing_impact": "Nakshatra-based conditional timing layer unavailable for qualifying charts." + }, + "yogini_dasha": { + "name": "Yogini Dasha / 36年周期", + "domains": [ + "timing", + "event", + "yogini" + ], + "status": "covered", + "knowledge_refs": [ + "references/alternative-dasha-systems.md" + ], + "commands": [ + "full-reading" + ], + "output_paths": [ + "modules.yogini_dasha" + ], + "audit_label": "Yogini", + "missing_impact": "Lagna-based 36-year timing layer unavailable." + }, + "kalachakra_dasha": { + "name": "Kalachakra Dasha / 时轮大运", + "domains": [ + "timing", + "event", + "nakshatra" + ], + "status": "covered", + "knowledge_refs": [ + "references/alternative-dasha-systems.md" + ], + "commands": [ + "full-reading" + ], + "output_paths": [ + "modules.kalachakra_dasha" + ], + "audit_label": "Kalachakra", + "missing_impact": "Savya/Apasavya dual-mode timing layer unavailable." + }, "dasha_sandhi": { "name": "Dasha Sandhi", "domains": [ diff --git a/scripts/orchestrator_bridge.py b/scripts/orchestrator_bridge.py index b327f756..1aeac844 100644 --- a/scripts/orchestrator_bridge.py +++ b/scripts/orchestrator_bridge.py @@ -15,8 +15,19 @@ orchestrator_bridge.py — 编排器桥接层 """ 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 ( @@ -136,6 +147,235 @@ class OrchestratorBridge: 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: + pass + + # 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: + pass + + # 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: + pass + + # 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) + # ── 端到端报告生成 ── def generate_full_report( @@ -155,6 +395,7 @@ class OrchestratorBridge: "reading_chapters": Dict[ReadingTheme, List[ReadingChapter]], "theme_reports": Dict[ThemeName, ThemeReport], "unified_narrative": str, + "dasha_results": Dict[str, Any], } """ if themes is None: @@ -166,6 +407,9 @@ class OrchestratorBridge: ReadingTheme.SPIRITUAL, ] + # Step 0: 注入推运结果（在所有主题分析之前） + dasha_results = self.inject_dasha_results(chart_data) + reading_chapters: Dict[str, List[ReadingChapter]] = {} theme_reports: Dict[str, Any] = {} @@ -199,6 +443,7 @@ class OrchestratorBridge: }, "theme_reports": theme_reports, "unified_narrative": unified, + "dasha_results": dasha_results, } # ── 内部辅助 ── diff --git a/scripts/report_orchestrator.py b/scripts/report_orchestrator.py index 9aacb3b4..e8fc8c10 100644 --- a/scripts/report_orchestrator.py +++ b/scripts/report_orchestrator.py @@ -629,6 +629,9 @@ class ThematicReportOrchestrator: t: NarrativeGenerator(t) for t in ThemeName } self._timing_builder = TimingAnchorBuilder(chart_data) + self._external_timings: Dict[ThemeName, List[TimingAnchor]] = { + t: [] for t in ThemeName + } def add_technique(self, theme: ThemeName, result: TechniqueResult) -> None: """向指定主题添加一个技法分析结果""" @@ -640,6 +643,10 @@ class ThematicReportOrchestrator: for r in results: self.add_technique(theme, r) + def add_timing(self, theme: ThemeName, timing: TimingAnchor) -> None: + """向指定主题添加外部注入的时间锚定（如推运结果）""" + self._external_timings[theme].append(timing) + def generate_report(self, theme: ThemeName) -> ThemeReport: """生成单一主题完整报告""" validator = self._validators[theme] @@ -650,8 +657,12 @@ class ThematicReportOrchestrator: strength = validator.compute_overall_strength() dominant = validator.get_dominant_sentiment() - # 2. 时间锚定 - timing = self._timing_builder.build_for_theme(theme) + # 2. 时间锚定（优先使用外部注入的推运结果） + if self._external_timings.get(theme): + # 使用最新注入的外部timing + timing = self._external_timings[theme][-1] + else: + timing = self._timing_builder.build_for_theme(theme) # 3. 叙事生成 narrator = self._narrators[theme]