"""Imports : formats, quarantaine, vérification humaine, application, annulation, sécurité des fichiers."""
import io
import uuid

import pytest

from edcf.core import db as dbmod
from edcf.imports import service
from edcf.imports.detect import FileRejected, safe_display_name, sniff
from edcf.models import ImportFile
from tests import fixtures_gen as fx
from tests.conftest import Api


def upload(api: Api, name: str, data: bytes, mime: str = "application/octet-stream"):
    return api.c.post("/api/imports", files={"file": (name, data, mime)}, headers={"X-CSRF-Token": api.csrf})


def run_jobs():
    with dbmod.session_scope() as db:
        while service.run_next_job(db):
            pass


def analysed(api: Api, name: str, data: bytes):
    r = upload(api, name, data)
    assert r.status_code == 200, r.text
    run_jobs()
    d = api.get(f"/api/imports/{r.json()['id']}").json()
    assert d["status"] == "analyse", d.get("error")
    return d


def selected(d):
    src = next(s for s in d["sources"] if s["index"] == d["selected_source"])
    return [p for p in d["proposals"] if p["source_id"] == src["id"]]


def by_indicator(props):
    return {p["indicator_id"]: p for p in props if p["indicator_id"]}


def test_import_csv_ordre_different_doublon_total_et_semaine(admin: Api):
    d = analysed(admin, "bilan S40.csv", fx.csv_bytes())
    assert d["format"] == "csv" and len(d["sha256"]) == 64 and d["stored_name"] != "bilan S40.csv"
    assert d["week_detection"]["iso_week"] == 40 and d["week_detection"]["iso_year"] == 2026
    props = selected(d)
    ok = by_indicator(props)
    assert len(ok) == 14 and ok[4]["value"] == 1  # « Vitesse > 50 km/h » reconnu
    dups = [p for p in props if p["indicator_id"] == 7]
    assert len(dups) == 2 and all(p["decision"] == "a_verifier" for p in dups)
    total = next(p for p in props if p["raw_label"] == "Total")
    assert total["decision"] == "ignore" and total["indicator_id"] is None


def test_import_xlsx_feuilles_formule_fusion(admin: Api):
    d = analysed(admin, "bilan.xlsx", fx.xlsx_bytes())
    names = [s["name"] for s in d["sources"]]
    assert names == ["Brouillon caché", "Bilan S40"] and d["sources"][0]["hidden"] is True
    props = selected(d)
    ok = by_indicator(props)
    assert 12 not in ok, "ESI absente du fichier : jamais inventée"
    assert ok[6]["value"] == 27 and ok[6]["value_ref"].startswith("B")
    formula = ok[14]
    assert any("formule" in w for w in formula["warnings"]) and formula["decision"] == "a_verifier"
    assert ok[1]["details"] == ["0.92", "1.10"]
    assert d["week_detection"]["iso_week"] == 40


def test_validation_humaine_application_et_annulation(admin: Api):
    d = analysed(admin, "bilan.csv", fx.csv_bytes())
    iid = d["id"]
    r = admin.post(f"/api/imports/{iid}/apply", {"iso_year": 2026, "iso_week": 40})
    assert r.status_code == 422 and r.json()["code"] == "pending"
    for p in selected(d):
        if p["decision"] == "a_verifier":
            decision = "ignore" if p["raw_value"] == "9" or p["indicator_id"] is None else "accepte"
            assert admin.patch(f"/api/imports/{iid}/proposals/{p['id']}", {"decision": decision}).status_code == 200
    r = admin.post(f"/api/imports/{iid}/apply", {"iso_year": 2026, "iso_week": 40})
    assert r.status_code == 200, r.text
    b = admin.get("/api/bilans/2026/40").json()
    assert b["status"] == "a_verifier", "un import ne valide jamais le bilan"
    row = next(x for x in b["rows"] if x["indicator_id"] == 6)
    assert row["value"] == 27 and row["source"]["kind"] == "import" and row["source"]["file_name"] == "bilan.csv"
    assert row["source"]["ref"]
    # Annulation complète : retour à l'état précédent.
    r = admin.post(f"/api/imports/{iid}/cancel")
    assert r.status_code == 200 and r.json()["result"]["reverted"] is True
    b = admin.get("/api/bilans/2026/40").json()
    assert all(x["value"] is None for x in b["rows"]), [(x["indicator_id"], x["value"]) for x in b["rows"]]


def test_correction_d_une_proposition(admin: Api):
    d = analysed(admin, "bilan.csv", fx.csv_bytes())
    p = by_indicator(selected(d))[2]
    r = admin.patch(f"/api/imports/{d['id']}/proposals/{p['id']}", {"value": "7"})
    assert r.status_code == 200 and r.json()["value"] == 7 and r.json()["edited"]
    assert admin.patch(f"/api/imports/{d['id']}/proposals/{p['id']}", {"value": "-3"}).status_code == 422


def test_import_png_ocr_local(admin: Api):
    d = analysed(admin, "photo.png", fx.table_image("PNG"))
    props = selected(d)
    ok = by_indicator(props)
    assert len(ok) >= 11, f"indicateurs reconnus : {len(ok)}"
    correct = sum(1 for i, v in enumerate(fx.VALUES, start=1) if i in ok and ok[i]["value"] == v)
    assert correct >= 10
    assert all(p["decision"] == "a_verifier" for p in props if p["indicator_id"]), "OCR : confirmation humaine"
    assert all(p["bbox"] for p in ok.values()), "zone source conservée"
    src = d["sources"][0]
    img = admin.get(f"/api/imports/{d['id']}/sources/{src['id']}/image")
    assert img.status_code == 200 and img.content[:4] == b"\x89PNG"


def test_import_jpeg_pivote(admin: Api):
    d = analysed(admin, "photo.jpeg", fx.table_image("JPEG", rotate=90))
    assert len(by_indicator(selected(d))) >= 10
    assert d["sources"][0]["width"] < d["sources"][0]["height"] or "Orientation" in " ".join(
        d["sources"][0]["meta"]["warnings"])


def test_import_pdf_texte_et_scanne(admin: Api):
    rows = [{"indicator_id": i, "value": v, "observation": ""} for i, v in enumerate(fx.VALUES, start=1)]
    b = admin.put("/api/bilans/2026/39/draft", {"revision": 0, "rows": rows}).json()
    pdf = admin.get("/api/bilans/2026/39/document/export.pdf").content
    d = analysed(admin, "export.pdf", pdf)
    ok = by_indicator(selected(d))
    assert len(ok) == 14 and all(ok[i]["value"] == v for i, v in enumerate(fx.VALUES, start=1))
    assert d["week_detection"]["iso_week"] == 39
    assert b["revision"] >= 1
    d2 = analysed(admin, "scan.pdf", fx.scanned_pdf())
    assert d2["sources"][0]["meta"]["ocr"] is True and len(by_indicator(selected(d2))) >= 10


def test_roundtrip_export_xlsx_ods(admin: Api):
    rows = [{"indicator_id": i, "value": v, "observation": ""} for i, v in enumerate(fx.VALUES, start=1)]
    rows[10]["value"] = None
    admin.put("/api/bilans/2026/38/draft", {"revision": 0, "rows": rows})
    for fmt in ("xlsx", "ods"):
        data = admin.get(f"/api/bilans/2026/38/document/export.{fmt}").content
        d = analysed(admin, f"retour.{fmt}", data)
        ok = by_indicator(selected(d))
        assert ok[1]["value"] == 3 and ok[6]["value"] == 27
        assert ok[11]["value"] is None, "cellule vide réimportée comme non renseignée"


def test_doublon_de_fichier_signale(admin: Api):
    analysed(admin, "a.csv", fx.csv_bytes())
    d = analysed(admin, "b.csv", fx.csv_bytes())
    assert d["duplicate_of"] and any("Doublon" in w for w in d["warnings"])


def test_lecteur_ne_peut_pas_importer(reader: Api):
    assert upload(reader, "a.csv", fx.csv_bytes()).status_code == 403


# --- Sécurité des fichiers -----------------------------------------------------------------------

@pytest.mark.parametrize("name,data,fragment", [
    ("image.xlsx", fx.table_image("PNG"), "extension"),
    ("tableau.png", fx.csv_bytes(), "extension"),
    ("ancien.xls", b"\xd0\xcf\x11\xe0\xa1\xb1\x1a\xe1" + b"\0" * 600, ".xls"),
    ("macro.xlsx", fx.zip_with({"[Content_Types].xml": b"<Types/>", "xl/workbook.xml": b"<w/>",
                                "xl/vbaProject.bin": b"x"}), "macros"),
    ("xxe.xlsx", fx.zip_with({"[Content_Types].xml": b'<?xml version="1.0"?><!DOCTYPE x [<!ENTITY e SYSTEM "file:///etc/passwd">]><Types/>',
                              "xl/workbook.xml": b"<w/>"}), "XXE"),
    ("traversal.xlsx", fx.zip_with({"[Content_Types].xml": b"<Types/>", "xl/workbook.xml": b"<w/>",
                                    "../../evil.sh": b"x"}), "traversée"),
    ("bombe.xlsx", fx.zip_with({"[Content_Types].xml": b"<Types/>", "xl/workbook.xml": b"<w/>",
                                "xl/worksheets/sheet1.xml": b"0" * (40 * 1024 * 1024)}), "zip bomb"),
    ("actif.pdf", b"%PDF-1.4\n1 0 obj << /OpenAction << /S /JavaScript /JS (app.alert(1)) >> >> endobj\n%%EOF", "actif"),
    ("polyglotte.png", b"\x89PNG\r\n\x1a\n" + b"\0" * 100 + b"<?php system($_GET['c']); ?>", "polyglotte"),
    ("binaire.csv", b"\x00\x01\x02\x03" * 100, "Format non accepté"),
], ids=["fausse-extension", "faux-png", "xls", "macro", "xxe", "traversee", "zip-bomb", "pdf-actif", "polyglotte",
        "binaire"])
def test_fichiers_malveillants_refuses(admin: Api, name, data, fragment):
    r = upload(admin, name, data)
    assert r.status_code == 422, r.text
    assert fragment.lower() in r.json()["message"].lower()


def test_fichier_trop_volumineux(admin: Api):
    r = upload(admin, "gros.csv", b"a;1\n" * (3 * 1024 * 1024))
    assert r.status_code == 413


def test_nom_de_fichier_malveillant(admin: Api):
    r = upload(admin, "../../../etc/passwd;rm -rf.csv", fx.csv_bytes(), "image/png")
    assert r.status_code == 200
    body = r.json()
    assert body["original_name"] == "passwd;rm -rf.csv" and "/" not in body["stored_name"]
    assert body["declared_mime"] == "image/png" and body["mime"] == "text/csv"
    with dbmod.session_scope() as db:
        imp = db.get(ImportFile, uuid.UUID(body["id"]))
        path = service.import_dir(imp.id) / imp.stored_name
        assert path.is_file() and oct(path.stat().st_mode)[-3:] == "600"


def test_noms_assainis():
    assert safe_display_name("C:\\Users\\x\\bilan<1>.xlsx") == "bilan1.xlsx"
    assert safe_display_name("\x00\x1b[31m.csv") == "[31m.csv"
    with pytest.raises(FileRejected):
        sniff(b"", "vide.csv")


def test_objet_d_un_autre_import_inaccessible(admin: Api):
    d1 = analysed(admin, "a.csv", fx.csv_bytes())
    d2 = analysed(admin, "b.xlsx", fx.xlsx_bytes())
    other = selected(d2)[0]
    r = admin.patch(f"/api/imports/{d1['id']}/proposals/{other['id']}", {"decision": "ignore"})
    assert r.status_code == 404
    src2 = d2["sources"][1]["id"]
    assert admin.get(f"/api/imports/{d1['id']}/sources/{src2}/image").status_code == 404