Spaces:
Sleeping
Sleeping
| 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() | |