Spaces:
Sleeping
Sleeping
| """ | |
| Unkor — Détection deepfake hybride, 100 % LOCAL (CPU) | |
| ====================================================== | |
| Duo : détecteur IA généraliste ONNX (umm-maybe/AI-image-detector par défaut) pour | |
| les images générées (SDXL, Midjourney, Gemini, SD…) + EfficientNet/FaceForensics++ | |
| (ONNX) pour les visages, complétés par ELA / FFT / SRM (30 filtres) / EXIF, | |
| fusion par vote pondéré. | |
| Débogage : tous les scores BRUTS sont journalisés sur stdout (désactivable via | |
| UNKOR_DEBUG=0). Voir [scores] dans les logs. | |
| """ | |
| import os | |
| import io | |
| import threading | |
| import numpy as np | |
| import cv2 | |
| from PIL import Image | |
| from concurrent.futures import ThreadPoolExecutor | |
| _VOTE_THRESHOLD = 0.50 # un analyseur compte comme anomalie dès 50 % | |
| _FACE_VOTE_THRESHOLD = 0.72 # visage : biais EfficientNet ~0.60 sur vrais visages -> seuil relevé | |
| _BASE_DIR = os.path.dirname(os.path.abspath(__file__)) | |
| _MODELS_DIR = os.path.join(_BASE_DIR, "models") | |
| _DEBUG = os.getenv("UNKOR_DEBUG", "1").lower() not in ("0", "false", "no") | |
| def _dbg(msg): | |
| if _DEBUG: | |
| line = f"[scores] {msg}" | |
| try: | |
| print(line, flush=True) | |
| except UnicodeEncodeError: | |
| print(line.encode("ascii", "replace").decode("ascii"), flush=True) | |
| # ═══════════════════════════════════════════════════════════════════════════════ | |
| # 1. ELA — Error Level Analysis (blocs 8×8 + énergie AC DCT) | |
| # ═══════════════════════════════════════════════════════════════════════════════ | |
| class ELAAnalyzer: | |
| """ | |
| Recompresse en JPEG q90, mesure l'erreur de recompression, puis l'analyse | |
| par blocs 8×8 (grille JPEG) : énergie moyenne, variance inter-blocs | |
| (uniformité), et énergie AC (coefficients DCT hors DC) des blocs d'erreur. | |
| Réel JPEG = erreur structurée le long de la grille ; synthétique = erreur | |
| faible et très homogène. | |
| """ | |
| QUALITY = 90 | |
| def analyze(self, rgb): | |
| try: | |
| return self._run(rgb) | |
| except Exception as e: | |
| _dbg(f"ELA exception: {e}") | |
| return {"ela_score": 0.5, "ela_mean": 0.0, "ela_blockvar": 0.0, "ela_ac": 0.0} | |
| def _run(self, rgb): | |
| buf = io.BytesIO() | |
| Image.fromarray(rgb).save(buf, "JPEG", quality=self.QUALITY) | |
| buf.seek(0) | |
| recomp = np.asarray(Image.open(buf).convert("RGB"), dtype=np.float32) | |
| err = np.abs(rgb.astype(np.float32) - recomp).mean(axis=2) # carte d'erreur | |
| h, w = err.shape | |
| H, W = (h // 8) * 8, (w // 8) * 8 | |
| if H < 8 or W < 8: | |
| return {"ela_score": 0.40, "ela_mean": float(err.mean()), "ela_cv": 0.0, "ela_ac": 0.0} | |
| blocks = (err[:H, :W].reshape(H // 8, 8, W // 8, 8) | |
| .swapaxes(1, 2).reshape(-1, 8, 8)) | |
| bmean = blocks.mean(axis=(1, 2)) | |
| mean_energy = float(bmean.mean()) | |
| block_cv = float(np.std(bmean) / (mean_energy + 1e-6)) # uniformité RELATIVE (scale-invariant) | |
| # Énergie AC (DCT) sur un échantillon de blocs (structure de l'erreur) | |
| n = len(blocks) | |
| idx = np.linspace(0, n - 1, min(n, 400)).astype(int) | |
| ac = 0.0 | |
| for i in idx: | |
| d = cv2.dct(blocks[i].astype(np.float32)) | |
| d[0, 0] = 0.0 | |
| ac += float(np.abs(d).sum()) | |
| ac_energy = ac / max(1, len(idx)) | |
| # ELA recalibré (anti-faux-positifs) : la compression JPEG NORMALE donne une | |
| # erreur FAIBLE mais STRUCTURÉE (concentrée sur textures/bords -> CV de bloc | |
| # élevé). On NE signale donc QUE les cartes d'erreur anormalement UNIFORMES | |
| # (CV bas) ET sans structure AC — les deux requis (produit), pour ne pas | |
| # pénaliser les photos lisses réelles (qui gardent de l'AC via le bruit capteur). | |
| flat = float(np.clip((0.45 - block_cv) / 0.45, 0, 1)) if block_cv < 0.45 else 0.0 | |
| noac = float(np.clip((8.0 - ac_energy) / 8.0, 0, 1)) if ac_energy < 8.0 else 0.0 | |
| synth = flat * noac # signature diffusion : uniforme ET sans AC | |
| # DIFFUSION moderne (SDXL/MJ/Gemini : aplats lisses, ~zéro bruit capteur) : | |
| # bonus de synergie quand la signature est FORTE -> score jusqu'à ~0.85, alors | |
| # que les JPEG réels (CV élevé) restent ~0.30 et les photos lisses réelles ~0.35. | |
| strong = float(np.clip((synth - 0.45) / 0.35, 0, 1)) | |
| score = float(np.clip(0.30 + 0.40 * synth + 0.20 * strong, 0.05, 0.85)) | |
| return {"ela_score": score, "ela_mean": mean_energy, | |
| "ela_cv": block_cv, "ela_ac": ac_energy} | |
| # ═══════════════════════════════════════════════════════════════════════════════ | |
| # 2. FFT — pics de grille λ/2 & λ/4 (diffusion/GAN) + pente spectrale | |
| # ═══════════════════════════════════════════════════════════════════════════════ | |
| class FFTAnalyzer: | |
| """ | |
| Profil spectral radial : pente 1/f, énergie HF, et surtout détection de | |
| PICS DE GRILLE par proéminence aux rayons Nyquist/2 et Nyquist/4 — signature | |
| du sur-échantillonnage des décodeurs GAN/diffusion. | |
| """ | |
| def analyze(self, rgb): | |
| try: | |
| return self._run(rgb) | |
| except Exception as e: | |
| _dbg(f"FFT exception: {e}") | |
| return {"fft_score": 0.5, "fft_grid2": 0.0, "fft_grid4": 0.0, "fft_slopedev": 0.0} | |
| def _run(self, rgb): | |
| gray = cv2.cvtColor(rgb, cv2.COLOR_RGB2GRAY).astype(np.float32) / 255.0 | |
| gray = gray - gray.mean() | |
| h, w = gray.shape | |
| F = np.abs(np.fft.fftshift(np.fft.fft2(gray))) | |
| cy, cx = h // 2, w // 2 | |
| max_r = min(cy, cx) | |
| yi, xi = np.ogrid[0:h, 0:w] | |
| radii = np.sqrt((yi - cy) ** 2 + (xi - cx) ** 2) | |
| # Profil radial moyen | |
| r_int = np.clip(radii.astype(int), 0, max_r - 1) | |
| sums = np.bincount(r_int.ravel(), weights=F.ravel(), minlength=max_r) | |
| cnts = np.bincount(r_int.ravel(), minlength=max_r).clip(1) | |
| prof = sums / cnts | |
| # Pente spectrale | |
| rs = np.arange(2, max_r); mask = prof[2:max_r] > 0 | |
| alpha = (-float(np.polyfit(np.log(rs[mask].astype(float)), | |
| np.log(prof[2:max_r][mask]), 1)[0]) | |
| if mask.sum() > 10 else 1.8) | |
| slopedev = abs(alpha - 1.8) / 0.8 | |
| # Proéminence d'un pic à un rayon donné vs voisinage (baseline médiane) | |
| def prominence(frac): | |
| idx = int(max_r * frac) | |
| if idx < 4 or idx >= max_r - 4: | |
| return 0.0 | |
| peak = float(prof[idx - 1:idx + 2].max()) | |
| neigh = np.concatenate([prof[max(0, idx - 8):idx - 2], prof[idx + 3:idx + 9]]) | |
| base = float(np.median(neigh)) if neigh.size else float(prof[idx]) | |
| return peak / (base + 1e-6) | |
| grid2 = prominence(0.5) # Nyquist/2 | |
| grid4 = prominence(0.25) # Nyquist/4 | |
| s_slope = float(np.clip(slopedev - 0.1, 0, 0.7)) | |
| s_g2 = float(np.clip((grid2 - 1.6) / 2.0, 0, 0.9)) if grid2 > 1.6 else 0.0 | |
| s_g4 = float(np.clip((grid4 - 1.6) / 2.0, 0, 0.9)) if grid4 > 1.6 else 0.0 | |
| grid = max(s_g2, s_g4) | |
| score = float(np.clip(0.55 * grid + 0.25 * s_slope + 0.20 * max(s_g2, s_g4), 0.02, 0.96)) | |
| return {"fft_score": score, "fft_grid2": grid2, "fft_grid4": grid4, "fft_slopedev": slopedev} | |
| # ═══════════════════════════════════════════════════════════════════════════════ | |
| # 3. SRM — banc de ~30 filtres passe-haut (Steganalysis Rich Model) | |
| # ═══════════════════════════════════════════════════════════════════════════════ | |
| def _build_srm_kernels(): | |
| K = [] | |
| # 1er ordre (4 directions) | |
| b1 = np.array([[0, 0, 0], [0, -1, 1], [0, 0, 0]], np.float32) | |
| K += [np.rot90(b1, k) for k in range(4)] | |
| # 1er ordre diagonal (4) | |
| b1d = np.array([[0, 0, 0], [0, -1, 0], [0, 0, 1]], np.float32) | |
| K += [np.rot90(b1d, k) for k in range(4)] | |
| # 2e ordre (2 axes) | |
| b2 = np.array([[0, 0, 0], [1, -2, 1], [0, 0, 0]], np.float32) | |
| K += [b2, np.rot90(b2, 1)] | |
| # 2e ordre diagonal (2) | |
| b2d = np.array([[1, 0, 0], [0, -2, 0], [0, 0, 1]], np.float32) | |
| K += [b2d, np.fliplr(b2d)] | |
| # 3e ordre (4 directions, 5×5) | |
| b3 = np.zeros((5, 5), np.float32); b3[2, 0:4] = [1, -3, 3, -1] | |
| K += [np.rot90(b3, k) for k in range(4)] | |
| # EDGE 3×3 (4 rotations) | |
| e3 = np.array([[-1, 2, -1], [2, -4, 2], [0, 0, 0]], np.float32) / 4.0 | |
| K += [np.rot90(e3, k) for k in range(4)] | |
| # SQUARE 3×3 | |
| K += [np.array([[-1, 2, -1], [2, -4, 2], [-1, 2, -1]], np.float32) / 4.0] | |
| # SQUARE 5×5 (KV) | |
| K += [np.array([[-1, 2, -2, 2, -1], [2, -6, 8, -6, 2], [-2, 8, -12, 8, -2], | |
| [2, -6, 8, -6, 2], [-1, 2, -2, 2, -1]], np.float32) / 12.0] | |
| # EDGE 5×5 (4 rotations) | |
| e5 = np.zeros((5, 5), np.float32) | |
| e5[0:3, 0:3] = np.array([[-1, 2, -2], [2, -6, 8], [-2, 8, -12]], np.float32) / 12.0 | |
| K += [np.rot90(e5, k) for k in range(4)] | |
| # 2e ordre tri-directionnel supplémentaire (3) → total ~30 | |
| b2b = np.array([[1, -2, 1], [0, 0, 0], [0, 0, 0]], np.float32) | |
| K += [np.rot90(b2b, k) for k in range(3)] | |
| return K | |
| class SRMAnalyzer: | |
| """Banc de ~30 filtres SRM. Le bruit capteur réel produit une énergie | |
| résiduelle élevée et cohérente sur l'ensemble du banc ; les images de | |
| diffusion ont un résidu faible et atypique.""" | |
| _KERNELS = _build_srm_kernels() | |
| def analyze(self, rgb): | |
| try: | |
| return self._run(rgb) | |
| except Exception as e: | |
| _dbg(f"SRM exception: {e}") | |
| return {"srm_score": 0.5, "srm_energy": 0.0, "srm_disp": 0.0, "srm_kurt": 0.0, "srm_n": 0} | |
| def _run(self, rgb): | |
| gray = cv2.cvtColor(rgb, cv2.COLOR_RGB2GRAY).astype(np.float32) / 255.0 | |
| stds, kurts = [], [] | |
| for k in self._KERNELS: | |
| r = cv2.filter2D(gray, cv2.CV_32F, k) | |
| s = float(np.std(r)) | |
| stds.append(s) | |
| if len(kurts) < 6: | |
| m2 = float(np.mean(r ** 2)); m4 = float(np.mean(r ** 4)) | |
| kurts.append(m4 / (m2 ** 2 + 1e-14)) | |
| stds = np.array(stds) | |
| energy = float(stds.mean()) # énergie résiduelle globale | |
| disp = float(stds.std() / (stds.mean() + 1e-9)) # dispersion inter-filtres | |
| kurt = float(np.mean(kurts)) if kurts else 0.0 | |
| # Énergie faible => synthétique (peu de bruit capteur) | |
| if energy < 0.004: s_e = 0.90 | |
| elif energy < 0.010: s_e = 0.90 - (energy - 0.004) / 0.006 * 0.34 | |
| elif energy < 0.022: s_e = 0.56 - (energy - 0.010) / 0.012 * 0.28 | |
| elif energy < 0.045: s_e = 0.28 - (energy - 0.022) / 0.023 * 0.16 | |
| else: s_e = max(0.05, 0.12 - (energy - 0.045) * 0.8) | |
| s_kurt = float(np.clip((kurt - 6.0) / 30.0, 0, 0.55)) | |
| score = float(np.clip(0.72 * s_e + 0.28 * s_kurt, 0.02, 0.95)) | |
| return {"srm_score": score, "srm_energy": energy, "srm_disp": disp, | |
| "srm_kurt": kurt, "srm_n": len(self._KERNELS)} | |
| # ═══════════════════════════════════════════════════════════════════════════════ | |
| # 3b. Cohérence chromatique — couleurs « trop parfaites » des images IA | |
| # ═══════════════════════════════════════════════════════════════════════════════ | |
| class ColorAnalyzer: | |
| """Cohérence chromatique. Les images IA ont souvent des couleurs « trop | |
| parfaites » : saturation très uniforme et transitions de couleur anormalement | |
| lisses (peu de bruit chromatique de capteur). On mesure l'uniformité de la | |
| saturation et des canaux RGB, et le bruit chromatique local (résidu passe-haut | |
| des canaux a*/b* Lab). Couleurs uniformes ET lisses => suspect.""" | |
| def analyze(self, rgb): | |
| try: | |
| return self._run(rgb) | |
| except Exception as e: | |
| _dbg(f"COLOR exception: {e}") | |
| return {"chroma_score": 0.40, "chroma_satstd": 0.0, | |
| "chroma_noise": 0.0, "chroma_satmean": 0.0} | |
| def _run(self, rgb): | |
| hsv = cv2.cvtColor(rgb, cv2.COLOR_RGB2HSV).astype(np.float32) | |
| sat = hsv[:, :, 1] / 255.0 | |
| sat_mean = float(sat.mean()) | |
| sat_std = float(sat.std()) | |
| # Variance (dispersion spatiale) moyenne des canaux RGB | |
| rgb_f = rgb.astype(np.float32) / 255.0 | |
| chan_std = float(rgb_f.reshape(-1, 3).std(axis=0).mean()) | |
| # Bruit chromatique local : résidu passe-haut des canaux a*/b* (Lab). | |
| # Réel = bruit chroma de capteur (résidu élevé) ; IA = couleurs lissées. | |
| lab = cv2.cvtColor(rgb, cv2.COLOR_RGB2LAB).astype(np.float32) | |
| ab = lab[:, :, 1:3] | |
| chroma_noise = float(np.abs(ab - cv2.GaussianBlur(ab, (0, 0), 1.5)).mean()) | |
| # Uniformité : saturation OU canaux RGB anormalement peu dispersés. | |
| flat_sat = float(np.clip((0.16 - sat_std) / 0.16, 0, 1)) if sat_std < 0.16 else 0.0 | |
| flat_rgb = float(np.clip((0.13 - chan_std) / 0.13, 0, 1)) if chan_std < 0.13 else 0.0 | |
| flat_col = max(flat_sat, flat_rgb) | |
| # Lissage chroma (peu de bruit capteur) — REQUIS pour ne pas pénaliser une | |
| # vraie photo peu colorée mais bruitée. | |
| smooth = float(np.clip((1.3 - chroma_noise) / 1.3, 0, 1)) if chroma_noise < 1.3 else 0.0 | |
| uniform = flat_col * smooth | |
| # Renfort « couleurs parfaites » : très saturé ET uniforme. | |
| oversat = float(np.clip((sat_mean - 0.45) / 0.40, 0, 1)) * flat_sat | |
| score = float(np.clip(0.30 + 0.46 * uniform + 0.10 * oversat, 0.05, 0.82)) | |
| return {"chroma_score": score, "chroma_satstd": sat_std, | |
| "chroma_noise": chroma_noise, "chroma_satmean": sat_mean} | |
| # ═══════════════════════════════════════════════════════════════════════════════ | |
| # 3c. Qualité image — bruit naturel (Laplacien) : les images IA sont « trop propres » | |
| # ═══════════════════════════════════════════════════════════════════════════════ | |
| class QualityAnalyzer: | |
| """Niveau de bruit naturel via Laplacien. Les images IA sont « trop nettes/ | |
| parfaites » : très peu de grain. On estime le PLANCHER de bruit par la médiane | |
| de |Laplacien| (robuste aux contours, contrairement à la variance) : un vrai | |
| capteur laisse du grain partout, l'IA non. img_noise bas => image trop propre.""" | |
| def analyze(self, rgb): | |
| try: | |
| return self._run(rgb) | |
| except Exception as e: | |
| _dbg(f"QUALITY exception: {e}") | |
| return {"img_noise": 5.0, "lap_var": 0.0, "too_sharp": False} | |
| def _run(self, rgb): | |
| gray = cv2.cvtColor(rgb, cv2.COLOR_RGB2GRAY).astype(np.float32) | |
| lap = cv2.Laplacian(gray, cv2.CV_32F, ksize=3) | |
| img_noise = float(np.median(np.abs(lap))) # plancher de bruit (robuste aux contours) | |
| lap_var = float(lap.var()) # netteté globale (info) | |
| too_sharp = bool(img_noise < 10.0) # seuil calibré : réel ~16 (grain) / IA Gemini ~8 (lisse) | |
| return {"img_noise": img_noise, "lap_var": lap_var, "too_sharp": too_sharp} | |
| # ═══════════════════════════════════════════════════════════════════════════════ | |
| # 4. EXIF forensique — signal FAIBLE (anti data-leakage) | |
| # ═══════════════════════════════════════════════════════════════════════════════ | |
| class EXIFForensicAnalyzer: | |
| _AI_SW = frozenset([ | |
| "midjourney", "stable diffusion", "dall-e", "dall·e", "runway", "flux", | |
| "firefly", "ideogram", "leonardo", "comfyui", "invoke", "automatic1111", | |
| "novelai", "diffusers", "adobe firefly", "adobe generative", | |
| ]) | |
| _PNG = b"\x89PNG" | |
| def analyze(self, pil_image, image_bytes): | |
| try: | |
| return self._run(pil_image, image_bytes) | |
| except Exception: | |
| return {"exif_score": 0.5, "exif_ai_tag": 0.0} | |
| def _run(self, pil_image, image_bytes): | |
| if image_bytes[:4] == self._PNG: | |
| raw = image_bytes[:20000].decode("latin-1", "replace").lower() | |
| if any(k in raw for k in ("stable diffusion", "comfyui", "cfg scale", | |
| "sampler", "model hash", "midjourney", "flux")): | |
| return {"exif_score": 0.85, "exif_ai_tag": 1.0} | |
| exif = {} | |
| try: | |
| exif = dict(pil_image.getexif() or {}) | |
| except Exception: | |
| pass | |
| for tag in (0x0131, 0x013B, 0x010E): | |
| if any(k in str(exif.get(tag, "")).lower() for k in self._AI_SW): | |
| return {"exif_score": 0.85, "exif_ai_tag": 1.0} | |
| camera = {0x010F, 0x0110, 0x829A, 0x829D, 0x8827, 0x9003, 0x8825} | |
| found = sum(1 for t in camera if t in exif) | |
| if found >= 3: | |
| return {"exif_score": 0.45, "exif_ai_tag": 0.0} | |
| if not exif: | |
| return {"exif_score": 0.55, "exif_ai_tag": 0.0} | |
| return {"exif_score": 0.50, "exif_ai_tag": 0.0} | |
| # ═══════════════════════════════════════════════════════════════════════════════ | |
| # 5. Détecteur global IA — Organika/sdxl-detector (ONNX) | |
| # ═══════════════════════════════════════════════════════════════════════════════ | |
| class SDXLDetectorAnalyzer: | |
| """ | |
| Détecteur d'images générées IA, généraliste (umm-maybe/AI-image-detector par | |
| défaut — SDXL, Midjourney, Gemini, SD…), exporté en ONNX (onnxruntime CPU). | |
| Le pré-traitement exact (taille, mean/std) ET les labels/index « fake » sont | |
| lus depuis le sidecar JSON ; l'index est auto-détecté depuis les labels du | |
| modèle (surchargeable via SDXL_FAKE_INDEX). | |
| """ | |
| _MODEL = os.path.join(_MODELS_DIR, "ai_detector.onnx") | |
| _SIDECAR = os.path.join(_MODELS_DIR, "ai_detector.json") | |
| _DEF_MEAN = np.array([0.485, 0.456, 0.406], dtype=np.float32) | |
| _DEF_STD = np.array([0.229, 0.224, 0.225], dtype=np.float32) | |
| def __init__(self): | |
| self._sess = None | |
| self._input = None | |
| self._tried = None | |
| self._lock = threading.Lock() | |
| self._size = 224 | |
| self._mean = self._DEF_MEAN | |
| self._std = self._DEF_STD | |
| self._labels = {} | |
| # SDXL_FAKE_INDEX (.env) PRIME sur le sidecar -> correctif manuel immédiat | |
| _env = os.getenv("SDXL_FAKE_INDEX") | |
| self._fake_idx_env = int(_env) if _env not in (None, "") else None | |
| self._fake_idx = self._fake_idx_env if self._fake_idx_env is not None else 0 | |
| def _model_path(self): | |
| p = os.getenv("SDXL_MODEL_PATH") | |
| if p and os.path.exists(p): | |
| return p | |
| return self._MODEL if os.path.exists(self._MODEL) else None | |
| def _load_sidecar(self, path): | |
| import json | |
| side = self._SIDECAR if os.path.exists(self._SIDECAR) else ( | |
| path[:-5] + ".json" if path.endswith(".onnx") else "") | |
| if side and os.path.exists(side): | |
| try: | |
| d = json.load(open(side, encoding="utf-8")) | |
| self._size = int(d.get("size", 224)) | |
| self._mean = np.array(d.get("mean", self._DEF_MEAN), dtype=np.float32) | |
| self._std = np.array(d.get("std", self._DEF_STD), dtype=np.float32) | |
| self._labels = d.get("labels", {}) or {} | |
| if self._fake_idx_env is None: # sinon le .env a déjà forcé l'index | |
| self._fake_idx = int(d.get("fake_index", self._fake_idx)) | |
| _dbg(f"SDXL sidecar: size={self._size} labels={self._labels} " | |
| f"fake_idx={self._fake_idx}" | |
| + (" (forcé par SDXL_FAKE_INDEX)" if self._fake_idx_env is not None else "")) | |
| except Exception as e: | |
| _dbg(f"SDXL sidecar illisible : {e}") | |
| def _ensure(self): | |
| if self._sess is not None: | |
| return | |
| with self._lock: | |
| if self._sess is not None: | |
| return | |
| path = self._model_path() | |
| if not path or path == self._tried: | |
| return | |
| self._tried = path | |
| try: | |
| import onnxruntime as ort | |
| self._load_sidecar(path) | |
| so = ort.SessionOptions() | |
| so.graph_optimization_level = ort.GraphOptimizationLevel.ORT_ENABLE_ALL | |
| so.intra_op_num_threads = 4 | |
| self._sess = ort.InferenceSession( | |
| path, sess_options=so, providers=["CPUExecutionProvider"]) | |
| self._input = self._sess.get_inputs()[0].name | |
| print(f"[SDXL] modèle ONNX chargé : {os.path.basename(path)}", flush=True) | |
| except Exception as e: | |
| print(f"[SDXL] chargement échoué : {e}", flush=True) | |
| self._sess = None | |
| def predict(self, rgb_full): | |
| self._ensure() | |
| if self._sess is None: | |
| return 0.5, False | |
| try: | |
| img = rgb_full | |
| if img is None or not isinstance(img, np.ndarray) or img.ndim < 2: | |
| return 0.5, False | |
| if img.dtype != np.uint8: | |
| img = np.clip(img, 0, 255).astype(np.uint8) | |
| if img.ndim == 2: | |
| img = cv2.cvtColor(img, cv2.COLOR_GRAY2RGB) | |
| elif img.shape[2] == 1: | |
| img = cv2.cvtColor(img[:, :, 0], cv2.COLOR_GRAY2RGB) | |
| elif img.shape[2] == 4: | |
| img = cv2.cvtColor(img, cv2.COLOR_RGBA2RGB) | |
| elif img.shape[2] != 3: | |
| img = np.ascontiguousarray(img[:, :, :3]) | |
| if img.shape[0] < 2 or img.shape[1] < 2: | |
| return 0.5, False | |
| s = max(32, int(self._size)) | |
| img = cv2.resize(img, (s, s), interpolation=cv2.INTER_AREA).astype(np.float32) / 255.0 | |
| img = (img - self._mean) / self._std | |
| x = np.ascontiguousarray(np.transpose(img, (2, 0, 1))[None, ...].astype(np.float32)) | |
| out = np.asarray(self._sess.run(None, {self._input: x})[0]).ravel() | |
| if out.size == 1: | |
| p = float(1.0 / (1.0 + np.exp(-out[0]))) | |
| probs = [1.0 - p, p] | |
| else: | |
| e = np.exp(out - out.max()); probs = (e / e.sum()).tolist() | |
| p = float(probs[min(self._fake_idx, out.size - 1)]) | |
| lbl = (self._labels.get(str(self._fake_idx)) | |
| or self._labels.get(self._fake_idx) or f"idx{self._fake_idx}") | |
| _dbg(f"SDXL labels={self._labels or 'n/a'} logits={np.round(out, 3).tolist()} " | |
| f"probs={np.round(probs, 3).tolist()} fake_idx={self._fake_idx}({lbl}) -> p={p:.3f}") | |
| return float(np.clip(p, 0.02, 0.98)), True | |
| except Exception as e: | |
| print(f"[SDXL] inférence échouée : {e}", flush=True) | |
| return 0.5, False | |
| # ═══════════════════════════════════════════════════════════════════════════════ | |
| # 6. ONNX visage — EfficientNet / FaceForensics++ | |
| # ═══════════════════════════════════════════════════════════════════════════════ | |
| class OnnxFaceAnalyzer: | |
| _MEAN = np.array([0.485, 0.456, 0.406], dtype=np.float32) | |
| _STD = np.array([0.229, 0.224, 0.225], dtype=np.float32) | |
| def __init__(self): | |
| self._sess = None | |
| self._input = None | |
| self._tried = None | |
| self._lock = threading.Lock() | |
| self._fake_idx = int(os.getenv("ONNX_FAKE_INDEX", "1")) | |
| def _local_path(self): | |
| p = os.getenv("ONNX_MODEL_PATH") | |
| if p and os.path.exists(p): | |
| return p | |
| default_local = os.path.join(_MODELS_DIR, "deepfake_efficientnet.onnx") | |
| return default_local if os.path.exists(default_local) else None | |
| def _maybe_int8(self, path): | |
| if os.getenv("ONNX_INT8", "1").lower() not in ("1", "true", "yes"): | |
| return path | |
| if ".int8." in path: | |
| return path | |
| q = path[:-5] + ".int8.onnx" if path.endswith(".onnx") else path + ".int8.onnx" | |
| if os.path.exists(q): | |
| return q | |
| try: | |
| from onnxruntime.quantization import quantize_dynamic, QuantType | |
| quantize_dynamic(path, q, weight_type=QuantType.QInt8) | |
| print(f"[FACE] quantification INT8 -> {os.path.basename(q)}", flush=True) | |
| return q | |
| except Exception as e: | |
| print(f"[FACE] INT8 indisponible ({e}) — modèle FP conservé", flush=True) | |
| return path | |
| def _ensure(self): | |
| if self._sess is not None: | |
| return | |
| with self._lock: | |
| if self._sess is not None: | |
| return | |
| path = self._local_path() | |
| if not path or path == self._tried: | |
| return | |
| self._tried = path | |
| try: | |
| import onnxruntime as ort | |
| path = self._maybe_int8(path) | |
| so = ort.SessionOptions() | |
| so.graph_optimization_level = ort.GraphOptimizationLevel.ORT_ENABLE_ALL | |
| so.intra_op_num_threads = 4 | |
| self._sess = ort.InferenceSession( | |
| path, sess_options=so, providers=["CPUExecutionProvider"]) | |
| self._input = self._sess.get_inputs()[0].name | |
| print(f"[FACE] modèle ONNX chargé : {os.path.basename(path)}", flush=True) | |
| except Exception as e: | |
| print(f"[FACE] chargement échoué : {e}", flush=True) | |
| self._sess = None | |
| def predict(self, rgb): | |
| self._ensure() | |
| if self._sess is None: | |
| return 0.5, False | |
| try: | |
| img = cv2.resize(rgb, (224, 224), interpolation=cv2.INTER_AREA).astype(np.float32) / 255.0 | |
| img = (img - self._MEAN) / self._STD | |
| x = np.ascontiguousarray(np.transpose(img, (2, 0, 1))[None, ...].astype(np.float32)) | |
| out = np.asarray(self._sess.run(None, {self._input: x})[0]).ravel() | |
| if out.size == 1: | |
| p = float(1.0 / (1.0 + np.exp(-out[0]))) | |
| else: | |
| e = np.exp(out - out.max()) | |
| p = float((e / e.sum())[min(self._fake_idx, out.size - 1)]) | |
| _dbg(f"FACE raw: out={np.round(out, 3).tolist()} -> p={p:.3f}") | |
| return float(np.clip(p, 0.02, 0.98)), True | |
| except Exception as e: | |
| print(f"[FACE] inférence échouée : {e}", flush=True) | |
| return 0.5, False | |
| # ═══════════════════════════════════════════════════════════════════════════════ | |
| # Localisation des visages — YuNet (DNN moderne, famille RetinaFace/SCRFD) | |
| # ═══════════════════════════════════════════════════════════════════════════════ | |
| class FaceLocator: | |
| """Détection de visages par YuNet (réseau one-stage type RetinaFace, intégré à | |
| OpenCV) — bien plus robuste que Haar sur les poses, éclairages et petits | |
| visages. Repli automatique sur Haar Cascade si le modèle ONNX est absent.""" | |
| def __init__(self): | |
| self._yunet = None | |
| # Repli Haar OPTIONNEL : l'API CascadeClassifier a été retirée d'OpenCV 5. | |
| self._haar = None | |
| try: | |
| if hasattr(cv2, "CascadeClassifier") and hasattr(cv2, "data"): | |
| self._haar = cv2.CascadeClassifier( | |
| cv2.data.haarcascades + "haarcascade_frontalface_default.xml") | |
| except Exception as e: | |
| print(f"[FACE-DET] Haar indisponible ({e})", flush=True) | |
| path = os.getenv("FACE_DETECTOR_PATH", | |
| os.path.join(_MODELS_DIR, "face_detector_yunet.onnx")) | |
| try: | |
| if os.path.exists(path) and hasattr(cv2, "FaceDetectorYN_create"): | |
| self._yunet = cv2.FaceDetectorYN_create(path, "", (320, 320), 0.6, 0.3, 5000) | |
| print(f"[FACE-DET] YuNet chargé : {os.path.basename(path)}", flush=True) | |
| else: | |
| print("[FACE-DET] YuNet indisponible -> repli Haar Cascade", flush=True) | |
| except Exception as e: | |
| self._yunet = None | |
| print(f"[FACE-DET] échec YuNet ({e}) -> repli Haar Cascade", flush=True) | |
| def detect(self, rgb, gray): | |
| """-> liste de (x, y, w, h).""" | |
| h, w = gray.shape[:2] | |
| if self._yunet is not None: | |
| try: | |
| self._yunet.setInputSize((w, h)) | |
| _, dets = self._yunet.detect(cv2.cvtColor(rgb, cv2.COLOR_RGB2BGR)) | |
| out = [] | |
| if dets is not None: | |
| for d in dets: | |
| x, y = max(0, int(d[0])), max(0, int(d[1])) | |
| fw, fh = min(int(d[2]), w - x), min(int(d[3]), h - y) | |
| if fw >= 24 and fh >= 24: | |
| out.append((x, y, fw, fh)) | |
| return out | |
| except Exception as e: | |
| _dbg(f"YuNet detect: {e} -> repli Haar") | |
| if self._haar is None: | |
| return [] | |
| r = self._haar.detectMultiScale(gray, 1.1, 4, minSize=(40, 40)) | |
| return [tuple(map(int, f)) for f in r] if len(r) > 0 else [] | |
| # ═══════════════════════════════════════════════════════════════════════════════ | |
| # Explications automatiques par analyseur (rapport) | |
| # ═══════════════════════════════════════════════════════════════════════════════ | |
| _EXPLAIN = { | |
| "sdxl": ("Détecteur IA", | |
| "Aucune signature d'image générée par IA détectée.", | |
| "Signature partielle d'image générée par IA.", | |
| "Forte signature d'image générée par IA (diffusion/GAN).", | |
| "Détecteur IA indisponible pour ce fichier."), | |
| "face": ("Analyse du visage", | |
| "Le visage ne présente pas d'artefacts de deepfake.", | |
| "Le visage présente des irrégularités à vérifier.", | |
| "Le visage présente des artefacts typiques de deepfake.", | |
| "Aucun visage détecté dans l'image."), | |
| "ela": ("Niveaux d'erreur (ELA)", | |
| "Erreur de recompression structurée, cohérente avec une photo réelle.", | |
| "Erreur de recompression inhabituellement uniforme.", | |
| "Carte d'erreur anormalement lisse, typique d'un rendu synthétique.", | |
| "Analyse non disponible."), | |
| "fft": ("Analyse spectrale (FFT)", | |
| "Aucun pic périodique suspect dans le spectre de Fourier.", | |
| "Pics périodiques modérés dans le spectre.", | |
| "Grille spectrale marquée, signature d'un décodeur génératif.", | |
| "Analyse non disponible."), | |
| "srm": ("Bruit résiduel (SRM)", | |
| "Bruit de capteur naturel présent (grain photo).", | |
| "Bruit de capteur plus faible qu'attendu.", | |
| "Bruit de capteur quasi absent, image probablement synthétique.", | |
| "Analyse non disponible."), | |
| "chroma": ("Cohérence chromatique", | |
| "Couleurs et saturation naturelles.", | |
| "Saturation inhabituellement uniforme.", | |
| "Couleurs « trop parfaites », typiques d'un rendu IA.", | |
| "Analyse non disponible."), | |
| "exif": ("Métadonnées (EXIF)", | |
| "Métadonnées cohérentes avec un appareil photo.", | |
| "Métadonnées absentes ou incomplètes.", | |
| "Trace d'un logiciel de génération d'images dans les métadonnées.", | |
| "Analyse non disponible."), | |
| } | |
| def _build_explanations(scores: dict) -> list: | |
| """Statut + phrase par analyseur : ok (<0.45), warn (0.45-0.65), alert (>0.65).""" | |
| out = [] | |
| for key, (label, t_ok, t_warn, t_alert, t_na) in _EXPLAIN.items(): | |
| s = scores.get(key) | |
| if s is None: | |
| out.append({"key": key, "label": label, "score": None, "status": "na", "text": t_na}) | |
| continue | |
| if s < 0.45: status, text = "ok", t_ok | |
| elif s < 0.65: status, text = "warn", t_warn | |
| else: status, text = "alert", t_alert | |
| out.append({"key": key, "label": label, "score": round(float(s), 3), | |
| "status": status, "text": text}) | |
| return out | |
| # ═══════════════════════════════════════════════════════════════════════════════ | |
| # Poids de fusion forensique — surchargeables via FUSION_WEIGHTS (JSON) | |
| # ═══════════════════════════════════════════════════════════════════════════════ | |
| # Cycle d'amélioration continue (Système Schrödinger) : les benchmarks anonymes | |
| # (/benchmarks/stats) révèlent quels analyseurs se trompent -> on ajuste ici les | |
| # poids SANS redéployer, ex. : FUSION_WEIGHTS='{"normal":{"fft":0.62},"boost":{"srm":0.70}}' | |
| _FUSION_DEFAULTS = { | |
| "normal": {"ela": 0.10, "fft": 0.58, "srm": 0.54, "chroma": 0.16}, | |
| "boost": {"ela": 0.08, "fft": 0.74, "srm": 0.66, "chroma": 0.18}, | |
| } | |
| def _load_fusion_weights(): | |
| import json | |
| w = {mode: dict(vals) for mode, vals in _FUSION_DEFAULTS.items()} | |
| raw = os.getenv("FUSION_WEIGHTS") | |
| if raw: | |
| try: | |
| user = json.loads(raw) | |
| for mode in w: | |
| for k, v in (user.get(mode) or {}).items(): | |
| if k in w[mode]: | |
| w[mode][k] = float(v) | |
| _dbg(f"FUSION_WEIGHTS surchargés : {w}") | |
| except Exception as e: | |
| _dbg(f"FUSION_WEIGHTS illisible ({e}) — défauts conservés") | |
| return w | |
| _FUSION_W = _load_fusion_weights() | |
| # ═══════════════════════════════════════════════════════════════════════════════ | |
| # Orchestrateur — fusion par vote pondéré (détecteur SDXL dominant) | |
| # ═══════════════════════════════════════════════════════════════════════════════ | |
| class DeepfakeDetector: | |
| def __init__(self): | |
| self.ela = ELAAnalyzer() | |
| self.fft = FFTAnalyzer() | |
| self.srm = SRMAnalyzer() | |
| self.color = ColorAnalyzer() | |
| self.quality = QualityAnalyzer() | |
| self.exif = EXIFForensicAnalyzer() | |
| self.sdxl = SDXLDetectorAnalyzer() | |
| self.face = OnnxFaceAnalyzer() | |
| self.facedet = FaceLocator() | |
| try: | |
| from model_bootstrap import start_background_bootstrap | |
| start_background_bootstrap(self) | |
| except Exception as e: | |
| print(f"[bootstrap] non démarré : {e}", flush=True) | |
| def _signal_score(f, forensic_boost=False): | |
| # FFT et SRM DOMINANTS : meilleurs détecteurs d'images de diffusion/GAN. | |
| # forensic_boost (zone SDXL incertaine) : on relève encore FFT/SRM pour trancher. | |
| # Cohérence chromatique en appui ; ELA volontairement sous-pondéré. | |
| # Poids surchargeables via FUSION_WEIGHTS (tuning piloté par les benchmarks). | |
| w = _FUSION_W["boost" if forensic_boost else "normal"] | |
| parts = [(f["ela_score"], w["ela"]), (f["fft_score"], w["fft"]), | |
| (f["srm_score"], w["srm"]), (f.get("chroma_score", 0.40), w["chroma"])] | |
| # EXIF inclus UNIQUEMENT s'il porte une information : | |
| # • tag logiciel IA détecté -> signal fort (exif_score ~0.85) | |
| # • métadonnées caméra (exif_score<0.50) -> léger signal « réel » | |
| # Un EXIF NEUTRE (~0.50/0.55, absence de métadonnées) est IGNORÉ : sinon le | |
| # 0.55 par défaut tire artificiellement la fusion vers le milieu. | |
| if f["exif_ai_tag"] > 0.5: | |
| parts.append((f["exif_score"], 0.20)) | |
| elif f["exif_score"] < 0.50: | |
| parts.append((f["exif_score"], 0.08)) | |
| tw = sum(w for _, w in parts) | |
| return sum(s * w for s, w in parts) / tw | |
| def analyze(self, image_bytes: bytes, video_frame: bool = False) -> dict: | |
| """video_frame=True (frames extraites d'une vidéo) : désactive les | |
| analyseurs NON PERTINENTS sur une frame ré-encodée en JPEG — | |
| • EfficientNet visage : biaisé (~0.55 sur vrais visages), il compressait | |
| tous les scores de frames vers 50 % ; le temporel est couvert par le | |
| ResNext50+LSTM dans la fusion vidéo ; | |
| • EXIF : une frame ré-encodée n'a jamais de métadonnées (0.55 constant, | |
| bruit pur) -> exclu et affiché « non disponible ». | |
| Économie CPU au passage (EfficientNet était le plus coûteux par frame).""" | |
| arr = np.frombuffer(image_bytes, np.uint8) | |
| bgr = cv2.imdecode(arr, cv2.IMREAD_COLOR) | |
| if bgr is None: | |
| raise ValueError("Impossible de décoder l'image.") | |
| rgb_full = cv2.cvtColor(bgr, cv2.COLOR_BGR2RGB) | |
| H, W = rgb_full.shape[:2] | |
| if H < 1 or W < 1: | |
| raise ValueError("Image décodée vide.") | |
| if max(H, W) > 640: | |
| sc = 640 / float(max(H, W)) | |
| rgb = cv2.resize(rgb_full, (max(1, int(round(W * sc))), max(1, int(round(H * sc)))), | |
| interpolation=cv2.INTER_AREA) | |
| else: | |
| rgb = rgb_full | |
| h, w = rgb.shape[:2] | |
| pil = Image.fromarray(rgb) | |
| gray = cv2.cvtColor(rgb, cv2.COLOR_RGB2GRAY) | |
| faces = self.facedet.detect(rgb, gray) | |
| face_crop = None | |
| if faces: | |
| fx, fy, fw, fh = max(faces, key=lambda f: f[2] * f[3]) | |
| px, py = int(fw * 0.20), int(fh * 0.20) | |
| x1, y1 = max(0, fx - px), max(0, fy - py) | |
| x2, y2 = min(w, fx + fw + px), min(h, fy + fh + py) | |
| if (x2 - x1) >= 32 and (y2 - y1) >= 32: | |
| face_crop = rgb[y1:y2, x1:x2] | |
| with ThreadPoolExecutor(max_workers=8) as ex: | |
| f_ela = ex.submit(self.ela.analyze, rgb) | |
| f_fft = ex.submit(self.fft.analyze, rgb) | |
| f_srm = ex.submit(self.srm.analyze, rgb) | |
| f_color = ex.submit(self.color.analyze, rgb) | |
| f_qual = ex.submit(self.quality.analyze, rgb) | |
| f_exif = None if video_frame else ex.submit(self.exif.analyze, pil, image_bytes) | |
| f_sdxl = ex.submit(self.sdxl.predict, rgb_full) | |
| f_face = (ex.submit(self.face.predict, face_crop) | |
| if (face_crop is not None and not video_frame) else None) | |
| feats = {} | |
| for fut in (f_ela, f_fft, f_srm, f_color, f_qual): | |
| feats.update(fut.result()) | |
| # EXIF neutre (exclu de la fusion) quand désactivé | |
| feats.update(f_exif.result() if f_exif is not None | |
| else {"exif_score": 0.55, "exif_ai_tag": 0.0}) | |
| sdxl_p, sdxl_ok = f_sdxl.result() | |
| face_p, face_ok = (f_face.result() if f_face is not None else (0.5, False)) | |
| # Zone SDXL incertaine [0.40, 0.60] -> on fait davantage confiance à FFT/SRM. | |
| uncertain = bool(sdxl_ok and 0.40 <= sdxl_p <= 0.60) | |
| signal = self._signal_score(feats, forensic_boost=uncertain) | |
| if sdxl_ok: | |
| w = float(os.getenv("SDXL_WEIGHT", "0.82")) # détecteur SDXL dominant | |
| # Garde-fou : un score saturé aux bornes (modèle/préproc inadapté) ne | |
| # doit pas aplatir toutes les images -> on réduit alors son poids. | |
| if sdxl_p <= 0.03 or sdxl_p >= 0.97: | |
| w = min(w, 0.15) | |
| _dbg(f"SDXL saturé (p={sdxl_p:.3f}) -> poids réduit à {w}") | |
| elif uncertain: | |
| w = min(w, 0.30) | |
| _dbg(f"SDXL incertain (p={sdxl_p:.3f}) -> poids réduit à {w} (FFT/SRM tranchent)") | |
| global_score = w * sdxl_p + (1 - w) * signal | |
| else: | |
| global_score = signal | |
| # ── Poids Face ADAPTATIF ──────────────────────────────────────────── | |
| # Le modèle EfficientNet visage est biaisé (~0.60 sur des visages RÉELS | |
| # nets) : on ne lui accorde le poids FORT (0.85) que s'il est franchement | |
| # affirmatif (face_p > 0.72). En dessous — y compris la zone bruitée | |
| # 0.65–0.72 — poids prudent 0.50, pour ne pas gonfler un vrai visage net. | |
| # face_prudent (visage ambigu ET SDXL très bas) neutralise en plus les | |
| # planchers visage plus bas. | |
| face_prudent = bool(face_ok and 0.50 <= face_p <= 0.70 and sdxl_ok and sdxl_p < 0.20) | |
| fw = 0.85 if (face_ok and face_p > 0.72) else 0.50 | |
| if face_crop is not None and face_ok: | |
| final = fw * face_p + (1 - fw) * global_score | |
| _dbg(f"poids Face adaptatif fw={fw:.2f} (prudent={face_prudent}, face={face_p:.2f}, sdxl={sdxl_p:.2f})") | |
| else: | |
| final = global_score | |
| # Bonus de convergence : si ELA, FFT et SRM dépassent TOUS 0.35, le faisceau | |
| # forensique concorde -> +0.08 au score final. | |
| if feats["ela_score"] > 0.35 and feats["fft_score"] > 0.35 and feats["srm_score"] > 0.35: | |
| final += 0.08 | |
| _dbg("bonus convergence ELA+FFT+SRM > 0.35 : +0.08") | |
| # (Bonus « image trop nette + visage -> +0.10 » SUPPRIMÉ : les vrais portraits | |
| # nets le déclenchaient systématiquement -> faux positifs.) | |
| final = float(np.clip(final, 0.01, 0.99)) | |
| # ── Recalibrage : planchers de score (appliqués APRÈS le calcul final) ── | |
| # Consensus : plus d'analyseurs dépassent 45 %, plus le plancher est haut | |
| # (1 -> 40 %, 2 -> 65 %, 3+ -> 80 %). | |
| # Plancher visage : Face > 72 % -> >= 65 %. Seuil RELEVÉ à 0.72 : | |
| # le modèle EfficientNet renvoie ~60 % sur des visages RÉELS nets (biais | |
| # constaté) et la bande 0.65-0.72 reste du bruit ; la zone 50-72 % est | |
| # donc ignorée — le visage ne compte (plancher ET consensus) que > 0.72. | |
| # Plancher SDXL : SDXL > 60 % -> >= 65 %. | |
| # NB : le comptage porte sur les scores BRUTS (indépendant des poids de fusion). | |
| ela_o = feats["ela_score"] > 0.45 | |
| fft_o = feats["fft_score"] > 0.45 | |
| srm_o = feats["srm_score"] > 0.45 | |
| sdxl_o = bool(sdxl_ok and sdxl_p > 0.45) | |
| face_o = bool(face_ok and face_p > 0.72 and not face_prudent) # vote consensus visage (biais ~0.60, bruit <= 0.72) | |
| over50 = sum((ela_o, fft_o, srm_o, sdxl_o, face_o)) | |
| floor = 0.80 if over50 >= 3 else 0.65 if over50 == 2 else 0.40 if over50 == 1 else 0.0 | |
| # Plancher visage : UNIQUEMENT hors zone prudente et au-delà du bruit (0.72). | |
| if face_ok and not face_prudent and face_p > 0.72: | |
| floor = max(floor, 0.65) # plancher visage (seuil relevé à 0.72) | |
| if sdxl_ok and sdxl_p > 0.60: | |
| floor = max(floor, 0.65) # plancher détecteur IA global | |
| _dbg("over50 check (seuil 0.45, visage 0.72): " | |
| f"ela={feats['ela_score']:.3f}[{'+' if ela_o else '-'}] " | |
| f"fft={feats['fft_score']:.3f}[{'+' if fft_o else '-'}] " | |
| f"srm={feats['srm_score']:.3f}[{'+' if srm_o else '-'}] " | |
| f"sdxl={('%.3f' % sdxl_p) if sdxl_ok else 'N/A'}[{'+' if sdxl_o else '-'}] " | |
| f"face={('%.3f' % face_p) if face_ok else 'N/A'}[{'+' if face_o else '-'}] " | |
| f"-> over50={over50}, floor={floor:.2f}, final_avant={final:.3f}") | |
| if final < floor: | |
| _dbg(f"plancher applique : {final:.3f} -> {floor:.2f}") | |
| final = floor | |
| votes = sum(1 for s in (feats["ela_score"], feats["fft_score"], feats["srm_score"]) | |
| if s > _VOTE_THRESHOLD) | |
| if sdxl_ok and sdxl_p > _VOTE_THRESHOLD: | |
| votes += 1 | |
| if face_ok and face_p > _FACE_VOTE_THRESHOLD: | |
| votes += 1 | |
| # ── Journalisation des VOTES (anomalie dès _VOTE_THRESHOLD) ───────── | |
| _dbg(f"votes (seuil {_VOTE_THRESHOLD:.2f}): " | |
| f"ela={feats['ela_score']:.3f}[{'+' if feats['ela_score'] > _VOTE_THRESHOLD else '-'}] " | |
| f"fft={feats['fft_score']:.3f}[{'+' if feats['fft_score'] > _VOTE_THRESHOLD else '-'}] " | |
| f"srm={feats['srm_score']:.3f}[{'+' if feats['srm_score'] > _VOTE_THRESHOLD else '-'}] " | |
| f"sdxl={('%.3f' % sdxl_p) if sdxl_ok else 'N/A'}[{'+' if (sdxl_ok and sdxl_p > _VOTE_THRESHOLD) else '-'}] " | |
| f"face={('%.3f' % face_p) if face_ok else 'N/A'}[{'+' if (face_ok and face_p > _FACE_VOTE_THRESHOLD) else '-'}] " | |
| f"-> votes={votes}/5") | |
| # ── Journalisation des scores BRUTS (débogage) ────────────────────── | |
| _dbg(f"sdxl={'%.3f' % sdxl_p if sdxl_ok else 'N/A'} " | |
| f"face={'%.3f' % face_p if face_ok else 'N/A'} " | |
| f"ela={feats['ela_score']:.3f}(mean={feats.get('ela_mean',0):.2f},ac={feats.get('ela_ac',0):.2f}) " | |
| f"fft={feats['fft_score']:.3f}(g2={feats.get('fft_grid2',0):.2f},g4={feats.get('fft_grid4',0):.2f}) " | |
| f"srm={feats['srm_score']:.3f}(E={feats.get('srm_energy',0):.4f},n={feats.get('srm_n',0)}) " | |
| f"chroma={feats.get('chroma_score',0):.3f}(satstd={feats.get('chroma_satstd',0):.3f},noise={feats.get('chroma_noise',0):.2f}) " | |
| f"qual(noise={feats.get('img_noise',0):.2f},sharp={feats.get('too_sharp',False)}) " | |
| f"exif={feats['exif_score']:.2f} | signal={signal:.3f} global={global_score:.3f} " | |
| f"FINAL={final:.3f} face_det={len(faces)>0}") | |
| scores_out = { | |
| "sdxl": round(sdxl_p, 3) if sdxl_ok else None, | |
| "face": round(face_p, 3) if face_ok else None, | |
| "ela": round(feats["ela_score"], 3), | |
| "fft": round(feats["fft_score"], 3), | |
| "srm": round(feats["srm_score"], 3), | |
| "exif": None if video_frame else round(feats["exif_score"], 3), | |
| "chroma": round(feats.get("chroma_score", 0.40), 3), | |
| } | |
| return { | |
| "confidence_score": round(final, 4), | |
| "is_deepfake": final > 0.50, | |
| "face_detected": len(faces) > 0, | |
| "face_count": int(len(faces)), | |
| "image_size": {"width": W, "height": H}, | |
| "votes": votes, | |
| "fusion": "vote_pondere", | |
| "scores": scores_out, | |
| "explanations": _build_explanations(scores_out), | |
| } | |