#!/usr/bin/env python3 """ YouTube Knowledge Pipeline Schema Migration Runner Connects to Elestio PostgreSQL and runs the YouTube schema migration. Usage: python run_youtube_migration.py [--verify-only] Options: --verify-only Only verify tables exist, don't run migration RULE 6 COMPLIANT: PostgreSQL only, no SQLite """ import sys import os from pathlib import Path from datetime import datetime # Add genesis-system paths for imports sys.path.insert(0, str(Path(__file__).parent.parent.parent.parent)) sys.path.insert(0, '/mnt/e/genesis-system/data/genesis-memory') try: import psycopg2 from psycopg2 import sql except ImportError: print("ERROR: psycopg2 not installed. Install with: pip install psycopg2-binary") sys.exit(1) # Import Elestio config try: from elestio_config import PostgresConfig except ImportError: # Fallback to hardcoded values if import fails class PostgresConfig: host = "postgresql-genesis-u50607.vm.elestio.app" port = 25432 user = "postgres" password = "etY0eog17tD-dDuj--IRH" database = "postgres" @classmethod def get_connection_params(cls): return { "host": cls.host, "port": cls.port, "user": cls.user, "password": cls.password, "database": cls.database } # Migration file path MIGRATION_FILE = Path(__file__).parent / "002_youtube_schema.sql" # Tables to verify after migration EXPECTED_TABLES = [ "youtube_videos", "video_entities", "video_insights", "processed_history", "supadata_usage" ] # Expected indexes (subset for verification) EXPECTED_INDEXES = [ "idx_youtube_videos_channel", "idx_youtube_videos_published_at", "idx_video_entities_video_id", "idx_video_entities_type", "idx_video_insights_video_id", "idx_processed_history_video_id", "idx_supadata_usage_date" ] def get_connection(): """Get PostgreSQL connection using Elestio config.""" try: conn = psycopg2.connect(**PostgresConfig.get_connection_params()) return conn except Exception as e: print(f"ERROR: Failed to connect to PostgreSQL: {e}") sys.exit(1) def run_migration(conn): """Execute the migration SQL file.""" print(f"\n{'='*60}") print("YOUTUBE KNOWLEDGE PIPELINE SCHEMA MIGRATION") print(f"{'='*60}") print(f"Migration file: {MIGRATION_FILE}") print(f"Started at: {datetime.now().isoformat()}") print() # Read migration SQL if not MIGRATION_FILE.exists(): print(f"ERROR: Migration file not found: {MIGRATION_FILE}") return False with open(MIGRATION_FILE, 'r') as f: migration_sql = f.read() print(f"Migration SQL loaded: {len(migration_sql)} bytes") # Execute migration cursor = conn.cursor() try: print("\nExecuting migration...") cursor.execute(migration_sql) conn.commit() print("Migration executed successfully!") return True except Exception as e: conn.rollback() print(f"ERROR: Migration failed: {e}") return False finally: cursor.close() def verify_tables(conn): """Verify all expected tables were created.""" print(f"\n{'='*60}") print("VERIFYING TABLES") print(f"{'='*60}") cursor = conn.cursor() all_verified = True for table in EXPECTED_TABLES: cursor.execute(""" SELECT EXISTS ( SELECT FROM information_schema.tables WHERE table_schema = 'public' AND table_name = %s ) """, (table,)) exists = cursor.fetchone()[0] status = "[OK]" if exists else "[MISSING]" print(f" {status} {table}") if not exists: all_verified = False cursor.close() return all_verified def verify_indexes(conn): """Verify key indexes were created.""" print(f"\n{'='*60}") print("VERIFYING INDEXES") print(f"{'='*60}") cursor = conn.cursor() all_verified = True for index in EXPECTED_INDEXES: cursor.execute(""" SELECT EXISTS ( SELECT FROM pg_indexes WHERE schemaname = 'public' AND indexname = %s ) """, (index,)) exists = cursor.fetchone()[0] status = "[OK]" if exists else "[MISSING]" print(f" {status} {index}") if not exists: all_verified = False cursor.close() return all_verified def verify_foreign_keys(conn): """Verify foreign key constraints.""" print(f"\n{'='*60}") print("VERIFYING FOREIGN KEYS") print(f"{'='*60}") cursor = conn.cursor() # Check for FK constraints cursor.execute(""" SELECT tc.table_name, kcu.column_name, ccu.table_name AS foreign_table_name, ccu.column_name AS foreign_column_name FROM information_schema.table_constraints AS tc JOIN information_schema.key_column_usage AS kcu ON tc.constraint_name = kcu.constraint_name AND tc.table_schema = kcu.table_schema JOIN information_schema.constraint_column_usage AS ccu ON ccu.constraint_name = tc.constraint_name AND ccu.table_schema = tc.table_schema WHERE tc.constraint_type = 'FOREIGN KEY' AND tc.table_name IN ('video_entities', 'video_insights') """) fks = cursor.fetchall() cursor.close() if fks: for fk in fks: print(f" [OK] {fk[0]}.{fk[1]} -> {fk[2]}.{fk[3]}") return True else: print(" [WARNING] No foreign keys found") return False def get_table_row_counts(conn): """Get row counts for all tables.""" print(f"\n{'='*60}") print("TABLE ROW COUNTS") print(f"{'='*60}") cursor = conn.cursor() for table in EXPECTED_TABLES: try: cursor.execute(sql.SQL("SELECT COUNT(*) FROM {}").format(sql.Identifier(table))) count = cursor.fetchone()[0] print(f" {table}: {count} rows") except Exception as e: print(f" {table}: ERROR - {e}") cursor.close() def print_summary(tables_ok, indexes_ok, fks_ok): """Print migration summary.""" print(f"\n{'='*60}") print("MIGRATION SUMMARY") print(f"{'='*60}") print(f" Tables: {'PASS' if tables_ok else 'FAIL'}") print(f" Indexes: {'PASS' if indexes_ok else 'FAIL'}") print(f" Foreign Keys: {'PASS' if fks_ok else 'FAIL'}") overall = tables_ok and indexes_ok and fks_ok print(f"\n Overall: {'SUCCESS' if overall else 'FAILED'}") print(f"\nCompleted at: {datetime.now().isoformat()}") print(f"{'='*60}\n") return overall def main(): """Main entry point.""" verify_only = "--verify-only" in sys.argv print("\n" + "="*60) print("GENESIS YOUTUBE SCHEMA MIGRATION") print("="*60) print(f"Host: {PostgresConfig.host}:{PostgresConfig.port}") print(f"Database: {PostgresConfig.database}") print(f"Mode: {'VERIFY ONLY' if verify_only else 'FULL MIGRATION'}") # Connect to database print("\nConnecting to PostgreSQL...") conn = get_connection() print("Connected successfully!") try: # Run migration unless verify-only if not verify_only: migration_ok = run_migration(conn) if not migration_ok: print("\nMigration failed. Exiting.") return 1 # Verify tables tables_ok = verify_tables(conn) # Verify indexes indexes_ok = verify_indexes(conn) # Verify foreign keys fks_ok = verify_foreign_keys(conn) # Show row counts get_table_row_counts(conn) # Print summary overall_ok = print_summary(tables_ok, indexes_ok, fks_ok) return 0 if overall_ok else 1 finally: conn.close() print("Database connection closed.") if __name__ == "__main__": sys.exit(main())