Jyotisha/scripts/research_timing/render.py

"""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 json
import logging
import re
from datetime import datetime, timedelta, timezone
from pathlib import Path

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:
    note = _pyjhora_note(title, language)
    text = f"## {title}\n\n{BANNER[language]}"
    if note:
        text += "\n\n" + note
    if body.strip():
        text += "\n\n" + body.strip()
    return text


_PYJHORA_NOTES: dict | None = None


def _pyjhora_notes() -> dict:
    """Committed offline comparison. Missing or unread JSON adds no label."""
    global _PYJHORA_NOTES
    if _PYJHORA_NOTES is None:
        path = Path(__file__).resolve().parents[2] / "docs" / "tasks" / "research-timing-pyjhora-20261023.json"
        notes: dict[tuple[str, str], str] = {}
        try:
            payload = json.loads(path.read_text(encoding="utf-8"))
        except (OSError, json.JSONDecodeError):
            payload = {}
        systems = payload.get("systems") if isinstance(payload, dict) else None
        for item in (systems or {}).values():
            if not isinstance(item, dict) or item.get("status") != "不一致":
                continue
            zh = str(item.get("heading_zh") or "")
            en = str(item.get("heading_en") or "")
            category = str(item.get("category") or "")
            category_en = str(item.get("category_en") or "")
            match = item.get("match") or "exact"
            if zh and category:
                notes[("zh", zh, match)] = f"与 PyJHora 不一致：差在 {category}"
            if en and category_en:
                notes[("en", en, match)] = f"Differs from PyJHora: {category_en}"
        _PYJHORA_NOTES = notes
    return _PYJHORA_NOTES


def _pyjhora_note(title: str, language: str) -> str:
    best = ""
    found = ""
    for (lang, heading, match), sentence in _pyjhora_notes().items():
        if lang != language or not heading:
            continue
        if title == heading or (match == "prefix" and title.startswith(heading + " ")):
            if len(heading) >= len(best):
                best = heading
                found = sentence
    return found


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)