"""full-reading keeps the upstream module keys and does not go online."""

from __future__ import annotations

import importlib.util
import json
from pathlib import Path

import pytest

from scripts.jaimini import SIGNS
from scripts.jyotish_engine import cmd_full_reading
from scripts.professional_report_reference import _export_args
from scripts.punarphoo import detect_punarphoo
from scripts.rashi_dashas_native import SIGN_LORDS, navamsha_sign

ROOT = Path(__file__).resolve().parents[1]
# 原值：docs/research/upstream_full_synastry_bde5d615/.../validate_full_synastry_packets.py.txt
# 新值：docs/research/upstream_full_synastry_7a773d13/.../validate_full_synastry_packets.py.txt
# 原因：修改单 M3。新校验器把 Niryaana Shoola 固定判为 not_applicable，合盘不用这套大运。
VALIDATOR = (
    ROOT
    / "docs/research/upstream_full_synastry_7a773d13/skill/scripts/validate_full_synastry_packets.py.txt"
)

# Baseline full-reading keys on origin/staging b3ef2d815, plus the five keys this
# change adds. 81 + 5 = the upstream 86-key set.
UPSTREAM_MODULE_KEYS = frozenset({
    "actionable_context", "argala", "ashtakavarga", "ashtottari_dasha", "aspects",
    "audit", "avasthas", "bhava_bala", "bhava_chalit", "bhrigu_pada_dasha",
    "career_strict_evidence", "chart", "chaturshitisama_dasha", "congregation",
    "d9_navamsa_expanded", "dasa_convergence", "dasha", "dasha_sandhi",
    "dispositor_chains", "double_transit_pac", "dwadashottari_dasha",
    "dwisaptatisama_dasha", "final_dispositors", "finance_strict_evidence",
    "functional_benefic_malefic", "guided_topics", "house_map", "inter_chart_linkage",
    "jaimini", "kalachakra_dasha", "kp", "lagna_vargottamamsa", "marriage_counting",
    "moon_chart", "muntha", "nakshatra_advanced", "nakshatra_dasha", "narayana_dasha",
    "navamsha_dasha", "niryaana_shoola_dasha", "pancha_pakshi", "panchanga",
    "panchottari_dasha", "planetary_congregation", "planetary_friendship", "prashna",
    "punarphoo", "pushkara", "pushkara_planets", "rashi_tulya_navamsa",
    "relationship_strict_evidence", "remedies", "sahams", "satabdika_dasha",
    "sensitive_points", "shadbala", "shastihayani_dasha", "shattrimshatsama_dasha",
    "shodashottari_dasha", "solar_return", "source_priority", "special_lagnas",
    "sudarshana", "tajika", "tajika_yogas", "tithi_lord", "transit_ll7l",
    "transit_multi_reference", "transit_positions", "tribhagi_40_dasha",
    "tribhagi_dasha", "trimshamsa_d30", "unified_western_reading", "upagrahas",
    "validation", "varga_extended", "varga_full", "varga_research_high", "vargottama",
    "vedastro_official_full_snapshot", "vedastro_range_scan_result", "vimsopaka",
    "vivah_saham", "yoga", "yogas_doshas", "yogini_dasha",
})


def _case(path: Path, case_id: str) -> dict:
    rows = json.loads(path.read_text(encoding="utf-8"))
    birth = next(row for row in rows if row["id"] == case_id)
    return {
        "year": birth["year"],
        "month": birth["month"],
        "day": birth["day"],
        "hour": birth["hour"],
        "minute": birth["minute"],
        "second": birth.get("second") or 0,
        "lat": birth["lat"],
        "lon": birth["lon"],
        "tz": birth["tz"],
        "ayanamsa": "lahiri",
        "node_mode": "mean",
        "today": "2026-10-18",
    }


@pytest.fixture(scope="module")
def reports():
    cases = {
        "pitt": _case(ROOT / "tests/indastro_cases.json", "brad_pitt"),
        "jolie": _case(ROOT / "tests/celebrity_cases.json", "jolie"),
    }
    return {name: cmd_full_reading(_export_args(birth)) for name, birth in cases.items()}


def _validator():
    from importlib.machinery import SourceFileLoader

    loader = SourceFileLoader("validate_full_synastry_packets", str(VALIDATOR))
    spec = importlib.util.spec_from_loader(loader.name, loader)
    module = importlib.util.module_from_spec(spec)
    loader.exec_module(module)
    return module


def _goel_method_1_first(d9_lagna: int, lord_sign_by_sign: dict[int, int]) -> tuple[int, int]:
    start = d9_lagna if d9_lagna % 2 == 0 else (d9_lagna + 6) % 12
    lord_sign = int(lord_sign_by_sign[start]) % 12
    if start % 2 == 0:
        distance = (lord_sign - start) % 12
    else:
        distance = (start - lord_sign) % 12
    years = 12 if distance == 0 else distance + 1
    return start, years


def test_unified_western_reading_module_imports() -> None:
    assert not (ROOT / "scripts" / "_script_root.py").exists()
    import classical_astrology
    import unified_western_reading

    assert callable(unified_western_reading.build_unified_western_reading)
    assert callable(classical_astrology.build_classical_astrology_packet)


def test_engine_does_not_import_the_retired_niryaana_file() -> None:
    source = (ROOT / "scripts/jyotish_engine.py").read_text(encoding="utf-8")
    assert "from niryaana_shoola_dasha import" not in source
    assert "import niryaana_shoola_dasha" not in source


def test_full_reading_module_keys_cover_the_upstream_set(reports) -> None:
    assert len(UPSTREAM_MODULE_KEYS) == 86
    for name, report in reports.items():
        missing = UPSTREAM_MODULE_KEYS - set(report["modules"])
        assert not missing, name


def test_simple_modules_match_the_upstream_shapes(reports) -> None:
    for report in reports.values():
        modules = report["modules"]
        lagna = modules["lagna_vargottamamsa"]
        assert set(lagna) == {"d1_sign", "d9_sign", "is_lagna_vargottamamsa", "source"}
        assert lagna["source"] == "raman_horoscope_v1_bounded_pp64"
        assert lagna["is_lagna_vargottamamsa"] is bool(
            lagna["d1_sign"] and lagna["d9_sign"] and lagna["d1_sign"] == lagna["d9_sign"]
        )
        flags = modules["pushkara"]
        expected = [
            f"{planet} ({data['sign']} {data['degree_in_sign']:.1f}°)"
            for planet, data in flags.items()
            if data.get("pushkara_navamsa") or data.get("pushkara_bhaga")
        ]
        assert modules["pushkara_planets"] == expected
        assert modules["punarphoo"] == detect_punarphoo(report["chart"]["planets"])


def test_secondary_dashas_use_native_profiles_and_match_goel_method_1(reports) -> None:
    for report in reports.values():
        modules = report["modules"]
        navamsha = modules["navamsha_dasha"]
        niryaana = modules["niryaana_shoola_dasha"]
        assert navamsha["status"] == "parameter_sensitive"
        assert navamsha["profile_id"] == "goel_method_1_v1"
        # 原值：niryaana == {execution_status: blocked, reason: 口径更正中，见 TASK-niryaana-shoola-pvr-20261018}
        # 新值：status parameter_sensitive，profile pvr_niryaana_shoola_v1，periods 是 PVR 主运
        # 原因：#84 用真结果换掉 #81 的占位。合盘校验器仍把这一行固定判为不适用。
        assert niryaana["status"] == "parameter_sensitive"
        assert niryaana["profile_id"] == "pvr_niryaana_shoola_v1"
        assert niryaana["periods"] == niryaana["native_result"]["periods"]
        assert niryaana["native_result"]["status"] == "listed"
        assert "antardasha" not in niryaana["periods"][0]
        assert sum(row["years"] for row in niryaana["periods"]) == 96
        assert navamsha["native_result"]["status"] == "listed"
        first = navamsha["periods"][0]
        d9 = modules["varga_full"]["D9_Navamsa"]
        d9_lagna = int(d9["Ascendant"]["sign_idx"])
        lord_signs = {
            index: int(d9[SIGN_LORDS[SIGNS[index]]]["sign_idx"])
            for index in range(12)
        }
        # Upstream goel_method_1_v1 reads the lord's D9 sign. Ours derives the
        # same sign from the natal longitude. Both must name the same first major.
        assert (first["sign_index"], first["years"]) == _goel_method_1_first(d9_lagna, lord_signs)
        longitudes = {
            name: float(row["degree_raw"])
            for name, row in report["chart"]["planets"].items()
            if isinstance(row, dict) and isinstance(row.get("degree_raw"), (int, float))
        }
        asc_lon = float(report["chart"]["ascendant"]["degree_raw"])
        longitude_lords = {
            index: navamsha_sign(longitudes[SIGN_LORDS[SIGNS[index]]])
            for index in range(12)
        }
        assert (first["sign_index"], first["years"]) == _goel_method_1_first(
            navamsha_sign(asc_lon), longitude_lords
        )


def test_synastry_validator_accepts_the_secondary_dashas(reports) -> None:
    validator = _validator()
    receipt = validator.build_execution_receipt(reports["pitt"], reports["jolie"])
    rows = {row["technique"]: row for row in receipt["individual_technique_audit"]}
    assert rows["Navamsha Dasha"]["status"] in {"executed", "parameter_sensitive"}
    # 原值：Niryaana Shoola 必须是 executed 或 parameter_sensitive
    # 新值：not_applicable
    # 原因：修改单 M3。上游 7a773d13 校验器对这一行固定判不适用，合盘任务不因它 blocked。
    assert rows["Niryaana Shoola Dasha"]["status"] == "not_applicable"
    tasks = {row["task"]: row for row in receipt["pair_execution_tasks"]}
    assert tasks["secondary-dasha convergence and disagreement review"]["status"] != "blocked"


def test_full_reading_still_succeeds_when_network_and_subprocess_raise() -> None:
    import subprocess
    import urllib.request

    from unified_western_reading import build_unified_western_reading

    calls = {"count": 0, "stacks": []}
    original_urlopen = urllib.request.urlopen
    original_run = subprocess.run

    def boom(*_args, **_kwargs):
        import traceback

        calls["count"] += 1
        calls["stacks"].append("".join(traceback.format_stack()))
        raise RuntimeError("network and external programs are not allowed")

    urllib.request.urlopen = boom
    subprocess.run = boom
    try:
        birth = _case(ROOT / "tests/indastro_cases.json", "brad_pitt")
        report = cmd_full_reading(_export_args(birth))
        western = report["modules"]["unified_western_reading"]
        assert western["status"] != "blocked"
        # VedAstro's existing local runner may attempt subprocess and fail closed.
        # The western packet must not be on that stack.
        for stack in calls["stacks"]:
            assert "unified_western_reading.py" not in stack
            assert "classical_astrology.py" not in stack
        calls["count"] = 0
        packet = build_unified_western_reading(
            year=birth["year"],
            month=birth["month"],
            day=birth["day"],
            hour=birth["hour"],
            minute=birth["minute"],
            second=birth["second"],
            latitude=birth["lat"],
            longitude=birth["lon"],
            timezone=birth["tz"],
            reference_date=birth["today"],
            duration_scan_end_date=birth["today"],
            lunar_return_months=3,
        )
        assert packet["system"] == "western_astrology"
        assert calls["count"] == 0
    finally:
        urllib.request.urlopen = original_urlopen
        subprocess.run = original_run