prog/video.py

# prog/video.py
import cv2
import torch
import numpy as np
from PIL import Image

# ── Model identifiers ─────────────────────────────────────────
# Only using your confirmed-working model.
# Simma7/ffeg has been removed — it is private/deleted.
MODEL_NAME = "simma7/image_model"

device = "cpu"

# ── Lazy model loading — loads ONCE on first video call ───────
_model     = None
_processor = None

def _load():
    global _model, _processor
    if _model is None:
        from transformers import AutoImageProcessor, SiglipForImageClassification
        print("Loading video model (simma7/image_model)...")
        _processor = AutoImageProcessor.from_pretrained(MODEL_NAME)
        _model     = SiglipForImageClassification.from_pretrained(MODEL_NAME).to(device)
        _model.eval()
        print("Video model loaded.")


def _score_frame(pil_img) -> tuple:
    """Run one frame through the model. Returns (real_score, fake_score)."""
    _load()
    inputs = _processor(images=pil_img, return_tensors="pt").to(device)
    with torch.no_grad():
        outputs = _model(**inputs)
        probs   = torch.nn.functional.softmax(outputs.logits, dim=-1)[0]

    labels = _model.config.id2label
    real_s, fake_s = 0.0, 0.0

    for i, prob in enumerate(probs):
        lbl = str(labels.get(i, "")).lower()
        s   = float(prob)
        if "deepfake" in lbl:
            fake_s += s
        elif any(k in lbl for k in ("real", "ai", "artificial")):
            real_s += s
        else:
            fake_s += s

    total = real_s + fake_s
    if total == 0:
        return 0.5, 0.5
    return real_s / total, fake_s / total


def _threat(fake_conf: float) -> tuple:
    if fake_conf < 0.50:
        return "NONE",   "✅ ALLOW"
    elif fake_conf < 0.60:
        return "LOW",    "📋 LOG"
    elif fake_conf < 0.85:
        return "MEDIUM", "⚠️ ALERT"
    else:
        return "HIGH",   "🚫 BLOCK"


def detect_video(video_path: str) -> str:
    if not video_path:
        return "⚠️ No video provided."

    try:
        cap = cv2.VideoCapture(video_path)
        if not cap.isOpened():
            return "❌ Could not open video file."

        total_frames = int(cap.get(cv2.CAP_PROP_FRAME_COUNT))
        fps_video    = cap.get(cv2.CAP_PROP_FPS) or 25

        # Sample up to 15 frames spread evenly across the video
        max_samples = 15
        if total_frames <= 0:
            indices = list(range(0, 300, 20))   # fallback
        else:
            indices = np.linspace(0, total_frames - 1,
                                  min(max_samples, total_frames),
                                  dtype=int).tolist()

        real_scores, fake_scores = [], []

        for idx in indices:
            cap.set(cv2.CAP_PROP_POS_FRAMES, int(idx))
            ret, frame = cap.read()
            if not ret:
                continue
            try:
                rgb  = cv2.cvtColor(frame, cv2.COLOR_BGR2RGB)
                img  = Image.fromarray(rgb).resize((224, 224))
                r, f = _score_frame(img)
                real_scores.append(r)
                fake_scores.append(f)
            except Exception:
                continue

        cap.release()

        if not real_scores:
            return "❌ Could not process any frames from this video."

        real_avg = float(np.mean(real_scores))
        fake_avg = float(np.mean(fake_scores))
        frames   = len(real_scores)

        # Normalise
        total    = real_avg + fake_avg
        real_avg /= total
        fake_avg /= total

        prediction     = "FAKE" if fake_avg >= 0.5 else "REAL"
        threat, action = _threat(fake_avg)
        bar = "█" * int(fake_avg * 20) + "░" * (20 - int(fake_avg * 20))

        return (
            f"🎥 VIDEO RESULT\n"
            f"{'='*38}\n\n"
            f"Prediction : {'⚠️ FAKE' if prediction == 'FAKE' else '✅ REAL'}\n"
            f"Fake score : {fake_avg:.2%}  [{bar}]\n"
            f"Real score : {real_avg:.2%}\n"
            f"Threat     : {threat}\n"
            f"Action     : {action}\n"
            f"Frames     : {frames} analysed\n\n"
            f"{'⚠️ Deepfake signals detected in video.' if prediction == 'FAKE' else '✅ Video appears authentic.'}"
        )

    except Exception as e:
        return f"❌ Video analysis error: {str(e)}"