video-accessibility-old/backend/migrate.py

#!/usr/bin/env python3
"""
Database migration CLI for the Accessible Video Platform.

Usage:
    python migrate.py status              # Show migration status
    python migrate.py up                  # Apply all pending migrations
    python migrate.py up <version>        # Apply migrations up to version
    python migrate.py down <version>      # Rollback to version
    python migrate.py create <description> # Create new migration template
"""

import asyncio
import sys
from datetime import datetime
from pathlib import Path
from typing import Optional

from app.migrations.migrator import MigrationManager
from app.core.config import get_settings
from app.core.database import connect_to_mongo, close_mongo_connection


class MigrationCLI:
    """Command-line interface for database migrations."""
    
    def __init__(self):
        self.manager = MigrationManager()
    
    async def initialize(self):
        """Initialize database connection and migration manager"""
        await connect_to_mongo()
        await self.manager.initialize()
    
    async def status(self) -> None:
        """Show current migration status."""
        status = await self.manager.get_migration_status()
        
        print("🗃️  Database Migration Status")
        print("=" * 40)
        print(f"Total migrations: {status['total_migrations']}")
        print(f"Applied migrations: {status['applied_migrations']}")
        print(f"Pending migrations: {status['pending_migrations']}")
        print(f"Latest applied: {status['latest_applied'] or 'None'}")
        
        if status['pending_migrations'] > 0:
            print(f"⚠️  {status['pending_migrations']} migrations pending")
        else:
            print("✅ Database is up to date")
        
        print("\nApplied migrations:")
        for version in status['all_applied']:
            print(f"  ✅ {version}")
    
    async def migrate_up(self, target_version: Optional[str] = None) -> None:
        """Apply migrations."""
        print(f"🚀 Applying migrations{f' up to {target_version}' if target_version else ''}...")
        
        try:
            applied = await self.manager.migrate_up(target_version)
            
            if applied:
                print(f"✅ Successfully applied {len(applied)} migrations:")
                for version in applied:
                    print(f"  ✅ {version}")
            else:
                print("ℹ️  No migrations to apply")
                
        except Exception as e:
            print(f"❌ Migration failed: {e}")
            sys.exit(1)
    
    async def migrate_down(self, target_version: str) -> None:
        """Rollback migrations."""
        print(f"⚠️  Rolling back migrations to {target_version}...")
        print("⚠️  This operation may be destructive!")
        
        # Confirmation prompt
        response = input("Are you sure you want to proceed? (y/N): ")
        if response.lower() != 'y':
            print("❌ Rollback cancelled")
            return
        
        try:
            rolled_back = await self.manager.migrate_down(target_version)
            
            if rolled_back:
                print(f"✅ Successfully rolled back {len(rolled_back)} migrations:")
                for version in rolled_back:
                    print(f"  ⬇️  {version}")
            else:
                print("ℹ️  No migrations to rollback")
                
        except Exception as e:
            print(f"❌ Rollback failed: {e}")
            sys.exit(1)
    
    def create_migration(self, description: str) -> None:
        """Create a new migration template."""
        # Generate version from current timestamp
        now = datetime.utcnow()
        version = now.strftime("%Y-%m-%d-%H%M%S")
        
        # Sanitize description for filename
        safe_description = "".join(c if c.isalnum() or c in "-_" else "_" for c in description.lower())
        
        filename = f"migration_{version}_{safe_description}.py"
        migrations_dir = Path(__file__).parent / "app" / "migrations" / "scripts"
        migrations_dir.mkdir(parents=True, exist_ok=True)
        
        filepath = migrations_dir / filename
        
        # Create migration template
        template = f'''"""Migration: {description}."""

from app.migrations.migrator import Migration


class Migration(Migration):
    """{description}."""
    
    def __init__(self):
        super().__init__()
        self.version = "{version}"
        self.description = "{description}"
    
    async def up(self) -> None:
        """Apply the migration."""
        
        # TODO: Implement your migration logic here
        # Example:
        # await self.db.collection_name.create_index([("field", 1)])
        # await self.db.collection_name.update_many(
        #     {{"old_field": {{"$exists": True}}}},
        #     {{"$rename": {{"old_field": "new_field"}}}}
        # )
        
        print(f"✅ Applied migration {{self.version}}: {{self.description}}")
    
    async def down(self) -> None:
        """Rollback the migration."""
        
        # TODO: Implement your rollback logic here
        # Example:
        # await self.db.collection_name.drop_index("index_name")
        # await self.db.collection_name.update_many(
        #     {{"new_field": {{"$exists": True}}}},
        #     {{"$rename": {{"new_field": "old_field"}}}}
        # )
        
        print(f"⚠️  Rolled back migration {{self.version}}: {{self.description}}")
'''
        
        filepath.write_text(template)
        print(f"✅ Created migration template: {filepath}")
        print(f"📝 Edit the file to implement your migration logic")


async def main():
    """Main CLI entry point."""
    
    if len(sys.argv) < 2:
        print(__doc__)
        sys.exit(1)
    
    command = sys.argv[1]
    cli = MigrationCLI()
    
    try:
        # Initialize database connection for all commands except create
        if command != "create":
            await cli.initialize()
        
        if command == "status":
            await cli.status()
            
        elif command == "up":
            target = sys.argv[2] if len(sys.argv) > 2 else None
            await cli.migrate_up(target)
            
        elif command == "down":
            if len(sys.argv) < 3:
                print("❌ Target version required for rollback")
                sys.exit(1)
            target = sys.argv[2]
            await cli.migrate_down(target)
            
        elif command == "create":
            if len(sys.argv) < 3:
                print("❌ Description required for new migration")
                sys.exit(1)
            description = " ".join(sys.argv[2:])
            cli.create_migration(description)
            
        else:
            print(f"❌ Unknown command: {command}")
            print(__doc__)
            sys.exit(1)
    
    finally:
        # Close database connection
        await close_mongo_connection()


if __name__ == "__main__":
    asyncio.run(main())