""" Supabase PostgreSQL Database Implementation Replaces SQLite with Supabase PostgreSQL for production deployment """ import os import time from datetime import datetime, timedelta from typing import Dict, List, Optional import psycopg2 from psycopg2.extras import RealDictCursor from contextlib import contextmanager class SupabaseDatabase: """Database management class for Supabase PostgreSQL""" def __init__(self, db_url: Optional[str] = None): """ Initialize Supabase database connection Args: db_url: PostgreSQL connection string Format: postgresql://postgres:[PASSWORD]@[HOST]:5432/postgres """ self.db_url = db_url or os.environ.get('SUPABASE_DB_URL') if not self.db_url: raise ValueError("SUPABASE_DB_URL environment variable is required") # Test connection and initialize schema self._init_database() @contextmanager def get_connection(self, retries=3): """Get database connection with automatic commit/rollback and retry logic""" last_error = None for attempt in range(retries): try: # Force IPv4 if IPv6 fails (connection pooling URLs typically work better) conn = psycopg2.connect( self.db_url, connect_timeout=10 # 10 second timeout ) try: yield conn conn.commit() return except Exception as e: conn.rollback() raise e finally: conn.close() except (psycopg2.OperationalError, psycopg2.InterfaceError) as e: # Transient connection errors - retry last_error = e error_msg = str(e) # Check for circuit breaker errors if "circuit breaker" in error_msg.lower() or "circuit breaker open" in error_msg.lower(): if attempt == 0: # Only print detailed message on first attempt print(f"⚠️ Circuit breaker error detected:") print(f" This usually means:") print(f" 1. Username format is wrong (for pooler: must be postgres.PROJECT_REF)") print(f" 2. Password is incorrect") print(f" 3. Connection string format is wrong") # Try to extract and show username from connection string for debugging try: from urllib.parse import urlparse parsed = urlparse(self.db_url) username = parsed.username or "NOT SET" hostname = parsed.hostname or "NOT SET" print(f" Current username: {username}") print(f" Current hostname: {hostname}") # Check if using pooler if "pooler" in hostname: if not username.startswith("postgres."): print(f" ❌ ERROR: For pooler, username must be 'postgres.PROJECT_REF'") print(f" ❌ Current: '{username}'") print(f" ✅ Should be: 'postgres.rhzontzgndybmjpeuvzm'") print(f" 💡 Fix: Update SUPABASE_DB_URL to include project ref in username") else: print(f" ✅ Username format looks correct for pooler") else: print(f" ℹ️ Using direct connection (not pooler)") except Exception as e: print(f" Could not parse connection string: {e}") # Check for timeout errors elif "timeout" in error_msg.lower() or "expired" in error_msg.lower(): if attempt == 0: # Only print detailed message on first attempt print(f"⚠️ Connection timeout detected:") print(f" This usually means:") print(f" 1. Connection string format is wrong (check username includes project ref)") print(f" 2. Password is incorrect or not URL-encoded") print(f" 3. Network/firewall blocking connection") print(f" For pooler URL, username must be: postgres.PROJECT_REF") print(f" Example: postgres.rhzontzgndybmjpeuvzm") # Try to extract and show username from connection string for debugging try: from urllib.parse import urlparse parsed = urlparse(self.db_url) username = parsed.username or "NOT SET" print(f" Current username in connection string: {username}") if not username.startswith("postgres."): print(f" ⚠️ WARNING: Username should be 'postgres.PROJECT_REF' for pooler!") except: pass # Check if it's a network unreachable error elif "Network is unreachable" in error_msg or "Name or service not known" in error_msg: if attempt == 0: # Only print detailed message on first attempt print(f"⚠️ Network connectivity issue detected:") print(f" This usually means:") print(f" 1. Supabase IP allowlist is blocking Hugging Face IPs") print(f" 2. You need to use Connection Pooler URL instead") print(f" Go to Supabase Dashboard > Settings > Database") print(f" Use the 'Connection pooling' URI (port 6543 or 5432)") if attempt < retries - 1: wait_time = (attempt + 1) * 0.5 # Exponential backoff: 0.5s, 1s, 1.5s print(f"⚠️ Database connection error (attempt {attempt + 1}/{retries}), retrying in {wait_time}s...") time.sleep(wait_time) continue else: # Last attempt failed print(f"❌ Database connection failed after {retries} attempts: {e}") raise except Exception as e: # Non-transient errors - don't retry raise def _init_database(self): """Initialize database tables""" with self.get_connection() as conn: cursor = conn.cursor() # Users table cursor.execute(''' CREATE TABLE IF NOT EXISTS users ( id SERIAL PRIMARY KEY, user_id TEXT UNIQUE NOT NULL, email TEXT, phone TEXT, password_hash TEXT, baby_name TEXT NOT NULL, baby_gender TEXT, baby_birthday TEXT, platform TEXT DEFAULT 'web', language TEXT DEFAULT 'ar', registration_date TIMESTAMPTZ NOT NULL DEFAULT NOW(), created_at TIMESTAMPTZ DEFAULT NOW(), updated_at TIMESTAMPTZ DEFAULT NOW() ) ''') # Add password_hash column if it doesn't exist (for existing databases) try: cursor.execute(''' ALTER TABLE users ADD COLUMN IF NOT EXISTS password_hash TEXT ''') except Exception as e: # Column might already exist, ignore error pass # Predictions table (recordings) cursor.execute(''' CREATE TABLE IF NOT EXISTS predictions ( id SERIAL PRIMARY KEY, user_id TEXT, prediction TEXT NOT NULL, confidence REAL NOT NULL, audio_id TEXT UNIQUE NOT NULL, model_type TEXT, timestamp TIMESTAMPTZ NOT NULL DEFAULT NOW(), created_at TIMESTAMPTZ DEFAULT NOW(), FOREIGN KEY (user_id) REFERENCES users(user_id) ON DELETE SET NULL ) ''') # Feedback table cursor.execute(''' CREATE TABLE IF NOT EXISTS feedback ( id SERIAL PRIMARY KEY, audio_id TEXT NOT NULL, user_id TEXT, predicted_label TEXT NOT NULL, correct_label TEXT NOT NULL, is_correct BOOLEAN NOT NULL, confidence REAL, category TEXT, file_path TEXT, submission_id TEXT, timestamp TIMESTAMPTZ NOT NULL DEFAULT NOW(), created_at TIMESTAMPTZ DEFAULT NOW(), FOREIGN KEY (user_id) REFERENCES users(user_id) ON DELETE SET NULL ) ''') # Add file_path and submission_id columns if they don't exist (for existing databases) try: cursor.execute('ALTER TABLE feedback ADD COLUMN IF NOT EXISTS file_path TEXT') cursor.execute('ALTER TABLE feedback ADD COLUMN IF NOT EXISTS submission_id TEXT') except Exception as e: # Column might already exist, ignore error pass # Add what_helped to predictions if it doesn't exist (for existing databases) try: cursor.execute('ALTER TABLE predictions ADD COLUMN IF NOT EXISTS what_helped TEXT') except Exception as e: pass # Analytics events table cursor.execute(''' CREATE TABLE IF NOT EXISTS analytics_events ( id SERIAL PRIMARY KEY, event_name TEXT NOT NULL, user_id TEXT, event_data JSONB, timestamp TIMESTAMPTZ NOT NULL DEFAULT NOW(), created_at TIMESTAMPTZ DEFAULT NOW() ) ''') # Create indexes for better query performance cursor.execute('CREATE INDEX IF NOT EXISTS idx_users_user_id ON users(user_id)') cursor.execute('CREATE INDEX IF NOT EXISTS idx_predictions_user_id ON predictions(user_id)') cursor.execute('CREATE INDEX IF NOT EXISTS idx_predictions_timestamp ON predictions(timestamp)') cursor.execute('CREATE INDEX IF NOT EXISTS idx_feedback_user_id ON feedback(user_id)') cursor.execute('CREATE INDEX IF NOT EXISTS idx_feedback_audio_id ON feedback(audio_id)') cursor.execute('CREATE INDEX IF NOT EXISTS idx_feedback_timestamp ON feedback(timestamp)') cursor.execute('CREATE INDEX IF NOT EXISTS idx_analytics_user_id ON analytics_events(user_id)') cursor.execute('CREATE INDEX IF NOT EXISTS idx_analytics_timestamp ON analytics_events(timestamp)') conn.commit() print("✅ Supabase database tables initialized") # ==================== USERS CRUD ==================== def create_user(self, user_data: Dict) -> str: """Create a new user""" try: with self.get_connection() as conn: cursor = conn.cursor() registration_date = user_data.get('registration_date', datetime.now().isoformat()) # Convert ISO format string to PostgreSQL timestamp if needed # PostgreSQL accepts ISO format strings directly cursor.execute(''' INSERT INTO users ( user_id, email, phone, password_hash, baby_name, baby_gender, baby_birthday, platform, language, registration_date ) VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s, %s) ''', ( user_data.get('user_id'), user_data.get('email'), user_data.get('phone'), user_data.get('password_hash'), user_data.get('baby_name'), user_data.get('baby_gender'), user_data.get('baby_birthday'), user_data.get('platform', 'web'), user_data.get('language', 'ar'), registration_date )) return user_data.get('user_id') except Exception as e: print(f"❌ Database error in create_user: {e}") raise def ensure_user_exists(self, user_id: str) -> None: """Create a minimal user record if it does not exist (so predictions FK is satisfied).""" if not user_id or not user_id.strip(): return if self.get_user(user_id.strip()): return try: self.create_user({ 'user_id': user_id.strip(), 'baby_name': 'Guest', 'platform': 'web', 'language': 'ar', }) except Exception as e: # Ignore duplicate (race) or other errors; prediction save may still fail pass def get_user(self, user_id: str) -> Optional[Dict]: """Get user by user_id""" with self.get_connection() as conn: cursor = conn.cursor(cursor_factory=RealDictCursor) cursor.execute('SELECT * FROM users WHERE user_id = %s', (user_id,)) row = cursor.fetchone() return dict(row) if row else None def get_user_by_email(self, email: str) -> Optional[Dict]: """Get user by email (for login)""" with self.get_connection() as conn: cursor = conn.cursor(cursor_factory=RealDictCursor) cursor.execute('SELECT * FROM users WHERE email = %s', (email,)) row = cursor.fetchone() return dict(row) if row else None def get_user_by_phone(self, phone: str) -> Optional[Dict]: """Get user by phone number (for login)""" with self.get_connection() as conn: cursor = conn.cursor(cursor_factory=RealDictCursor) cursor.execute('SELECT * FROM users WHERE phone = %s', (phone,)) row = cursor.fetchone() return dict(row) if row else None def get_user_by_email_or_phone(self, identifier: str) -> Optional[Dict]: """Get user by email or phone number (for login)""" with self.get_connection() as conn: cursor = conn.cursor(cursor_factory=RealDictCursor) # Try email first (case-insensitive) cursor.execute('SELECT * FROM users WHERE LOWER(email) = LOWER(%s) OR phone = %s', (identifier, identifier)) row = cursor.fetchone() return dict(row) if row else None def get_all_users(self, limit: Optional[int] = None, offset: int = 0) -> List[Dict]: """Get all users with pagination""" with self.get_connection() as conn: cursor = conn.cursor(cursor_factory=RealDictCursor) if limit: cursor.execute( 'SELECT * FROM users ORDER BY registration_date DESC LIMIT %s OFFSET %s', (limit, offset) ) else: cursor.execute('SELECT * FROM users ORDER BY registration_date DESC') return [dict(row) for row in cursor.fetchall()] def update_user(self, user_id: str, user_data: Dict) -> bool: """Update user information""" with self.get_connection() as conn: cursor = conn.cursor() updates = [] values = [] for key in ['email', 'phone', 'baby_name', 'baby_gender', 'baby_birthday', 'platform', 'language']: if key in user_data: updates.append(f'{key} = %s') values.append(user_data[key]) if not updates: return False updates.append('updated_at = %s') values.append(datetime.now()) values.append(user_id) cursor.execute(f''' UPDATE users SET {", ".join(updates)} WHERE user_id = %s ''', values) return cursor.rowcount > 0 def delete_user(self, user_id: str) -> bool: """Delete user""" with self.get_connection() as conn: cursor = conn.cursor() cursor.execute('DELETE FROM users WHERE user_id = %s', (user_id,)) return cursor.rowcount > 0 def count_users(self) -> int: """Count total users""" with self.get_connection() as conn: cursor = conn.cursor() cursor.execute('SELECT COUNT(*) as count FROM users') return cursor.fetchone()[0] # ==================== PREDICTIONS ==================== def create_prediction(self, prediction_data: Dict): """Save a prediction (recording)""" with self.get_connection() as conn: cursor = conn.cursor() timestamp = prediction_data.get('timestamp', datetime.now().isoformat()) cursor.execute(''' INSERT INTO predictions ( user_id, prediction, confidence, audio_id, model_type, timestamp ) VALUES (%s, %s, %s, %s, %s, %s) ''', ( prediction_data.get('user_id'), prediction_data.get('prediction'), prediction_data.get('confidence', 0), prediction_data.get('audio_id'), prediction_data.get('model_type'), timestamp )) def count_predictions(self, user_id: Optional[str] = None) -> int: """Count total predictions/recordings""" with self.get_connection() as conn: cursor = conn.cursor() if user_id: cursor.execute('SELECT COUNT(*) FROM predictions WHERE user_id = %s', (user_id,)) else: cursor.execute('SELECT COUNT(*) FROM predictions') return cursor.fetchone()[0] def get_predictions_list(self, limit: Optional[int] = None, offset: int = 0, user_id: Optional[str] = None) -> List[Dict]: """Get list of predictions with pagination""" with self.get_connection() as conn: cursor = conn.cursor(cursor_factory=RealDictCursor) query = 'SELECT * FROM predictions' params = [] if user_id: query += ' WHERE user_id = %s' params.append(user_id) query += ' ORDER BY timestamp DESC' if limit: query += ' LIMIT %s OFFSET %s' params.extend([limit, offset]) cursor.execute(query, tuple(params)) return [dict(row) for row in cursor.fetchall()] def update_prediction_what_helped(self, prediction_id: int, user_id: str, what_helped: Optional[str]) -> bool: """Update the what_helped field for a prediction. Only updates if prediction belongs to user_id.""" with self.get_connection() as conn: cursor = conn.cursor() cursor.execute( 'UPDATE predictions SET what_helped = %s WHERE id = %s AND user_id = %s', (what_helped, prediction_id, user_id) ) return cursor.rowcount > 0 # ==================== FEEDBACK ==================== def create_feedback(self, feedback_data: Dict): """Save feedback""" try: with self.get_connection() as conn: cursor = conn.cursor() timestamp = feedback_data.get('timestamp', datetime.now().isoformat()) # Ensure is_correct is a proper boolean is_correct = bool(feedback_data.get('is_correct', False)) # Handle confidence - convert None to NULL confidence = feedback_data.get('confidence') if confidence is not None: try: confidence = float(confidence) except (ValueError, TypeError): confidence = None cursor.execute(''' INSERT INTO feedback ( audio_id, user_id, predicted_label, correct_label, is_correct, confidence, category, file_path, submission_id, timestamp ) VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s, %s) ''', ( feedback_data.get('audio_id'), feedback_data.get('user_id'), feedback_data.get('predicted_label'), feedback_data.get('correct_label'), is_correct, confidence, feedback_data.get('correct_label'), # category feedback_data.get('file_path'), # file path where audio is stored feedback_data.get('submission_id'), # submission ID from feedback_manager timestamp )) except Exception as e: print(f"❌ Database error in create_feedback: {e}") import traceback traceback.print_exc() raise def count_feedback(self, is_correct: Optional[bool] = None) -> int: """Count feedback (optionally filtered by is_correct)""" with self.get_connection() as conn: cursor = conn.cursor() if is_correct is not None: cursor.execute('SELECT COUNT(*) FROM feedback WHERE is_correct = %s', (is_correct,)) else: cursor.execute('SELECT COUNT(*) FROM feedback') return cursor.fetchone()[0] def get_feedback_list(self, limit: Optional[int] = None, offset: int = 0, is_correct: Optional[bool] = None) -> List[Dict]: """Get feedback entries with pagination and optional filter""" with self.get_connection() as conn: cursor = conn.cursor(cursor_factory=RealDictCursor) query = 'SELECT * FROM feedback' params = [] if is_correct is not None: query += ' WHERE is_correct = %s' params.append(is_correct) query += ' ORDER BY timestamp DESC' if limit: query += ' LIMIT %s OFFSET %s' params.extend([limit, offset]) cursor.execute(query, tuple(params)) return [dict(row) for row in cursor.fetchall()] def delete_feedback(self, feedback_id: int) -> bool: """Delete a feedback entry by ID""" with self.get_connection() as conn: cursor = conn.cursor() cursor.execute('DELETE FROM feedback WHERE id = %s', (feedback_id,)) return cursor.rowcount > 0 def delete_feedback_by_submission_id(self, submission_id: str) -> bool: """Delete a feedback entry by submission_id""" with self.get_connection() as conn: cursor = conn.cursor() cursor.execute('DELETE FROM feedback WHERE submission_id = %s', (submission_id,)) return cursor.rowcount > 0 def delete_feedbacks_by_date_range(self, start_date: str, end_date: str) -> int: """Delete feedbacks within a date range. Returns number of deleted rows.""" with self.get_connection() as conn: cursor = conn.cursor() cursor.execute( 'DELETE FROM feedback WHERE timestamp >= %s AND timestamp <= %s', (start_date, end_date) ) return cursor.rowcount def delete_feedbacks_by_user_id(self, user_id: str) -> int: """Delete all feedbacks for a specific user. Returns number of deleted rows.""" with self.get_connection() as conn: cursor = conn.cursor() cursor.execute('DELETE FROM feedback WHERE user_id = %s', (user_id,)) return cursor.rowcount def get_feedback_by_id(self, feedback_id: int) -> Optional[Dict]: """Get a single feedback entry by ID""" with self.get_connection() as conn: cursor = conn.cursor(cursor_factory=RealDictCursor) cursor.execute('SELECT * FROM feedback WHERE id = %s', (feedback_id,)) row = cursor.fetchone() return dict(row) if row else None # ==================== ANALYTICS EVENTS ==================== def create_analytics_event(self, event_data: Dict): """Save an analytics event""" try: with self.get_connection() as conn: cursor = conn.cursor() timestamp = event_data.get('timestamp', datetime.now().isoformat()) import json cursor.execute(''' INSERT INTO analytics_events ( event_name, user_id, event_data, timestamp ) VALUES (%s, %s, %s, %s) ''', ( event_data.get('event_name'), event_data.get('user_id'), json.dumps(event_data.get('event_data', {})), timestamp )) except Exception as e: print(f"❌ Database error in create_analytics_event: {e}") raise # ==================== STATISTICS ==================== def get_user_stats(self) -> Dict: """Get comprehensive user statistics""" with self.get_connection() as conn: cursor = conn.cursor() # Total users cursor.execute('SELECT COUNT(*) FROM users') total_users = cursor.fetchone()[0] # Active users (last 7/30 days) - users who made predictions seven_days_ago = datetime.now() - timedelta(days=7) thirty_days_ago = datetime.now() - timedelta(days=30) cursor.execute(''' SELECT COUNT(DISTINCT user_id) FROM predictions WHERE timestamp >= %s AND user_id IS NOT NULL ''', (seven_days_ago,)) active_7d = cursor.fetchone()[0] cursor.execute(''' SELECT COUNT(DISTINCT user_id) FROM predictions WHERE timestamp >= %s AND user_id IS NOT NULL ''', (thirty_days_ago,)) active_30d = cursor.fetchone()[0] # New users cursor.execute(''' SELECT COUNT(*) FROM users WHERE registration_date >= %s ''', (seven_days_ago,)) new_users_7d = cursor.fetchone()[0] cursor.execute(''' SELECT COUNT(*) FROM users WHERE registration_date >= %s ''', (thirty_days_ago,)) new_users_30d = cursor.fetchone()[0] return { 'total_users': total_users, 'active_users_7d': active_7d, 'active_users_30d': active_30d, 'new_users_7d': new_users_7d, 'new_users_30d': new_users_30d } def get_feedback_stats(self) -> Dict: """Get feedback statistics""" with self.get_connection() as conn: cursor = conn.cursor(cursor_factory=RealDictCursor) # Total feedback total_feedback = self.count_feedback() # Positive feedback (is_correct = true) positive_feedback = self.count_feedback(is_correct=True) # Negative feedback (is_correct = false) negative_feedback = self.count_feedback(is_correct=False) # Feedback by category cursor.execute(''' SELECT category, COUNT(*) as count FROM feedback GROUP BY category ''') category_counts = {row['category']: row['count'] for row in cursor.fetchall()} # Total recordings/predictions total_recordings = self.count_predictions() return { 'total_feedback': total_feedback, 'positive_feedback': positive_feedback, 'negative_feedback': negative_feedback, 'category_counts': category_counts, 'total_recordings': total_recordings } def get_platform_distribution(self) -> Dict: """Get platform distribution""" with self.get_connection() as conn: cursor = conn.cursor(cursor_factory=RealDictCursor) cursor.execute(''' SELECT platform, COUNT(*) as count FROM users GROUP BY platform ''') return {row['platform']: row['count'] for row in cursor.fetchall()} def get_language_distribution(self) -> Dict: """Get language distribution""" with self.get_connection() as conn: cursor = conn.cursor(cursor_factory=RealDictCursor) cursor.execute(''' SELECT language, COUNT(*) as count FROM users GROUP BY language ''') return {row['language']: row['count'] for row in cursor.fetchall()} def get_user_timeline(self, days: int = 30) -> List[Dict]: """Get user registration timeline""" with self.get_connection() as conn: cursor = conn.cursor(cursor_factory=RealDictCursor) days_ago = datetime.now() - timedelta(days=days) cursor.execute(''' SELECT DATE(registration_date) as date, COUNT(*) as count FROM users WHERE registration_date >= %s GROUP BY DATE(registration_date) ORDER BY date ASC ''', (days_ago,)) return [{'date': str(row['date']), 'count': row['count']} for row in cursor.fetchall()] def get_feedback_timeline(self, days: int = 30) -> List[Dict]: """Get feedback submission timeline""" with self.get_connection() as conn: cursor = conn.cursor(cursor_factory=RealDictCursor) days_ago = datetime.now() - timedelta(days=days) cursor.execute(''' SELECT DATE(timestamp) as date, COUNT(*) as count FROM feedback WHERE timestamp >= %s GROUP BY DATE(timestamp) ORDER BY date ASC ''', (days_ago,)) return [{'date': str(row['date']), 'count': row['count']} for row in cursor.fetchall()]