Files
Jyotisha/scripts/reading_orchestrator.py
732642856 0f0c5dbdfa v6.1.0: Add 6 core missing modules + infrastructure
New modules (from Chinese reference articles):
- darakaraka_reader.py: DK spouse deep analysis (planet/sign/house/aspects)
- rashi_tulya_navamsa.py: RTN mapping engine (D9->D1, Gunas, curse detection)
- curse_yoga_detector.py: Curse yoga naming (Yama/Preta/Rakshasa/Pisacha)
- spouse_status_yoga.py: High-status spouse detection (7 principles)
- pancha_pakshi.py: Five-bird electional timing (daily schedule)
- tithi_analyzer.py: Tithi lord emotional pattern analysis

Infrastructure:
- technique_registry.py: Unified technique registry (31 techniques tracked)
- reading_orchestrator.py: Theme-based reading orchestrator (5 themes)

All 6 modules tested and runnable.
Current baseline: Precision 94.66%, Recall 92.95%, F1 93.80%
2026-06-07 12:11:15 +08:00

657 lines
24 KiB
Python
Raw Permalink Blame History

This file contains ambiguous Unicode characters
This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
reading_orchestrator.py — 主题化解读编排器
===========================================
"模块罗列"升级为"主题叙事"
核心能力:
1. 主题路由 — 根据用户问题选择分析主题
2. 技法编排 — 按优先级调用相关技法
3. 交叉验证 — 多维度矛盾裁决
4. 叙事生成 — 从数据生成连贯的人生故事
版本: v1.0 | 2026-06-07
"""
from typing import Dict, List, Any, Optional
from dataclasses import dataclass, field
from enum import Enum
import json
class ReadingTheme(Enum):
"""解读主题"""
OVERVIEW = "overview" # 生命概览
MARRIAGE = "marriage" # 婚姻情感
CAREER = "career" # 事业财富
HEALTH = "health" # 健康危机
SPIRITUAL = "spiritual" # 灵性成长
WEALTH = "wealth" # 财富积累
TIMING = "timing" # 时间预测
RELATIONSHIP = "relationship" # 人际关系
EDUCATION = "education" # 教育学习
FAMILY = "family" # 家庭父母
class ConflictResolution(Enum):
"""矛盾裁决策略"""
D9_OVERRIDES = "d9_overrides" # D9优先(婚姻/灵性)
D1_OVERRIDES = "d1_overrides" # D1优先(世俗事务)
STRONGEST_WINS = "strongest_wins" # 最强力量优先
CONTEXTUAL = "contextual" # 上下文叙事(推荐)
ALL_VALID = "all_valid" # 全部保留,标注差异
@dataclass
class ThemeConfig:
"""主题配置"""
theme: ReadingTheme
title: str # 主题标题
description: str # 主题描述
technique_chain: List[str] # 技法调用链
primary_varga: List[int] # 主要分盘
conflict_strategy: ConflictResolution # 矛盾裁决策略
narrative_template: str # 叙事模板
@dataclass
class TechniqueResult:
"""技法计算结果包装"""
technique_name: str
technique_name_cn: str
status: str # success / warning / error
result: Dict[str, Any]
interpretation: Optional[str] = None
confidence: float = 0.5 # 置信度 0-1
weight: float = 1.0 # 在主题中的权重
@dataclass
class ConflictItem:
"""矛盾项"""
dimension: str # 维度(如"婚姻质量"
d1_finding: str # D1发现
d9_finding: str # D9发现
other_findings: Dict[str, str] # 其他分盘发现
resolution: str # 裁决结果
narrative: str # 叙事表达
@dataclass
class ReadingChapter:
"""报告章节"""
title: str
subtitle: str
priority: int # 优先级(1最高)
techniques_used: List[str] # 使用的技法
findings: List[str] # 核心发现
conflicts: List[ConflictItem] # 矛盾与裁决
narrative: str # 叙事文本
actionable: List[str] # 行动建议
timing: Optional[str] = None # 时间锚定
class ReadingOrchestrator:
"""解读编排器"""
def __init__(self, registry):
self.registry = registry
self._themes: Dict[ReadingTheme, ThemeConfig] = self._init_themes()
def _init_themes(self) -> Dict[ReadingTheme, ThemeConfig]:
"""初始化主题配置"""
return {
ReadingTheme.OVERVIEW: ThemeConfig(
theme=ReadingTheme.OVERVIEW,
title="生命蓝图概览",
description="从全局视角解读先天格局",
technique_chain=[
"chart", "shadbala", "yoga", "karaka",
"nakshatra", "vimsopaka", "avastha"
],
primary_varga=[1],
conflict_strategy=ConflictResolution.ALL_VALID,
narrative_template="overview"
),
ReadingTheme.MARRIAGE: ThemeConfig(
theme=ReadingTheme.MARRIAGE,
title="婚姻与情感深度解析",
description="从D1、D9、DK、Upapada多维度解读婚姻质量",
technique_chain=[
"darakaraka",
"spouse_status_yoga",
"navamsa_analysis",
"rtn_mapping",
"vivah_saham",
"tithi_lord",
"mangal_dosha",
"synastry",
"upapada_lagna",
"dasha_marriage",
"transit_marriage"
],
primary_varga=[1, 9],
conflict_strategy=ConflictResolution.D9_OVERRIDES,
narrative_template="marriage"
),
ReadingTheme.CAREER: ThemeConfig(
theme=ReadingTheme.CAREER,
title="事业与社会成就",
description="从D1、D10、AmK多维度解读事业轨迹",
technique_chain=[
"amatyakaraka",
"dashamsa_analysis",
"d10_yoga",
"shadbala",
"ashtakavarga",
"tajika_varshaphala",
"dasha_career",
"transit_career"
],
primary_varga=[1, 10],
conflict_strategy=ConflictResolution.D1_OVERRIDES,
narrative_template="career"
),
ReadingTheme.HEALTH: ThemeConfig(
theme=ReadingTheme.HEALTH,
title="健康与生命危机",
description="从D1、D30、凶星合相检测健康风险",
technique_chain=[
"curse_yoga",
"trimshamsa",
"shadbala",
"avastha",
"drekkena",
"dasha_health",
"transit_health"
],
primary_varga=[1, 30],
conflict_strategy=ConflictResolution.STRONGEST_WINS,
narrative_template="health"
),
ReadingTheme.WEALTH: ThemeConfig(
theme=ReadingTheme.WEALTH,
title="财富与资源",
description="从D1、D2、Dhana Yoga解读财富格局",
technique_chain=[
"dhana_yoga",
"hora_analysis",
"dhanakaraka",
"vasihikamsa",
"ashtakavarga",
"dasha_wealth"
],
primary_varga=[1, 2],
conflict_strategy=ConflictResolution.D1_OVERRIDES,
narrative_template="wealth"
),
ReadingTheme.SPIRITUAL: ThemeConfig(
theme=ReadingTheme.SPIRITUAL,
title="灵性与灵魂成长",
description="从AK、D20、Karakamsha解读灵性道路",
technique_chain=[
"atmakaraka",
"karakamsha",
"vimsamsa",
"chara_dasha",
"nakshatra_deity",
"dasha_spiritual"
],
primary_varga=[1, 9, 20],
conflict_strategy=ConflictResolution.D9_OVERRIDES,
narrative_template="spiritual"
),
ReadingTheme.TIMING: ThemeConfig(
theme=ReadingTheme.TIMING,
title="时间预测",
description="Dasha + Transit精准时间锚定",
technique_chain=[
"dasha",
"transit",
"double_transit",
"tajika",
"gochara"
],
primary_varga=[1],
conflict_strategy=ConflictResolution.CONTEXTUAL,
narrative_template="timing"
),
}
def analyze(self, chart_data: Dict, theme: ReadingTheme,
question: Optional[str] = None) -> List[ReadingChapter]:
"""
执行主题化分析
Args:
chart_data: 星盘数据
theme: 分析主题
question: 用户具体问题(可选)
Returns:
章节列表
"""
config = self._themes.get(theme)
if not config:
return []
chapters = []
# Step 1: 执行技法链
technique_results = []
for tech_name in config.technique_chain:
result = self._execute_technique(tech_name, chart_data)
if result:
technique_results.append(result)
# Step 2: 生成核心发现
findings = self._extract_findings(technique_results, config)
# Step 3: 检测矛盾
conflicts = self._detect_conflicts(technique_results, config)
# Step 4: 裁决矛盾
resolved_conflicts = self._resolve_conflicts(conflicts, config)
# Step 5: 生成叙事
narrative = self._generate_narrative(findings, resolved_conflicts, config, question)
# Step 6: 生成行动建议
actionable = self._generate_actionable(findings, resolved_conflicts, config)
# Step 7: 时间锚定
timing = self._generate_timing(technique_results, config)
# 组装章节
chapter = ReadingChapter(
title=config.title,
subtitle=config.description,
priority=1,
techniques_used=[r.technique_name for r in technique_results if r.status == "success"],
findings=findings,
conflicts=resolved_conflicts,
narrative=narrative,
actionable=actionable,
timing=timing
)
chapters.append(chapter)
return chapters
def _execute_technique(self, name: str, chart_data: Dict) -> Optional[TechniqueResult]:
"""执行单个技法"""
spec = self.registry.get(name)
if not spec:
return None
# 如果技法状态是PLANNED,返回占位
if spec.status.value == "planned":
return TechniqueResult(
technique_name=name,
technique_name_cn=spec.name_cn,
status="planned",
result={},
interpretation=f"[{spec.name_cn}] 技法开发中,将在后续版本中提供。",
confidence=0.0
)
# 执行计算
try:
result = self.registry.compute(name, chart_data)
if result is None:
return None
# 获取解读
interpretation = self.registry.interpret(name, result)
# 评估置信度
confidence = self._assess_confidence(result, spec)
return TechniqueResult(
technique_name=name,
technique_name_cn=spec.name_cn,
status="success",
result=result,
interpretation=interpretation,
confidence=confidence
)
except Exception as e:
return TechniqueResult(
technique_name=name,
technique_name_cn=spec.name_cn if spec else name,
status="error",
result={"error": str(e)},
confidence=0.0
)
def _assess_confidence(self, result: Dict, spec) -> float:
"""评估结果置信度"""
# 基础置信度
confidence = 0.5
# 如果有明确检测结果,提高置信度
if result.get("detected") or result.get("present"):
confidence += 0.2
# 如果有多个支持证据
evidence_count = len(result.get("evidence", []))
confidence += min(evidence_count * 0.05, 0.2)
# 如果技法是STABLE状态
if spec and spec.status.value == "stable":
confidence += 0.1
return min(confidence, 1.0)
def _extract_findings(self, results: List[TechniqueResult],
config: ThemeConfig) -> List[str]:
"""提取核心发现"""
findings = []
for r in results:
if r.status != "success":
continue
# 从结果中提取关键发现
result = r.result
# Yoga检测类
if "yogas" in result:
yogas = result["yogas"]
if yogas:
findings.append(f"检测到 {len(yogas)} 个相关Yoga: {', '.join([y.get('name','') for y in yogas[:3]])}")
# 力量评估类
if "score" in result:
score = result["score"]
if isinstance(score, (int, float)) and score > 0.7:
findings.append(f"{r.technique_name_cn}得分较高 ({score:.2f}),显示此领域有显著影响")
# 相位/合相类
if "conjunctions" in result:
conj = result["conjunctions"]
if conj:
findings.append(f"发现关键合相: {', '.join([str(c) for c in conj[:2]])}")
# 解读文本
if r.interpretation:
findings.append(r.interpretation)
# 去重并排序(按置信度)
seen = set()
unique = []
for f in findings:
if f and f not in seen:
seen.add(f)
unique.append(f)
return unique[:7] # 最多7条核心发现
def _detect_conflicts(self, results: List[TechniqueResult],
config: ThemeConfig) -> List[ConflictItem]:
"""检测多维度矛盾"""
conflicts = []
# 按分盘分组
d1_results = [r for r in results if "d1" in r.technique_name or r.technique_name in ["chart", "yoga"]]
d9_results = [r for r in results if "navamsa" in r.technique_name or r.technique_name == "rtn_mapping"]
# 简单示例:D1和D9的婚姻判断矛盾
d1_marriage = self._extract_marriage_signal(d1_results)
d9_marriage = self._extract_marriage_signal(d9_results)
if d1_marriage and d9_marriage and d1_marriage != d9_marriage:
conflicts.append(ConflictItem(
dimension="婚姻质量",
d1_finding=d1_marriage,
d9_finding=d9_marriage,
other_findings={},
resolution="待裁决",
narrative=""
))
return conflicts
def _extract_marriage_signal(self, results: List[TechniqueResult]) -> str:
"""从结果中提取婚姻信号"""
# 简化实现
positive = sum(1 for r in results if r.confidence > 0.6)
negative = sum(1 for r in results if r.confidence < 0.4)
if positive > negative:
return "有利"
elif negative > positive:
return "有挑战"
return "中性"
def _resolve_conflicts(self, conflicts: List[ConflictItem],
config: ThemeConfig) -> List[ConflictItem]:
"""裁决矛盾"""
resolved = []
for c in conflicts:
strategy = config.conflict_strategy
if strategy == ConflictResolution.D9_OVERRIDES:
winner = c.d9_finding
reason = "Navamsa (D9) 在婚姻分析中具有最高权威"
elif strategy == ConflictResolution.D1_OVERRIDES:
winner = c.d1_finding
reason = "本命盘 (D1) 在世俗事务中具有最高权威"
elif strategy == ConflictResolution.STRONGEST_WINS:
# 这里简化处理,实际应比较力量
winner = c.d9_finding if len(c.d9_finding) > len(c.d1_finding) else c.d1_finding
reason = "基于力量评估的综合判断"
elif strategy == ConflictResolution.CONTEXTUAL:
winner = f"表面{c.d1_finding},但深层{c.d9_finding}"
reason = "需要结合具体情境综合判断"
else:
winner = f"D1: {c.d1_finding}, D9: {c.d9_finding}"
reason = "两种可能性并存"
c.resolution = winner
c.narrative = self._conflict_to_narrative(c, reason)
resolved.append(c)
return resolved
def _conflict_to_narrative(self, conflict: ConflictItem, reason: str) -> str:
"""将矛盾转换为叙事"""
if conflict.dimension == "婚姻质量":
if "有利" in conflict.resolution and "挑战" in conflict.resolution:
return f"婚姻表面看起来顺利,但深层存在需要关注的议题。{reason}"
return f"{conflict.resolution}{reason}"
return f"{conflict.resolution}{reason}"
def _generate_narrative(self, findings: List[str],
conflicts: List[ConflictItem],
config: ThemeConfig,
question: Optional[str]) -> str:
"""生成主题叙事文本"""
parts = []
# 开场
parts.append(f"## {config.title}\n")
parts.append(f"{config.description}\n")
# 核心发现
if findings:
parts.append("### 核心发现\n")
for i, f in enumerate(findings, 1):
parts.append(f"{i}. {f}\n")
# 矛盾与裁决
if conflicts:
parts.append("\n### 深层洞察\n")
for c in conflicts:
parts.append(f"\n{c.narrative}\n")
# 回应问题
if question:
parts.append(f"\n### 关于您的问题\n")
parts.append(f"针对「{question}」,综合以上分析:\n")
# 这里可以接入AI生成更具体的回答
return "\n".join(parts)
def _generate_actionable(self, findings: List[str],
conflicts: List[ConflictItem],
config: ThemeConfig) -> List[str]:
"""生成行动建议"""
actionable = []
if config.theme == ReadingTheme.MARRIAGE:
actionable.extend([
"关注Navamsa中7宫主星的状态,这是婚姻质量的核心指标",
"在DK激活的Dasha周期特别注意关系维护",
"如果Mangal Dosha存在,考虑传统补救措施"
])
elif config.theme == ReadingTheme.CAREER:
actionable.extend([
"在AmK相关的Dasha周期积极寻求事业突破",
"关注D10中10宫主星与Lagna的相位",
"利用Double Transit激活期推进重要项目"
])
elif config.theme == ReadingTheme.HEALTH:
actionable.extend([
"定期进行健康检查,特别是6宫和8宫相关的身体部位",
"在凶星Dasha周期特别注意安全",
"考虑传统 Ayurveda 调理"
])
return actionable
def _generate_timing(self, results: List[TechniqueResult],
config: ThemeConfig) -> Optional[str]:
"""生成时间锚定"""
# 从Dasha和Transit结果中提取时间信息
timing_parts = []
for r in results:
if "dasha" in r.technique_name and r.result:
current = r.result.get("current", {})
if current:
timing_parts.append(f"当前处于 {current.get('lord', '?')} 大运")
if timing_parts:
return "; ".join(timing_parts)
return None
def generate_full_report(self, chart_data: Dict,
themes: Optional[List[ReadingTheme]] = None) -> Dict:
"""
生成完整报告
Args:
chart_data: 星盘数据
themes: 要分析的主题列表(None=全部)
Returns:
结构化报告
"""
if themes is None:
themes = [ReadingTheme.OVERVIEW, ReadingTheme.MARRIAGE,
ReadingTheme.CAREER, ReadingTheme.HEALTH]
report = {
"version": "2.0-theme-based",
"themes_analyzed": [],
"chapters": [],
"summary": "",
"missing_techniques": []
}
for theme in themes:
chapters = self.analyze(chart_data, theme)
report["chapters"].extend(chapters)
report["themes_analyzed"].append(theme.value)
# 记录缺失的技法
config = self._themes.get(theme)
if config:
for tech_name in config.technique_chain:
spec = self.registry.get(tech_name)
if spec and spec.status.value == "planned":
report["missing_techniques"].append({
"name": tech_name,
"name_cn": spec.name_cn,
"theme": theme.value
})
# 去重缺失技法
seen = set()
unique_missing = []
for m in report["missing_techniques"]:
if m["name"] not in seen:
seen.add(m["name"])
unique_missing.append(m)
report["missing_techniques"] = unique_missing
# 生成总摘要
report["summary"] = self._generate_summary(report["chapters"])
return report
def _generate_summary(self, chapters: List[ReadingChapter]) -> str:
"""生成报告总摘要"""
parts = ["# 印度占星综合解读报告\n"]
for ch in chapters:
parts.append(f"\n## {ch.title}\n")
if ch.findings:
parts.append("关键发现: " + "; ".join(ch.findings[:3]) + "\n")
if ch.timing:
parts.append(f"时间锚定: {ch.timing}\n")
return "\n".join(parts)
# ============================================================================
# CLI 调试
# ============================================================================
if __name__ == "__main__":
from technique_registry import registry
print("=" * 60)
print("主题化解读编排器")
print("=" * 60)
orchestrator = ReadingOrchestrator(registry)
# 模拟星盘数据
mock_chart = {
"ascendant": {"sign": "Aries", "degree": 15.5},
"planets": {
"Sun": {"sign": "Aries", "degree": 20.0, "house": 1},
"Moon": {"sign": "Taurus", "degree": 10.0, "house": 2},
}
}
# 生成婚姻主题报告
print("\n--- 婚姻主题分析 ---\n")
chapters = orchestrator.analyze(mock_chart, ReadingTheme.MARRIAGE)
for ch in chapters:
print(ch.narrative)
print(f"\n使用技法: {', '.join(ch.techniques_used)}")
print(f"行动建议: {ch.actionable}")
# 生成完整报告
print("\n" + "=" * 60)
print("完整报告预览")
print("=" * 60)
report = orchestrator.generate_full_report(mock_chart)
print(f"\n分析主题: {report['themes_analyzed']}")
print(f"章节数: {len(report['chapters'])}")
print(f"\n待开发技法:")
for m in report['missing_techniques'][:10]:
print(f" - {m['name_cn']} ({m['name']})")