jesse-uxandGrok (agent loop)
d45e65a9ec
研究章在渲染时计算，不写入资料包。四种星座大运次小运和 Kalachakra 第三层不改原来的主运、子运。D81、D108、D144 只留在研究章。 自测：test_research_timing、星座大运、域读取、卫生预算、真恒星、jhora 扫描、API 增长、六章标题、英文版。三张虚构盘中英出稿都低于 180 秒。PyJHora 已导入但未做逐层对照。 Co-Authored-By: Grok (agent loop)
501 lines
23 KiB
Python
501 lines
23 KiB
Python
"""Snake-free bilingual chapters for the research appendix.
|
|
|
|
The chapters are computed while the report is rendered. They are not stored
|
|
on the packet, and they are not sent to chat cards or rectification.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import logging
|
|
import re
|
|
from datetime import datetime, timedelta, timezone
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
try:
|
|
from report_render.parity_chapters import _first, _major_list
|
|
from report_render.text import PLANET_LABEL_CN, SIGNS_CN
|
|
from research_timing.bundle import chart_from_parts, research_bundle
|
|
from research_timing.rashi_pd import attach_rashi_pratyantar
|
|
from rashi_dashas_native import calculate_rashi_dashas
|
|
except ImportError: # pragma: no cover - package import
|
|
from scripts.report_render.parity_chapters import _first, _major_list
|
|
from scripts.report_render.text import PLANET_LABEL_CN, SIGNS_CN
|
|
from scripts.research_timing.bundle import chart_from_parts, research_bundle
|
|
from scripts.research_timing.rashi_pd import attach_rashi_pratyantar
|
|
from scripts.rashi_dashas_native import calculate_rashi_dashas
|
|
|
|
try:
|
|
from annual_native.dasha_children import attach_kalachakra_simplified_children
|
|
except ImportError: # pragma: no cover - package import
|
|
from scripts.annual_native.dasha_children import attach_kalachakra_simplified_children
|
|
|
|
BANNER = {
|
|
"zh": "研究口径：依赖所选方法，未经外部软件核对。",
|
|
"en": "Research basis: method-dependent; not checked against external software.",
|
|
}
|
|
_CHILD_KEYS = ("antardashas", "pratyantardashas", "antardasha", "pratyantar", "pratyantardasha")
|
|
_LEVELS = {"MD": "MD", "AD": "AD", "PD": "PD", "antardasha": "AD", "pratyantardasha": "PD", "pratyantar": "PD"}
|
|
_FIXED = {
|
|
"tithi-natal-forward": ("Natal Tithi Ashtottari forward", "本命 Tithi Ashtottari 顺行"),
|
|
"tithi-natal-reverse": ("Natal Tithi Ashtottari reverse", "本命 Tithi Ashtottari 逆行"),
|
|
"yoga-natal": ("Natal Yoga Vimshottari", "本命 Yoga Vimshottari"),
|
|
"saptavarga": ("Saptavarga lords", "七分盘星座主"),
|
|
"buddhi-gati": ("Buddhi Gati", "Buddhi Gati"),
|
|
"kaala": ("Kaala", "Kaala"),
|
|
"sudasa-confirmed": ("Sudasa confirmed start", "Sudasa 已确认起点"),
|
|
"sudasa-birth": ("Sudasa birth cycle", "Sudasa 出生周期"),
|
|
"varnada": ("Varnada", "Varnada"),
|
|
"karana": ("Karana year", "Karana 年"),
|
|
}
|
|
_TECH = {"tithi": "Tithi Ashtottari", "yoga": "Yoga Pravesha", "nakshatra": "Nakshatra Pravesha"}
|
|
_KIND = {
|
|
"en": {
|
|
"arc": "arc balance", "time": "elapsed balance", "successor": "successor arc",
|
|
"tithi-year": "360 tithi year", "forward": "forward", "reverse": "reverse",
|
|
},
|
|
"zh": {
|
|
"arc": "角距", "time": "时间余额", "successor": "后继角距",
|
|
"tithi-year": "360 太阴月年", "forward": "顺行", "reverse": "逆行",
|
|
},
|
|
}
|
|
_KARANA = {
|
|
"julene": ("Julene lords", "Julene 主星"),
|
|
"pvr": ("PVR quadrant lords", "PVR 象限主星"),
|
|
"pvr-tithi-year": ("PVR 360 tithi year", "PVR 360 太阴月年"),
|
|
}
|
|
_RASHI = (
|
|
("sthira", "Sthira", "Sthira"),
|
|
("drig", "Drig", "Drig"),
|
|
("shoola", "Shoola", "Shoola"),
|
|
("navamsha", "Navamsha", "Navamsha"),
|
|
)
|
|
_DIVISION = {1: "D1", 2: "D2", 3: "D3", 7: "D7", 9: "D9", 12: "D12", 30: "D30", 81: "D81", 108: "D108", 144: "D144", 150: "D150"}
|
|
|
|
|
|
def render_research_chapters(packet: dict, language: str) -> str:
|
|
"""Three research chapters. One failed family does not remove the others."""
|
|
if language not in ("zh", "en"):
|
|
raise ValueError("language must be zh or en")
|
|
try:
|
|
chart = context_from_packet(packet if isinstance(packet, dict) else {})
|
|
bundle = research_bundle(chart)
|
|
except (ValueError, RuntimeError, KeyError, TypeError) as exc:
|
|
logger.warning("research timing chapter unavailable: %s: %s", type(exc).__name__, exc)
|
|
text = _chapter(
|
|
"Additional timing calculations" if language == "en" else "补充推运计算数据",
|
|
language,
|
|
_reason(language, "This chapter did not finish. The runtime log has the cause.", "这一章没有算成，原因记在运行日志里。"),
|
|
)
|
|
return text
|
|
timing = _timing_chapter(bundle, language)
|
|
auxiliary = _auxiliary_chapter(packet, chart, language)
|
|
vargas = _varga_chapter(bundle, language)
|
|
return "\n".join(part for part in (timing, auxiliary, vargas) if part)
|
|
|
|
|
|
def context_from_packet(packet: dict) -> dict:
|
|
birth = packet.get("birth_info") if isinstance(packet.get("birth_info"), dict) else {}
|
|
profile = packet.get("calculation_profile") if isinstance(packet.get("calculation_profile"), dict) else {}
|
|
settings = profile.get("effective_settings") if isinstance(profile.get("effective_settings"), dict) else {}
|
|
year, month, day = _birth_date(birth)
|
|
hour = int(birth.get("hour") if birth.get("hour") is not None else 0)
|
|
minute = int(birth.get("minute") if birth.get("minute") is not None else 0)
|
|
second = int(birth.get("second") if birth.get("second") is not None else 0)
|
|
if birth.get("lat") is None and birth.get("latitude") is None:
|
|
raise ValueError("birth place is missing")
|
|
if birth.get("lon") is None and birth.get("longitude") is None:
|
|
raise ValueError("birth place is missing")
|
|
lat = float(birth.get("lat") if birth.get("lat") is not None else birth.get("latitude"))
|
|
lon = float(birth.get("lon") if birth.get("lon") is not None else birth.get("longitude"))
|
|
tz = _tz(settings.get("timezone_offset") if settings.get("timezone_offset") not in (None, "") else birth.get("tz", birth.get("timezone")))
|
|
ayanamsa = str(birth.get("ayanamsa_name") or birth.get("ayanamsa") or settings.get("ayanamsa") or "lahiri")
|
|
node_mode = str(birth.get("node_mode") or settings.get("node_mode") or "mean")
|
|
year_days = float(settings.get("dasha_year_days") or 365.25)
|
|
target_year, query = _query(packet, year, tz)
|
|
return chart_from_parts(
|
|
year=year, month=month, day=day, hour=hour, minute=minute, second=second,
|
|
lat=lat, lon=lon, tz=tz, ayanamsa=ayanamsa, node_mode=node_mode,
|
|
target_year=target_year, query=query, year_days=year_days,
|
|
)
|
|
|
|
|
|
def _timing_chapter(bundle: dict, language: str) -> str:
|
|
parts = [_chapter("Additional timing calculations" if language == "en" else "补充推运计算数据", language, "")]
|
|
note = bundle["reference_note_en"] if language == "en" else bundle["reference_note_zh"]
|
|
counts = _return_counts(bundle)
|
|
count_line = (
|
|
f"Tithi returns across the target year and the next two years: {counts['tithi']}. "
|
|
f"Yoga returns in the target year: {counts['yoga']}. "
|
|
f"Nakshatra returns in the target year: {counts['nakshatra']}."
|
|
if language == "en"
|
|
else (
|
|
f"目标年及后两年的 Tithi 回归次数：{counts['tithi']}。"
|
|
f"目标年的 Yoga 回归次数：{counts['yoga']}。"
|
|
f"目标年的 Nakshatra 回归次数：{counts['nakshatra']}。"
|
|
)
|
|
)
|
|
parts.append(note + "\n\n" + count_line)
|
|
families = bundle.get("families") or {}
|
|
for key, family in families.items():
|
|
if key in ("annual-returns", "research-vargas"):
|
|
continue
|
|
if key == "karana":
|
|
groups = family.get("returns") if isinstance(family, dict) else None
|
|
if isinstance(groups, list):
|
|
for group in groups:
|
|
number = group.get("index") or ""
|
|
suffix = f"return {number}" if language == "en" else f"第 {number} 次"
|
|
parts.extend(_variant_sections(group, language, "Karana", suffix))
|
|
else:
|
|
parts.extend(_variant_sections(family, language, "Karana", ""))
|
|
continue
|
|
if key == "varnada" and isinstance(family.get("variants"), list):
|
|
parts.extend(_variant_sections(family, language, "Varnada"))
|
|
continue
|
|
title = _heading(key, language)
|
|
if not title:
|
|
continue
|
|
parts.append(_family_section(title, family, language, bundle.get("query")))
|
|
return "\n\n".join(parts)
|
|
|
|
|
|
def _auxiliary_chapter(packet: dict, chart: dict, language: str) -> str:
|
|
parts = [_chapter("Auxiliary dasha subperiods" if language == "en" else "辅助大运次小运数据", language, "")]
|
|
birth = datetime(chart["year"], chart["month"], chart["day"]).date()
|
|
longitudes = chart["longitudes"]
|
|
original = calculate_rashi_dashas(birth, chart["ascendant"], longitudes)
|
|
copied = attach_rashi_pratyantar(original)
|
|
rule = (
|
|
"The third level starts at the sub-period sign and runs forward, splitting that sub-period into twelve equal parts. Major and sub-period rows stay as they were."
|
|
if language == "en"
|
|
else "次小运从子运星座起顺行，把该子运的时间等分成十二段。主运和子运与原来的表相同。"
|
|
)
|
|
for key, en, zh in _RASHI:
|
|
title = en if language == "en" else zh
|
|
section = copied.get(key) if isinstance(copied.get(key), dict) else {}
|
|
periods = _third_only(section.get("periods") or [])
|
|
body = rule + "\n\n" + (_table(periods, language) if periods else _reason(
|
|
language,
|
|
"This system has no sub-period rows, so the third level is not listed.",
|
|
"这个体系没有子运，所以不列次小运。",
|
|
))
|
|
parts.append(_family_block(title, language, body))
|
|
niryaana = (
|
|
"Niryaana Shoola child periods and the third level are omitted. The upstream program states that the child-period rule is not independently specified. Rows from an unpublished frozen engine are not a written source."
|
|
if language == "en"
|
|
else "Niryaana Shoola 的子运和次小运这次不列。上游主程序写明子运规则没有单独的出处，冻结引擎里的几行不能当作书页依据。"
|
|
)
|
|
parts.append(_family_block("Niryaana Shoola", language, niryaana))
|
|
parts.append(_family_block("Kalachakra", language, _kalachakra(packet, language)))
|
|
return "\n\n".join(parts)
|
|
|
|
|
|
def _varga_chapter(bundle: dict, language: str) -> str:
|
|
family = (bundle.get("families") or {}).get("research-vargas") or {}
|
|
rows = family.get("rows") or []
|
|
intro = (
|
|
"D81, D108 and D144 are printed only in this chapter."
|
|
if language == "en"
|
|
else "D81、D108、D144 只出现在这一章。"
|
|
)
|
|
body = intro if family.get("status") == "calculated" else _text(family, language)
|
|
if rows:
|
|
body += "\n\n" + _varga_table(rows, language)
|
|
return _chapter("Research divisional charts" if language == "en" else "研究型分盘", language, body)
|
|
|
|
|
|
def _variant_sections(family: dict, language: str, fallback: str, suffix: str = "") -> list[str]:
|
|
variants = family.get("variants")
|
|
if not isinstance(variants, list) or not variants:
|
|
return [_family_block(fallback, language, _text(family, language))]
|
|
sections = []
|
|
for index, variant in enumerate(variants, start=1):
|
|
profile = variant.get("profile")
|
|
if profile in _KARANA:
|
|
title = _KARANA[profile][0 if language == "en" else 1]
|
|
else:
|
|
seed = variant.get("seed_sign")
|
|
title = f"{fallback} {_name(seed, language)}".strip() if seed else f"{fallback} {index}"
|
|
if suffix:
|
|
title = f"{title} {suffix}"
|
|
sections.append(_family_section(title, variant, language, None))
|
|
return sections
|
|
|
|
|
|
def _family_section(title: str, family: dict, language: str, query: str | None) -> str:
|
|
body = [_text(family, language)]
|
|
periods = family.get("periods") or []
|
|
if periods and family.get("status") != "omitted":
|
|
if query:
|
|
current = _current(periods, query)
|
|
if current:
|
|
label = "Reference row" if language == "en" else "参考时刻所在"
|
|
body.append(label + "\n\n" + _table(current, language))
|
|
body.append(_table(periods, language))
|
|
rows = family.get("rows")
|
|
if rows:
|
|
body.append(_varga_table(rows, language))
|
|
return _family_block(title, language, "\n\n".join(part for part in body if part))
|
|
|
|
|
|
def _third_only(periods: list[dict]) -> list[dict]:
|
|
rows = []
|
|
for major in periods:
|
|
if not isinstance(major, dict):
|
|
continue
|
|
for child in major.get("antardasha") or []:
|
|
third = child.get("pratyantar") if isinstance(child, dict) else None
|
|
if isinstance(third, list) and third:
|
|
rows.extend(third)
|
|
return rows
|
|
|
|
|
|
def _kalachakra(packet: dict, language: str) -> str:
|
|
note = (
|
|
"The third level uses the simplified model, split in proportion from the majors already on the chart. Major and sub-period rows stay as they were."
|
|
if language == "en"
|
|
else "次小运按简化模型，从现有主运按比例分出。主运和子运与原来的表相同。"
|
|
)
|
|
module = _first(packet, ("timing_and_predictive_systems", "kalachakra_dasha"))
|
|
rows = [item for item in _major_list(module) if isinstance(item, dict) and "antardasha" not in item and "subperiods" not in item]
|
|
if not rows:
|
|
rows = [item for item in _major_list(module) if isinstance(item, dict)]
|
|
if not rows:
|
|
missing = "The chart has no Kalachakra majors, so the third level is not listed." if language == "en" else "这张盘没有 Kalachakra 主运，所以不列次小运。"
|
|
return note + "\n\n" + missing
|
|
try:
|
|
_enriched, _children, grands = attach_kalachakra_simplified_children(rows)
|
|
except (KeyError, TypeError, ValueError) as exc:
|
|
logger.warning("research kalachakra third level failed: %s: %s", type(exc).__name__, exc)
|
|
failed = "The third level did not finish. The runtime log has the cause." if language == "en" else "次小运没有算成，原因记在运行日志里。"
|
|
return note + "\n\n" + failed
|
|
shaped = [{
|
|
"level": "PD",
|
|
"sign": item.get("rashi") or item.get("sign"),
|
|
"lord": item.get("lord"),
|
|
"start": item.get("start_date") or item.get("start"),
|
|
"end": item.get("end_date") or item.get("end"),
|
|
} for item in grands if isinstance(item, dict)]
|
|
if not shaped:
|
|
empty = "No third-level rows were produced." if language == "en" else "没有产生次小运行。"
|
|
return note + "\n\n" + empty
|
|
return note + "\n\n" + _table(shaped, language)
|
|
|
|
|
|
def _chapter(title: str, language: str, body: str) -> str:
|
|
text = f"## {title}\n\n{BANNER[language]}"
|
|
if body:
|
|
text += "\n\n" + body.strip()
|
|
return text
|
|
|
|
|
|
def _family_block(title: str, language: str, body: str) -> str:
|
|
return f"## {title}\n\n{BANNER[language]}\n\n{body.strip()}"
|
|
|
|
|
|
def _heading(key: str, language: str) -> str:
|
|
if key in _FIXED:
|
|
en, zh = _FIXED[key]
|
|
return en if language == "en" else zh
|
|
matched = re.fullmatch(r"tithi-(\d{4})-(\d+)-pair-(\d+)-(arc|time)-(forward|reverse)", key)
|
|
if matched:
|
|
year, number, pair, kind, direction = matched.groups()
|
|
if language == "zh":
|
|
return (
|
|
f"年度 Tithi Ashtottari {_KIND['zh'][kind]} {_KIND['zh'][direction]} "
|
|
f"{year} 年第 {number} 次，配对 {pair}"
|
|
)
|
|
return (
|
|
f"Annual Tithi Ashtottari {_KIND['en'][kind]} {_KIND['en'][direction]} "
|
|
f"{year} return {number} pair {pair}"
|
|
)
|
|
matched = re.fullmatch(r"tithi-(\d{4})-(\d+)", key)
|
|
if matched:
|
|
year, number = matched.groups()
|
|
if language == "zh":
|
|
return f"年度 Tithi Ashtottari {year} 年第 {number} 次"
|
|
return f"Annual Tithi Ashtottari {year} return {number}"
|
|
matched = re.fullmatch(r"(yoga|nakshatra)-(\d+)-successor-(\d+)", key)
|
|
if matched:
|
|
tech, number, pair = matched.groups()
|
|
if language == "zh":
|
|
return f"年度 {_TECH[tech]} 后继角距 第 {number} 次，配对 {pair}"
|
|
return f"Annual {_TECH[tech]} successor arc return {number} pair {pair}"
|
|
matched = re.fullmatch(r"(yoga|nakshatra)-(\d+)-tithi-year", key)
|
|
if matched:
|
|
tech, number = matched.groups()
|
|
if language == "zh":
|
|
return f"年度 {_TECH[tech]} 360 太阴月年 第 {number} 次"
|
|
return f"Annual {_TECH[tech]} 360 tithi year return {number}"
|
|
matched = re.fullmatch(r"(yoga|nakshatra)-(\d+)-successor", key)
|
|
if matched:
|
|
tech, number = matched.groups()
|
|
if language == "zh":
|
|
return f"年度 {_TECH[tech]} 后继角距 第 {number} 次"
|
|
return f"Annual {_TECH[tech]} successor arc return {number}"
|
|
return ""
|
|
|
|
|
|
def _text(family: dict, language: str) -> str:
|
|
if not isinstance(family, dict):
|
|
return ""
|
|
custom = family.get("reason_en") if language == "en" else family.get("reason_zh")
|
|
if custom:
|
|
return str(custom)
|
|
if family.get("pd_status") in ("omitted", "blocked"):
|
|
return _reason(
|
|
language,
|
|
"No third level is listed for this method.",
|
|
"这个方法这里不列次小运。",
|
|
)
|
|
if family.get("status") == "omitted":
|
|
return _reason(language, "This branch is not listed.", "这一支不列表。")
|
|
return ""
|
|
|
|
|
|
def _reason(language: str, en: str, zh: str) -> str:
|
|
return en if language == "en" else zh
|
|
|
|
|
|
def _table(periods: list[dict], language: str) -> str:
|
|
headers = ("Level", "Planet or name", "Start", "End") if language == "en" else ("层级", "主星/名称", "起始", "结束")
|
|
lines = ["| " + " | ".join(headers) + " |", "| --- | --- | --- | --- |"]
|
|
|
|
def walk(rows: list, fallback: str) -> None:
|
|
for row in rows:
|
|
if not isinstance(row, dict):
|
|
continue
|
|
level = _LEVELS.get(str(row.get("level") or fallback), fallback)
|
|
label = "MD" if language == "en" else "主运"
|
|
if level == "AD":
|
|
label = "AD" if language == "en" else "子运"
|
|
elif level == "PD":
|
|
label = "PD" if language == "en" else "次小运"
|
|
name = row.get("lord") or row.get("planet") or row.get("sign") or ""
|
|
lines.append("| " + " | ".join((
|
|
label,
|
|
_name(name, language),
|
|
_plain(row.get("start")),
|
|
_plain(row.get("end")),
|
|
)) + " |")
|
|
for key in _CHILD_KEYS:
|
|
children = row.get(key)
|
|
if isinstance(children, list) and children:
|
|
walk(children, "AD" if level == "MD" else "PD")
|
|
|
|
walk(periods, "MD")
|
|
if len(lines) <= 2:
|
|
return ""
|
|
return "\n".join(lines)
|
|
|
|
|
|
def _current(periods: list[dict], query: str) -> list[dict]:
|
|
moment = datetime.fromisoformat(query)
|
|
found: list[dict] = []
|
|
|
|
def walk(rows: list) -> bool:
|
|
for row in rows:
|
|
if not isinstance(row, dict) or "start" not in row or "end" not in row:
|
|
continue
|
|
try:
|
|
start = _align(datetime.fromisoformat(str(row["start"])), moment)
|
|
end = _align(datetime.fromisoformat(str(row["end"])), moment)
|
|
except (TypeError, ValueError):
|
|
continue
|
|
if start <= moment < end:
|
|
found.append({key: row.get(key) for key in ("level", "lord", "planet", "sign", "start", "end")})
|
|
for key in _CHILD_KEYS:
|
|
children = row.get(key)
|
|
if isinstance(children, list) and walk(children):
|
|
break
|
|
return True
|
|
return False
|
|
|
|
walk(periods)
|
|
return found
|
|
|
|
|
|
def _align(value: datetime, sample: datetime) -> datetime:
|
|
if value.utcoffset() is None and sample.utcoffset() is not None:
|
|
return value.replace(tzinfo=sample.tzinfo)
|
|
if value.utcoffset() is not None and sample.utcoffset() is None:
|
|
return value.replace(tzinfo=None)
|
|
return value
|
|
|
|
|
|
def _varga_table(rows: list[dict], language: str) -> str:
|
|
headers = ("Division", "Body", "Sign", "Degree", "Sign lord") if language == "en" else ("分盘", "星体", "星座", "度数", "星座主")
|
|
lines = ["| " + " | ".join(headers) + " |", "| --- | --- | --- | --- | --- |"]
|
|
for row in rows:
|
|
division = _DIVISION.get(int(row.get("division") or 0), str(row.get("division") or ""))
|
|
degree = row.get("degree")
|
|
degree_text = f"{float(degree):.4f}" if isinstance(degree, (int, float)) else ""
|
|
lines.append("| " + " | ".join((
|
|
division,
|
|
_name(row.get("body"), language),
|
|
_name(row.get("sign"), language),
|
|
degree_text,
|
|
_name(row.get("lord"), language),
|
|
)) + " |")
|
|
return "\n".join(lines)
|
|
|
|
|
|
def _name(value, language: str) -> str:
|
|
text = "" if value is None else str(value)
|
|
if language == "zh":
|
|
text = PLANET_LABEL_CN.get(text, SIGNS_CN.get(text, text))
|
|
if text == "Lagna":
|
|
text = "上升"
|
|
return _plain(text)
|
|
|
|
|
|
def _plain(value) -> str:
|
|
return str(value or "").replace("|", "/").replace("\n", " ")
|
|
|
|
|
|
def _return_counts(bundle: dict) -> dict[str, int]:
|
|
found = ((bundle.get("families") or {}).get("annual-returns") or {}).get("returns") or {}
|
|
return {key: len(found.get(key) or []) for key in ("tithi", "yoga", "nakshatra")}
|
|
|
|
|
|
def _birth_date(birth: dict) -> tuple[int, int, int]:
|
|
if all(birth.get(key) is not None for key in ("year", "month", "day")):
|
|
return int(birth["year"]), int(birth["month"]), int(birth["day"])
|
|
text = str(birth.get("date") or "")
|
|
year, month, day = text.split("-")[:3]
|
|
return int(year), int(month), int(day)
|
|
|
|
|
|
def _tz(value) -> float:
|
|
if isinstance(value, (int, float)) and not isinstance(value, bool):
|
|
return float(value)
|
|
text = str(value or "").strip().upper().replace("UTC", "")
|
|
matched = re.fullmatch(r"([+-])(\d{2}):(\d{2})", text)
|
|
if matched:
|
|
sign = 1 if matched.group(1) == "+" else -1
|
|
return sign * (int(matched.group(2)) + int(matched.group(3)) / 60.0)
|
|
if not text:
|
|
raise ValueError("timezone is missing")
|
|
return float(text)
|
|
|
|
|
|
def _query(packet: dict, birth_year: int, tz: float) -> tuple[int, datetime]:
|
|
pack = _first(packet, ("timing_and_predictive_systems", "annual_tajika_pack"))
|
|
profile = pack.get("profile") if isinstance(pack, dict) else {}
|
|
target = profile.get("target_year") if isinstance(profile, dict) else None
|
|
if target in (None, ""):
|
|
target = packet.get("birth_info", {}).get("target_year") if isinstance(packet.get("birth_info"), dict) else None
|
|
ephemeris = _first(packet, ("timing_and_predictive_systems", "report_month_ephemeris"))
|
|
rows = ephemeris.get("report_rows") if isinstance(ephemeris, dict) else None
|
|
month_text = rows[0].get("month") if isinstance(rows, list) and rows and isinstance(rows[0], dict) else None
|
|
zone = timezone(timedelta(hours=tz))
|
|
if isinstance(month_text, str) and len(month_text) >= 7 and month_text[4] == "-":
|
|
year = int(month_text[0:4])
|
|
month = int(month_text[5:7])
|
|
if target in (None, ""):
|
|
target = year
|
|
return int(target), datetime(year, month, 1, 12, 0, tzinfo=zone)
|
|
year = int(target if target not in (None, "") else birth_year)
|
|
return year, datetime(year, 1, 1, 12, 0, tzinfo=zone)
|