Spaces:
Running
Running
| """ | |
| 紫微斗数四层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) | |