from __future__ import annotations

import hashlib
import json
import sqlite3
import subprocess
import sys
from datetime import date, timedelta
from pathlib import Path

import pytest
from openpyxl import Workbook

ROOT = Path(__file__).resolve().parents[1]
sys.path.insert(0, str(ROOT / "scripts" / "health"))
sys.path.insert(0, str(ROOT / "tests"))

from dashboard_v5.read_api import APIError, dispatch_api  # noqa: E402
from fixtures.dashboard_v5_fixture import build_dashboard_v5_fixture  # noqa: E402
import lab_source_reconciliation as reconciliation  # noqa: E402
from lab_source_reconciliation import (  # noqa: E402
    _digest,
    apply_exact_backfills,
    run_audit,
    staging_plan,
)


def _fixture(tmp_path: Path) -> Path:
    db = tmp_path / "a2.db"
    build_dashboard_v5_fixture(db)
    connection = sqlite3.connect(db)
    connection.executemany(
        "INSERT OR IGNORE INTO dokumente_status(status) VALUES(?)",
        [("neu",), ("eingearbeitet",), ("archiviert",)],
    )
    connection.commit()
    connection.close()
    subprocess.run(
        [sys.executable, str(ROOT / "tests" / "fixtures" / "dashboard_v5_original_fixture.py"),
         "--db", str(db), "--root", str(tmp_path / "originals")],
        check=True,
    )
    return db


def _query(payload: dict, metric: str) -> list[dict]:
    return [item for item in payload["observations"] if item["metric_id"] == metric]


def test_lab_history_metadata_matches_series_and_ignores_clicked_row(tmp_path):
    db = _fixture(tmp_path)
    record = dispatch_api(db, "/api/v1/record-labs", "q=CRP")
    crp = _query(record, "lab.crp")
    assert len(crp) == 5
    assert crp[0]["history"] == crp[2]["history"] == crp[-1]["history"]
    history = crp[0]["history"]
    assert history == {
        "metric_id": "lab.crp",
        "earliest_date": "2024-01-10",
        "latest_date": "2026-05-01",
        "observation_count": 5,
        "verified_observation_count": 5,
        "missing_reference_count": 1,
        "linked_document_count": 5,
        "unlinked_document_count": 0,
        "range_from": "2024-01-10",
        "range_to": record["today"],
    }
    series = dispatch_api(
        db,
        "/api/v1/series",
        f"metric=lab.crp&resolution=day&from={history['range_from']}&to={history['range_to']}",
    )
    assert len(series["points"]) == len(crp) == 5
    assert [point["date"] for point in series["points"]] == sorted(point["date"] for point in series["points"])
    assert len({point["date"] for point in series["points"]}) == 5
    newest_only = dispatch_api(
        db,
        "/api/v1/record-labs",
        f"q=CRP&from={crp[0]['date']}&to={record['today']}",
    )
    assert len(_query(newest_only, "lab.crp")) == 1
    assert newest_only["histories"][0]["observation_count"] == 5


def test_record_labs_and_series_remain_count_consistent_above_100_rows(tmp_path):
    db = _fixture(tmp_path)
    connection = sqlite3.connect(db)
    connection.row_factory = sqlite3.Row
    template = dict(connection.execute(
        "SELECT * FROM laborwerte WHERE parameter_name LIKE '%CRP%' ORDER BY id LIMIT 1"
    ).fetchone())
    columns = [name for name in template if name != "id"]
    placeholders = ",".join("?" for _ in columns)
    for index in range(101):
        row = dict(template)
        day = (date(2020, 1, 1) + timedelta(days=index)).isoformat()
        row["abnahme_datum"] = day
        if "befund_datum" in row:
            row["befund_datum"] = day
        connection.execute(
            f"INSERT INTO laborwerte({','.join(columns)}) VALUES({placeholders})",
            [row[name] for name in columns],
        )
    connection.commit()
    connection.close()
    record = dispatch_api(db, "/api/v1/record-labs", "q=CRP")
    crp = _query(record, "lab.crp")
    history = next(item for item in record["histories"] if item["metric_id"] == "lab.crp")
    series = dispatch_api(
        db,
        "/api/v1/series",
        f"metric=lab.crp&resolution=day&from={history['range_from']}&to={history['range_to']}",
    )
    assert len(crp) == history["observation_count"] == len(series["points"]) == 106
    assert record["truncated"] is False


def test_reference_ranges_are_observation_specific_and_missing_stays_unknown(tmp_path):
    db = _fixture(tmp_path)
    series = dispatch_api(
        db,
        "/api/v1/series",
        "metric=lab.crp&resolution=day&from=2024-01-10&to=2026-07-17",
    )
    assert len(series["points"]) == 5
    references = [point.get("reference") for point in series["points"]]
    assert references[0] != references[1]
    missing = next(point for point in series["points"] if point["date"] == "2025-05-20")
    assert missing["reference_status"] == "missing_reference"
    assert "reference" not in missing


def test_document_review_status_never_changes_lab_series_count(tmp_path):
    db = _fixture(tmp_path)
    before = dispatch_api(db, "/api/v1/series", "metric=lab.crp&resolution=day&from=2024-01-10&to=2026-07-17")
    connection = sqlite3.connect(db)
    connection.execute("UPDATE dokumente SET review_status='nicht_geprueft' WHERE id=(SELECT canonical_document_id FROM laborwerte WHERE canonical_document_id IS NOT NULL LIMIT 1)")
    connection.commit()
    connection.close()
    after = dispatch_api(db, "/api/v1/series", "metric=lab.crp&resolution=day&from=2024-01-10&to=2026-07-17")
    assert len(before["points"]) == len(after["points"]) == 5
    labs = dispatch_api(db, "/api/v1/record-labs", "q=CRP")
    assert all(item["document_link_status"] == "linked" for item in _query(labs, "lab.crp"))
    assert all(item["document_id"] is None for item in _query(labs, "lab.crp"))


def test_single_observation_contract_and_duplicate_parameters_fail_closed(tmp_path):
    db = _fixture(tmp_path)
    record = dispatch_api(db, "/api/v1/record-labs", "q=Faktor%20VIII")
    factor = _query(record, "lab.factor_viii")
    assert len(factor) == 1
    assert factor[0]["history"]["observation_count"] == 1
    series = dispatch_api(
        db,
        "/api/v1/series",
        f"metric=lab.factor_viii&resolution=day&from={factor[0]['history']['range_from']}&to={factor[0]['history']['range_to']}",
    )
    assert len(series["points"]) == 1
    for query in (
        "metric=lab.crp&metric=lab.factor_viii&resolution=day&from=2024-01-10&to=2026-07-17",
        "metric=lab.crp&resolution=day&from=2024-01-10&from=2025-01-01&to=2026-07-17",
        "metric=lab.crp&resolution=day&from=2024-01-10&to=2026-07-17&to=2026-07-16",
    ):
        with pytest.raises(APIError) as error:
            dispatch_api(db, "/api/v1/series", query)
        assert error.value.status == 400


def _workbook(path: Path) -> None:
    workbook = Workbook()
    sheet = workbook.active
    assert sheet is not None
    sheet.cell(1, 3, "2024-01-10")
    sheet.cell(1, 4, "2023-02-02")
    sheet.cell(2, 1, "C-Reaktives Protein (CRP)")
    sheet.cell(2, 2, "0-5 mg/L")
    sheet.cell(2, 3, "4,2")
    sheet.cell(2, 4, "3,1")
    workbook.save(path)


def test_reconciliation_is_read_only_and_workbook_plan_never_verifies(tmp_path):
    db = _fixture(tmp_path)
    xlsx = tmp_path / "synthetic.xlsx"
    _workbook(xlsx)
    before = hashlib.sha256(db.read_bytes()).hexdigest()
    audit = run_audit(db, xlsx)
    after = hashlib.sha256(db.read_bytes()).hexdigest()
    assert before == after
    aggregate = audit["aggregate"]
    assert aggregate["mode"] == "read_only_fail_closed"
    assert aggregate["classifications"]["exact_match"] == 1
    assert aggregate["classifications"]["workbook_only"] == 1
    assert aggregate["parity"]["workbook_exact_in_canonical_v5"] == 1
    plan = staging_plan(audit)
    assert plan["mode"] == "plan_only_no_database_write"
    assert plan["candidates"]
    assert all(item["status"] == "zur_pruefung" for item in plan["candidates"])
    assert all(item["provenance"] == "legacy_reference_xlsx" for item in plan["candidates"])
    assert all(item["verified_against_original"] == 0 for item in plan["candidates"])
    assert all(item["reference_range_source"] != "scanned_original" for item in plan["candidates"])
    assert plan["covered_count"] == plan["discrepancy_count"]
    assert len(plan["candidates"]) + len(plan["excluded"]) == plan["discrepancy_count"]


def test_exact_link_backfill_is_bounded_idempotent_and_preserves_validation(tmp_path, monkeypatch):
    monkeypatch.setattr(reconciliation, "PRIVATE_REPORT_DIR", tmp_path / "private")
    db = _fixture(tmp_path)
    connection = sqlite3.connect(db)
    document_id = connection.execute("SELECT id FROM dokumente ORDER BY id LIMIT 1").fetchone()[0]
    lab_id = connection.execute("SELECT id FROM laborwerte WHERE parameter_name='Faktor VIII'").fetchone()[0]
    connection.execute("UPDATE laborwerte SET dokument_id=?,canonical_document_id=NULL WHERE id=?", (document_id, lab_id))
    before_validation = connection.execute("SELECT validierungsstatus,verified_against_original,reference_range_source FROM laborwerte WHERE id=?", (lab_id,)).fetchone()
    connection.commit()
    connection.close()
    candidates = [{"lab_id": lab_id, "document_id": document_id}]
    result = apply_exact_backfills(db, candidates, tmp_path / "private" / "backups", _digest(candidates))
    assert result["changed"] == 1
    assert result["idempotent_second_run_changes"] == 0
    assert result["integrity"] == "ok" and result["restore_test"] is True
    connection = sqlite3.connect(db)
    linked, *validation = connection.execute("SELECT canonical_document_id,validierungsstatus,verified_against_original,reference_range_source FROM laborwerte WHERE id=?", (lab_id,)).fetchone()
    connection.close()
    assert linked == document_id
    assert tuple(validation) == before_validation
    with pytest.raises(RuntimeError):
        apply_exact_backfills(db, [{"lab_id": lab_id, "document_id": document_id + 1}], tmp_path / "private" / "backups", "wrong")


def test_exact_link_backfill_rolls_back_trigger_side_effects(tmp_path, monkeypatch):
    monkeypatch.setattr(reconciliation, "PRIVATE_REPORT_DIR", tmp_path / "private")
    db = _fixture(tmp_path)
    connection = sqlite3.connect(db)
    document_id = connection.execute("SELECT id FROM dokumente ORDER BY id LIMIT 1").fetchone()[0]
    lab_id = connection.execute("SELECT id FROM laborwerte WHERE parameter_name='Faktor VIII'").fetchone()[0]
    before_status = connection.execute("SELECT validierungsstatus FROM laborwerte WHERE id=?", (lab_id,)).fetchone()[0]
    connection.execute("UPDATE laborwerte SET dokument_id=?,canonical_document_id=NULL WHERE id=?", (document_id, lab_id))
    connection.execute(
        """CREATE TRIGGER mutate_validation_after_link AFTER UPDATE OF canonical_document_id ON laborwerte
           WHEN NEW.canonical_document_id IS NOT NULL
           BEGIN UPDATE laborwerte SET validierungsstatus='triggered' WHERE id=NEW.id; END"""
    )
    connection.commit()
    connection.close()
    candidates = [{"lab_id": lab_id, "document_id": document_id}]
    with pytest.raises(RuntimeError, match="validation fields changed"):
        apply_exact_backfills(
            db, candidates, tmp_path / "private" / "backups", _digest(candidates)
        )
    connection = sqlite3.connect(db)
    linked, status = connection.execute(
        "SELECT canonical_document_id,validierungsstatus FROM laborwerte WHERE id=?", (lab_id,)
    ).fetchone()
    connection.close()
    assert linked is None
    assert status == before_status


def test_exact_link_backfill_rolls_back_cross_table_trigger_side_effects(tmp_path, monkeypatch):
    monkeypatch.setattr(reconciliation, "PRIVATE_REPORT_DIR", tmp_path / "private")
    db = _fixture(tmp_path)
    connection = sqlite3.connect(db)
    document_id = connection.execute("SELECT id FROM dokumente ORDER BY id LIMIT 1").fetchone()[0]
    lab_id = connection.execute("SELECT id FROM laborwerte WHERE parameter_name='Faktor VIII'").fetchone()[0]
    before_review = connection.execute("SELECT review_status FROM dokumente WHERE id=?", (document_id,)).fetchone()[0]
    connection.execute("UPDATE laborwerte SET dokument_id=?,canonical_document_id=NULL WHERE id=?", (document_id, lab_id))
    connection.execute(
        f"""CREATE TRIGGER mutate_document_after_link AFTER UPDATE OF canonical_document_id ON laborwerte
            WHEN NEW.canonical_document_id IS NOT NULL
            BEGIN UPDATE dokumente SET review_status='triggered' WHERE id={int(document_id)}; END"""
    )
    connection.commit()
    connection.close()
    candidates = [{"lab_id": lab_id, "document_id": document_id}]
    with pytest.raises(RuntimeError, match="database changed outside intended canonical links"):
        apply_exact_backfills(
            db, candidates, tmp_path / "private" / "backups", _digest(candidates)
        )
    connection = sqlite3.connect(db)
    linked = connection.execute("SELECT canonical_document_id FROM laborwerte WHERE id=?", (lab_id,)).fetchone()[0]
    review = connection.execute("SELECT review_status FROM dokumente WHERE id=?", (document_id,)).fetchone()[0]
    connection.close()
    assert linked is None
    assert review == before_review


def test_private_writer_rejects_symlinks_and_replaces_permissive_file(tmp_path, monkeypatch):
    private_root = tmp_path / "private"
    monkeypatch.setattr(reconciliation, "PRIVATE_REPORT_DIR", private_root)
    destination = private_root / "report.json"
    destination.parent.mkdir(parents=True)
    destination.write_text("old", encoding="utf-8")
    destination.chmod(0o644)
    reconciliation._write_private(destination, {"safe": True})
    assert destination.stat().st_mode % 4096 == 0o600
    outside = tmp_path / "outside.json"
    link = private_root / "linked.json"
    link.symlink_to(outside)
    with pytest.raises(RuntimeError, match="non-symlink"):
        reconciliation._write_private(link, {"safe": False})
    assert not outside.exists()
    with pytest.raises(RuntimeError, match="private report root"):
        reconciliation._write_private(tmp_path / "repository-like.json", {"safe": False})


def test_private_writer_rejects_symlink_root_without_chmod_side_effect(tmp_path, monkeypatch):
    external = tmp_path / "external"
    external.mkdir(mode=0o755)
    linked_root = tmp_path / "linked-private-root"
    linked_root.symlink_to(external, target_is_directory=True)
    monkeypatch.setattr(reconciliation, "PRIVATE_REPORT_DIR", linked_root)
    before_mode = external.stat().st_mode % 4096
    with pytest.raises(RuntimeError, match="must not be symlinks"):
        reconciliation._write_private(linked_root / "report.json", {"safe": False})
    assert external.stat().st_mode % 4096 == before_mode == 0o755
    assert not (external / "report.json").exists()


def test_ambiguous_relations_are_not_backfill_candidates(tmp_path):
    db = _fixture(tmp_path)
    connection = sqlite3.connect(db)
    document_ids = [row[0] for row in connection.execute("SELECT id FROM dokumente ORDER BY id LIMIT 2")]
    lab_id = connection.execute("SELECT id FROM laborwerte ORDER BY id LIMIT 1").fetchone()[0]
    connection.execute("UPDATE laborwerte SET dokument_id=?,canonical_document_id=? WHERE id=?", (*document_ids, lab_id))
    connection.commit()
    connection.close()
    xlsx = tmp_path / "synthetic.xlsx"
    _workbook(xlsx)
    audit = run_audit(db, xlsx)
    assert audit["aggregate"]["classifications"]["ambiguous_document_link"] == 1
    assert all(candidate["lab_id"] != lab_id for candidate in audit["exact_backfills"])
    serialized = json.dumps(audit["aggregate"])
    assert "synthetic.pdf" not in serialized
    assert "/tmp/" not in serialized