Deploy VeriVid Backend to Hugging Face Spaces

.dockerignore

@@ -0,0 +1,6 @@
+__pycache__
+*.pyc
+.git
+.env
+temp/
+cache/

.env.example

@@ -0,0 +1,6 @@
+# Backend Environment Variables
+# For production, set these in Render dashboard
+
+HF_TOKEN=your_huggingface_token_here
+SIGHTENGINE_API_USER=your_sightengine_user
+SIGHTENGINE_API_SECRET=your_sightengine_secret

.gitignore

@@ -0,0 +1,6 @@
+.env
+__pycache__/
+*.pyc
+temp/
+cache/
+*.log

Dockerfile

@@ -0,0 +1,30 @@
+# Use Python 3.9
+FROM python:3.9
+
+# Install FFmpeg (Required for VeriVid pipeline)
+RUN apt-get update && apt-get install -y ffmpeg && rm -rf /var/lib/apt/lists/*
+
+# Set up a new user named "user" with user ID 1000
+RUN useradd -m -u 1000 user
+
+# Switch to the "user" user
+USER user
+
+# Set home to the user's home directory
+ENV HOME=/home/user \
+	PATH=/home/user/.local/bin:$PATH
+
+# Set the working directory to the user's home directory
+WORKDIR $HOME/app
+
+# Copy the current directory contents into the container at $HOME/app setting the owner to the user
+COPY --chown=user . $HOME/app
+
+# Install requirements
+RUN pip install --no-cache-dir --upgrade -r requirements.txt
+
+# Create temp directory for processing
+RUN mkdir -p $HOME/app/temp
+
+# Run the application on port 7860
+CMD ["uvicorn", "main:app", "--host", "0.0.0.0", "--port", "7860"]

app/api/routes.py

@@ -0,0 +1,90 @@
+# C:\Users\bahae\.gemini\antigravity\scratch\verivid-ai\backend\app\api\routes.py
+from fastapi import APIRouter, BackgroundTasks, UploadFile, File, Form, Request
+from fastapi.responses import JSONResponse
+from slowapi import Limiter
+from slowapi.util import get_remote_address
+from app.services.pipeline import run_analysis_pipeline
+import uuid
+import os
+import shutil
+
+router = APIRouter()
+limiter = Limiter(key_func=get_remote_address)
+
+JOBS = {}
+TEMP_DIR = os.path.join(os.path.dirname(__file__), '..', '..', 'temp')
+
+@router.post("/analyze")
+@limiter.limit("10/hour")  # 10 requests per hour per IP
+async def start_analysis(
+    request: Request,
+    background_tasks: BackgroundTasks, 
+    url: str = Form(None), 
+    file: UploadFile = File(None)
+):
+    """
+    Start video analysis.
+    Rate limited to 10 requests per hour per IP.
+    """
+    # Validate input
+    if not url and not (file and file.filename):
+        return JSONResponse(
+            status_code=400,
+            content={"error": "Please provide either a URL or upload a file"}
+        )
+    
+    # Basic URL validation
+    if url:
+        if len(url) > 500:
+            return JSONResponse(status_code=400, content={"error": "URL too long"})
+        if not url.startswith(("http://", "https://")):
+            return JSONResponse(status_code=400, content={"error": "Invalid URL format"})
+    
+    job_id = str(uuid.uuid4())
+    JOBS[job_id] = {"status": "queued"}
+    
+    # If file uploaded, save it with size limit
+    file_path = None
+    if file and file.filename:
+        # 100MB limit
+        MAX_SIZE = 100 * 1024 * 1024
+        contents = await file.read()
+        if len(contents) > MAX_SIZE:
+            return JSONResponse(status_code=400, content={"error": "File too large (max 100MB)"})
+        
+        os.makedirs(TEMP_DIR, exist_ok=True)
+        file_path = os.path.join(TEMP_DIR, f"{job_id}_upload.mp4")
+        with open(file_path, 'wb') as f:
+            f.write(contents)
+    
+    # Start async processing
+    background_tasks.add_task(
+        run_analysis_pipeline, 
+        job_id, 
+        url, 
+        file_path,
+        JOBS
+    )
+    
+    return {"job_id": job_id, "status": "queued"}
+
+@router.get("/analyze/{job_id}")
+@limiter.limit("60/minute")  # 60 result checks per minute
+async def get_result(request: Request, job_id: str):
+    """Get analysis result by job ID"""
+    # Validate job_id format (UUID)
+    try:
+        uuid.UUID(job_id)
+    except ValueError:
+        return JSONResponse(status_code=400, content={"error": "Invalid job ID"})
+    
+    result = JOBS.get(job_id)
+    if not result:
+        return JSONResponse(status_code=404, content={"status": "not_found"})
+    
+    return result
+
+@router.get("/health")
+async def health_check():
+    """Health check endpoint"""
+    return {"status": "ok", "jobs_in_memory": len(JOBS)}

app/core/config.py

@@ -0,0 +1,19 @@
+from pydantic_settings import BaseSettings
+import os
+
+class Settings(BaseSettings):
+    """
+    Settings resolve from environment variables.
+    Secrets must be set in the deployment environment (Render, Hugging Face, etc).
+    """
+
+    HF_TOKEN: str = os.getenv("HF_TOKEN", "")
+    SIGHTENGINE_API_USER: str = os.getenv("SIGHTENGINE_API_USER", "")
+    SIGHTENGINE_API_SECRET: str = os.getenv("SIGHTENGINE_API_SECRET", "")
+
+    class Config:
+        env_file = ".env"
+        env_file_encoding = "utf-8"
+
+
+settings = Settings()

app/core/scoring.py

@@ -0,0 +1,56 @@
+# C:\Users\bahae\.gemini\antigravity\scratch\verivid-ai\backend\app\core\scoring.py
+
+def calculate_risk(signals: dict):
+    """
+    Calculate final risk score using SightEngine-calibrated thresholds
+    """
+    visual = signals.get('visual', {})
+    audio = signals.get('audio', {})
+    meta = signals.get('metadata', {})
+    heur = signals.get('heuristics', {})
+    
+    v_avg = visual.get('avg_prob', 0)
+    v_max = visual.get('max_prob', 0)
+    frame_count = visual.get('frame_count', 1)
+    
+    a_score = audio.get('spoof_prob', 0)
+    m_score = meta.get('risk_score', 0)
+    h_score = heur.get('risk_score', 0)
+    
+    # Use max between avg and max (catches localized AI)
+    visual_prob = max(v_avg, v_max * 0.9)
+    
+    # SightEngine-calibrated weights
+    # SightEngine is more reliable, so we trust visual more
+    weights = {
+        "visual": 0.60,
+        "audio": 0.10,
+        "metadata": 0.20,
+        "heuristics": 0.10
+    }
+    
+    # Calculate weighted score
+    final_score = (
+        visual_prob * 100 * weights['visual'] +
+        a_score * 100 * weights['audio'] +
+        m_score * 100 * weights['metadata'] +
+        h_score * 100 * weights['heuristics']
+    )
+    
+    # Confidence based on frame count and visual strength
+    if frame_count >= 3 and visual_prob > 0.5:
+        confidence = "HIGH"
+    elif frame_count >= 2 or visual_prob > 0.3:
+        confidence = "MEDIUM"
+    else:
+        confidence = "LOW"
+    
+    # Recommendation thresholds
+    if final_score >= 50:
+        rec = "HIGH RISK"
+    elif final_score >= 25:
+        rec = "REVIEW"
+    else:
+        rec = "SAFE"
+    
+    return round(final_score), confidence, rec

app/services/downloader.py

@@ -0,0 +1,442 @@
+# C:\Users\bahae\.gemini\antigravity\scratch\verivid-ai\backend\app\services\downloader.py
+"""
+Smart Scraper Pipeline with Cobalt API
+=======================================
+This module uses a 3-layer strategy for video extraction:
+1. Cobalt API (external, avoids IP blocks)
+2. yt-dlp streaming (local, zero storage)
+3. yt-dlp download (fallback, uses temp storage)
+
+Storage usage: ~500KB per analysis (5 frames) instead of 50-100MB per video.
+"""
+
+import subprocess
+import os
+import glob
+import requests
+from urllib.parse import urlparse, urlunparse
+import yt_dlp
+
+TEMP_DIR = os.path.join(os.path.dirname(__file__), '..', '..', 'temp')
+
+# Cobalt API endpoints (public instances)
+COBALT_ENDPOINTS = [
+    "https://api.cobalt.tools",
+    "https://co.wuk.sh",  # Backup instance
+]
+
+
+def ensure_temp_dir():
+    os.makedirs(TEMP_DIR, exist_ok=True)
+    
+
+def clean_temp(job_id: str):
+    """Clean up temp files for a job"""
+    pattern = os.path.join(TEMP_DIR, f"{job_id}*")
+    for f in glob.glob(pattern):
+        try:
+            os.remove(f)
+        except:
+            pass
+
+
+def normalize_url(url: str) -> str:
+    """Clean common tracking params from YouTube Shorts to improve yt-dlp success."""
+    try:
+        parsed = urlparse(url)
+        if "youtube.com" in parsed.netloc and "/shorts/" in parsed.path:
+            parsed = parsed._replace(query="")
+            return urlunparse(parsed)
+    except Exception:
+        pass
+    return url
+
+
+# ============================================================
+# LAYER 1: COBALT API (Best - External service, avoids IP blocks)
+# ============================================================
+
+def get_cobalt_url(url: str) -> tuple[str, str]:
+    """
+    Query Cobalt API to get direct stream URL.
+    Returns (video_url, audio_url) or (None, None) on failure.
+    
+    Cobalt handles TikTok, Instagram, YouTube, Twitter, etc. perfectly
+    without our server IP getting blocked.
+    """
+    url = normalize_url(url)
+    
+    headers = {
+        "Accept": "application/json",
+        "Content-Type": "application/json",
+    }
+    
+    payload = {
+        "url": url,
+        "vCodec": "h264",
+        "vQuality": "720",
+        "aFormat": "mp3",
+        "isNoTTWatermark": True,  # TikTok without watermark
+        "isAudioOnly": False,
+    }
+    
+    for endpoint in COBALT_ENDPOINTS:
+        try:
+            print(f"[Cobalt] Trying {endpoint}...")
+            response = requests.post(
+                f"{endpoint}/api/json",
+                json=payload,
+                headers=headers,
+                timeout=15
+            )
+            
+            if response.status_code != 200:
+                print(f"[Cobalt] {endpoint} returned {response.status_code}")
+                continue
+                
+            data = response.json()
+            status = data.get("status")
+            
+            if status == "stream" or status == "redirect":
+                # Direct stream URL available
+                stream_url = data.get("url")
+                print(f"[Cobalt] Success! Got stream URL from {endpoint}")
+                return stream_url, None
+                
+            elif status == "picker":
+                # Multiple options (e.g., video + audio separate)
+                picker = data.get("picker", [])
+                video_url = None
+                audio_url = None
+                for item in picker:
+                    if item.get("type") == "video" and not video_url:
+                        video_url = item.get("url")
+                    elif item.get("type") == "audio" and not audio_url:
+                        audio_url = item.get("url")
+                if video_url:
+                    print(f"[Cobalt] Success! Got picker URLs from {endpoint}")
+                    return video_url, audio_url
+                    
+            elif status == "error":
+                print(f"[Cobalt] Error from {endpoint}: {data.get('text', 'Unknown error')}")
+                continue
+                
+        except requests.Timeout:
+            print(f"[Cobalt] Timeout from {endpoint}")
+            continue
+        except Exception as e:
+            print(f"[Cobalt] Exception from {endpoint}: {e}")
+            continue
+    
+    print("[Cobalt] All endpoints failed, falling back to yt-dlp")
+    return None, None
+
+
+def smart_get_stream_url(url: str) -> tuple[str, str, str]:
+    """
+    Smart 3-layer strategy to get the best stream URL:
+    
+    Layer 1: Cobalt API (external, fast, avoids blocks)
+    Layer 2: yt-dlp direct URL (local, may get blocked)
+    Layer 3: Returns None (caller should use download fallback)
+    
+    Returns: (video_url, audio_url, source) where source is 'cobalt', 'ytdlp', or None
+    """
+    # Layer 1: Try Cobalt first
+    video_url, audio_url = get_cobalt_url(url)
+    if video_url:
+        return video_url, audio_url, "cobalt"
+    
+    # Layer 2: Try yt-dlp direct URL
+    video_url, audio_url = get_direct_url(url)
+    if video_url:
+        return video_url, audio_url, "ytdlp"
+    
+    # Layer 3: Signal caller to use download fallback
+    return None, None, None
+
+
+def get_video_info(url: str):
+    """Extract metadata without downloading"""
+    url = normalize_url(url)
+    ydl_opts = {
+        'quiet': True,
+        'no_warnings': True,
+        'skip_download': True,
+        'noplaylist': True,
+        'geo_bypass': True,
+        'extractor_args': {
+            'youtube': {
+                'player_client': ['android'],
+                'geo_bypass_country': ['US']
+            }
+        },
+        'http_headers': {
+            'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64)',
+        },
+    }
+    
+    try:
+        with yt_dlp.YoutubeDL(ydl_opts) as ydl:
+            info = ydl.extract_info(url, download=False)
+            return {
+                "title": info.get('title'),
+                "thumbnail": info.get('thumbnail'),
+                "duration": info.get('duration'),
+                "uploader": info.get('uploader'),
+                "view_count": info.get('view_count'),
+                "fps": info.get('fps'),
+                "vcodec": info.get('vcodec'),
+                "acodec": info.get('acodec'),
+                "width": info.get('width'),
+                "height": info.get('height'),
+            }
+    except Exception as e:
+        print(f"yt-dlp info error: {e}")
+        return None
+
+
+def get_direct_url(url: str) -> tuple[str, str]:
+    """
+    Get direct video and audio URLs without downloading.
+    Returns (video_url, audio_url) - audio_url may be None.
+    """
+    url = normalize_url(url)
+    ydl_opts = {
+        'quiet': True,
+        'no_warnings': True,
+        'skip_download': True,
+        'noplaylist': True,
+        'geo_bypass': True,
+        'format': 'bestvideo[ext=mp4][height<=720]/best[ext=mp4]/best',
+        'extractor_args': {
+            'youtube': {
+                'player_client': ['android'],
+                'geo_bypass_country': ['US']
+            }
+        },
+        'http_headers': {
+            'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64)',
+        },
+    }
+    
+    try:
+        with yt_dlp.YoutubeDL(ydl_opts) as ydl:
+            info = ydl.extract_info(url, download=False)
+            video_url = info.get('url')
+            
+            # Try to get audio URL separately for platforms with separated streams
+            audio_url = None
+            if info.get('requested_formats'):
+                for fmt in info['requested_formats']:
+                    if fmt.get('acodec') != 'none' and fmt.get('vcodec') == 'none':
+                        audio_url = fmt.get('url')
+                        break
+            
+            return video_url, audio_url
+    except Exception as e:
+        print(f"yt-dlp URL extraction error: {e}")
+        return None, None
+
+
+def stream_extract_frames(url: str, job_id: str, max_frames: int = 5, duration: int = 30) -> list:
+    """
+    SMART STREAMING: Uses Cobalt API first, then yt-dlp fallback.
+    
+    Pipeline: URL → Cobalt/yt-dlp → ffmpeg → frame JPEGs
+    
+    Only the small frame images (~100KB each) are saved to disk.
+    """
+    ensure_temp_dir()
+    frame_pattern = os.path.join(TEMP_DIR, f"{job_id}_frame_%03d.jpg")
+    
+    # Get streaming URL using smart 3-layer strategy
+    video_url, _, source = smart_get_stream_url(url)
+    if not video_url:
+        print(f"[{job_id}] Could not get stream URL from any source")
+        return []
+    
+    print(f"[{job_id}] Using stream URL from: {source}")
+    
+    try:
+        # Calculate fps to get max_frames from duration seconds
+        # e.g., 5 frames from 30 seconds = 1 frame every 6 seconds = 0.167 fps
+        fps = max_frames / duration
+        
+        # FFmpeg reads directly from the URL (no disk write for video)
+        cmd = [
+            'ffmpeg',
+            '-t', str(duration),           # Only process first N seconds
+            '-i', video_url,               # Input directly from URL
+            '-vf', f'fps={fps}',           # Extract at calculated fps
+            '-frames:v', str(max_frames),  # Max frames
+            '-q:v', '2',                   # High quality JPEG
+            '-y',                          # Overwrite
+            frame_pattern
+        ]
+        
+        result = subprocess.run(
+            cmd, 
+            capture_output=True, 
+            timeout=90,
+            env={**os.environ, 'FFMPEG_HTTP_SEEKABLE': '1'}
+        )
+        
+        if result.returncode != 0:
+            print(f"FFmpeg stderr: {result.stderr.decode()[:500]}")
+        
+        # Find extracted frames
+        frames = sorted(glob.glob(os.path.join(TEMP_DIR, f"{job_id}_frame_*.jpg")))
+        return frames
+        
+    except subprocess.TimeoutExpired:
+        print("FFmpeg streaming timeout")
+        return []
+    except Exception as e:
+        print(f"FFmpeg streaming error: {e}")
+        return []
+
+
+def stream_extract_audio(url: str, job_id: str, duration: int = 30) -> str:
+    """
+    Extract audio using smart 3-layer strategy (Cobalt first, then yt-dlp).
+    Output is a small WAV file (~500KB for 30s mono 16kHz).
+    """
+    ensure_temp_dir()
+    audio_path = os.path.join(TEMP_DIR, f"{job_id}_audio.wav")
+    
+    # Get URL using smart 3-layer strategy
+    video_url, audio_url, source = smart_get_stream_url(url)
+    source_url = audio_url or video_url
+    
+    if not source_url:
+        print(f"[{job_id}] Could not get URL for audio extraction")
+        return None
+    
+    print(f"[{job_id}] Using audio source from: {source}")
+    
+    try:
+        cmd = [
+            'ffmpeg',
+            '-t', str(duration),       # Only first N seconds
+            '-i', source_url,          # Input from URL
+            '-vn',                     # No video
+            '-acodec', 'pcm_s16le',    # PCM 16-bit
+            '-ar', '16000',            # 16kHz (speech model compatible)
+            '-ac', '1',                # Mono
+            '-y',
+            audio_path
+        ]
+        
+        result = subprocess.run(cmd, capture_output=True, timeout=60)
+        
+        if result.returncode != 0:
+            print(f"Audio extraction stderr: {result.stderr.decode()[:300]}")
+            
+        return audio_path if os.path.exists(audio_path) else None
+        
+    except Exception as e:
+        print(f"Audio streaming error: {e}")
+        return None
+
+
+# ============================================================
+# LEGACY FUNCTIONS (kept for backward compatibility / fallback)
+# ============================================================
+
+def download_video(url: str, job_id: str) -> str:
+    """
+    LEGACY: Download video to temp directory.
+    Use stream_extract_frames() instead to avoid disk usage.
+    """
+    url = normalize_url(url)
+    ensure_temp_dir()
+    output_path = os.path.join(TEMP_DIR, f"{job_id}.mp4")
+    
+    ydl_opts = {
+        'quiet': True,
+        'no_warnings': True,
+        'outtmpl': output_path,
+        'noplaylist': True,
+        'geo_bypass': True,
+        'retries': 3,
+        'fragment_retries': 3,
+        'extractor_args': {
+            'youtube': {
+                'player_client': ['android'],
+                'geo_bypass_country': ['US']
+            }
+        },
+        'download_ranges': lambda info, ydl: [{'start_time': 0, 'end_time': 30}],
+        'force_keyframes_at_cuts': True,
+        'format': 'bestvideo[ext=mp4][height<=720]+bestaudio[ext=m4a]/best[ext=mp4]/best',
+        'merge_output_format': 'mp4',
+        'http_headers': {
+            'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64)',
+        },
+    }
+    
+    try:
+        with yt_dlp.YoutubeDL(ydl_opts) as ydl:
+            ydl.download([url])
+        return output_path if os.path.exists(output_path) else None
+    except Exception as e:
+        print(f"yt-dlp download error: {e}")
+        return None
+
+
+def extract_frames(video_path: str, job_id: str, fps: float = 0.5, max_frames: int = 5) -> list:
+    """
+    LEGACY: Extract frames from a local video file.
+    Use stream_extract_frames() instead to avoid disk usage.
+    """
+    ensure_temp_dir()
+    frame_pattern = os.path.join(TEMP_DIR, f"{job_id}_frame_%03d.jpg")
+    
+    try:
+        cmd = [
+            'ffmpeg', '-i', video_path,
+            '-vf', f'fps={fps}',
+            '-frames:v', str(max_frames),
+            '-q:v', '2',
+            frame_pattern,
+            '-y'
+        ]
+        
+        subprocess.run(cmd, capture_output=True, timeout=60)
+        frames = sorted(glob.glob(os.path.join(TEMP_DIR, f"{job_id}_frame_*.jpg")))
+        return frames
+        
+    except subprocess.TimeoutExpired:
+        print("FFmpeg timeout")
+        return []
+    except Exception as e:
+        print(f"FFmpeg error: {e}")
+        return []
+
+
+def extract_audio(video_path: str, job_id: str) -> str:
+    """
+    LEGACY: Extract audio from a local video file.
+    Use stream_extract_audio() instead.
+    """
+    ensure_temp_dir()
+    audio_path = os.path.join(TEMP_DIR, f"{job_id}_audio.wav")
+    
+    try:
+        cmd = [
+            'ffmpeg', '-i', video_path,
+            '-vn',
+            '-acodec', 'pcm_s16le',
+            '-ar', '16000',
+            '-ac', '1',
+            audio_path,
+            '-y'
+        ]
+        
+        subprocess.run(cmd, capture_output=True, timeout=60)
+        return audio_path if os.path.exists(audio_path) else None
+        
+    except Exception as e:
+        print(f"Audio extraction error: {e}")
+        return None

app/services/hf_inference.py

@@ -0,0 +1,216 @@
+# C:\Users\bahae\.gemini\antigravity\scratch\verivid-ai\backend\app\services\hf_inference.py
+import os
+import requests
+from app.core.config import settings
+
+# Fallback HuggingFace models
+MODELS = [("Organika/sdxl-detector", ["artificial", "ai", "synthetic"])]
+
+def call_hf_model(model_name: str, image_bytes: bytes, ai_labels: list) -> float:
+    """Call HuggingFace model as fallback"""
+    if not settings.HF_TOKEN:
+        return None
+    
+    headers = {
+        "Authorization": f"Bearer {settings.HF_TOKEN}",
+        "Content-Type": "image/jpeg",
+    }
+    
+    try:
+        url = f"https://router.huggingface.co/hf-inference/models/{model_name}"
+        response = requests.post(url, headers=headers, data=image_bytes, timeout=45)
+        
+        if response.status_code != 200:
+            return None
+        if response.text.startswith('<!doctype'):
+            return None
+        
+        result = response.json()
+        
+        for item in result:
+            label = str(item.get('label', '')).lower()
+            score = float(item.get('score', 0))
+            
+            for ai_label in ai_labels:
+                if ai_label in label:
+                    return score
+            if 'human' in label or 'real' in label:
+                return 1 - score
+        return 0
+    except:
+        return None
+
+def analyze_visual_fallback(frame_paths: list) -> dict:
+    """Fallback visual analysis using HuggingFace"""
+    scores = []
+    
+    for path in frame_paths[:3]:  # Only 3 frames for fallback
+        try:
+            with open(path, 'rb') as f:
+                img_bytes = f.read()
+            
+            for model_name, ai_labels in MODELS:
+                score = call_hf_model(model_name, img_bytes, ai_labels)
+                if score is not None:
+                    scores.append(score)
+                    break
+        except:
+            continue
+    
+    if scores:
+        return {
+            "avg_prob": sum(scores) / len(scores),
+            "max_prob": max(scores),
+            "frame_count": len(scores),
+            "details": f"HuggingFace fallback: {len(scores)} frames analyzed"
+        }
+    return {"avg_prob": 0, "max_prob": 0, "frame_count": 0, "details": "Fallback failed"}
+
+def analyze_audio_ai(file_path: str, audio_path: str = None):
+    """
+    Real audio analysis for deepfake/synthetic speech detection.
+    Uses HuggingFace audio classification models.
+    """
+    if not audio_path or not os.path.exists(audio_path):
+        return {"spoof_prob": 0, "details": "No audio track.", "confidence": "high"}
+    
+    audio_size = os.path.getsize(audio_path)
+    if audio_size < 1000:
+        return {"spoof_prob": 0.1, "details": "Silent or very short audio.", "confidence": "low"}
+    
+    if not settings.HF_TOKEN:
+        # Fallback to heuristic analysis without API
+        return _analyze_audio_heuristic(audio_path, audio_size)
+    
+    # Try HuggingFace audio deepfake detection model
+    try:
+        with open(audio_path, 'rb') as f:
+            audio_bytes = f.read()
+        
+        # Use speech/audio classification model for deepfake detection
+        # Model: microsoft/wavlm-base-plus-sv (speaker verification - good for detecting synthetic)
+        # Alternative: facebook/wav2vec2-base for general audio analysis
+        headers = {
+            "Authorization": f"Bearer {settings.HF_TOKEN}",
+            "Content-Type": "audio/wav",
+        }
+        
+        # Try audio classification
+        model_url = "https://router.huggingface.co/hf-inference/models/facebook/wav2vec2-base-960h"
+        response = requests.post(model_url, headers=headers, data=audio_bytes, timeout=30)
+        
+        if response.status_code == 200 and not response.text.startswith('<!doctype'):
+            result = response.json()
+            # Analyze response for synthetic speech indicators
+            return _parse_audio_result(result, audio_size)
+        else:
+            # API failed, use heuristic
+            return _analyze_audio_heuristic(audio_path, audio_size)
+            
+    except Exception as e:
+        print(f"Audio HF inference error: {e}")
+        return _analyze_audio_heuristic(audio_path, audio_size)
+
+
+def _analyze_audio_heuristic(audio_path: str, audio_size: int) -> dict:
+    """
+    Heuristic audio analysis when API is unavailable.
+    Analyzes file characteristics as proxy signals.
+    """
+    import wave
+    
+    try:
+        with wave.open(audio_path, 'rb') as wav:
+            framerate = wav.getframerate()
+            nchannels = wav.getnchannels()
+            nframes = wav.getnframes()
+            duration = nframes / framerate if framerate > 0 else 0
+            
+            signals = []
+            spoof_prob = 0.0
+            
+            # Check for TTS-typical characteristics
+            # Many TTS systems output at exactly 16kHz or 22050Hz
+            if framerate in [16000, 22050, 24000]:
+                spoof_prob += 0.15
+                signals.append(f"TTS-common sample rate ({framerate}Hz)")
+            
+            # Very short duration with speech suggests clip
+            if 0 < duration < 3:
+                spoof_prob += 0.1
+                signals.append("Very short audio clip")
+            
+            # Mono audio is common in TTS
+            if nchannels == 1:
+                spoof_prob += 0.05
+                signals.append("Mono audio (common in TTS)")
+            
+            # File size vs duration ratio (synthetic often has consistent bitrate)
+            if duration > 0:
+                kb_per_second = (audio_size / 1024) / duration
+                if 28 < kb_per_second < 35:  # Very consistent 256kbps range
+                    spoof_prob += 0.1
+                    signals.append("Uniform bitrate pattern")
+            
+            # Cap probability
+            spoof_prob = min(spoof_prob, 0.5)
+            
+            details = "; ".join(signals) if signals else f"Audio analyzed ({audio_size // 1024}KB, {duration:.1f}s)"
+            confidence = "medium" if signals else "low"
+            
+            return {
+                "spoof_prob": round(spoof_prob, 2),
+                "details": details,
+                "confidence": confidence,
+                "duration_s": round(duration, 1),
+                "sample_rate": framerate
+            }
+            
+    except Exception as e:
+        # Can't parse as WAV, return minimal info
+        return {
+            "spoof_prob": 0.15,
+            "details": f"Audio present but unparseable ({audio_size // 1024}KB)",
+            "confidence": "low"
+        }
+
+
+def _parse_audio_result(result, audio_size: int) -> dict:
+    """Parse HuggingFace audio model response."""
+    # wav2vec2 returns transcription, not classification
+    # We use transcription quality as a proxy signal
+    
+    if isinstance(result, dict) and 'text' in result:
+        text = result.get('text', '').strip()
+        
+        if not text:
+            return {
+                "spoof_prob": 0.1,
+                "details": "No speech detected in audio.",
+                "confidence": "medium"
+            }
+        
+        # Very clean transcription can indicate TTS (natural speech has more disfluencies)
+        word_count = len(text.split())
+        
+        # Short, clean phrases are more likely TTS
+        if word_count < 5 and len(text) > 10:
+            return {
+                "spoof_prob": 0.25,
+                "details": f"Short clear speech detected: '{text[:50]}...'",
+                "confidence": "medium",
+                "transcript_preview": text[:100]
+            }
+        
+        return {
+            "spoof_prob": 0.15,
+            "details": f"Speech detected ({word_count} words).",
+            "confidence": "medium",
+            "transcript_preview": text[:100]
+        }
+    
+    return {
+        "spoof_prob": 0.15,
+        "details": f"Audio analyzed ({audio_size // 1024}KB).",
+        "confidence": "low"
+    }

app/services/local_signals.py

@@ -0,0 +1,123 @@
+# C:\Users\bahae\.gemini\antigravity\scratch\verivid-ai\backend\app\services\local_signals.py
+import subprocess
+import json
+import os
+
+def get_video_metadata(file_path: str) -> dict:
+    """Extract detailed metadata using ffprobe"""
+    if not file_path or not os.path.exists(file_path):
+        return {}
+    
+    try:
+        cmd = [
+            'ffprobe', '-v', 'quiet',
+            '-print_format', 'json',
+            '-show_format', '-show_streams',
+            file_path
+        ]
+        result = subprocess.run(cmd, capture_output=True, text=True, timeout=30)
+        return json.loads(result.stdout) if result.stdout else {}
+    except:
+        return {}
+
+def analyze_metadata(file_path: str, video_info: dict = None):
+    """
+    Step 2: Local Signals (Metadata Analysis)
+    """
+    risk_score = 0
+    flags = []
+    
+    # If we have yt-dlp info
+    if video_info:
+        # Check for unusual resolution
+        width = video_info.get('width', 0)
+        height = video_info.get('height', 0)
+        
+        if width and height:
+            # 9:16 ratio with very high res might be suspicious for UGC
+            if width > 1080 and height > 1920:
+                risk_score += 0.1
+                flags.append("Unusually high resolution for short-form content")
+        
+        # Check FPS
+        fps = video_info.get('fps', 0)
+        if fps and fps > 30:
+            risk_score += 0.05
+            flags.append(f"High framerate ({fps}fps) uncommon for UGC")
+        
+        # Check codec
+        vcodec = video_info.get('vcodec', '')
+        if vcodec and 'av1' in vcodec.lower():
+            risk_score += 0.1
+            flags.append("AV1 codec often used by AI rendering tools")
+    
+    # If we have the actual file, do deeper analysis
+    if file_path and os.path.exists(file_path):
+        meta = get_video_metadata(file_path)
+        
+        if meta:
+            fmt = meta.get('format', {})
+            
+            # Check for missing encoder info (stripped metadata)
+            tags = fmt.get('tags', {})
+            if not tags.get('encoder') and not tags.get('creation_time'):
+                risk_score += 0.15
+                flags.append("Missing encoder/creation metadata (possibly stripped)")
+            
+            # Check bitrate
+            bitrate = int(fmt.get('bit_rate', 0))
+            if bitrate > 0:
+                # Very low bitrate for resolution = re-encoding
+                duration = float(fmt.get('duration', 0))
+                size = int(fmt.get('size', 0))
+                if duration > 0 and size > 0:
+                    expected_bitrate = size * 8 / duration
+                    if expected_bitrate < 500000:  # Less than 500kbps
+                        risk_score += 0.1
+                        flags.append("Low bitrate suggests heavy re-encoding")
+    
+    # Default if no issues found
+    if not flags:
+        flags.append("No metadata anomalies detected")
+    
+    return {
+        "risk_score": min(risk_score, 1.0),  # Cap at 1.0
+        "flags": flags,
+        "details": "; ".join(flags)
+    }
+
+def analyze_heuristics(file_path: str, meta: dict, video_info: dict = None):
+    """
+    Step 5: Heuristic Analysis
+    """
+    risk_score = 0
+    flags = []
+    
+    # Platform-quality mismatch
+    if video_info:
+        # TikTok with 4K = suspicious
+        width = video_info.get('width', 0)
+        if width and width >= 2160:
+            risk_score += 0.2
+            flags.append("Studio-quality resolution unusual for platform")
+        
+        # Very short video with high production value
+        duration = video_info.get('duration', 0)
+        if duration and duration < 15 and width and width > 1080:
+            risk_score += 0.1
+            flags.append("Short clip with high production quality")
+    
+    # Check for consistent frame rate (from metadata)
+    if meta and 'flags' in meta:
+        if 'Low bitrate' in str(meta.get('flags', [])):
+            risk_score += 0.05
+            flags.append("Compression artifacts may hide manipulation")
+    
+    if not flags:
+        flags.append("No heuristic red flags")
+    
+    return {
+        "risk_score": min(risk_score, 1.0),
+        "flags": flags,
+        "details": "; ".join(flags)
+    }

app/services/pipeline.py

@@ -0,0 +1,226 @@
+# C:\Users\bahae\.gemini\antigravity\scratch\verivid-ai\backend\app\services\pipeline.py
+"""
+Analysis Pipeline with Zero-Storage Streaming
+==============================================
+For URL-based analysis: Uses streaming to avoid saving full video files.
+For uploaded files: Uses traditional file-based processing.
+"""
+
+import os
+import json
+import hashlib
+from datetime import datetime
+
+from app.services.downloader import (
+    get_video_info, 
+    clean_temp,
+    # Streaming functions (zero storage)
+    stream_extract_frames,
+    stream_extract_audio,
+    # Legacy functions (for uploaded files)
+    extract_frames,
+    extract_audio
+)
+from app.services.local_signals import analyze_metadata, analyze_heuristics
+from app.services.sightengine import analyze_frames_with_sightengine
+from app.services.hf_inference import analyze_visual_fallback, analyze_audio_ai
+from app.core.scoring import calculate_risk
+
+# Cache
+CACHE_DIR = os.path.join(os.path.dirname(__file__), '..', '..', 'cache')
+
+def get_cache_key(url: str) -> str:
+    return hashlib.md5(url.encode()).hexdigest()
+
+def get_cached_result(url: str):
+    os.makedirs(CACHE_DIR, exist_ok=True)
+    cache_file = os.path.join(CACHE_DIR, f"{get_cache_key(url)}.json")
+    if os.path.exists(cache_file):
+        try:
+            with open(cache_file, 'r') as f:
+                data = json.load(f)
+            cached_time = datetime.fromisoformat(data.get('cached_at', '2000-01-01'))
+            if (datetime.now() - cached_time).total_seconds() < 86400:
+                return data.get('result')
+        except:
+            pass
+    return None
+
+def save_to_cache(url: str, result: dict):
+    os.makedirs(CACHE_DIR, exist_ok=True)
+    cache_file = os.path.join(CACHE_DIR, f"{get_cache_key(url)}.json")
+    try:
+        with open(cache_file, 'w') as f:
+            json.dump({'cached_at': datetime.now().isoformat(), 'url': url, 'result': result}, f)
+    except:
+        pass
+
+
+async def run_analysis_pipeline(job_id: str, url: str, uploaded_file_path: str, jobs_db: dict):
+    """
+    Main analysis pipeline with ZERO-STORAGE streaming for URL analysis.
+    
+    For URLs: Streams video directly from platform → ffmpeg → frames (no video saved to disk)
+    For uploads: Uses traditional file-based processing
+    """
+    print(f"[{job_id}] Starting analysis for URL: {url}")
+    jobs_db[job_id]["status"] = "processing"
+    
+    try:
+        # Check cache
+        if url:
+            cached = get_cached_result(url)
+            if cached:
+                print(f"[{job_id}] Cache hit!")
+                cached['id'] = job_id
+                jobs_db[job_id] = {"status": "completed", "result": cached}
+                return
+        
+        # Get video info (does not download)
+        video_info = None
+        if url:
+            print(f"[{job_id}] Fetching video info...")
+            video_info = get_video_info(url)
+        if not video_info:
+            video_info = {"thumbnail": None, "title": "Unknown"}
+        
+        frame_paths = []
+        audio_path = None
+        video_path = None  # Only set for uploaded files
+        
+        # ============================================
+        # PATH A: URL-based analysis (try streaming first, fallback to download)
+        # ============================================
+        if url and not uploaded_file_path:
+            print(f"[{job_id}] STREAMING MODE: Attempting to extract frames directly from URL...")
+            frame_paths = stream_extract_frames(url, job_id, max_frames=5, duration=30)
+            
+            # If streaming failed, fallback to traditional download
+            if not frame_paths:
+                print(f"[{job_id}] Streaming failed, falling back to traditional download...")
+                from app.services.downloader import download_video
+                video_path = download_video(url, job_id)
+                
+                if video_path and os.path.exists(video_path):
+                    print(f"[{job_id}] Downloaded video, extracting frames...")
+                    frame_paths = extract_frames(video_path, job_id, fps=0.5, max_frames=5)
+                    
+                    if frame_paths:
+                        print(f"[{job_id}] Extracted {len(frame_paths)} frames via fallback")
+                        audio_path = extract_audio(video_path, job_id)
+                    else:
+                        jobs_db[job_id] = {"status": "failed", "error": "Could not extract frames from video"}
+                        print(f"[{job_id}] Failed: fallback extraction also failed")
+                        return
+                else:
+                    jobs_db[job_id] = {"status": "failed", "error": "Could not download video from URL"}
+                    print(f"[{job_id}] Failed: download failed")
+                    return
+            else:
+                print(f"[{job_id}] Streaming success! Extracted {len(frame_paths)} frames")
+                print(f"[{job_id}] Extracting audio via streaming...")
+                audio_path = stream_extract_audio(url, job_id, duration=30)
+        
+        # ============================================
+        # PATH B: Uploaded file (traditional processing)
+        # ============================================
+        elif uploaded_file_path and os.path.exists(uploaded_file_path):
+            print(f"[{job_id}] FILE MODE: Processing uploaded file...")
+            video_path = uploaded_file_path
+            
+            print(f"[{job_id}] Extracting frames from file...")
+            frame_paths = extract_frames(video_path, job_id, fps=0.5, max_frames=5)
+            
+            if not frame_paths:
+                jobs_db[job_id] = {"status": "failed", "error": "No frames extracted from uploaded file"}
+                print(f"[{job_id}] Failed: 0 frames extracted from upload")
+                return
+            
+            print(f"[{job_id}] Extracted {len(frame_paths)} frames from file")
+            
+            print(f"[{job_id}] Extracting audio from file...")
+            audio_path = extract_audio(video_path, job_id)
+        
+        else:
+            jobs_db[job_id] = {"status": "failed", "error": "No URL or file provided"}
+            print(f"[{job_id}] Failed: no input provided")
+            return
+        
+        # ============================================
+        # ANALYSIS (same for both paths)
+        # ============================================
+        
+        # PRIMARY: SightEngine Analysis
+        from app.core.config import settings
+        se_configured = bool(settings.SIGHTENGINE_API_USER and settings.SIGHTENGINE_API_SECRET)
+        print(f"[{job_id}] Running SightEngine analysis... configured={se_configured}")
+        sightengine_result = analyze_frames_with_sightengine(frame_paths)
+        
+        # Build visual result
+        if sightengine_result.get("avg_score") is not None:
+            visual = {
+                "avg_prob": sightengine_result["avg_score"],
+                "max_prob": sightengine_result["max_score"],
+                "frame_count": sightengine_result["frame_count"],
+                "frame_scores": sightengine_result["frame_scores"],
+                "details": sightengine_result["details"],
+                "source": "SightEngine"
+            }
+        else:
+            # FALLBACK: HuggingFace
+            print(f"[{job_id}] SightEngine failed or not configured, using HuggingFace fallback...")
+            fallback = analyze_visual_fallback(frame_paths)
+            visual = {
+                "avg_prob": fallback["avg_prob"],
+                "max_prob": fallback["max_prob"],
+                "frame_count": fallback["frame_count"],
+                "frame_scores": [],
+                "details": fallback["details"],
+                "source": "HuggingFace (fallback)"
+            }
+        
+        print(f"[{job_id}] Running audio analysis...")
+        audio = analyze_audio_ai(video_path, audio_path=audio_path)
+        
+        print(f"[{job_id}] Running metadata analysis...")
+        # For streaming mode, we don't have a video file, so use video_info
+        meta = analyze_metadata(video_path, video_info=video_info)
+        
+        print(f"[{job_id}] Running heuristics...")
+        heuristics = analyze_heuristics(video_path, meta, video_info=video_info)
+        
+        # Calculate score
+        signals = {"visual": visual, "audio": audio, "metadata": meta, "heuristics": heuristics}
+        score, confidence, rec = calculate_risk(signals)
+        
+        # Build result
+        result = {
+            "score": score,
+            "confidence": confidence,
+            "recommendation": rec,
+            "signals": signals,
+            "video_info": {
+                "title": video_info.get("title"),
+                "duration": video_info.get("duration"),
+                "resolution": f"{video_info.get('width', '?')}x{video_info.get('height', '?')}",
+                "frames_analyzed": len(frame_paths)
+            },
+            "explanation": f"Analyzed {len(frame_paths)} frames using {visual.get('source', 'AI')}. Risk score: {score}/100 ({rec}). {confidence} confidence.",
+            "disclaimer": "This assessment estimates the likelihood of AI generation. It does not guarantee absolute authenticity."
+        }
+        
+        # Cache and cleanup
+        if url:
+            save_to_cache(url, result)
+        clean_temp(job_id)
+        
+        result['id'] = job_id
+        jobs_db[job_id] = {"status": "completed", "result": result}
+        print(f"[{job_id}] Completed: {score}/100 ({rec})")
+        
+    except Exception as e:
+        print(f"[{job_id}] Failed: {e}")
+        import traceback
+        traceback.print_exc()
+        jobs_db[job_id] = {"status": "failed", "error": str(e)}
+        clean_temp(job_id)

app/services/sightengine.py

@@ -0,0 +1,99 @@
+# C:\Users\bahae\.gemini\antigravity\scratch\verivid-ai\backend\app\services\sightengine.py
+import requests
+from app.core.config import settings
+
+SIGHTENGINE_CHECK_URL = "https://api.sightengine.com/1.0/check.json"
+
+def analyze_with_sightengine(image_url: str = None, image_bytes: bytes = None) -> dict:
+    """
+    Use SightEngine's professional AI detection.
+    Returns: {"ai_score": 0-1, "details": str, "raw": dict}
+    """
+    if not settings.SIGHTENGINE_API_USER or not settings.SIGHTENGINE_API_SECRET:
+        return {"ai_score": None, "details": "SightEngine not configured", "raw": None}
+    
+    try:
+        if image_url:
+            # URL-based check
+            response = requests.post(
+                SIGHTENGINE_CHECK_URL,
+                data={
+                    "url": image_url,
+                    "models": "genai",
+                    "api_user": settings.SIGHTENGINE_API_USER,
+                    "api_secret": settings.SIGHTENGINE_API_SECRET
+                },
+                timeout=30
+            )
+        elif image_bytes:
+            # File-based check
+            response = requests.post(
+                SIGHTENGINE_CHECK_URL,
+                data={
+                    "models": "genai",
+                    "api_user": settings.SIGHTENGINE_API_USER,
+                    "api_secret": settings.SIGHTENGINE_API_SECRET
+                },
+                files={"media": ("image.jpg", image_bytes, "image/jpeg")},
+                timeout=30
+            )
+        else:
+            return {"ai_score": None, "details": "No image provided", "raw": None}
+        
+        if response.status_code != 200:
+            return {"ai_score": None, "details": f"API error: {response.status_code}", "raw": response.text[:200]}
+        
+        data = response.json()
+        
+        # SightEngine returns: {"type": {"ai_generated": 0.95, ...}}
+        if data.get("status") == "success":
+            genai_data = data.get("type", {})
+            ai_score = genai_data.get("ai_generated", 0)
+            
+            return {
+                "ai_score": ai_score,
+                "details": f"SightEngine AI detection: {round(ai_score * 100)}% AI probability",
+                "raw": data
+            }
+        else:
+            return {"ai_score": None, "details": f"API error: {data.get('error', {}).get('message', 'Unknown')}", "raw": data}
+            
+    except Exception as e:
+        return {"ai_score": None, "details": f"Exception: {str(e)}", "raw": None}
+
+def analyze_frames_with_sightengine(frame_paths: list) -> dict:
+    """Analyze multiple frames and aggregate scores"""
+    scores = []
+    details = []
+    
+    for path in frame_paths[:5]:  # Limit to 5 frames to save API calls
+        try:
+            with open(path, 'rb') as f:
+                img_bytes = f.read()
+            
+            result = analyze_with_sightengine(image_bytes=img_bytes)
+            
+            if result["ai_score"] is not None:
+                scores.append(result["ai_score"])
+                details.append(f"Frame: {round(result['ai_score'] * 100)}%")
+        except Exception as e:
+            details.append(f"Error: {str(e)[:50]}")
+    
+    if scores:
+        avg_score = sum(scores) / len(scores)
+        max_score = max(scores)
+        return {
+            "avg_score": avg_score,
+            "max_score": max_score,
+            "frame_count": len(scores),
+            "frame_scores": [round(s, 3) for s in scores],
+            "details": f"SightEngine analyzed {len(scores)} frames. Avg: {round(avg_score*100)}%, Max: {round(max_score*100)}%"
+        }
+    else:
+        return {
+            "avg_score": None,
+            "max_score": None,
+            "frame_count": 0,
+            "frame_scores": [],
+            "details": "SightEngine analysis failed: " + "; ".join(details)
+        }

main.py

@@ -0,0 +1,76 @@
+# C:\Users\bahae\.gemini\antigravity\scratch\verivid-ai\backend\main.py
+from fastapi import FastAPI, Request
+from fastapi.middleware.cors import CORSMiddleware
+from fastapi.responses import JSONResponse
+from slowapi import Limiter, _rate_limit_exceeded_handler
+from slowapi.util import get_remote_address
+from slowapi.errors import RateLimitExceeded
+from app.api import routes
+
+# Rate limiter
+limiter = Limiter(key_func=get_remote_address)
+
+app = FastAPI(
+    title="VeriVid Risk Engine",
+    version="1.0.0",
+    description="AI Video Authenticity Detection API",
+    docs_url="/docs" if True else None,  # Disable in production
+    redoc_url=None
+)
+
+# Add rate limiter
+app.state.limiter = limiter
+app.add_exception_handler(RateLimitExceeded, _rate_limit_exceeded_handler)
+
+# CORS - restrict origins in production
+ALLOWED_ORIGINS = [
+    "http://localhost:3000",
+    "http://127.0.0.1:3000",
+    "https://verivid.ai",
+    "https://www.verivid.ai",
+    "https://verivid-ai.vercel.app",
+    "https://verivid-ai-final.vercel.app",
+    "https://verivid-ai-final.onrender.com",
+    "https://verivid-ai-final-1.onrender.com",
+]
+
+# For development/preview, also allow Vercel preview URLs
+import re
+def is_allowed_origin(origin: str) -> bool:
+    if origin in ALLOWED_ORIGINS:
+        return True
+    # Allow ANY Vercel deployment (covers all preview URLs)
+    if re.match(r"https://.*\.vercel\.app", origin):
+        return True
+    return False
+
+app.add_middleware(
+    CORSMiddleware,
+    allow_origins=ALLOWED_ORIGINS,
+    # Allow ANY vercel.app subdomain (covers preview deployments)
+    allow_origin_regex=r"https://.*\.vercel\.app",
+    allow_credentials=True,
+    allow_methods=["GET", "POST", "OPTIONS"],
+    allow_headers=["*"],
+)
+
+# Security headers middleware
+@app.middleware("http")
+async def add_security_headers(request: Request, call_next):
+    response = await call_next(request)
+    response.headers["X-Content-Type-Options"] = "nosniff"
+    response.headers["X-Frame-Options"] = "DENY"
+    response.headers["X-XSS-Protection"] = "1; mode=block"
+    response.headers["Referrer-Policy"] = "strict-origin-when-cross-origin"
+    return response
+
+# Include routes
+app.include_router(routes.router, prefix="/api/v1")
+
+@app.get("/")
+def health():
+    return {"status": "ok", "service": "VeriVid Engine", "version": "1.0.0"}
+
+@app.get("/health")
+def health_check():
+    return {"status": "healthy"}

requirements.txt

@@ -0,0 +1,8 @@
+fastapi>=0.115.0
+uvicorn[standard]>=0.32.0
+python-dotenv>=1.0.0
+pydantic-settings>=2.6.0
+requests>=2.32.0
+yt-dlp>=2024.12.0
+slowapi>=0.1.9
+python-multipart>=0.0.9