Change db to Mysql
This commit is contained in:
@@ -32,7 +32,7 @@ class User(db.Model):
|
||||
full_name = db.Column(db.String(100), nullable=False)
|
||||
email = db.Column(db.String(120), unique=True, nullable=False)
|
||||
username = db.Column(db.String(80), unique=True, nullable=False)
|
||||
password_hash = db.Column(db.String(128), nullable=False)
|
||||
password_hash = db.Column(db.String(255), nullable=False)
|
||||
role = db.Column(db.String(20), nullable=False, default='staff') # admin or staff
|
||||
created_by = db.Column(db.Integer, db.ForeignKey('users.id'), nullable=True)
|
||||
created_date = db.Column(db.DateTime, default=datetime.utcnow)
|
||||
@@ -83,6 +83,7 @@ class QRCode(db.Model):
|
||||
def has_coordinates(self):
|
||||
"""Check if this QR code has address coordinates"""
|
||||
return self.address_latitude is not None and self.address_longitude is not None
|
||||
|
||||
@property
|
||||
def coordinates_display(self):
|
||||
"""Get formatted coordinates for display"""
|
||||
@@ -854,19 +855,24 @@ def migrate_to_enhanced_location_accuracy():
|
||||
return False
|
||||
|
||||
def check_location_accuracy_column_exists():
|
||||
"""Check if the location_accuracy column exists in the attendance_data table"""
|
||||
"""
|
||||
Check if location_accuracy column exists in attendance_data table (MySQL compatible)
|
||||
"""
|
||||
try:
|
||||
# MySQL-compatible query for checking column existence
|
||||
result = db.session.execute(text("""
|
||||
SELECT column_name
|
||||
FROM information_schema.columns
|
||||
WHERE table_name='attendance_data' AND column_name='location_accuracy'
|
||||
SELECT COUNT(*) as count
|
||||
FROM INFORMATION_SCHEMA.COLUMNS
|
||||
WHERE TABLE_SCHEMA = DATABASE()
|
||||
AND TABLE_NAME = 'attendance_data'
|
||||
AND COLUMN_NAME = 'location_accuracy'
|
||||
"""))
|
||||
|
||||
column_exists = result.fetchone() is not None
|
||||
return column_exists
|
||||
count = result.fetchone().count
|
||||
return count > 0
|
||||
|
||||
except Exception as e:
|
||||
print(f"⚠️ Error checking location_accuracy column: {e}")
|
||||
print(f"Error checking location_accuracy column: {e}")
|
||||
return False
|
||||
|
||||
def get_employee_checkin_history(employee_id, qr_code_id, date_filter=None):
|
||||
@@ -2667,17 +2673,18 @@ def update_existing_qr_codes():
|
||||
db.session.rollback()
|
||||
|
||||
def add_coordinate_columns():
|
||||
"""Add coordinate columns to existing qr_codes table"""
|
||||
"""Add coordinate columns to existing qr_codes table (MySQL compatible)"""
|
||||
try:
|
||||
# Check if columns already exist
|
||||
# Check if columns already exist - MySQL compatible query
|
||||
result = db.session.execute(text("""
|
||||
SELECT column_name
|
||||
FROM information_schema.columns
|
||||
WHERE table_name='qr_codes' AND column_name IN
|
||||
('address_latitude', 'address_longitude', 'coordinate_accuracy', 'coordinates_updated_date')
|
||||
SELECT COLUMN_NAME
|
||||
FROM INFORMATION_SCHEMA.COLUMNS
|
||||
WHERE TABLE_SCHEMA = DATABASE()
|
||||
AND TABLE_NAME = 'qr_codes'
|
||||
AND COLUMN_NAME IN ('address_latitude', 'address_longitude', 'coordinate_accuracy', 'coordinates_updated_date')
|
||||
"""))
|
||||
|
||||
existing_columns = [row.column_name for row in result.fetchall()]
|
||||
existing_columns = [row.COLUMN_NAME for row in result.fetchall()]
|
||||
|
||||
# Add missing columns
|
||||
if 'address_latitude' not in existing_columns:
|
||||
|
||||
+2721
File diff suppressed because it is too large
Load Diff
@@ -0,0 +1,586 @@
|
||||
#!/usr/bin/env python3
|
||||
"""
|
||||
MySQL Database Verification Script for QR Attendance System
|
||||
|
||||
This script verifies that the MySQL database migration was successful
|
||||
and all functionality is working correctly.
|
||||
|
||||
Usage:
|
||||
python verify_mysql_database.py
|
||||
|
||||
This script will:
|
||||
1. Test MySQL database connection
|
||||
2. Verify all tables exist
|
||||
3. Check data integrity
|
||||
4. Test application functionality
|
||||
5. Validate performance
|
||||
|
||||
Author: QR Attendance System Verification Team
|
||||
Version: 1.0
|
||||
"""
|
||||
|
||||
import sys
|
||||
import os
|
||||
from datetime import datetime, date, timedelta
|
||||
import time
|
||||
|
||||
# Add your app directory to path
|
||||
sys.path.append(os.path.dirname(os.path.abspath(__file__)))
|
||||
|
||||
try:
|
||||
from app import app, db, User, QRCode, AttendanceData
|
||||
from app import get_location_accuracy_level_enhanced, calculate_location_accuracy_enhanced
|
||||
from sqlalchemy import text, inspect
|
||||
except ImportError as e:
|
||||
print(f"❌ Error importing modules: {e}")
|
||||
sys.exit(1)
|
||||
|
||||
class MySQLVerifier:
|
||||
"""MySQL database verification class"""
|
||||
|
||||
def __init__(self):
|
||||
self.app = app
|
||||
self.issues = []
|
||||
self.test_results = {}
|
||||
|
||||
def log_result(self, test_name, passed, message=""):
|
||||
"""Log test result"""
|
||||
self.test_results[test_name] = {
|
||||
'passed': passed,
|
||||
'message': message,
|
||||
'timestamp': datetime.now()
|
||||
}
|
||||
|
||||
if not passed:
|
||||
self.issues.append(f"{test_name}: {message}")
|
||||
|
||||
def test_database_connection(self):
|
||||
"""Test basic MySQL database connectivity"""
|
||||
print("🔌 Testing MySQL Database Connection...")
|
||||
|
||||
try:
|
||||
with self.app.app_context():
|
||||
# Test basic connection
|
||||
result = db.session.execute(text("SELECT 1 as test")).fetchone()
|
||||
|
||||
if result.test == 1:
|
||||
print(" ✅ MySQL connection successful")
|
||||
|
||||
# Get MySQL version
|
||||
version_result = db.session.execute(text("SELECT VERSION() as version")).fetchone()
|
||||
print(f" 📊 MySQL Version: {version_result.version}")
|
||||
|
||||
# Get database name
|
||||
db_result = db.session.execute(text("SELECT DATABASE() as db_name")).fetchone()
|
||||
print(f" 📊 Database: {db_result.db_name}")
|
||||
|
||||
self.log_result("database_connection", True, "MySQL connection working")
|
||||
return True
|
||||
else:
|
||||
self.log_result("database_connection", False, "Connection test failed")
|
||||
return False
|
||||
|
||||
except Exception as e:
|
||||
print(f" ❌ MySQL connection failed: {e}")
|
||||
self.log_result("database_connection", False, str(e))
|
||||
return False
|
||||
|
||||
def test_table_structure(self):
|
||||
"""Verify all required tables exist with correct structure"""
|
||||
print("\n📋 Testing Table Structure...")
|
||||
|
||||
try:
|
||||
with self.app.app_context():
|
||||
inspector = inspect(db.engine)
|
||||
tables = inspector.get_table_names()
|
||||
|
||||
required_tables = ['users', 'qr_codes', 'attendance_data']
|
||||
|
||||
print(f" 📊 Found tables: {', '.join(tables)}")
|
||||
|
||||
all_tables_exist = True
|
||||
for table in required_tables:
|
||||
if table in tables:
|
||||
print(f" ✅ {table} table exists")
|
||||
|
||||
# Check key columns
|
||||
columns = inspector.get_columns(table)
|
||||
column_names = [col['name'] for col in columns]
|
||||
|
||||
if table == 'users':
|
||||
required_cols = ['id', 'username', 'email', 'password_hash', 'role']
|
||||
elif table == 'qr_codes':
|
||||
required_cols = ['id', 'name', 'location', 'qr_code_image']
|
||||
elif table == 'attendance_data':
|
||||
required_cols = ['id', 'qr_code_id', 'employee_id', 'check_in_date']
|
||||
|
||||
missing_cols = [col for col in required_cols if col not in column_names]
|
||||
if missing_cols:
|
||||
print(f" ⚠️ Missing columns in {table}: {missing_cols}")
|
||||
all_tables_exist = False
|
||||
else:
|
||||
print(f" ✅ {table} has all required columns")
|
||||
else:
|
||||
print(f" ❌ {table} table missing")
|
||||
all_tables_exist = False
|
||||
|
||||
self.log_result("table_structure", all_tables_exist,
|
||||
"All tables exist" if all_tables_exist else "Missing tables/columns")
|
||||
return all_tables_exist
|
||||
|
||||
except Exception as e:
|
||||
print(f" ❌ Error checking table structure: {e}")
|
||||
self.log_result("table_structure", False, str(e))
|
||||
return False
|
||||
|
||||
def test_data_integrity(self):
|
||||
"""Test data integrity and relationships"""
|
||||
print("\n🔍 Testing Data Integrity...")
|
||||
|
||||
try:
|
||||
with self.app.app_context():
|
||||
# Count records
|
||||
user_count = User.query.count()
|
||||
qr_count = QRCode.query.count()
|
||||
attendance_count = AttendanceData.query.count()
|
||||
|
||||
print(f" 📊 Record counts:")
|
||||
print(f" Users: {user_count:,}")
|
||||
print(f" QR Codes: {qr_count:,}")
|
||||
print(f" Attendance: {attendance_count:,}")
|
||||
|
||||
# Test relationships
|
||||
relationship_issues = []
|
||||
|
||||
# Check foreign key relationships
|
||||
orphaned_qr_codes = db.session.execute(text("""
|
||||
SELECT COUNT(*) as count FROM qr_codes qr
|
||||
LEFT JOIN users u ON qr.created_by = u.id
|
||||
WHERE u.id IS NULL AND qr.created_by IS NOT NULL
|
||||
""")).fetchone().count
|
||||
|
||||
if orphaned_qr_codes > 0:
|
||||
relationship_issues.append(f"{orphaned_qr_codes} QR codes with invalid creator")
|
||||
print(f" ⚠️ {orphaned_qr_codes} QR codes with invalid creator references")
|
||||
|
||||
orphaned_attendance = db.session.execute(text("""
|
||||
SELECT COUNT(*) as count FROM attendance_data ad
|
||||
LEFT JOIN qr_codes qr ON ad.qr_code_id = qr.id
|
||||
WHERE qr.id IS NULL
|
||||
""")).fetchone().count
|
||||
|
||||
if orphaned_attendance > 0:
|
||||
relationship_issues.append(f"{orphaned_attendance} attendance records with invalid QR code")
|
||||
print(f" ⚠️ {orphaned_attendance} attendance records with invalid QR code references")
|
||||
|
||||
# Check for duplicate usernames/emails
|
||||
duplicate_usernames = db.session.execute(text("""
|
||||
SELECT username, COUNT(*) as count FROM users
|
||||
GROUP BY username HAVING COUNT(*) > 1
|
||||
""")).fetchall()
|
||||
|
||||
if duplicate_usernames:
|
||||
relationship_issues.append(f"Duplicate usernames found")
|
||||
print(f" ⚠️ Duplicate usernames: {[row.username for row in duplicate_usernames]}")
|
||||
|
||||
# Check location accuracy data
|
||||
attendance_with_location = AttendanceData.query.filter(
|
||||
AttendanceData.latitude.isnot(None)
|
||||
).count()
|
||||
|
||||
print(f" 📊 Attendance records with location data: {attendance_with_location:,}")
|
||||
|
||||
if len(relationship_issues) == 0:
|
||||
print(" ✅ All data integrity checks passed")
|
||||
self.log_result("data_integrity", True, "Data integrity verified")
|
||||
return True
|
||||
else:
|
||||
print(f" ❌ Data integrity issues found")
|
||||
self.log_result("data_integrity", False, "; ".join(relationship_issues))
|
||||
return False
|
||||
|
||||
except Exception as e:
|
||||
print(f" ❌ Error checking data integrity: {e}")
|
||||
self.log_result("data_integrity", False, str(e))
|
||||
return False
|
||||
|
||||
def test_mysql_specific_features(self):
|
||||
"""Test MySQL-specific features and compatibility"""
|
||||
print("\n🔧 Testing MySQL-Specific Features...")
|
||||
|
||||
try:
|
||||
with self.app.app_context():
|
||||
# Test MySQL engine type
|
||||
engine_result = db.session.execute(text("""
|
||||
SELECT ENGINE FROM information_schema.TABLES
|
||||
WHERE TABLE_SCHEMA = DATABASE() AND TABLE_NAME = 'users'
|
||||
""")).fetchone()
|
||||
|
||||
if engine_result:
|
||||
engine = engine_result.ENGINE
|
||||
print(f" 📊 MySQL Storage Engine: {engine}")
|
||||
|
||||
if engine == 'InnoDB':
|
||||
print(" ✅ Using InnoDB engine (supports transactions)")
|
||||
else:
|
||||
print(f" ⚠️ Using {engine} engine (consider InnoDB for ACID compliance)")
|
||||
|
||||
# Test character set
|
||||
charset_result = db.session.execute(text("""
|
||||
SELECT DEFAULT_CHARACTER_SET_NAME, DEFAULT_COLLATION_NAME
|
||||
FROM information_schema.SCHEMATA
|
||||
WHERE SCHEMA_NAME = DATABASE()
|
||||
""")).fetchone()
|
||||
|
||||
if charset_result:
|
||||
charset = charset_result.DEFAULT_CHARACTER_SET_NAME
|
||||
collation = charset_result.DEFAULT_COLLATION_NAME
|
||||
print(f" 📊 Character Set: {charset}, Collation: {collation}")
|
||||
|
||||
if charset in ['utf8mb4', 'utf8']:
|
||||
print(" ✅ UTF-8 character set configured")
|
||||
else:
|
||||
print(f" ⚠️ Consider using utf8mb4 character set")
|
||||
|
||||
# Test auto-increment functionality
|
||||
test_user = User(
|
||||
full_name='Test User',
|
||||
email='test@example.com',
|
||||
username=f'testuser_{int(time.time())}',
|
||||
role='staff'
|
||||
)
|
||||
test_user.set_password('testpass')
|
||||
|
||||
db.session.add(test_user)
|
||||
db.session.commit()
|
||||
|
||||
if test_user.id:
|
||||
print(f" ✅ Auto-increment working (generated ID: {test_user.id})")
|
||||
|
||||
# Clean up test user
|
||||
db.session.delete(test_user)
|
||||
db.session.commit()
|
||||
|
||||
self.log_result("mysql_features", True, "MySQL features working correctly")
|
||||
return True
|
||||
else:
|
||||
print(" ❌ Auto-increment not working")
|
||||
self.log_result("mysql_features", False, "Auto-increment failed")
|
||||
return False
|
||||
|
||||
except Exception as e:
|
||||
print(f" ❌ Error testing MySQL features: {e}")
|
||||
self.log_result("mysql_features", False, str(e))
|
||||
return False
|
||||
|
||||
def test_application_functionality(self):
|
||||
"""Test core application functionality"""
|
||||
print("\n🧪 Testing Application Functionality...")
|
||||
|
||||
try:
|
||||
with self.app.app_context():
|
||||
# Test user authentication functions
|
||||
admin_user = User.query.filter_by(role='admin').first()
|
||||
if admin_user:
|
||||
print(f" ✅ Admin user found: {admin_user.username}")
|
||||
|
||||
# Test password hashing
|
||||
if admin_user.check_password('admin123'): # Default password
|
||||
print(" ⚠️ Default admin password detected - please change it")
|
||||
|
||||
print(" ✅ Password hashing functionality working")
|
||||
else:
|
||||
print(" ⚠️ No admin user found")
|
||||
|
||||
# Test QR code functionality
|
||||
qr_code_sample = QRCode.query.first()
|
||||
if qr_code_sample:
|
||||
print(f" ✅ QR code data accessible")
|
||||
|
||||
# Test coordinate functionality if available
|
||||
if qr_code_sample.has_coordinates:
|
||||
print(f" ✅ QR code coordinates available: {qr_code_sample.coordinates_display}")
|
||||
else:
|
||||
print(" ℹ️ QR code coordinates not set (optional feature)")
|
||||
|
||||
# Test attendance functionality
|
||||
attendance_sample = AttendanceData.query.first()
|
||||
if attendance_sample:
|
||||
print(f" ✅ Attendance data accessible")
|
||||
|
||||
# Test location accuracy if available
|
||||
if attendance_sample.has_location_data:
|
||||
accuracy_level = attendance_sample.location_accuracy_level
|
||||
print(f" ✅ Location accuracy calculation working: {accuracy_level}")
|
||||
else:
|
||||
print(" ℹ️ No location data in sample (feature works when GPS available)")
|
||||
|
||||
# Test relationships
|
||||
if qr_code_sample and attendance_sample:
|
||||
qr_with_attendance = QRCode.query.join(AttendanceData).first()
|
||||
if qr_with_attendance:
|
||||
print(" ✅ Database relationships working correctly")
|
||||
else:
|
||||
print(" ⚠️ No QR codes with attendance records found")
|
||||
|
||||
self.log_result("application_functionality", True, "Core functionality verified")
|
||||
return True
|
||||
|
||||
except Exception as e:
|
||||
print(f" ❌ Error testing application functionality: {e}")
|
||||
self.log_result("application_functionality", False, str(e))
|
||||
return False
|
||||
|
||||
def test_performance(self):
|
||||
"""Test basic database performance"""
|
||||
print("\n⚡ Testing Database Performance...")
|
||||
|
||||
try:
|
||||
with self.app.app_context():
|
||||
# Test query performance
|
||||
start_time = time.time()
|
||||
|
||||
# Complex join query
|
||||
complex_query = db.session.query(AttendanceData, QRCode, User).join(
|
||||
QRCode, AttendanceData.qr_code_id == QRCode.id
|
||||
).join(
|
||||
User, QRCode.created_by == User.id
|
||||
).limit(100).all()
|
||||
|
||||
query_time = time.time() - start_time
|
||||
print(f" 📊 Complex join query time: {query_time:.3f} seconds")
|
||||
|
||||
if query_time < 1.0:
|
||||
print(" ✅ Query performance acceptable")
|
||||
performance_ok = True
|
||||
else:
|
||||
print(" ⚠️ Query performance slow - consider adding indexes")
|
||||
performance_ok = False
|
||||
|
||||
# Test bulk operations
|
||||
start_time = time.time()
|
||||
bulk_count = AttendanceData.query.count()
|
||||
count_time = time.time() - start_time
|
||||
|
||||
print(f" 📊 Count query time ({bulk_count:,} records): {count_time:.3f} seconds")
|
||||
|
||||
# Test connection pooling
|
||||
start_time = time.time()
|
||||
for i in range(10):
|
||||
db.session.execute(text("SELECT 1")).fetchone()
|
||||
pool_time = time.time() - start_time
|
||||
|
||||
print(f" 📊 Connection pool test (10 queries): {pool_time:.3f} seconds")
|
||||
|
||||
self.log_result("performance", performance_ok,
|
||||
f"Query time: {query_time:.3f}s" if performance_ok else "Performance issues detected")
|
||||
return performance_ok
|
||||
|
||||
except Exception as e:
|
||||
print(f" ❌ Error testing performance: {e}")
|
||||
self.log_result("performance", False, str(e))
|
||||
return False
|
||||
|
||||
def test_migration_completeness(self):
|
||||
"""Test that migration preserved all data correctly"""
|
||||
print("\n📊 Testing Migration Completeness...")
|
||||
|
||||
try:
|
||||
with self.app.app_context():
|
||||
# Check for common migration issues
|
||||
issues_found = []
|
||||
|
||||
# Test datetime handling
|
||||
recent_records = AttendanceData.query.filter(
|
||||
AttendanceData.created_timestamp >= datetime.now() - timedelta(days=30)
|
||||
).count()
|
||||
|
||||
if recent_records > 0:
|
||||
print(f" ✅ Recent timestamps preserved ({recent_records} records)")
|
||||
else:
|
||||
print(" ℹ️ No recent records found (expected if migrating old data)")
|
||||
|
||||
# Test text field preservation
|
||||
qr_with_long_text = QRCode.query.filter(
|
||||
db.func.length(QRCode.qr_code_image) > 1000
|
||||
).first()
|
||||
|
||||
if qr_with_long_text:
|
||||
print(" ✅ Large text fields (QR images) preserved correctly")
|
||||
else:
|
||||
print(" ⚠️ No large text fields found - check QR code images")
|
||||
issues_found.append("QR code images may not be preserved")
|
||||
|
||||
# Test float/decimal precision
|
||||
attendance_with_coords = AttendanceData.query.filter(
|
||||
AttendanceData.latitude.isnot(None)
|
||||
).first()
|
||||
|
||||
if attendance_with_coords:
|
||||
lat_precision = len(str(attendance_with_coords.latitude).split('.')[-1]) if '.' in str(attendance_with_coords.latitude) else 0
|
||||
if lat_precision >= 6:
|
||||
print(f" ✅ Coordinate precision preserved ({lat_precision} decimal places)")
|
||||
else:
|
||||
print(f" ⚠️ Coordinate precision may be reduced ({lat_precision} decimal places)")
|
||||
issues_found.append("Coordinate precision reduced")
|
||||
|
||||
# Test boolean field handling
|
||||
active_users = User.query.filter_by(active_status=True).count()
|
||||
inactive_users = User.query.filter_by(active_status=False).count()
|
||||
|
||||
print(f" 📊 User status: {active_users} active, {inactive_users} inactive")
|
||||
if active_users > 0:
|
||||
print(" ✅ Boolean fields working correctly")
|
||||
|
||||
# Test foreign key constraints
|
||||
try:
|
||||
# Try to insert invalid foreign key
|
||||
invalid_attendance = AttendanceData(
|
||||
qr_code_id=99999, # Non-existent QR code
|
||||
employee_id='TEST',
|
||||
location_name='Test',
|
||||
check_in_date=date.today(),
|
||||
check_in_time=datetime.now().time()
|
||||
)
|
||||
db.session.add(invalid_attendance)
|
||||
db.session.commit()
|
||||
|
||||
# If we get here, foreign key constraint failed
|
||||
db.session.delete(invalid_attendance)
|
||||
db.session.commit()
|
||||
print(" ⚠️ Foreign key constraints not enforced")
|
||||
issues_found.append("Foreign key constraints not working")
|
||||
|
||||
except Exception:
|
||||
# This is expected - foreign key constraint should prevent the insert
|
||||
db.session.rollback()
|
||||
print(" ✅ Foreign key constraints working correctly")
|
||||
|
||||
migration_complete = len(issues_found) == 0
|
||||
self.log_result("migration_completeness", migration_complete,
|
||||
"Migration complete" if migration_complete else "; ".join(issues_found))
|
||||
|
||||
return migration_complete
|
||||
|
||||
except Exception as e:
|
||||
print(f" ❌ Error testing migration completeness: {e}")
|
||||
self.log_result("migration_completeness", False, str(e))
|
||||
return False
|
||||
|
||||
def generate_report(self):
|
||||
"""Generate comprehensive verification report"""
|
||||
print("\n📋 GENERATING VERIFICATION REPORT")
|
||||
print("=" * 60)
|
||||
|
||||
passed_tests = sum(1 for result in self.test_results.values() if result['passed'])
|
||||
total_tests = len(self.test_results)
|
||||
success_rate = (passed_tests / total_tests * 100) if total_tests > 0 else 0
|
||||
|
||||
print(f"📊 SUMMARY:")
|
||||
print(f" Tests Passed: {passed_tests}/{total_tests} ({success_rate:.1f}%)")
|
||||
print(f" Issues Found: {len(self.issues)}")
|
||||
|
||||
print(f"\n📝 DETAILED RESULTS:")
|
||||
for test_name, result in self.test_results.items():
|
||||
status = "✅ PASS" if result['passed'] else "❌ FAIL"
|
||||
print(f" {status} {test_name}")
|
||||
if result['message']:
|
||||
print(f" {result['message']}")
|
||||
|
||||
if self.issues:
|
||||
print(f"\n⚠️ ISSUES TO ADDRESS:")
|
||||
for i, issue in enumerate(self.issues, 1):
|
||||
print(f" {i}. {issue}")
|
||||
|
||||
print(f"\n🎯 RECOMMENDATIONS:")
|
||||
if success_rate >= 90:
|
||||
print(" ✅ Migration appears successful - ready for production use")
|
||||
print(" ✅ Perform additional application testing")
|
||||
print(" ✅ Set up regular MySQL backups")
|
||||
print(" ✅ Monitor performance in production")
|
||||
elif success_rate >= 70:
|
||||
print(" ⚠️ Migration mostly successful but has issues")
|
||||
print(" ⚠️ Address the issues listed above before production use")
|
||||
print(" ⚠️ Consider additional testing")
|
||||
else:
|
||||
print(" ❌ Migration has significant issues")
|
||||
print(" ❌ Do not use in production until issues are resolved")
|
||||
print(" ❌ Consider re-running migration process")
|
||||
|
||||
# Save report to file
|
||||
report_filename = f"mysql_verification_report_{datetime.now().strftime('%Y%m%d_%H%M%S')}.txt"
|
||||
try:
|
||||
with open(report_filename, 'w') as f:
|
||||
f.write(f"MySQL Database Verification Report\n")
|
||||
f.write(f"Generated: {datetime.now().isoformat()}\n")
|
||||
f.write(f"=" * 50 + "\n\n")
|
||||
|
||||
f.write(f"Summary:\n")
|
||||
f.write(f"Tests Passed: {passed_tests}/{total_tests} ({success_rate:.1f}%)\n")
|
||||
f.write(f"Issues Found: {len(self.issues)}\n\n")
|
||||
|
||||
f.write(f"Detailed Results:\n")
|
||||
for test_name, result in self.test_results.items():
|
||||
status = "PASS" if result['passed'] else "FAIL"
|
||||
f.write(f"{status}: {test_name}\n")
|
||||
if result['message']:
|
||||
f.write(f" Message: {result['message']}\n")
|
||||
f.write(f" Time: {result['timestamp'].isoformat()}\n\n")
|
||||
|
||||
if self.issues:
|
||||
f.write(f"Issues:\n")
|
||||
for i, issue in enumerate(self.issues, 1):
|
||||
f.write(f"{i}. {issue}\n")
|
||||
|
||||
print(f"\n📄 Report saved to: {report_filename}")
|
||||
|
||||
except Exception as e:
|
||||
print(f"⚠️ Could not save report to file: {e}")
|
||||
|
||||
return success_rate >= 90
|
||||
|
||||
def main():
|
||||
"""Main verification process"""
|
||||
print("MYSQL DATABASE VERIFICATION")
|
||||
print("=" * 60)
|
||||
print(f"Timestamp: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}")
|
||||
print()
|
||||
|
||||
verifier = MySQLVerifier()
|
||||
|
||||
# Run all verification tests
|
||||
tests = [
|
||||
verifier.test_database_connection,
|
||||
verifier.test_table_structure,
|
||||
verifier.test_data_integrity,
|
||||
verifier.test_mysql_specific_features,
|
||||
verifier.test_application_functionality,
|
||||
verifier.test_performance,
|
||||
verifier.test_migration_completeness
|
||||
]
|
||||
|
||||
try:
|
||||
for test in tests:
|
||||
if not test():
|
||||
print(f"\n⚠️ Test failed: {test.__name__}")
|
||||
|
||||
except KeyboardInterrupt:
|
||||
print("\n\n⚠️ Verification interrupted by user")
|
||||
return False
|
||||
|
||||
except Exception as e:
|
||||
print(f"\n❌ Unexpected error during verification: {e}")
|
||||
import traceback
|
||||
traceback.print_exc()
|
||||
return False
|
||||
|
||||
# Generate final report
|
||||
success = verifier.generate_report()
|
||||
|
||||
print(f"\nVerification completed at {datetime.now().strftime('%H:%M:%S')}")
|
||||
|
||||
return success
|
||||
|
||||
if __name__ == '__main__':
|
||||
success = main()
|
||||
sys.exit(0 if success else 1)
|
||||
@@ -0,0 +1,609 @@
|
||||
#!/usr/bin/env python3
|
||||
"""
|
||||
Fixed PostgreSQL to MySQL Migration Script with Table Creation
|
||||
|
||||
This script handles empty MySQL databases by creating tables first.
|
||||
|
||||
Usage:
|
||||
python migrate_fixed.py [options]
|
||||
|
||||
Options:
|
||||
--export-pg Export data from PostgreSQL
|
||||
--import-mysql Import data to MySQL
|
||||
--full-migrate Complete migration (export + import)
|
||||
--verify Verify migration success
|
||||
--help Show this help message
|
||||
|
||||
Author: QR Attendance System Migration Team
|
||||
Version: 1.2 (Fixed for empty databases)
|
||||
"""
|
||||
|
||||
import sys
|
||||
import os
|
||||
import json
|
||||
import argparse
|
||||
from datetime import datetime, date, time
|
||||
import tempfile
|
||||
import decimal
|
||||
|
||||
# Add app directory to path
|
||||
sys.path.append(os.path.dirname(os.path.abspath(__file__)))
|
||||
|
||||
try:
|
||||
from app import app, db, User, QRCode, AttendanceData
|
||||
from sqlalchemy import create_engine, text, inspect
|
||||
except ImportError as e:
|
||||
print(f"❌ Error importing required modules: {e}")
|
||||
print("Make sure you have installed all requirements: pip install -r requirements.txt")
|
||||
sys.exit(1)
|
||||
|
||||
class DatabaseMigrator:
|
||||
"""Main class for handling PostgreSQL to MySQL migration with table creation"""
|
||||
|
||||
def __init__(self, pg_connection_string=None, mysql_connection_string=None):
|
||||
self.pg_connection = pg_connection_string
|
||||
self.mysql_connection = mysql_connection_string or os.environ.get('DATABASE_URL')
|
||||
self.backup_dir = tempfile.mkdtemp(prefix='db_migration_')
|
||||
self.migration_stats = {
|
||||
'users': {'exported': 0, 'imported': 0},
|
||||
'qr_codes': {'exported': 0, 'imported': 0},
|
||||
'attendance_data': {'exported': 0, 'imported': 0}
|
||||
}
|
||||
|
||||
print(f"📁 Migration workspace: {self.backup_dir}")
|
||||
|
||||
def serialize_value(self, value):
|
||||
"""Convert database values to JSON-serializable format"""
|
||||
if value is None:
|
||||
return None
|
||||
elif isinstance(value, (date, datetime)):
|
||||
return value.isoformat()
|
||||
elif isinstance(value, time):
|
||||
return value.isoformat()
|
||||
elif isinstance(value, decimal.Decimal):
|
||||
return float(value)
|
||||
elif isinstance(value, (bytes, bytearray)):
|
||||
# Handle binary data (like QR code images stored as binary)
|
||||
try:
|
||||
return value.decode('utf-8')
|
||||
except UnicodeDecodeError:
|
||||
import base64
|
||||
return base64.b64encode(value).decode('utf-8')
|
||||
else:
|
||||
return value
|
||||
|
||||
def deserialize_value(self, value, field_name):
|
||||
"""Convert JSON values back to appropriate Python types"""
|
||||
if value is None:
|
||||
return None
|
||||
|
||||
# Handle datetime fields
|
||||
datetime_fields = ['created_date', 'last_login_date', 'coordinates_updated_date',
|
||||
'created_timestamp', 'updated_timestamp']
|
||||
date_fields = ['check_in_date']
|
||||
time_fields = ['check_in_time']
|
||||
|
||||
if field_name in datetime_fields and isinstance(value, str):
|
||||
try:
|
||||
return datetime.fromisoformat(value.replace('Z', '+00:00'))
|
||||
except ValueError:
|
||||
return datetime.fromisoformat(value)
|
||||
elif field_name in date_fields and isinstance(value, str):
|
||||
return datetime.fromisoformat(value).date()
|
||||
elif field_name in time_fields and isinstance(value, str):
|
||||
if 'T' in value: # Full datetime string
|
||||
return datetime.fromisoformat(value).time()
|
||||
else: # Time-only string
|
||||
return datetime.strptime(value, '%H:%M:%S').time()
|
||||
|
||||
return value
|
||||
|
||||
def check_and_create_tables(self):
|
||||
"""Check if tables exist in MySQL and create them if needed"""
|
||||
print("🔧 Checking and creating MySQL tables...")
|
||||
|
||||
try:
|
||||
with app.app_context():
|
||||
# Check if tables exist
|
||||
inspector = inspect(db.engine)
|
||||
existing_tables = inspector.get_table_names()
|
||||
|
||||
required_tables = ['users', 'qr_codes', 'attendance_data']
|
||||
missing_tables = [table for table in required_tables if table not in existing_tables]
|
||||
|
||||
if missing_tables:
|
||||
print(f" 📋 Missing tables: {', '.join(missing_tables)}")
|
||||
print(" 🔨 Creating database tables...")
|
||||
|
||||
# Create all tables
|
||||
db.create_all()
|
||||
|
||||
# Verify creation
|
||||
inspector = inspect(db.engine)
|
||||
new_tables = inspector.get_table_names()
|
||||
created_tables = [table for table in required_tables if table in new_tables]
|
||||
|
||||
if len(created_tables) == len(required_tables):
|
||||
print(f" ✅ Successfully created tables: {', '.join(created_tables)}")
|
||||
return True
|
||||
else:
|
||||
print(f" ❌ Failed to create some tables")
|
||||
return False
|
||||
else:
|
||||
print(f" ✅ All required tables exist: {', '.join(existing_tables)}")
|
||||
return True
|
||||
|
||||
except Exception as e:
|
||||
print(f" ❌ Error creating tables: {e}")
|
||||
import traceback
|
||||
traceback.print_exc()
|
||||
return False
|
||||
|
||||
def safe_count_records(self, model_class):
|
||||
"""Safely count records, returning 0 if table doesn't exist"""
|
||||
try:
|
||||
with app.app_context():
|
||||
return model_class.query.count()
|
||||
except Exception as e:
|
||||
if "doesn't exist" in str(e) or "does not exist" in str(e):
|
||||
return 0
|
||||
else:
|
||||
raise e
|
||||
|
||||
def export_postgresql_data(self):
|
||||
"""Export data from PostgreSQL database using native SQL"""
|
||||
print("🔄 EXPORTING POSTGRESQL DATA")
|
||||
print("=" * 50)
|
||||
|
||||
if not self.pg_connection:
|
||||
print("❌ PostgreSQL connection string not provided")
|
||||
return False
|
||||
|
||||
try:
|
||||
# Create PostgreSQL engine
|
||||
pg_engine = create_engine(self.pg_connection)
|
||||
|
||||
# Export Users table
|
||||
print("📤 Exporting users table...")
|
||||
with pg_engine.connect() as conn:
|
||||
result = conn.execute(text("SELECT * FROM users ORDER BY id"))
|
||||
users_data = []
|
||||
|
||||
for row in result:
|
||||
user_dict = {}
|
||||
for key, value in row._mapping.items():
|
||||
user_dict[key] = self.serialize_value(value)
|
||||
users_data.append(user_dict)
|
||||
|
||||
users_file = os.path.join(self.backup_dir, 'users.json')
|
||||
with open(users_file, 'w') as f:
|
||||
json.dump(users_data, f, indent=2)
|
||||
|
||||
self.migration_stats['users']['exported'] = len(users_data)
|
||||
print(f" ✅ Exported {len(users_data)} user records")
|
||||
|
||||
# Export QR Codes table
|
||||
print("📤 Exporting qr_codes table...")
|
||||
with pg_engine.connect() as conn:
|
||||
result = conn.execute(text("SELECT * FROM qr_codes ORDER BY id"))
|
||||
qr_codes_data = []
|
||||
|
||||
for row in result:
|
||||
qr_dict = {}
|
||||
for key, value in row._mapping.items():
|
||||
qr_dict[key] = self.serialize_value(value)
|
||||
qr_codes_data.append(qr_dict)
|
||||
|
||||
qr_codes_file = os.path.join(self.backup_dir, 'qr_codes.json')
|
||||
with open(qr_codes_file, 'w') as f:
|
||||
json.dump(qr_codes_data, f, indent=2)
|
||||
|
||||
self.migration_stats['qr_codes']['exported'] = len(qr_codes_data)
|
||||
print(f" ✅ Exported {len(qr_codes_data)} QR code records")
|
||||
|
||||
# Export Attendance Data table
|
||||
print("📤 Exporting attendance_data table...")
|
||||
with pg_engine.connect() as conn:
|
||||
result = conn.execute(text("SELECT * FROM attendance_data ORDER BY id"))
|
||||
attendance_data = []
|
||||
|
||||
for row in result:
|
||||
att_dict = {}
|
||||
for key, value in row._mapping.items():
|
||||
att_dict[key] = self.serialize_value(value)
|
||||
attendance_data.append(att_dict)
|
||||
|
||||
attendance_file = os.path.join(self.backup_dir, 'attendance_data.json')
|
||||
with open(attendance_file, 'w') as f:
|
||||
json.dump(attendance_data, f, indent=2)
|
||||
|
||||
self.migration_stats['attendance_data']['exported'] = len(attendance_data)
|
||||
print(f" ✅ Exported {len(attendance_data)} attendance records")
|
||||
|
||||
# Create metadata file
|
||||
metadata = {
|
||||
'export_timestamp': datetime.now().isoformat(),
|
||||
'source_database': 'PostgreSQL',
|
||||
'target_database': 'MySQL',
|
||||
'stats': self.migration_stats,
|
||||
'pg_connection': self.pg_connection.split('@')[1] if '@' in self.pg_connection else 'hidden'
|
||||
}
|
||||
|
||||
metadata_file = os.path.join(self.backup_dir, 'migration_metadata.json')
|
||||
with open(metadata_file, 'w') as f:
|
||||
json.dump(metadata, f, indent=2)
|
||||
|
||||
print(f"\n✅ Export completed successfully")
|
||||
print(f"📊 Total records exported: {sum(table['exported'] for table in self.migration_stats.values())}")
|
||||
print(f"📁 Export location: {self.backup_dir}")
|
||||
|
||||
return True
|
||||
|
||||
except Exception as e:
|
||||
print(f"❌ Export failed: {e}")
|
||||
import traceback
|
||||
traceback.print_exc()
|
||||
return False
|
||||
|
||||
def import_to_mysql(self):
|
||||
"""Import data to MySQL database with table creation"""
|
||||
print("\n🔄 IMPORTING TO MYSQL")
|
||||
print("=" * 50)
|
||||
|
||||
try:
|
||||
# First, ensure tables exist
|
||||
if not self.check_and_create_tables():
|
||||
print("❌ Failed to create required tables")
|
||||
return False
|
||||
|
||||
with app.app_context():
|
||||
# Check if we should clear existing data
|
||||
print("🧹 Checking existing data...")
|
||||
existing_users = self.safe_count_records(User)
|
||||
existing_qr_codes = self.safe_count_records(QRCode)
|
||||
existing_attendance = self.safe_count_records(AttendanceData)
|
||||
|
||||
print(f" Found existing data: {existing_users} users, {existing_qr_codes} QR codes, {existing_attendance} attendance records")
|
||||
|
||||
if existing_users > 0 or existing_qr_codes > 0 or existing_attendance > 0:
|
||||
response = input(" Clear existing data before import? (y/n): ").lower().strip()
|
||||
|
||||
if response == 'y':
|
||||
print(" Clearing existing data...")
|
||||
try:
|
||||
# Disable foreign key checks temporarily
|
||||
db.session.execute(text("SET FOREIGN_KEY_CHECKS = 0"))
|
||||
|
||||
# Delete in proper order (child tables first)
|
||||
db.session.execute(text("DELETE FROM attendance_data"))
|
||||
db.session.execute(text("DELETE FROM qr_codes"))
|
||||
db.session.execute(text("DELETE FROM users"))
|
||||
|
||||
# Reset auto-increment counters
|
||||
db.session.execute(text("ALTER TABLE attendance_data AUTO_INCREMENT = 1"))
|
||||
db.session.execute(text("ALTER TABLE qr_codes AUTO_INCREMENT = 1"))
|
||||
db.session.execute(text("ALTER TABLE users AUTO_INCREMENT = 1"))
|
||||
|
||||
db.session.execute(text("SET FOREIGN_KEY_CHECKS = 1"))
|
||||
db.session.commit()
|
||||
print(" ✅ Existing data cleared")
|
||||
except Exception as e:
|
||||
print(f" ⚠️ Could not clear existing data: {e}")
|
||||
db.session.rollback()
|
||||
|
||||
# Import Users
|
||||
print("📥 Importing users...")
|
||||
users_file = os.path.join(self.backup_dir, 'users.json')
|
||||
if os.path.exists(users_file):
|
||||
with open(users_file, 'r') as f:
|
||||
users_data = json.load(f)
|
||||
|
||||
imported_count = 0
|
||||
for user_data in users_data:
|
||||
try:
|
||||
# Deserialize datetime fields
|
||||
for field in list(user_data.keys()):
|
||||
user_data[field] = self.deserialize_value(user_data[field], field)
|
||||
|
||||
# Remove 'id' field to let MySQL auto-increment
|
||||
if 'id' in user_data:
|
||||
del user_data['id']
|
||||
|
||||
# Create user object - filter out None values
|
||||
user_fields = {k: v for k, v in user_data.items() if v is not None}
|
||||
user = User(**user_fields)
|
||||
db.session.add(user)
|
||||
imported_count += 1
|
||||
|
||||
except Exception as e:
|
||||
print(f" ⚠️ Error importing user {user_data.get('username', 'unknown')}: {e}")
|
||||
|
||||
try:
|
||||
db.session.commit()
|
||||
self.migration_stats['users']['imported'] = imported_count
|
||||
print(f" ✅ Imported {imported_count} user records")
|
||||
except Exception as e:
|
||||
print(f" ❌ Error committing users: {e}")
|
||||
db.session.rollback()
|
||||
return False
|
||||
else:
|
||||
print(" ⚠️ users.json not found")
|
||||
|
||||
# Import QR Codes
|
||||
print("📥 Importing qr_codes...")
|
||||
qr_codes_file = os.path.join(self.backup_dir, 'qr_codes.json')
|
||||
if os.path.exists(qr_codes_file):
|
||||
with open(qr_codes_file, 'r') as f:
|
||||
qr_codes_data = json.load(f)
|
||||
|
||||
imported_count = 0
|
||||
for qr_data in qr_codes_data:
|
||||
try:
|
||||
# Deserialize datetime fields
|
||||
for field in list(qr_data.keys()):
|
||||
qr_data[field] = self.deserialize_value(qr_data[field], field)
|
||||
|
||||
# Remove 'id' field to let MySQL auto-increment
|
||||
if 'id' in qr_data:
|
||||
del qr_data['id']
|
||||
|
||||
# Create QR code object - filter out None values
|
||||
qr_fields = {k: v for k, v in qr_data.items() if v is not None}
|
||||
qr_code = QRCode(**qr_fields)
|
||||
db.session.add(qr_code)
|
||||
imported_count += 1
|
||||
|
||||
except Exception as e:
|
||||
print(f" ⚠️ Error importing QR code {qr_data.get('name', 'unknown')}: {e}")
|
||||
|
||||
try:
|
||||
db.session.commit()
|
||||
self.migration_stats['qr_codes']['imported'] = imported_count
|
||||
print(f" ✅ Imported {imported_count} QR code records")
|
||||
except Exception as e:
|
||||
print(f" ❌ Error committing QR codes: {e}")
|
||||
db.session.rollback()
|
||||
return False
|
||||
else:
|
||||
print(" ⚠️ qr_codes.json not found")
|
||||
|
||||
# Import Attendance Data
|
||||
print("📥 Importing attendance_data...")
|
||||
attendance_file = os.path.join(self.backup_dir, 'attendance_data.json')
|
||||
if os.path.exists(attendance_file):
|
||||
with open(attendance_file, 'r') as f:
|
||||
attendance_data = json.load(f)
|
||||
|
||||
imported_count = 0
|
||||
for att_data in attendance_data:
|
||||
try:
|
||||
# Deserialize datetime and date fields
|
||||
for field in list(att_data.keys()):
|
||||
att_data[field] = self.deserialize_value(att_data[field], field)
|
||||
|
||||
# Remove 'id' field to let MySQL auto-increment
|
||||
if 'id' in att_data:
|
||||
del att_data['id']
|
||||
|
||||
# Create attendance object - filter out None values
|
||||
att_fields = {k: v for k, v in att_data.items() if v is not None}
|
||||
attendance = AttendanceData(**att_fields)
|
||||
db.session.add(attendance)
|
||||
imported_count += 1
|
||||
|
||||
except Exception as e:
|
||||
print(f" ⚠️ Error importing attendance record {att_data.get('employee_id', 'unknown')}: {e}")
|
||||
|
||||
try:
|
||||
db.session.commit()
|
||||
self.migration_stats['attendance_data']['imported'] = imported_count
|
||||
print(f" ✅ Imported {imported_count} attendance records")
|
||||
except Exception as e:
|
||||
print(f" ❌ Error committing attendance data: {e}")
|
||||
db.session.rollback()
|
||||
return False
|
||||
else:
|
||||
print(" ⚠️ attendance_data.json not found")
|
||||
|
||||
print(f"\n✅ Import completed successfully")
|
||||
print(f"📊 Total records imported: {sum(table['imported'] for table in self.migration_stats.values())}")
|
||||
|
||||
return True
|
||||
|
||||
except Exception as e:
|
||||
print(f"❌ Import failed: {e}")
|
||||
import traceback
|
||||
traceback.print_exc()
|
||||
try:
|
||||
db.session.rollback()
|
||||
except:
|
||||
pass
|
||||
return False
|
||||
|
||||
def verify_migration(self):
|
||||
"""Verify that migration was successful"""
|
||||
print("\n🔍 VERIFYING MIGRATION")
|
||||
print("=" * 50)
|
||||
|
||||
try:
|
||||
with app.app_context():
|
||||
# Count records in MySQL
|
||||
users_count = self.safe_count_records(User)
|
||||
qr_codes_count = self.safe_count_records(QRCode)
|
||||
attendance_count = self.safe_count_records(AttendanceData)
|
||||
|
||||
print(f"📊 Record counts in MySQL:")
|
||||
print(f" Users: {users_count}")
|
||||
print(f" QR Codes: {qr_codes_count}")
|
||||
print(f" Attendance: {attendance_count}")
|
||||
|
||||
# Compare with exported counts
|
||||
print(f"\n📊 Comparison with exported data:")
|
||||
verification_results = []
|
||||
|
||||
for table, stats in self.migration_stats.items():
|
||||
exported = stats['exported']
|
||||
imported = stats['imported']
|
||||
|
||||
if table == 'users':
|
||||
actual = users_count
|
||||
elif table == 'qr_codes':
|
||||
actual = qr_codes_count
|
||||
elif table == 'attendance_data':
|
||||
actual = attendance_count
|
||||
|
||||
# For new IDs, we expect imported == actual, but exported might be different
|
||||
status = "✅" if imported == actual else "❌"
|
||||
print(f" {table}: Exported={exported}, Imported={imported}, Actual={actual} {status}")
|
||||
verification_results.append(imported == actual)
|
||||
|
||||
# Test basic functionality
|
||||
print(f"\n🧪 Testing basic functionality:")
|
||||
|
||||
# Test user authentication
|
||||
admin_user = User.query.filter_by(role='admin').first()
|
||||
if admin_user:
|
||||
print(f" ✅ Admin user found: {admin_user.username}")
|
||||
else:
|
||||
print(f" ⚠️ No admin user found")
|
||||
|
||||
# Test relationships if data exists
|
||||
if users_count > 0 and qr_codes_count > 0:
|
||||
qr_with_creator = QRCode.query.join(User, QRCode.created_by == User.id).first()
|
||||
if qr_with_creator:
|
||||
print(f" ✅ QR code relationships working")
|
||||
else:
|
||||
print(f" ⚠️ No QR codes with valid creators found")
|
||||
|
||||
if qr_codes_count > 0 and attendance_count > 0:
|
||||
attendance_with_qr = AttendanceData.query.join(QRCode).first()
|
||||
if attendance_with_qr:
|
||||
print(f" ✅ Attendance relationships working")
|
||||
else:
|
||||
print(f" ⚠️ No attendance records with valid QR codes found")
|
||||
|
||||
# Test location data
|
||||
attendance_with_location = AttendanceData.query.filter(
|
||||
AttendanceData.latitude.isnot(None)
|
||||
).first()
|
||||
if attendance_with_location:
|
||||
print(f" ✅ Location data preserved")
|
||||
else:
|
||||
print(f" ℹ️ No location data found (may be expected)")
|
||||
|
||||
migration_success = all(verification_results)
|
||||
|
||||
if migration_success:
|
||||
print(f"\n🎉 MIGRATION VERIFICATION PASSED")
|
||||
print(f" All data successfully migrated to MySQL")
|
||||
else:
|
||||
print(f"\n⚠️ MIGRATION VERIFICATION ISSUES DETECTED")
|
||||
print(f" Please review the counts above")
|
||||
|
||||
return migration_success
|
||||
|
||||
except Exception as e:
|
||||
print(f"❌ Verification failed: {e}")
|
||||
import traceback
|
||||
traceback.print_exc()
|
||||
return False
|
||||
|
||||
def cleanup(self):
|
||||
"""Clean up temporary files"""
|
||||
try:
|
||||
import shutil
|
||||
shutil.rmtree(self.backup_dir)
|
||||
print(f"🧹 Cleaned up temporary files")
|
||||
except Exception as e:
|
||||
print(f"⚠️ Could not clean up temporary files: {e}")
|
||||
|
||||
def full_migration(self, pg_connection_string):
|
||||
"""Perform complete migration process"""
|
||||
print("🚀 STARTING FULL MIGRATION PROCESS")
|
||||
print("=" * 60)
|
||||
|
||||
self.pg_connection = pg_connection_string
|
||||
|
||||
# Step 1: Export from PostgreSQL
|
||||
if not self.export_postgresql_data():
|
||||
print("❌ Migration failed during export phase")
|
||||
return False
|
||||
|
||||
# Step 2: Import to MySQL
|
||||
if not self.import_to_mysql():
|
||||
print("❌ Migration failed during import phase")
|
||||
return False
|
||||
|
||||
# Step 3: Verify migration
|
||||
if not self.verify_migration():
|
||||
print("⚠️ Migration completed but verification detected issues")
|
||||
return False
|
||||
|
||||
print("\n🎉 MIGRATION COMPLETED SUCCESSFULLY!")
|
||||
print("=" * 60)
|
||||
print("Next steps:")
|
||||
print("1. Test all application functionality thoroughly")
|
||||
print("2. Update backup procedures for MySQL")
|
||||
print("3. Consider removing old PostgreSQL database after verification")
|
||||
|
||||
return True
|
||||
|
||||
def main():
|
||||
"""Main entry point"""
|
||||
parser = argparse.ArgumentParser(description='PostgreSQL to MySQL Migration Tool (Fixed)')
|
||||
parser.add_argument('--export-pg', action='store_true',
|
||||
help='Export data from PostgreSQL')
|
||||
parser.add_argument('--import-mysql', action='store_true',
|
||||
help='Import data to MySQL')
|
||||
parser.add_argument('--full-migrate', action='store_true',
|
||||
help='Complete migration (export + import)')
|
||||
parser.add_argument('--verify', action='store_true',
|
||||
help='Verify migration success')
|
||||
parser.add_argument('--pg-connection', type=str,
|
||||
help='PostgreSQL connection string')
|
||||
parser.add_argument('--mysql-connection', type=str,
|
||||
help='MySQL connection string (default: from .env)')
|
||||
|
||||
args = parser.parse_args()
|
||||
|
||||
if not any([args.export_pg, args.import_mysql, args.full_migrate, args.verify]):
|
||||
parser.print_help()
|
||||
print("\nExample usage:")
|
||||
print(" python migrate_fixed.py --export-pg --pg-connection 'postgresql://user:pass@localhost/dbname'")
|
||||
print(" python migrate_fixed.py --import-mysql")
|
||||
print(" python migrate_fixed.py --full-migrate --pg-connection 'postgresql://user:pass@localhost/dbname'")
|
||||
sys.exit(1)
|
||||
|
||||
migrator = DatabaseMigrator(args.pg_connection, args.mysql_connection)
|
||||
|
||||
try:
|
||||
if args.export_pg:
|
||||
if not args.pg_connection:
|
||||
print("❌ PostgreSQL connection string required for export")
|
||||
print("Example: postgresql://username:password@localhost:5432/database_name")
|
||||
sys.exit(1)
|
||||
migrator.export_postgresql_data()
|
||||
|
||||
elif args.import_mysql:
|
||||
migrator.import_to_mysql()
|
||||
|
||||
elif args.verify:
|
||||
migrator.verify_migration()
|
||||
|
||||
elif args.full_migrate:
|
||||
if not args.pg_connection:
|
||||
print("❌ PostgreSQL connection string required for full migration")
|
||||
print("Example: postgresql://username:password@localhost:5432/database_name")
|
||||
sys.exit(1)
|
||||
migrator.full_migration(args.pg_connection)
|
||||
|
||||
finally:
|
||||
# Don't auto-cleanup if export was successful - user might want to review files
|
||||
if not (args.export_pg and sum(migrator.migration_stats[table]['exported'] for table in migrator.migration_stats) > 0):
|
||||
migrator.cleanup()
|
||||
else:
|
||||
print(f"\n📁 Export files preserved at: {migrator.backup_dir}")
|
||||
print(" Run with --import-mysql to complete migration")
|
||||
|
||||
if __name__ == '__main__':
|
||||
main()
|
||||
+10
-6
@@ -1,12 +1,13 @@
|
||||
# QR Code Management System - Python Dependencies
|
||||
# Production-ready Flask application with PostgreSQL support
|
||||
# Production-ready Flask application with MySQL support
|
||||
|
||||
# Core Flask Framework
|
||||
Flask==2.3.3
|
||||
Flask-SQLAlchemy==3.0.5
|
||||
|
||||
# Database Support
|
||||
psycopg2-binary==2.9.7 # PostgreSQL adapter
|
||||
# Database Support - MySQL
|
||||
PyMySQL==1.1.0 # Pure Python MySQL client
|
||||
mysql-connector-python==8.2.0 # Official MySQL connector (alternative)
|
||||
SQLAlchemy==2.0.21
|
||||
|
||||
# Security and Authentication
|
||||
@@ -16,7 +17,7 @@ Werkzeug==2.3.7 # Security utilities and password hashing
|
||||
qrcode==7.4.2 # QR code generation library
|
||||
Pillow==10.0.1 # Image processing for QR codes
|
||||
|
||||
# User Agent Detection (NEW)
|
||||
# User Agent Detection
|
||||
user-agents==2.2.0 # Device and browser detection from user agent strings
|
||||
|
||||
# URL and Regex Processing
|
||||
@@ -43,7 +44,7 @@ itsdangerous==2.1.2 # Secure data serialization
|
||||
Jinja2==3.1.2 # Template engine
|
||||
MarkupSafe==2.1.3 # Safe string handling
|
||||
|
||||
# Data Export and Processing (NEW)
|
||||
# Data Export and Processing
|
||||
openpyxl==3.1.2 # Excel file generation for attendance reports
|
||||
pandas==2.1.1 # Data manipulation for reports (optional)
|
||||
|
||||
@@ -54,4 +55,7 @@ requests==2.31.0 # HTTP library for external API calls
|
||||
Flask-Caching==2.1.0 # Caching support for Flask
|
||||
|
||||
# Logging and Monitoring (optional)
|
||||
python-json-logger==2.0.7 # Structured logging support
|
||||
python-json-logger==2.0.7 # Structured logging support
|
||||
|
||||
# Cryptography dependencies (required for some MySQL features)
|
||||
cryptography==41.0.7 # Required for MySQL SSL connections
|
||||
@@ -0,0 +1,57 @@
|
||||
# QR Code Management System - Python Dependencies
|
||||
# Production-ready Flask application with PostgreSQL support
|
||||
|
||||
# Core Flask Framework
|
||||
Flask==2.3.3
|
||||
Flask-SQLAlchemy==3.0.5
|
||||
|
||||
# Database Support
|
||||
psycopg2-binary==2.9.7 # PostgreSQL adapter
|
||||
SQLAlchemy==2.0.21
|
||||
|
||||
# Security and Authentication
|
||||
Werkzeug==2.3.7 # Security utilities and password hashing
|
||||
|
||||
# QR Code Generation
|
||||
qrcode==7.4.2 # QR code generation library
|
||||
Pillow==10.0.1 # Image processing for QR codes
|
||||
|
||||
# User Agent Detection (NEW)
|
||||
user-agents==2.2.0 # Device and browser detection from user agent strings
|
||||
|
||||
# URL and Regex Processing
|
||||
regex==2023.8.8 # Enhanced regex support for URL generation
|
||||
|
||||
# Environment and Configuration
|
||||
python-dotenv==1.0.0 # Environment variable management
|
||||
|
||||
# Date and Time Processing
|
||||
python-dateutil==2.8.2 # Extended date/time processing
|
||||
|
||||
# Development and Testing (optional)
|
||||
pytest==7.4.2 # Testing framework
|
||||
pytest-flask==1.2.0 # Flask testing utilities
|
||||
Flask-Testing==0.8.1 # Additional Flask testing tools
|
||||
|
||||
# Production Server (optional)
|
||||
gunicorn==21.2.0 # WSGI HTTP Server for production
|
||||
gevent==23.7.0 # Async worker support
|
||||
|
||||
# Utilities
|
||||
click==8.1.7 # Command line interface creation
|
||||
itsdangerous==2.1.2 # Secure data serialization
|
||||
Jinja2==3.1.2 # Template engine
|
||||
MarkupSafe==2.1.3 # Safe string handling
|
||||
|
||||
# Data Export and Processing (NEW)
|
||||
openpyxl==3.1.2 # Excel file generation for attendance reports
|
||||
pandas==2.1.1 # Data manipulation for reports (optional)
|
||||
|
||||
# HTTP Requests (for potential integrations)
|
||||
requests==2.31.0 # HTTP library for external API calls
|
||||
|
||||
# Caching (optional for performance)
|
||||
Flask-Caching==2.1.0 # Caching support for Flask
|
||||
|
||||
# Logging and Monitoring (optional)
|
||||
python-json-logger==2.0.7 # Structured logging support
|
||||
+214
@@ -0,0 +1,214 @@
|
||||
#!/usr/bin/env python3
|
||||
"""
|
||||
Database Schema Fix Script
|
||||
|
||||
This script fixes the password_hash column length issue and other potential
|
||||
schema mismatches between PostgreSQL and MySQL.
|
||||
|
||||
Usage:
|
||||
python fix_schema.py
|
||||
"""
|
||||
|
||||
import sys
|
||||
import os
|
||||
|
||||
# Add app directory to path
|
||||
sys.path.append(os.path.dirname(os.path.abspath(__file__)))
|
||||
|
||||
try:
|
||||
from app import app, db
|
||||
from sqlalchemy import text, inspect
|
||||
except ImportError as e:
|
||||
print(f"❌ Error importing modules: {e}")
|
||||
sys.exit(1)
|
||||
|
||||
def fix_mysql_schema():
|
||||
"""Fix MySQL schema to accommodate PostgreSQL data types"""
|
||||
print("🔧 FIXING MYSQL SCHEMA")
|
||||
print("=" * 50)
|
||||
|
||||
try:
|
||||
with app.app_context():
|
||||
# Check current schema
|
||||
inspector = inspect(db.engine)
|
||||
|
||||
# Fix users table
|
||||
print("📋 Checking users table schema...")
|
||||
users_columns = inspector.get_columns('users')
|
||||
|
||||
schema_fixes = []
|
||||
|
||||
# Check password_hash column length
|
||||
password_hash_col = next((col for col in users_columns if col['name'] == 'password_hash'), None)
|
||||
if password_hash_col:
|
||||
# Check if it's too small (PostgreSQL scrypt hashes can be 200+ characters)
|
||||
if password_hash_col['type'].length and password_hash_col['type'].length < 255:
|
||||
schema_fixes.append("ALTER TABLE users MODIFY COLUMN password_hash VARCHAR(255)")
|
||||
print(f" ⚠️ password_hash column too small ({password_hash_col['type'].length} chars)")
|
||||
else:
|
||||
print(f" ✅ password_hash column size OK")
|
||||
|
||||
# Check for other potential issues
|
||||
qr_codes_columns = inspector.get_columns('qr_codes')
|
||||
|
||||
# Check qr_code_image column (base64 images can be very long)
|
||||
qr_image_col = next((col for col in qr_codes_columns if col['name'] == 'qr_code_image'), None)
|
||||
if qr_image_col:
|
||||
# QR code images should be TEXT or LONGTEXT, not VARCHAR
|
||||
if hasattr(qr_image_col['type'], 'length') and qr_image_col['type'].length:
|
||||
schema_fixes.append("ALTER TABLE qr_codes MODIFY COLUMN qr_code_image LONGTEXT")
|
||||
print(f" ⚠️ qr_code_image should be LONGTEXT")
|
||||
else:
|
||||
print(f" ✅ qr_code_image column type OK")
|
||||
|
||||
# Check location_address column
|
||||
location_addr_col = next((col for col in qr_codes_columns if col['name'] == 'location_address'), None)
|
||||
if location_addr_col:
|
||||
if hasattr(location_addr_col['type'], 'length') and location_addr_col['type'].length and location_addr_col['type'].length < 500:
|
||||
schema_fixes.append("ALTER TABLE qr_codes MODIFY COLUMN location_address TEXT")
|
||||
print(f" ⚠️ location_address column too small")
|
||||
else:
|
||||
print(f" ✅ location_address column size OK")
|
||||
|
||||
# Check attendance_data table
|
||||
print("📋 Checking attendance_data table schema...")
|
||||
attendance_columns = inspector.get_columns('attendance_data')
|
||||
|
||||
# Check address column
|
||||
address_col = next((col for col in attendance_columns if col['name'] == 'address'), None)
|
||||
if address_col:
|
||||
if hasattr(address_col['type'], 'length') and address_col['type'].length and address_col['type'].length < 500:
|
||||
schema_fixes.append("ALTER TABLE attendance_data MODIFY COLUMN address VARCHAR(500)")
|
||||
print(f" ⚠️ address column too small")
|
||||
else:
|
||||
print(f" ✅ address column size OK")
|
||||
|
||||
# Check user_agent column (can be very long)
|
||||
user_agent_col = next((col for col in attendance_columns if col['name'] == 'user_agent'), None)
|
||||
if user_agent_col:
|
||||
if hasattr(user_agent_col['type'], 'length') and user_agent_col['type'].length:
|
||||
schema_fixes.append("ALTER TABLE attendance_data MODIFY COLUMN user_agent TEXT")
|
||||
print(f" ⚠️ user_agent should be TEXT")
|
||||
else:
|
||||
print(f" ✅ user_agent column type OK")
|
||||
|
||||
# Apply fixes
|
||||
if schema_fixes:
|
||||
print(f"\n🔨 Applying {len(schema_fixes)} schema fixes...")
|
||||
for i, fix in enumerate(schema_fixes, 1):
|
||||
try:
|
||||
print(f" {i}. {fix}")
|
||||
db.session.execute(text(fix))
|
||||
print(f" ✅ Applied successfully")
|
||||
except Exception as e:
|
||||
print(f" ❌ Failed: {e}")
|
||||
db.session.rollback()
|
||||
return False
|
||||
|
||||
db.session.commit()
|
||||
print(f"\n✅ All schema fixes applied successfully!")
|
||||
return True
|
||||
else:
|
||||
print(f"\n✅ No schema fixes needed - schema looks good!")
|
||||
return True
|
||||
|
||||
except Exception as e:
|
||||
print(f"❌ Error fixing schema: {e}")
|
||||
import traceback
|
||||
traceback.print_exc()
|
||||
return False
|
||||
|
||||
def show_current_schema():
|
||||
"""Show current MySQL schema for debugging"""
|
||||
print("\n📋 CURRENT MYSQL SCHEMA")
|
||||
print("=" * 30)
|
||||
|
||||
try:
|
||||
with app.app_context():
|
||||
inspector = inspect(db.engine)
|
||||
|
||||
tables = ['users', 'qr_codes', 'attendance_data']
|
||||
|
||||
for table in tables:
|
||||
if table in inspector.get_table_names():
|
||||
print(f"\n📊 {table} table:")
|
||||
columns = inspector.get_columns(table)
|
||||
|
||||
for col in columns:
|
||||
col_type = str(col['type'])
|
||||
nullable = "NULL" if col['nullable'] else "NOT NULL"
|
||||
default = f" DEFAULT {col['default']}" if col['default'] else ""
|
||||
print(f" {col['name']:<20} {col_type:<20} {nullable}{default}")
|
||||
else:
|
||||
print(f"\n❌ {table} table not found")
|
||||
|
||||
except Exception as e:
|
||||
print(f"❌ Error showing schema: {e}")
|
||||
|
||||
def test_data_compatibility():
|
||||
"""Test if sample data would fit in current schema"""
|
||||
print("\n🧪 TESTING DATA COMPATIBILITY")
|
||||
print("=" * 35)
|
||||
|
||||
# Test password hash length
|
||||
sample_password_hash = "scrypt:32768:8:1$GMnEGe3K4utb6mNW$3472a6003b60b095d4a9cbb201fe85e662a4d03b7162cf0c5ca896112f04b8f04c202311002ec17d56ef3ad15b232175762ad80be289cb83d546095ddfe14c77"
|
||||
print(f"📏 Sample password hash length: {len(sample_password_hash)} characters")
|
||||
|
||||
# Test QR code image (typical base64 image)
|
||||
sample_qr_image = "iVBORw0KGgoAAAANSUhEUgAAASwAAAEsCAYAAAB5fY51" + "A" * 2000 # Simulate base64 image
|
||||
print(f"📏 Sample QR image length: {len(sample_qr_image)} characters")
|
||||
|
||||
# Test user agent string
|
||||
sample_user_agent = "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/91.0.4472.124 Safari/537.36"
|
||||
print(f"📏 Sample user agent length: {len(sample_user_agent)} characters")
|
||||
|
||||
try:
|
||||
with app.app_context():
|
||||
inspector = inspect(db.engine)
|
||||
|
||||
# Check users table
|
||||
users_columns = inspector.get_columns('users')
|
||||
password_col = next((col for col in users_columns if col['name'] == 'password_hash'), None)
|
||||
|
||||
if password_col and hasattr(password_col['type'], 'length') and password_col['type'].length:
|
||||
if len(sample_password_hash) > password_col['type'].length:
|
||||
print(f"❌ Password hash won't fit (needs {len(sample_password_hash)}, has {password_col['type'].length})")
|
||||
else:
|
||||
print(f"✅ Password hash will fit")
|
||||
else:
|
||||
print(f"✅ Password hash column is TEXT/unlimited")
|
||||
|
||||
except Exception as e:
|
||||
print(f"❌ Error testing compatibility: {e}")
|
||||
|
||||
def main():
|
||||
"""Main function"""
|
||||
print("MYSQL SCHEMA FIX UTILITY")
|
||||
print("=" * 60)
|
||||
|
||||
# Show current schema
|
||||
show_current_schema()
|
||||
|
||||
# Test compatibility
|
||||
test_data_compatibility()
|
||||
|
||||
# Ask user if they want to apply fixes
|
||||
print(f"\n" + "="*60)
|
||||
response = input("Apply schema fixes? (y/n): ").lower().strip()
|
||||
|
||||
if response == 'y':
|
||||
success = fix_mysql_schema()
|
||||
|
||||
if success:
|
||||
print(f"\n🎉 Schema fixes completed!")
|
||||
print(f"💡 You can now run the migration script again:")
|
||||
print(f" python migrate_fixed.py --import-mysql")
|
||||
else:
|
||||
print(f"\n❌ Schema fixes failed. Check the errors above.")
|
||||
else:
|
||||
print(f"\n📋 Schema fixes skipped.")
|
||||
print(f"💡 To manually fix the password_hash issue, run:")
|
||||
print(f" ALTER TABLE users MODIFY COLUMN password_hash VARCHAR(255);")
|
||||
|
||||
if __name__ == '__main__':
|
||||
main()
|
||||
Reference in New Issue
Block a user