jesse-ux
9bf4731e2c
Add the interpretation pages from fields the engine already computed, project the annual tables the reader prints, and clear English Han on the four measured charts. Patyayini is not implemented. The generate button copy is unchanged. This commit stays on the feature branch and is not pushed to staging.
1229 lines
57 KiB
Python
1229 lines
57 KiB
Python
"""Template interpretation packs for the full-data report.
|
||
|
||
These functions only read fields the engine has already calculated. They do
|
||
not compute charts, dashas, or Saturn cycles, and they do not import PyJHora.
|
||
Wording stays conditional: a period may bring a topic forward. It does not
|
||
assert death, lifespan, diagnosis, divorce, miscarriage, infertility, remarriage,
|
||
or a count of relationships. Neecha Bhanga is not labeled a Raja Yoga.
|
||
Functional nature is taken from this site's functional benefic/malefic layer.
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import re
|
||
from datetime import date
|
||
from typing import Any
|
||
|
||
PLANETS = ("Sun", "Moon", "Mars", "Mercury", "Jupiter", "Venus", "Saturn", "Rahu", "Ketu")
|
||
|
||
PLANET_ZH = {
|
||
"Sun": "太阳",
|
||
"Moon": "月亮",
|
||
"Mars": "火星",
|
||
"Mercury": "水星",
|
||
"Jupiter": "木星",
|
||
"Venus": "金星",
|
||
"Saturn": "土星",
|
||
"Rahu": "北交点",
|
||
"Ketu": "南交点",
|
||
}
|
||
|
||
ROLE_ZH = {
|
||
"functional_benefic": "功能吉星",
|
||
"functional_malefic": "功能凶星",
|
||
"functional_neutral": "功能中性",
|
||
}
|
||
ROLE_EN = {
|
||
"functional_benefic": "functional benefic",
|
||
"functional_malefic": "functional malefic",
|
||
"functional_neutral": "functional neutral",
|
||
}
|
||
|
||
# Themes only. No diagnosis, lifespan, or relationship-count claims.
|
||
_THEMES: dict[str, dict[str, tuple[str, str]]] = {
|
||
"Sun": {
|
||
"zh": ("这段时间可能突出责任、可见度与需要拍板的公共角色。", "父亲、地位或对外代表一类议题更容易被看见。"),
|
||
"en": ("This period may bring forward responsibility, visibility, and public roles that need a decision.", "Father, status, or representing a group can become easier to notice."),
|
||
},
|
||
"Moon": {
|
||
"zh": ("这段时间可能突出情绪节奏、居住与照料关系。", "家庭日常、变动和公众情绪更容易成为主题。"),
|
||
"en": ("This period may bring forward emotional pace, home, and care in relationships.", "Household routine, change, and public mood can become easier to notice."),
|
||
},
|
||
"Mars": {
|
||
"zh": ("这段时间可能突出行动、竞争与把事情做完的压力。", "勇气、技术操练和冲突处理更容易被看见。"),
|
||
"en": ("This period may bring forward action, competition, and pressure to finish work.", "Courage, technical practice, and handling conflict can become easier to notice."),
|
||
},
|
||
"Mercury": {
|
||
"zh": ("这段时间可能突出学习、书写、交易与来回沟通。", "合同、计算和技能交换更容易成为主题。"),
|
||
"en": ("This period may bring forward study, writing, trade, and back-and-forth communication.", "Contracts, calculation, and skill exchange can become easier to notice."),
|
||
},
|
||
"Jupiter": {
|
||
"zh": ("这段时间可能突出判断、教导、信仰与扩张机会。", "顾问、远行和意义感更容易被看见。"),
|
||
"en": ("This period may bring forward judgment, teaching, belief, and chances to expand.", "Advising, long travel, and a search for meaning can become easier to notice."),
|
||
},
|
||
"Venus": {
|
||
"zh": ("这段时间可能突出合作、审美、合意与资源交换。", "艺术、协商和关系里的舒适感更容易成为主题。"),
|
||
"en": ("This period may bring forward cooperation, taste, agreement, and exchange of resources.", "Art, negotiation, and comfort inside a relationship can become easier to notice."),
|
||
},
|
||
"Saturn": {
|
||
"zh": ("这段时间可能突出结构、延迟、责任与长期功课。", "界限、资历和工作负担更容易被看见。"),
|
||
"en": ("This period may bring forward structure, delay, duty, and long-term work.", "Boundaries, seniority, and the weight of work can become easier to notice."),
|
||
},
|
||
"Rahu": {
|
||
"zh": ("这段时间可能突出放大、突破与不按旧习惯走的路径。", "外来机会和强烈欲望更容易成为主题。"),
|
||
"en": ("This period may bring forward amplification, breaks with habit, and unfamiliar routes.", "Outside opportunities and strong wanting can become easier to notice."),
|
||
},
|
||
"Ketu": {
|
||
"zh": ("这段时间可能突出收束、分离与向内整理。", "放下、专精和减少外界噪声更容易被看见。"),
|
||
"en": ("This period may bring forward release, separation, and sorting things inwardly.", "Letting go, specialization, and less outer noise can become easier to notice."),
|
||
},
|
||
}
|
||
|
||
_HOUSE_TOPIC_ZH = {
|
||
1: "自我呈现与人生主轴",
|
||
2: "资源、表达与家庭积累",
|
||
3: "行动、学习与沟通",
|
||
4: "家庭基础与居住",
|
||
5: "创造、判断与教育",
|
||
6: "服务、竞争与日常责任",
|
||
7: "合作与一对一关系",
|
||
8: "共享资源、风险与深层变化",
|
||
9: "信念、远行与高阶学习",
|
||
10: "职业、职责与社会角色",
|
||
11: "收益、社群与长期目标",
|
||
12: "休整、支出与幕后事务",
|
||
}
|
||
_HOUSE_TOPIC_EN = {
|
||
1: "self-presentation and the main life axis",
|
||
2: "resources, speech, and family accumulation",
|
||
3: "initiative, study, and communication",
|
||
4: "home base and residence",
|
||
5: "creativity, judgment, and education",
|
||
6: "service, competition, and daily duty",
|
||
7: "cooperation and one-to-one relationship",
|
||
8: "shared resources, risk, and deep change",
|
||
9: "belief, long travel, and higher study",
|
||
10: "career, duty, and public role",
|
||
11: "gains, networks, and long-range aims",
|
||
12: "retreat, spending, and behind-the-scenes work",
|
||
}
|
||
|
||
DECISION_4_DASHAS = (
|
||
"sthira",
|
||
"drig",
|
||
"shoola",
|
||
"navamsha",
|
||
"niryaana_shoola",
|
||
"lagna_kendradi",
|
||
"chara",
|
||
)
|
||
|
||
# These two producers are named in the catalog and left for a later pass.
|
||
DEFERRED_ANNUAL_FIELDS = frozenset({"monthly_chart_pack", "annual_chart_series_pack"})
|
||
|
||
# Upstream annual chapters. A missing producer is listed, not calculated here.
|
||
ANNUAL_CHAPTERS = (
|
||
("Mudda Dasha", "mudda_dasha"),
|
||
("Patyayini Dasha", "patyayini_dasha"),
|
||
("Patyayini ending dates", "patyayini_dasha"),
|
||
("Sahams", "sahams"),
|
||
("Raman Sahams", "sahams"),
|
||
("Tajika yogas", "tajika_yogas"),
|
||
("Monthly mudda windows", "monthly_windows"),
|
||
("Solar return", "solar_return"),
|
||
("Monthly return charts", "monthly_chart_pack"),
|
||
("Eight-year overview", "annual_chart_series_pack"),
|
||
("Lifetime annual charts", "lifetime_annual_chart_pages"),
|
||
("Tripataki", "tripataki_chakra"),
|
||
("Annual special points", "annual_special_points"),
|
||
("Current-year result notes", "pl9_current_year_results"),
|
||
)
|
||
|
||
|
||
def _lang(packet: dict | None) -> str:
|
||
language = str((packet or {}).get("report_language") or "").strip().lower()
|
||
return language if language in {"zh", "en"} else "zh"
|
||
|
||
|
||
def _dict(value: Any) -> dict:
|
||
return value if isinstance(value, dict) else {}
|
||
|
||
|
||
def _list(value: Any) -> list:
|
||
return value if isinstance(value, list) else []
|
||
|
||
|
||
def _planet(name: Any, language: str) -> str:
|
||
text = str(name or "").strip()
|
||
if not text:
|
||
return "未列出" if language == "zh" else "not listed"
|
||
if language == "zh":
|
||
return PLANET_ZH.get(text, text)
|
||
return text
|
||
|
||
|
||
def _role(raw: Any, language: str) -> str:
|
||
key = str(raw or "").strip()
|
||
table = ROLE_ZH if language == "zh" else ROLE_EN
|
||
if key in table:
|
||
return table[key]
|
||
return "未单独列出" if language == "zh" else "not listed separately"
|
||
|
||
|
||
def _dignity(value: Any, language: str) -> str:
|
||
text = str(value or "").strip()
|
||
if not text:
|
||
return "未列出" if language == "zh" else "not listed"
|
||
match = re.fullmatch(r"([^()（）]+?)\s*[(（]([A-Za-z][A-Za-z \-]*)[)）]", text)
|
||
if not match:
|
||
if language == "en" and re.search(r"[\u4e00-\u9fff]", text):
|
||
return "not listed separately"
|
||
return text
|
||
return match.group(1).strip() if language == "zh" else match.group(2).strip()
|
||
|
||
|
||
def _date_text(value: Any) -> str:
|
||
text = str(value or "").strip()
|
||
if "T" in text and len(text) >= 10:
|
||
return text[:10]
|
||
return text[:10] if len(text) >= 10 and text[4:5] == "-" else text
|
||
|
||
|
||
def _worksheets(packet: dict) -> dict:
|
||
return _dict(packet.get("worksheets"))
|
||
|
||
|
||
def _timing(packet: dict) -> dict:
|
||
return _dict(_worksheets(packet).get("timing_and_predictive_systems"))
|
||
|
||
|
||
def _advanced(packet: dict) -> dict:
|
||
return _dict(_worksheets(packet).get("advanced_systems"))
|
||
|
||
|
||
def _planets(packet: dict) -> dict:
|
||
d1 = _dict(_worksheets(packet).get("d1_rasi_bhava"))
|
||
planets = _dict(d1.get("planets"))
|
||
if planets:
|
||
return planets
|
||
return _dict(_dict(packet.get("core_chart")).get("planets"))
|
||
|
||
|
||
def _houses(packet: dict) -> dict:
|
||
d1 = _dict(_worksheets(packet).get("d1_rasi_bhava"))
|
||
houses = _dict(d1.get("houses"))
|
||
if houses:
|
||
return houses
|
||
return _dict(_dict(packet.get("core_chart")).get("houses"))
|
||
|
||
|
||
def _house_row(houses: dict, number: int) -> dict:
|
||
return _dict(houses.get(f"house_{number}") or houses.get(str(number)) or houses.get(number))
|
||
|
||
|
||
def _functional_classification(packet: dict) -> dict[str, str]:
|
||
layer = _dict(_dict(_worksheets(packet).get("strengths_and_scores")).get("functional_benefic_malefic"))
|
||
if not layer:
|
||
layer = _dict(_advanced(packet).get("functional_benefic_malefic"))
|
||
classified: dict[str, str] = {}
|
||
for label, key in (
|
||
("functional_benefic", "functional_benefics"),
|
||
("functional_malefic", "functional_malefics"),
|
||
("functional_neutral", "functional_neutrals"),
|
||
):
|
||
for planet in _list(layer.get(key)):
|
||
classified[str(planet)] = label
|
||
return classified
|
||
|
||
|
||
def _owned_houses(packet: dict, houses: dict) -> dict[str, list[int]]:
|
||
layer = _dict(_dict(_worksheets(packet).get("strengths_and_scores")).get("functional_benefic_malefic"))
|
||
raw = layer.get("owned_houses")
|
||
owned: dict[str, list[int]] = {}
|
||
if isinstance(raw, dict):
|
||
for planet, houses_owned in raw.items():
|
||
if isinstance(houses_owned, list):
|
||
owned[str(planet)] = [int(item) for item in houses_owned if isinstance(item, int)]
|
||
if owned:
|
||
return owned
|
||
for number in range(1, 13):
|
||
lord = _house_row(houses, number).get("lord")
|
||
if lord:
|
||
owned.setdefault(str(lord), []).append(number)
|
||
return owned
|
||
|
||
|
||
def _current_lords(packet: dict) -> dict[str, str]:
|
||
dasha = _dict(_timing(packet).get("dasha"))
|
||
current = _dict(dasha.get("current_dasha"))
|
||
md = str(current.get("lord") or "")
|
||
ad = ""
|
||
pd = ""
|
||
for row in _list(current.get("antardasha_timeline")):
|
||
if isinstance(row, dict) and row.get("is_current"):
|
||
ad = str(row.get("lord") or "")
|
||
for child in _list(row.get("pratyantar_dasha_timeline")):
|
||
if isinstance(child, dict) and child.get("is_current"):
|
||
pd = str(child.get("lord") or "")
|
||
break
|
||
break
|
||
if not md:
|
||
for period in _list(dasha.get("timeline")):
|
||
if isinstance(period, dict) and period.get("is_current"):
|
||
md = str(period.get("lord") or "")
|
||
break
|
||
return {"md": md, "ad": ad, "pd": pd}
|
||
|
||
|
||
def _lord_facts(packet: dict) -> dict[str, dict[str, Any]]:
|
||
planets = _planets(packet)
|
||
houses = _houses(packet)
|
||
owned = _owned_houses(packet, houses)
|
||
facts: dict[str, dict[str, Any]] = {}
|
||
for planet, row in planets.items():
|
||
if not isinstance(row, dict):
|
||
continue
|
||
item = {
|
||
"house": row.get("house"),
|
||
"ownership": owned.get(str(planet), []),
|
||
"dignity": row.get("status") or row.get("dignity"),
|
||
"nakshatra": row.get("nakshatra"),
|
||
"pada": row.get("nakshatra_pada") or row.get("pada"),
|
||
"sign": row.get("sign"),
|
||
}
|
||
facts[str(planet)] = item
|
||
return facts
|
||
|
||
|
||
def _themes(lord: str, language: str) -> list[str]:
|
||
bank = _THEMES.get(lord) or {}
|
||
pair = bank.get(language) or bank.get("en") or ()
|
||
return list(pair)
|
||
|
||
|
||
def _native_vimshottari_periods(packet: dict) -> list[dict[str, Any]]:
|
||
dasha = _dict(_timing(packet).get("dasha"))
|
||
timeline = _list(dasha.get("timeline"))
|
||
periods = []
|
||
for period in timeline:
|
||
if not isinstance(period, dict) or not period.get("lord"):
|
||
continue
|
||
antardashas = []
|
||
for ad in _list(period.get("antardasha_timeline")):
|
||
if not isinstance(ad, dict):
|
||
continue
|
||
antardashas.append({
|
||
**ad,
|
||
"start": _date_text(ad.get("start") or ad.get("start_date")),
|
||
"end": _date_text(ad.get("end") or ad.get("end_date")),
|
||
"pratyantar_dasha_timeline": [
|
||
{
|
||
**child,
|
||
"start": _date_text(child.get("start") or child.get("start_date")),
|
||
"end": _date_text(child.get("end") or child.get("end_date")),
|
||
}
|
||
for child in _list(ad.get("pratyantar_dasha_timeline"))
|
||
if isinstance(child, dict)
|
||
],
|
||
})
|
||
periods.append({
|
||
"lord": period.get("lord"),
|
||
"start": _date_text(period.get("start") or period.get("start_date")),
|
||
"end": _date_text(period.get("end") or period.get("end_date")),
|
||
"status": "verified",
|
||
"raw_period": {
|
||
"is_current": bool(period.get("is_current")),
|
||
"antardasha_timeline": antardashas,
|
||
},
|
||
})
|
||
return periods
|
||
|
||
|
||
def build_dasha_master_view(packet: dict) -> dict[str, Any]:
|
||
"""Reshape the native Vimshottari timeline. No new periods are calculated."""
|
||
|
||
language = _lang(packet)
|
||
classification = _functional_classification(packet)
|
||
facts = _lord_facts(packet)
|
||
periods = _native_vimshottari_periods(packet)
|
||
items = []
|
||
for period in periods:
|
||
lord = str(period.get("lord") or "")
|
||
if any(row.get("lord") == lord for row in items):
|
||
continue
|
||
row_facts = facts.get(lord, {})
|
||
blocked = [key for key in ("house", "ownership", "dignity", "nakshatra", "pada") if row_facts.get(key) in (None, "", [])]
|
||
items.append({
|
||
"kind": "planet_lord_trigger",
|
||
"family": "vimshottari",
|
||
"lord": lord,
|
||
"facts": row_facts,
|
||
"functional_classification": classification.get(lord, "blocked"),
|
||
"blocked_fields": blocked,
|
||
"status": "partial" if blocked else "verified",
|
||
"evidence_label": "single_system_inference" if not blocked else "parameter_sensitive",
|
||
})
|
||
return {
|
||
"schema": "dasha_master_report_pack_v1",
|
||
"audit": {"status": "partial_verified" if items else "blocked"},
|
||
"families": {"vimshottari": {"periods": periods, "execution_status": "partial_verified" if periods else "blocked"}},
|
||
"interpretation_triggers": {
|
||
"status": "partial" if items else "blocked",
|
||
"planet_lord_items": items,
|
||
"sign_items": [],
|
||
},
|
||
"language": language,
|
||
}
|
||
|
||
|
||
def build_dasha_interpretation_pack(dasha_master_pack: dict[str, Any], profile: dict[str, Any] | None = None) -> dict[str, Any]:
|
||
"""Turn already-built Vimshottari facts into report pages."""
|
||
|
||
language = str((dasha_master_pack or {}).get("language") or (profile or {}).get("language") or "zh")
|
||
triggers = _dict(dasha_master_pack.get("interpretation_triggers"))
|
||
items = [row for row in _list(triggers.get("planet_lord_items")) if isinstance(row, dict)]
|
||
source_status = str(_dict(dasha_master_pack.get("audit")).get("status") or triggers.get("status") or "blocked")
|
||
pages = _dasha_pages(dasha_master_pack, items, language)
|
||
status = "partial_verified" if pages else "blocked"
|
||
return {
|
||
"schema": "pl9.dasha_interpretation_pack.v1",
|
||
"status": status,
|
||
"language": language,
|
||
"page_series": pages,
|
||
"domain_paragraphs": _domain_paragraphs(items, language),
|
||
"summary": {"page_series_count": len(pages), "narrative_item_count": len(items)},
|
||
"audit": {"status": status, "source_dasha_master_status": source_status},
|
||
}
|
||
|
||
|
||
def _dasha_pages(master: dict, items: list[dict], language: str) -> list[dict[str, Any]]:
|
||
periods = _list(_dict(_dict(master.get("families")).get("vimshottari")).get("periods"))
|
||
by_lord = {str(row.get("lord")): row for row in items if row.get("lord")}
|
||
current_index = next((idx for idx, period in enumerate(periods) if _dict(period.get("raw_period")).get("is_current")), 0)
|
||
ordered = list(periods[current_index:]) + list(periods[:current_index])
|
||
pages = []
|
||
for period in ordered:
|
||
if not isinstance(period, dict) or not period.get("lord"):
|
||
continue
|
||
lord = str(period["lord"])
|
||
trigger = by_lord.get(lord, {})
|
||
facts = _dict(trigger.get("facts") or trigger.get("trigger_facts"))
|
||
raw = _dict(period.get("raw_period"))
|
||
antardashas = _list(raw.get("antardasha_timeline"))
|
||
if raw.get("is_current"):
|
||
current_ad_index = next((idx for idx, row in enumerate(antardashas) if isinstance(row, dict) and row.get("is_current")), 0)
|
||
report_ads = antardashas[current_ad_index:]
|
||
else:
|
||
report_ads = antardashas
|
||
start = period.get("start") or "?"
|
||
end = period.get("end") or "?"
|
||
title = (
|
||
f"{_planet(lord, language)} 主运（MD）：{start} 至 {end}"
|
||
if language == "zh"
|
||
else f"{lord} Maha Dasha (MD): {start} to {end}"
|
||
)
|
||
current_ad = next((row for row in antardashas if isinstance(row, dict) and row.get("is_current")), None)
|
||
if language == "zh":
|
||
general = f"{_planet(lord, language)} 主运期间，这段时间可能突出该星的自然主题。下面只按已算好的落宫、守护宫、尊贵、星宿和功能吉凶来读，不是确定会发生的事。"
|
||
natal = _natal_zh(lord, facts, trigger)
|
||
current_text = (
|
||
f"当前子运（AD）是 {_planet(_dict(current_ad).get('lord'), language)}，{current_ad.get('start') or '?'} 至 {current_ad.get('end') or '?'}。它只细化当前节奏。"
|
||
if isinstance(current_ad, dict)
|
||
else "这一主运不是当前阶段，子运保留在时间表里。"
|
||
)
|
||
else:
|
||
general = f"During the Maha Dasha of {lord}, this period may bring the planet's natural topics forward. The reading uses only the calculated house, ownership, dignity, nakshatra, and functional nature. It is not a certain event."
|
||
natal = _natal_en(lord, facts, trigger)
|
||
current_text = (
|
||
f"The current Antar Dasha (AD) is {current_ad.get('lord')}, {current_ad.get('start') or '?'} to {current_ad.get('end') or '?'} . It only refines the current pace."
|
||
if isinstance(current_ad, dict)
|
||
else "This Maha Dasha is not the current period. Sub-periods stay in the timeline."
|
||
)
|
||
flow_blocks = [{
|
||
"kind": "mahadasha",
|
||
"title": title,
|
||
"heading": title,
|
||
"sections": [("General", line) for line in (general, *_themes(lord, language))],
|
||
}]
|
||
subperiods = []
|
||
for ad in report_ads:
|
||
if not isinstance(ad, dict) or not ad.get("lord"):
|
||
continue
|
||
ad_lord = str(ad["lord"])
|
||
ad_facts = _dict(by_lord.get(ad_lord, {}).get("facts") or by_lord.get(ad_lord, {}).get("trigger_facts"))
|
||
paragraph = _sub_paragraph(lord, ad_lord, ad, ad_facts, language, level="AD")
|
||
pratyantars = []
|
||
for pd in _list(ad.get("pratyantar_dasha_timeline")):
|
||
if not isinstance(pd, dict) or not pd.get("lord"):
|
||
continue
|
||
pd_lord = str(pd["lord"])
|
||
pd_facts = _dict(by_lord.get(pd_lord, {}).get("facts") or by_lord.get(pd_lord, {}).get("trigger_facts"))
|
||
pd_paragraph = _sub_paragraph(ad_lord, pd_lord, pd, pd_facts, language, level="PD")
|
||
pratyantars.append({"lord": pd_lord, "start": pd.get("start"), "end": pd.get("end"), "paragraph": pd_paragraph})
|
||
flow_blocks.append({
|
||
"kind": "pratyantar",
|
||
"title": f"{lord}-{ad_lord}-{pd_lord}",
|
||
"heading": f"{_planet(pd_lord, language)} PD" if language == "zh" else f"Pratyantar Dasha of {pd_lord}",
|
||
"sections": [("Pratyantar", pd_paragraph), *[( "Theme", line) for line in _themes(pd_lord, language)[:1]]],
|
||
})
|
||
subperiods.append({"lord": ad_lord, "start": ad.get("start"), "end": ad.get("end"), "paragraph": paragraph, "pratyantar_periods": pratyantars})
|
||
flow_blocks.append({
|
||
"kind": "antardasha",
|
||
"title": f"{lord}-{ad_lord}",
|
||
"heading": f"{_planet(ad_lord, language)} AD" if language == "zh" else f"Antar Dasha of {ad_lord}",
|
||
"sections": [("Antar", paragraph), *[("Theme", line) for line in _themes(ad_lord, language)[:1]]],
|
||
})
|
||
pages.append({
|
||
"title": title,
|
||
"maha_lord": lord,
|
||
"start": start,
|
||
"end": end,
|
||
"subperiods": subperiods,
|
||
"flow_blocks": flow_blocks,
|
||
"sections": {
|
||
"general": general,
|
||
"natal_context": natal,
|
||
"domain_context": _domain_line(lord, facts, language),
|
||
"current_subperiod": current_text,
|
||
},
|
||
})
|
||
return pages
|
||
|
||
|
||
def _natal_zh(lord: str, facts: dict, trigger: dict) -> str:
|
||
house = facts.get("house")
|
||
owned = facts.get("ownership") if isinstance(facts.get("ownership"), list) else []
|
||
bits = [f"{_planet(lord, 'zh')}在本命盘位于第{house}宫" if isinstance(house, int) else f"{_planet(lord, 'zh')}的落宫未列出"]
|
||
if owned:
|
||
bits.append("守护第" + "、".join(str(item) for item in owned) + "宫")
|
||
bits.append(f"尊贵为{_dignity(facts.get('dignity'), 'zh')}")
|
||
bits.append(f"星宿为{facts.get('nakshatra') or '未列出'}")
|
||
bits.append(f"功能吉凶为{_role(trigger.get('functional_classification'), 'zh')}")
|
||
return "盘内依据：" + "；".join(bits) + "。功能吉凶用本仓功能吉凶层，不另做一套判断。"
|
||
|
||
|
||
def _natal_en(lord: str, facts: dict, trigger: dict) -> str:
|
||
house = facts.get("house") if isinstance(facts.get("house"), int) else "not listed"
|
||
owned = facts.get("ownership") if isinstance(facts.get("ownership"), list) else []
|
||
owned_text = ", ".join(str(item) for item in owned) if owned else "not listed"
|
||
nakshatra = facts.get("nakshatra") or "not listed"
|
||
if re.search(r"[\u4e00-\u9fff]", str(nakshatra)):
|
||
nakshatra = "not listed separately"
|
||
return (
|
||
f"Natal basis: {lord} is in house {house}, owns houses {owned_text}, "
|
||
f"dignity {_dignity(facts.get('dignity'), 'en')}, nakshatra {nakshatra}, "
|
||
f"functional nature {_role(trigger.get('functional_classification'), 'en')}. "
|
||
"Functional nature comes from this site's functional benefic/malefic layer."
|
||
)
|
||
|
||
|
||
def _domain_line(lord: str, facts: dict, language: str) -> str:
|
||
house = facts.get("house")
|
||
if language == "zh":
|
||
if not isinstance(house, int):
|
||
return f"领域展开：{_planet(lord, language)}的宫位未列出，这里只保留主运索引。"
|
||
return f"领域展开：{_planet(lord, language)}位于第{house}宫。事业、财富、关系与日常责任要再看对应宫主和分盘，这段时间可能突出这些主题。"
|
||
if not isinstance(house, int):
|
||
return f"Domain note: house placement for {lord} is not listed, so this page keeps the period index only."
|
||
return f"Domain note: {lord} is in house {house}. Career, resources, relationship, and daily duty still need their house lords and divisional charts. This period may bring those topics forward."
|
||
|
||
|
||
def _sub_paragraph(parent: str, lord: str, row: dict, facts: dict, language: str, *, level: str) -> str:
|
||
house = facts.get("house") if isinstance(facts.get("house"), int) else None
|
||
owned = facts.get("ownership") if isinstance(facts.get("ownership"), list) else []
|
||
start = row.get("start") or "?"
|
||
end = row.get("end") or "?"
|
||
if language == "zh":
|
||
owned_text = "、".join(str(item) for item in owned) if owned else "未列出"
|
||
house_text = str(house) if house is not None else "未列出"
|
||
return (
|
||
f"{_planet(parent, language)} / {_planet(lord, language)}（{level}）：{start} 至 {end}。"
|
||
f"这段时间可能突出{_planet(lord, language)}的第{house_text}宫、守护第{owned_text}宫、"
|
||
f"尊贵{_dignity(facts.get('dignity'), language)}与星宿{facts.get('nakshatra') or '未列出'}。"
|
||
"这是阶段线索，不是确定会发生的事。"
|
||
)
|
||
owned_text = ", ".join(str(item) for item in owned) if owned else "not listed"
|
||
house_text = str(house) if house is not None else "not listed"
|
||
nakshatra = facts.get("nakshatra") or "not listed"
|
||
if re.search(r"[\u4e00-\u9fff]", str(nakshatra)):
|
||
nakshatra = "not listed separately"
|
||
return (
|
||
f"{parent} / {lord} ({level}): {start} to {end}. "
|
||
f"This period may bring forward {lord}'s house {house_text}, owned houses {owned_text}, "
|
||
f"dignity {_dignity(facts.get('dignity'), language)}, and nakshatra {nakshatra}. "
|
||
"This is a period clue, not a certain event."
|
||
)
|
||
|
||
|
||
def _domain_paragraphs(items: list[dict], language: str) -> list[dict[str, Any]]:
|
||
groups = {
|
||
"career": {1, 2, 6, 10, 11},
|
||
"wealth": {2, 5, 9, 11},
|
||
"relationship": {1, 5, 7, 11},
|
||
"health": {1, 6, 8, 12},
|
||
}
|
||
titles = {
|
||
"zh": {"career": "事业", "wealth": "财富", "relationship": "关系", "health": "日常状态"},
|
||
"en": {"career": "Career", "wealth": "Resources", "relationship": "Relationship", "health": "Daily condition"},
|
||
}
|
||
rows = []
|
||
for domain, houses in groups.items():
|
||
matched = []
|
||
for item in items:
|
||
facts = _dict(item.get("facts") or item.get("trigger_facts"))
|
||
found = set()
|
||
if isinstance(facts.get("house"), int):
|
||
found.add(facts["house"])
|
||
if isinstance(facts.get("ownership"), list):
|
||
found.update(value for value in facts["ownership"] if isinstance(value, int))
|
||
if found & houses:
|
||
matched.append(str(item.get("lord")))
|
||
names = ", ".join(_planet(name, language) for name in matched[:5]) or ("未列出" if language == "zh" else "not listed")
|
||
if language == "zh":
|
||
paragraph = f"{titles['zh'][domain]}：这段时间可能突出 {names} 所守护的宫位主题，需与分盘和年度层一起看，不是确定会发生的事。"
|
||
else:
|
||
paragraph = f"{titles['en'][domain]}: this period may bring forward topics of {names}. Read them with the divisional charts and the annual layer. This is not a certain event."
|
||
rows.append({"domain": domain, "paragraph": paragraph, "trigger_count": len(matched)})
|
||
return rows
|
||
|
||
|
||
def build_vimshottari_dasha_report_pack(
|
||
native_payload: dict[str, Any],
|
||
pl9_payload: dict[str, Any] | None = None,
|
||
profile: dict[str, Any] | None = None,
|
||
tolerance_days: int = 1,
|
||
) -> dict[str, Any]:
|
||
"""Compare native and supplied PL9 periods by period_key. It does not calculate dashas."""
|
||
|
||
native_periods = [row for row in _list((native_payload or {}).get("periods")) if isinstance(row, dict)]
|
||
pl9_periods = [row for row in _list((pl9_payload or {}).get("periods")) if isinstance(row, dict)]
|
||
|
||
def _key(row: dict) -> str:
|
||
if row.get("period_key"):
|
||
return str(row["period_key"])
|
||
return "|".join(str(row.get(part) or "") for part in ("lord", "start", "end"))
|
||
|
||
native_by_key = {_key(row): row for row in native_periods if _key(row).strip("|")}
|
||
pl9_by_key = {_key(row): row for row in pl9_periods if _key(row).strip("|")}
|
||
rows = []
|
||
matched = mismatched = blocked = 0
|
||
for period_key in sorted(set(native_by_key) | set(pl9_by_key)):
|
||
native_period = native_by_key.get(period_key)
|
||
pl9_period = pl9_by_key.get(period_key)
|
||
start_delta = end_delta = None
|
||
if native_period is None or pl9_period is None:
|
||
status = "blocked"
|
||
blocked += 1
|
||
else:
|
||
start_delta = _delta_days(native_period.get("start"), pl9_period.get("start"))
|
||
end_delta = _delta_days(native_period.get("end"), pl9_period.get("end"))
|
||
deltas = [item for item in (start_delta, end_delta) if item is not None]
|
||
if len(deltas) != 2:
|
||
status = "blocked"
|
||
blocked += 1
|
||
elif max(deltas) <= tolerance_days:
|
||
status = "verified"
|
||
matched += 1
|
||
else:
|
||
status = "parameter_sensitive"
|
||
mismatched += 1
|
||
rows.append({
|
||
"period_key": period_key,
|
||
"start_delta_days": start_delta,
|
||
"end_delta_days": end_delta,
|
||
"status": status,
|
||
})
|
||
if not pl9_periods:
|
||
audit_status = "blocked"
|
||
reason = "no_external_period_list"
|
||
elif blocked:
|
||
audit_status = "blocked"
|
||
reason = "period_missing"
|
||
elif mismatched:
|
||
audit_status = "parameter_sensitive"
|
||
reason = "period_delta"
|
||
elif matched:
|
||
audit_status = "verified"
|
||
reason = "within_tolerance"
|
||
else:
|
||
audit_status = "blocked"
|
||
reason = "no_periods"
|
||
return {
|
||
"schema": "pl9.vimshottari_dasha_report_pack.v1",
|
||
"status": audit_status,
|
||
"tolerance_days": tolerance_days,
|
||
"rows": rows,
|
||
"summary": {"matched": matched, "mismatched": mismatched, "blocked": blocked},
|
||
"audit": {"status": audit_status, "reason": reason, "profile": profile or {}},
|
||
}
|
||
|
||
|
||
def _delta_days(left: Any, right: Any) -> int | None:
|
||
if not left or not right:
|
||
return None
|
||
try:
|
||
return abs((date.fromisoformat(str(left)[:10]) - date.fromisoformat(str(right)[:10])).days)
|
||
except ValueError:
|
||
return None
|
||
|
||
|
||
def build_graha_interpretation_pack(
|
||
chart_planets: dict[str, Any],
|
||
owned_houses: dict[str, Any] | None = None,
|
||
functional_classification: dict[str, str] | None = None,
|
||
active_dasha_lords: set[str] | None = None,
|
||
remedy_snapshot: dict[str, Any] | None = None,
|
||
strength_snapshot: dict[str, Any] | None = None,
|
||
*,
|
||
language: str = "zh",
|
||
) -> dict[str, Any]:
|
||
"""Planet pages from calculated placement, dignity, and functional nature."""
|
||
|
||
owned_houses = owned_houses or {}
|
||
functional_classification = functional_classification or {}
|
||
active = active_dasha_lords or set()
|
||
strength_snapshot = strength_snapshot or {}
|
||
pages = []
|
||
for planet in PLANETS:
|
||
row = chart_planets.get(planet) if isinstance(chart_planets.get(planet), dict) else {}
|
||
if not row:
|
||
continue
|
||
house = row.get("house")
|
||
sign = row.get("sign")
|
||
owned = owned_houses.get(planet) if isinstance(owned_houses.get(planet), list) else []
|
||
role = _role(functional_classification.get(planet), language)
|
||
dignity = _dignity(row.get("status") or row.get("dignity"), language)
|
||
nakshatra = row.get("nakshatra") or ("未列出" if language == "zh" else "not listed")
|
||
if language == "en" and re.search(r"[\u4e00-\u9fff]", str(nakshatra)):
|
||
nakshatra = "not listed separately"
|
||
if language == "en" and re.search(r"[\u4e00-\u9fff]", str(sign or "")):
|
||
sign = "not listed separately"
|
||
strength = _dict(_dict(strength_snapshot.get("planets")).get(planet))
|
||
total = strength.get("total") or strength.get("total_shadbala")
|
||
active_text = planet in active
|
||
if language == "zh":
|
||
sections = {
|
||
"placement": f"{_planet(planet, language)}落在{sign or '未列出'}第{house if isinstance(house, int) else '未列出'}宫，尊贵为{dignity}。",
|
||
"sign_and_dispositor": f"守护宫为{'、'.join(str(item) for item in owned) or '未列出'}。这段时间可能突出这些宫的主题。",
|
||
"nakshatra": f"星宿为{nakshatra}。星宿主只作为已算好的修饰，不单独下结论。",
|
||
"functional_role": f"功能吉凶为{role}。这一层用本仓功能吉凶 v2，与自然吉凶分开读。",
|
||
"strength_and_context": f"六维力量合计为{total if total is not None else '未列出'}。力量只说明表现条件，不是确定会发生的事。",
|
||
"activation": (
|
||
f"{_planet(planet, language)}在当前 MD、AD 或 PD 中，这段时间可能突出它守护的主题。"
|
||
if active_text
|
||
else f"{_planet(planet, language)}不在当前三级大运里，先作为本命背景。"
|
||
),
|
||
}
|
||
else:
|
||
sections = {
|
||
"placement": f"{planet} is in {sign or 'not listed'}, house {house if isinstance(house, int) else 'not listed'}, dignity {dignity}.",
|
||
"sign_and_dispositor": f"Owned houses: {', '.join(str(item) for item in owned) or 'not listed'}. This period may bring those house topics forward.",
|
||
"nakshatra": f"Nakshatra: {nakshatra}. The star lord only qualifies the calculated placement.",
|
||
"functional_role": f"Functional nature: {role}. This uses the site's functional benefic/malefic layer, separate from natural benefic or malefic.",
|
||
"strength_and_context": f"Shadbala total: {total if total is not None else 'not listed'}. Strength describes condition, not a certain event.",
|
||
"activation": (
|
||
f"{planet} is inside the current MD, AD, or PD, so this period may bring its topics forward."
|
||
if active_text
|
||
else f"{planet} is not inside the current three Dasha levels, so it stays natal background."
|
||
),
|
||
}
|
||
pages.append({"planet": planet if language == "en" else _planet(planet, language), "sections": sections})
|
||
return {
|
||
"schema": "pl9.graha_interpretation_pack.v1",
|
||
"status": "partial_verified" if pages else "blocked",
|
||
"planet_pages": pages,
|
||
"audit": {"status": "partial_verified" if pages else "blocked", "remedy_used": bool(remedy_snapshot)},
|
||
}
|
||
|
||
|
||
def build_yoga_interpretation_pack(
|
||
yoga_snapshot: dict[str, Any] | None,
|
||
yogas_doshas_snapshot: dict[str, Any] | None = None,
|
||
active_dasha_lords: set[str] | None = None,
|
||
*,
|
||
language: str = "zh",
|
||
) -> dict[str, Any]:
|
||
"""Pages for yogas the detector already returned."""
|
||
|
||
yoga_snapshot = yoga_snapshot or {}
|
||
yogas_doshas_snapshot = yogas_doshas_snapshot or {}
|
||
active = active_dasha_lords or set()
|
||
groups = (
|
||
yoga_snapshot.get("detected_yogas"),
|
||
yoga_snapshot.get("yogas"),
|
||
_dict(yogas_doshas_snapshot.get("raj_yogas")).get("yogas"),
|
||
_dict(yogas_doshas_snapshot.get("dhana_yogas")).get("yogas"),
|
||
_dict(yogas_doshas_snapshot.get("pancha_mahapurusha")).get("yogas"),
|
||
_dict(yogas_doshas_snapshot.get("nicha_bhanga_raj")).get("yogas"),
|
||
)
|
||
pages = []
|
||
seen: set[str] = set()
|
||
for rows in groups:
|
||
for raw in _list(rows):
|
||
if not isinstance(raw, dict):
|
||
continue
|
||
name = _yoga_name(raw, language)
|
||
if name in seen:
|
||
continue
|
||
seen.add(name)
|
||
combination = str(raw.get("combination_en") or raw.get("combination") or raw.get("reason") or "")
|
||
if language == "en":
|
||
combination = str(raw.get("combination_en") or raw.get("name") or combination)
|
||
if re.search(r"[\u4e00-\u9fff]", combination):
|
||
combination = "combination text is not listed separately"
|
||
planets = [str(item) for item in _list(raw.get("planets"))]
|
||
overlap = sorted(set(planets) & set(active))
|
||
if language == "zh":
|
||
sections = {
|
||
"detected": f"本案检出：{name}。组合：{combination or '未列出'}。",
|
||
"theme": "主题层只保留传统候选。落陷取消不是王者格。这段时间可能突出相关行星守护的主题，不是确定会发生的事。",
|
||
"result": "结果只作为条件说明，不把单一瑜伽写成职位、收入或关系的保证。",
|
||
"activation": (
|
||
f"当前大运重叠：{'、'.join(_planet(item, language) for item in overlap)}。"
|
||
if overlap
|
||
else "当前三级大运没有与这条组合的行星直接重叠。"
|
||
),
|
||
}
|
||
else:
|
||
sections = {
|
||
"detected": f"Detected: {name}. Combination: {combination or 'not listed'}.",
|
||
"theme": "The theme stays a traditional candidate. Neecha Bhanga is not a Raja Yoga. This period may bring the involved planets' topics forward. It is not a certain event.",
|
||
"result": "The result is a condition note. One yoga is not a certain outcome for office, income, or relationship.",
|
||
"activation": (
|
||
f"Current Dasha overlap: {', '.join(overlap)}."
|
||
if overlap
|
||
else "The current three Dasha levels do not directly overlap the planets in this combination."
|
||
),
|
||
}
|
||
pages.append({"name": name, "sections": sections})
|
||
return {"schema": "pl9.yoga_interpretation_pack.v1", "status": "partial_verified" if pages else "blocked", "pages": pages}
|
||
|
||
|
||
def _yoga_name(raw: dict, language: str) -> str:
|
||
blob = " ".join(str(raw.get(key) or "") for key in ("name", "name_cn", "type", "category", "subtype")).lower()
|
||
if "neecha" in blob or "nicha" in blob or "落陷" in blob:
|
||
if language == "zh":
|
||
return "落陷取消（Neecha Bhanga，不是王者格）"
|
||
return "Neecha Bhanga (debilitation cancellation, not a Raja Yoga)"
|
||
if language == "en":
|
||
name = str(raw.get("name") or raw.get("type") or "Yoga")
|
||
if re.search(r"[\u4e00-\u9fff]", name):
|
||
return "Yoga (name not listed separately)"
|
||
return name
|
||
return str(raw.get("name_cn") or raw.get("name") or raw.get("type") or "瑜伽")
|
||
|
||
|
||
def build_bhavesh_interpretation_pack(
|
||
houses: dict[str, Any] | None,
|
||
chart_planets: dict[str, Any] | None,
|
||
functional_classification: dict[str, str] | None = None,
|
||
active_dasha_lords: set[str] | None = None,
|
||
*,
|
||
language: str = "zh",
|
||
) -> dict[str, Any]:
|
||
"""One page per house lord, using calculated placement and functional nature."""
|
||
|
||
houses = houses or {}
|
||
chart_planets = chart_planets or {}
|
||
functional_classification = functional_classification or {}
|
||
active = active_dasha_lords or set()
|
||
topics = _HOUSE_TOPIC_ZH if language == "zh" else _HOUSE_TOPIC_EN
|
||
pages = []
|
||
for number in range(1, 13):
|
||
house = _house_row(houses, number)
|
||
lord = house.get("lord")
|
||
lord_row = chart_planets.get(lord) if isinstance(lord, str) and isinstance(chart_planets.get(lord), dict) else {}
|
||
if not house or not lord or not lord_row:
|
||
continue
|
||
lord_house = lord_row.get("house")
|
||
role = _role(functional_classification.get(str(lord)), language)
|
||
dignity = _dignity(lord_row.get("status") or lord_row.get("dignity"), language)
|
||
sign = lord_row.get("sign") or ("未列出" if language == "zh" else "not listed")
|
||
if language == "en" and re.search(r"[\u4e00-\u9fff]", str(sign)):
|
||
sign = "not listed separately"
|
||
cusp = house.get("cusp_sign") or house.get("sign") or ("未列出" if language == "zh" else "not listed")
|
||
if language == "en" and re.search(r"[\u4e00-\u9fff]", str(cusp)):
|
||
cusp = "not listed separately"
|
||
if language == "zh":
|
||
sections = {
|
||
"house_context": f"第{number}宫：{topics[number]}。宫头为{cusp}，宫主为{_planet(lord, language)}。",
|
||
"lord_placement": f"宫主{_planet(lord, language)}位于{sign}第{lord_house if isinstance(lord_house, int) else '未列出'}宫，尊贵为{dignity}，功能吉凶为{role}。",
|
||
"result": f"第{number}宫的主题会通过宫主所在的第{lord_house if isinstance(lord_house, int) else '未列出'}宫表现。这段时间可能突出这一层，不是确定会发生的事。",
|
||
"context": "同宫与相位仍以本命表为准，这里不另算。",
|
||
"activation": (
|
||
f"{_planet(lord, language)}在当前大运里，这段时间可能突出第{number}宫主题。"
|
||
if lord in active
|
||
else f"{_planet(lord, language)}不在当前三级大运里，第{number}宫以本命背景为主。"
|
||
),
|
||
}
|
||
else:
|
||
sections = {
|
||
"house_context": f"House {number}: {topics[number]}. Cusp {cusp}. Lord {lord}.",
|
||
"lord_placement": f"Lord {lord} is in {sign}, house {lord_house if isinstance(lord_house, int) else 'not listed'}, dignity {dignity}, functional nature {role}.",
|
||
"result": f"House {number} shows through the lord's house {lord_house if isinstance(lord_house, int) else 'not listed'}. This period may bring that forward. It is not a certain event.",
|
||
"context": "Conjunctions and aspects stay in the natal tables. This page does not recompute them.",
|
||
"activation": (
|
||
f"{lord} is inside the current Dasha, so this period may bring house {number} forward."
|
||
if lord in active
|
||
else f"{lord} is not inside the current three Dasha levels, so house {number} stays natal background."
|
||
),
|
||
}
|
||
pages.append({"house": number, "lord": lord, "sections": sections})
|
||
return {"schema": "pl9.bhavesh_interpretation_pack.v1", "status": "partial_verified" if pages else "blocked", "pages": pages}
|
||
|
||
|
||
def build_dosha_remedy_pack(
|
||
yogas_doshas: dict[str, Any] | None,
|
||
chart_planets: dict[str, Any] | None = None,
|
||
ascendant: dict[str, Any] | None = None,
|
||
transit_snapshot: dict[str, Any] | None = None,
|
||
remedy_snapshot: dict[str, Any] | None = None,
|
||
ashtakavarga: dict[str, Any] | None = None,
|
||
*,
|
||
language: str = "zh",
|
||
) -> dict[str, Any]:
|
||
"""Explain dosha rows the detector already returned. Saturn cycles are not recalculated."""
|
||
|
||
yogas_doshas = yogas_doshas or {}
|
||
rows = []
|
||
labels = {
|
||
"mangal_dosha": ("火星婚姻煞", "Mangala / Kuja Dosha"),
|
||
"kaal_sarp_dosha": ("时蛇煞", "Kaal Sarp Dosha"),
|
||
"pitra_dosha": ("祖业煞", "Pitra Dosha"),
|
||
}
|
||
for key, (zh_label, en_label) in labels.items():
|
||
raw = yogas_doshas.get(key)
|
||
if not isinstance(raw, dict):
|
||
continue
|
||
present = raw.get("has_dosha") is True
|
||
title = zh_label if language == "zh" else en_label
|
||
facts = ", ".join(
|
||
f"{field}={raw.get(field)}"
|
||
for field in ("mars_house", "mars_sign", "severity", "type", "rahu_house", "ketu_house")
|
||
if raw.get(field) not in (None, "", [], {})
|
||
) or ("未列出" if language == "zh" else "not listed")
|
||
if language == "en" and re.search(r"[\u4e00-\u9fff]", facts):
|
||
facts = "not listed separately"
|
||
if language == "zh":
|
||
sections = {
|
||
"detected": f"{title}：{'本案检测到条件' if present else '本案未检测到条件'}。",
|
||
"consideration": "只核对检测器已经给出的宫位和星座，不把这一条写成关系的确定结果。",
|
||
"facts": f"结构字段：{facts}。",
|
||
"result": "若条件成立，这段时间可能突出合作里的压力或节奏；若未成立，就不把它当成主导障碍。",
|
||
"cancellation": "取消条件以检测器返回的说明为准，这里不新判。",
|
||
}
|
||
else:
|
||
sections = {
|
||
"detected": f"{title}: {'condition detected' if present else 'condition not detected'}.",
|
||
"consideration": "Only the houses and signs already returned by the detector are repeated. This is not a certain relationship outcome.",
|
||
"facts": f"Structured fields: {facts}.",
|
||
"result": "When the condition is present, this period may bring forward pressure or pace in cooperation. When it is absent, it is not treated as the leading obstacle.",
|
||
"cancellation": "Cancellation notes stay with the detector. This page does not add a new judgment.",
|
||
}
|
||
rows.append({"id": key, "title": title, "present": present, "sections": sections})
|
||
sade = yogas_doshas.get("sade_sati") if isinstance(yogas_doshas.get("sade_sati"), dict) else {}
|
||
phase = sade.get("phase") or sade.get("status") or ("未列出" if language == "zh" else "not listed")
|
||
if language == "en" and re.search(r"[\u4e00-\u9fff]", str(phase)):
|
||
phase = "not listed separately"
|
||
if language == "zh":
|
||
saturn_note = f"Sade Sati 使用检测器已给出的状态：{phase}。这里不重算土星周期。"
|
||
else:
|
||
saturn_note = f"Sade Sati uses the detector status already calculated: {phase}. This page does not recompute the Saturn cycle."
|
||
return {
|
||
"schema": "pl9.dosha_remedy_pack.v1",
|
||
"status": "partial_verified" if rows or sade else "blocked",
|
||
"dosha_rows": rows,
|
||
"saturn_note": saturn_note,
|
||
"supportive_practices": {
|
||
"recommendations": _list(_dict(remedy_snapshot).get("recommendations")),
|
||
},
|
||
"audit": {
|
||
"status": "partial_verified" if rows or sade else "blocked",
|
||
"ashtakavarga_present": bool(ashtakavarga),
|
||
"transit_present": bool(transit_snapshot),
|
||
"ascendant_present": bool(ascendant),
|
||
"chart_planets_present": bool(chart_planets),
|
||
},
|
||
}
|
||
|
||
|
||
def _field_usable(field: Any) -> bool:
|
||
if isinstance(field, list):
|
||
return bool(field)
|
||
if not isinstance(field, dict) or not field:
|
||
return False
|
||
if field.get("status") in {"blocked", "not_applicable"}:
|
||
return False
|
||
if field.get("periods") or field.get("rows") or field.get("data") or field.get("dasha_sequence"):
|
||
return True
|
||
return field.get("status") in {"partial_verified", "verified", "conflict", "parameter_sensitive"}
|
||
|
||
|
||
def project_annual_tables(annual: dict) -> None:
|
||
"""Copy existing annual fields onto the table names the reader already prints."""
|
||
|
||
if not isinstance(annual, dict):
|
||
return
|
||
normalized = annual.get("normalized_tables")
|
||
if not isinstance(normalized, dict):
|
||
normalized = {}
|
||
annual["normalized_tables"] = normalized
|
||
mudda = _dict(annual.get("mudda_dasha"))
|
||
if not normalized.get("mudda_rows"):
|
||
periods = _list(mudda.get("periods") or _dict(mudda.get("data")).get("periods") or _dict(mudda.get("data")).get("dasha_sequence"))
|
||
rows = []
|
||
for period in periods:
|
||
if not isinstance(period, dict):
|
||
continue
|
||
rows.append({
|
||
"balance_profile": period.get("balance_profile") or "native",
|
||
"lord": period.get("lord"),
|
||
"start": period.get("start") or period.get("start_date"),
|
||
"end": period.get("end") or period.get("end_date"),
|
||
"boundary_status": mudda.get("status") or "partial_verified",
|
||
})
|
||
if rows:
|
||
normalized["mudda_rows"] = rows
|
||
if not normalized.get("saham_rows"):
|
||
normalized["saham_rows"] = _saham_rows(annual.get("sahams"))
|
||
if not normalized.get("tajika_yoga_rows"):
|
||
normalized["tajika_yoga_rows"] = _tajika_rows(annual.get("tajika_yogas"))
|
||
|
||
|
||
def _saham_rows(field: Any) -> list[dict[str, Any]]:
|
||
data = field.get("data") if isinstance(field, dict) else field
|
||
rows = []
|
||
if isinstance(data, list):
|
||
source = data
|
||
elif isinstance(data, dict):
|
||
nested = data.get("rows") or data.get("sahams") or data.get("items")
|
||
source = nested if isinstance(nested, list) else [{"name": key, **value} if isinstance(value, dict) else {"name": key, "value": value} for key, value in data.items() if key not in {"status", "reason", "producer"}]
|
||
else:
|
||
source = []
|
||
for item in source:
|
||
if not isinstance(item, dict):
|
||
continue
|
||
if item.get("sign") in (None, "") and item.get("degree_in_sign") in (None, "") and item.get("longitude") in (None, ""):
|
||
continue
|
||
rows.append({
|
||
"name": item.get("name") or item.get("saham"),
|
||
"saham": item.get("saham") or item.get("name"),
|
||
"sign": item.get("sign"),
|
||
"degree_in_sign": item.get("degree_in_sign") or item.get("degree"),
|
||
})
|
||
return rows
|
||
|
||
|
||
def _tajika_rows(field: Any) -> list[dict[str, Any]]:
|
||
data = field.get("data") if isinstance(field, dict) else field
|
||
rows = []
|
||
candidates = []
|
||
if isinstance(data, list):
|
||
candidates = data
|
||
elif isinstance(data, dict):
|
||
for key in ("yogas", "rows", "named_yogas", "items", "results"):
|
||
if isinstance(data.get(key), list):
|
||
candidates = data[key]
|
||
break
|
||
for item in candidates:
|
||
if not isinstance(item, dict):
|
||
continue
|
||
rows.append({
|
||
"yoga": item.get("yoga") or item.get("name") or item.get("id"),
|
||
"name": item.get("name") or item.get("yoga"),
|
||
"present": item.get("present") if item.get("present") is not None else item.get("status") or item.get("formed"),
|
||
"forming_planets": item.get("forming_planets") or item.get("planets"),
|
||
})
|
||
return [row for row in rows if row.get("yoga") or row.get("name")]
|
||
|
||
|
||
def annual_coverage(annual: dict | None) -> list[dict[str, str]]:
|
||
"""Each upstream annual chapter is present or named with the missing field."""
|
||
|
||
annual = annual or {}
|
||
rows = []
|
||
seen_missing = set()
|
||
for title, field in ANNUAL_CHAPTERS:
|
||
usable = _field_usable(annual.get(field))
|
||
if field == "patyayini_dasha":
|
||
usable = False
|
||
if field == "sahams" and title == "Raman Sahams":
|
||
usable = bool(_dict(annual.get("normalized_tables")).get("raman_saham_rows"))
|
||
if field == "mudda_dasha":
|
||
usable = bool(_dict(annual.get("normalized_tables")).get("mudda_rows") or _field_usable(annual.get("mudda_dasha")))
|
||
if field == "tajika_yogas":
|
||
usable = bool(_dict(annual.get("normalized_tables")).get("tajika_yoga_rows") or _field_usable(annual.get("tajika_yogas")))
|
||
if usable:
|
||
rows.append({"section": title, "status": "present", "field": field})
|
||
else:
|
||
# Patyayini's two chapters share one producer. Count the field once.
|
||
status = "deferred" if field in DEFERRED_ANNUAL_FIELDS else "missing_field"
|
||
if field == "patyayini_dasha" and field in seen_missing:
|
||
status = "missing_field_same_producer"
|
||
seen_missing.add(field)
|
||
rows.append({"section": title, "status": status, "field": field})
|
||
return rows
|
||
|
||
|
||
def section_coverage(packet: dict) -> list[dict[str, str]]:
|
||
"""Catalog used by the progress note. Decision 4 dashas are out of scope."""
|
||
|
||
rows = []
|
||
for family in DECISION_4_DASHAS:
|
||
rows.append({"section": family, "status": "out_of_scope", "field": "decision_4"})
|
||
timing = _timing(packet)
|
||
advanced = _advanced(packet)
|
||
checks = (
|
||
("Dasha interpretation", bool(_list(_dict(timing.get("dasha_interpretation_pack")).get("page_series")))),
|
||
("Graha interpretation", bool(_list(_dict(advanced.get("graha_interpretation_pack")).get("planet_pages")))),
|
||
("Bhavesh interpretation", bool(_list(_dict(advanced.get("bhavesh_interpretation_pack")).get("pages")))),
|
||
("Dosha interpretation", bool(_list(_dict(advanced.get("dosha_remedy_pack")).get("dosha_rows")))),
|
||
("Yoga interpretation", bool(_list(_dict(advanced.get("yoga_interpretation_pack")).get("pages")))),
|
||
)
|
||
for title, present in checks:
|
||
rows.append({"section": title, "status": "present" if present else "missing_field", "field": title})
|
||
rows.extend(annual_coverage(_dict(timing.get("annual_tajika_pack"))))
|
||
return rows
|
||
|
||
|
||
def missing_field_count(packet: dict) -> int:
|
||
fields = {
|
||
row["field"]
|
||
for row in section_coverage(packet)
|
||
if row["status"] == "missing_field"
|
||
}
|
||
return len(fields)
|
||
|
||
|
||
def opening_focus_block(packet: dict) -> str:
|
||
"""Current MD/AD/PD plus career, wealth, and relationship house lords."""
|
||
|
||
language = _lang(packet)
|
||
lords = _current_lords(packet)
|
||
houses = _houses(packet)
|
||
planets = _planets(packet)
|
||
classification = _functional_classification(packet)
|
||
|
||
def lord_line(number: int, topic_zh: str, topic_en: str) -> str:
|
||
house = _house_row(houses, number)
|
||
lord = str(house.get("lord") or "")
|
||
row = planets.get(lord) if isinstance(planets.get(lord), dict) else {}
|
||
placed = row.get("house")
|
||
dignity = _dignity(row.get("status") or row.get("dignity"), language)
|
||
role = _role(classification.get(lord), language)
|
||
if language == "zh":
|
||
return (
|
||
f"- {topic_zh}第{number}宫主 {_planet(lord, language)} 落在第{placed if isinstance(placed, int) else '未列出'}宫，"
|
||
f"尊贵为{dignity}，功能吉凶为{role}。"
|
||
)
|
||
return (
|
||
f"- {topic_en} house {number} lord {lord or 'not listed'} is in house "
|
||
f"{placed if isinstance(placed, int) else 'not listed'}, dignity {dignity}, functional nature {role}."
|
||
)
|
||
|
||
if language == "zh":
|
||
lines = [
|
||
"重点：当前主运 MD {md}，子运 AD {ad}，次子运 PD {pd}。".format(
|
||
md=_planet(lords["md"], language),
|
||
ad=_planet(lords["ad"], language),
|
||
pd=_planet(lords["pd"], language),
|
||
),
|
||
"",
|
||
"## 重点",
|
||
lord_line(10, "事业", "Career"),
|
||
lord_line(2, "财富", "Resources"),
|
||
lord_line(7, "婚恋", "Relationship"),
|
||
"- 这段时间可能突出这些宫主所守护的主题，不是确定会发生的事。",
|
||
]
|
||
else:
|
||
lines = [
|
||
"Focus: current Maha Dasha MD {md}, Antar Dasha AD {ad}, Pratyantar Dasha PD {pd}.".format(
|
||
md=lords["md"] or "not listed",
|
||
ad=lords["ad"] or "not listed",
|
||
pd=lords["pd"] or "not listed",
|
||
),
|
||
"",
|
||
"## Focus",
|
||
lord_line(10, "事业", "Career"),
|
||
lord_line(2, "财富", "Resources"),
|
||
lord_line(7, "婚恋", "Relationship"),
|
||
"- This period may bring those house-lord topics forward. It is not a certain event.",
|
||
]
|
||
return "\n".join(lines)
|
||
|
||
|
||
def _strength_snapshot(packet: dict) -> dict:
|
||
strengths = _dict(_worksheets(packet).get("strengths_and_scores"))
|
||
shadbala = _dict(strengths.get("shadbala"))
|
||
return {"planets": _dict(shadbala.get("planets"))}
|
||
|
||
|
||
def attach_full_data_interpretation(packet: dict) -> dict:
|
||
"""Fill interpretation packs and annual table aliases on a report packet."""
|
||
|
||
if not isinstance(packet, dict):
|
||
return packet
|
||
language = _lang(packet)
|
||
worksheets = packet.setdefault("worksheets", {})
|
||
if not isinstance(worksheets, dict):
|
||
return packet
|
||
timing = worksheets.setdefault("timing_and_predictive_systems", {})
|
||
advanced = worksheets.setdefault("advanced_systems", {})
|
||
if not isinstance(timing, dict) or not isinstance(advanced, dict):
|
||
return packet
|
||
houses = _houses(packet)
|
||
planets = _planets(packet)
|
||
classification = _functional_classification(packet)
|
||
owned = _owned_houses(packet, houses)
|
||
current = _current_lords(packet)
|
||
active = {lord for lord in (current["md"], current["ad"], current["pd"]) if lord}
|
||
master = build_dasha_master_view(packet)
|
||
timing["dasha_interpretation_pack"] = build_dasha_interpretation_pack(master, {"language": language})
|
||
native_periods = []
|
||
for period in _list(_dict(_dict(master.get("families")).get("vimshottari")).get("periods")):
|
||
native_periods.append({
|
||
"period_key": f"{period.get('lord')}|{period.get('start')}|{period.get('end')}",
|
||
"lord": period.get("lord"),
|
||
"start": period.get("start"),
|
||
"end": period.get("end"),
|
||
})
|
||
timing["vimshottari_dasha_report_pack"] = build_vimshottari_dasha_report_pack(
|
||
{"periods": native_periods, "lord_facts": _lord_facts(packet)},
|
||
{},
|
||
{"language": language},
|
||
)
|
||
advanced["graha_interpretation_pack"] = build_graha_interpretation_pack(
|
||
planets,
|
||
owned,
|
||
classification,
|
||
active,
|
||
strength_snapshot=_strength_snapshot(packet),
|
||
language=language,
|
||
)
|
||
advanced["bhavesh_interpretation_pack"] = build_bhavesh_interpretation_pack(
|
||
houses,
|
||
planets,
|
||
classification,
|
||
active,
|
||
language=language,
|
||
)
|
||
yoga = _dict(advanced.get("yoga"))
|
||
dosha = _dict(advanced.get("yogas_doshas") or _worksheets(packet).get("yogas_and_doshas"))
|
||
advanced["yoga_interpretation_pack"] = build_yoga_interpretation_pack(
|
||
yoga,
|
||
dosha,
|
||
active,
|
||
language=language,
|
||
)
|
||
advanced["dosha_remedy_pack"] = build_dosha_remedy_pack(
|
||
dosha,
|
||
planets,
|
||
_dict(_dict(_worksheets(packet).get("d1_rasi_bhava")).get("ascendant")),
|
||
_dict(timing.get("transit_multi_reference")),
|
||
language=language,
|
||
)
|
||
annual = timing.get("annual_tajika_pack")
|
||
if isinstance(annual, dict):
|
||
project_annual_tables(annual)
|
||
return packet
|