"""API dasha periods must match the full-reading modules (BUG-1341, BUG-1344).

fixtureProvenance:
  fictional chart from docs/tasks/TASK-dasha-unify-20261013.md section 1
  public charts already stored in tests/smoke_test_runner.py (Elvis Presley)
  and tests/verify_jyotish_v6_1.py (Britney Spears, UT clock, tz 0)
  PyJHora 4.8.7: drik.set_ayanamsa_mode("LAHIRI");
  graha get_dhasa_bhukthi(dob, tob, place, dhasa_level_index=1);
  ashtottari get_ashtottari_dhasa_bhukthi(julian_day_number(dob, tob), place, dhasa_level_index=1)
"""

from __future__ import annotations

import importlib.util
import os
import random
import re
import sys
from datetime import date, datetime
from functools import lru_cache
from pathlib import Path

import pytest

ROOT = Path(__file__).resolve().parents[1]
SCRIPTS = ROOT / "scripts"
os.environ.setdefault("VEDASTRO_ENABLE_NETWORK", "0")

FICTION = {
    "year": 1990,
    "month": 1,
    "day": 1,
    "hour": 12,
    "minute": 0,
    "second": 0,
    "lat": 28.6139,
    "lon": 77.2090,
    "tz": 5.5,
    "ayanamsa": "lahiri",
    "node_mode": "mean",
    "today": "2026-01-01",
}

# Registry key -> full-reading module id. These are the systems T1 compares.
READING_KEYS = (
    ("panchottari", "panchottari_dasha"),
    ("satabdika", "satabdika_dasha"),
    ("chaturaaseeti", "chaturshitisama_dasha"),
    ("shasti_hayani", "shastihayani_dasha"),
    ("shodasottari", "shodashottari_dasha"),
    ("dwadashottari", "dwadashottari_dasha"),
    ("dwisaptati", "dwisaptatisama_dasha"),
    ("shattrimsa", "shattrimshatsama_dasha"),
    ("yogini", "yogini_dasha"),
    ("ashtottari", "ashtottari_dasha"),
    ("kalachakra", "kalachakra_dasha"),
    ("narayana", "narayana_dasha"),
)

# Anchors copied from the six modules before they shared one helper.
BALANCE_ANCHORS = (
    ("panchottari_dasha", "ANURADHA_INDEX"),
    ("satabdika_dasha", "REVATI_INDEX"),
    ("chaturshitisama_dasha", "SWATI_INDEX"),
    ("dwisaptatisama_dasha", "MULA_INDEX"),
    ("shattrimshatsama_dasha", "SRAVANA_INDEX"),
    ("shodashottari_dasha", "PUSHYA_INDEX"),
)

# 原值：白名单只有 dasha_analyzer.py、divisional_charts_extended.py
# 新值：加上 tajika.py
# 原因：calc_mudda_with_dasha_balance 被 test_ported_function_is_verbatim 逐字锁定，
#       函数里的年数字面量不能改成共享表引用。其余非白名单文件仍只能命中 dasha_common.py。
FROZEN_YEAR_TABLES = {
    "dasha_analyzer.py",
    "divisional_charts_extended.py",
    "tajika.py",
}
YEAR_TABLE = re.compile(
    r"""['"]Ketu['"]\s*:\s*7\b.*?['"]Venus['"]\s*:\s*20\b.*?['"]Mercury['"]\s*:\s*17\b""",
    re.DOTALL,
)
PYJHORA_LORDS = {
    0: "Sun",
    1: "Moon",
    2: "Mars",
    3: "Mercury",
    4: "Jupiter",
    5: "Venus",
    6: "Saturn",
    7: "Rahu",
    8: "Ketu",
}


def _handler():
    path = ROOT / "tests" / "test_api_server_security.py"
    spec = importlib.util.spec_from_file_location("dasha_parity_security_helpers", path)
    module = importlib.util.module_from_spec(spec)
    spec.loader.exec_module(module)
    return module._handler()


def _drop_injected_repo_paths() -> None:
    """PyJHora inserts the main checkout onto sys.path. Keep this worktree."""
    banned = {
        os.path.normcase(str(ROOT.parent)),
        os.path.normcase(str(ROOT.parent.parent)),
    }
    sys.path[:] = [
        item
        for item in sys.path
        if not item or os.path.normcase(os.path.abspath(item)) not in banned
    ]
    scripts = str(SCRIPTS)
    if scripts not in sys.path:
        sys.path.insert(0, scripts)


def _major_rows(payload):
    if not isinstance(payload, dict):
        return []
    if isinstance(payload.get("major"), list):
        return payload["major"]
    if isinstance(payload.get("mahadasha_sequence"), list):
        return payload["mahadasha_sequence"]
    if isinstance(payload.get("periods"), list):
        return payload["periods"]
    return []


def _chart_body(report: dict, dasha: str) -> dict:
    chart = report["chart"]
    planets = {
        name: {"lon": row["lon"]}
        for name, row in chart["planets"].items()
        if isinstance(row, dict) and isinstance(row.get("lon"), (int, float))
    }
    return {
        "dasha": dasha,
        "year": FICTION["year"],
        "month": FICTION["month"],
        "day": FICTION["day"],
        "hour": FICTION["hour"],
        "minute": FICTION["minute"],
        "lat": FICTION["lat"],
        "lon": FICTION["lon"],
        "tz": FICTION["tz"],
        "today": FICTION["today"],
        "planets": planets,
        "ascendant": {"lon": chart["ascendant"]["lon"]},
    }


@lru_cache(maxsize=1)
def _fiction_reading() -> dict:
    from scripts.jyotish_engine import cmd_full_reading
    from scripts.professional_report_reference import _export_args

    return cmd_full_reading(_export_args(dict(FICTION)))


def _legacy_birth_balance(moon_longitude, anchor_index, sequence):
    """Source copy of the six helpers before they called dasha_common.

    Operation order matches the deleted bodies: modulo, nakshatra index,
    offset, lord from the sequence, fraction, then years times (1 - fraction).
    """
    moon_longitude = float(moon_longitude) % 360.0
    span = 360.0 / 27.0
    nakshatra_index = int(moon_longitude // span)
    offset = (nakshatra_index - anchor_index) % 27
    starting_lord = sequence[offset % len(sequence)][0]
    years_by_lord = dict(sequence)
    fraction_elapsed = (moon_longitude % span) / span
    remaining_years = years_by_lord[starting_lord] * (1.0 - fraction_elapsed)
    return starting_lord, remaining_years, nakshatra_index


def _as_date(value) -> date:
    if isinstance(value, datetime):
        return value.date()
    if isinstance(value, date):
        return value
    text = str(value)
    return datetime.fromisoformat(text[:19]).date()


def _period_covering(majors, moment: datetime):
    first = datetime.fromisoformat(majors[0]["start_date"])
    last = datetime.fromisoformat(majors[-1]["end_date"])
    span = last - first
    if span.total_seconds() <= 0 or moment < first:
        return None
    cycles = int((moment - first).total_seconds() // span.total_seconds())
    for period in majors:
        start = datetime.fromisoformat(period["start_date"]) + cycles * span
        end = datetime.fromisoformat(period["end_date"]) + cycles * span
        if start <= moment < end:
            return period["planet"], start, end
    return None


def test_api_major_periods_match_full_reading_modules() -> None:
    report = _fiction_reading()
    modules = report["modules"]
    handler = _handler()
    for key, module_id in READING_KEYS:
        engine_rows = _major_rows(modules[module_id])
        result = handler._compute_dasha_system(_chart_body(report, key))
        assert result["precision"] == "engine_module", key
        assert result["precision"] not in {"generic", "calculator"}
        assert result["periods"] == engine_rows, key


def test_unimplemented_dasha_keys_are_rejected() -> None:
    handler = _handler()
    body = _chart_body(_fiction_reading(), "tara")
    from scripts.jyotish_api_server import BadRequest

    with pytest.raises(BadRequest, match="dasha system not implemented"):
        handler._compute_dasha_system(body)
    body["dasha"] = "chara"
    with pytest.raises(BadRequest, match=r"/api/dasha/chara"):
        handler._compute_dasha_system(body)
    source = (SCRIPTS / "jyotish_api_server.py").read_text(encoding="utf-8")
    assert "except BadRequest as e:" in source
    assert "_error_json(str(e), 400, 'ERR_BAD_REQUEST')" in source


def test_narayana_keeps_the_legacy_label() -> None:
    result = _handler()._compute_dasha_system(_chart_body(_fiction_reading(), "narayana"))
    assert "旧算法" in result["algorithm_label"]
    assert result["profile"] == "legacy_reference_only"
    assert "Sanjay Rath" not in result["algorithm_label"]


def test_shoola_uses_the_native_rashi_section() -> None:
    report = _fiction_reading()
    from scripts.rashi_dashas_native import calculate_rashi_dashas

    chart = report["chart"]
    planet_lons = {name: row["lon"] for name, row in chart["planets"].items()}
    expected = calculate_rashi_dashas(
        date(1990, 1, 1),
        chart["ascendant"]["lon"],
        planet_lons,
    )["shoola"]["periods"]
    result = _handler()._compute_dasha_system(_chart_body(report, "shoola"))
    assert result["precision"] == "engine_module"
    assert result["periods"] == expected


def test_birth_balance_matches_the_old_formula() -> None:
    from scripts.dasha_common import nakshatra_seeded_birth_balance

    rng = random.Random(1341)
    samples = [rng.uniform(-40.0, 400.0) for _ in range(2000)]
    for module_name, anchor_name in BALANCE_ANCHORS:
        module = importlib.import_module(f"scripts.{module_name}")
        sequence = module.DASHA_SEQUENCE
        anchor = getattr(module, anchor_name)
        for moon in samples:
            assert module._birth_lord_and_balance(moon) == nakshatra_seeded_birth_balance(
                moon, anchor_index=anchor, sequence=sequence
            )
            assert nakshatra_seeded_birth_balance(
                moon, anchor_index=anchor, sequence=sequence
            ) == _legacy_birth_balance(moon, anchor, sequence)


def test_vimshottari_year_literals_stay_at_the_shared_table() -> None:
    matches = sorted(
        path.name
        for path in SCRIPTS.glob("*.py")
        if path.name not in FROZEN_YEAR_TABLES and YEAR_TABLE.search(path.read_text(encoding="utf-8"))
    )
    assert matches == ["dasha_common.py"]
    frozen = SCRIPTS / "dasha_analyzer.py"
    assert YEAR_TABLE.search(frozen.read_text(encoding="utf-8"))


def _pyjhora_rows(module_name: str, birth: dict):
    pytest.importorskip("jhora")
    _drop_injected_repo_paths()
    from jhora import utils
    from jhora.panchanga import drik

    drik.set_ayanamsa_mode("LAHIRI")
    place = drik.Place("f", birth["lat"], birth["lon"], birth["tz"])
    dob = drik.Date(birth["year"], birth["month"], birth["day"])
    tob = (birth["hour"], birth["minute"], int(birth.get("second", 0)))
    if module_name == "ashtottari":
        from jhora.horoscope.dhasa.graha.ashtottari import get_ashtottari_dhasa_bhukthi

        jd = utils.julian_day_number((birth["year"], birth["month"], birth["day"]), tob)
        return get_ashtottari_dhasa_bhukthi(jd, place, dhasa_level_index=1)
    module = importlib.import_module(f"jhora.horoscope.dhasa.graha.{module_name}")
    return module.get_dhasa_bhukthi(dob, tob, place, dhasa_level_index=1)


def _assert_same_boundaries(engine_rows, py_rows, birth: datetime, slack_days: int) -> None:
    engine = [(_row_lord(row), _as_date(row.get("start_date"))) for row in engine_rows]
    oracle = [(PYJHORA_LORDS[row[0][0]], date(*row[1][:3])) for row in py_rows]
    assert engine[0][0] == oracle[0][0]
    birth_day = birth.date()
    if engine[0][1] == birth_day and oracle[0][1] < birth_day:
        pairs = zip(engine[1:], oracle[1:])
        limit = 0
    else:
        pairs = zip(engine, oracle)
        limit = slack_days
    compared = 0
    for (engine_lord, engine_start), (oracle_lord, oracle_start) in pairs:
        assert engine_lord == oracle_lord
        assert abs((engine_start - oracle_start).days) <= limit
        compared += 1
    assert compared >= 3


def _row_lord(row: dict) -> str:
    return row.get("lord") or row.get("planet")


def test_conditional_tables_match_pyjhora_start_days() -> None:
    from scripts.chaturshitisama_dasha import calculate_chaturshitisama_dasha
    from scripts.dasha_registry import build_reading_birth_info
    from scripts.domain_calculation_service import compute_chart
    from scripts.panchottari_dasha import calculate_panchottari_dasha
    from scripts.satabdika_dasha import calculate_satabdika_dasha
    from scripts.source_bounded_child_periods import PYJHORA_PARENT_SEEDED_EQUAL_SPLIT_PROFILE

    chart = compute_chart({**FICTION, "ayanamsa": "lahiri"})
    birth = datetime(1990, 1, 1, 12, 0)
    _alt, conditional, _lons = build_reading_birth_info(
        birth,
        chart["planets"],
        chart["ascendant"],
        {"lat": FICTION["lat"], "lon": FICTION["lon"], "tz": FICTION["tz"], "year": 1990},
    )
    info = {
        **conditional,
        "child_period_profile": PYJHORA_PARENT_SEEDED_EQUAL_SPLIT_PROFILE,
        "standard_table_profile": "pl9_reports_all_dasha_tables_v1",
    }
    calls = (
        ("panchottari", calculate_panchottari_dasha),
        ("sataatbika", calculate_satabdika_dasha),
        ("chathuraaseethi_sama", calculate_chaturshitisama_dasha),
    )
    for module_name, calculator in calls:
        rows = calculator(info)["major"]
        assert rows
        _assert_same_boundaries(rows, _pyjhora_rows(module_name, FICTION), birth, 0)


def _public_case(name: str, birth: dict, family: str) -> None:
    from scripts.ashtottari_dasha import calculate_ashtottari_dasha
    from scripts.ashtottari_rule_profiles import (
        RULE_FAMILY_MOON_NAKSHATRA_PLUS_PAKSHA,
        RULE_FAMILY_RAHU_FROM_LAGNA_LORD,
    )
    from scripts.dasha_registry import build_reading_birth_info
    from scripts.domain_calculation_service import compute_chart

    families = {
        "moon": RULE_FAMILY_MOON_NAKSHATRA_PLUS_PAKSHA,
        "rahu": RULE_FAMILY_RAHU_FROM_LAGNA_LORD,
    }
    chart = compute_chart({**birth, "second": 0, "node_mode": "mean", "ayanamsa": "lahiri"})
    born = datetime(birth["year"], birth["month"], birth["day"], birth["hour"], birth["minute"])
    alt, conditional, _lons = build_reading_birth_info(
        born,
        chart["planets"],
        chart["ascendant"],
        {"lat": birth["lat"], "lon": birth["lon"], "tz": birth["tz"], "year": birth["year"]},
    )
    result = calculate_ashtottari_dasha({
        **alt,
        "moon_longitude": conditional["moon_longitude"],
        "birth_is_daytime": conditional["birth_is_daytime"],
        "rahu_sign_index": conditional["rahu_sign_index"],
        "lagna_lord_sign_index": conditional["lagna_lord_sign_index"],
        "applicability_rule_family": families[family],
    })
    assert result["applicable"] is True, name
    majors = result["major"]
    start = datetime.fromisoformat(majors[0]["start_date"])
    end = datetime.fromisoformat(majors[0]["end_date"])
    assert start <= born < end
    now = datetime.now()
    covered = _period_covering(majors, now)
    assert covered is not None, name
    current = result["current"]
    assert current["planet"] == covered[0]
    assert datetime.fromisoformat(current["start_date"]) == covered[1]
    assert datetime.fromisoformat(current["end_date"]) == covered[2]
    _assert_same_boundaries(majors, _pyjhora_rows("ashtottari", birth), born, 1)


def test_ashtottari_birth_balance_matches_pyjhora_within_one_day() -> None:
    fiction = {key: FICTION[key] for key in ("year", "month", "day", "hour", "minute", "lat", "lon", "tz")}
    presley = {
        "year": 1935,
        "month": 1,
        "day": 8,
        "hour": 4,
        "minute": 35,
        "lat": 34.3,
        "lon": -88.4,
        "tz": -6.0,
    }
    spears = {
        "year": 1981,
        "month": 12,
        "day": 2,
        "hour": 19,
        "minute": 55,
        "lat": 31.17,
        "lon": -89.18,
        "tz": 0.0,
    }
    _public_case("fiction", fiction, "moon")
    _public_case("presley", presley, "rahu")
    _public_case("spears", spears, "moon")


def test_placeholder_modules_are_gone() -> None:
    assert not (SCRIPTS / "extended_dashas.py").exists()
    assert not (SCRIPTS / "conditional_dashas.py").exists()
    assert not (ROOT / "skills" / "jyotish-vedic-astrology" / "scripts" / "extended_dashas.py").exists()