Jyotisha/scripts/research/research_timing_pyjhora_compare.py

"""Offline PyJHora comparison for the research timing appendix.

Pytest does not import this file. The runtime report does not import it either.
PyJHora is imported only inside ``pyjhora_fingerprints``, and only when this
file is run as a worker process. A missing public function is recorded as
「无此方法」 and is not a mismatch.
"""

from __future__ import annotations

import importlib.util
import json
import subprocess
import sys
import tempfile
import warnings
from datetime import datetime, timedelta, timezone
from pathlib import Path

ROOT = Path(__file__).resolve().parents[2]
OUT = ROOT / "docs" / "tasks" / "research-timing-pyjhora-20261023.json"
CHARTS = tuple(range(10))
YEAR_BASIS = 365.25
RETURN_MINUTE = 1.0
CATEGORIES = (
    ("起点", "start"),
    ("方向", "direction"),
    ("年数", "year count"),
    ("子运划分", "sub-period division"),
    ("回归时刻", "return moment"),
)
HEADINGS = {
    "tithi-natal-forward": ("Natal Tithi Ashtottari forward", "本命 Tithi Ashtottari 顺行", "exact"),
    "tithi-natal-reverse": ("Natal Tithi Ashtottari reverse", "本命 Tithi Ashtottari 逆行", "exact"),
    "yoga-natal": ("Natal Yoga Vimshottari", "本命 Yoga Vimshottari", "exact"),
    "buddhi-gati": ("Buddhi Gati", "Buddhi Gati", "exact"),
    "kaala": ("Kaala", "Kaala", "exact"),
    "sudasa-birth": ("Sudasa birth cycle", "Sudasa 出生周期", "exact"),
    "varnada": ("Varnada", "Varnada", "prefix"),
    "karana-julene": ("Julene lords", "Julene 主星", "prefix"),
    "karana-pvr": ("PVR quadrant lords", "PVR 象限主星", "prefix"),
    "karana-pvr-tithi-year": ("PVR 360 tithi year", "PVR 360 太阴月年", "prefix"),
    "sthira": ("Sthira", "Sthira", "exact"),
    "drig": ("Drig", "Drig", "exact"),
    "shoola": ("Shoola", "Shoola", "exact"),
    "navamsha": ("Navamsha", "Navamsha", "exact"),
    "kalachakra": ("Kalachakra", "Kalachakra", "exact"),
}


def _ensure_root() -> None:
    """Match the test path: ``scripts/`` first so engine modules import, repo root second."""
    for entry in (str(ROOT / "scripts"), str(ROOT)):
        if entry in sys.path:
            sys.path.remove(entry)
        sys.path.insert(0, entry)


def _span_years(start, end) -> float | None:
    if not start or not end:
        return None
    try:
        left = datetime.fromisoformat(str(start))
        right = datetime.fromisoformat(str(end))
    except ValueError:
        return None
    days = (right - left).total_seconds() / 86400.0
    if days <= 0:
        return None
    return round(days / YEAR_BASIS, 2)


def _label(row: dict, kind: str) -> str:
    if kind == "sign":
        value = row.get("sign") or row.get("rashi") or row.get("lord")
    else:
        value = row.get("lord") or row.get("planet") or row.get("sign")
    return "" if value is None else str(value)


def fingerprint_periods(periods: list, *, kind: str) -> dict:
    """Lords or signs, nominal years, first period's children, and its first grandchildren."""
    maha: list[str] = []
    years: list[float | None] = []
    children: list[str] = []
    grand: list[str] = []
    for index, row in enumerate(periods or []):
        if not isinstance(row, dict):
            continue
        maha.append(_label(row, kind))
        if "full_end_days" in row and "full_start_days" in row:
            span = (float(row["full_end_days"]) - float(row["full_start_days"])) / YEAR_BASIS
            years.append(round(span, 2))
        elif isinstance(row.get("years"), (int, float)) and not isinstance(row.get("years"), bool):
            years.append(round(float(row["years"]), 2))
        else:
            years.append(_span_years(row.get("start") or row.get("start_date"), row.get("end") or row.get("end_date")))
        if index != 0:
            continue
        kids = row.get("antardashas") or row.get("antardasha") or []
        for child in kids:
            if not isinstance(child, dict):
                continue
            children.append(_label(child, kind))
            if grand:
                continue
            thirds = child.get("pratyantardashas") or child.get("pratyantar") or child.get("pratyantardasha") or []
            for item in thirds:
                if isinstance(item, dict):
                    grand.append(_label(item, kind))
    if not maha:
        return {"skipped": True}
    return {"maha": maha, "years": years, "children": children, "grand": grand}


def _chart_input(index: int) -> tuple[dict, dict]:
    _ensure_root()
    from tests.report_english_fictional import fictional_case
    from scripts.research_timing.bundle import chart_from_parts

    case = fictional_case(index)
    zone = timezone(timedelta(hours=float(case["tz"])))
    query = datetime(int(case["target_year"]), 1, 1, 12, 0, tzinfo=zone)
    chart = chart_from_parts(
        year=case["year"], month=case["month"], day=case["day"],
        hour=case["hour"], minute=case["minute"], second=case["second"],
        lat=case["lat"], lon=case["lon"], tz=case["tz"],
        ayanamsa=case["ayanamsa"], node_mode=case["node_mode"],
        target_year=case["target_year"], query=query,
    )
    return case, chart


def _tithi_return_days(chart: dict) -> float | None:
    from scripts.research_timing.frame import SiderealFrame, birth_jd
    from scripts.research_timing.search import find_returns

    frame = SiderealFrame(chart["ayanamsa"], chart["node_mode"])
    jd = birth_jd(chart["year"], chart["month"], chart["day"], chart["hour_decimal"], chart["tz"])
    found = find_returns(frame, "tithi_pravesha", jd, jd, jd + 400.0)
    for row in found.get("returns") or []:
        elapsed = float(row["jd_ut"]) - jd
        if elapsed > 1e-3:
            return elapsed
    return None


def collect_native(indexes: tuple[int, ...] = CHARTS) -> dict:
    """Our research periods for the fixed fictional charts. This process does not import jhora."""
    _ensure_root()
    from scripts.annual_native.dasha_children import attach_kalachakra_simplified_children
    from scripts.kalachakra_dasha import calculate_kalachakra_dasha
    from scripts.research_timing.bundle import _arc_remaining, _tithi_number
    from scripts.research_timing.cycles import (
        TITHI_FORWARD,
        TITHI_REVERSE,
        karana_year,
        tithi_ashtottari,
        yoga_vimshottari,
    )
    from scripts.research_timing.families import (
        buddhi_gati,
        hora_sign_from_events,
        kaala_from_events,
        karana_seed,
        sudasa_birth,
        varnada_jss,
    )
    from scripts.research_timing.frame import SiderealFrame, birth_jd
    from scripts.research_timing.rashi_pd import attach_rashi_pratyantar
    from scripts.research_timing.search import hindu_solar_events
    from scripts.rashi_dashas_native import calculate_rashi_dashas

    payload = {}
    for index in indexes:
        try:
            _case, chart = _chart_input(index)
        except OSError:
            print(f"native chart {index} skipped", file=sys.stderr)
            continue
        year_days = float(chart["year_days"])
        frame = SiderealFrame(chart["ayanamsa"], chart["node_mode"])
        jd = birth_jd(chart["year"], chart["month"], chart["day"], chart["hour_decimal"], chart["tz"])
        sun, moon = frame.sun_moon(jd)
        epoch = chart["epoch"]
        systems: dict[str, dict] = {}

        def keep(name: str, producer) -> None:
            try:
                systems[name] = producer()
            except (ValueError, RuntimeError, OverflowError, KeyError, TypeError, ZeroDivisionError, OSError) as exc:
                systems[name] = {"error": type(exc).__name__}

        number = _tithi_number(sun, moon)
        fraction = _arc_remaining(sun, moon)
        keep("tithi-natal-forward", lambda: fingerprint_periods(tithi_ashtottari(
            epoch, number, fraction, 108 * year_days, TITHI_FORWARD, natal=True,
        )["periods"], kind="planet"))
        keep("tithi-natal-reverse", lambda: fingerprint_periods(tithi_ashtottari(
            epoch, number, fraction, 108 * year_days, TITHI_REVERSE, natal=True,
        )["periods"], kind="planet"))
        keep("yoga-natal", lambda: fingerprint_periods(yoga_vimshottari(
            epoch, (sun + moon) % 360.0, 120 * year_days, natal=True, depth=3,
        )["periods"], kind="planet"))
        keep("buddhi-gati", lambda: fingerprint_periods(
            buddhi_gati(int(chart["ascendant"] // 30), chart["longitudes"], epoch, year_days).get("periods") or [],
            kind="planet",
        ))
        try:
            events = hindu_solar_events(epoch, chart["lat"], chart["lon"])
        except (ValueError, RuntimeError, OverflowError, KeyError, TypeError, OSError) as exc:
            events = None
            systems["kaala"] = {"error": type(exc).__name__}
            systems["varnada"] = {"error": type(exc).__name__}
        if events is not None:
            keep("kaala", lambda: fingerprint_periods(
                kaala_from_events(epoch, events, year_days).get("periods") or [], kind="planet",
            ))
            try:
                hora = hora_sign_from_events(epoch, int(chart["ascendant"] // 30), events)
                if hora.get("status") != "calculated":
                    systems["varnada"] = {"skipped": True}
                else:
                    variants = varnada_jss(
                        int(chart["ascendant"] // 30), int(hora["sign_index"]),
                        chart["longitudes"], epoch, year_days,
                    )
                    parts = [fingerprint_periods(item.get("periods") or [], kind="sign") for item in variants]
                    systems["varnada"] = {"variants": parts}
            except (ValueError, RuntimeError, OverflowError, KeyError, TypeError, OSError) as exc:
                systems["varnada"] = {"error": type(exc).__name__}
        for profile, span, key in (
            ("julene", year_days, "karana-julene"),
            ("pvr", year_days, "karana-pvr"),
            ("pvr", 360.0, "karana-pvr-tithi-year"),
        ):
            def karana_fp(profile=profile, span=span):
                seed = karana_seed(sun, moon, chart["ascendant"], "pvr" if profile == "pvr" else "julene")
                if seed.get("status") != "calculated":
                    return {"skipped": True}
                built = karana_year(epoch, seed["lord"], float(seed["remaining_fraction"]), float(span))
                return fingerprint_periods(built.get("periods") or [], kind="planet")
            keep(key, karana_fp)
        birth = datetime(chart["year"], chart["month"], chart["day"]).date()
        try:
            copied = attach_rashi_pratyantar(calculate_rashi_dashas(birth, chart["ascendant"], chart["longitudes"]))
            for name in ("sthira", "drig", "shoola", "navamsha"):
                section = copied.get(name) if isinstance(copied.get(name), dict) else {}
                systems[name] = fingerprint_periods(section.get("periods") or [], kind="sign")
        except (ValueError, RuntimeError, OverflowError, KeyError, TypeError, OSError) as exc:
            for name in ("sthira", "drig", "shoola", "navamsha"):
                systems[name] = {"error": type(exc).__name__}
        try:
            moon_lon = float(chart["longitudes"]["Moon"])
            nak = int(moon_lon // (360.0 / 27.0)) % 27
            pada = int((moon_lon % (360.0 / 27.0)) // (360.0 / 108.0)) + 1
            # This calculator subtracts datetime.now(), which is naive. Keep the
            # clock time and drop the zone so the call can finish.
            birth_dt = epoch.replace(tzinfo=None) if isinstance(epoch, datetime) and epoch.tzinfo is not None else epoch
            calculated = calculate_kalachakra_dasha({
                "moon_nakshatra_index": nak,
                "moon_pada": pada,
                "birth_datetime": birth_dt,
            })
            enriched, _children, _grands = attach_kalachakra_simplified_children(calculated.get("major") or [])
            systems["kalachakra"] = fingerprint_periods(enriched, kind="sign")
        except (ValueError, RuntimeError, OverflowError, KeyError, TypeError, OSError) as exc:
            systems["kalachakra"] = {"error": type(exc).__name__}
        try:
            systems["sudasa-birth"] = fingerprint_periods(
                sudasa_birth(chart["ascendant"], chart["longitudes"], epoch, year_days).get("periods") or [],
                kind="sign",
            )
        except (ValueError, RuntimeError, OverflowError, KeyError, TypeError, OSError) as exc:
            systems["sudasa-birth"] = {"error": type(exc).__name__}
        try:
            return_days = _tithi_return_days(chart)
        except (ValueError, RuntimeError, OverflowError, KeyError, TypeError, OSError):
            return_days = None
        payload[str(index)] = {"systems": systems, "tithi_return_days": return_days}
        print(f"native chart {index}", file=sys.stderr)
    return payload


def _as_name(value, kind: str) -> str:
    planets = ("Sun", "Moon", "Mars", "Mercury", "Jupiter", "Venus", "Saturn", "Rahu", "Ketu")
    signs = (
        "Aries", "Taurus", "Gemini", "Cancer", "Leo", "Virgo",
        "Libra", "Scorpio", "Sagittarius", "Capricorn", "Aquarius", "Pisces",
    )
    if isinstance(value, str):
        return value
    if isinstance(value, bool) or not isinstance(value, (int, float)):
        return str(value)
    index = int(value)
    table = planets if kind == "planet" else signs
    if 0 <= index < len(table):
        return table[index]
    return str(value)


def _start_days(value):
    if isinstance(value, (list, tuple)) and len(value) >= 3:
        year, month, day = int(value[0]), int(value[1]), int(value[2])
        hours = float(value[3]) if len(value) > 3 else 0.0
        try:
            stamp = datetime(year, month, day) + timedelta(hours=hours)
        except ValueError:
            return None
        return stamp.toordinal() + hours / 24.0
    return None


def _unpack_row(row):
    if not isinstance(row, (list, tuple)) or not row or not isinstance(row[0], (list, tuple)):
        return None
    lords = tuple(row[0])
    start = None
    duration = None
    for item in row[1:]:
        if isinstance(item, (list, tuple)) and start is None:
            start = _start_days(item)
        elif isinstance(item, (int, float)) and not isinstance(item, bool):
            duration = float(item)
    return lords, start, duration


def fingerprint_rows(rows: list, *, kind: str) -> dict:
    parsed = []
    for row in rows or []:
        unpacked = _unpack_row(row)
        if unpacked is None:
            continue
        lords, start, duration = unpacked
        if not lords:
            continue
        parsed.append((tuple(_as_name(item, kind) for item in lords), start, duration))
    if not parsed:
        return {"skipped": True}
    groups = []
    for lords, start, duration in parsed:
        if not groups or groups[-1]["lord"] != lords[0]:
            groups.append({"lord": lords[0], "start": start, "durs": [], "children": [], "grands": {}})
        group = groups[-1]
        if duration is not None:
            group["durs"].append(duration)
        if len(lords) < 2:
            continue
        child = lords[1]
        if child not in group["grands"]:
            group["children"].append(child)
            group["grands"][child] = []
        if len(lords) >= 3:
            group["grands"][child].append(lords[2])
    maha = [group["lord"] for group in groups]
    years: list[float | None] = []
    for index, group in enumerate(groups):
        if group["durs"]:
            years.append(round(sum(group["durs"]), 2))
            continue
        nxt = groups[index + 1]["start"] if index + 1 < len(groups) else None
        if group["start"] is not None and nxt is not None:
            years.append(round((nxt - group["start"]) / YEAR_BASIS, 2))
        else:
            years.append(None)
    first = groups[0]
    children = first["children"]
    grand = first["grands"].get(children[0], []) if children else []
    return {"maha": maha, "years": years, "children": children, "grand": grand}


def _rows_of(result):
    if isinstance(result, list):
        return result
    if isinstance(result, tuple):
        for item in result:
            if isinstance(item, list):
                return item
    return []


def pyjhora_fingerprints(indexes: tuple[int, ...] = CHARTS) -> dict:
    """PyJHora periods. Importing jhora here is intentional and stays in this process."""
    warnings.filterwarnings("ignore")
    _ensure_root()
    import jhora
    from jhora import const, utils
    from jhora.panchanga import drik
    from jhora.horoscope.dhasa.graha import buddhi_gathi, kaala, karana_chathuraaseethi_sama, tithi_ashtottari, yoga_vimsottari
    from jhora.horoscope.dhasa.raasi import drig, kalachakra, navamsa, shoola, sthira, sudasa, varnada

    version = getattr(jhora, "__version__", None)
    if not version:
        try:
            from importlib.metadata import version as pkg_version
            version = pkg_version("PyJHora")
        except (ImportError, OSError, ValueError):
            version = "installed"
    charts = {}
    for index in indexes:
        try:
            case, _chart = _chart_input(index)
            mode = {"lahiri": "LAHIRI", "raman": "RAMAN", "kp": "KP"}.get(str(case["ayanamsa"]).lower(), "LAHIRI")
            drik.set_planet_list(set_rahu_ketu_as_true_nodes=str(case["node_mode"]).lower() == "true", include_western_planets=False)
            drik.set_ayanamsa_mode(mode)
            place = drik.Place("fictional", case["lat"], case["lon"], case["tz"], elevation=0.0)
            dob = drik.Date(int(case["year"]), int(case["month"]), int(case["day"]))
            tob = (int(case["hour"]), int(case["minute"]), int(case["second"]))
            jd = utils.julian_day_number(dob, tob)
        except OSError:
            print(f"pyjhora chart {index} skipped", file=sys.stderr)
            continue
        systems: dict[str, dict] = {}

        def run(name: str, kind: str, producer) -> None:
            try:
                produced = producer()
            except AttributeError:
                # The public names are imported above. An AttributeError from the
                # call itself is a broken invocation, not 「无此方法」.
                systems[name] = {"error": "AttributeError"}
                return
            except (TypeError, ValueError, RuntimeError, OverflowError, KeyError, ZeroDivisionError, OSError) as exc:
                systems[name] = {"error": type(exc).__name__}
                return
            try:
                systems[name] = fingerprint_rows(_rows_of(produced), kind=kind)
            except (TypeError, ValueError, RuntimeError, OverflowError, KeyError, ZeroDivisionError, OSError, AttributeError) as exc:
                systems[name] = {"error": type(exc).__name__}

        run("tithi-natal-forward", "planet", lambda: tithi_ashtottari.get_dhasa_bhukthi(jd, place, dhasa_level_index=2))
        systems["tithi-natal-reverse"] = {"missing": True}
        run("yoga-natal", "planet", lambda: yoga_vimsottari.get_dhasa_bhukthi(jd, place, dhasa_level_index=3))
        run("buddhi-gati", "planet", lambda: buddhi_gathi.get_dhasa_bhukthi(dob, tob, place, dhasa_level_index=2))
        run("kaala", "planet", lambda: kaala.get_dhasa_antardhasa(dob, tob, place, dhasa_level_index=2))
        run("sudasa-birth", "sign", lambda: sudasa.get_dhasa_bhukthi(dob, tob, place, dhasa_level_index=2))
        run("varnada", "sign", lambda: varnada.get_dhasa_antardhasa(dob, tob, place, dhasa_level_index=2))
        run("karana-julene", "planet", lambda: karana_chathuraaseethi_sama.get_dhasa_bhukthi(dob, tob, place, dhasa_level_index=2))
        systems["karana-pvr"] = systems["karana-julene"]
        systems["karana-pvr-tithi-year"] = systems["karana-julene"]
        run("sthira", "sign", lambda: sthira.get_dhasa_antardhasa(dob, tob, place, dhasa_level_index=3))
        run("drig", "sign", lambda: drig.get_dhasa_antardhasa(jd, place, dhasa_level_index=3))
        run("shoola", "sign", lambda: shoola.get_dhasa_bhukthi(dob, tob, place, dhasa_level_index=3))
        run("navamsha", "sign", lambda: navamsa.get_dhasa_antardhasa(dob, tob, place, dhasa_level_index=3))
        def kala_rows():
            jd_utc = jd - float(case["tz"]) / 24.0
            moon = drik.sidereal_longitude(jd_utc, const._MOON)
            return kalachakra.kalachakra_dhasa(moon, jd, dhasa_level_index=3, place=place)
        run("kalachakra", "sign", kala_rows)
        try:
            systems["tithi_return_days"] = _py_tithi_return_days(jd, float(case["tz"]))
        except (TypeError, ValueError, RuntimeError, OverflowError, KeyError, ZeroDivisionError, OSError):
            systems["tithi_return_days"] = None
        charts[str(index)] = systems
        print(f"pyjhora chart {index}", file=sys.stderr)
    return {"version": str(version), "charts": charts}


def _py_tithi_return_days(jd_local: float, tz: float) -> float | None:
    from jhora import const
    from jhora.panchanga import drik

    jd_utc = jd_local - tz / 24.0
    phase0 = drik.lunar_phase(jd_utc)
    sign0 = int(drik.sidereal_longitude(jd_utc, const._SUN) // 30)

    def residual(offset: float) -> float:
        phase = drik.lunar_phase(jd_utc + offset)
        return (phase - phase0 + 180.0) % 360.0 - 180.0

    left = 0.05
    left_value = residual(left)
    while left < 400.0:
        right = min(left + 0.25, 400.0)
        right_value = residual(right)
        if left_value * right_value <= 0 and abs(right_value - left_value) < 180.0:
            lo, hi, lo_value = left, right, left_value
            for _ in range(40):
                mid = (lo + hi) / 2.0
                mid_value = residual(mid)
                if lo_value * mid_value <= 0:
                    hi = mid
                else:
                    lo, lo_value = mid, mid_value
            root = (lo + hi) / 2.0
            if abs(residual(root)) < 1e-3 and int(drik.sidereal_longitude(jd_utc + root, const._SUN) // 30) == sign0:
                return root
        left, left_value = right, right_value
    return None


def _base_cycle(seq: list[str]) -> list[str]:
    """Shortest prefix that tiles ``seq``, so one extra lord of the same cycle still matches."""
    for size in range(1, len(seq) + 1):
        cycle = seq[:size]
        if all(seq[index] == cycle[index % size] for index in range(len(seq))):
            return cycle
    return seq


def _same_lords(left: list[str], right: list[str]) -> bool:
    """Same order, including when one side continues a partial repeat of the other's cycle."""
    if left == right:
        return True
    if not left or not right:
        return False
    short, long = (left, right) if len(left) <= len(right) else (right, left)
    cycle = _base_cycle(short)
    if not cycle:
        return False
    return all(name == cycle[index % len(cycle)] for index, name in enumerate(long))


def _is_reverse(left: list[str], right: list[str]) -> bool:
    if not left or not right:
        return False
    return _same_lords(list(reversed(left)), right)


def _order_example(left: list[str], right: list[str]) -> str:
    """Name the first period that differs. Never say X vs X when the opening lords match."""
    count = min(len(left), len(right))
    for index in range(count):
        if left[index] != right[index]:
            return f"period {index + 1} native {left[index]} vs PyJHora {right[index]}"
    return f"period count native {len(left)} vs PyJHora {len(right)}"


def _years_match(left: list, right: list, lords_left: list[str], lords_right: list[str]) -> bool:
    count = min(len(left), len(right), len(lords_left), len(lords_right))
    pairs = []
    for index in range(count):
        a, b = left[index], right[index]
        if isinstance(a, (int, float)) and isinstance(b, (int, float)):
            pairs.append((float(a), float(b)))
    if not pairs:
        return True
    return all(abs(a - b) <= 0.35 for a, b in pairs)


def _slice_match(native: list[str], other: list[str]) -> bool:
    if native == other:
        return True
    if not native or not other:
        return False
    if len(native) > len(other):
        native, other = other, native
    return any(other[index:index + len(native)] == native for index in range(len(other) - len(native) + 1))


def _usable(fp: dict | None) -> bool:
    return isinstance(fp, dict) and not fp.get("skipped") and not fp.get("error") and not fp.get("missing") and bool(fp.get("maha"))


def classify_pair(native: dict, other: dict, *, chart: int, return_days: tuple[float | None, float | None] | None = None) -> dict:
    if isinstance(other, dict) and other.get("missing"):
        return {"status": "无此方法", "chart": chart}
    if (isinstance(native, dict) and native.get("error")) or (isinstance(other, dict) and other.get("error")):
        return {"status": "error", "chart": chart}
    if not isinstance(native, dict) or not isinstance(other, dict):
        return {"status": "skipped", "chart": chart}
    native_rows = native.get("variants") if isinstance(native.get("variants"), list) else [native]
    other_rows = other.get("variants") if isinstance(other.get("variants"), list) else [other]
    usable_native = [row for row in native_rows if _usable(row)]
    usable_other = [row for row in other_rows if _usable(row)]
    if not usable_native or not usable_other:
        if not usable_native and not usable_other:
            return {"status": "skipped", "chart": chart}
        return _mismatch("起点", chart, "one side listed no periods")
    verdicts = []
    for native_fp in usable_native:
        for other_fp in usable_other:
            verdicts.append(_classify_rows(native_fp, other_fp, chart, return_days if len(verdicts) == 0 else None))
    mismatches = [item for item in verdicts if item["status"] == "不一致"]
    if not mismatches and any(item["status"] == "一致" for item in verdicts):
        agreed = next(item for item in verdicts if item["status"] == "一致")
        return agreed
    if mismatches:
        return mismatches[0]
    return {"status": "skipped", "chart": chart}


def _classify_rows(native: dict, other: dict, chart: int, return_days) -> dict:
    left = [str(item) for item in native.get("maha") or []]
    right = [str(item) for item in other.get("maha") or []]
    if not _same_lords(left, right):
        if _is_reverse(left, right):
            return _mismatch("方向", chart, f"period order runs the other way, {_order_example(left, right)}")
        return _mismatch("起点", chart, _order_example(left, right))
    if not _years_match(native.get("years") or [], other.get("years") or [], left, right):
        return _mismatch("年数", chart, f"year count differs at the first compared period of {left[0]}")
    if not _slice_match(list(native.get("children") or []), list(other.get("children") or [])):
        return _mismatch("子运划分", chart, f"child periods under {left[0]} do not line up")
    if not _slice_match(list(native.get("grand") or []), list(other.get("grand") or [])):
        return _mismatch("子运划分", chart, f"third level under {left[0]} does not line up")
    note = _return_note(return_days)
    if note and note["status"] == "不一致":
        return note
    agreed = {"status": "一致", "chart": chart}
    if note and note.get("within_tolerance") is not None:
        agreed["within_tolerance"] = note["within_tolerance"]
    return agreed


def _mismatch(category: str, chart: int, example: str) -> dict:
    english = dict(CATEGORIES)[category]
    return {
        "status": "不一致",
        "category": category,
        "category_en": english,
        "example": f"chart {chart} {example}",
        "chart": chart,
    }


def _return_note(pair) -> dict | None:
    if pair is None:
        return None
    native, other = pair
    if native is None and other is None:
        return None
    if native is None or other is None:
        return {
            "status": "不一致",
            "category": "回归时刻",
            "category_en": "return moment",
            "example": "tithi return is present on only one side",
        }
    minutes = abs(float(native) - float(other)) * 1440.0
    if minutes <= RETURN_MINUTE:
        return {"status": "一致", "within_tolerance": round(minutes, 3)}
    return {
        "status": "不一致",
        "category": "回归时刻",
        "category_en": "return moment",
        "example": f"tithi return differs by {minutes:.1f} minutes",
    }


def _merge_status(rows: list[dict]) -> dict:
    errors = [row for row in rows if row.get("status") == "error"]
    comparable = [row for row in rows if row.get("status") not in {"skipped", "error"}]
    if not comparable:
        return {"status": "error"} if errors else {"status": "无此方法"}
    if all(row.get("status") == "无此方法" for row in comparable):
        return {"status": "无此方法"}
    mismatches = [row for row in comparable if row.get("status") == "不一致"]
    if mismatches:
        chosen = sorted(mismatches, key=lambda row: (CATEGORIES.index((row["category"], row["category_en"])), row.get("chart", 0)))[0]
        return {
            "status": "不一致",
            "category": chosen["category"],
            "category_en": chosen["category_en"],
            "example": chosen.get("example") or "",
        }
    if any(row.get("status") == "一致" for row in comparable):
        return {"status": "一致"}
    return {"status": "无此方法"}


def _system_record(key: str, merged: dict) -> dict:
    en, zh, match = HEADINGS[key]
    record = {
        "status": merged["status"],
        "heading_en": en,
        "heading_zh": zh,
        "match": match,
    }
    if merged["status"] == "不一致":
        record["category"] = merged["category"]
        record["category_en"] = merged["category_en"]
        record["example"] = merged.get("example") or ""
    return record


def build_report(native: dict, pyjhora: dict) -> dict:
    indexes = sorted({int(key) for key in native} & {int(key) for key in pyjhora.get("charts") or {}})
    within = []
    systems = {}
    for key in HEADINGS:
        rows = []
        for index in indexes:
            ours = (native[str(index)].get("systems") or {}).get(key) or {"skipped": True}
            theirs = (pyjhora["charts"][str(index)] or {}).get(key) or {"skipped": True}
            pair = None
            if key == "tithi-natal-forward":
                pair = (
                    native[str(index)].get("tithi_return_days"),
                    (pyjhora["charts"][str(index)] or {}).get("tithi_return_days"),
                )
            verdict = classify_pair(ours, theirs, chart=index, return_days=pair)
            if verdict.get("status") == "不一致" and verdict.get("category") == "回归时刻":
                verdict["example"] = f"chart {index} {verdict.get('example') or ''}".replace(f"chart {index} chart {index} ", f"chart {index} ")
            if verdict.get("within_tolerance") is not None:
                within.append({"system": key, "chart": index, "minutes": verdict["within_tolerance"]})
            rows.append(verdict)
        systems[key] = _system_record(key, _merge_status(rows))
    return {
        "status": "compared",
        "charts": len(indexes),
        "chart_indexes": indexes,
        "pyjhora_version": pyjhora.get("version") or "",
        "return_rule": (
            "A return moment within 1 minute is listed in within_tolerance and is not a mismatch. "
            "The two sides use different ayanamsa implementations and ephemeris files."
        ),
        "omitted_from_labels": [
            "Sudasa confirmed start has no written full major start on our side, so it is not compared.",
            "Niryaana Shoola child periods stay omitted by the 2026-10-23 decision.",
            "Annual Tithi, Yoga, and Nakshatra headings are outside this pass.",
        ],
        "within_tolerance": within,
        "systems": systems,
    }


def _write(payload: dict) -> None:
    OUT.parent.mkdir(parents=True, exist_ok=True)
    OUT.write_text(json.dumps(payload, ensure_ascii=False, indent=2) + "\n", encoding="utf-8")


def main() -> int:
    _ensure_root()
    if "--pyjhora-worker" in sys.argv:
        slot = sys.argv.index("--pyjhora-worker")
        target = Path(sys.argv[slot + 1])
        indexes = tuple(int(item) for item in sys.argv[slot + 2:]) or CHARTS
        target.write_text(json.dumps(pyjhora_fingerprints(indexes), ensure_ascii=False), encoding="utf-8")
        print(target)
        return 0
    if importlib.util.find_spec("jhora") is None:
        _write({
            "status": "not_run",
            "charts": 0,
            "reason": "PyJHora is not installed in this interpreter",
            "error_type": "ImportError",
            "systems": {},
        })
        print("not_run")
        return 0
    native = collect_native(CHARTS)
    worker_out = Path(tempfile.gettempdir()) / "jyotisha-92-pyjhora.json"
    command = [
        sys.executable, "-m", "scripts.research.research_timing_pyjhora_compare",
        "--pyjhora-worker", str(worker_out), *[str(index) for index in CHARTS],
    ]
    completed = subprocess.run(command, cwd=str(ROOT), check=False)
    if completed.returncode != 0 or not worker_out.exists():
        print("pyjhora worker failed", file=sys.stderr)
        return 1
    pyjhora = json.loads(worker_out.read_text(encoding="utf-8"))
    worker_out.unlink(missing_ok=True)
    report = build_report(native, pyjhora)
    if report["charts"] < 5:
        print(f"only {report['charts']} charts", file=sys.stderr)
        return 1
    failed = [key for key, item in report["systems"].items() if item.get("status") == "error"]
    if failed:
        print("call failed on every chart: " + ",".join(failed), file=sys.stderr)
        return 1
    _write(report)
    counts = {}
    for item in report["systems"].values():
        counts[item["status"]] = counts.get(item["status"], 0) + 1
    print(json.dumps({"charts": report["charts"], "counts": counts}, ensure_ascii=False))
    return 0


if __name__ == "__main__":
    raise SystemExit(main())