"""Recette pédagogique Nymphar, données fictives, aucune connexion distante.

python recette.py > resultat.json
Les scénarios négatifs doivent être détectés ; leur détection fait réussir la recette.
Les valeurs attendues appartiennent uniquement à factures.csv fourni.
"""
import csv
import hashlib
import io
import json
import platform
import tempfile
from pathlib import Path

import duckdb


def quote_path(path):
    return str(path).replace("'", "''")


def profile(con, relation):
    row = con.execute(f"""
        SELECT count(*), count(DISTINCT facture_id),
               count(*) FILTER (WHERE montant_eur IS NULL),
               count(*) FILTER (WHERE montant_eur = 0),
               sum(montant_eur), min(facture_id), max(facture_id)
        FROM {relation}
    """).fetchone()
    return dict(zip(["lignes", "identifiants", "montants_absents", "zeros",
                     "total_eur", "premier_id", "dernier_id"],
                    [*row[:4], str(row[4]), *row[5:]]))


EXPECTED = {"lignes": 10, "identifiants": 10, "montants_absents": 1,
            "zeros": 1, "total_eur": "500.00", "premier_id": "0001", "dernier_id": "0010"}


def evaluate(source, parquet_root, missing_partition=False):
    con = duckdb.connect(":memory:")
    try:
        con.execute("SET threads = 2")
        con.execute("SET memory_limit = '512MB'")
        con.execute(f"""
            CREATE TABLE factures AS
            SELECT * FROM read_csv('{quote_path(source)}',
              auto_detect = false, header = true, delim = ';',
              decimal_separator = ',', nullstr = '',
              columns = {{'facture_id':'VARCHAR', 'date_facture':'DATE',
                          'pays':'VARCHAR', 'montant_eur':'DECIMAL(12,2)'}})
        """)
        source_profile = profile(con, "factures")
        con.execute(f"""
            COPY (SELECT *, year(date_facture) AS annee,
                            month(date_facture) AS mois FROM factures)
            TO '{quote_path(parquet_root)}'
            (FORMAT parquet, PARTITION_BY (annee, mois))
        """)
        files = sorted(parquet_root.rglob("*.parquet"))
        if missing_partition:
            files = [p for p in files if "mois=9" in p.parts]
        # Liste explicite : l'inventaire des fichiers est une partie de la recette.
        sql_files = "[" + ",".join("'" + quote_path(p) + "'" for p in files) + "]"
        con.execute(f"""CREATE VIEW relu AS SELECT * FROM read_parquet(
            {sql_files}, hive_partitioning = true)""")
        parquet_profile = profile(con, "relu")
        # EXCEPT ALL conserve la multiplicité ; une comparaison de sommes seule ne suffit pas.
        cols = "facture_id, date_facture, pays, montant_eur"
        difference = con.execute(f"""
            SELECT count(*) FROM (
                (SELECT {cols} FROM factures EXCEPT ALL SELECT {cols} FROM relu)
                UNION ALL
                (SELECT {cols} FROM relu EXCEPT ALL SELECT {cols} FROM factures)
            )
        """).fetchone()[0]
        countries = {r[0]: str(r[1]) for r in con.execute(
            "SELECT pays, sum(montant_eur) FROM relu GROUP BY pays ORDER BY pays").fetchall()}
        months = con.execute("""SELECT mois, count(*), sum(montant_eur)
                                FROM relu GROUP BY mois ORDER BY mois""").fetchall()
        max_partition = con.execute("SELECT max(mois) FROM relu WHERE mois = 9").fetchone()[0]
        plan = con.execute("EXPLAIN SELECT max(mois) FROM relu WHERE mois = 9").fetchone()[1]
        checks = {
            "reference_metier_csv": source_profile == EXPECTED,
            "reference_metier_parquet": parquet_profile == EXPECTED,
            "egalite_lignes_csv_parquet": difference == 0,
            "totaux_pays": countries == {"DE": "100.00", "FR": "400.00"},
            "partitions": [(r[0], r[1], str(r[2])) for r in months] ==
                          [(9, 5, "210.00"), (10, 5, "290.00")],
            "max_partition_filtree": max_partition == 9,
        }
        return {"checks": checks, "accepted": all(checks.values()),
                "csv": source_profile, "parquet": parquet_profile,
                "difference_lignes": difference, "totaux_pays": countries,
                "partitions": [[r[0], r[1], str(r[2])] for r in months],
                "max_partition_filtree": max_partition,
                "fichiers_lus": [str(p.relative_to(parquet_root)) for p in files],
                "plan_max_partition": plan}
    finally:
        con.close()


def main():
    fixture = Path(__file__).with_name("factures.csv")
    original = fixture.read_text(encoding="utf-8")
    rows = list(csv.DictReader(io.StringIO(original), delimiter=";"))
    scenarios = ["nominal", "doublon", "centimes", "partition_manquante", "valeur_invalide"]
    results = []
    # Répertoire unique, aucune écriture dans un fichier de base existant.
    with tempfile.TemporaryDirectory(prefix="nymphar-duckdb-recette-") as work:
        root = Path(work)
        for name in scenarios:
            selected = [dict(row) for row in rows]
            if name == "doublon":
                selected.append(dict(selected[0]))
            elif name == "centimes":
                selected[0]["montant_eur"] = "10010,00"
            elif name == "valeur_invalide":
                selected[0]["montant_eur"] = "inconnu"
            path = root / f"{name}.csv"
            with path.open("w", encoding="utf-8", newline="") as handle:
                writer = csv.DictWriter(handle, fieldnames=rows[0].keys(), delimiter=";")
                writer.writeheader()
                writer.writerows(selected)
            try:
                observed = evaluate(path, root / name, name == "partition_manquante")
            except duckdb.Error as exc:
                if name != "valeur_invalide":
                    raise
                observed = {"accepted": False, "error_type": type(exc).__name__,
                            "error": str(exc).replace(str(root), "<atelier>")}
            expected_accept = name == "nominal"
            results.append({"scenario": name, "expected_accept": expected_accept,
                            "behaviour_as_expected": observed["accepted"] == expected_accept,
                            **observed})
    report = {"duckdb_version": duckdb.__version__, "python": platform.python_version(),
              "platform": platform.platform(), "data": "synthetic",
              "source_sha256": hashlib.sha256(fixture.read_bytes()).hexdigest(),
              "expected": EXPECTED, "results": results,
              "success": all(r["behaviour_as_expected"] for r in results)}
    print(json.dumps(report, ensure_ascii=False, indent=2))
    raise SystemExit(0 if report["success"] else 1)


if __name__ == "__main__":
    main()