"""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()