"""Byte-identical report shortcuts and the full_data markdown cache."""

from __future__ import annotations

import copy
import hashlib
import json
import os
import re
import threading
import time
from typing import Any

import pytest

from scripts.calculation_profile_contract import hash_bound_result
from scripts.pl9_full_data_export import render_full_data_markdown
from scripts.pl9_language_terms import _thicken_zh_dasha_density, replace_natural_language_term
from scripts.report_perf_flags import (
    REFERENCE_CACHE_DEFAULT,
    RENDER_FAST_DEFAULT,
    report_reference_cache_enabled,
    report_render_fast_enabled,
)
from scripts.professional_report_reference import build_professional_report_reference
from tests.test_report_english_edition import CASES, _packet


def _legacy_drop_key_recursive(payload: Any, key_name: str) -> Any:
    if isinstance(payload, dict):
        return {
            key: _legacy_drop_key_recursive(value, key_name)
            for key, value in payload.items()
            if key != key_name
        }
    if isinstance(payload, list):
        return [_legacy_drop_key_recursive(item, key_name) for item in payload]
    return payload


def _legacy_drop_dotted_path(payload: Any, path: tuple[str, ...]) -> Any:
    if len(path) == 2 and path[0] == "**":
        return _legacy_drop_key_recursive(payload, path[1])
    if not path or not isinstance(payload, dict) or path[0] not in payload:
        return payload
    cloned = dict(payload)
    key, *rest = path
    if not rest:
        cloned.pop(key, None)
        return cloned
    cloned[key] = _legacy_drop_dotted_path(payload[key], tuple(rest))
    return cloned


def _legacy_canonicalize(value: Any) -> Any:
    if isinstance(value, dict):
        normalized: dict[str, Any] = {}
        for key, child in value.items():
            normalized_key = key if isinstance(key, str) else f"__{type(key).__name__}__:{key}"
            normalized[normalized_key] = _legacy_canonicalize(child)
        return normalized
    if isinstance(value, (list, tuple)):
        return [_legacy_canonicalize(item) for item in value]
    return value


def _legacy_payload(result: dict, scope: dict) -> dict:
    excluded_top = {"result_hash", "result_binding", "calculation_profile", "calculation_profile_id"} | {
        str(key) for key in (scope.get("excluded_top_keys") or ())
    }
    payload = {key: value for key, value in result.items() if key not in excluded_top}
    for raw_path in scope.get("excluded_paths") or ():
        parts = tuple(part for part in str(raw_path).split(".") if part)
        if parts:
            payload = _legacy_drop_dotted_path(payload, parts)
    return payload


def _legacy_hash(result: dict, input_hash: str, scope: dict) -> str:
    encoded = json.dumps(
        _legacy_canonicalize({"input_hash": input_hash, "result": _legacy_payload(result, scope)}),
        ensure_ascii=False,
        sort_keys=True,
        separators=(",", ":"),
        default=str,
    )
    return hashlib.sha256(encoded.encode("utf-8")).hexdigest()


@pytest.fixture(scope="module")
def day_views():
    """One real day packet. ``delivered`` is the bound object; render uses a copy."""
    from scripts.jyotish_engine import build_professional_report_reference_packet, cmd_full_reading
    from scripts.professional_report_reference import _export_args

    os.environ["JYOTISH_REPORT_RENDER_FAST"] = "0"
    os.environ["JYOTISH_REPORT_REFERENCE_CACHE"] = "0"
    try:
        case = CASES["day"]
        reading = cmd_full_reading(type("Args", (), dict(case))())
        delivered = build_professional_report_reference_packet(reading, _export_args(dict(case)), [])
        packet = dict(delivered)
        packet["report_version"] = "pl9_personal_long_report.v3"
        zh = render_full_data_markdown(copy.deepcopy(packet))
        english = copy.deepcopy(packet)
        english["report_language"] = "en"
        en = render_full_data_markdown(english)
    finally:
        os.environ.pop("JYOTISH_REPORT_RENDER_FAST", None)
        os.environ.pop("JYOTISH_REPORT_REFERENCE_CACHE", None)
    return delivered, packet, zh, en


def test_flags_follow_the_environment_at_call_time(monkeypatch) -> None:
    monkeypatch.delenv("JYOTISH_REPORT_RENDER_FAST", raising=False)
    monkeypatch.delenv("JYOTISH_REPORT_REFERENCE_CACHE", raising=False)
    assert report_render_fast_enabled() is RENDER_FAST_DEFAULT
    assert report_reference_cache_enabled() is REFERENCE_CACHE_DEFAULT
    monkeypatch.setenv("JYOTISH_REPORT_RENDER_FAST", "0")
    monkeypatch.setenv("JYOTISH_REPORT_REFERENCE_CACHE", "0")
    assert report_render_fast_enabled() is False
    assert report_reference_cache_enabled() is False
    monkeypatch.setenv("JYOTISH_REPORT_RENDER_FAST", "1")
    monkeypatch.setenv("JYOTISH_REPORT_REFERENCE_CACHE", "1")
    assert report_render_fast_enabled() is True
    assert report_reference_cache_enabled() is True


def test_fast_render_matches_the_legacy_bytes(day_views, monkeypatch) -> None:
    _delivered, packet, zh_off, en_off = day_views
    monkeypatch.setenv("JYOTISH_REPORT_RENDER_FAST", "1")
    monkeypatch.setenv("JYOTISH_REPORT_REFERENCE_CACHE", "0")
    zh_on = render_full_data_markdown(copy.deepcopy(packet))
    english = copy.deepcopy(packet)
    english["report_language"] = "en"
    en_on = render_full_data_markdown(english)
    assert zh_on == zh_off
    assert en_on == en_off


def _both(function, monkeypatch, *args):
    monkeypatch.setenv("JYOTISH_REPORT_RENDER_FAST", "0")
    off = function(*args)
    monkeypatch.setenv("JYOTISH_REPORT_RENDER_FAST", "1")
    on = function(*args)
    assert on == off
    return off


def test_dasha_density_samples_come_from_the_real_render(day_views, monkeypatch) -> None:
    _delivered, _packet_view, zh, _en = day_views
    lines = [line for line in zh.splitlines() if line.strip()]
    # This fictional chart never emits the density sentence. Fill the existing
    # sanitizer sentence with lords that do appear in the render.
    lords = [token for token in ("木星", "北交点", "土星", "金星") if token in zh]
    assert len(lords) >= 2
    hit = f"{lords[0]}主大运中的{lords[1]}子运，会把{lords[0]}的自然主题带入当前阶段。"
    miss = next(line for line in lines if "子运" not in line and "`" not in line)
    quoted = next(line for line in lines if "`" in line)
    hit_out = _both(_thicken_zh_dasha_density, monkeypatch, hit + "\n")
    assert "这一子运不是单独结论" in hit_out or "这个次子运只承担短周期细化" in hit_out
    miss_out = _both(_thicken_zh_dasha_density, monkeypatch, miss + "\n")
    assert miss_out == miss
    quoted_out = _both(_thicken_zh_dasha_density, monkeypatch, quoted + "\n")
    assert "`" in quoted_out
    assert quoted in quoted_out


def test_term_samples_come_from_the_real_render(day_views, monkeypatch) -> None:
    _delivered, _packet_view, zh, _en = day_views
    lines = zh.splitlines()
    raw = "Ashtakavarga"
    hit = next(line for line in lines if raw in re.sub(r"`[^`]*`", "", line))
    miss = next(line for line in lines if line.strip() and raw not in line)
    span = next(span for span in re.findall(r"`[^`]*`", zh) if raw not in span)
    hit_out = _both(replace_natural_language_term, monkeypatch, hit, raw, "探针词")
    assert "探针词" in hit_out or hit_out == hit
    if re.search(rf"(?<![A-Za-z0-9_]){raw}(?![A-Za-z0-9_])", re.sub(r"`[^`]*`", "", hit)):
        assert "探针词" in hit_out
    miss_out = _both(replace_natural_language_term, monkeypatch, miss, raw, "探针词")
    assert miss_out == miss
    quoted_out = _both(replace_natural_language_term, monkeypatch, span, raw, "探针词")
    assert quoted_out == span


def test_fast_hash_matches_the_legacy_algorithm_on_a_real_packet(day_views, monkeypatch) -> None:
    packet, _render_packet, _zh, _en = day_views
    scope = packet["result_binding"]["binding_scope"]
    input_hash = packet["calculation_profile"]["input_hash"]
    monkeypatch.setenv("JYOTISH_REPORT_RENDER_FAST", "1")
    assert hash_bound_result(packet, input_hash, scope) == _legacy_hash(packet, input_hash, scope)
    assert hash_bound_result(packet, input_hash, scope) == packet["result_hash"]
    variant = copy.deepcopy(packet)
    variant["probe_layer"] = {
        "elapsed_seconds": 1.5,
        "called_at": "2001-01-01T00:00:00Z",
        "cache_created_at": "created",
        "cache_expires_at": "expires",
        "nested": {"elapsed_seconds": 9, (1, 2): {"called_at": "inside"}},
    }
    summary = variant.setdefault("ai_and_audit", {}).setdefault("summary", {})
    summary["elapsed_seconds"] = 99
    summary["kept"] = "yes"
    assert hash_bound_result(variant, input_hash, scope) == _legacy_hash(variant, input_hash, scope)
    monkeypatch.setenv("JYOTISH_REPORT_RENDER_FAST", "0")
    assert hash_bound_result(packet, input_hash) == packet["result_hash"]


class _CountingHandler:
    def __init__(self, birth: dict, packet: dict) -> None:
        self.birth = birth
        self.packet = packet
        self.calls = 0
        self.fail_once = False
        self.delay = 0.0
        self.lock = threading.Lock()

    def _high_rigor_birth_payload(self, body: dict) -> dict:
        payload = dict(self.birth)
        if body.get("today"):
            payload["today"] = body["today"]
        payload["candidate_range"] = body.get("candidate_range")
        return payload

    def _compute_full_reading_for_thematic(self, _birth: dict) -> dict:
        with self.lock:
            self.calls += 1
            call = self.calls
        if self.delay:
            time.sleep(self.delay)
        if self.fail_once and call == 1:
            raise RuntimeError("report_build_failed")
        return {"stub": True}


def _body(**extra) -> dict:
    body = {
        "format": "markdown",
        "edition": "full_data",
        "include_fact_tables": True,
        "languages": ["zh", "en"],
        "today": "2026-04-07",
    }
    body.update(extra)
    return body


@pytest.fixture
def cached_report(monkeypatch, tmp_path, day_views):
    import scripts.consult_full_data_cache as cache
    import scripts.professional_report_reference as reference

    def _dir(name: str = "consult_full_data_cache"):
        path = tmp_path / name
        path.mkdir(parents=True, exist_ok=True)
        return path

    monkeypatch.setattr(cache, "cache_dir", _dir)
    monkeypatch.setattr(reference, "render_full_data_markdown", lambda packet: "en-body" if packet.get("report_language") == "en" else "zh-body")
    monkeypatch.setattr(reference, "report_density_packet", lambda _packet: {"facts": 1})
    monkeypatch.setattr(reference, "render_dasha_applicability", lambda _packet, language="zh": f"app-{language}")
    monkeypatch.setattr(reference, "clean_reader_appendix_markdown", lambda text: text)
    packet = day_views[1]
    birth = {key: value for key, value in CASES["day"].items() if key != "name"}
    handler = _CountingHandler(birth, packet)

    class _Engine:
        def build_professional_report_reference_packet(self, _reading, _args, _packs):
            return copy.deepcopy(packet)

        def render_pl9_markdown(self, _packet):
            return "reference"

    monkeypatch.setenv("JYOTISH_REPORT_REFERENCE_CACHE", "1")
    monkeypatch.setenv("JYOTISH_REPORT_RENDER_FAST", "0")
    monkeypatch.setenv("GITHUB_SHA", "report-cache-test")
    return handler, _Engine(), _dir("report_reference_cache")


def _dumps(result: dict) -> str:
    return json.dumps(result, ensure_ascii=False, default=str)


def test_second_call_reuses_the_cached_response(cached_report) -> None:
    handler, engine, folder = cached_report
    first = build_professional_report_reference(handler, _body(), engine=engine)
    second = build_professional_report_reference(handler, _body(), engine=engine)
    assert handler.calls == 1
    assert _dumps(first) == _dumps(second)
    assert list(folder.glob("*.json"))


@pytest.mark.parametrize("extra", [
    {"today": "2026-05-01"},
    {"languages": ["zh"]},
    {"include_fact_tables": False},
    {"candidate_range": {"kind": "probe"}},
])
def test_changed_inputs_miss_the_report_cache(cached_report, extra) -> None:
    handler, engine, _folder = cached_report
    build_professional_report_reference(handler, _body(), engine=engine)
    build_professional_report_reference(handler, _body(**extra), engine=engine)
    assert handler.calls == 2


def test_github_sha_is_part_of_the_report_cache_key(cached_report, monkeypatch) -> None:
    handler, engine, _folder = cached_report
    build_professional_report_reference(handler, _body(), engine=engine)
    monkeypatch.setenv("GITHUB_SHA", "report-cache-other")
    build_professional_report_reference(handler, _body(), engine=engine)
    assert handler.calls == 2


def test_json_format_and_disabled_cache_write_nothing(cached_report, monkeypatch) -> None:
    handler, engine, folder = cached_report
    for path in folder.glob("*.json"):
        path.unlink()
    build_professional_report_reference(
        handler,
        {"format": "json", "edition": "full_data", "today": "2026-04-07"},
        engine=engine,
    )
    assert list(folder.glob("*.json")) == []
    monkeypatch.setenv("JYOTISH_REPORT_REFERENCE_CACHE", "0")
    build_professional_report_reference(handler, _body(), engine=engine)
    assert list(folder.glob("*.json")) == []
    monkeypatch.setenv("JYOTISH_REPORT_REFERENCE_CACHE", "1")
    monkeypatch.setenv("GITHUB_SHA", "unknown")
    build_professional_report_reference(handler, _body(), engine=engine)
    assert list(folder.glob("*.json")) == []


def test_concurrent_same_key_builds_once(cached_report) -> None:
    handler, engine, _folder = cached_report
    handler.delay = 0.2
    barrier = threading.Barrier(2)
    errors = []

    def run() -> None:
        try:
            barrier.wait()
            build_professional_report_reference(handler, _body(), engine=engine)
        except Exception as exc:  # pragma: no cover - recorded for the assertion
            errors.append(exc)

    threads = [threading.Thread(target=run) for _ in range(2)]
    for thread in threads:
        thread.start()
    for thread in threads:
        thread.join()
    assert errors == []
    assert handler.calls == 1


def test_failed_build_is_not_cached(cached_report) -> None:
    handler, engine, folder = cached_report
    handler.fail_once = True
    with pytest.raises(RuntimeError):
        build_professional_report_reference(handler, _body(), engine=engine)
    assert list(folder.glob("*.json")) == []
    build_professional_report_reference(handler, _body(), engine=engine)
    assert handler.calls == 2
    build_professional_report_reference(handler, _body(), engine=engine)
    assert handler.calls == 2


def test_report_cache_directory_prunes_old_files(cached_report) -> None:
    handler, engine, folder = cached_report
    stale = folder / "stale-probe.json"
    stale.write_text("{}", encoding="utf-8")
    old = time.time() - (3 * 24 * 60 * 60)
    os.utime(stale, (old, old))
    build_professional_report_reference(handler, _body(), engine=engine)
    assert not stale.exists()