import os import sys import time import base64 import tempfile import statistics import httpx import cv2 from dotenv import load_dotenv # Logs robustes sur consoles non-UTF-8 (Windows cp1252, locale C…) for _stream in (sys.stdout, sys.stderr): try: _stream.reconfigure(encoding="utf-8", errors="replace") except Exception: pass load_dotenv() from fastapi import FastAPI, File, UploadFile, HTTPException from fastapi.middleware.cors import CORSMiddleware from fastapi.exceptions import RequestValidationError from fastapi.responses import JSONResponse from starlette.exceptions import HTTPException as StarletteHTTPException from starlette.requests import Request from detector import DeepfakeDetector from video_model import VideoDeepfakeAnalyzer import benchmarks from benchmarks import MODEL_VERSION, SYSTEM_NAME app = FastAPI(title="Unkor — Detection Engine (Système Schrödinger)") # Origines de confiance — référence documentaire (et repli si l'API repasse un # jour en mode restreint) : le middleware ci-dessous autorise TOUTES les origines. CORS_ORIGINS = [ "http://localhost:5173", "http://localhost:5174", "http://127.0.0.1:5173", "http://127.0.0.1:5174", "https://unkor.fr", "https://www.unkor.fr", ] # API publique BÊTA : ouverte à toutes les origines, sans clé d'API. # Pas de cookies ni d'authentification -> allow_credentials=False, obligatoire # avec "*" (la spec CORS interdit credentials + wildcard). app.add_middleware( CORSMiddleware, allow_origins=["*"], allow_credentials=False, allow_methods=["*"], allow_headers=["*"], expose_headers=["*"], ) def _cors(request: Request) -> dict: """En-têtes CORS des réponses d'erreur (défensif — l'API est ouverte à tous).""" return {"Access-Control-Allow-Origin": "*"} if request.headers.get("origin") else {} @app.exception_handler(StarletteHTTPException) async def http_exception_handler(request: Request, exc: StarletteHTTPException): return JSONResponse( status_code=exc.status_code, content={"detail": exc.detail}, headers=_cors(request), ) @app.exception_handler(RequestValidationError) async def validation_exception_handler(request: Request, exc: RequestValidationError): return JSONResponse( status_code=422, content={"detail": exc.errors()}, headers=_cors(request), ) detector = DeepfakeDetector() video_net = VideoDeepfakeAnalyzer() ALLOWED_TYPES = {"image/jpeg", "image/png", "image/webp"} MAX_SIZE_MB = 10 @app.get("/") def root(): return {"status": "ok", "service": "Unkor Detection Engine", "system": SYSTEM_NAME, "model_version": MODEL_VERSION} @app.get("/benchmarks/stats") async def benchmarks_stats(): """Statistiques agrégées des analyses (données 100 % anonymes).""" from fastapi.concurrency import run_in_threadpool return JSONResponse(await run_in_threadpool(benchmarks.stats)) @app.post("/analyze") async def analyze_image(file: UploadFile = File(...)): if file.content_type not in ALLOWED_TYPES: raise HTTPException(400, f"Format non supporté : {file.content_type}") contents = await file.read() if len(contents) > MAX_SIZE_MB * 1024 * 1024: raise HTTPException(400, f"Fichier trop volumineux (max {MAX_SIZE_MB} Mo)") t0 = time.perf_counter() try: result = detector.analyze(contents) except Exception as e: raise HTTPException(500, f"Erreur d'analyse : {e}") duration_ms = (time.perf_counter() - t0) * 1000.0 size = result.get("image_size") or {} benchmarks.record( "image", result["confidence_score"], result.get("scores"), duration_ms, f"{size.get('width', 0)}x{size.get('height', 0)}") return JSONResponse({ **result, "filename": file.filename, "model_version": MODEL_VERSION, "system": SYSTEM_NAME, }) @app.post("/new-user") async def new_user(nickname: str = ""): name = (nickname or "").strip()[:40] or "anonyme" try: async with httpx.AsyncClient(timeout=5) as client: await client.post( "https://ntfy.sh/unkor-notifs", content=f"Nouvel utilisateur : {name}".encode("utf-8"), headers={"Title": "Unkor"}, ) except Exception as e: print(f"[ntfy] erreur : {e}", flush=True) return {"ok": True} # ── Détection vidéo — Système Schrödinger ──────────────────────────────────── # Deux passes complémentaires, puis fusion : # 1. SPÉCIALISTE VIDÉO : ResNext50+LSTM (FaceForensics-style) sur une séquence # de VIDEO_SEQ_LEN crops de visage équi-espacés -> cohérence TEMPORELLE # apprise (le modèle voit le "clignotement" des deepfakes). # 2. FORENSIQUE IMAGE : le pipeline 7 analyseurs sur VIDEO_PIPE_FRAMES frames # -> attrape les vidéos 100 % synthétiques (Runway/Sora) sans visage, # fournit vignettes, explications et écart-type temporel. MAX_VIDEO_MB = 50 VIDEO_SEQ_LEN = max(4, int(os.getenv("VIDEO_SEQ_LEN", "20"))) VIDEO_PIPE_FRAMES = max(3, int(os.getenv("VIDEO_PIPE_FRAMES", "12"))) VIDEO_EXT = {".mp4", ".mov", ".avi", ".mkv", ".webm"} def _thumb_b64(frame, max_w=360): h, w = frame.shape[:2] if w > max_w: frame = cv2.resize(frame, (max_w, int(h * max_w / w)), interpolation=cv2.INTER_AREA) ok, buf = cv2.imencode(".jpg", frame, [int(cv2.IMWRITE_JPEG_QUALITY), 80]) return "data:image/jpeg;base64," + base64.b64encode(buf.tobytes()).decode("ascii") def _even_timestamps(duration, n): if duration > 0: secs = int(duration) + 1 if secs <= n: return [float(t) for t in range(secs)] return [duration * (i + 0.5) / n for i in range(n)] return [float(t) for t in range(n)] def _face_box(rgb): """Boîte SERRÉE du plus grand visage (YuNet, sans marge — fidèle aux crops dlib du régime d'entraînement du checkpoint). -> (x1, y1, x2, y2) ou None.""" try: gray = cv2.cvtColor(rgb, cv2.COLOR_RGB2GRAY) faces = detector.facedet.detect(rgb, gray) if not faces: return None fx, fy, fw, fh = max(faces, key=lambda f: f[2] * f[3]) h, w = rgb.shape[:2] x1, y1 = max(0, fx), max(0, fy) x2, y2 = min(w, fx + fw), min(h, fy + fh) if (x2 - x1) >= 24 and (y2 - y1) >= 24: return (x1, y1, x2, y2) return None except Exception: return None def _schrodinger_fuse(video_ok, video_p, face_seq_ratio, boosted_avg, boosted_max): """Fusion Schrödinger V2 — pilotée par la CONFIANCE du spécialiste vidéo. -> (vw, final_avg, final_max). Historique : la V1 bridait le ResNext50+LSTM (gates + plafonds de poids) parce qu'il était servi hors distribution (canaux RGB au lieu de BGR, frames équi-espacées au lieu de consécutives) et produisait des hyper-confiances aberrantes. Depuis la correction du contrat d'entrée, il discrimine fortement (mesuré : ~3 % sur vidéo réelle, ~98 % sur vidéo Runway) — les anciens garde-fous plafonnaient ses VRAIS positifs à ~60 % et sont donc supprimés. Principe V2 : le poids du modèle vidéo suit sa confiance, SYMÉTRIQUEMENT (il tire vers le réel comme vers le fake) : • confiance c = |video_p − 0.5| × 2 (0 = indécis, 1 = certain) • vw = 0.20 + (vw_max − 0.20) × c, vw_max = 0.85 (0.60 si < 30 % de visages, hors distribution FaceForensics++) • accord des deux voies (≥ 0.55) -> bonus +0.08 • video_p ≥ 0.90 avec visages -> plancher 0.75 (objectif : vidéo IA > 70 %) • forensique ≥ 0.70 -> plancher 0.60 (deepfake qui tromperait le ResNext) • video_p ≤ 0.10 sans alarme forensique (< 0.60) -> plafond 0.30 (objectif : vraie vidéo < 30 %)""" if not video_ok: return 0.0, boosted_avg, boosted_max conf = abs(video_p - 0.5) * 2.0 vw_max = 0.85 if face_seq_ratio >= 0.3 else 0.60 vw = 0.20 + (vw_max - 0.20) * conf final_avg = vw * video_p + (1 - vw) * boosted_avg final_max = vw * video_p + (1 - vw) * boosted_max if video_p >= 0.55 and boosted_max >= 0.55: final_max = min(1.0, final_max + 0.08) # bonus d'accord des deux voies if video_p >= 0.90 and face_seq_ratio >= 0.3: final_max = max(final_max, 0.75) # spécialiste vidéo quasi certain if boosted_max >= 0.70: final_max = max(final_max, 0.60) # alarme forensique indépendante if video_p <= 0.10 and boosted_max < 0.60: final_max = min(final_max, 0.30) # spécialiste quasi certain du RÉEL final_avg = min(final_avg, final_max) return vw, final_avg, final_max def _grab_frames(cap, timestamps, seekable): """-> liste de (timestamp, frame BGR).""" out = [] for ts in timestamps: if seekable: cap.set(cv2.CAP_PROP_POS_MSEC, ts * 1000.0) ok_read, frame = cap.read() if not ok_read: if seekable: continue break out.append((float(ts), frame)) return out def _process_video(contents: bytes, filename: str) -> dict: suffix = os.path.splitext(filename or "")[1].lower() if suffix not in VIDEO_EXT: suffix = ".mp4" tmp = tempfile.NamedTemporaryFile(delete=False, suffix=suffix) try: tmp.write(contents) tmp.close() cap = cv2.VideoCapture(tmp.name) if not cap.isOpened(): return {"error": "Vidéo illisible."} fps = cap.get(cv2.CAP_PROP_FPS) or 0.0 if fps <= 0: fps = 25.0 total = int(cap.get(cv2.CAP_PROP_FRAME_COUNT) or 0) duration = total / fps if total > 0 else 0.0 seekable = duration > 0 # ── Passe 1 : pipeline forensique image (frames pleines) ──────────── results = [] resolution = None for ts, frame in _grab_frames(cap, _even_timestamps(duration, VIDEO_PIPE_FRAMES), seekable): if resolution is None: resolution = f"{frame.shape[1]}x{frame.shape[0]}" ok_enc, buf = cv2.imencode(".jpg", frame, [int(cv2.IMWRITE_JPEG_QUALITY), 90]) if not ok_enc: continue try: # Mode frame vidéo : EfficientNet visage (biaisé ~0.55) et EXIF # (toujours vide sur une frame ré-encodée) sont désactivés — ils # compressaient tous les scores de frames vers ~50 %. res = detector.analyze(buf.tobytes(), video_frame=True) results.append((float(res["confidence_score"]), ts, frame, res)) except Exception: continue # ── Passe 2 : séquence pour ResNext50+LSTM — fidèle à l'ENTRAÎNEMENT ─ # Le checkpoint a appris sur des frames CONSÉCUTIVES (~30 fps), en crops # de visage SERRÉS, et en ordre de canaux BGR (lecture OpenCV sans # conversion dans le projet d'origine). On reproduit exactement ce # régime : départ à ~20 % de la vidéo, lecture séquentielle, on garde # les crops de visage ; repli plein cadre si la vidéo n'a pas de visage. seq_crops, seq_faces = [], 0 start_ts = duration * 0.20 if duration > 4 else 0.0 if seekable: cap.set(cv2.CAP_PROP_POS_MSEC, start_ts * 1000.0) reads, max_reads = 0, VIDEO_SEQ_LEN * 3 no_face_frames = [] while len(seq_crops) < VIDEO_SEQ_LEN and reads < max_reads: ok_read, frame = cap.read() if not ok_read: break reads += 1 rgb = cv2.cvtColor(frame, cv2.COLOR_BGR2RGB) box = _face_box(rgb) if box is not None: x1, y1, x2, y2 = box seq_crops.append(frame[y1:y2, x1:x2]) # crop BGR serré seq_faces += 1 elif len(no_face_frames) < VIDEO_SEQ_LEN: no_face_frames.append(frame) # candidates plein cadre (BGR) if len(seq_crops) < 4 and no_face_frames: # Vidéo (quasi) sans visage : séquence plein cadre consécutive. seq_crops = no_face_frames[:VIDEO_SEQ_LEN] seq_faces = 0 cap.release() finally: try: os.unlink(tmp.name) except Exception: pass if not results: return {"error": "Aucune frame analysable dans la vidéo."} video_p, video_ok = video_net.predict(seq_crops) face_seq_ratio = seq_faces / len(seq_crops) if seq_crops else 0.0 # ── Forensique image : agrégats + cohérence temporelle ────────────────── scores = [r[0] for r in results] top = sorted(results, key=lambda r: r[0], reverse=True)[:3] t_std = float(statistics.pstdev(scores)) if len(scores) > 1 else 0.0 temporal_bonus = min(t_std * 2.0, 0.25) # std élevée = deepfake qui "clignote" boosted_avg = min(sum(scores) / len(scores) + temporal_bonus, 1.0) boosted_max = min(max(scores) + temporal_bonus, 1.0) # Part des frames où le modèle visage a signalé (>0.5). Conservé pour la # compatibilité API/benchmarks — vaut 0.0 depuis que EfficientNet est # désactivé sur les frames vidéo (video_frame=True). face_flags = sum(1 for *_x, r in results if isinstance(r["scores"].get("face"), (int, float)) and r["scores"]["face"] > 0.5) face_flag_ratio = face_flags / len(results) # ── Fusion Schrödinger V2 (détails dans _schrodinger_fuse) ─────────────── vw, final_avg, final_max = _schrodinger_fuse( video_ok, video_p, face_seq_ratio, boosted_avg, boosted_max) final_max = min(final_max, 1.0) final_avg = min(final_avg, final_max) print(f"[schrodinger] video_p={('%.3f' % video_p) if video_ok else 'N/A'} vw={vw:.2f} " f"faces_seq={face_seq_ratio:.2f} pipe_avg={boosted_avg:.3f} pipe_max={boosted_max:.3f} " f"t_std={t_std:.3f} face_flags={face_flag_ratio:.2f} " f"-> avg={final_avg:.3f} max={final_max:.3f}", flush=True) # Moyennes par analyseur sur les frames (pour les benchmarks anonymes) acc = {} for *_x, r in results: for k, v in (r.get("scores") or {}).items(): if isinstance(v, (int, float)): a = acc.setdefault(k, [0.0, 0]) a[0] += float(v) a[1] += 1 bench_scores = {k: s / n for k, (s, n) in acc.items() if n > 0} if video_ok: bench_scores["video_model"] = video_p # Explications : celles de la frame la plus suspecte, SANS visage/EXIF # (désactivés par design sur les frames vidéo), avec le spécialiste # ResNext50+LSTM en tête de liste. expl = [e for e in (top[0][3].get("explanations") or []) if e.get("key") not in ("face", "exif")] if video_ok: if video_p < 0.45: v_st, v_txt = "ok", "Cohérence temporelle naturelle — aucune signature de deepfake vidéo." elif video_p < 0.65: v_st, v_txt = "warn", "Dynamique temporelle partiellement atypique, à vérifier." else: v_st, v_txt = "alert", "Signature de deepfake vidéo détectée (incohérences temporelles)." expl.insert(0, {"key": "video_model", "label": "Modèle vidéo (ResNext50+LSTM)", "score": round(float(video_p), 3), "status": v_st, "text": v_txt}) return { "frames_analyzed": len(results), "avg_score": round(final_avg, 4), "max_score": round(final_max, 4), "temporal_std": round(t_std, 4), "is_deepfake": final_max >= 0.55, "video_model_score": round(video_p, 4) if video_ok else None, "video_model_frames": len(seq_crops) if video_ok else 0, "face_flag_ratio": round(face_flag_ratio, 3), "suspicious_frames": [ {"timestamp": round(ts, 1), "score": round(sc, 4), "image": _thumb_b64(fr)} for sc, ts, fr, _res in top ], "explanations": expl, "_bench_scores": bench_scores, "_resolution": resolution, } @app.post("/analyze-video") async def analyze_video(file: UploadFile = File(...)): from fastapi.concurrency import run_in_threadpool name = (file.filename or "").lower() ctype = file.content_type or "" if not any(name.endswith(e) for e in VIDEO_EXT) and not ctype.startswith("video/"): raise HTTPException(400, f"Format vidéo non supporté : {file.content_type}") contents = await file.read() if len(contents) > MAX_VIDEO_MB * 1024 * 1024: raise HTTPException(400, f"Vidéo trop volumineuse (max {MAX_VIDEO_MB} Mo)") t0 = time.perf_counter() result = await run_in_threadpool(_process_video, contents, file.filename) if "error" in result: raise HTTPException(400, result["error"]) duration_ms = (time.perf_counter() - t0) * 1000.0 bench_scores = result.pop("_bench_scores", None) resolution = result.pop("_resolution", None) benchmarks.record("video", result["max_score"], bench_scores, duration_ms, resolution) return JSONResponse({ **result, "filename": file.filename, "model_version": MODEL_VERSION, "system": SYSTEM_NAME, })