Spaces:
Running
Running
| """A-EYE backend — HYBRID entrypoint (real-use tuned). | |
| Real-world feedback: model 49 discriminates better than the PatchGuard image | |
| verdict, but its heatmap (old model 24) was weak. So: | |
| - VERDICT = model 49 (the proven champion discriminator) | |
| - + composite rescue: if 49 says REAL but PatchGuard finds a CONCENTRATED | |
| high-confidence AI region (real photo with an inserted/AI object), escalate | |
| to AI — the case 49 structurally misses. | |
| - HEATMAP = dense PatchGuard (model 63) map, rendered vivid, only when AI. | |
| Same /api/analyze contract as the existing app (schemas.AnalyzeResponse), so the | |
| Expo app needs no changes. New file; existing backend files are untouched. | |
| Run (from a-eye/backend): | |
| venv\\Scripts\\python -m uvicorn main_hybrid:app --host 0.0.0.0 --port 8000 | |
| """ | |
| from __future__ import annotations | |
| import base64 | |
| import io | |
| import os | |
| import sys | |
| import time | |
| from contextlib import asynccontextmanager | |
| from typing import Any, AsyncIterator | |
| import numpy as np | |
| import torch | |
| import torchvision.transforms as T | |
| from PIL import Image, ImageOps | |
| from scipy.ndimage import gaussian_filter | |
| from fastapi import FastAPI, File, HTTPException, UploadFile | |
| from fastapi.concurrency import run_in_threadpool | |
| from fastapi.middleware.cors import CORSMiddleware | |
| from fastapi.responses import HTMLResponse | |
| from fastapi.staticfiles import StaticFiles | |
| import provenance_gate | |
| import synthid_gate | |
| from heatmap_nextgen import overlay as render_overlay, pure_heatmap, residual_var | |
| from patchguard import PATCHGUARD_ARCH, PatchGuardDetector | |
| from schemas import AnalyzeResponse, HealthResponse, VersionResponse | |
| from zero_shot_v4 import CLIP_MEAN, CLIP_STD, ZeroShotV4Detector | |
| # Model 63 = dense-decoder architecture from the detector repo. Reuse that exact | |
| # (trained) code so loading can never drift from how the checkpoint was produced. | |
| _DETECTOR_ROOT = os.getenv("AEYE_DETECTOR_ROOT", r"C:\Users\user\Desktop\a-eye(vit_clip_LoRA)") | |
| if os.path.isdir(_DETECTOR_ROOT) and _DETECTOR_ROOT not in sys.path: | |
| sys.path.insert(0, _DETECTOR_ROOT) | |
| from aeye_next.models.dense_patchguard import DensePatchGuardDetector | |
| MAX_IMAGE_BYTES = 28 * 1024 * 1024 | |
| DEVICE = os.environ.get("AEYE_DEVICE", "cpu") | |
| INPUT_SIZE = 224 | |
| VERDICT_PATH = os.getenv("VERDICT_MODEL_PATH", "./models/best49.pt") | |
| HEATMAP_PATH = os.getenv("HEATMAP_MODEL_PATH", "./models/best63.pt") | |
| HEATMAP_KIND = os.getenv("HEATMAP_MODEL_KIND", "dense").lower() | |
| # model 49 architecture (matches the app's model_loader.MODEL_49) | |
| MODEL_49_ARCH = dict( | |
| clip_backbone="clip-vit-l-14", clip_layer=13, semantic_dim=512, forensic_dim=256, | |
| frequency_dim=192, fft_bins=48, image_size=224, num_classes=2, num_sources=20, | |
| dropout=0.26, source_grl_lambda=0.0, freeze_clip=True, | |
| ) | |
| # model 63 = dense localization decoder (replaces 61 as heatmap + rescue). Native | |
| # 32x32 dense patch map -> sharper localization; image head powers the rescue gate. | |
| DENSE_ARCH = dict( | |
| clip_backbone="clip-vit-l-14", clip_layer=13, semantic_dim=512, forensic_dim=256, | |
| frequency_dim=192, fft_bins=48, num_classes=2, num_sources=2, | |
| dropout=0.26, source_grl_lambda=0.0, freeze_clip=True, patch_hidden=256, | |
| patch_topk_frac=0.06, decoder_grid=32, decoder_channels=192, decoder_blocks=3, | |
| decoder_dropout=0.04, | |
| ) | |
| # model 53 = SigLIP-large/256 verdict head. DIFFERENT backbone than 49/63 (which | |
| # share CLIP-L), so it makes uncorrelated errors -> used as a 2nd rescue. Measured: | |
| # adds ~+9pt AI recall over 49 alone with ~0 extra real false positives. (Does NOT | |
| # fix GPT — it shares the SD/FLUX-only training gap.) Loaded via the detector repo. | |
| # OFF by default: the 53 ensemble was validated and does NOT help under the | |
| # real-precision priority — at an FPR-safe gate it adds ~0 recall, and a low gate | |
| # triples real false positives. Kept for re-use after a GPT-data retrain fixes 53. | |
| ENABLE_SIGLIP = os.getenv("ENABLE_SIGLIP", "0") == "1" | |
| SIGLIP_PATH = os.getenv("SIGLIP_MODEL_PATH", "./models/best53.pt") | |
| SIGLIP_SIZE = 256 | |
| SIGLIP_THR = float(os.getenv("SIGLIP_THRESHOLD", "0.90")) | |
| SIGLIP_53_ARCH = dict( | |
| clip_backbone="siglip-large-256", clip_layer=13, semantic_dim=512, forensic_dim=256, | |
| frequency_dim=192, fft_bins=48, num_classes=2, num_sources=2, freeze_clip=True, | |
| source_grl_lambda=0.0, dropout=0.26, image_size=SIGLIP_SIZE, | |
| ) | |
| # composite rescue: only when model 49 says REAL but PatchGuard's TRAINED image | |
| # verdict (p_ai) is confidently AI. Measured on model 63: real-paste/control p_ai | |
| # clears 0.72 only ~3% of the time while AI inserts clear it ~36% — model 49 alone | |
| # catches just ~9% of inserts, so this rescue is the main insert detector. | |
| RESCUE_P59 = float(os.getenv("PATCHGUARD_RESCUE_THRESHOLD", "0.72")) | |
| # real-guard: model 49 sometimes false-positives on real CAMERA photos (out of its | |
| # training distribution). When 49 only *weakly* says AI, PatchGuard (model 61, which | |
| # has stronger real precision) confidently says REAL, there is no provenance and no | |
| # localized insert, we trust REAL — a camera photo must stay real. | |
| REAL_GUARD_MARGIN = 0.12 # only override when score49 is within this of the threshold | |
| REAL_GUARD_P59 = 0.20 # model 61 must be confidently real | |
| REAL_GUARD_PEAK = 0.45 # and find no localized insert | |
| def _read_threshold(path: str, default: float) -> float: | |
| try: | |
| ck = torch.load(path, map_location="cpu", weights_only=False) | |
| metrics = ck.get("metrics", {}) if isinstance(ck, dict) else {} | |
| for key in ("calibrated_threshold", "threshold"): | |
| v = metrics.get(key) | |
| if isinstance(v, (int, float)) and 0.0 < float(v) < 1.0: | |
| return float(v) | |
| except Exception: | |
| pass | |
| return default | |
| class HybridDetector: | |
| def __init__(self) -> None: | |
| self.device = torch.device(DEVICE) | |
| # VERDICT preprocessing must match how the models were trained/evaluated: | |
| # resize the shorter side to 256, then center-crop 224 (aspect preserved). | |
| # The old square Resize((224,224)) DISTORTED non-square images (screenshots | |
| # especially), which measurably hurt detection — e.g. a screenshotted GPT | |
| # image scored 0.60 squished vs 0.77 center-cropped. | |
| self.transform = T.Compose([ | |
| T.Resize(256, interpolation=T.InterpolationMode.BICUBIC), | |
| T.CenterCrop(INPUT_SIZE), | |
| T.ToTensor(), | |
| T.Normalize(CLIP_MEAN, CLIP_STD), | |
| ]) | |
| # HEATMAP preprocessing keeps the WHOLE image (square resize) so the overlay | |
| # lines up with the full displayed photo instead of only the center crop. | |
| self.transform_full = T.Compose([ | |
| T.Resize((INPUT_SIZE, INPUT_SIZE), interpolation=T.InterpolationMode.BICUBIC), | |
| T.ToTensor(), | |
| T.Normalize(CLIP_MEAN, CLIP_STD), | |
| ]) | |
| # SigLIP (model 53) takes 256px input (its own backbone resolution). | |
| self.transform_53 = T.Compose([ | |
| T.Resize(round(SIGLIP_SIZE / 0.875), interpolation=T.InterpolationMode.BICUBIC), | |
| T.CenterCrop(SIGLIP_SIZE), | |
| T.ToTensor(), | |
| T.Normalize(CLIP_MEAN, CLIP_STD), | |
| ]) | |
| self.verdict = ZeroShotV4Detector(**MODEL_49_ARCH) | |
| self._load(self.verdict, VERDICT_PATH) | |
| self.threshold = _read_threshold(VERDICT_PATH, 0.4625) | |
| if HEATMAP_KIND == "dense" or os.path.basename(HEATMAP_PATH).lower().startswith("best63"): | |
| self.patch = DensePatchGuardDetector(**DENSE_ARCH) | |
| else: | |
| self.patch = PatchGuardDetector(**PATCHGUARD_ARCH) | |
| self._load(self.patch, HEATMAP_PATH) | |
| # 2nd verdict head — SigLIP (model 53). OFF by default (see ENABLE_SIGLIP). | |
| self.siglip = None | |
| if ENABLE_SIGLIP: | |
| self.siglip = ZeroShotV4Detector(**SIGLIP_53_ARCH) | |
| self._load(self.siglip, SIGLIP_PATH) | |
| def _load(self, model: torch.nn.Module, path: str) -> None: | |
| ck = torch.load(path, map_location="cpu", weights_only=False) | |
| state = ck.get("model_trainable_state_dict", ck.get("model_state_dict", ck)) | |
| model.load_trainable_state_dict(state) | |
| model.to(self.device).eval() | |
| def _dense_patch(self, pil: Image.Image) -> np.ndarray: | |
| """Model 63's native dense localization map (32x32) over the WHOLE image | |
| (full-frame resize) so the heatmap aligns with the displayed photo.""" | |
| x = self.transform_full(pil).unsqueeze(0).to(self.device) | |
| out = self.patch(x) | |
| return torch.sigmoid(out["patch_logits"].float())[0].cpu().numpy() | |
| def _hires_loc(self, original: Image.Image, base: np.ndarray | None = None) -> np.ndarray: | |
| """Dense 32x32 localization with horizontal-flip TTA (averaging the map over the | |
| image and its mirror steadies it), fused with the noise-residual map and | |
| smoothed. `base` reuses the patch map already computed in analyze() so only the | |
| mirror pass is extra.""" | |
| ph = self._dense_patch(original) if base is None else base | |
| ph_flip = self._dense_patch(original.transpose(Image.FLIP_LEFT_RIGHT))[:, ::-1] | |
| ph = (ph + ph_flip) / 2.0 | |
| gh = ph.shape[0] | |
| rv = residual_var(original, gh) | |
| rng = float(np.ptp(rv)) | |
| rv = (rv - rv.min()) / (rng + 1e-6) if rng > 1e-6 else np.zeros_like(rv) | |
| return gaussian_filter(ph * (0.5 + 0.5 * rv), 0.7) | |
| def analyze(self, image_bytes: bytes, filename: str | None = None) -> dict[str, Any]: | |
| started = time.perf_counter() | |
| # Phone photos often carry EXIF orientation instead of rotated pixels. | |
| # Normalize it before model inference and overlay rendering so the | |
| # original/heatmap tabs do not appear rotated relative to each other. | |
| original = ImageOps.exif_transpose(Image.open(io.BytesIO(image_bytes))).convert("RGB") | |
| x = self.transform(original).unsqueeze(0).to(self.device) | |
| provenance = provenance_gate.check_bytes(image_bytes) | |
| filename_signals = provenance_gate.check_filename(filename) | |
| signals = sorted(set((provenance.get("signals") or []) + filename_signals)) | |
| # Optional SynthID pixel-watermark check (no-op unless SYNTHID_API_* env set). | |
| # Unlike byte/metadata provenance, this survives screenshots / re-encoding. | |
| synthid = synthid_gate.check_image(image_bytes) | |
| if synthid.get("signals"): | |
| signals = sorted(set(signals + synthid["signals"])) | |
| provenance.setdefault("details", {})["synthid"] = synthid.get("details") | |
| provenance["signals"] = signals | |
| if filename_signals: | |
| provenance.setdefault("details", {})["filename_hint"] = filename | |
| out49 = self.verdict(x) | |
| logits = out49["logits"] if isinstance(out49, dict) else out49 | |
| score49 = float(torch.softmax(logits.float().reshape(-1)[:2], dim=0)[1]) | |
| score = score49 | |
| pg = self.patch(x) | |
| patch = torch.sigmoid(pg["patch_logits"].float())[0].cpu().numpy() # 16x16 | |
| peak = float(patch.max()) | |
| p_ai_59 = float(torch.sigmoid(pg["z_img"].float())[0]) | |
| # 2nd verdict head (SigLIP/53) — only when enabled (off by default). | |
| s53, ai53 = 0.0, False | |
| if self.siglip is not None: | |
| out53 = self.siglip(self.transform_53(original).unsqueeze(0).to(self.device)) | |
| logits53 = out53["logits"] if isinstance(out53, dict) else out53 | |
| s53 = float(torch.softmax(logits53.float().reshape(-1)[:2], dim=0)[1]) | |
| ai53 = s53 >= SIGLIP_THR | |
| source = "model" | |
| verdict = "AI" if score >= self.threshold else "REAL" | |
| model_verdict = verdict # the pixel model's OWN call, before provenance/rescue | |
| reason = "model 49 verdict" | |
| patchguard_would_rescue = model_verdict == "REAL" and p_ai_59 >= RESCUE_P59 | |
| rescued = False | |
| if signals: | |
| verdict = "AI" | |
| score = max(score, 0.99) | |
| source = "provenance" | |
| reason = "AI provenance found (" + ", ".join(str(s) for s in signals) + ")" | |
| rescued = patchguard_would_rescue | |
| if model_verdict == "AI" or patchguard_would_rescue or ai53: | |
| reason += "; pixel model also positive" | |
| elif patchguard_would_rescue: | |
| verdict = "AI" | |
| score = max(score, p_ai_59) | |
| source = "patchguard_rescue" | |
| rescued = True | |
| reason = "model 49 + AI-insert rescue (PatchGuard)" | |
| elif model_verdict == "REAL" and ai53: | |
| # 49 said REAL, but the SigLIP head (different backbone) flags AI. | |
| verdict = "AI" | |
| score = max(score, s53) | |
| source = "siglip_rescue" | |
| rescued = True | |
| reason = "model 49 REAL but SigLIP(53) ensemble flags AI" | |
| elif (model_verdict == "AI" and not ai53 and score49 < self.threshold + REAL_GUARD_MARGIN | |
| and p_ai_59 < REAL_GUARD_P59 and peak < REAL_GUARD_PEAK): | |
| # camera/real photo that model 49 only weakly flags as AI, with no | |
| # provenance and no localized insert, and PatchGuard says clearly REAL. | |
| verdict = "REAL" | |
| source = "real_guard" | |
| score = min(score, self.threshold - 0.01) | |
| reason = "model 49 weak-AI overridden by PatchGuard real-guard (camera/real)" | |
| # Downscale for display so the base64 payload stays small. A full-res | |
| # PNG of a phone photo is 10-25MB and crashes the app on decode. 1024px | |
| # is plenty for the heatmap overlay; the real-verdict thumbnail is smaller. | |
| display = _downscale(original, 768) | |
| if verdict == "AI": | |
| # Always show the REAL model heatmap — even when provenance is what flagged | |
| # the image. (Previously the provenance-only case painted a flat red wash; | |
| # the user prefers a real localized heatmap. For a fully-AI image the model | |
| # has no single region to point at, so this highlights its most-suspect | |
| # area — approximate, but a real heatmap rather than a flat blanket.) | |
| # Dense localization over the WHOLE image (residual-fused + flip-TTA), so | |
| # the overlay lines up with the full displayed photo. | |
| loc = self._hires_loc(original) | |
| # Always render VIVID — the hot region is normalized to full color, no | |
| # confidence dimming (product preference: heatmap should never look faint). | |
| conf = 1.0 | |
| heatmap_b64 = _png_b64(pure_heatmap(loc, display.size, conf, blanket=False)) | |
| overlay_b64 = _png_b64(render_overlay(display, loc, conf, blanket=False)) | |
| flat = np.sort(patch.ravel())[::-1] | |
| k = max(1, int(round(flat.size * 0.08))) | |
| structure_score = float(flat[:k].mean()) | |
| detail_score = peak | |
| else: | |
| # proper PNG (the frontend labels data-uris image/png; a JPEG payload | |
| # there can crash the native image decoder on some phones) | |
| clean = _png_b64(_downscale(original, 600)) | |
| heatmap_b64 = clean | |
| overlay_b64 = clean | |
| structure_score = detail_score = score | |
| reason = "model 49 verdict - real, heatmap suppressed" | |
| return { | |
| "score": score, | |
| "verdict": verdict, | |
| "heatmap_b64": heatmap_b64, | |
| "overlay_b64": overlay_b64, | |
| "model_version": ("49+53" if ENABLE_SIGLIP else "49") + "_verdict+63_heatmap" + ("+provenance" if source == "provenance" else "") + ("+rescue" if rescued else ""), | |
| "elapsed_ms": int((time.perf_counter() - started) * 1000), | |
| "structure_score": structure_score, | |
| "context_score": score, | |
| "detail_score": detail_score, | |
| "threshold": self.threshold, | |
| "reason": reason, | |
| "source": source, | |
| "provenance": provenance, | |
| "ensemble": {"model49_score": score49, "siglip53_score": s53, "patchguard_score": p_ai_59, "patch_peak": peak, "rescue_threshold": RESCUE_P59}, | |
| } | |
| def _png_b64(image: Image.Image) -> str: | |
| buf = io.BytesIO() | |
| image.save(buf, format="PNG") | |
| return base64.b64encode(buf.getvalue()).decode("ascii") | |
| def _downscale(image: Image.Image, max_side: int) -> Image.Image: | |
| w, h = image.size | |
| if max(w, h) <= max_side: | |
| return image | |
| s = max_side / float(max(w, h)) | |
| return image.resize((max(1, int(w * s)), max(1, int(h * s))), Image.Resampling.BILINEAR) | |
| _detector: HybridDetector | None = None | |
| async def lifespan(app: FastAPI) -> AsyncIterator[None]: | |
| global _detector | |
| print(f"[hybrid] loading 49 (verdict) + 63 (dense heatmap/rescue) on {DEVICE} ...", flush=True) | |
| _detector = HybridDetector() | |
| print(f"[hybrid] ready: verdict=model49 thr={_detector.threshold} heatmap=model63 dense", flush=True) | |
| yield | |
| app = FastAPI(title="A-EYE Inference API (hybrid)", version="3.0.0", lifespan=lifespan) | |
| app.add_middleware(CORSMiddleware, allow_origins=["*"], allow_credentials=True, | |
| allow_methods=["*"], allow_headers=["*"]) | |
| _WEBAPP_DIR = os.path.join(os.path.dirname(os.path.abspath(__file__)), "webapp") | |
| _TEST_PAGE = """<!doctype html><html lang=ko><head><meta charset=utf-8> | |
| <meta name=viewport content="width=device-width,initial-scale=1"><title>A-EYE hybrid</title> | |
| <style>body{font-family:system-ui,sans-serif;background:#0b1020;color:#e8ecf5;margin:0;padding:24px} | |
| .wrap{max-width:1000px;margin:0 auto}h1{font-size:20px;margin:0 0 4px}.sub{color:#8a93a8;font-size:13px;margin-bottom:18px} | |
| .drop{display:block;border:2px dashed #34415f;border-radius:14px;padding:26px;text-align:center;background:#0f1730;cursor:pointer} | |
| input[type=file]{display:none}button{margin-top:12px;padding:11px 18px;border:0;border-radius:10px;background:#2563eb;color:#fff;font-size:15px;cursor:pointer} | |
| .verdict{margin-top:20px;padding:15px;border-radius:14px;font-size:18px;font-weight:700}.ai{background:rgba(239,68,68,.15);color:#ff6b6b}.real{background:rgba(16,185,129,.15);color:#34d399} | |
| .meta{color:#8a93a8;font-size:13px;font-weight:400;margin-top:6px}.grid{display:grid;grid-template-columns:repeat(3,1fr);gap:12px;margin-top:16px} | |
| .card{background:#0f1730;border:1px solid #25304d;border-radius:12px;padding:9px;text-align:center}.card h3{font-size:12px;color:#8a93a8;margin:0 0 8px}.card img{width:100%;border-radius:8px;display:block} | |
| @media(max-width:700px){.grid{grid-template-columns:1fr}}</style></head><body><div class=wrap> | |
| <h1>A-EYE <span style=color:#8a93a8;font-size:13px>· 판정=49 / 히트맵=59</span></h1><div class=sub>판정은 검증된 49번, 히트맵은 PatchGuard 국소맵(진하게). 실사진+AI합성은 보강 검출.</div> | |
| <label class=drop><div id=dt>사진을 클릭해서 선택</div><input type=file id=f accept=image/*></label><button id=go disabled>분석</button><div id=out></div></div> | |
| <script>const f=document.getElementById('f'),dt=document.getElementById('dt'),go=document.getElementById('go'),out=document.getElementById('out');let p=null; | |
| f.onchange=()=>{p=f.files[0]||null;dt.textContent=p?p.name:'사진을 클릭해서 선택';go.disabled=!p}; | |
| go.onclick=async()=>{if(!p)return;go.disabled=1;out.innerHTML='<p class=meta>분석 중…</p>';const fd=new FormData();fd.append('image',p,p.name); | |
| try{const r=await fetch('/api/analyze',{method:'POST',body:fd});const d=await r.json();if(!r.ok)throw new Error(d.detail||r.status); | |
| const ai=d.verdict==='AI',pct=Math.round((d.score||0)*100),o=URL.createObjectURL(p); | |
| out.innerHTML='<div class="verdict '+(ai?'ai':'real')+'">'+(ai?'AI 생성 의심':'실제 사진')+' · AI 확률 '+pct+'%<div class=meta>'+d.model_version+' · 임계값 '+d.threshold.toFixed(3)+' · '+d.elapsed_ms+'ms</div></div>'+ | |
| '<div class=grid><div class=card><h3>원본</h3><img src="'+o+'"></div><div class=card><h3>히트맵</h3><img src="data:image/png;base64,'+d.heatmap_b64+'"></div><div class=card><h3>오버레이</h3><img src="data:image/png;base64,'+d.overlay_b64+'"></div></div>'; | |
| }catch(e){out.innerHTML='<p style=color:#ff6b6b>오류: '+e.message+'</p>'}go.disabled=0}</script></body></html>""" | |
| async def index() -> str: | |
| page = os.path.join(os.path.dirname(os.path.abspath(__file__)), "webapp", "index.html") | |
| if os.path.isfile(page): | |
| with open(page, encoding="utf-8") as fh: | |
| return fh.read() | |
| return _TEST_PAGE | |
| async def health() -> HealthResponse: | |
| return HealthResponse(status="ok") | |
| async def version() -> VersionResponse: | |
| return VersionResponse(model_version=("49+53" if ENABLE_SIGLIP else "49") + "_verdict+63_heatmap", model_path=VERDICT_PATH) | |
| async def predict(image: UploadFile = File(...)) -> dict[str, Any]: | |
| """Compat endpoint for the a-Eye mobile/web app (fake_prob/real_prob shape).""" | |
| if not (image.content_type or "").startswith("image/"): | |
| raise HTTPException(status_code=400, detail="Invalid image") | |
| image_bytes = await image.read() | |
| if not image_bytes or len(image_bytes) > MAX_IMAGE_BYTES: | |
| raise HTTPException(status_code=400, detail="Invalid image") | |
| if _detector is None: | |
| raise HTTPException(status_code=503, detail="Model not loaded") | |
| try: | |
| result = await run_in_threadpool(_detector.analyze, image_bytes, image.filename) | |
| except Exception as exc: # noqa: BLE001 | |
| return {"error": f"Invalid image: {exc}"} | |
| fake = float(result["score"]) * 100.0 | |
| en = result.get("ensemble") or {} | |
| print( | |
| f"[predict] {len(image_bytes)//1024}KB -> {result['verdict']} " | |
| f"src={result.get('source')} score={result['score']:.3f}", | |
| flush=True, | |
| ) | |
| return { | |
| "fake_prob": fake, | |
| "real_prob": 100.0 - fake, | |
| "confidence": max(fake, 100.0 - fake), | |
| "model_version": result.get("model_version"), | |
| "model_probs": { | |
| "model49 verdict": float(en.get("model49_score", 0.0)) * 100.0, | |
| "patchguard63 insert": float(en.get("patchguard_score", 0.0)) * 100.0, | |
| }, | |
| "verdict": result.get("verdict"), | |
| "reason": result.get("reason"), | |
| "heatmap_b64": result.get("heatmap_b64"), | |
| "overlay_b64": result.get("overlay_b64"), | |
| } | |
| async def analyze(image: UploadFile = File(...)) -> AnalyzeResponse: | |
| if (image.content_type or "") not in {"image/jpeg", "image/png", "image/webp"}: | |
| raise HTTPException(status_code=400, detail="Invalid image") | |
| image_bytes = await image.read() | |
| if not image_bytes or len(image_bytes) > MAX_IMAGE_BYTES: | |
| raise HTTPException(status_code=400, detail="Invalid image") | |
| if _detector is None: | |
| raise HTTPException(status_code=503, detail="Model not loaded") | |
| try: | |
| # run the blocking torch inference off the event loop so concurrent | |
| # requests don't serialize behind each other. | |
| result = await run_in_threadpool(_detector.analyze, image_bytes, image.filename) | |
| print( | |
| f"[req] {len(image_bytes)//1024}KB ct={image.content_type} name={image.filename} " | |
| f"-> {result['verdict']} src={result.get('source')} score={result['score']:.3f} " | |
| f"m49={result.get('ensemble', {}).get('model49_score', 0.0):.3f} " | |
| f"pg={result.get('ensemble', {}).get('patchguard_score', 0.0):.3f} " | |
| f"reason={result.get('reason')}", | |
| flush=True, | |
| ) | |
| return AnalyzeResponse(**result) | |
| except Exception as exc: # noqa: BLE001 | |
| raise HTTPException(status_code=400, detail=f"Invalid image: {exc}") from exc | |
| from fastapi.responses import FileResponse # noqa: E402 | |
| async def webapp_files(path: str) -> FileResponse: | |
| full = os.path.normpath(os.path.join(_WEBAPP_DIR, path)) | |
| if not full.startswith(os.path.normpath(_WEBAPP_DIR)): | |
| raise HTTPException(status_code=404, detail="Not found") | |
| if os.path.isfile(full): | |
| return FileResponse(full) | |
| if os.path.isfile(full + ".html"): | |
| return FileResponse(full + ".html") | |
| index = os.path.join(_WEBAPP_DIR, "index.html") | |
| if os.path.isfile(index): | |
| return FileResponse(index) | |
| raise HTTPException(status_code=404, detail="Not found") | |