#!/usr/bin/env python """ Updated Migration Final Simple Script ==================================== This script provides a comprehensive workflow for resetting and rebuilding Django migrations for the rogaining_srv project using a simplified approach. Usage: python migration_simple_reset.py [options] Options: --backup-only : Only create backup of existing migrations --reset-only : Only reset migrations (requires backup to exist) --apply-only : Only apply migrations (requires simple migration to exist) --full : Run complete workflow (default) Features: - Creates timestamped backup of existing migrations - Clears migration history from database - Creates simplified initial migration with only managed models - Applies migrations with proper error handling - Provides detailed logging and status reporting """ import os import sys import shutil import datetime import subprocess import json from pathlib import Path # Add Django project to path project_root = Path(__file__).parent sys.path.insert(0, str(project_root)) # Setup Django environment os.environ.setdefault('DJANGO_SETTINGS_MODULE', 'config.settings') import django django.setup() from django.core.management import execute_from_command_line from django.db import connection, transaction from django.core.management.base import CommandError class MigrationManager: def __init__(self): self.project_root = Path(__file__).parent self.migrations_dir = self.project_root / 'rog' / 'migrations' self.backup_timestamp = datetime.datetime.now().strftime('%Y%m%d_%H%M%S') self.backup_dir = self.project_root / f'rog/migrations_backup_{self.backup_timestamp}' def log(self, message, level='INFO'): """Log message with timestamp and level""" timestamp = datetime.datetime.now().strftime('%Y-%m-%d %H:%M:%S') print(f"[{timestamp}] {level}: {message}") def run_command(self, command, check=True): """Execute shell command with logging""" self.log(f"Executing: {' '.join(command)}") try: result = subprocess.run(command, capture_output=True, text=True, check=check) if result.stdout: self.log(f"STDOUT: {result.stdout.strip()}") if result.stderr: self.log(f"STDERR: {result.stderr.strip()}", 'WARNING') return result except subprocess.CalledProcessError as e: self.log(f"Command failed with exit code {e.returncode}", 'ERROR') self.log(f"STDOUT: {e.stdout}", 'ERROR') self.log(f"STDERR: {e.stderr}", 'ERROR') raise def backup_migrations(self): """Create backup of existing migrations""" self.log("Creating backup of existing migrations...") if not self.migrations_dir.exists(): self.log("No migrations directory found, creating empty one") self.migrations_dir.mkdir(parents=True, exist_ok=True) (self.migrations_dir / '__init__.py').touch() return # Create backup directory self.backup_dir.mkdir(parents=True, exist_ok=True) # Copy all files for item in self.migrations_dir.iterdir(): if item.is_file(): shutil.copy2(item, self.backup_dir) self.log(f"Backed up: {item.name}") self.log(f"Backup completed: {self.backup_dir}") def clear_migration_history(self): """Clear migration history from database""" self.log("Clearing migration history from database...") try: with connection.cursor() as cursor: # Check if django_migrations table exists cursor.execute(""" SELECT COUNT(*) FROM information_schema.tables WHERE table_name = 'django_migrations' AND table_schema = 'public' """) if cursor.fetchone()[0] == 0: self.log("django_migrations table does not exist, skipping clear") return # Count existing migrations cursor.execute("SELECT COUNT(*) FROM django_migrations WHERE app = 'rog'") count = cursor.fetchone()[0] self.log(f"Found {count} existing rog migrations") # Delete rog migrations cursor.execute("DELETE FROM django_migrations WHERE app = 'rog'") self.log(f"Deleted {count} migration records") except Exception as e: self.log(f"Error clearing migration history: {e}", 'ERROR') raise def remove_migration_files(self): """Remove existing migration files except __init__.py""" self.log("Removing existing migration files...") if not self.migrations_dir.exists(): return removed_count = 0 for item in self.migrations_dir.iterdir(): if item.is_file() and item.name != '__init__.py': item.unlink() self.log(f"Removed: {item.name}") removed_count += 1 self.log(f"Removed {removed_count} migration files") def create_simple_migration(self): """Create simplified initial migration""" self.log("Creating simplified initial migration...") # Create simple migration content migration_content = '''# Generated by migration_simple_reset.py from django.contrib.gis.db import models from django.contrib.auth.models import AbstractUser from django.db import migrations import django.contrib.gis.db.models.fields import django.core.validators class Migration(migrations.Migration): initial = True dependencies = [ ('auth', '0012_alter_user_first_name_max_length'), ] operations = [ migrations.CreateModel( name='CustomUser', fields=[ ('id', models.AutoField(auto_created=True, primary_key=True, serialize=False, verbose_name='ID')), ('password', models.CharField(max_length=128, verbose_name='password')), ('last_login', models.DateTimeField(blank=True, null=True, verbose_name='last login')), ('is_superuser', models.BooleanField(default=False, help_text='Designates that this user has all permissions without explicitly assigning them.', verbose_name='superuser status')), ('username', models.CharField(error_messages={'unique': 'A user with that username already exists.'}, help_text='Required. 150 characters or fewer. Letters, digits and @/./+/-/_ only.', max_length=150, unique=True, validators=[django.contrib.auth.validators.UnicodeUsernameValidator()], verbose_name='username')), ('first_name', models.CharField(blank=True, max_length=150, verbose_name='first name')), ('last_name', models.CharField(blank=True, max_length=150, verbose_name='last name')), ('email', models.EmailField(blank=True, max_length=254, verbose_name='email address')), ('is_staff', models.BooleanField(default=False, help_text='Designates whether the user can log into this admin site.', verbose_name='staff status')), ('is_active', models.BooleanField(default=True, help_text='Designates whether this user should be treated as active. Unselect this instead of deleting accounts.', verbose_name='active')), ('date_joined', models.DateTimeField(auto_now_add=True, verbose_name='date joined')), ('groups', models.ManyToManyField(blank=True, help_text='The groups this user belongs to. A user will get all permissions granted to each of their groups.', related_name='user_set', related_query_name='user', to='auth.Group', verbose_name='groups')), ('user_permissions', models.ManyToManyField(blank=True, help_text='Specific permissions for this user.', related_name='user_set', related_query_name='user', to='auth.Permission', verbose_name='user permissions')), ], options={ 'verbose_name': 'user', 'verbose_name_plural': 'users', 'abstract': False, }, managers=[ ('objects', django.contrib.auth.models.UserManager()), ], ), migrations.CreateModel( name='Category', fields=[ ('id', models.AutoField(auto_created=True, primary_key=True, serialize=False, verbose_name='ID')), ('name', models.CharField(max_length=50)), ('description', models.TextField(blank=True, null=True)), ('created_at', models.DateTimeField(auto_now_add=True)), ('updated_at', models.DateTimeField(auto_now=True)), ], options={ 'verbose_name_plural': 'categories', }, ), migrations.CreateModel( name='NewEvent', fields=[ ('id', models.AutoField(auto_created=True, primary_key=True, serialize=False, verbose_name='ID')), ('name', models.CharField(max_length=200)), ('description', models.TextField(blank=True, null=True)), ('event_date', models.DateField()), ('start_time', models.TimeField()), ('end_time', models.TimeField()), ('event_boundary', django.contrib.gis.db.models.fields.PolygonField(blank=True, null=True, srid=4326)), ('max_participants', models.PositiveIntegerField(default=100)), ('is_active', models.BooleanField(default=True)), ('created_at', models.DateTimeField(auto_now_add=True)), ('updated_at', models.DateTimeField(auto_now=True)), ('category', models.ForeignKey(blank=True, null=True, on_delete=django.db.models.deletion.SET_NULL, to='rog.category')), ], ), migrations.CreateModel( name='Team', fields=[ ('id', models.AutoField(auto_created=True, primary_key=True, serialize=False, verbose_name='ID')), ('name', models.CharField(max_length=200)), ('max_members', models.PositiveIntegerField(default=5)), ('is_active', models.BooleanField(default=True)), ('created_at', models.DateTimeField(auto_now_add=True)), ('updated_at', models.DateTimeField(auto_now=True)), ('event', models.ForeignKey(on_delete=django.db.models.deletion.CASCADE, related_name='teams', to='rog.newevent')), ], ), migrations.CreateModel( name='Location', fields=[ ('id', models.AutoField(auto_created=True, primary_key=True, serialize=False, verbose_name='ID')), ('name', models.CharField(max_length=200)), ('description', models.TextField(blank=True, null=True)), ('coordinate', django.contrib.gis.db.models.fields.PointField(srid=4326)), ('altitude', models.FloatField(blank=True, null=True)), ('location_type', models.CharField(choices=[('checkpoint', 'Checkpoint'), ('start', 'Start Point'), ('finish', 'Finish Point'), ('water', 'Water Station'), ('emergency', 'Emergency Point'), ('viewpoint', 'View Point'), ('other', 'Other')], default='checkpoint', max_length=20)), ('points', models.IntegerField(default=0)), ('is_active', models.BooleanField(default=True)), ('created_at', models.DateTimeField(auto_now_add=True)), ('updated_at', models.DateTimeField(auto_now=True)), ('event', models.ForeignKey(on_delete=django.db.models.deletion.CASCADE, related_name='locations', to='rog.newevent')), ], ), migrations.CreateModel( name='Entry', fields=[ ('id', models.AutoField(auto_created=True, primary_key=True, serialize=False, verbose_name='ID')), ('role', models.CharField(choices=[('leader', 'Team Leader'), ('member', 'Team Member')], default='member', max_length=10)), ('joined_at', models.DateTimeField(auto_now_add=True)), ('is_active', models.BooleanField(default=True)), ('team', models.ForeignKey(on_delete=django.db.models.deletion.CASCADE, related_name='entries', to='rog.team')), ('user', models.ForeignKey(on_delete=django.db.models.deletion.CASCADE, to='rog.customuser')), ], options={ 'verbose_name_plural': 'entries', 'unique_together': {('user', 'team')}, }, ), ] ''' # Write migration file migration_file = self.migrations_dir / '0001_simple_initial.py' with open(migration_file, 'w', encoding='utf-8') as f: f.write(migration_content) self.log(f"Created: {migration_file}") def apply_migrations(self): """Apply migrations using Django management command""" self.log("Applying migrations...") try: # Use fake-initial to treat as initial migration result = self.run_command([ 'python', 'manage.py', 'migrate', '--fake-initial' ], check=False) if result.returncode == 0: self.log("Migrations applied successfully") else: self.log("Migration application failed", 'ERROR') raise CommandError("Migration application failed") except Exception as e: self.log(f"Error applying migrations: {e}", 'ERROR') raise def check_status(self): """Check current migration status""" self.log("Checking migration status...") try: result = self.run_command([ 'python', 'manage.py', 'showmigrations', 'rog' ], check=False) if result.returncode == 0: self.log("Migration status checked successfully") else: self.log("Failed to check migration status", 'WARNING') except Exception as e: self.log(f"Error checking migration status: {e}", 'WARNING') def run_full_workflow(self): """Execute complete migration reset workflow""" self.log("Starting full migration reset workflow...") try: # Step 1: Backup existing migrations self.backup_migrations() # Step 2: Clear migration history self.clear_migration_history() # Step 3: Remove migration files self.remove_migration_files() # Step 4: Create simplified migration self.create_simple_migration() # Step 5: Apply migrations self.apply_migrations() # Step 6: Check final status self.check_status() self.log("Full migration reset workflow completed successfully!") self.log(f"Backup created at: {self.backup_dir}") except Exception as e: self.log(f"Workflow failed: {e}", 'ERROR') self.log(f"You can restore from backup at: {self.backup_dir}", 'INFO') raise def run_backup_only(self): """Create backup only""" self.log("Running backup-only mode...") self.backup_migrations() self.log("Backup completed successfully!") def run_reset_only(self): """Reset migrations only (requires backup)""" self.log("Running reset-only mode...") self.clear_migration_history() self.remove_migration_files() self.create_simple_migration() self.log("Reset completed successfully!") def run_apply_only(self): """Apply migrations only""" self.log("Running apply-only mode...") self.apply_migrations() self.check_status() self.log("Apply completed successfully!") def main(): """Main function with command line argument handling""" import argparse parser = argparse.ArgumentParser( description='Reset and rebuild Django migrations for rogaining_srv project' ) parser.add_argument( '--backup-only', action='store_true', help='Only create backup of existing migrations' ) parser.add_argument( '--reset-only', action='store_true', help='Only reset migrations (requires backup to exist)' ) parser.add_argument( '--apply-only', action='store_true', help='Only apply migrations (requires simple migration to exist)' ) parser.add_argument( '--full', action='store_true', help='Run complete workflow (default)' ) args = parser.parse_args() # If no specific mode is selected, default to full if not any([args.backup_only, args.reset_only, args.apply_only, args.full]): args.full = True manager = MigrationManager() try: if args.backup_only: manager.run_backup_only() elif args.reset_only: manager.run_reset_only() elif args.apply_only: manager.run_apply_only() elif args.full: manager.run_full_workflow() except Exception as e: print(f"\nError: {e}") print("Please check the logs above for details.") sys.exit(1) if __name__ == '__main__': main()