app/analysis.py

"""Pipeline d'analyse IA pour une demande de vérification Kelo ID.

Trois vérifications automatiques :
1. Correspondance visage pièce d'identité <-> selfie (DeepFace.verify)
2. Estimation d'âge à partir du visage (DeepFace.analyze)
3. Score de probabilité "deepfake" sur les deux images (modèle ViT
   Hugging Face dédié)

Aucune de ces trois vérifications n'est fiable à 100% — voir le
README pour le détail des limites. Le verdict final est volontairement
prudent : tout résultat ambigu route vers la file de modération
humaine plutôt que de trancher seul.
"""

import io
import logging
from dataclasses import dataclass

import cv2
import numpy as np
from PIL import Image

logger = logging.getLogger("kelo_id_ai.analysis")

FACE_MATCH_THRESHOLD = 0.55
DEEPFAKE_REJECT_THRESHOLD = 0.75
DEEPFAKE_AMBIGUOUS_THRESHOLD = 0.4
MIN_AGE_REQUIRED = 13


@dataclass
class AnalysisResult:
    face_match_score: float | None
    estimated_age: float | None
    deepfake_score: float | None
    verdict: str
    reason: str


def _extract_frame_from_video(video_bytes: bytes) -> np.ndarray:
    import tempfile

    with tempfile.NamedTemporaryFile(suffix=".webm") as tmp:
        tmp.write(video_bytes)
        tmp.flush()
        cap = cv2.VideoCapture(tmp.name)
        frame_count = int(cap.get(cv2.CAP_PROP_FRAME_COUNT))
        middle_frame_index = max(frame_count // 2, 0)
        cap.set(cv2.CAP_PROP_POS_FRAMES, middle_frame_index)
        success, frame = cap.read()
        cap.release()

        if not success or frame is None:
            raise ValueError("Impossible d'extraire une frame de la vidéo selfie.")

        return cv2.cvtColor(frame, cv2.COLOR_BGR2RGB)


def _bytes_to_pil(image_bytes: bytes) -> Image.Image:
    return Image.open(io.BytesIO(image_bytes)).convert("RGB")


def run_face_match_and_age(
    id_document_bytes: bytes, selfie_frame_rgb: np.ndarray
) -> tuple[float | None, float | None, str | None]:
    from deepface import DeepFace

    id_document_array = np.array(_bytes_to_pil(id_document_bytes))

    try:
        verify_result = DeepFace.verify(
            img1_path=id_document_array,
            img2_path=selfie_frame_rgb,
            model_name="VGG-Face",
            detector_backend="opencv",
            enforce_detection=True,
        )
        distance = verify_result["distance"]
        threshold = verify_result["threshold"]
        similarity = max(0.0, 1 - (distance / (threshold * 2)))
    except ValueError as exc:
        logger.warning("Échec détection visage: %s", exc)
        return None, None, "Aucun visage détecté sur la pièce d'identité ou le selfie."

    try:
        analyze_result = DeepFace.analyze(
            img_path=selfie_frame_rgb,
            actions=["age"],
            detector_backend="opencv",
            enforce_detection=True,
        )
        estimated_age = float(analyze_result[0]["age"])
    except ValueError as exc:
        logger.warning("Échec estimation d'âge: %s", exc)
        estimated_age = None

    return similarity, estimated_age, None


_deepfake_pipeline = None


def _get_deepfake_pipeline():
    global _deepfake_pipeline
    if _deepfake_pipeline is None:
        from transformers import pipeline

        _deepfake_pipeline = pipeline(
            "image-classification",
            model="prithivMLmods/Deep-Fake-Detector-v2-Model",
        )
    return _deepfake_pipeline


def run_deepfake_score(image_bytes: bytes) -> float | None:
    try:
        pipe = _get_deepfake_pipeline()
        image = _bytes_to_pil(image_bytes)
        results = pipe(image)
        fake_score = next(
            (r["score"] for r in results if r["label"].lower() in ("fake", "deepfake")),
            None,
        )
        return fake_score
    except Exception as exc:  # noqa: BLE001
        logger.error("Erreur détection deepfake: %s", exc)
        return None


def analyze_verification(id_document_bytes: bytes, selfie_video_bytes: bytes) -> AnalysisResult:
    try:
        selfie_frame = _extract_frame_from_video(selfie_video_bytes)
    except Exception as exc:  # noqa: BLE001
        logger.error("Erreur extraction frame vidéo: %s", exc)
        return AnalysisResult(
            face_match_score=None,
            estimated_age=None,
            deepfake_score=None,
            verdict="ambigu",
            reason="Impossible de lire la vidéo selfie. Routé vers la modération humaine.",
        )

    face_match_score, estimated_age, face_error = run_face_match_and_age(
        id_document_bytes, selfie_frame
    )

    selfie_frame_bytes = _ndarray_to_jpeg_bytes(selfie_frame)
    deepfake_score_document = run_deepfake_score(id_document_bytes)
    deepfake_score_selfie = run_deepfake_score(selfie_frame_bytes)

    deepfake_scores = [s for s in (deepfake_score_document, deepfake_score_selfie) if s is not None]
    deepfake_score = max(deepfake_scores) if deepfake_scores else None

    if face_error:
        return AnalysisResult(
            face_match_score=None,
            estimated_age=estimated_age,
            deepfake_score=deepfake_score,
            verdict="ambigu",
            reason=face_error,
        )

    if deepfake_score is not None and deepfake_score >= DEEPFAKE_REJECT_THRESHOLD:
        return AnalysisResult(
            face_match_score=face_match_score,
            estimated_age=estimated_age,
            deepfake_score=deepfake_score,
            verdict="refuse",
            reason="Indice élevé de contenu généré ou manipulé par IA.",
        )

    if face_match_score is not None and face_match_score < FACE_MATCH_THRESHOLD:
        return AnalysisResult(
            face_match_score=face_match_score,
            estimated_age=estimated_age,
            deepfake_score=deepfake_score,
            verdict="refuse",
            reason="Le visage du selfie ne correspond pas suffisamment à celui de la pièce d'identité.",
        )

    if estimated_age is not None and estimated_age < MIN_AGE_REQUIRED:
        return AnalysisResult(
            face_match_score=face_match_score,
            estimated_age=estimated_age,
            deepfake_score=deepfake_score,
            verdict="refuse",
            reason="Âge estimé inférieur à l'âge minimum requis.",
        )

    is_ambiguous = (
        (deepfake_score is not None and deepfake_score >= DEEPFAKE_AMBIGUOUS_THRESHOLD)
        or face_match_score is None
        or estimated_age is None
    )

    if is_ambiguous:
        return AnalysisResult(
            face_match_score=face_match_score,
            estimated_age=estimated_age,
            deepfake_score=deepfake_score,
            verdict="ambigu",
            reason="Résultat incertain — routé vers la modération humaine.",
        )

    return AnalysisResult(
        face_match_score=face_match_score,
        estimated_age=estimated_age,
        deepfake_score=deepfake_score,
        verdict="accepte",
        reason="Toutes les vérifications automatiques sont passées.",
    )


def _ndarray_to_jpeg_bytes(frame_rgb: np.ndarray) -> bytes:
    image = Image.fromarray(frame_rgb)
    buffer = io.BytesIO()
    image.save(buffer, format="JPEG")
    return buffer.getvalue()