8 Phases refactor
This commit is contained in:
IGNY8 VPS (Salman)
204
backend/verify_migrations.py
Normal file
204
backend/verify_migrations.py
Normal file
@@ -0,0 +1,204 @@
|
||||
#!/usr/bin/env python3
|
||||
"""
|
||||
Database Migration Verification Script
|
||||
Checks for orphaned SiteBlueprint tables and verifies new migrations
|
||||
"""
|
||||
|
||||
import os
|
||||
import sys
|
||||
import django
|
||||
|
||||
# Setup Django
|
||||
os.environ.setdefault('DJANGO_SETTINGS_MODULE', 'igny8_core.settings')
|
||||
django.setup()
|
||||
|
||||
from django.db import connection
|
||||
from django.core.management import call_command
|
||||
|
||||
|
||||
def check_orphaned_tables():
|
||||
"""Check for orphaned blueprint tables"""
|
||||
print("\n" + "="*60)
|
||||
print("CHECKING FOR ORPHANED SITEBLUEPRINT TABLES")
|
||||
print("="*60 + "\n")
|
||||
|
||||
with connection.cursor() as cursor:
|
||||
cursor.execute("""
|
||||
SELECT table_name
|
||||
FROM information_schema.tables
|
||||
WHERE table_schema = 'public'
|
||||
AND table_name LIKE '%blueprint%'
|
||||
ORDER BY table_name;
|
||||
""")
|
||||
tables = cursor.fetchall()
|
||||
|
||||
if tables:
|
||||
print("⚠️ Found blueprint-related tables:")
|
||||
for table in tables:
|
||||
print(f" - {table[0]}")
|
||||
print("\n💡 These tables can be safely dropped if no longer needed.")
|
||||
else:
|
||||
print("✅ No orphaned blueprint tables found.")
|
||||
|
||||
return len(tables) if tables else 0
|
||||
|
||||
|
||||
def verify_cluster_constraint():
|
||||
"""Verify cluster unique constraint is per-site/sector"""
|
||||
print("\n" + "="*60)
|
||||
print("VERIFYING CLUSTER UNIQUE CONSTRAINT")
|
||||
print("="*60 + "\n")
|
||||
|
||||
with connection.cursor() as cursor:
|
||||
cursor.execute("""
|
||||
SELECT
|
||||
tc.constraint_name,
|
||||
tc.constraint_type,
|
||||
string_agg(kcu.column_name, ', ' ORDER BY kcu.ordinal_position) as columns
|
||||
FROM information_schema.table_constraints tc
|
||||
JOIN information_schema.key_column_usage kcu
|
||||
ON tc.constraint_name = kcu.constraint_name
|
||||
AND tc.table_schema = kcu.table_schema
|
||||
WHERE tc.table_name = 'igny8_clusters'
|
||||
AND tc.constraint_type = 'UNIQUE'
|
||||
GROUP BY tc.constraint_name, tc.constraint_type;
|
||||
""")
|
||||
constraints = cursor.fetchall()
|
||||
|
||||
if constraints:
|
||||
print("Found unique constraints on igny8_clusters:")
|
||||
for constraint in constraints:
|
||||
name, ctype, columns = constraint
|
||||
print(f" {name}: {columns}")
|
||||
|
||||
# Check if it includes site and sector
|
||||
if 'site' in columns.lower() and 'sector' in columns.lower():
|
||||
print(f" ✅ Constraint is scoped per-site/sector")
|
||||
else:
|
||||
print(f" ⚠️ Constraint may need updating")
|
||||
else:
|
||||
print("⚠️ No unique constraints found on igny8_clusters")
|
||||
|
||||
|
||||
def verify_automation_delays():
|
||||
"""Verify automation delay fields exist"""
|
||||
print("\n" + "="*60)
|
||||
print("VERIFYING AUTOMATION DELAY CONFIGURATION")
|
||||
print("="*60 + "\n")
|
||||
|
||||
with connection.cursor() as cursor:
|
||||
cursor.execute("""
|
||||
SELECT
|
||||
column_name,
|
||||
data_type,
|
||||
column_default
|
||||
FROM information_schema.columns
|
||||
WHERE table_name = 'igny8_automationconfig'
|
||||
AND column_name IN ('within_stage_delay', 'between_stage_delay')
|
||||
ORDER BY column_name;
|
||||
""")
|
||||
columns = cursor.fetchall()
|
||||
|
||||
if len(columns) == 2:
|
||||
print("✅ Delay configuration fields found:")
|
||||
for col in columns:
|
||||
name, dtype, default = col
|
||||
print(f" {name}: {dtype} (default: {default})")
|
||||
else:
|
||||
print(f"⚠️ Expected 2 delay fields, found {len(columns)}")
|
||||
|
||||
|
||||
def check_migration_status():
|
||||
"""Check migration status"""
|
||||
print("\n" + "="*60)
|
||||
print("CHECKING MIGRATION STATUS")
|
||||
print("="*60 + "\n")
|
||||
|
||||
with connection.cursor() as cursor:
|
||||
cursor.execute("""
|
||||
SELECT app, name, applied
|
||||
FROM django_migrations
|
||||
WHERE name LIKE '%cluster%' OR name LIKE '%delay%'
|
||||
ORDER BY applied DESC
|
||||
LIMIT 10;
|
||||
""")
|
||||
migrations = cursor.fetchall()
|
||||
|
||||
if migrations:
|
||||
print("Recent relevant migrations:")
|
||||
for mig in migrations:
|
||||
app, name, applied = mig
|
||||
status = "✅" if applied else "⏳"
|
||||
print(f" {status} {app}.{name}")
|
||||
print(f" Applied: {applied}")
|
||||
else:
|
||||
print("No relevant migrations found in history")
|
||||
|
||||
|
||||
def check_data_integrity():
|
||||
"""Check for data integrity issues"""
|
||||
print("\n" + "="*60)
|
||||
print("DATA INTEGRITY CHECKS")
|
||||
print("="*60 + "\n")
|
||||
|
||||
from igny8_core.business.planning.models import Clusters, Keywords
|
||||
|
||||
# Check for clusters with 'active' status (should all be 'new' or 'mapped')
|
||||
active_clusters = Clusters.objects.filter(status='active').count()
|
||||
if active_clusters > 0:
|
||||
print(f"⚠️ Found {active_clusters} clusters with status='active'")
|
||||
print(" These should be updated to 'new' or 'mapped'")
|
||||
else:
|
||||
print("✅ No clusters with invalid 'active' status")
|
||||
|
||||
# Check for duplicate cluster names in same site/sector
|
||||
with connection.cursor() as cursor:
|
||||
cursor.execute("""
|
||||
SELECT name, site_id, sector_id, COUNT(*) as count
|
||||
FROM igny8_clusters
|
||||
GROUP BY name, site_id, sector_id
|
||||
HAVING COUNT(*) > 1;
|
||||
""")
|
||||
duplicates = cursor.fetchall()
|
||||
|
||||
if duplicates:
|
||||
print(f"\n⚠️ Found {len(duplicates)} duplicate cluster names in same site/sector:")
|
||||
for dup in duplicates[:5]: # Show first 5
|
||||
print(f" - '{dup[0]}' (site={dup[1]}, sector={dup[2]}): {dup[3]} duplicates")
|
||||
else:
|
||||
print("✅ No duplicate cluster names within same site/sector")
|
||||
|
||||
|
||||
def main():
|
||||
print("\n" + "#"*60)
|
||||
print("# IGNY8 DATABASE MIGRATION VERIFICATION")
|
||||
print("# Date:", __import__('datetime').datetime.now().strftime('%Y-%m-%d %H:%M:%S'))
|
||||
print("#"*60)
|
||||
|
||||
try:
|
||||
orphaned = check_orphaned_tables()
|
||||
verify_cluster_constraint()
|
||||
verify_automation_delays()
|
||||
check_migration_status()
|
||||
check_data_integrity()
|
||||
|
||||
print("\n" + "="*60)
|
||||
print("VERIFICATION COMPLETE")
|
||||
print("="*60)
|
||||
|
||||
if orphaned > 0:
|
||||
print(f"\n⚠️ {orphaned} orphaned table(s) found - review recommended")
|
||||
else:
|
||||
print("\n✅ All verifications passed!")
|
||||
|
||||
print("\n")
|
||||
|
||||
except Exception as e:
|
||||
print(f"\n❌ ERROR: {e}")
|
||||
import traceback
|
||||
traceback.print_exc()
|
||||
sys.exit(1)
|
||||
|
||||
|
||||
if __name__ == '__main__':
|
||||
main()
|
||||
Reference in New Issue
Block a user