Jyotisha/scripts/report_render/annual_method_reference.py

"""Parent side of the full-data annual method reference.

This module does not import PyJHora. The worker runs in a child process.
The worker module name appears only as the ``-m`` argument below.
"""

from __future__ import annotations

import hashlib
import json
import logging
import subprocess
import sys
from datetime import datetime, timedelta
from pathlib import Path
from typing import Any, Callable

log = logging.getLogger("jyotisha.annual_method_reference")

REPO_ROOT = Path(__file__).resolve().parents[2]
WORKER_MODULE = "scripts.report_render.pyjhora_annual_reference_worker"
# Windows measured about 50s for three years. 45s killed a finished call.
# Linux is expected to be faster. 120s still leaves the report under the 180s wait.
DEFAULT_TIMEOUT_SECONDS = 120
TABLES = ("mudda_birth", "mudda_annual", "patyayini", "tajika")
CHAPTER_ZH = "独立年度方法参考"
CHAPTER_EN = "Independent Annual Method Reference"
DASHA_LORDS = {
    0: "Sun",
    1: "Moon",
    2: "Mars",
    3: "Mercury",
    4: "Jupiter",
    5: "Venus",
    6: "Saturn",
    7: "Rahu",
    8: "Ketu",
}
YOGA_NAMES = {
    "ithasala": "Ithasala",
    "eesarpha": "Esharpha",
    "manahoo": "Manahoo",
    "kamboola": "Kamboola",
    "gairi_kamboola": "Gairi-Kamboola",
    "khallasara": "Khallsara",
    "radda": "Radda",
    "nakta": "Nakta",
    "yamaya": "Yamaya",
}
YOGA_ORDER = (
    "ithasala",
    "eesarpha",
    "manahoo",
    "kamboola",
    "gairi_kamboola",
    "khallasara",
    "radda",
    "nakta",
    "yamaya",
)
ZH_PLANETS = {
    "Sun": "太阳",
    "Moon": "月亮",
    "Mars": "火星",
    "Mercury": "水星",
    "Jupiter": "木星",
    "Venus": "金星",
    "Saturn": "土星",
    "Rahu": "罗睺",
    "Ketu": "计都",
    "Lagna": "上升",
}

_CACHE: dict[str, dict[str, Any]] = {}
_RUNNER: Callable[[dict[str, Any], int], dict[str, Any]] | None = None


def set_reference_runner(runner: Callable[[dict[str, Any], int], dict[str, Any]] | None) -> None:
    """Tests and the 200-chart sweep install a stand-in. ``None`` restores the child process."""
    global _RUNNER
    _RUNNER = runner


def reset_reference_cache() -> None:
    _CACHE.clear()


def mask_report_fingerprint_lines(markdown: str) -> str:
    """Blank the report-id and result-fingerprint lines before a byte compare.

    Including this chapter changes ``result_hash``, so those two identity lines
    move. Everything else, after the chapter itself is stripped, must not.
    """
    output: list[str] = []
    for line in (markdown or "").splitlines(keepends=True):
        body = line.rstrip("\r\n")
        ending = line[len(body):]
        stripped = body.strip()
        if stripped.startswith("- 报告 ID: `") and stripped.endswith("`"):
            output.append(f"- 报告 ID: `FINGERPRINT`{ending}")
        elif stripped.startswith("- 计算结果摘要: `") and stripped.endswith("`"):
            output.append(f"- 计算结果摘要: `FINGERPRINT`{ending}")
        else:
            output.append(line)
    return "".join(output)


def strip_annual_method_chapter(markdown: str) -> str:
    """Drop this chapter so older full-text checks still see the previous report."""
    lines = (markdown or "").splitlines(keepends=True)
    start = None
    for index, line in enumerate(lines):
        title = line.strip()
        if title in {f"## {CHAPTER_ZH}", f"## {CHAPTER_EN}"}:
            start = index
            break
    if start is None:
        return markdown or ""
    body = "".join(lines[:start]).rstrip()
    return (body + "\n") if body else ""


def _request_hash(payload: dict[str, Any]) -> str:
    encoded = json.dumps(payload, ensure_ascii=False, sort_keys=True, separators=(",", ":"))
    return hashlib.sha256(encoded.encode("utf-8")).hexdigest()


def _pick(args: Any, packet: dict[str, Any], name: str) -> Any:
    value = getattr(args, name, None)
    if value not in (None, ""):
        return value
    birth = packet.get("birth_info") if isinstance(packet.get("birth_info"), dict) else {}
    if birth.get(name) not in (None, ""):
        return birth.get(name)
    profile = packet.get("calculation_profile") if isinstance(packet.get("calculation_profile"), dict) else {}
    if profile.get(name) not in (None, ""):
        return profile.get(name)
    return None


def _payload_from(args: Any, packet: dict[str, Any]) -> dict[str, Any] | None:
    try:
        year = int(_pick(args, packet, "year"))
        target_year = int(_pick(args, packet, "target_year") or year)
        birth = {
            "name": "request",
            "year": year,
            "month": int(_pick(args, packet, "month")),
            "day": int(_pick(args, packet, "day")),
            "hour": int(_pick(args, packet, "hour") or 0),
            "minute": int(_pick(args, packet, "minute") or 0),
            "second": int(_pick(args, packet, "second") or 0),
            "lat": float(_pick(args, packet, "lat")),
            "lon": float(_pick(args, packet, "lon")),
            "tz": float(_pick(args, packet, "tz")),
        }
    except (TypeError, ValueError):
        return None
    ayanamsa = _pick(args, packet, "ayanamsa") or "lahiri"
    node_mode = _pick(args, packet, "node_mode") or "mean"
    return {
        "birth": birth,
        "ayanamsa": str(ayanamsa),
        "node_mode": str(node_mode),
        "target_year": target_year,
    }


def _invoke_worker(payload: dict[str, Any], timeout: int) -> dict[str, Any]:
    if _RUNNER is not None:
        return _RUNNER(payload, timeout)
    completed = subprocess.run(
        [sys.executable, "-m", WORKER_MODULE],
        input=json.dumps(payload, ensure_ascii=False),
        capture_output=True,
        text=True,
        timeout=timeout,
        check=False,
        cwd=str(REPO_ROOT),
    )
    if completed.returncode != 0:
        log.warning(
            "annual method reference worker exit %s: %s",
            completed.returncode,
            (completed.stderr or "")[-500:],
        )
        return {"ok": False, "reason": f"exit_{completed.returncode}"}
    text = (completed.stdout or "").strip()
    try:
        parsed = json.loads(text)
    except json.JSONDecodeError:
        parsed = None
        for line in reversed(text.splitlines()):
            if line.strip().startswith("{"):
                try:
                    parsed = json.loads(line)
                    break
                except json.JSONDecodeError:
                    continue
        if parsed is None:
            log.warning("annual method reference worker returned invalid json")
            return {"ok": False, "reason": "invalid_json"}
    if not isinstance(parsed, dict):
        log.warning("annual method reference worker returned %s", type(parsed).__name__)
        return {"ok": False, "reason": "invalid_payload"}
    if completed.stderr and not parsed.get("ok"):
        log.warning("annual method reference worker stderr: %s", completed.stderr[-500:])
    return parsed


def _year_shell(year: int, *, ayanamsa: str | None = None, node_header: str = "default_true") -> dict[str, Any]:
    return {
        "requested_year": year,
        "ayanamsa": ayanamsa,
        "ayanamsa_ok": False,
        "node_header": node_header,
        "patyayini_anchor": None,
        "tajika_anchor": None,
        "tables": {name: {"ok": False, "rows": []} for name in TABLES},
    }


def _unavailable(target_year: int) -> dict[str, Any]:
    return {"years": [_year_shell(target_year + offset) for offset in range(3)]}


def _lord_label(value: Any) -> str:
    if isinstance(value, str) and value.strip().upper() == "L":
        return "Lagna"
    try:
        return str(DASHA_LORDS.get(int(value), value))
    except (TypeError, ValueError):
        return str(value)


def _display_clock(start: Any) -> str | None:
    if not isinstance(start, (list, tuple)) or len(start) != 4:
        return None
    try:
        year, month, day = (int(part) for part in start[:3])
        hour_decimal = float(start[3])
        stamp = datetime(year, month, day) + timedelta(hours=hour_decimal)
        return stamp.strftime("%Y-%m-%d %H:%M:%S")
    except (TypeError, ValueError, OverflowError):
        return None


def _duration_text(value: Any) -> str:
    try:
        number = float(value)
    except (TypeError, ValueError):
        return ""
    text = f"{number:.4f}".rstrip("0").rstrip(".")
    return text


def _period_rows(periods: Any, *, with_end: bool) -> list[dict[str, Any]]:
    rows: list[dict[str, Any]] = []
    if not isinstance(periods, list):
        return rows
    for order, period in enumerate(periods, start=1):
        if not isinstance(period, (list, tuple)) or len(period) < 3:
            continue
        lords, start, duration = period[0], period[1], period[2]
        if not isinstance(lords, (list, tuple)) or len(lords) < 2:
            continue
        display = _display_clock(start)
        if display is None:
            continue
        rows.append({
            "order": order,
            "main_lord": _lord_label(lords[0]),
            "sub_lord": _lord_label(lords[1]),
            "start": display,
            "end": None,
            "duration": _duration_text(duration),
        })
    if with_end:
        for index, row in enumerate(rows):
            following = rows[index + 1] if index + 1 < len(rows) else None
            row["end"] = following["start"] if following else None
    return rows


def _anchor_text(anchor: Any) -> str | None:
    if not isinstance(anchor, dict):
        return None
    date = anchor.get("date")
    time = anchor.get("time")
    if not isinstance(date, list) or len(date) != 3 or not isinstance(time, str):
        return None
    try:
        return f"{int(date[0]):04d}-{int(date[1]):02d}-{int(date[2]):02d} {time}"
    except (TypeError, ValueError):
        return None


def _planet_name(value: Any) -> str:
    return _lord_label(value)


def _relationship_groups(raw: Any) -> list[list[str]]:
    if raw in (None, []):
        return []
    if isinstance(raw, (list, tuple)):
        if raw and all(not isinstance(item, (list, tuple, dict)) for item in raw):
            return [[_planet_name(item) for item in raw]]
        groups = []
        for item in raw:
            if isinstance(item, (list, tuple)):
                groups.append([_planet_name(part) for part in item])
            else:
                groups.append([_planet_name(item)])
        return groups
    return [[_planet_name(raw)]]


def _tajika_rows(raw: Any) -> list[dict[str, Any]]:
    yogas = raw.get("yogas") if isinstance(raw, dict) and isinstance(raw.get("yogas"), dict) else {}
    failures = raw.get("failures") if isinstance(raw, dict) and isinstance(raw.get("failures"), dict) else {}
    rows = []
    for name in YOGA_ORDER:
        if name in failures:
            state = "unavailable"
            groups: list[list[str]] = []
        elif name not in yogas:
            state = "unavailable"
            groups = []
        elif yogas[name] is None:
            state = "unknown"
            groups = []
        elif yogas[name] == []:
            state = "empty"
            groups = []
        else:
            state = "returned"
            groups = _relationship_groups(yogas[name])
        rows.append({
            "yoga": YOGA_NAMES.get(name, name.title()),
            "state": state,
            "relationships": groups,
        })
    return rows


def _normalize_year(raw: dict[str, Any], *, ayanamsa: str | None, ayanamsa_ok: bool, node_header: str) -> dict[str, Any]:
    year = int(raw.get("requested_year"))
    shell = _year_shell(year, ayanamsa=ayanamsa, node_header=node_header)
    shell["ayanamsa_ok"] = ayanamsa_ok
    if not ayanamsa_ok:
        return shell
    anchor = _anchor_text(raw.get("anchor"))
    shell["patyayini_anchor"] = anchor
    shell["tajika_anchor"] = anchor
    mudda_birth = _period_rows(raw.get("mudda_birth"), with_end=False)
    mudda_annual = _period_rows(raw.get("mudda_annual"), with_end=False)
    patyayini = _period_rows(raw.get("patyayini"), with_end=True)
    tajika = _tajika_rows(raw.get("tajika")) if isinstance(raw.get("tajika"), dict) else []
    shell["tables"] = {
        "mudda_birth": {"ok": bool(mudda_birth) and "mudda_birth_error" not in raw, "rows": mudda_birth},
        "mudda_annual": {"ok": bool(mudda_annual) and "annual_return_error" not in raw and "annual_chart_error" not in raw, "rows": mudda_annual},
        "patyayini": {"ok": bool(patyayini) and "patyayini_error" not in raw, "rows": patyayini},
        "tajika": {"ok": bool(tajika) and "annual_chart_error" not in raw, "rows": tajika},
    }
    return shell


def normalize_worker_payload(payload: dict[str, Any], target_year: int) -> dict[str, Any]:
    """Turn the child JSON into reader rows. Failure text stays out of the packet."""
    if not isinstance(payload, dict) or not payload.get("ok") or not payload.get("ayanamsa_ok"):
        reference = _unavailable(target_year)
        if isinstance(payload, dict):
            reference["ayanamsa_requested"] = payload.get("ayanamsa_requested")
            reference["node_header"] = payload.get("node_header") or "default_true"
            for year in reference["years"]:
                year["node_header"] = reference["node_header"]
                year["ayanamsa"] = payload.get("ayanamsa_readback") or payload.get("ayanamsa_requested")
        return reference
    years = []
    for raw in payload.get("years") or []:
        if isinstance(raw, dict):
            years.append(_normalize_year(
                raw,
                ayanamsa=payload.get("ayanamsa_readback"),
                ayanamsa_ok=True,
                node_header=str(payload.get("node_header") or "default_true"),
            ))
    if len(years) != 3:
        return _unavailable(target_year)
    return {"years": years}


def build_pyjhora_annual_method_reference(args: Any, packet: dict[str, Any], *, timeout: int = DEFAULT_TIMEOUT_SECONDS) -> dict[str, Any]:
    payload = _payload_from(args, packet if isinstance(packet, dict) else {})
    if payload is None:
        return _unavailable(0)
    key = _request_hash(payload)
    if _RUNNER is None and key in _CACHE:
        return _CACHE[key]
    try:
        raw = _invoke_worker(payload, timeout)
    except subprocess.TimeoutExpired:
        log.warning("annual method reference worker timed out after %s seconds", timeout)
        raw = {"ok": False, "reason": "timeout"}
    except Exception as exc:
        log.warning("annual method reference worker could not start: %s", exc.__class__.__name__)
        raw = {"ok": False, "reason": exc.__class__.__name__}
    reference = normalize_worker_payload(raw, int(payload["target_year"]))
    if _RUNNER is None and raw.get("ok"):
        _CACHE[key] = reference
    return reference


def attach_pyjhora_annual_method_reference(packet: dict[str, Any], args: Any) -> dict[str, Any]:
    """Store the chapter under the new packet key. Existing keys stay untouched."""
    if not isinstance(packet, dict):
        return packet
    worksheets = packet.get("worksheets")
    timing = worksheets.get("timing_and_predictive_systems") if isinstance(worksheets, dict) else None
    if not isinstance(timing, dict):
        return packet
    timing["pyjhora_annual_method_reference"] = build_pyjhora_annual_method_reference(args, packet)
    return packet


def _copy(packet: dict[str, Any]) -> dict[str, Any]:
    worksheets = packet.get("worksheets") if isinstance(packet.get("worksheets"), dict) else {}
    timing = worksheets.get("timing_and_predictive_systems") if isinstance(worksheets.get("timing_and_predictive_systems"), dict) else {}
    reference = timing.get("pyjhora_annual_method_reference")
    return reference if isinstance(reference, dict) else {}


def _planet(name: str, language: str) -> str:
    if language == "zh":
        return ZH_PLANETS.get(name, name)
    return name


def _relationships(groups: list[list[str]], language: str) -> str:
    rendered = []
    for group in groups:
        rendered.append("–".join(_planet(name, language) for name in group))
    return "; ".join(item for item in rendered if item)


def _state_label(state: str, language: str) -> str:
    labels = {
        "returned": ("Combination returned; conditions still need checking", "已返回组合，需核对条件"),
        "empty": ("No combination detected", "检测为空"),
        "unknown": ("Unknown", "未知"),
        "unavailable": ("Result unavailable", "未获得结果"),
    }
    pair = labels.get(state, labels["unavailable"])
    return pair[1] if language == "zh" else pair[0]


def _table(headers: list[str], rows: list[list[str]]) -> str:
    head = "| " + " | ".join(headers) + " |"
    rule = "| " + " | ".join("---" for _ in headers) + " |"
    body = ["| " + " | ".join(row) + " |" for row in rows]
    return "\n".join([head, rule, *body])


def _one_year(year: dict[str, Any], language: str) -> str:
    requested = year.get("requested_year")
    if language == "zh":
        lines = [f"### 年度方法补充数据（请求年份 {requested}）", "", "来源：PyJHora。本节数字由 PyJHora 计算，与前面的原生周期分开读。"]
        node = {"mean": "平交点", "true": "真交点"}.get(str(year.get("node_header")), "PyJHora 默认（真交点）")
        ayanamsa = year.get("ayanamsa") or "未能按请求设定"
        lines.append(f"岁差：{ayanamsa}。交点：{node}。")
        lines.append(f"Patyayini 回归锚点：{year.get('patyayini_anchor') or '未取得'}")
        lines.append(f"Tajika 回归锚点：{year.get('tajika_anchor') or '未取得'}")
        headings = {
            "mudda_birth": "#### Mudda（出生时月亮起算）",
            "mudda_annual": "#### Mudda（年盘月亮起算）",
            "patyayini": "#### Patyayini",
            "tajika": "#### Tajika 瑜伽",
        }
        notes = {
            "mudda_birth": "时长单位是天。这一表没有结束时间。",
            "mudda_annual": "时长单位是天。这一表没有结束时间。",
            "patyayini": "时长单位是年。结束时间是下一段的开始。最后一段没有结束时间。",
            "tajika": "空的组合写检测为空。没有返回结果时写未获得结果。",
        }
        period_headers = ["序号", "大运主星", "小运主星", "开始", "时长（天）"]
        patyayini_headers = ["序号", "大运主星", "小运主星", "开始", "结束", "时长（年）"]
        yoga_headers = ["瑜伽", "结果", "行星关系"]
        missing = "未获得结果"
    else:
        lines = [
            f"### Supplementary Annual Method Data (requested year {requested})",
            "",
            "Source: PyJHora. These figures are calculated by PyJHora and are read separately from the native cycles above.",
        ]
        node = {"mean": "mean", "true": "true"}.get(str(year.get("node_header")), "PyJHora default (true node)")
        ayanamsa = year.get("ayanamsa") or "could not be set to the requested mode"
        lines.append(f"Ayanamsa: {ayanamsa}. Nodes: {node}.")
        lines.append(f"Patyayini return anchor: {year.get('patyayini_anchor') or 'Not obtained'}")
        lines.append(f"Tajika return anchor: {year.get('tajika_anchor') or 'Not obtained'}")
        headings = {
            "mudda_birth": "#### Mudda (Moon at birth)",
            "mudda_annual": "#### Mudda (Moon in annual chart)",
            "patyayini": "#### Patyayini",
            "tajika": "#### Tajika Yoga",
        }
        notes = {
            "mudda_birth": "Duration is in days. This table has no end time.",
            "mudda_annual": "Duration is in days. This table has no end time.",
            "patyayini": "Duration is in years. The end is the next row's start. The last row has no end time.",
            "tajika": "An empty detection is written as no combination. A failed call is written as result unavailable.",
        }
        period_headers = ["No.", "MD lord", "AD lord", "Start", "Duration (days)"]
        patyayini_headers = ["No.", "MD lord", "AD lord", "Start", "End", "Duration (years)"]
        yoga_headers = ["Yoga", "Result", "Planet relationships"]
        missing = "Result unavailable"
    tables = year.get("tables") if isinstance(year.get("tables"), dict) else {}
    for key in TABLES:
        lines.extend(["", headings[key], "", notes[key], ""])
        table = tables.get(key) if isinstance(tables.get(key), dict) else {}
        rows = table.get("rows") if table.get("ok") and isinstance(table.get("rows"), list) else []
        if not rows:
            lines.append(missing)
            continue
        if key == "tajika":
            rendered = [[
                str(row.get("yoga") or ""),
                _state_label(str(row.get("state") or ""), language),
                _relationships(row.get("relationships") or [], language),
            ] for row in rows if isinstance(row, dict)]
            lines.append(_table(yoga_headers, rendered))
            continue
        headers = patyayini_headers if key == "patyayini" else period_headers
        rendered = []
        for row in rows:
            if not isinstance(row, dict):
                continue
            end = row.get("end") or ("—" if language == "zh" else "-")
            base = [
                str(row.get("order") or ""),
                _planet(str(row.get("main_lord") or ""), language),
                _planet(str(row.get("sub_lord") or ""), language),
                str(row.get("start") or ""),
            ]
            if key == "patyayini":
                base.append(str(end))
            base.append(str(row.get("duration") or ""))
            rendered.append(base)
        lines.append(_table(headers, rendered))
    return "\n".join(lines)


def render_annual_method_reference(packet: dict[str, Any], language: str) -> str:
    reference = _copy(packet if isinstance(packet, dict) else {})
    years = [year for year in reference.get("years") or [] if isinstance(year, dict)]
    if not years:
        return ""
    if language == "zh":
        intro = "## 独立年度方法参考\n\n本节与前面的原生周期分开使用。数字不替换、不合并前面的年运。"
    else:
        intro = "## Independent Annual Method Reference\n\nUse this section separately from the native cycles above. These figures do not replace the annual cycles earlier in the report."
    body = "\n\n".join(_one_year(year, "en" if language == "en" else "zh") for year in years)
    return intro + "\n\n" + body + "\n"