"""Issue #93: PyJHora annual method reference stays in one child process."""

from __future__ import annotations

import json
import subprocess
import sys
from pathlib import Path

import pytest

from scripts.diagnose_pyjhora_adapter import build_product_report
from scripts.pl9_full_data_export import read_full_data_domain, render_full_data_markdown
from scripts.rectification_three_engine_packet import _pyjhora_d1
from scripts.report_render.annual_method_reference import (
    CHAPTER_EN,
    CHAPTER_ZH,
    build_pyjhora_annual_method_reference,
    normalize_worker_payload,
    mask_report_fingerprint_lines,
    render_annual_method_reference,
    reset_reference_cache,
    set_reference_runner,
    strip_annual_method_chapter,
)
from tests.report_english_fictional import fictional_case

ROOT = Path(__file__).resolve().parents[1]


def setup_function() -> None:
    reset_reference_cache()
    set_reference_runner(None)


def test_result_hash_includes_the_annual_method_chapter() -> None:
    # 原值: 两个标记的 result_hash 相同（本章被摘出指纹）。
    # 新值: 两个标记的 result_hash 不同，且本章仍留在资料包里。
    # 原因: 交付包里有这一章时，重算指纹必须对得上（BUG-1442；与 staging 撞号后改号）。
    from scripts.calculation_profile_contract import bind_result_to_profile

    profile = {"input_hash": "ab" * 32}

    def packet(marker: str) -> dict:
        return {
            "value": 1,
            "worksheets": {
                "timing_and_predictive_systems": {
                    "kept": "same",
                    "pyjhora_annual_method_reference": {"marker": marker},
                }
            },
        }

    installed = bind_result_to_profile(packet("installed"), profile)
    blocked = bind_result_to_profile(packet("blocked"), profile)
    assert installed["result_hash"] != blocked["result_hash"]
    assert installed["worksheets"]["timing_and_predictive_systems"]["pyjhora_annual_method_reference"]["marker"] == "installed"
    assert installed["worksheets"]["timing_and_predictive_systems"]["kept"] == "same"


def test_fingerprint_lines_match_after_masking_for_the_chapter_compare() -> None:
    left = "\n".join([
        "## 报告身份与来源",
        "- 报告 ID: `full-report://one/aaaaaaaaaaaaaaaaaaaaaaaa`",
        "- 计算结果摘要: `0123456789abcdef...`",
        "## 前面的章",
        "same",
        f"## {CHAPTER_ZH}",
        "only-in-left",
    ]) + "\n"
    right = "\n".join([
        "## 报告身份与来源",
        "- 报告 ID: `full-report://two/bbbbbbbbbbbbbbbbbbbbbbbb`",
        "- 计算结果摘要: `fedcba9876543210...`",
        "## 前面的章",
        "same",
        f"## {CHAPTER_ZH}",
        "only-in-right",
    ]) + "\n"
    assert mask_report_fingerprint_lines(strip_annual_method_chapter(left)) == mask_report_fingerprint_lines(
        strip_annual_method_chapter(right)
    )
    assert "aaaaaaaaaaaaaaaaaaaaaaaa" not in mask_report_fingerprint_lines(left)
    assert "0123456789abcdef" not in mask_report_fingerprint_lines(left)


def test_reference_edition_does_not_request_the_annual_method_worker(monkeypatch) -> None:
    from types import SimpleNamespace

    from scripts.professional_report_reference import (
        FULL_DATA_EDITION,
        _assemble_professional_report_reference,
    )

    seen: list[bool] = []

    def build(_reading, _args, _packs, include_annual_method_reference=True):
        seen.append(include_annual_method_reference)
        return {"report_version": "v", "worksheets": {}}

    engine = SimpleNamespace(
        build_professional_report_reference_packet=build,
        render_pl9_markdown=lambda _packet: "standard",
    )
    handler = SimpleNamespace(_compute_full_reading_for_thematic=lambda _birth: {})
    birth = fictional_case(0)
    reference = _assemble_professional_report_reference(
        handler, birth, "markdown", "reference", [], [], False, engine,
    )
    assert reference["markdown"] == "standard"
    assert seen == [False]

    monkeypatch.setattr(
        "scripts.professional_report_reference.render_full_data_markdown",
        lambda _packet: "full-data",
    )
    full = _assemble_professional_report_reference(
        handler, birth, "markdown", FULL_DATA_EDITION, [], [], False, engine,
    )
    assert full["markdown"] == "full-data"
    assert seen == [False, True]


def test_product_report_does_not_probe(monkeypatch) -> None:
    def _boom(*_args, **_kwargs):
        raise AssertionError("product diagnostics must not probe jhora")

    monkeypatch.setattr("scripts.diagnose_pyjhora_adapter.importlib.util.find_spec", _boom)
    report = build_product_report()
    assert report["status"] == "missing_dependency"
    assert report["dependency_available"] is False
    assert report["missing_dependency"] == "jhora"


def test_pyjhora_d1_stays_on_the_missing_install_row() -> None:
    source = (ROOT / "scripts" / "rectification_three_engine_packet.py").read_text(encoding="utf-8")
    assert "import_module" not in source
    assert "import jhora" not in source
    with pytest.raises(ModuleNotFoundError):
        _pyjhora_d1({"year": 1990})


def test_deploy_does_not_set_year_lord_pyjhora_reference() -> None:
    hits = []
    for path in (ROOT / "deploy").rglob("*"):
        if not path.is_file():
            continue
        text = path.read_text(encoding="utf-8", errors="ignore")
        if "JYOTISH_YEAR_LORD_PYJHORA_REFERENCE" in text:
            hits.append(path.relative_to(ROOT).as_posix())
    assert hits == []


def test_timeout_and_nonzero_exit_leave_every_table_unavailable() -> None:
    def _timeout(_payload, timeout):
        raise subprocess.TimeoutExpired(cmd="worker", timeout=timeout)

    set_reference_runner(_timeout)
    reference = build_pyjhora_annual_method_reference(_args(0), {"birth_info": {}})
    assert _all_unavailable(reference)

    def _exit(_payload, _timeout):
        return {"ok": False, "reason": "exit_3"}

    set_reference_runner(_exit)
    reference = build_pyjhora_annual_method_reference(_args(0), {"birth_info": {}})
    assert _all_unavailable(reference)


def test_normalize_keeps_nine_yogas_and_two_mudda_tables() -> None:
    raw = {
        "ok": True,
        "ayanamsa_ok": True,
        "ayanamsa_readback": "RAMAN",
        "node_header": "mean",
        "years": [{
            "requested_year": 2026,
            "anchor": {"date": [2026, 6, 15], "time": "16:35:04"},
            "mudda_birth": [[[0, 1], [2026, 1, 2, 3.5], 30]],
            "mudda_annual": [[[2, "L"], [2026, 6, 15, 1], 10]],
            "patyayini": [
                [[3, 4], [2026, 6, 15, 0], 0.4],
                [[5, 6], [2026, 11, 1, 0], 0.6],
            ],
            "tajika": {
                "yogas": {name: [] for name in (
                    "ithasala", "eesarpha", "manahoo", "kamboola", "nakta", "yamaya",
                )},
                "failures": {"radda": "NameError"},
            },
        }],
    }
    raw["years"][0]["tajika"]["yogas"]["gairi_kamboola"] = None
    raw["years"][0]["tajika"]["yogas"]["khallasara"] = None
    second = json.loads(json.dumps(raw["years"][0]))
    third = json.loads(json.dumps(raw["years"][0]))
    second["requested_year"] = 2027
    third["requested_year"] = 2028
    raw["years"].extend([second, third])
    reference = normalize_worker_payload(raw, 2026)
    tables = reference["years"][0]["tables"]
    assert tables["mudda_birth"]["rows"][0]["main_lord"] == "Sun"
    assert tables["mudda_annual"]["rows"][0]["sub_lord"] == "Lagna"
    assert tables["patyayini"]["rows"][0]["end"] == "2026-11-01 00:00:00"
    assert tables["patyayini"]["rows"][1]["end"] is None
    states = {row["yoga"]: row["state"] for row in tables["tajika"]["rows"]}
    assert len(states) == 9
    assert states["Radda"] == "unavailable"
    assert states["Gairi-Kamboola"] == "unknown"
    assert states["Ithasala"] == "empty"


def test_chapter_is_last_and_domain_view_omits_the_key() -> None:
    packet = {
        "report_language": "zh",
        "worksheets": {"timing_and_predictive_systems": {"pyjhora_annual_method_reference": {
            "years": [_shell(2026), _shell(2027), _shell(2028)],
        }}},
    }
    chinese = render_full_data_markdown(packet)
    english_packet = dict(packet)
    english_packet["report_language"] = "en"
    english = render_full_data_markdown(english_packet)
    assert chinese.rstrip().splitlines()[-1] != ""
    assert _h2(chinese)[-1] == CHAPTER_ZH
    assert _h2(english)[-1] == CHAPTER_EN
    assert chinese.count("来源：PyJHora") == 3
    assert english.count("Source: PyJHora") == 3
    assert "未获得结果" in chinese and "Result unavailable" in english
    assert strip_annual_method_chapter(chinese) == strip_annual_method_chapter(strip_annual_method_chapter(chinese))
    assert CHAPTER_ZH not in strip_annual_method_chapter(chinese)
    domain = read_full_data_domain(packet, "career")
    timing = domain["worksheets"]["timing_and_predictive_systems"]
    assert "pyjhora_annual_method_reference" not in timing


def test_three_fictional_charts_match_a_second_annual_chart_call() -> None:
    """Real PyJHora, three fictional charts. Birth fields stay out of the assertion text."""
    script = r"""
import json, sys
sys.path[:0] = [sys.argv[1], sys.argv[1] + "\\scripts"]
from tests.report_english_fictional import fictional_case
from scripts.report_render.annual_method_reference import _invoke_worker, _payload_from
class Args:
    pass
out = []
for index in (0, 1, 9):
    birth = fictional_case(index)
    args = Args()
    for key, value in birth.items():
        setattr(args, key, value)
    payload = _payload_from(args, {})
    if index == 0:
        payload["prove_reuse"] = True
        payload["include_upstream_diff"] = True
    raw = _invoke_worker(payload, 120)
    years = raw.get("years") or []
    diff = (years[0].get("upstream_diff") if years else None) or {}
    out.append({
        "i": index,
        "ok": bool(raw.get("ok")),
        "ayanamsa_ok": bool(raw.get("ayanamsa_ok")),
        "readback": raw.get("ayanamsa_readback"),
        "node": raw.get("node_header"),
        "reuse": [bool(year.get("reuse_matches")) for year in years],
        "documented_anchor": (diff.get("documented_anchor") or {}).get("date"),
        "upstream_anchor": (diff.get("upstream_anchor") or {}).get("date"),
        "documented_mudda": diff.get("documented_mudda_start"),
        "upstream_mudda": diff.get("upstream_mudda_start"),
        "yoga_keys": sorted(((years[0].get("tajika") or {}).get("yogas") or {}).keys()) if years else [],
        "radda": ((years[0].get("tajika") or {}).get("failures") or {}).get("radda"),
        "jhora_in_parent": any(name == "jhora" or name.startswith("jhora.") for name in sys.modules),
    })
print(json.dumps(out, ensure_ascii=False))
"""
    completed = subprocess.run(
        [sys.executable, "-c", script, str(ROOT)],
        cwd=ROOT,
        text=True,
        capture_output=True,
        timeout=400,
        check=False,
    )
    assert completed.returncode == 0, completed.stderr[-1000:]
    rows = json.loads(completed.stdout)
    assert [row["i"] for row in rows] == [0, 1, 9]
    for row in rows:
        assert row["ok"] and row["ayanamsa_ok"], row
        assert row["readback"]
        if row["i"] == 0:
            assert row["reuse"] == [True, True, True]
            assert row["documented_anchor"] and row["upstream_anchor"]
        assert row["jhora_in_parent"] is False
        assert "radda" not in row["yoga_keys"]
        assert row["radda"]


def test_parent_moon_and_ayanamsa_stay_put() -> None:
    script = r"""
import json, sys
sys.path[:0] = [sys.argv[1], sys.argv[1] + "\\scripts"]
import swisseph as swe
from tests.report_english_fictional import fictional_case
from scripts.report_render.annual_method_reference import build_pyjhora_annual_method_reference
birth = fictional_case(0)
class Args:
    pass
args = Args()
for key, value in birth.items():
    setattr(args, key, value)
jd = swe.julday(2000, 1, 1, 0)
before_ay = swe.get_ayanamsa_ut(jd)
before_moon = swe.calc_ut(jd, swe.MOON)[0][0]
reference = build_pyjhora_annual_method_reference(args, {})
after_ay = swe.get_ayanamsa_ut(jd)
after_moon = swe.calc_ut(jd, swe.MOON)[0][0]
print(json.dumps({
    "same_ayanamsa": before_ay == after_ay,
    "same_moon": before_moon == after_moon,
    "jhora": any(name == "jhora" or name.startswith("jhora.") for name in sys.modules),
    "years": len(reference.get("years") or []),
    "tables": sorted((reference.get("years") or [{}])[0].get("tables", {})),
}))
"""
    completed = subprocess.run(
        [sys.executable, "-c", script, str(ROOT)],
        cwd=ROOT,
        text=True,
        capture_output=True,
        timeout=90,
        check=False,
    )
    assert completed.returncode == 0, completed.stderr[-1000:]
    payload = json.loads(completed.stdout.strip().splitlines()[-1])
    assert payload["same_ayanamsa"] and payload["same_moon"]
    assert payload["jhora"] is False
    assert payload["years"] == 3
    assert payload["tables"] == ["mudda_annual", "mudda_birth", "patyayini", "tajika"]


def _args(index: int):
    birth = fictional_case(index)

    class Args:
        pass

    args = Args()
    for key, value in birth.items():
        setattr(args, key, value)
    return args


def _shell(year: int) -> dict:
    return {
        "requested_year": year,
        "ayanamsa": "RAMAN",
        "ayanamsa_ok": False,
        "node_header": "mean",
        "patyayini_anchor": None,
        "tajika_anchor": None,
        "tables": {name: {"ok": False, "rows": []} for name in ("mudda_birth", "mudda_annual", "patyayini", "tajika")},
    }


def _all_unavailable(reference: dict) -> bool:
    years = reference.get("years") or []
    return len(years) == 3 and all(
        all(not year["tables"][name]["ok"] for name in year["tables"])
        for year in years
    )


def _h2(markdown: str) -> list[str]:
    return [line[3:].strip() for line in markdown.splitlines() if line.startswith("## ")]


def test_rendered_english_chapter_has_no_han() -> None:
    text = render_annual_method_reference({
        "worksheets": {"timing_and_predictive_systems": {"pyjhora_annual_method_reference": {
            "years": [_shell(2026)],
        }}},
    }, "en")
    assert "Source: PyJHora" in text
    assert not any("\u4e00" <= char <= "\u9fff" for char in text)