diff --git a/db_verification.py b/db_verification.py deleted file mode 100644 index d2def70..0000000 --- a/db_verification.py +++ /dev/null @@ -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) \ No newline at end of file diff --git a/migrate_project.py b/migrate_project.py deleted file mode 100644 index 2494c0d..0000000 --- a/migrate_project.py +++ /dev/null @@ -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() \ No newline at end of file diff --git a/migrate_roles.py b/migrate_roles.py deleted file mode 100644 index 6b6ace3..0000000 --- a/migrate_roles.py +++ /dev/null @@ -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() \ No newline at end of file diff --git a/postgres2mysql.py b/postgres2mysql.py deleted file mode 100644 index ff8d6f0..0000000 --- a/postgres2mysql.py +++ /dev/null @@ -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() diff --git a/requirements.txt.bak b/requirements.txt.bak deleted file mode 100644 index b9156f5..0000000 --- a/requirements.txt.bak +++ /dev/null @@ -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 \ No newline at end of file diff --git a/schema_fix.py b/schema_fix.py deleted file mode 100644 index 8519b16..0000000 --- a/schema_fix.py +++ /dev/null @@ -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()