Spaces:
Sleeping
Sleeping
| from fastapi import APIRouter, UploadFile, File, Form, HTTPException, Request | |
| from typing import Optional | |
| import logging | |
| 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() | |
| logger = logging.getLogger("uvicorn") | |
| async def health_check(): | |
| return {"status": "ok", "message": "VerifAI Backend is running (Text, Image, Video)"} | |
| async def analyze( | |
| request: Request, | |
| text: Optional[str] = Form(None), | |
| file: Optional[UploadFile] = File(None) | |
| ): | |
| """ | |
| Multi-Modal Analysis Endpoint. | |
| """ | |
| try: | |
| # Debug Logging | |
| logger.info(f"Analyze Request: Content-Type={request.headers.get('content-type')}") | |
| # 1. Direct Text Input | |
| if text: | |
| logger.info("Analyzing Text Input") | |
| # Run text analysis in threadpool to avoid blocking event loop | |
| return await run_in_threadpool(text_service.analyze, text) | |
| # 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')} | |
| } | |
| # --- 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 Exception as e: | |
| logger.error(f"Analysis Error: {str(e)}") | |
| raise HTTPException(status_code=500, detail=str(e)) | |