""" 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}")