Spaces:
Sleeping
Sleeping
| """ | |
| Supabase Integration for Deepfake Detection Backend | |
| Handles PDF storage and analysis result persistence | |
| """ | |
| import os | |
| import json | |
| import hashlib | |
| from datetime import datetime | |
| from typing import Optional, Dict, Any, List | |
| from supabase import create_client, Client | |
| from dotenv import load_dotenv | |
| import base64 | |
| import logging | |
| # Load environment variables | |
| load_dotenv() | |
| logger = logging.getLogger(__name__) | |
| class SupabaseManager: | |
| """ | |
| Manages all Supabase operations for the deepfake detection system | |
| """ | |
| def __init__(self): | |
| """Initialize Supabase client with configuration.""" | |
| self.supabase_url = os.getenv("SUPABASE_URL") | |
| self.supabase_anon_key = os.getenv("SUPABASE_ANON_KEY") | |
| self.service_role_key = os.getenv("SUPABASE_SERVICE_ROLE_KEY") | |
| self.bucket_name = os.getenv("SUPABASE_BUCKET_NAME", "deepfake-reports") | |
| self.table_name = os.getenv("SUPABASE_TABLE_NAME", "detection_results") | |
| if not all([self.supabase_url, self.supabase_anon_key]): | |
| raise ValueError( | |
| "Missing required Supabase configuration. " | |
| "Please set SUPABASE_URL and SUPABASE_ANON_KEY in your .env file" | |
| ) | |
| # Create Supabase client - Use service role key if available (bypasses RLS) | |
| try: | |
| # Prefer service role key for backend operations to bypass RLS | |
| api_key = self.service_role_key if self.service_role_key else self.supabase_anon_key | |
| self.supabase: Client = create_client(self.supabase_url, api_key) | |
| if self.service_role_key: | |
| logger.info("✅ Supabase client initialized with SERVICE ROLE KEY") | |
| else: | |
| logger.warning("⚠️ Using ANON KEY - Set SUPABASE_SERVICE_ROLE_KEY for backend operations") | |
| except Exception as e: | |
| logger.error(f"❌ Failed to initialize Supabase client: {e}") | |
| raise | |
| def upload_pdf_report(self, pdf_bytes: bytes, filename: str, | |
| classification: str) -> Optional[Dict[str, Any]]: | |
| """ | |
| Upload PDF report to Supabase storage | |
| Args: | |
| pdf_bytes: PDF content as bytes | |
| filename: Original filename (without extension) | |
| classification: 'deepfake' or 'authentic' | |
| Returns: | |
| Dictionary with upload information or None if failed | |
| """ | |
| try: | |
| # Generate unique filename with timestamp | |
| timestamp = datetime.now().strftime('%Y%m%d_%H%M%S_%f')[:-3] | |
| pdf_filename = f"gradcam_{classification}_{filename}_{timestamp}.pdf" | |
| # Upload to Supabase storage | |
| result = self.supabase.storage.from_(self.bucket_name).upload( | |
| path=pdf_filename, | |
| file=pdf_bytes, | |
| file_options={ | |
| "content-type": "application/pdf", | |
| "cache-control": "3600" # Cache for 1 hour | |
| } | |
| ) | |
| if hasattr(result, 'error') and result.error: | |
| logger.error(f"Upload error: {result.error}") | |
| return None | |
| # Get public URL | |
| public_url_result = self.supabase.storage.from_(self.bucket_name).get_public_url(pdf_filename) | |
| public_url = public_url_result if isinstance(public_url_result, str) else public_url_result.get('publicUrl') | |
| upload_info = { | |
| "pdf_filename": pdf_filename, | |
| "pdf_url": public_url, | |
| "bucket_name": self.bucket_name, | |
| "file_size_mb": round(len(pdf_bytes) / (1024 * 1024), 2), | |
| "uploaded_at": datetime.now().isoformat(), | |
| "upload_path": f"{self.bucket_name}/{pdf_filename}" | |
| } | |
| logger.info(f"✅ PDF uploaded successfully: {pdf_filename}") | |
| return upload_info | |
| except Exception as e: | |
| logger.error(f"❌ Failed to upload PDF: {e}") | |
| return None | |
| def save_detection_result(self, analysis_data: Dict[str, Any], | |
| upload_info: Optional[Dict[str, Any]] = None, | |
| client_metadata: Optional[Dict[str, Any]] = None) -> Optional[str]: | |
| """ | |
| Save detection analysis result to database | |
| Args: | |
| analysis_data: Detection analysis results (must include 'user_id') | |
| upload_info: PDF upload information | |
| client_metadata: Optional client metadata (IP, user agent, etc.) | |
| Returns: | |
| Record ID if successful, None if failed | |
| """ | |
| try: | |
| # Prepare SIMPLIFIED database record with only essential fields | |
| db_record = { | |
| # Essential fields only | |
| "filename": analysis_data.get('filename', 'unknown'), | |
| "classification": analysis_data.get('classification', | |
| 'deepfake' if analysis_data['prediction'] > 0.5 else 'authentic'), | |
| "image_size": analysis_data.get('image_size'), | |
| "prediction": analysis_data['prediction'], | |
| "face_confidence": analysis_data.get('face_confidence'), | |
| "processing_time_ms": analysis_data.get('processing_time_ms'), | |
| "user_id": analysis_data.get('user_id'), # Add user_id field | |
| # PDF storage info | |
| "pdf_filename": upload_info.get('pdf_filename') if upload_info else None, | |
| "pdf_url": upload_info.get('pdf_url') if upload_info else None, | |
| "bucket_name": upload_info.get('bucket_name') if upload_info else None, | |
| "file_size_mb": upload_info.get('file_size_mb') if upload_info else None | |
| } | |
| # Insert into database | |
| result = self.supabase.table(self.table_name).insert(db_record).execute() | |
| if result.data and len(result.data) > 0: | |
| record_id = result.data[0]['id'] | |
| logger.info(f"✅ Detection result saved with ID: {record_id}") | |
| return record_id | |
| else: | |
| logger.error("❌ No data returned from insert operation") | |
| return None | |
| except Exception as e: | |
| logger.error(f"❌ Failed to save detection result: {e}") | |
| return None | |
| def get_detection_results(self, limit: int = 100, | |
| filter_deepfake: Optional[bool] = None) -> List[Dict[str, Any]]: | |
| """ | |
| Retrieve detection results from database | |
| Args: | |
| limit: Maximum number of results to return | |
| filter_deepfake: Filter by deepfake classification (None for all) | |
| Returns: | |
| List of detection results | |
| """ | |
| try: | |
| query = self.supabase.table(self.table_name).select("*") | |
| if filter_deepfake is not None: | |
| query = query.eq("is_deepfake", filter_deepfake) | |
| query = query.order("created_at", desc=True).limit(limit) | |
| result = query.execute() | |
| if result.data: | |
| logger.info(f"✅ Retrieved {len(result.data)} detection results") | |
| return result.data | |
| else: | |
| logger.info("No detection results found") | |
| return [] | |
| except Exception as e: | |
| logger.error(f"❌ Failed to retrieve detection results: {e}") | |
| return [] | |
| def get_detection_stats(self) -> Dict[str, Any]: | |
| """ | |
| Get statistics about detection results | |
| Returns: | |
| Dictionary with statistics | |
| """ | |
| try: | |
| # Get total count | |
| total_result = self.supabase.table(self.table_name).select("id", count="exact").execute() | |
| total_count = total_result.count if total_result.count is not None else 0 | |
| # Get deepfake count | |
| deepfake_result = self.supabase.table(self.table_name).select("id", count="exact").eq("is_deepfake", True).execute() | |
| deepfake_count = deepfake_result.count if deepfake_result.count is not None else 0 | |
| # Get authentic count | |
| authentic_count = total_count - deepfake_count | |
| # Get recent activity (last 24 hours) | |
| from datetime import timedelta | |
| yesterday = (datetime.now() - timedelta(days=1)).isoformat() | |
| recent_result = self.supabase.table(self.table_name).select("id", count="exact").gte("created_at", yesterday).execute() | |
| recent_count = recent_result.count if recent_result.count is not None else 0 | |
| stats = { | |
| "total_analyses": total_count, | |
| "deepfake_detected": deepfake_count, | |
| "authentic_detected": authentic_count, | |
| "deepfake_percentage": round((deepfake_count / total_count * 100), 2) if total_count > 0 else 0, | |
| "recent_24h": recent_count, | |
| "last_updated": datetime.now().isoformat() | |
| } | |
| logger.info("✅ Detection statistics retrieved successfully") | |
| return stats | |
| except Exception as e: | |
| logger.error(f"❌ Failed to retrieve detection statistics: {e}") | |
| return { | |
| "total_analyses": 0, | |
| "deepfake_detected": 0, | |
| "authentic_detected": 0, | |
| "deepfake_percentage": 0, | |
| "recent_24h": 0, | |
| "error": str(e) | |
| } | |
| def delete_old_reports(self, days_old: int = 30) -> int: | |
| """ | |
| Delete old PDF reports and database records (cleanup function) | |
| Args: | |
| days_old: Delete reports older than this many days | |
| Returns: | |
| Number of records deleted | |
| """ | |
| try: | |
| from datetime import timedelta | |
| cutoff_date = (datetime.now() - timedelta(days=days_old)).isoformat() | |
| # Get old records with PDF filenames | |
| old_records = self.supabase.table(self.table_name).select("pdf_filename").lt("created_at", cutoff_date).execute() | |
| deleted_count = 0 | |
| if old_records.data: | |
| # Delete PDF files from storage | |
| for record in old_records.data: | |
| if record.get('pdf_filename'): | |
| try: | |
| self.supabase.storage.from_(self.bucket_name).remove([record['pdf_filename']]) | |
| except Exception as e: | |
| logger.warning(f"Failed to delete PDF {record['pdf_filename']}: {e}") | |
| # Delete database records | |
| delete_result = self.supabase.table(self.table_name).delete().lt("created_at", cutoff_date).execute() | |
| deleted_count = len(delete_result.data) if delete_result.data else 0 | |
| logger.info(f"✅ Cleaned up {deleted_count} old records") | |
| return deleted_count | |
| except Exception as e: | |
| logger.error(f"❌ Failed to cleanup old reports: {e}") | |
| return 0 | |
| def health_check(self) -> Dict[str, Any]: | |
| """ | |
| Perform health check on Supabase services | |
| Returns: | |
| Dictionary with health status | |
| """ | |
| health_status = { | |
| "database": False, | |
| "storage": False, | |
| "overall": False, | |
| "timestamp": datetime.now().isoformat() | |
| } | |
| try: | |
| # Test database connection | |
| test_query = self.supabase.table(self.table_name).select("id").limit(1).execute() | |
| health_status["database"] = True | |
| # Test storage connection | |
| buckets = self.supabase.storage.list_buckets() | |
| health_status["storage"] = any(bucket.name == self.bucket_name for bucket in buckets) | |
| health_status["overall"] = health_status["database"] and health_status["storage"] | |
| except Exception as e: | |
| health_status["error"] = str(e) | |
| logger.error(f"❌ Supabase health check failed: {e}") | |
| return health_status | |
| # Global instance | |
| _supabase_manager = None | |
| def get_supabase_manager() -> SupabaseManager: | |
| """Get or create global Supabase manager instance.""" | |
| global _supabase_manager | |
| if _supabase_manager is None: | |
| _supabase_manager = SupabaseManager() | |
| return _supabase_manager | |
| def initialize_supabase() -> bool: | |
| """ | |
| Initialize Supabase connection (call this at application startup) | |
| Note: Bucket and table setup must be done manually in Supabase dashboard | |
| Returns: | |
| True if connection successful, False otherwise | |
| """ | |
| try: | |
| manager = get_supabase_manager() | |
| health = manager.health_check() | |
| if health["overall"]: | |
| logger.info("✅ Supabase connection verified") | |
| return True | |
| else: | |
| logger.warning("⚠️ Supabase connection has issues") | |
| return False | |
| except Exception as e: | |
| logger.error(f"❌ Supabase initialization failed: {e}") | |
| return False | |
| # Configuration validation | |
| def validate_supabase_config() -> Dict[str, Any]: | |
| """Validate Supabase configuration and return status.""" | |
| config_status = { | |
| "valid": False, | |
| "missing_vars": [], | |
| "recommendations": [] | |
| } | |
| required_vars = ["SUPABASE_URL", "SUPABASE_ANON_KEY"] | |
| optional_vars = ["SUPABASE_SERVICE_ROLE_KEY", "SUPABASE_BUCKET_NAME", "SUPABASE_TABLE_NAME"] | |
| for var in required_vars: | |
| if not os.getenv(var): | |
| config_status["missing_vars"].append(var) | |
| if not config_status["missing_vars"]: | |
| config_status["valid"] = True | |
| # Add recommendations | |
| if not os.getenv("SUPABASE_SERVICE_ROLE_KEY"): | |
| config_status["recommendations"].append("Set SUPABASE_SERVICE_ROLE_KEY for admin operations") | |
| return config_status | |
| if __name__ == "__main__": | |
| # Test configuration | |
| config_status = validate_supabase_config() | |
| print("Supabase Configuration Status:") | |
| print(f"Valid: {config_status['valid']}") | |
| if config_status["missing_vars"]: | |
| print(f"Missing variables: {config_status['missing_vars']}") | |
| if config_status["recommendations"]: | |
| print(f"Recommendations: {config_status['recommendations']}") | |
| if config_status["valid"]: | |
| print("\nTesting Supabase connection...") | |
| try: | |
| manager = get_supabase_manager() | |
| health = manager.health_check() | |
| print(f"Health check: {health}") | |
| except Exception as e: | |
| print(f"Connection test failed: {e}") |