Files
Jyotisha/scripts/pl9_interpretation_packs.py
T

1524 lines
75 KiB
Python
Raw 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.
"""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."),
},
}
# Traditional period topics, rewritten as possibilities. No diagnosis, death,
# lifespan, or certain reputation loss.
_PHALA: dict[str, dict[str, tuple[str, str, str]]] = {
"Sun": {
"zh": ("责任、可见职务、父亲或权威人物、土地,以及需要当场决定的公共事务", "被认可、得到行政上的协助,或把一件公共事务做完", "与权威的摩擦、离开熟悉环境办事、心里发热或坐不住,名声也可能受到议论"),
"en": ("responsibility, visible office, a father or authority figure, land, and public decisions", "recognition, administrative help, or finishing a public task", "friction with authority, leaving a familiar place for duty, heat or restlessness, and talk about reputation"),
},
"Moon": {
"zh": ("情绪节奏、母亲或照料、居住、出行、公众气氛和学习", "家里更安稳、交易或学习有进展,或换一个地方之后事情变顺", "判断摇摆、家人意见不一致、出行变密,旧的安全感变弱"),
"en": ("emotional pace, care, residence, travel, public mood, and study", "a steadier home, progress in trade or study, or a change of place that helps", "wavering judgment, family disagreement, denser travel, and a weaker sense of security"),
},
"Mars": {
"zh": ("行动、勇气、土地、兄弟姐妹、竞争和技术上的决断", "体力更够用、不动产或操作类工作有进展,竞争里更容易把事做完", "话说得硬、争执、火或器械方面的压力,冲动之下把资源花在搬迁或对抗上"),
"en": ("action, courage, land, siblings, competition, and technical decisions", "more stamina, progress in property or hands-on work, and finishing a contest", "harsh speech, quarrels, pressure around fire or tools, and spending resources on a move or a fight"),
},
"Mercury": {
"zh": ("学习、书写、交易、计算、合同和来回沟通", "文书更清楚、结盟更有用,或靠技能把问题拆开", "判断变散、文书带来额外开支,生意节奏忽快忽慢"),
"en": ("study, writing, trade, calculation, contracts, and back-and-forth talk", "clearer paperwork, more useful alliances, or taking a problem apart by skill", "scattered judgment, extra cost from paperwork, and an uneven business pace"),
},
"Jupiter": {
"zh": ("老师、判断、信仰、顾问、学习和体面的扩张", "得到尊重、顾问角色更清楚,或一件长期目标有着落", "建议不一致、目标推迟、家族义务变重"),
"en": ("teachers, judgment, belief, counsel, study, and a respectable expansion", "more respect, a clearer advisory role, or a long aim that finds a place", "conflicting advice, delayed aims, and heavier family obligations"),
},
"Venus": {
"zh": ("合作、审美、合意、资源交换、艺术和社交里的舒适", "协商更顺、作品或设计被接受,关系里的舒适感增加", "开销变大、享乐挤占正事,合作约定需要重新谈"),
"en": ("cooperation, taste, agreement, exchange, art, and social comfort", "smoother negotiation, accepted design, and more comfort in a relationship", "larger spending, pleasure crowding out work, and agreements that need to be discussed again"),
},
"Saturn": {
"zh": ("劳动、延迟、资历、界限、土地和长期功课", "工作更稳、责任被看见,后半段比开头更容易出成果", "拖延、负担变重、和熟人的距离拉开,事情要慢慢熬"),
"en": ("labor, delay, seniority, boundaries, land, and long-term work", "steadier work, visible duty, and more results in the later part than at the start", "delay, a heavier load, more distance from familiar people, and work that has to be endured"),
},
"Rahu": {
"zh": ("放大、外来机会、不按旧习惯走的路径,以及突然的转向", "非传统渠道带来开口,远方或圈层之外的事情更活跃", "想要得很急、信息前后不一致,名声可能受到议论"),
"en": ("amplification, outside opportunities, unfamiliar routes, and sudden turns", "an opening through a nontraditional channel, with distant or outside circles more active", "urgent wanting, inconsistent information, and talk about reputation"),
},
"Ketu": {
"zh": ("收束、分离、向内整理、专精和减少外界噪声", "看清一件旧目标可以放下,或在压力之后出现间歇的进展", "突然的停顿、来回奔波,或把注意力从外面收回来"),
"en": ("release, separation, inward sorting, specialization, and less outer noise", "seeing that an old aim can be set down, or intermittent progress after pressure", "a sudden pause, restless travel, or attention pulled back from the outside"),
},
}
_NODE_LORDS = {"Rahu", "Ketu"}
_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 _period_beats(lord: str, language: str) -> str:
"""Four traditional beats. Wording stays possible, not certain.
The "possible, not certain" caveat is said once at the top of the Dasha
chapter (DASHA_CHAPTER_NOTE), not repeated in every period (product 2026-10-08).
"""
bank = _PHALA.get(lord) or _PHALA["Saturn"]
topic, good, pressure = bank["en" if language == "en" else "zh"]
name = _planet(lord, language)
if language == "zh":
return (
f"这段时间可能突出的主题:传统阅读里,{name}容易把{topic}推到眼前。"
f"条件好时可能:若落宫、尊贵和当前大运同向,可能出现{good}。"
f"受压时可能表现为:若该星受压或与当前节奏不一致,可能感到{pressure}。"
f"怎么细读:先核对{name}的落宫、守护宫、尊贵、星宿和功能吉凶,"
"再看它是否出现在主运、子运或次子运。"
)
return (
f"Topics this period may bring forward: in the traditional reading, {name} may bring {topic} forward. "
f"When conditions are supportive, it may show {good}. "
f"Under pressure, it may show as {pressure}. "
f"How to read it more closely: check the house, ownership, dignity, nakshatra, and functional nature of {name}, "
"then whether it is the Maha, Antar, or Pratyantar lord."
)
DASHA_CHAPTER_NOTE = {
"zh": "以下每段大运、子运、次子运都只写这段时间可能突出的主题与条件,既不是事实陈述,也不是预言;每一段都要和宫主、分盘、星宿以及当前大运层级合看,不能单凭一段下结论。",
"en": "Every Maha, Antar and Pratyantar section below describes topics and conditions that may come forward in that period, not events that have happened or must happen. Read each with the house lords, divisional charts, nakshatra and the running period level; no single section is a conclusion on its own.",
}
def _ruled_houses_clause(lord: str, owned: list, language: str) -> str:
"""Rahu and Ketu rule no houses. Other empty lists stay an explicit gap."""
if lord in _NODE_LORDS:
return "不守护宫位" if language == "zh" else "rules no house"
if language == "zh":
if owned:
return "守护第" + "、".join(str(item) for item in owned) + "宫"
return "守护宫未单独列出"
if owned:
return "owns houses " + ", ".join(str(item) for item in owned)
return "owned houses are not listed"
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,
"chapter_note": dict(DASHA_CHAPTER_NOTE),
"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)} 主运期间,这段时间可能突出该星的自然主题。"
"下面只按已算好的落宫、守护宫、尊贵、星宿和功能吉凶来读,不是确定会发生的事。"
+ _period_beats(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. "
+ _period_beats(lord, language)
)
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')}的落宫未列出"]
bits.append(_ruled_houses_clause(lord, owned, "zh"))
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 []
ownership = _ruled_houses_clause(lord, owned, "en")
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}, {ownership}, "
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 "?"
ruled = _ruled_houses_clause(lord, owned, language)
if language == "zh":
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}宫、{ruled}、"
f"尊贵{_dignity(facts.get('dignity'), language)}与星宿{facts.get('nakshatra') or '未列出'}。"
+ _period_beats(lord, language)
)
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}, {ruled}, "
f"dignity {_dignity(facts.get('dignity'), language)}, and nakshatra {nakshatra}. "
+ _period_beats(lord, language)
)
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, language: str = "zh") -> None:
"""Copy existing annual fields onto the table names the reader already prints.
Nothing here calculates a new chart. Empty producers stay empty.
"""
if not isinstance(annual, dict):
return
language = language if language in {"zh", "en"} else "zh"
normalized = annual.get("normalized_tables")
if not isinstance(normalized, dict):
normalized = {}
annual["normalized_tables"] = normalized
mudda = _dict(annual.get("mudda_dasha"))
mudda_data = _annual_data(mudda)
if not normalized.get("mudda_rows"):
periods = _list(mudda.get("periods") or mudda_data.get("periods") or mudda_data.get("dasha_sequence"))
rows = []
for period in periods:
if not isinstance(period, dict):
continue
lord = period.get("lord") or period.get("main_lord")
rows.append({
"balance_profile": period.get("balance_profile") or mudda_data.get("balance_profile") or "native",
"lord": lord,
"start": period.get("start") or period.get("start_date"),
"end": period.get("end") or period.get("end_date"),
"boundary_status": "listed",
"reading": _mudda_reading(str(lord or ""), language),
})
for period in _list(mudda.get("antardasha_periods") or mudda_data.get("antardasha_periods")):
if not isinstance(period, dict):
continue
main_lord = period.get("main_lord")
sub_lord = period.get("sub_lord")
rows.append({
"balance_profile": "antardasha",
"lord": f"{main_lord}/{sub_lord}",
"start": period.get("start"),
"end": period.get("end"),
"boundary_status": "listed",
"reading": _mudda_reading(str(sub_lord or main_lord or ""), language),
})
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"))
_project_annual_surfaces(annual, normalized, language)
def _project_annual_surfaces(annual: dict, normalized: dict, language: str) -> None:
"""Fill reader tables from fields the pack already holds."""
solar = _annual_data(annual.get("solar_return"))
chart = _annual_data(annual.get("annual_chart"))
year_lord = _annual_data(annual.get("year_lord"))
muntha = _annual_data(annual.get("muntha"))
if not normalized.get("true_solar_return_rows"):
rows = []
for key in ("dt_local", "dt_ut", "sun_lon"):
value = solar.get(key)
if value not in (None, ""):
rows.append({"field": key, "value": value, "dt_local": solar.get("dt_local")})
if rows:
normalized["true_solar_return_rows"] = rows
if not normalized.get("panchadhikari_rows"):
rows = []
for item in _list(year_lord.get("candidates")):
if not isinstance(item, dict):
continue
rows.append({
"planet": item.get("planet"),
"role": item.get("source") or item.get("role"),
"value": item.get("panchavargiya_score"),
"sign": item.get("planet_sign"),
"house": item.get("planet_house"),
})
if not rows and year_lord.get("year_lord"):
rows.append({
"planet": year_lord.get("year_lord"),
"role": year_lord.get("selection_basis"),
"value": year_lord.get("selection_basis_text_en") if language == "en" else year_lord.get("selection_basis_text"),
"sign": year_lord.get("year_lord_sign"),
"house": year_lord.get("year_lord_house"),
})
if rows:
normalized["panchadhikari_rows"] = rows
if not normalized.get("annual_anchor_rows"):
anchors = []
if solar.get("dt_local"):
anchors.append({"name": "Varshapravesha", "label": "solar return", "value": solar.get("dt_local")})
if chart.get("asc_sign"):
anchors.append({"name": "annual Lagna", "label": "ascendant", "sign": chart.get("asc_sign"), "value": chart.get("asc_sign")})
if muntha.get("muntha_sign"):
anchors.append({"name": "Muntha", "label": "muntha", "sign": muntha.get("muntha_sign"), "house": muntha.get("muntha_house"), "value": muntha.get("muntha_lord")})
if year_lord.get("year_lord"):
anchors.append({"name": "year lord", "label": "Panchadhikari", "value": year_lord.get("year_lord")})
if anchors:
normalized["annual_anchor_rows"] = anchors
longitudes = chart.get("planet_longitudes") if isinstance(chart.get("planet_longitudes"), dict) else {}
if longitudes and not normalized.get("annual_planet_rows"):
rows = []
for planet in PLANETS:
longitude = longitudes.get(planet)
if not isinstance(longitude, (int, float)) or isinstance(longitude, bool):
continue
rows.append({"planet": planet, "longitude": longitude, "degree": round(float(longitude) % 30, 2)})
if rows:
normalized["annual_planet_rows"] = rows
yoga_rows = _list(normalized.get("tajika_yoga_rows"))
if yoga_rows and not normalized.get("tajika_presence_rows"):
normalized["tajika_presence_rows"] = [
{"yoga": row.get("yoga"), "name": row.get("name"), "present": row.get("present"), "forming_planets": row.get("forming_planets")}
for row in yoga_rows
]
normalized["tajika_condition_status_rows"] = [
{"yoga": row.get("yoga"), "name": row.get("name"), "condition": row.get("condition"), "present": row.get("present")}
for row in yoga_rows
]
normalized["tajika_forming_pair_rows"] = [
{
"yoga": row.get("yoga"),
"name": row.get("name"),
"planet_a": row.get("planet_a"),
"planet_b": row.get("planet_b"),
"forming_planets": row.get("forming_planets"),
}
for row in yoga_rows
]
strength = [
{"planet": row.get("planet_a"), "yoga": row.get("yoga"), "strength": row.get("average_deeptamsa") or row.get("residual"), "value": row.get("average_deeptamsa") or row.get("residual")}
for row in yoga_rows
if row.get("average_deeptamsa") not in (None, "") or row.get("residual") not in (None, "")
]
if strength:
normalized["tajika_strength_rows"] = strength
if not normalized.get("tajika_strength_rows"):
scored = _strength_rows(annual.get("tajika_strength"))
if scored:
normalized["tajika_strength_rows"] = scored
if not annual.get("pl9_current_year_results"):
sections = _current_year_sections(solar, chart, year_lord, muntha, language)
if sections:
annual["pl9_current_year_results"] = {"status": "partial_verified", "sections": sections}
if not annual.get("pl9_annual_dasha_results"):
dasha_rows = []
for row in _list(normalized.get("mudda_rows")):
if row.get("balance_profile") == "antardasha":
continue
dasha_rows.append({
"order": len(dasha_rows) + 1,
"lord": row.get("lord"),
"start": row.get("start"),
"end": row.get("end"),
"themes": [row.get("reading")],
})
if dasha_rows:
annual["pl9_annual_dasha_results"] = {"status": "partial_verified", "rows": dasha_rows}
def _current_year_sections(solar: dict, chart: dict, year_lord: dict, muntha: dict, language: str) -> list[dict[str, Any]]:
sections = []
if solar.get("dt_local") or chart.get("asc_sign"):
if language == "zh":
theme = "真实太阳返照已经算出。这一年可能从年返上升和返照时刻看出环境变化,不是确定会发生的事。"
else:
theme = "The solar return is already calculated. The year may show a change of setting from the annual ascendant and the return time. This is not a certain event."
sections.append({"title": "solar return", "anchor": {"dt_local": solar.get("dt_local"), "asc_sign": chart.get("asc_sign")}, "themes": [theme]})
lord = year_lord.get("year_lord")
if lord:
if language == "zh":
theme = f"年主是{_planet(lord, 'zh')}。条件好时可能突出该星的年度主题;受压时可能表现为节奏需要重看。年主依据来自 Panchadhikari,不是另选一颗星。"
else:
theme = f"The year lord is {lord}. Supportive conditions may bring that planet's annual topics forward. Under pressure the pace may need another look. The basis is Panchadhikari."
sections.append({"title": "year lord", "anchor": {"year_lord": lord}, "themes": [theme]})
if muntha.get("muntha_sign") or muntha.get("muntha_lord"):
if language == "zh":
theme = "Muntha 标出这一年更容易被注意到的宫位。它可能加重该宫主题,不是单独的事件结论。"
else:
theme = "Muntha marks the house that may be easier to notice this year. It may emphasize that house. It is not a separate event conclusion."
sections.append({"title": "Muntha", "anchor": {"sign": muntha.get("muntha_sign"), "lord": muntha.get("muntha_lord")}, "themes": [theme]})
return sections
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
planets = item.get("forming_planets") or item.get("planets")
pair = list(planets) if isinstance(planets, list) else []
rows.append({
"yoga": item.get("yoga") or item.get("named_yoga") or item.get("name") or item.get("candidate_name") or item.get("id"),
"name": item.get("name") or item.get("named_yoga") or item.get("candidate_name") or item.get("yoga"),
"present": item.get("present") if item.get("present") is not None else item.get("motion") or item.get("formed"),
"forming_planets": planets,
"motion": item.get("motion"),
"aspect": item.get("aspect"),
"average_deeptamsa": item.get("average_deeptamsa"),
"residual": item.get("residual"),
"planet_a": pair[0] if pair else None,
"planet_b": pair[1] if len(pair) > 1 else None,
"condition": item.get("condition") or item.get("motion") or item.get("aspect"),
})
return [row for row in rows if row.get("yoga") or row.get("name")]
def _annual_data(field: Any) -> dict:
if not isinstance(field, dict):
return {}
data = field.get("data")
return data if isinstance(data, dict) else field
def _mudda_reading(lord: str, language: str) -> str:
name = _planet(lord, "zh" if language == "zh" else "en")
if language == "zh":
return f"{name}这一段可能突出该星的年度主题。条件好时可能事情更顺,受压时可能表现为节奏变慢或需要重谈。这不是确定会发生的事。"
return f"The {lord} span may bring that planet's annual topics forward. Supportive conditions may make the work easier. Under pressure the pace may slow or an agreement may need another look. This is not a certain event."
def _strength_rows(field: Any) -> list[dict[str, Any]]:
"""Numeric Harsha and five-fold scores only. Do not print the internal status word."""
data = _annual_data(field)
rows = []
for layer, key in (("Harsha", "harsha_bala"), ("five-fold", "panchavargiya_bala")):
table = data.get(key)
if not isinstance(table, dict):
continue
for planet, item in table.items():
if not isinstance(item, dict):
continue
score = item.get("score")
if not isinstance(score, (int, float)) or isinstance(score, bool):
continue
rows.append({"planet": planet, "yoga": layer, "strength": score, "value": score})
return rows
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, language)
return packet