Spaces:
Sleeping
Sleeping
bahaeddinmselmi
fix(analyzer): robust doubled-URL normalization and enhanced error reporting
c9f75f9 | # C:\Users\bahae\.gemini\antigravity\scratch\verivid-ai\hf_space\app\services\hf_inference.py | |
| import os | |
| import requests | |
| from app.core.config import settings | |
| # Fallback HuggingFace models | |
| VISUAL_MODELS = [("Organika/sdxl-detector", ["artificial", "ai", "synthetic"])] | |
| AUDIO_MODELS = [("mel06/Whisper-Deepfake-Detection", ["fake", "spoof", "synthetic"])] | |
| def call_hf_model(model_name: str, image_bytes: bytes, ai_labels: list) -> float: | |
| """Call HuggingFace model for AI detection""" | |
| 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]: # Target 3 key frames | |
| try: | |
| with open(path, 'rb') as f: | |
| img_bytes = f.read() | |
| for model_name, ai_labels in VISUAL_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: {len(scores)} frames analyzed" | |
| } | |
| return {"avg_prob": 0, "max_prob": 0, "frame_count": 0, "details": "No frames provided" if not frame_paths else "Model inference 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"} | |
| if not settings.HF_TOKEN: | |
| return {"spoof_prob": 0.1, "details": "Audio engine requires HF_TOKEN.", "confidence": "high"} | |
| try: | |
| with open(audio_path, 'rb') as f: | |
| audio_bytes = f.read() | |
| headers = { | |
| "Authorization": f"Bearer {settings.HF_TOKEN}", | |
| "Content-Type": "audio/wav", | |
| } | |
| # Try current best audio deepfake detection model | |
| for model_name, ai_labels in AUDIO_MODELS: | |
| model_url = f"https://router.huggingface.co/hf-inference/models/{model_name}" | |
| 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() | |
| score = 0 | |
| if isinstance(result, list): | |
| for item in result: | |
| label = str(item.get('label', '')).lower() | |
| s = float(item.get('score', 0)) | |
| for ai_label in ai_labels: | |
| if ai_label in label: | |
| score = s | |
| break | |
| if 'human' in label or 'real' in label: | |
| score = 1 - s | |
| return { | |
| "spoof_prob": round(score, 3), | |
| "details": f"AI Audio Detection ({model_name})", | |
| "confidence": "high" if score > 0.8 or score < 0.2 else "medium" | |
| } | |
| except Exception as e: | |
| print(f"Audio HF inference error: {e}") | |
| return {"spoof_prob": 0.1, "details": "Audio engine fallback (Heuristic)", "confidence": "low"} | |