""" 紫微斗数四层Schema组装：meta / strength / timing / evidence。 只追加新字段，不改动任何旧字段——见 API_UNIFICATION_PLAN.md §3。 调用方：app.py 的 v4系列endpoint (analyze_ziwei_knowledge_enhanced / analyze_ziwei_v4)。 """ from datetime import datetime from typing import Any, Dict, List, Optional from terms import star_term, palace_term, brightness_term, mutagen_term, star_with_brightness WEAK_BRIGHTNESS = {"不", "陷"} GAN = ["甲", "乙", "丙", "丁", "戊", "己", "庚", "辛", "壬", "癸"] ZHI = ["子", "丑", "寅", "卯", "辰", "巳", "午", "未", "申", "酉", "戌", "亥"] # 十天干四化表（三合派通行版，禄/权/科/忌顺序）。 # 与iztro horoscope()的流年四化输出已用2026丙午/2027丁未/2028戊申/2029己酉 # 四个年份对照验证一致（见API_UNIFICATION_PLAN.md §2.1 golden基线）—— # 因此流年四化不需要动Node子进程，查表+本命盘落宫即可。 FLOW_SIHUA = { "甲": ("廉贞", "破军", "武曲", "太阳"), "乙": ("天机", "天梁", "紫微", "太阴"), "丙": ("天同", "天机", "文昌", "廉贞"), "丁": ("太阴", "天同", "天机", "巨门"), "戊": ("贪狼", "太阴", "右弼", "天机"), "己": ("武曲", "贪狼", "天梁", "文曲"), "庚": ("太阳", "武曲", "太阴", "天同"), "辛": ("巨门", "太阳", "文曲", "文昌"), "壬": ("天梁", "紫微", "左辅", "武曲"), "癸": ("破军", "巨门", "太阴", "贪狼"), } def _year_ganzhi(year: int) -> str: """公历年→干支（立春边界忽略：流年四化按整年粒度）""" return GAN[(year - 4) % 10] + ZHI[(year - 4) % 12] def build_flow_year_sihua(palaces: List[Dict[str, Any]], year: Optional[int] = None) -> Optional[Dict[str, Any]]: """ 流年四化（2026-07-13补齐Phase 3规范缺口"查询年流年四化"）： 流年天干查FLOW_SIHUA表，四化星的落宫/亮度取自本命盘palaces—— 流年四化星落本命某宫 = 该宫被这一年引动，是标准判读方式。 """ if year is None: year = datetime.now().year gz = _year_ganzhi(year) stars = FLOW_SIHUA.get(gz[0]) if not stars: return None star_loc: Dict[str, Dict[str, str]] = {} for p in palaces: for group in ("major_stars", "minor_stars"): for s in p.get(group) or []: if s.get("name"): star_loc[s["name"]] = { "palace": p.get("name", ""), "brightness": s.get("brightness", ""), } transformations = {} for label, star in zip(("化禄", "化权", "化科", "化忌"), stars): loc = star_loc.get(star, {}) transformations[label] = { "star": star, "palace": loc.get("palace", ""), "brightness": loc.get("brightness", ""), } return {"year": year, "year_ganzhi": gz, "transformations": transformations} def build_meta() -> Dict[str, str]: return {"engine": "iztro", "schema_version": "1.0"} def build_strength(palaces: List[Dict[str, Any]]) -> Dict[str, Any]: """ 亮度汇总——尤其标出"带四化但亮度弱"的星(2026-07-12发现的真实案例：小鱼盘太阳 化权却落陷、巨门化禄却亮度"不"，这两条此前完全没有被结构化标注出来，只能靠 人工核对十二宫才能发现，容易被AI报告当成纯粹的利好来写)。 """ weak_mutagen_stars = [] strong_stars = [] for p in palaces: for s in p.get("major_stars", []) or []: name, brightness, mutagen = s.get("name"), s.get("brightness"), s.get("mutagen") if not name: continue if mutagen and brightness in WEAK_BRIGHTNESS: weak_mutagen_stars.append({ "star": name, "palace": p.get("name"), "brightness": brightness, "mutagen": mutagen, }) if brightness in ("庙", "旺"): strong_stars.append({"star": name, "palace": p.get("name"), "brightness": brightness}) return { "weak_mutagen_stars": weak_mutagen_stars, "strong_stars": strong_stars, } def build_lu_ji_flow(sihua_analysis: Optional[Dict[str, Any]]) -> Optional[Dict[str, Any]]: """ 禄因忌果（小鱼原创方法论，见 agent-knowledge/TH&SiSi/business-methodology/ lu-yin-ji-guo-theory.md）：化禄=因，资源的入口与流动（资源从哪来）； 化忌=果，仓库——资源最终沉淀、真正属于命主的位置。忌所在宫位是要经营的 一生课题，不是要逃避的缺陷。 这是确定性推导（禄/忌落宫直接可得，不是AI判读），放在API层让所有产品 复用，避免每条产品线的prompt各自教AI这套读法、各教一个版本。 """ if not sihua_analysis: return None year_sihua = sihua_analysis.get("year_sihua", {}) lu, ji = year_sihua.get("化禄"), year_sihua.get("化忌") if not (lu and ji): return None return { "inlet": {"star": lu.get("star", ""), "palace": lu.get("palace", "")}, "vault": {"star": ji.get("star", ""), "palace": ji.get("palace", "")}, "reading": ( "Lu marks where resources flow IN; Ji marks the VAULT where they finally settle — " "a lifelong arena to cultivate, not a flaw. Always pair the vault's friction with its settlement." ), } def build_timing(daxun_analysis: Optional[Dict[str, Any]]) -> List[Dict[str, Any]]: if not daxun_analysis: return [] timing = [] for key, is_current in (("current", True), ("next", False)): d = daxun_analysis.get(key) if not d: continue # "next" daxun(predict_next_daxun的输出)不带end_age，只有start_age—— # 每步大限固定10年，缺失时用start_age+9兜底，跟evidence文本口径保持一致 end_age = d.get("end_age") if end_age is None and d.get("start_age") is not None: end_age = d["start_age"] + 9 timing.append({ "layer": "daxun", "label": d.get("palace_name", ""), "label_en": palace_term(d.get("palace_name", "")) if d.get("palace_name") else "", "start_age": d.get("start_age"), "end_age": end_age, "is_current": is_current, "flags": ["empty_palace"] if d.get("is_empty_palace") else [], }) return timing def build_evidence( chart: Dict[str, Any], palaces: List[Dict[str, Any]], sihua_analysis: Optional[Dict[str, Any]], daxun_analysis: Optional[Dict[str, Any]], ) -> str: lines: List[str] = [] ming_gong = next((p for p in palaces if p.get("name") == "命宫"), None) body_palace = next((p for p in palaces if p.get("is_body_palace")), None) if ming_gong: stars = ", ".join( star_with_brightness(s["name"], s.get("brightness", "")) for s in ming_gong.get("major_stars", []) if s.get("name") ) lines.append(f"- {palace_term('命宫')}: {stars or '(no major star / 借星安宫)'}.") lines.append(f"- Soul star (命主): {star_term(chart.get('soul', ''))}. Body star (身主): {star_term(chart.get('body', ''))}.") if body_palace: lines.append(f"- Body Palace (身宫) falls in: {palace_term(body_palace.get('name', ''))}.") if sihua_analysis: year_sihua = sihua_analysis.get("year_sihua", {}) parts = [] for mutagen_key, mutagen_char in (("化禄", "禄"), ("化权", "权"), ("化科", "科"), ("化忌", "忌")): entry = year_sihua.get(mutagen_key) if entry: parts.append(f"{mutagen_term(mutagen_char)} -> {star_term(entry.get('star',''))} @ {palace_term(entry.get('palace',''))}") if parts: lines.append("- Natal Four Transformations (生年四化): " + "; ".join(parts) + ".") flow = build_lu_ji_flow(sihua_analysis) if flow: lines.append( f"- Lu-Ji flow (禄因忌果): resources flow IN via {star_term(flow['inlet']['star'])} @ {palace_term(flow['inlet']['palace'])}; " f"they finally SETTLE in the VAULT at {star_term(flow['vault']['star'])} @ {palace_term(flow['vault']['palace'])}. " f"The Ji palace is the lifelong vault to cultivate — what truly belongs to this person accumulates there — " f"not a flaw to avoid; always pair its friction with its settlement." ) fy = build_flow_year_sihua(palaces) if fy: parts = [] for label in ("化禄", "化权", "化科", "化忌"): t = fy["transformations"].get(label, {}) if t.get("star"): loc = f" @ {palace_term(t['palace'])}" if t.get("palace") else "" bright = f" (brightness={brightness_term(t['brightness'])})" if t.get("brightness") in WEAK_BRIGHTNESS else "" parts.append(f"{mutagen_term(label[-1])} -> {star_term(t['star'])}{loc}{bright}") lines.append( f"- Flow-year Four Transformations (流年四化, {fy['year']} {fy['year_ganzhi']}): " + "; ".join(parts) + ". The flow-year Ji palace is the arena stress-tested THIS year (annual friction, not lifelong verdict)." ) strength = build_strength(palaces) if strength["weak_mutagen_stars"]: flagged = "; ".join( f"{star_term(w['star'])} {mutagen_term(w['mutagen'][-1] if len(w['mutagen'])>1 else w['mutagen'])} " f"but brightness={brightness_term(w['brightness'])}" for w in strength["weak_mutagen_stars"] ) lines.append( f"- CAUTION (signal present but weak execution): {flagged}. " f"The mutagen signal exists structurally but the star's own brightness is low — " f"deliver at a discount, must state both sides (structure vs. execution)." ) if daxun_analysis: current = daxun_analysis.get("current", {}) if current: major = ", ".join(current.get("palace_stars", {}).get("major", [])) lines.append( f"- Current Daxun (现行大限): {palace_term(current.get('palace_name',''))} " f"({current.get('start_age')}-{current.get('end_age')}), stars: {major or '(empty)'}." ) nxt = daxun_analysis.get("next", {}) if nxt: lines.append( f"- Next Daxun (下一大限): {palace_term(nxt.get('palace_name',''))} " f"({nxt.get('start_age')}-{nxt.get('start_age', 0) + 9})." ) return "\n".join(lines)