nngo/GOV_QR_Codes_Management
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

clean up

Browse Source
This commit is contained in:
nngo
2025-08-15 19:40:10 -04:00
parent 455105f832
commit 7e2aaf1852
6 changed files with 0 additions and 2143 deletions
Download Patch File Download Diff File Expand all files Collapse all files
db_verification.py
-586
View File
@@ -1,586 +0,0 @@
#!/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)
migrate_project.py
-347
View File
@@ -1,347 +0,0 @@
#!/usr/bin/env python3
"""
Database Migration Script for Project Model
==========================================
This script safely migrates your existing database to add the Project model
and associate QR codes with projects.
The script will:
1. Backup your current database
2. Create the new projects table
3. Add project_id column to qr_codes table
4. Create some sample projects (optional)
5. Provide rollback instructions
Usage:
python migrate_projects.py
Requirements:
- Your existing Flask app with database models
- Database write permissions
"""
import os
import sys
import shutil
from datetime import datetime
from sqlalchemy import create_engine, text, inspect
from sqlalchemy.orm import sessionmaker
# Add your app to the Python path
sys.path.append(os.path.dirname(os.path.abspath(__file__)))
try:
from app import app, db, User, QRCode, Project
from dotenv import load_dotenv
except ImportError as e:
print(f"Error importing app modules: {e}")
print("Make sure this script is in the same directory as your app.py file")
sys.exit(1)
# Load environment variables
load_dotenv()
# Configuration
BACKUP_DIR = "database_backups"
MIGRATION_VERSION = "v1.1_add_project_model"
def create_backup_directory():
"""Create backup directory if it doesn't exist"""
if not os.path.exists(BACKUP_DIR):
os.makedirs(BACKUP_DIR)
print(f"✓ Created backup directory: {BACKUP_DIR}")
def backup_database():
"""Create a backup of the current database"""
timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
backup_filename = f"backup_{MIGRATION_VERSION}_{timestamp}.db"
backup_path = os.path.join(BACKUP_DIR, backup_filename)
# Get database path from config
db_url = app.config['SQLALCHEMY_DATABASE_URI']
if db_url.startswith('sqlite:///'):
# SQLite database
db_path = db_url.replace('sqlite:///', '')
if os.path.exists(db_path):
shutil.copy2(db_path, backup_path)
print(f"✓ Database backed up to: {backup_path}")
return backup_path
else:
print(f"⚠ Database file not found: {db_path}")
return None
else:
print("⚠ Non-SQLite databases require manual backup")
print("Please ensure you have a recent backup before proceeding")
return None
def validate_current_database():
"""Validate the current database structure and data"""
print("\n🔍 Validating current database...")
try:
with app.app_context():
# Check if required tables exist
inspector = inspect(db.engine)
tables = inspector.get_table_names()
required_tables = ['users', 'qr_codes']
for table in required_tables:
if table not in tables:
print(f"❌ Required table '{table}' not found")
return False
print(f"✓ Table '{table}' exists")
# Check current data
users = User.query.all()
qr_codes = QRCode.query.all()
print(f"✓ Found {len(users)} users in database")
print(f"✓ Found {len(qr_codes)} QR codes in database")
# Check if projects table already exists
if 'projects' in tables:
print("⚠ Projects table already exists - migration may have been run before")
projects = Project.query.all()
print(f"✓ Found {len(projects)} existing projects")
print("✓ Database validation passed")
return True
except Exception as e:
print(f"❌ Database validation failed: {e}")
return False
def perform_migration():
"""Perform the actual migration"""
print("\n🚀 Starting migration...")
try:
with app.app_context():
# Create all tables (this will create the projects table if it doesn't exist)
db.create_all()
print("✓ Database tables created/updated")
# Check if project_id column exists in qr_codes table
inspector = inspect(db.engine)
qr_columns = inspector.get_columns('qr_codes')
qr_column_names = [col['name'] for col in qr_columns]
if 'project_id' not in qr_column_names:
# Add project_id column to qr_codes table
print("➕ Adding project_id column to qr_codes table...")
db.session.execute(text('ALTER TABLE qr_codes ADD COLUMN project_id INTEGER'))
db.session.commit()
print("✓ project_id column added to qr_codes table")
else:
print("✓ project_id column already exists in qr_codes table")
print("✓ Migration completed successfully!")
return True
except Exception as e:
print(f"❌ Migration failed: {e}")
import traceback
traceback.print_exc()
return False
def create_sample_projects():
"""Create some sample projects (optional)"""
print("\n📁 Creating sample projects...")
try:
with app.app_context():
# Get the first admin user to assign as creator
admin_user = User.query.filter_by(role='admin').first()
creator_id = admin_user.id if admin_user else None
# Check if any projects exist
existing_projects = Project.query.count()
if existing_projects > 0:
print(f"✓ Found {existing_projects} existing projects - skipping sample creation")
return True
# Sample projects
sample_projects = [
{
'name': 'Office Locations',
'description': 'QR codes for various office locations and facilities'
},
{
'name': 'Events',
'description': 'QR codes for company events and meetings'
},
{
'name': 'Training Materials',
'description': 'QR codes for training sessions and educational content'
}
]
created_count = 0
for project_data in sample_projects:
# Check if project with this name already exists
existing = Project.query.filter_by(name=project_data['name']).first()
if not existing:
project = Project(
name=project_data['name'],
description=project_data['description'],
created_by=creator_id
)
db.session.add(project)
created_count += 1
if created_count > 0:
db.session.commit()
print(f"✓ Created {created_count} sample projects")
else:
print("✓ Sample projects already exist")
return True
except Exception as e:
print(f"❌ Failed to create sample projects: {e}")
return False
def test_new_functionality():
"""Test the new project functionality"""
print("\n🧪 Testing new project functionality...")
try:
with app.app_context():
# Test Project model
projects = Project.query.all()
print(f"✓ Can query projects: {len(projects)} found")
# Test QRCode project relationship
qr_codes = QRCode.query.all()
for qr in qr_codes[:3]: # Test first 3 QR codes
project = qr.project # This should not raise an error
print(f"✓ QR code '{qr.name}' project: {project.name if project else 'None'}")
# Test Project.qr_codes relationship
if projects:
first_project = projects[0]
qr_count = first_project.qr_count
print(f"✓ Project '{first_project.name}' has {qr_count} QR codes")
print("✓ New functionality tests passed")
return True
except Exception as e:
print(f"❌ Functionality tests failed: {e}")
import traceback
traceback.print_exc()
return False
def display_summary():
"""Display migration summary"""
print("\n📊 MIGRATION SUMMARY")
print("=" * 50)
try:
with app.app_context():
users = User.query.count()
projects = Project.query.count()
qr_codes = QRCode.query.count()
print(f"👥 Users: {users}")
print(f"📁 Projects: {projects}")
print(f"🔗 QR Codes: {qr_codes}")
# Show project distribution
if projects > 0:
print(f"\n📁 Project Details:")
for project in Project.query.all():
print(f" • {project.name}: {project.qr_count} QR codes")
# Show unassigned QR codes
unassigned = QRCode.query.filter_by(project_id=None).count()
if unassigned > 0:
print(f"\n⚠️ {unassigned} QR codes are not assigned to any project")
except Exception as e:
print(f"Error generating summary: {e}")
def rollback_instructions():
"""Show rollback instructions"""
print("\n🔄 ROLLBACK INSTRUCTIONS")
print("=" * 50)
print("If you need to rollback this migration:")
print("1. Stop your application")
print("2. Restore the database backup:")
print(" - For SQLite: Replace your database file with the backup")
print(" - For other databases: Restore from your backup")
print("3. Remove the project_id column from qr_codes table:")
print(" ALTER TABLE qr_codes DROP COLUMN project_id;")
print("4. Drop the projects table:")
print(" DROP TABLE projects;")
print("5. Update your app.py to remove Project model and related code")
print("6. Restart your application")
def main():
"""Main migration process"""
print("🗃️ PROJECT MODEL MIGRATION")
print("=" * 50)
print("This will add project functionality to your QR code system.")
print("Projects allow you to organize QR codes into logical groups.")
print("\nWhat this migration does:")
print("• Creates a new 'projects' table")
print("• Adds 'project_id' column to 'qr_codes' table")
print("• Creates sample projects (optional)")
print("• Updates relationships between models")
# Confirm migration
response = input("\nProceed with migration? (y/N): ").strip().lower()
if response not in ['y', 'yes']:
print("Migration cancelled.")
sys.exit(0)
# Step 1: Create backup directory
create_backup_directory()
# Step 2: Backup database
backup_path = backup_database()
if not backup_path:
response = input("No backup created. Continue anyway? (y/N): ").strip().lower()
if response not in ['y', 'yes']:
print("Migration cancelled for safety.")
sys.exit(0)
# Step 3: Validate current database
if not validate_current_database():
print("❌ Database validation failed. Migration cancelled.")
sys.exit(1)
# Step 4: Perform migration
if not perform_migration():
print("❌ Migration failed. Please check the errors above.")
sys.exit(1)
# Step 5: Create sample projects
create_sample = input("\nCreate sample projects? (Y/n): ").strip().lower()
if create_sample not in ['n', 'no']:
create_sample_projects()
# Step 6: Test new functionality
if not test_new_functionality():
print("❌ Functionality testing failed. Migration may be incomplete.")
sys.exit(1)
# Step 7: Display summary
display_summary()
# Step 8: Show rollback instructions
show_rollback = input("\nWould you like to see rollback instructions? (y/N): ").strip().lower()
if show_rollback in ['y', 'yes']:
rollback_instructions()
print("\n🚀 Migration completed successfully!")
print("You can now:")
print("• Create and manage projects")
print("• Associate QR codes with projects")
print("• Use the project dropdown in QR code forms")
print("• View project statistics and organization")
if __name__ == "__main__":
main()
migrate_roles.py
-330
View File
@@ -1,330 +0,0 @@
#!/usr/bin/env python3
"""
Database Migration Script for New User Roles
=============================================
This script safely migrates your existing database to support the new roles:
- payroll
- project_manager
The script will:
1. Backup your current database
2. Check for any data integrity issues
3. Add the new roles to your system
4. Provide a rollback option if needed
Usage:
python migrate_roles.py
Requirements:
- Your existing Flask app with database models
- Backup directory permissions
- Database write permissions
"""
import os
import sys
import shutil
from datetime import datetime
from sqlalchemy import create_engine, text
from sqlalchemy.orm import sessionmaker
# Add your app to the Python path
sys.path.append(os.path.dirname(os.path.abspath(__file__)))
try:
from app import app, db, User, QRCode
from dotenv import load_dotenv
except ImportError as e:
print(f"Error importing app modules: {e}")
print("Make sure this script is in the same directory as your app.py file")
sys.exit(1)
# Load environment variables
load_dotenv()
# Configuration
BACKUP_DIR = "database_backups"
VALID_ROLES = ['admin', 'staff', 'payroll', 'project_manager']
MIGRATION_VERSION = "v1.0_add_payroll_project_manager_roles"
def create_backup_directory():
"""Create backup directory if it doesn't exist"""
if not os.path.exists(BACKUP_DIR):
os.makedirs(BACKUP_DIR)
print(f"✓ Created backup directory: {BACKUP_DIR}")
def backup_database():
"""Create a backup of the current database"""
timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
backup_filename = f"backup_{MIGRATION_VERSION}_{timestamp}.db"
backup_path = os.path.join(BACKUP_DIR, backup_filename)
# Get database path from config
db_url = app.config['SQLALCHEMY_DATABASE_URI']
if db_url.startswith('sqlite:///'):
# SQLite database
db_path = db_url.replace('sqlite:///', '')
if os.path.exists(db_path):
shutil.copy2(db_path, backup_path)
print(f"✓ Database backed up to: {backup_path}")
return backup_path
else:
print(f"⚠ Database file not found: {db_path}")
return None
else:
print("⚠ Non-SQLite databases require manual backup")
print("Please ensure you have a recent backup before proceeding")
return None
def validate_current_database():
"""Validate the current database structure and data"""
print("\n🔍 Validating current database...")
try:
with app.app_context():
# Check if User table exists and has required columns
users = User.query.all()
print(f"✓ Found {len(users)} users in database")
# Check current roles
current_roles = db.session.query(User.role.distinct()).all()
current_roles = [role[0] for role in current_roles]
print(f"✓ Current roles in database: {current_roles}")
# Check for any invalid roles
invalid_roles = [role for role in current_roles if role not in ['admin', 'staff']]
if invalid_roles:
print(f"⚠ Found unexpected roles: {invalid_roles}")
return False
# Check QR codes
qr_codes = QRCode.query.all()
print(f"✓ Found {len(qr_codes)} QR codes in database")
print("✓ Database validation passed")
return True
except Exception as e:
print(f"❌ Database validation failed: {e}")
return False
def perform_migration():
"""Perform the actual migration"""
print("\n🚀 Starting migration...")
try:
with app.app_context():
# The migration is actually just code changes since we're not changing the database schema
# We're just allowing new values in the existing role column
# Check if any existing users need to be updated (optional)
admin_count = User.query.filter_by(role='admin').count()
staff_count = User.query.filter_by(role='staff').count()
print(f"✓ Current user distribution:")
print(f" - Administrators: {admin_count}")
print(f" - Staff Users: {staff_count}")
# Create a migration record (optional - for tracking)
migration_record = {
'version': MIGRATION_VERSION,
'timestamp': datetime.now(),
'description': 'Added support for payroll and project_manager roles'
}
print("✓ Migration completed successfully!")
print("✓ New roles 'payroll' and 'project_manager' are now supported")
return True
except Exception as e:
print(f"❌ Migration failed: {e}")
return False
def test_new_roles():
"""Test that new roles work correctly"""
print("\n🧪 Testing new role functionality...")
try:
with app.app_context():
# Test creating users with new roles (without actually saving them)
test_payroll_user = User(
full_name="Test Payroll User",
email="test_payroll@example.com",
username="test_payroll",
role="payroll"
)
test_pm_user = User(
full_name="Test Project Manager",
email="test_pm@example.com",
username="test_pm",
role="project_manager"
)
# Validate the objects (without saving)
if test_payroll_user.role in VALID_ROLES:
print("✓ Payroll role validation passed")
else:
print("❌ Payroll role validation failed")
return False
if test_pm_user.role in VALID_ROLES:
print("✓ Project Manager role validation passed")
else:
print("❌ Project Manager role validation failed")
return False
# Test role display names
if hasattr(test_payroll_user, 'get_role_display_name'):
payroll_display = test_payroll_user.get_role_display_name()
print(f"✓ Payroll display name: {payroll_display}")
if hasattr(test_pm_user, 'get_role_display_name'):
pm_display = test_pm_user.get_role_display_name()
print(f"✓ Project Manager display name: {pm_display}")
# Test permission methods
if hasattr(test_payroll_user, 'has_staff_permissions'):
if test_payroll_user.has_staff_permissions():
print("✓ Payroll user has staff-level permissions")
else:
print("❌ Payroll user missing staff-level permissions")
return False
if hasattr(test_pm_user, 'has_staff_permissions'):
if test_pm_user.has_staff_permissions():
print("✓ Project Manager has staff-level permissions")
else:
print("❌ Project Manager missing staff-level permissions")
return False
print("✓ All role functionality tests passed")
return True
except Exception as e:
print(f"❌ Role testing failed: {e}")
return False
def display_summary():
"""Display migration summary and next steps"""
print("\n" + "="*60)
print("🎉 MIGRATION COMPLETE!")
print("="*60)
print()
print("WHAT'S NEW:")
print("• Added support for 'payroll' role")
print("• Added support for 'project_manager' role")
print("• Both new roles have staff-level permissions")
print("• Updated templates support new role creation")
print("• Enhanced user management interface")
print()
print("NEXT STEPS:")
print("1. Restart your Flask application")
print("2. Test creating users with new roles via admin interface")
print("3. Verify new role badges appear correctly in user management")
print("4. Consider updating any custom permissions as needed")
print()
print("FILES UPDATED:")
print("• app.py - Core application logic")
print("• templates/create_user.html - User creation form")
print("• templates/edit_user.html - User editing form")
print("• templates/users.html - User management page")
print()
print("BACKUP LOCATION:")
backup_files = [f for f in os.listdir(BACKUP_DIR) if f.startswith('backup_')]
if backup_files:
latest_backup = sorted(backup_files)[-1]
print(f"• {os.path.join(BACKUP_DIR, latest_backup)}")
print()
def rollback_instructions():
"""Display rollback instructions"""
print("\n" + "="*60)
print("🔄 ROLLBACK INSTRUCTIONS")
print("="*60)
print()
print("If you need to rollback this migration:")
print()
print("1. Stop your Flask application")
print("2. Restore your database from backup:")
backup_files = [f for f in os.listdir(BACKUP_DIR) if f.startswith('backup_')]
if backup_files:
latest_backup = sorted(backup_files)[-1]
backup_path = os.path.join(BACKUP_DIR, latest_backup)
db_url = app.config['SQLALCHEMY_DATABASE_URI']
if db_url.startswith('sqlite:///'):
db_path = db_url.replace('sqlite:///', '')
print(f" cp {backup_path} {db_path}")
print("3. Revert your code files to previous versions")
print("4. Restart your application")
print()
def main():
"""Main migration function"""
print("="*60)
print("🔧 USER ROLES MIGRATION SCRIPT")
print("="*60)
print(f"Migration: {MIGRATION_VERSION}")
print(f"Date: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}")
print()
# Check if we're in the right directory
if not os.path.exists('app.py'):
print("❌ app.py not found in current directory")
print("Please run this script from your Flask application directory")
sys.exit(1)
# Create backup directory
create_backup_directory()
# Ask for confirmation
print("This migration will add support for new user roles:")
print("• payroll")
print("• project_manager")
print()
response = input("Do you want to proceed? (y/N): ").strip().lower()
if response not in ['y', 'yes']:
print("Migration cancelled.")
sys.exit(0)
# Step 1: Backup database
backup_path = backup_database()
if not backup_path:
response = input("No backup created. Continue anyway? (y/N): ").strip().lower()
if response not in ['y', 'yes']:
print("Migration cancelled for safety.")
sys.exit(0)
# Step 2: Validate current database
if not validate_current_database():
print("❌ Database validation failed. Migration cancelled.")
sys.exit(1)
# Step 3: Perform migration
if not perform_migration():
print("❌ Migration failed. Please check the errors above.")
sys.exit(1)
# Step 4: Test new functionality
if not test_new_roles():
print("❌ Role testing failed. Migration may be incomplete.")
sys.exit(1)
# Step 5: Display summary
display_summary()
# Step 6: Show rollback instructions
show_rollback = input("\nWould you like to see rollback instructions? (y/N): ").strip().lower()
if show_rollback in ['y', 'yes']:
rollback_instructions()
print("\n🚀 Migration completed successfully!")
print("You can now create users with payroll and project_manager roles.")
if __name__ == "__main__":
main()
postgres2mysql.py
-609
View File
@@ -1,609 +0,0 @@
#!/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()
requirements.txt.bak
-57
View File
@@ -1,57 +0,0 @@
# 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
schema_fix.py
-214
View File
@@ -1,214 +0,0 @@
#!/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()