Final Deploy

.gitignore

@@ -0,0 +1,5 @@
+__pycache__/
+*.pyc
+venv/
+.env
+.DS_Store

app/api/routes.py

@@ -1,8 +1,11 @@
-from fastapi import APIRouter, UploadFile, File, Form, HTTPException
+from fastapi import APIRouter, UploadFile, File, Form, HTTPException, Request
 from typing import Optional
 import logging
-from app.services.detector import analysis_service
+from fastapi.concurrency import run_in_threadpool
+from app.services.detector import analysis_service as text_service
 from app.services.ocr import ocr_service
+from app.services.image_detector import image_analysis_service
+from app.services.video_detector import video_analysis_service
 from app.models.schemas import AnalysisResponse, HealthResponse
 
 router = APIRouter()
@@ -10,50 +13,82 @@ logger = logging.getLogger("uvicorn")
 
 @router.get("/health", response_model=HealthResponse)
 async def health_check():
-    return {"status": "ok", "message": "VerifAI Backend is running"}
+    return {"status": "ok", "message": "VerifAI Backend is running (Text, Image, Video)"}
 
 @router.post("/analyze", response_model=AnalysisResponse)
 async def analyze(
+    request: Request,
     text: Optional[str] = Form(None),
     file: Optional[UploadFile] = File(None)
 ):
     """
-    Main analysis endpoint.
-    Accepts 'text' (Form data) OR 'file' (UploadFile).
+    Multi-Modal Analysis Endpoint.
     """
-    content_to_analyze = ""
-    source_type = "text"
-    
     try:
+        # Debug Logging
+        logger.info(f"Analyze Request: Content-Type={request.headers.get('content-type')}")
+        
+        # 1. Direct Text Input
         if text:
-            content_to_analyze = text
-            logger.info("Received text analysis request")
-        elif file:
-            logger.info(f"Received file analysis request: {file.filename}")
-            source_type = "image"
-            # Read file bytes
-            contents = await file.read()
-            # Run OCR
-            extracted_text = ocr_service.extract_text(contents)
-            if not extracted_text:
-                logger.warning("OCR failed to extract text")
-                raise HTTPException(status_code=400, detail="Could not extract text from image")
-            content_to_analyze = extracted_text
-            logger.info(f"OCR Success. Extracted {len(content_to_analyze)} chars")
-        else:
-            raise HTTPException(status_code=400, detail="No text or file provided")
+            logger.info("Analyzing Text Input")
+            # Run text analysis in threadpool to avoid blocking event loop
+            return await run_in_threadpool(text_service.analyze, text)
 
-        if len(content_to_analyze.split()) < 5:
-            raise HTTPException(status_code=400, detail="Text too short for analysis (min 5 words)")
+        # 2. File Input
+        if file:
+            content_type = file.content_type or ""
+            logger.info(f"Analyzing File: {file.filename} ({content_type})")
+            
+            file_bytes = await file.read()
+            
+            # --- VIDEO ANALYSIS ---
+            if "video" in content_type:
+                logger.info("Running Video Analysis")
+                result = await run_in_threadpool(video_analysis_service.analyze_video, file_bytes)
+                return {
+                    "score": result['score'],
+                    "verdict": result['verdict'],
+                    "media_details": {"type": "video", "frames": result.get('frame_details')}
+                }
 
-        # Run Analysis
-        result = analysis_service.analyze(content_to_analyze)
-        logger.info(f"Analysis Complete. Score: {result['score']}")
-        
-        return result
+            # --- IMAGE ANALYSIS (Dual Mode: AI Check + OCR) ---
+            if "image" in content_type:
+                logger.info("Running Image Analysis")
+                
+                # A. Check if Image is AI-Generated (Threadpool)
+                image_result = await run_in_threadpool(image_analysis_service.analyze_image, file_bytes)
+                logger.info(f"Image AI Score: {image_result['score']}")
+                
+                # If heavily AI, return immediately
+                if image_result['score'] > 80:
+                     return {
+                        "score": image_result['score'],
+                        "verdict": "AI-Generated Image",
+                        "media_details": {"type": "image", "detail": "Image detected as AI generated"}
+                    }
+                
+                # B. OCR Check (Threadpool)
+                extracted_text = await run_in_threadpool(ocr_service.extract_text, file_bytes)
+                
+                if extracted_text and len(extracted_text.split()) > 5:
+                    logger.info("OCR found text, analyzing integrity...")
+                    text_result = await run_in_threadpool(text_service.analyze, extracted_text)
+                    
+                    text_result["media_details"] = {
+                        "type": "image_ocr", 
+                        "image_ai_score": image_result['score']
+                    }
+                    return text_result
+                
+                # C. No text found? Return Image Analysis
+                return {
+                    "score": image_result['score'],
+                    "verdict": image_result['verdict'] if image_result['score'] > 50 else "Real/No Text",
+                    "media_details": {"type": "image", "detail": "No readable text found"}
+                }
+
+        raise HTTPException(status_code=400, detail="No text or valid file provided")
 
-    except HTTPException as he:
-        raise he
     except Exception as e:
         logger.error(f"Analysis Error: {str(e)}")
-        raise HTTPException(status_code=500, detail="Internal Server Error during analysis")
+        raise HTTPException(status_code=500, detail=str(e))

app/core/config.py

@@ -1,14 +1,25 @@
 import os
+from pydantic_settings import BaseSettings
 
-class Settings:
+class Settings(BaseSettings):
     PROJECT_NAME: str = "VerifAI Backend"
     API_V1_STR: str = "/api/v1"
     
-    # Models
-    DETECTOR_MODEL_NAME: str = "Hello-SimpleAI/chatgpt-detector-roberta" 
+    # Cloud Deployment Configuration
+    # We use '0.0.0.0' for binding to all interfaces in Docker
+    HOST: str = "0.0.0.0" 
+    # Port 7860 is the default for Hugging Face Spaces
+    PORT: int = int(os.environ.get("PORT", 7860))
+
+    # AI Models
+    # Text: Fakespot-AI model (RoBERTa-base finetuned)
+    TEXT_MODEL_NAME: str = "fakespot-ai/roberta-base-ai-text-detection-v1"
     METRIC_MODEL_NAME: str = "gpt2"
     
-    # OCR
-    OCR_LANGUAGES: list = ['en']
+    # Image: Distilled model for detecting AI generated images
+    IMAGE_MODEL_NAME: str = "umm-maybe/AI-image-detector"
+    
+    class Config:
+        case_sensitive = True
 
 settings = Settings()

app/models/schemas.py

@@ -13,8 +13,10 @@ class Segment(BaseModel):
 class AnalysisResponse(BaseModel):
     score: float
     verdict: str
-    metrics: Metrics
-    segments: list[Segment]
+    metrics: Optional[Metrics] = None
+    segments: Optional[list[Segment]] = None
+    error: Optional[str] = None
+    media_details: Optional[dict] = None # For Image/Video specific details
 
 class HealthResponse(BaseModel):
     status: str

app/services/detector.py

@@ -4,22 +4,42 @@ import math
 from transformers import AutoModelForSequenceClassification, AutoTokenizer, AutoModelForCausalLM
 from app.core.config import settings
 
-class AnalysisService:
+class TextAnalysisService:
     def __init__(self):
-        print("Loading AI Models... This might take a moment.")
+        print("Loading Text AI Models... This might take a moment.")
         
-        # 1. Load Detector (Classification)
-        self.detector_tokenizer = AutoTokenizer.from_pretrained(settings.DETECTOR_MODEL_NAME)
-        self.detector_model = AutoModelForSequenceClassification.from_pretrained(settings.DETECTOR_MODEL_NAME)
+        # 1. Load Detector (Classification) - Fakespot-AI
+        try:
+            # Use custom cache to be safe
+            cache_dir = "/tmp/hf_cache"
+            
+            self.detector_tokenizer = AutoTokenizer.from_pretrained(
+                settings.TEXT_MODEL_NAME, 
+                cache_dir=cache_dir
+            )
+            self.detector_model = AutoModelForSequenceClassification.from_pretrained(
+                settings.TEXT_MODEL_NAME,
+                cache_dir=cache_dir
+            )
+            print(f"Model ID2LABEL: {self.detector_model.config.id2label}")
+        except Exception as e:
+            print(f"Error loading text detector: {e}")
+            raise e
         
         # 2. Load Metric Model (Perplexity - GPT2)
-        self.metric_tokenizer = AutoTokenizer.from_pretrained(settings.METRIC_MODEL_NAME)
-        self.metric_model = AutoModelForCausalLM.from_pretrained(settings.METRIC_MODEL_NAME)
+        try:
+            self.metric_tokenizer = AutoTokenizer.from_pretrained(settings.METRIC_MODEL_NAME)
+            self.metric_model = AutoModelForCausalLM.from_pretrained(settings.METRIC_MODEL_NAME)
+        except Exception as e:
+            print(f"Error loading metric model: {e}")
+            # Non-critical failure for metrics
+            self.metric_model = None
         
         self.device = "cuda" if torch.cuda.is_available() else "cpu"
         self.detector_model.to(self.device)
-        self.metric_model.to(self.device)
-        print(f"Models loaded on {self.device}")
+        if self.metric_model:
+            self.metric_model.to(self.device)
+        print(f"Text Models loaded on {self.device}")
 
     def calculate_perplexity(self, text):
         """
@@ -27,6 +47,9 @@ class AnalysisService:
         Lower perplexity = implementation of model training data = likely AI.
         Higher perplexity = more random/creative = likely Human.
         """
+        if not self.metric_model:
+            return 0.0
+
         encodings = self.metric_tokenizer(text, return_tensors="pt")
         input_ids = encodings.input_ids.to(self.device)
 
@@ -69,7 +92,9 @@ class AnalysisService:
             logits = outputs.logits
             probs = torch.softmax(logits, dim=1)
             
-        # Hello-SimpleAI/chatgpt-detector-roberta: Label 0 is "Human", Label 1 is "ChatGPT" (AI)
+        # Fakespot uses Label 0 = Real, Label 1 = Fake (AI)
+        # Our test showed "Messy Human" got 99% on index 0 (Real).
+        # So prob[1] is the AI Probability.
         ai_prob = probs[0][1].item() * 100
         
         return ai_prob
@@ -103,7 +128,7 @@ class AnalysisService:
         verdict = "Human"
         if ai_probability > 80:
             verdict = "AI-Generated"
-        elif ai_probability > 40:
+        elif ai_probability > 50: # Lower threshold slightly for this robust model
             verdict = "Mixed/Uncertain"
             
         return {
@@ -116,4 +141,4 @@ class AnalysisService:
             "segments": segments
         }
 
-analysis_service = AnalysisService()
+analysis_service = TextAnalysisService() # Keep variable name compatible for now or update routes

app/services/image_detector.py

@@ -0,0 +1,61 @@
+from transformers import pipeline
+from PIL import Image
+import io
+from app.core.config import settings
+
+class ImageAnalysisService:
+    def __init__(self):
+        print("Loading Image AI Model... This might take a moment.")
+        try:
+            # Load Image Classification Pipeline
+            # umm-maybe/AI-image-detector typically returns labels like "artificial" or "human"
+            self.classifier = pipeline("image-classification", model=settings.IMAGE_MODEL_NAME)
+            print("Image Model loaded successfully.")
+        except Exception as e:
+            print(f"Error loading image model: {e}")
+            self.classifier = None
+
+    def analyze_image(self, image_data: bytes):
+        if not self.classifier:
+            return {"score": 0, "verdict": "Error: Model not loaded"}
+
+        try:
+            # Convert bytes to PIL Image
+            image = Image.open(io.BytesIO(image_data)).convert("RGB")
+            
+            # Run prediction
+            results = self.classifier(image)
+            # Results are typically a list of dicts: [{'label': 'artificial', 'score': 0.99}, {'label': 'human', 'score': 0.01}]
+            
+            ai_score = 0.0
+            
+            # Parse results to find 'artificial' or equivalent label
+            for result in results:
+                label = result['label'].lower()
+                if label in ['artificial', 'ai', 'fake', 'generated']:
+                    ai_score = result['score'] * 100
+                    break
+                elif label in ['human', 'real']:
+                    # If label is explicitly human, the inverse is AI score (roughly)
+                    # But usually the classifier returns both, so we just look for the AI one.
+                    pass
+            
+            verdict = "Human"
+            if ai_score > 90:
+                verdict = "AI-Generated"
+            elif ai_score > 60:
+                verdict = "Likely AI"
+            elif ai_score > 40:
+                verdict = "Mixed/Uncertain"
+
+            return {
+                "score": round(ai_score, 2),
+                "verdict": verdict,
+                "details": results # Return raw details for debugging if needed
+            }
+
+        except Exception as e:
+            print(f"Image Analysis Error: {e}")
+            return {"score": 0, "verdict": "Analysis Failed"}
+
+image_analysis_service = ImageAnalysisService()

app/services/video_detector.py

@@ -0,0 +1,98 @@
+import cv2
+import os
+import tempfile
+import numpy as np
+from app.services.image_detector import image_analysis_service
+
+class VideoAnalysisService:
+    def __init__(self):
+        # Relies on Image Service
+        pass
+
+    def analyze_video(self, video_bytes: bytes):
+        """
+        Analyzes a video by extracting keyframes and checking them for AI content.
+        """
+        if not image_analysis_service.classifier:
+             return {"score": 0, "verdict": "Error: Image Model not loaded"}
+             
+        # 1. Save bytes to temp file for OpenCV
+        with tempfile.NamedTemporaryFile(delete=False, suffix='.mp4') as temp:
+            temp.write(video_bytes)
+            temp_path = temp.name
+            
+        try:
+            cap = cv2.VideoCapture(temp_path)
+            if not cap.isOpened():
+                return {"score": 0, "verdict": "Error: Could not open video"}
+                
+            total_frames = int(cap.get(cv2.CAP_PROP_FRAME_COUNT))
+            fps = cap.get(cv2.CAP_PROP_FPS)
+            
+            # 2. Extract Keyframes (Start, 25%, 50%, 75%, End)
+            # Limit analysis to at most 5 frames to save speed
+            sample_points = [0, 0.25, 0.5, 0.75, 0.95]
+            frames_to_check = []
+            
+            for point in sample_points:
+                frame_idx = int(point * total_frames)
+                cap.set(cv2.CAP_PROP_POS_FRAMES, frame_idx)
+                ret, frame = cap.read()
+                if ret:
+                    frames_to_check.append(frame)
+                    
+            cap.release()
+            
+            # 3. Analyze Frames
+            frame_scores = []
+            frame_details = []
+            
+            for i, frame in enumerate(frames_to_check):
+                # Convert BGR (OpenCV) to RGB (PIL/Transformers)
+                # We need to encode it back to bytes for the image service (or refactor image service to accept arrays)
+                # To verify keeping it simple, let's encode to jpg bytes
+                _, buffer = cv2.imencode('.jpg', frame)
+                jpg_as_text = buffer.tobytes()
+                
+                result = image_analysis_service.analyze_image(jpg_as_text)
+                frame_scores.append(result['score'])
+                frame_details.append({
+                    "frame_index": i,
+                    "score": result['score'],
+                    "verdict": result['verdict']
+                })
+                
+            # 4. Aggregate Results
+            if not frame_scores:
+                return {"score": 0, "verdict": "Could not extract frames"}
+            
+            # Use MAX score as the indicator. If one frame is clearly deepfake, the video is suspect.
+            max_score = max(frame_scores)
+            avg_score = sum(frame_scores) / len(frame_scores)
+            
+            # Weighted score: Bias towards the Max score
+            final_score = (max_score * 0.7) + (avg_score * 0.3)
+            
+            verdict = "Real Video"
+            if final_score > 85:
+                verdict = "Deepfake/AI"
+            elif final_score > 60:
+                verdict = "Suspicious"
+                
+            return {
+                "score": round(final_score, 2),
+                "verdict": verdict,
+                "frames_analyzed": len(frames_to_check),
+                "frame_details": frame_details
+            }
+            
+        except Exception as e:
+            print(f"Video Analysis Error: {e}")
+            return {"score": 0, "verdict": "Video Analysis Failed"}
+            
+        finally:
+            # Cleanup temp file
+            if os.path.exists(temp_path):
+                os.remove(temp_path)
+
+video_analysis_service = VideoAnalysisService()

requirements.txt

@@ -1,10 +1,15 @@
-fastapi
-uvicorn
-python-multipart
-torch --index-url https://download.pytorch.org/whl/cpu
-transformers
-scipy
-numpy
-easyocr
-pillow
-opencv-python-headless
+fastapi==0.104.1
+uvicorn==0.24.0
+python-multipart>=0.0.9
+torch>=2.2.0
+transformers>=4.40.0
+accelerate>=0.26.0
+numpy==1.26.2
+slowapi==0.1.9
+easyocr==1.7.1
+pillow==10.2.0
+opencv-python-headless==4.8.1.78
+tf-keras==2.15.0
+pydantic-settings>=2.0.0
+protobuf==4.25.3
+sentencepiece