from __future__ import annotations

import hashlib
import json
import os
import re
import sqlite3
from pathlib import Path
from typing import Any

import jarvis_finance.services.household_classification as classification
import jarvis_finance.services.household_import as household_import
from jarvis_finance.services.household_import import preview_household_import

DB = Path("/home/agent/jarvis_runtime/finance-system/data/finance.sqlite3")
PAYLOAD = Path("/home/agent/jarvis_runtime/finance-system/sprint16.1-uat/real-preview-payload-private.json")
OLD = Path("/home/agent/jarvis_runtime/finance-system/sprint16.2-analysis/sprint16.2-local-preview-private.json")
ROOT = Path("/home/agent/jarvis_runtime/finance-system/sprint16.3-analysis")
SOURCES = [Path("/home/agent/.hermes/private/finance/sprint16") / name for name in ("raiffeisen.csv", "akb_transactions.csv", "visa.csv", "migros.csv")]
NEW_ONLY_TOKENS = {
    "h&m", "c&a", "c+a", "dr med dent", "puregym", "justeat", "comobitdefender",
    "restaurants und verpflegung", "tanken parkieren und fahrzeug",
}


def sha_file(path: Path) -> str:
    digest = hashlib.sha256()
    with path.open("rb") as handle:
        for chunk in iter(lambda: handle.read(1024 * 1024), b""):
            digest.update(chunk)
    return digest.hexdigest()


def table_digest(conn: sqlite3.Connection, table: str) -> str:
    digest = hashlib.sha256(table.encode())
    for row in conn.execute(f'SELECT * FROM "{table}" ORDER BY rowid'):
        digest.update(json.dumps(list(row), ensure_ascii=False, default=str, separators=(",", ":")).encode())
    return digest.hexdigest()


def snapshot() -> dict[str, Any]:
    conn = sqlite3.connect(f"file:{DB}?mode=ro", uri=True)
    tables = [row[0] for row in conn.execute("SELECT name FROM sqlite_master WHERE type='table' AND name NOT LIKE 'sqlite_%' ORDER BY name")]
    result = {
        "tables": {table: {"count": conn.execute(f'SELECT COUNT(*) FROM "{table}"').fetchone()[0], "digest": table_digest(conn, table)} for table in tables},
        "foreign_keys": list(conn.execute("PRAGMA foreign_key_check")),
        "integrity": conn.execute("PRAGMA integrity_check").fetchone()[0],
    }
    conn.close()
    return result


def private_write(path: Path, value: Any) -> None:
    temporary = path.with_suffix(path.suffix + ".tmp")
    temporary.write_text(json.dumps(value, ensure_ascii=False, indent=2), encoding="utf-8")
    os.chmod(temporary, 0o600)
    os.replace(temporary, path)
    os.chmod(path, 0o600)


def legacy_family(value: Any) -> dict[str, str] | None:
    merchant = classification.normalize_merchant(value)
    padded = f" {merchant} "
    for family_id, label, category_name, tokens in classification.MERCHANT_FAMILY_RULES:
        for token in tokens:
            if token in NEW_ONLY_TOKENS:
                continue
            normalized = classification.normalize_merchant(token)
            if normalized and f" {normalized} " in padded:
                return {"family_id": family_id, "label": label, "category_name": category_name}
    return None


def stable_response(value: dict[str, Any]) -> dict[str, Any]:
    return {key: item for key, item in value.items() if key != "performance"}


def run() -> None:
    before = snapshot()
    source_before = {str(path): sha_file(path) for path in SOURCES}
    payload = json.loads(PAYLOAD.read_text(encoding="utf-8"))
    old = json.loads(OLD.read_text(encoding="utf-8"))

    new_family = classification.merchant_family
    new_classification_version = classification.CLASSIFICATION_VERSION
    classification.merchant_family = legacy_family
    household_import.merchant_family = legacy_family
    classification.CLASSIFICATION_VERSION = "household_classification_v2.2"
    household_import.CLASSIFICATION_VERSION = "household_classification_v2.2"
    conn = sqlite3.connect(f"file:{DB}?mode=ro", uri=True)
    conn.row_factory = sqlite3.Row
    parity = preview_household_import(conn, payload)
    conn.close()
    classification.merchant_family = new_family
    household_import.merchant_family = new_family
    classification.CLASSIFICATION_VERSION = new_classification_version
    household_import.CLASSIFICATION_VERSION = new_classification_version

    normalized_old = json.loads(json.dumps(old))
    normalized_parity = stable_response(parity)
    normalized_old.pop("preview_fingerprint", None)
    normalized_parity.pop("preview_fingerprint", None)
    normalized_old["review_threshold"]["maximum_individual_review_count"] = 49
    parity_checks = {
        "documented_threshold_delta_passed": normalized_parity == normalized_old,
        "preview_fingerprint_changed_for_threshold": parity["preview_fingerprint"] != old["preview_fingerprint"],
        "counts_identical": parity["counts"] == old["counts"],
        "rows_identical": parity["rows"] == old["rows"],
        "items_identical": parity["items"] == old["items"],
        "clusters_identical": parity["merchant_clusters"] == old["merchant_clusters"],
        "hard_uat_gate_seconds": parity["performance"]["api_total_seconds"],
        "hard_uat_gate_passed": parity["performance"]["api_total_seconds"] <= 120,
    }
    parity_debug = ROOT / "sprint16.3-parity-debug-private.json"
    parity_debug.write_text(json.dumps({"checks": parity_checks, "parity": parity}, ensure_ascii=False), encoding="utf-8")
    os.chmod(parity_debug, 0o600)
    if not all(value for key, value in parity_checks.items() if key.endswith("identical") or key.endswith("passed")):
        raise RuntimeError(f"performance parity or UAT gate failed: {parity_checks}")

    conn = sqlite3.connect(f"file:{DB}?mode=ro", uri=True)
    conn.row_factory = sqlite3.Row
    calibrated = preview_household_import(conn, payload)
    conn.close()

    after = snapshot()
    source_after = {str(path): sha_file(path) for path in SOURCES}
    evidence = {
        "parity_checks": parity_checks,
        "calibrated_summary": {
            "classification_version": calibrated["classification_version"],
            "counts": calibrated["counts"],
            "review_threshold": calibrated["review_threshold"],
            "technically_confirmable": calibrated["technically_confirmable"],
            "business_ready_for_confirm": calibrated["business_ready_for_confirm"],
            "performance": calibrated["performance"],
        },
        "db_changed_tables": [table for table in before["tables"] if before["tables"][table] != after["tables"].get(table)],
        "integrity": [before["integrity"], after["integrity"]],
        "foreign_key_counts": [len(before["foreign_keys"]), len(after["foreign_keys"])],
        "source_hashes_unchanged": source_before == source_after,
        "confirm_called": False,
    }
    if evidence["db_changed_tables"] or evidence["integrity"] != ["ok", "ok"] or evidence["foreign_key_counts"] != [0, 0] or not evidence["source_hashes_unchanged"]:
        raise RuntimeError("read-only evidence failed")
    private_write(ROOT / "sprint16.3-performance-parity-private.json", parity)
    private_write(ROOT / "sprint16.3-calibrated-preview-private.json", calibrated)
    private_write(ROOT / "sprint16.3-optimized-preview-evidence-private.json", evidence)
    print(json.dumps(evidence, ensure_ascii=False, indent=2))


if __name__ == "__main__":
    run()