From 0e517b110fbe5199b4a666925fc611962d42ea14 Mon Sep 17 00:00:00 2001 From: Nguyen Ngo Date: Sun, 10 Aug 2025 10:51:19 -0400 Subject: [PATCH] Change db to Mysql --- app.py | 37 +- app.py.bak | 2721 ++++++++++++++++++++++++++++++++++++++++++ db_verification.py | 586 +++++++++ postgres2mysql.py | 609 ++++++++++ requirements.txt | 16 +- requirements.txt.bak | 57 + schema_fix.py | 214 ++++ 7 files changed, 4219 insertions(+), 21 deletions(-) create mode 100644 app.py.bak create mode 100644 db_verification.py create mode 100644 postgres2mysql.py create mode 100644 requirements.txt.bak create mode 100644 schema_fix.py diff --git a/app.py b/app.py index 27c91e8..edf6700 100644 --- a/app.py +++ b/app.py @@ -32,7 +32,7 @@ class User(db.Model): full_name = db.Column(db.String(100), nullable=False) email = db.Column(db.String(120), unique=True, nullable=False) username = db.Column(db.String(80), unique=True, nullable=False) - password_hash = db.Column(db.String(128), nullable=False) + password_hash = db.Column(db.String(255), nullable=False) role = db.Column(db.String(20), nullable=False, default='staff') # admin or staff created_by = db.Column(db.Integer, db.ForeignKey('users.id'), nullable=True) created_date = db.Column(db.DateTime, default=datetime.utcnow) @@ -83,6 +83,7 @@ class QRCode(db.Model): def has_coordinates(self): """Check if this QR code has address coordinates""" return self.address_latitude is not None and self.address_longitude is not None + @property def coordinates_display(self): """Get formatted coordinates for display""" @@ -854,19 +855,24 @@ def migrate_to_enhanced_location_accuracy(): return False def check_location_accuracy_column_exists(): - """Check if the location_accuracy column exists in the attendance_data table""" + """ + Check if location_accuracy column exists in attendance_data table (MySQL compatible) + """ try: + # MySQL-compatible query for checking column existence result = db.session.execute(text(""" - SELECT column_name - FROM information_schema.columns - WHERE table_name='attendance_data' AND column_name='location_accuracy' + SELECT COUNT(*) as count + FROM INFORMATION_SCHEMA.COLUMNS + WHERE TABLE_SCHEMA = DATABASE() + AND TABLE_NAME = 'attendance_data' + AND COLUMN_NAME = 'location_accuracy' """)) - column_exists = result.fetchone() is not None - return column_exists + count = result.fetchone().count + return count > 0 except Exception as e: - print(f"⚠️ Error checking location_accuracy column: {e}") + print(f"Error checking location_accuracy column: {e}") return False def get_employee_checkin_history(employee_id, qr_code_id, date_filter=None): @@ -2667,17 +2673,18 @@ def update_existing_qr_codes(): db.session.rollback() def add_coordinate_columns(): - """Add coordinate columns to existing qr_codes table""" + """Add coordinate columns to existing qr_codes table (MySQL compatible)""" try: - # Check if columns already exist + # Check if columns already exist - MySQL compatible query result = db.session.execute(text(""" - SELECT column_name - FROM information_schema.columns - WHERE table_name='qr_codes' AND column_name IN - ('address_latitude', 'address_longitude', 'coordinate_accuracy', 'coordinates_updated_date') + SELECT COLUMN_NAME + FROM INFORMATION_SCHEMA.COLUMNS + WHERE TABLE_SCHEMA = DATABASE() + AND TABLE_NAME = 'qr_codes' + AND COLUMN_NAME IN ('address_latitude', 'address_longitude', 'coordinate_accuracy', 'coordinates_updated_date') """)) - existing_columns = [row.column_name for row in result.fetchall()] + existing_columns = [row.COLUMN_NAME for row in result.fetchall()] # Add missing columns if 'address_latitude' not in existing_columns: diff --git a/app.py.bak b/app.py.bak new file mode 100644 index 0000000..5aa97af --- /dev/null +++ b/app.py.bak @@ -0,0 +1,2721 @@ +from flask import Flask, render_template, request, redirect, url_for, flash, session, jsonify +from flask_sqlalchemy import SQLAlchemy +from werkzeug.security import generate_password_hash, check_password_hash +from functools import wraps +from datetime import datetime, date, time, timedelta +from sqlalchemy import text +from user_agents import parse +from math import radians, cos, sin, asin, sqrt +import io, os, base64, re, uuid, requests, json, qrcode +from dotenv import load_dotenv + +# Load environment variables in .env +load_dotenv() + +# Initialize Flask application +app = Flask(__name__) +app.config['SECRET_KEY'] = os.environ.get('SECRET_KEY') +app.config['SQLALCHEMY_DATABASE_URI'] = os.environ.get('DATABASE_URL') +app.config['SQLALCHEMY_TRACK_MODIFICATIONS'] = os.environ.get('SQLALCHEMY_TRACK_MODIFICATIONS') + +# Initialize database +db = SQLAlchemy(app) + +# User Model +class User(db.Model): + """ + User model to manage system users with role-based access control + """ + __tablename__ = 'users' + + id = db.Column(db.Integer, primary_key=True) + full_name = db.Column(db.String(100), nullable=False) + email = db.Column(db.String(120), unique=True, nullable=False) + username = db.Column(db.String(80), unique=True, nullable=False) + password_hash = db.Column(db.String(128), nullable=False) + role = db.Column(db.String(20), nullable=False, default='staff') # admin or staff + created_by = db.Column(db.Integer, db.ForeignKey('users.id'), nullable=True) + created_date = db.Column(db.DateTime, default=datetime.utcnow) + active_status = db.Column(db.Boolean, default=True) + last_login_date = db.Column(db.DateTime, nullable=True) + + # Relationships + created_users = db.relationship('User', backref=db.backref('creator', remote_side=[id])) + created_qr_codes = db.relationship('QRCode', backref='creator', lazy='dynamic') + + def set_password(self, password): + """Hash and set user password""" + self.password_hash = generate_password_hash(password) + + def check_password(self, password): + """Verify user password""" + return check_password_hash(self.password_hash, password) + + def is_admin(self): + """Check if user has admin privileges""" + return self.role == 'admin' + +# QR Code Model +class QRCode(db.Model): + """ + Enhanced QR Code model to manage QR code records and metadata with address coordinates + """ + __tablename__ = 'qr_codes' + + id = db.Column(db.Integer, primary_key=True) + name = db.Column(db.String(100), nullable=False) + location = db.Column(db.String(100), nullable=False) + location_address = db.Column(db.Text, nullable=False) + location_event = db.Column(db.String(200), nullable=False) + qr_code_image = db.Column(db.Text, nullable=False) # Base64 encoded image + created_by = db.Column(db.Integer, db.ForeignKey('users.id'), nullable=False) + created_date = db.Column(db.DateTime, default=datetime.utcnow) + active_status = db.Column(db.Boolean, default=True) + qr_url = db.Column(db.String(255), unique=True, nullable=True) + + # NEW: Address Coordinates Fields + address_latitude = db.Column(db.Float, nullable=True) + address_longitude = db.Column(db.Float, nullable=True) + coordinate_accuracy = db.Column(db.String(50), nullable=True, default='geocoded') + coordinates_updated_date = db.Column(db.DateTime, nullable=True) + + @property + def has_coordinates(self): + """Check if this QR code has address coordinates""" + return self.address_latitude is not None and self.address_longitude is not None + @property + def coordinates_display(self): + """Get formatted coordinates for display""" + if self.has_coordinates: + return f"{self.address_latitude:.10f}, {self.address_longitude:.10f}" + return "Coordinates not available" + + def update_coordinates(self, latitude, longitude, accuracy='geocoded'): + """Update the address coordinates for this QR code""" + self.address_latitude = latitude + self.address_longitude = longitude + self.coordinate_accuracy = accuracy + self.coordinates_updated_date = datetime.utcnow() + +# Attendance Data Model +class AttendanceData(db.Model): + """Enhanced attendance tracking model with location support""" + __tablename__ = 'attendance_data' + + # Existing fields + id = db.Column(db.Integer, primary_key=True) + qr_code_id = db.Column(db.Integer, db.ForeignKey('qr_codes.id', ondelete='CASCADE'), nullable=False) + employee_id = db.Column(db.String(50), nullable=False) + check_in_date = db.Column(db.Date, nullable=False, default=datetime.today) + check_in_time = db.Column(db.Time, nullable=False, default=datetime.now().time) + device_info = db.Column(db.String(200)) + user_agent = db.Column(db.Text) + ip_address = db.Column(db.String(45)) + location_name = db.Column(db.String(100), nullable=False) + status = db.Column(db.String(20), default='present') + created_timestamp = db.Column(db.DateTime, default=datetime.utcnow) + updated_timestamp = db.Column(db.DateTime, default=datetime.utcnow, onupdate=datetime.utcnow) + + # LOCATION FIELDS - Add these if missing + latitude = db.Column(db.Float, nullable=True) + longitude = db.Column(db.Float, nullable=True) + accuracy = db.Column(db.Float, nullable=True) + location_accuracy = db.Column(db.Float, nullable=True) + altitude = db.Column(db.Float, nullable=True) + location_source = db.Column(db.String(50), default='manual') + address = db.Column(db.String(500), nullable=True) + + # Relationships + qr_code = db.relationship('QRCode', backref=db.backref('attendance_records', lazy='dynamic')) + + def __repr__(self): + return f'' + + @property + def has_location_data(self): + """Check if this record has GPS coordinates""" + return self.latitude is not None and self.longitude is not None + + @property + def location_accuracy_level(self): + """Get human-readable accuracy level""" + if not self.accuracy: + return 'unknown' + elif self.accuracy <= 50: + return 'high' + elif self.accuracy <= 100: + return 'medium' + else: + return 'low' + + @property + def coordinates_display(self): + """Get formatted coordinates for display""" + if self.has_location_data: + return f"{self.latitude:.10f}, {self.longitude:.10f}" + return "No GPS data" + + def to_dict(self): + """Convert to dictionary for JSON responses""" + return { + 'id': self.id, + 'employee_id': self.employee_id, + 'check_in_date': self.check_in_date.isoformat(), + 'check_in_time': self.check_in_time.isoformat(), + 'location_name': self.location_name, + 'status': self.status, + 'has_location': self.has_location_data, + 'coordinates': self.coordinates_display, + 'accuracy': self.accuracy, + 'address': self.address, + 'location_source': self.location_source + } + +# Utility functions +def get_coordinates_from_address(address): + """ + Get latitude and longitude from address using geocoding service + Returns (lat, lng) tuple or (None, None) if failed + """ + if not address or address.strip() == '': + return None, None + + try: + # Using a free geocoding service (Nominatim/OpenStreetMap) + # In production, consider using Google Maps Geocoding API for better accuracy + url = "https://nominatim.openstreetmap.org/search" + params = { + 'q': address, + 'format': 'json', + 'limit': 1, + 'addressdetails': 1 + } + + headers = { + 'User-Agent': 'QR-Attendance-System/1.0' + } + + response = requests.get(url, params=params, headers=headers, timeout=10) + + if response.status_code == 200: + data = response.json() + if data and len(data) > 0: + lat = float(data[0]['lat']) + lng = float(data[0]['lon']) + print(f"✅ Geocoded address '{address[:50]}...' to coordinates: {lat}, {lng}") + return lat, lng + + print(f"⚠️ Could not geocode address: {address}") + return None, None + + except Exception as e: + print(f"❌ Error geocoding address '{address}': {e}") + return None, None + +def get_coordinates_from_address_enhanced(address): + """ + Enhanced geocoding function with better error handling + Returns (latitude, longitude, accuracy_level) + """ + if not address or address.strip() == "": + print("⚠️ Empty address provided for geocoding") + return None, None, None + + address = address.strip() + print(f"🌍 Enhanced geocoding for: {address}") + + try: + # Primary geocoding using Nominatim (OpenStreetMap) + nominatim_url = "https://nominatim.openstreetmap.org/search" + params = { + 'q': address, + 'format': 'json', + 'limit': 1, + 'addressdetails': 1, + 'extratags': 1 + } + + headers = { + 'User-Agent': 'QR-Attendance-System/1.0 (Enhanced Location Accuracy)' + } + + response = requests.get(nominatim_url, params=params, headers=headers, timeout=10) + + if response.status_code == 200: + results = response.json() + + if results: + result = results[0] + lat = float(result['lat']) + lng = float(result['lon']) + + # Enhanced accuracy assessment + place_type = result.get('type', 'unknown') + osm_type = result.get('osm_type', 'unknown') + importance = float(result.get('importance', 0)) + + # More sophisticated accuracy determination + if place_type in ['house', 'building', 'shop', 'office'] or osm_type == 'way': + accuracy = 'excellent' + elif place_type in ['neighbourhood', 'suburb', 'quarter', 'residential']: + accuracy = 'good' + elif place_type in ['city', 'town', 'village'] and importance > 0.5: + accuracy = 'fair' + else: + accuracy = 'poor' + + print(f"✅ Enhanced geocoding successful:") + print(f" Coordinates: {lat:.10f}, {lng:.10f}") + print(f" Accuracy: {accuracy}") + + return lat, lng, accuracy + + print(f"⚠️ No results from enhanced geocoding for: {address}") + return None, None, None + + except Exception as e: + print(f"❌ Enhanced geocoding error: {e}") + return None, None, None + +def geocode_address_enhanced(address): + """ + Enhanced geocoding function for new coordinate features + Returns (lat, lng, accuracy) tuple or (None, None, None) if failed + """ + if not address or address.strip() == '': + return None, None, None + + try: + # Using Nominatim (OpenStreetMap) geocoding service + url = "https://nominatim.openstreetmap.org/search" + params = { + 'q': address, + 'format': 'json', + 'limit': 1, + 'addressdetails': 1 + } + + headers = { + 'User-Agent': 'QR-Attendance-System/1.0' + } + + response = requests.get(url, params=params, headers=headers, timeout=10) + + if response.status_code == 200: + data = response.json() + + if data and len(data) > 0: + result = data[0] + lat = float(result['lat']) + lng = float(result['lon']) + + # Determine accuracy based on result type + place_type = result.get('type', 'unknown') + osm_type = result.get('osm_type', 'unknown') + + if place_type in ['house', 'building'] or osm_type == 'way': + accuracy = 'high' + elif place_type in ['neighbourhood', 'suburb', 'quarter']: + accuracy = 'medium' + else: + accuracy = 'low' + + print(f"✅ Geocoded address: {address}") + print(f" Coordinates: {lat:.10f}, {lng:.10f}") + print(f" Accuracy: {accuracy} ({place_type})") + + return lat, lng, accuracy + + print(f"⚠️ No geocoding results for address: {address}") + return None, None, None + + except Exception as e: + print(f"❌ Geocoding error: {e}") + return None, None, None + +def calculate_distance_miles(lat1, lng1, lat2, lng2): + """ + Enhanced Haversine formula to calculate distance between two points in miles + Improved with better precision and error handling + """ + if any(coord is None for coord in [lat1, lng1, lat2, lng2]): + print("⚠️ Missing coordinates for distance calculation") + return None + + try: + # Validate coordinate ranges + if not (-90 <= lat1 <= 90) or not (-90 <= lat2 <= 90): + print(f"⚠️ Invalid latitude values: {lat1}, {lat2}") + return None + + if not (-180 <= lng1 <= 180) or not (-180 <= lng2 <= 180): + print(f"⚠️ Invalid longitude values: {lng1}, {lng2}") + return None + + # Convert decimal degrees to radians + lat1, lng1, lat2, lng2 = map(radians, [float(lat1), float(lng1), float(lat2), float(lng2)]) + + # Enhanced Haversine formula for better precision + dlng = lng2 - lng1 + dlat = lat2 - lat1 + + # Haversine calculation + a = sin(dlat/2)**2 + cos(lat1) * cos(lat2) * sin(dlng/2)**2 + c = 2 * asin(sqrt(a)) + + # Earth's radius in miles (more precise value) + r_miles = 3959.87433 + + # Calculate distance with enhanced precision + distance = c * r_miles + + # Round to 4 decimal places for better precision + distance = round(distance, 4) + + print(f"📏 Enhanced distance calculation:") + print(f" Point 1: {lat1*180/3.14159:.10f}, {lng1*180/3.14159:.10f}") + print(f" Point 2: {lat2*180/3.14159:.10f}, {lng2*180/3.14159:.10f}") + print(f" Distance: {distance:.4f} miles") + + return distance + + except Exception as e: + print(f"❌ Error in enhanced distance calculation: {e}") + return None + +def calculate_location_accuracy(qr_address, checkin_address, checkin_lat=None, checkin_lng=None): + """ + Calculate location accuracy by comparing QR code address with check-in location + Returns distance in miles between the two locations + """ + print(f"\n📍 CALCULATING LOCATION ACCURACY:") + print(f" QR Address: {qr_address}") + print(f" Check-in Address: {checkin_address}") + print(f" Check-in Coordinates: {checkin_lat}, {checkin_lng}") + + # Get QR code coordinates from address + qr_lat, qr_lng = get_coordinates_from_address(qr_address) + + if qr_lat is None or qr_lng is None: + print(f"⚠️ Could not geocode QR address, cannot calculate accuracy") + return None + + # Use check-in coordinates if available, otherwise geocode check-in address + if checkin_lat is not None and checkin_lng is not None: + checkin_coords_lat, checkin_coords_lng = checkin_lat, checkin_lng + print(f"✅ Using GPS coordinates for check-in location") + else: + checkin_coords_lat, checkin_coords_lng = get_coordinates_from_address(checkin_address) + if checkin_coords_lat is None or checkin_coords_lng is None: + print(f"⚠️ Could not geocode check-in address, cannot calculate accuracy") + return None + print(f"✅ Using geocoded coordinates for check-in address") + + # Calculate distance + distance = calculate_distance_miles(qr_lat, qr_lng, checkin_coords_lat, checkin_coords_lng) + + if distance is not None: + print(f"✅ Location accuracy calculated: {distance} miles") + + return distance + +def calculate_location_accuracy_enhanced(qr_address, checkin_address, checkin_lat=None, checkin_lng=None): + """ + ENHANCED location accuracy calculation comparing QR address with check-in location + This function provides improved precision and better error handling + + Parameters: + - qr_address: Address associated with the QR code + - checkin_address: Address where user checked in (from reverse geocoding) + - checkin_lat: GPS latitude from check-in (if available) + - checkin_lng: GPS longitude from check-in (if available) + + Returns: + - Distance in miles between QR location and check-in location + """ + print(f"\n🎯 ENHANCED LOCATION ACCURACY CALCULATION:") + print(f" QR Address: {qr_address}") + print(f" Check-in Address: {checkin_address}") + print(f" Check-in GPS: {checkin_lat}, {checkin_lng}") + print(f" Timestamp: {datetime.now()}") + + # Validate input parameters + if not qr_address or qr_address.strip() == "": + print(f"❌ QR address is empty or invalid") + return None + + # Step 1: Get coordinates for QR address using enhanced geocoding + print(f"\n📍 Step 1: Geocoding QR address...") + qr_lat, qr_lng, qr_accuracy = get_coordinates_from_address_enhanced(qr_address) + + if qr_lat is None or qr_lng is None: + print(f"❌ Could not geocode QR address: {qr_address}") + return None + + print(f"✅ QR location coordinates: {qr_lat:.10f}, {qr_lng:.10f} (accuracy: {qr_accuracy})") + + # Step 2: Determine check-in coordinates + print(f"\n📱 Step 2: Determining check-in coordinates...") + + checkin_coords_lat = None + checkin_coords_lng = None + checkin_source = "unknown" + + # Priority 1: Use GPS coordinates if available and valid + if checkin_lat is not None and checkin_lng is not None: + try: + lat_val = float(checkin_lat) + lng_val = float(checkin_lng) + + # Validate GPS coordinates + if -90 <= lat_val <= 90 and -180 <= lng_val <= 180: + checkin_coords_lat = lat_val + checkin_coords_lng = lng_val + checkin_source = "gps" + print(f"✅ Using GPS coordinates: {lat_val:.10f}, {lng_val:.10f}") + else: + print(f"⚠️ Invalid GPS coordinates: {lat_val}, {lng_val}") + except (ValueError, TypeError): + print(f"⚠️ Could not parse GPS coordinates") + + # Priority 2: Fallback to geocoding check-in address + if checkin_coords_lat is None and checkin_address: + print(f"🌍 Falling back to geocoding check-in address...") + checkin_coords_lat, checkin_coords_lng, checkin_accuracy = get_coordinates_from_address_enhanced(checkin_address) + if checkin_coords_lat is not None: + checkin_source = "address" + print(f"✅ Using geocoded coordinates: {checkin_coords_lat:.10f}, {checkin_coords_lng:.10f} (accuracy: {checkin_accuracy})") + + # Check if we have valid coordinates for both locations + if checkin_coords_lat is None or checkin_coords_lng is None: + print(f"❌ Could not determine check-in coordinates") + print(f" GPS: {checkin_lat}, {checkin_lng}") + print(f" Address: {checkin_address}") + return None + + # Step 3: Calculate enhanced distance + print(f"\n📏 Step 3: Calculating enhanced distance...") + distance = calculate_distance_miles(qr_lat, qr_lng, checkin_coords_lat, checkin_coords_lng) + + if distance is not None: + print(f"✅ Enhanced location accuracy calculated successfully!") + print(f" QR Location: {qr_lat:.10f}, {qr_lng:.10f}") + print(f" Check-in Location: {checkin_coords_lat:.10f}, {checkin_coords_lng:.10f}") + print(f" Source: {checkin_source}") + print(f" Distance: {distance:.4f} miles") + print(f" Accuracy Level: {get_location_accuracy_level_enhanced(distance)}") + else: + print(f"❌ Failed to calculate distance") + + return distance + +def generate_qr_url(name, qr_id): + """Generate a unique URL for QR code destination""" + # Clean the name for URL use + clean_name = re.sub(r'[^a-zA-Z0-9\s-]', '', name) + clean_name = re.sub(r'\s+', '-', clean_name.strip()) + clean_name = clean_name.lower() + + # Create unique URL + url_slug = f"qr-{qr_id}-{clean_name}" + return url_slug[:200] # Limit length + +def detect_device_info(user_agent_string): + """Extract device information from user agent""" + try: + user_agent = parse(user_agent_string) + device_info = f"{user_agent.device.family}" + + if user_agent.os.family: + device_info += f" - {user_agent.os.family}" + if user_agent.os.version_string: + device_info += f" {user_agent.os.version_string}" + + if user_agent.browser.family: + device_info += f" ({user_agent.browser.family})" + + return device_info[:200] # Limit length + except: + return "Unknown Device" + +def get_client_ip(): + """Get client IP address""" + if request.environ.get('HTTP_X_FORWARDED_FOR') is None: + return request.environ['REMOTE_ADDR'] + else: + return request.environ['HTTP_X_FORWARDED_FOR'] + +def calculate_distance_miles(lat1, lng1, lat2, lng2): + """ + Calculate the great circle distance between two points on Earth in miles + Using the Haversine formula + """ + if any(coord is None for coord in [lat1, lng1, lat2, lng2]): + return None + + try: + # Convert decimal degrees to radians + lat1, lng1, lat2, lng2 = map(radians, [lat1, lng1, lat2, lng2]) + + # Haversine formula + dlng = lng2 - lng1 + dlat = lat2 - lat1 + a = sin(dlat/2)**2 + cos(lat1) * cos(lat2) * sin(dlng/2)**2 + c = 2 * asin(sqrt(a)) + + # Radius of Earth in miles + r_miles = 3959 + + # Calculate the result + distance = c * r_miles + + print(f"📏 Calculated distance: {distance:.3f} miles") + return round(distance, 3) + + except Exception as e: + print(f"❌ Error calculating distance: {e}") + return None + +def get_location_accuracy_level(location_accuracy): + """Get human-readable location accuracy level based on distance""" + if not location_accuracy: + return 'unknown' + elif location_accuracy <= 0.1: # Within 0.1 mile (528 feet) + return 'excellent' + elif location_accuracy <= 0.5: # Within 0.5 mile + return 'good' + elif location_accuracy <= 1.0: # Within 1 mile + return 'fair' + else: + return 'poor' + +def get_location_accuracy_level_enhanced(location_accuracy): + """ + Enhanced function to categorize location accuracy with more granular levels + """ + if not location_accuracy or location_accuracy is None: + return 'unknown' + + # More precise accuracy thresholds + if location_accuracy <= 0.05: # Within 264 feet (50 meters) + return 'excellent' + elif location_accuracy <= 0.1: # Within 528 feet (100 meters) + return 'very_good' + elif location_accuracy <= 0.25: # Within 0.25 mile (1320 feet) + return 'good' + elif location_accuracy <= 0.5: # Within 0.5 mile + return 'fair' + elif location_accuracy <= 1.0: # Within 1 mile + return 'poor' + else: # Greater than 1 mile + return 'very_poor' + +def process_location_data(location_data): + """ + Process and validate location data from form + Returns clean location data or None values for invalid data + """ + processed = { + 'latitude': None, + 'longitude': None, + 'accuracy': None, + 'altitude': None, + 'source': location_data.get('location_source', 'manual'), + 'address': location_data.get('address', '')[:500] if location_data.get('address') else None + } + + try: + # Process latitude + if location_data.get('latitude') and location_data['latitude'] not in ['null', '']: + lat = float(location_data['latitude']) + if -90 <= lat <= 90: # Valid latitude range + processed['latitude'] = lat + else: + print(f"⚠️ Invalid latitude: {lat}") + + # Process longitude + if location_data.get('longitude') and location_data['longitude'] not in ['null', '']: + lng = float(location_data['longitude']) + if -180 <= lng <= 180: # Valid longitude range + processed['longitude'] = lng + else: + print(f"⚠️ Invalid longitude: {lng}") + + # Process accuracy + if location_data.get('accuracy') and location_data['accuracy'] not in ['null', '']: + acc = float(location_data['accuracy']) + if acc >= 0: # Accuracy should be positive + processed['accuracy'] = acc + else: + print(f"⚠️ Invalid accuracy: {acc}") + + # Process altitude + if location_data.get('altitude') and location_data['altitude'] not in ['null', '']: + alt = float(location_data['altitude']) + # Altitude can be negative (below sea level) + processed['altitude'] = alt + + except (ValueError, TypeError) as e: + print(f"⚠️ Error processing location data: {e}") + + return processed + +def reverse_geocode_coordinates(latitude, longitude): + """ + Convert GPS coordinates to human-readable address using reverse geocoding + Returns address string or None if failed + """ + if not latitude or not longitude: + return None + + try: + print(f"🌍 Reverse geocoding coordinates: {latitude}, {longitude}") + + # Using Nominatim (OpenStreetMap) reverse geocoding service + url = "https://nominatim.openstreetmap.org/reverse" + params = { + 'lat': latitude, + 'lon': longitude, + 'format': 'json', + 'addressdetails': 1, + 'zoom': 18 # High detail level + } + + headers = { + 'User-Agent': 'QR-Attendance-System/1.0' + } + + response = requests.get(url, params=params, headers=headers, timeout=10) + + if response.status_code == 200: + data = response.json() + + if data and 'display_name' in data: + address = data['display_name'] + print(f"✅ Reverse geocoded address: {address}") + return address + else: + print(f"⚠️ No address found for coordinates") + return None + else: + print(f"⚠️ Reverse geocoding API returned status: {response.status_code}") + return None + + except Exception as e: + print(f"❌ Error in reverse geocoding: {e}") + return None + +def process_location_data_enhanced(form_data): + """ + Enhanced processing of location data from form submission + Validates and cleans location data for storage, including reverse geocoding + """ + processed = { + 'latitude': None, + 'longitude': None, + 'accuracy': None, + 'altitude': None, + 'source': form_data.get('location_source', 'manual'), + 'address': None + } + + try: + # Process latitude + if form_data.get('latitude') and form_data['latitude'] not in ['null', '', 'undefined']: + lat = float(form_data['latitude']) + if -90 <= lat <= 90: # Valid latitude range + processed['latitude'] = lat + else: + print(f"⚠️ Invalid latitude: {lat}") + + # Process longitude + if form_data.get('longitude') and form_data['longitude'] not in ['null', '', 'undefined']: + lng = float(form_data['longitude']) + if -180 <= lng <= 180: # Valid longitude range + processed['longitude'] = lng + else: + print(f"⚠️ Invalid longitude: {lng}") + + # Process GPS accuracy + if form_data.get('accuracy') and form_data['accuracy'] not in ['null', '', 'undefined']: + acc = float(form_data['accuracy']) + if acc >= 0: # Accuracy should be positive + processed['accuracy'] = acc + else: + print(f"⚠️ Invalid GPS accuracy: {acc}") + + # Process altitude + if form_data.get('altitude') and form_data['altitude'] not in ['null', '', 'undefined']: + alt = float(form_data['altitude']) + processed['altitude'] = alt + + # Process address - First check if address was provided + if form_data.get('address'): + address = form_data['address'].strip() + if address and address not in ['null', '', 'undefined']: + # Check if the address is just coordinates (like "38.8104192000, -77.1850240000") + if re.match(r'^-?\d+\.\d+,?\s*-?\d+\.\d+$', address.replace(' ', '')): + print(f"🔍 Detected coordinate-format address: {address}") + # This is just coordinates, we need to reverse geocode + processed['address'] = None # Reset so reverse geocoding will trigger + else: + # This is a real address + processed['address'] = address[:500] # Limit to 500 characters + print(f"✅ Using provided address: {processed['address'][:100]}...") + + # CRITICAL: If we have coordinates but no real address, perform reverse geocoding + if (processed['latitude'] is not None and processed['longitude'] is not None + and not processed['address']): + print(f"🌍 Performing reverse geocoding for coordinates: {processed['latitude']}, {processed['longitude']}") + reverse_geocoded_address = reverse_geocode_coordinates(processed['latitude'], processed['longitude']) + if reverse_geocoded_address: + processed['address'] = reverse_geocoded_address[:500] + print(f"✅ Reverse geocoded address: {processed['address']}") + else: + print(f"⚠️ Could not reverse geocode coordinates, keeping coordinates as fallback") + processed['address'] = f"{processed['latitude']:.10f}, {processed['longitude']:.10f}" + + print(f"📍 Final processed location data:") + print(f" Coordinates: {processed['latitude']}, {processed['longitude']}") + print(f" GPS Accuracy: {processed['accuracy']}m") + print(f" Source: {processed['source']}") + print(f" Address: {processed['address'][:100] if processed['address'] else 'None'}...") + + return processed + + except Exception as e: + print(f"❌ Error processing location data: {e}") + return processed + +def migrate_to_enhanced_location_accuracy(): + """ + Migration function to recalculate all existing records with enhanced accuracy + """ + try: + print("🔄 Starting enhanced location accuracy migration...") + + # Get all records that need recalculation + records = db.session.execute(text(""" + SELECT ad.id, qc.location_address, ad.address, ad.latitude, ad.longitude, ad.location_accuracy + FROM attendance_data ad + LEFT JOIN qr_codes qc ON ad.qr_code_id = qc.id + WHERE qc.location_address IS NOT NULL + """)).fetchall() + + print(f"📊 Found {len(records)} records to process") + + updated_count = 0 + improved_count = 0 + + for record in records: + try: + # Calculate enhanced location accuracy + new_accuracy = calculate_location_accuracy_enhanced( + qr_address=record.location_address, + checkin_address=record.address, + checkin_lat=record.latitude, + checkin_lng=record.longitude + ) + + if new_accuracy is not None: + # Update the record + db.session.execute(text(""" + UPDATE attendance_data + SET location_accuracy = :accuracy + WHERE id = :record_id + """), { + 'accuracy': new_accuracy, + 'record_id': record.id + }) + + updated_count += 1 + + # Check if this is an improvement + if record.location_accuracy is None or abs(new_accuracy - (record.location_accuracy or 0)) > 0.001: + improved_count += 1 + print(f" ✅ Updated record {record.id}: {record.location_accuracy} → {new_accuracy:.4f} miles") + + except Exception as e: + print(f" ⚠️ Error processing record {record.id}: {e}") + + # Commit all changes + db.session.commit() + + print(f"✅ Enhanced migration completed!") + print(f" 📊 Records processed: {len(records)}") + print(f" ✅ Records updated: {updated_count}") + print(f" 📈 Records improved: {improved_count}") + + return True + + except Exception as e: + print(f"❌ Enhanced migration failed: {e}") + db.session.rollback() + return False + +def check_location_accuracy_column_exists(): + """Check if the location_accuracy column exists in the attendance_data table""" + try: + result = db.session.execute(text(""" + SELECT column_name + FROM information_schema.columns + WHERE table_name='attendance_data' AND column_name='location_accuracy' + """)) + + column_exists = result.fetchone() is not None + return column_exists + + except Exception as e: + print(f"⚠️ Error checking location_accuracy column: {e}") + return False + +def get_employee_checkin_history(employee_id, qr_code_id, date_filter=None): + """ + NEW HELPER FUNCTION: Get check-in history for an employee at a specific location + """ + try: + if date_filter is None: + date_filter = date.today() + + checkins = AttendanceData.query.filter_by( + employee_id=employee_id.upper(), + qr_code_id=qr_code_id, + check_in_date=date_filter + ).order_by(AttendanceData.check_in_time.asc()).all() + + return checkins + + except Exception as e: + print(f"❌ Error retrieving checkin history: {e}") + return [] + +def format_checkin_intervals(checkins): + """ + NEW HELPER FUNCTION: Format time intervals between check-ins for display + """ + if len(checkins) < 2: + return [] + + intervals = [] + for i in range(1, len(checkins)): + previous_time = datetime.combine(checkins[i-1].check_in_date, checkins[i-1].check_in_time) + current_time = datetime.combine(checkins[i].check_in_date, checkins[i].check_in_time) + + interval = current_time - previous_time + interval_minutes = int(interval.total_seconds() / 60) + + intervals.append({ + 'from_time': checkins[i-1].check_in_time.strftime('%H:%M'), + 'to_time': checkins[i].check_in_time.strftime('%H:%M'), + 'interval_minutes': interval_minutes, + 'interval_text': format_time_interval(interval_minutes) + }) + + return intervals + +def format_time_interval(minutes): + """ + NEW HELPER FUNCTION: Format minutes into human-readable time interval + """ + if minutes < 60: + return f"{minutes} minutes" + elif minutes < 1440: # Less than 24 hours + hours = minutes // 60 + remaining_minutes = minutes % 60 + if remaining_minutes == 0: + return f"{hours} hour{'s' if hours != 1 else ''}" + else: + return f"{hours}h {remaining_minutes}m" + else: + days = minutes // 1440 + remaining_hours = (minutes % 1440) // 60 + if remaining_hours == 0: + return f"{days} day{'s' if days != 1 else ''}" + else: + return f"{days}d {remaining_hours}h" + +# Authentication decorator +def login_required(f): + """Decorator to ensure user is logged in""" + @wraps(f) + def decorated_function(*args, **kwargs): + if 'user_id' not in session: + flash('Please log in to access this page.', 'error') + return redirect(url_for('login')) + return f(*args, **kwargs) + return decorated_function + +def admin_required(f): + @wraps(f) + def decorated_function(*args, **kwargs): + if 'username' not in session: + flash('Please log in to access this page.', 'error') + return redirect(url_for('login')) + + if session.get('role') != 'admin': + flash('Administrator privileges required for this action.', 'error') + return redirect(url_for('dashboard')) + + return f(*args, **kwargs) + return decorated_function + +# Utility function to generate QR code +def generate_qr_code(data): + """Generate QR code image and return as base64 string""" + qr = qrcode.QRCode( + version=1, + error_correction=qrcode.constants.ERROR_CORRECT_L, + box_size=10, + border=4, + ) + qr.add_data(data) + qr.make(fit=True) + + img = qr.make_image(fill_color="black", back_color="white") + + # Convert to base64 + buffer = io.BytesIO() + img.save(buffer, format='PNG') + img_str = base64.b64encode(buffer.getvalue()).decode() + + return img_str + +@app.template_filter('strftime') +def strftime_filter(value, format='%m/%d/%Y'): + """Format datetime/date/string as strftime""" + if isinstance(value, str): + if value.lower() == 'now': + return datetime.now().strftime(format) + try: + # Try to parse string as datetime + dt = datetime.fromisoformat(value) + return dt.strftime(format) + except (ValueError, TypeError): + return value + + if hasattr(value, 'strftime'): + return value.strftime(format) + + return str(value) + +# Routes +@app.route('/') +def index(): + """Home page - redirect to login if not authenticated""" + if 'user_id' in session: + return redirect(url_for('dashboard')) + return redirect(url_for('login')) + +@app.route('/register', methods=['GET', 'POST']) +def register(): + """User registration endpoint""" + if request.method == 'POST': + full_name = request.form['full_name'] + email = request.form['email'] + username = request.form['username'] + password = request.form['password'] + + # Check if user already exists + if User.query.filter_by(username=username).first(): + flash('Username already exists.', 'error') + return render_template('register.html') + + if User.query.filter_by(email=email).first(): + flash('Email already registered.', 'error') + return render_template('register.html') + + # Create new user (default role: staff) + new_user = User( + full_name=full_name, + email=email, + username=username, + role='staff' + ) + new_user.set_password(password) + + db.session.add(new_user) + db.session.commit() + + flash('Registration successful! Please log in.', 'success') + return redirect(url_for('login')) + + return render_template('register.html') + +@app.route('/logout') +def logout(): + """User logout endpoint""" + session.clear() + flash('You have been logged out.', 'info') + return redirect(url_for('login')) + +@app.route('/dashboard') +@login_required +def dashboard(): + """Main dashboard after login - Fixed to show all QR codes""" + user = User.query.get(session['user_id']) + + # Get ALL QR codes (both active and inactive) with proper error handling + # The frontend filtering will handle display logic + try: + qr_codes = QRCode.query.order_by(QRCode.created_date.desc()).all() # ✅ Fixed: removed filter + except Exception as e: + print(f"Error fetching QR codes: {e}") + qr_codes = [] + + return render_template('dashboard.html', user=user, qr_codes=qr_codes) + +# User management routes +@app.route('/profile', methods=['GET', 'POST']) +@login_required +def profile(): + """User profile management""" + user = User.query.get(session['user_id']) + + if request.method == 'POST': + form_type = request.form.get('form_type') + + if form_type == 'profile': + # Update profile information + user.full_name = request.form['full_name'] + user.email = request.form['email'] + + db.session.commit() + flash('Profile updated successfully!', 'success') + + elif form_type == 'password': + # Update password + current_password = request.form['current_password'] + new_password = request.form['new_password'] + + if user.check_password(current_password): + user.set_password(new_password) + db.session.commit() + flash('Password updated successfully!', 'success') + else: + flash('Current password is incorrect.', 'error') + + return redirect(url_for('profile')) + + return render_template('profile.html', user=user) + +@app.route('/users') +@admin_required +def users(): + """User management page (Admin only) - Enhanced with better data""" + try: + # Get all users with their QR code counts + all_users = db.session.query(User).all() + + # Add QR code counts to each user + for user in all_users: + user.qr_code_count = user.created_qr_codes.count() + user.active_qr_count = user.created_qr_codes.filter_by(active_status=True).count() + + print(f"Found {len(all_users)} users for admin view") + return render_template('users.html', users=all_users) + + except Exception as e: + print(f"Error fetching users: {e}") + flash('Error loading users. Please try again.', 'error') + return redirect(url_for('dashboard')) + +@app.route('/users/create', methods=['GET', 'POST']) +@admin_required +def create_user(): + """Create new user (Admin only)""" + if request.method == 'POST': + try: + full_name = request.form.get('full_name', '').strip() + email = request.form.get('email', '').strip().lower() + username = request.form.get('username', '').strip().lower() + password = request.form.get('password', '') + role = request.form.get('role', '') + + # Validation + if not all([full_name, email, username, password, role]): + flash('All fields are required.', 'error') + return render_template('create_user.html') + + if len(password) < 6: + flash('Password must be at least 6 characters long.', 'error') + return render_template('create_user.html') + + if role not in ['staff', 'admin']: + flash('Invalid role specified.', 'error') + return render_template('create_user.html') + + # Check if user already exists + if User.query.filter_by(username=username).first(): + flash('Username already exists. Please choose a different username.', 'error') + return render_template('create_user.html') + + if User.query.filter_by(email=email).first(): + flash('Email already registered. Please use a different email.', 'error') + return render_template('create_user.html') + + # Create user + new_user = User( + full_name=full_name, + email=email, + username=username, + role=role, + created_by=session['user_id'], + created_date=datetime.utcnow(), + active_status=True + ) + new_user.set_password(password) + + db.session.add(new_user) + db.session.commit() + + flash(f'User "{full_name}" created successfully!', 'success') + print(f"Admin {session['username']} created user: {username} with role: {role}") + + return redirect(url_for('users')) + + except Exception as e: + db.session.rollback() + print(f"Error creating user: {e}") + flash('Error creating user. Please try again.', 'error') + return render_template('create_user.html') + + return render_template('create_user.html') + +@app.route('/users//delete', methods=['GET', 'POST']) +@admin_required +def delete_user(user_id): + """Deactivate user (Admin only) - Fixed with proper validation""" + try: + user_to_delete = User.query.get(user_id) + current_user = User.query.get(session['user_id']) + + if not user_to_delete: + flash('User not found.', 'error') + return redirect(url_for('users')) + + # Prevent self-deletion + if user_to_delete.id == current_user.id: + flash('You cannot deactivate your own account. Ask another admin to do this.', 'error') + return redirect(url_for('users')) + + # Check if trying to delete the last admin + if user_to_delete.role == 'admin': + active_admin_count = User.query.filter_by(role='admin', active_status=True).count() + if active_admin_count <= 1: + flash('Cannot deactivate the last admin user. Promote another user to admin first.', 'error') + return redirect(url_for('users')) + + # Deactivate the user instead of deleting + user_to_delete.active_status = False + db.session.commit() + + flash(f'User "{user_to_delete.full_name}" has been deactivated successfully.', 'success') + print(f"Admin {current_user.username} deactivated user: {user_to_delete.username}") + + return redirect(url_for('users')) + + except Exception as e: + db.session.rollback() + print(f"Error deactivating user: {e}") + flash('Error deactivating user. Please try again.', 'error') + return redirect(url_for('users')) + +@app.route('/users//reactivate', methods=['GET', 'POST']) +@admin_required +def reactivate_user(user_id): + """Reactivate a deactivated user (Admin only)""" + try: + user_to_reactivate = User.query.get(user_id) + current_user = User.query.get(session['user_id']) + + if not user_to_reactivate: + flash('User not found.', 'error') + return redirect(url_for('users')) + + if user_to_reactivate.active_status: + flash('User is already active.', 'info') + else: + user_to_reactivate.active_status = True + db.session.commit() + flash(f'User "{user_to_reactivate.full_name}" has been reactivated successfully.', 'success') + print(f"Admin {current_user.username} reactivated user: {user_to_reactivate.username}") + + return redirect(url_for('users')) + + except Exception as e: + db.session.rollback() + print(f"Error reactivating user: {e}") + flash('Error reactivating user. Please try again.', 'error') + return redirect(url_for('users')) + +@app.route('/users//promote', methods=['GET', 'POST']) +@admin_required +def promote_user(user_id): + """Promote a staff user to admin (Admin only)""" + try: + user_to_promote = User.query.get(user_id) + current_user = User.query.get(session['user_id']) + + if not user_to_promote: + flash('User not found.', 'error') + return redirect(url_for('users')) + + if user_to_promote.role == 'admin': + flash('User is already an admin.', 'info') + else: + user_to_promote.role = 'admin' + db.session.commit() + flash(f'"{user_to_promote.full_name}" has been promoted to admin.', 'success') + print(f"Admin {current_user.username} promoted user {user_to_promote.username} to admin") + + return redirect(url_for('users')) + + except Exception as e: + db.session.rollback() + print(f"Error promoting user: {e}") + flash('Error promoting user. Please try again.', 'error') + return redirect(url_for('users')) + +@app.route('/users//demote', methods=['GET', 'POST']) +@admin_required +def demote_user(user_id): + """Demote an admin user to staff (Admin only)""" + try: + user_to_demote = User.query.get(user_id) + current_user = User.query.get(session['user_id']) + + if not user_to_demote: + flash('User not found.', 'error') + return redirect(url_for('users')) + + # Prevent self-demotion + if user_to_demote.id == current_user.id: + flash('You cannot demote yourself. Have another admin do this.', 'error') + return redirect(url_for('users')) + + # Check if this is the last admin + active_admin_count = User.query.filter_by(role='admin', active_status=True).count() + if active_admin_count <= 1 and user_to_demote.role == 'admin': + flash('Cannot demote the last admin user. Promote another user to admin first.', 'error') + return redirect(url_for('users')) + + if user_to_demote.role == 'staff': + flash('User is already staff.', 'info') + else: + user_to_demote.role = 'staff' + db.session.commit() + flash(f'"{user_to_demote.full_name}" has been demoted to staff.', 'success') + print(f"Admin {current_user.username} demoted user {user_to_demote.username} to staff") + + return redirect(url_for('users')) + + except Exception as e: + db.session.rollback() + print(f"Error demoting user: {e}") + flash('Error demoting user. Please try again.', 'error') + return redirect(url_for('users')) + +@app.route('/users//edit', methods=['GET', 'POST']) +@admin_required +def edit_user(user_id): + """Edit user information (Admin only)""" + try: + user_to_edit = User.query.get(user_id) + current_user = User.query.get(session['user_id']) + + if not user_to_edit: + flash('User not found.', 'error') + return redirect(url_for('users')) + + if request.method == 'POST': + full_name = request.form.get('full_name', '').strip() + email = request.form.get('email', '').strip().lower() + new_role = request.form.get('role', '') + new_password = request.form.get('new_password', '').strip() + + # Validation + if not all([full_name, email, new_role]): + flash('Name, email, and role are required.', 'error') + return render_template('edit_user.html', user=user_to_edit) + + if new_role not in ['staff', 'admin']: + flash('Invalid role specified.', 'error') + return render_template('edit_user.html', user=user_to_edit) + + # Check for email conflicts (excluding current user) + existing_email_user = User.query.filter_by(email=email).first() + if existing_email_user and existing_email_user.id != user_to_edit.id: + flash('Email already in use by another user.', 'error') + return render_template('edit_user.html', user=user_to_edit) + + # Prevent self-demotion + if (user_to_edit.id == current_user.id and + user_to_edit.role == 'admin' and new_role == 'staff'): + flash('You cannot demote yourself. Have another admin do this.', 'error') + return render_template('edit_user.html', user=user_to_edit) + + # Check if demoting the last admin + if (user_to_edit.role == 'admin' and new_role == 'staff'): + active_admin_count = User.query.filter_by(role='admin', active_status=True).count() + if active_admin_count <= 1: + flash('Cannot demote the last admin user. Promote another user to admin first.', 'error') + return render_template('edit_user.html', user=user_to_edit) + + # Update user information + user_to_edit.full_name = full_name + user_to_edit.email = email + user_to_edit.role = new_role + + # Handle password change if provided + if new_password: + if len(new_password) < 6: + flash('Password must be at least 6 characters long.', 'error') + return render_template('edit_user.html', user=user_to_edit) + user_to_edit.set_password(new_password) + + db.session.commit() + flash(f'User "{user_to_edit.full_name}" updated successfully.', 'success') + print(f"Admin {current_user.username} updated user: {user_to_edit.username}") + + return redirect(url_for('users')) + + return render_template('edit_user.html', user=user_to_edit) + + except Exception as e: + db.session.rollback() + print(f"Error editing user: {e}") + flash('Error updating user. Please try again.', 'error') + return redirect(url_for('users')) + +# ENHANCED USER STATISTICS API +@app.route('/api/users/stats') +@admin_required +def user_stats_api(): + """API endpoint for user statistics""" + try: + total_users = User.query.count() + active_users = User.query.filter_by(active_status=True).count() + admin_users = User.query.filter_by(role='admin', active_status=True).count() + staff_users = User.query.filter_by(role='staff', active_status=True).count() + inactive_users = User.query.filter_by(active_status=False).count() + + # Recent registrations (last 30 days) + thirty_days_ago = datetime.utcnow() - timedelta(days=30) + recent_registrations = User.query.filter(User.created_date >= thirty_days_ago).count() + + # Recent logins (last 7 days) + seven_days_ago = datetime.utcnow() - timedelta(days=7) + recent_logins = User.query.filter( + User.last_login_date >= seven_days_ago, + User.active_status == True + ).count() + + return jsonify({ + 'total_users': total_users, + 'active_users': active_users, + 'admin_users': admin_users, + 'staff_users': staff_users, + 'inactive_users': inactive_users, + 'recent_registrations': recent_registrations, + 'recent_logins': recent_logins + }) + + except Exception as e: + print(f"Error fetching user stats: {e}") + return jsonify({'error': 'Failed to fetch user statistics'}), 500 + +@app.route('/api/geocode', methods=['POST']) +@login_required # Add this decorator if you have it +def geocode_address(): + """ + API endpoint to geocode an address and return coordinates + """ + try: + data = request.get_json() + address = data.get('address', '').strip() + + if not address: + return jsonify({ + 'success': False, + 'message': 'Address is required' + }), 400 + + # Use the enhanced function that returns 3 values + lat, lng, accuracy = geocode_address_enhanced(address) + + if lat is not None and lng is not None: + return jsonify({ + 'success': True, + 'data': { + 'latitude': lat, + 'longitude': lng, + 'accuracy': accuracy, + 'coordinates_display': f"{lat:.10f}, {lng:.10f}" + }, + 'message': f'Address geocoded successfully with {accuracy} accuracy' + }) + else: + return jsonify({ + 'success': False, + 'message': 'Unable to geocode the provided address. Please verify the address is complete and accurate.' + }), 400 + + except Exception as e: + print(f"❌ Geocoding API error: {e}") + return jsonify({ + 'success': False, + 'message': 'An error occurred while geocoding the address' + }), 500 + +@app.route('/users//permanently-delete', methods=['GET', 'POST']) +@admin_required +def permanently_delete_user(user_id): + """Permanently delete user and all associated data (Admin only)""" + try: + user_to_delete = User.query.get_or_404(user_id) + current_user = User.query.get(session['user_id']) + + # Security checks + if user_to_delete.id == current_user.id: + flash('You cannot delete your own account.', 'error') + return redirect(url_for('users')) + + # Only allow deletion of inactive users for safety + if user_to_delete.active_status: + flash('User must be deactivated before permanent deletion.', 'error') + return redirect(url_for('users')) + + # If deleting an admin, ensure at least one admin remains + if user_to_delete.role == 'admin': + active_admin_count = User.query.filter_by(role='admin', active_status=True).count() + if active_admin_count <= 1: + flash('Cannot delete the last admin user in the system.', 'error') + return redirect(url_for('users')) + + user_name = user_to_delete.full_name + user_qr_count = user_to_delete.created_qr_codes.count() + + # Delete all QR codes created by this user + QRCode.query.filter_by(created_by=user_id).delete() + + # Update any users that were created by this user (set created_by to None) + created_users = User.query.filter_by(created_by=user_id).all() + for created_user in created_users: + created_user.created_by = None + + # Delete the user + db.session.delete(user_to_delete) + db.session.commit() + + flash(f'User "{user_name}" and {user_qr_count} associated QR codes have been permanently deleted.', 'success') + print(f"Admin {current_user.username} permanently deleted user: {user_to_delete.username}") + + return redirect(url_for('users')) + + except Exception as e: + db.session.rollback() + print(f"Error permanently deleting user: {e}") + flash('Error deleting user. Please try again.', 'error') + return redirect(url_for('users')) + +# BULK USER OPERATIONS +@app.route('/users/bulk/deactivate', methods=['POST']) +@admin_required +def bulk_deactivate_users(): + """Bulk deactivate multiple users (Admin only)""" + try: + user_ids = request.json.get('user_ids', []) + current_user_id = session['user_id'] + current_user = User.query.get(current_user_id) + + if not user_ids: + return jsonify({'error': 'No users selected'}), 400 + + # Filter out current user and validate + valid_user_ids = [] + admin_count = User.query.filter_by(role='admin', active_status=True).count() + admins_to_deactivate = 0 + + for user_id in user_ids: + if user_id == current_user_id: + continue # Skip current user + + user = User.query.get(user_id) + if user and user.active_status: + if user.role == 'admin': + admins_to_deactivate += 1 + valid_user_ids.append(user_id) + + # Check if we're trying to deactivate all admins + if admin_count - admins_to_deactivate < 1: + return jsonify({'error': 'Cannot deactivate all admin users'}), 400 + + # Deactivate users + deactivated_count = 0 + for user_id in valid_user_ids: + user = User.query.get(user_id) + if user: + user.active_status = False + deactivated_count += 1 + + db.session.commit() + + return jsonify({ + 'success': True, + 'message': f'Successfully deactivated {deactivated_count} users', + 'deactivated_count': deactivated_count + }) + + except Exception as e: + db.session.rollback() + print(f"Error in bulk deactivate: {e}") + return jsonify({'error': 'Failed to deactivate users'}), 500 + +@app.route('/users/bulk/activate', methods=['POST']) +@admin_required +def bulk_activate_users(): + """Bulk activate multiple users (Admin only)""" + try: + user_ids = request.json.get('user_ids', []) + + if not user_ids: + return jsonify({'error': 'No users selected'}), 400 + + # Activate users + activated_count = 0 + for user_id in user_ids: + user = User.query.get(user_id) + if user and not user.active_status: + user.active_status = True + activated_count += 1 + + db.session.commit() + + return jsonify({ + 'success': True, + 'message': f'Successfully activated {activated_count} users', + 'activated_count': activated_count + }) + + except Exception as e: + db.session.rollback() + print(f"Error in bulk activate: {e}") + return jsonify({'error': 'Failed to activate users'}), 500 + +@app.route('/users/bulk/permanently-delete', methods=['POST']) +@admin_required +def bulk_permanently_delete_users(): + """Bulk permanently delete multiple users and all associated data (Admin only)""" + try: + user_ids = request.json.get('user_ids', []) + current_user_id = session['user_id'] + current_user = User.query.get(current_user_id) + + if not user_ids: + return jsonify({'error': 'No users selected'}), 400 + + # Convert string IDs to integers for safety + try: + user_ids = [int(uid) for uid in user_ids] + except (ValueError, TypeError): + return jsonify({'error': 'Invalid user IDs provided'}), 400 + + # Security validations + deleted_users = [] + deleted_qr_count = 0 + errors = [] + + for user_id in user_ids: + try: + # Skip current user + if user_id == current_user_id: + errors.append(f"Cannot delete your own account") + continue + + user_to_delete = User.query.get(user_id) + if not user_to_delete: + errors.append(f"User with ID {user_id} not found") + continue + + # Only allow deletion of inactive users for safety + if user_to_delete.active_status: + errors.append(f"User '{user_to_delete.full_name}' must be deactivated before permanent deletion") + continue + + # If deleting an admin, ensure at least one admin remains + if user_to_delete.role == 'admin': + active_admin_count = User.query.filter_by(role='admin', active_status=True).count() + if active_admin_count <= 1: + errors.append(f"Cannot delete the last admin user '{user_to_delete.full_name}'") + continue + + # Count QR codes before deletion for reporting + user_qr_count = user_to_delete.created_qr_codes.count() + deleted_qr_count += user_qr_count + + # Delete all QR codes created by this user + QRCode.query.filter_by(created_by=user_id).delete() + + # Update any users that were created by this user (set created_by to None) + created_users = User.query.filter_by(created_by=user_id).all() + for created_user in created_users: + created_user.created_by = None + + # Delete the user + deleted_users.append({ + 'name': user_to_delete.full_name, + 'username': user_to_delete.username, + 'qr_count': user_qr_count + }) + + db.session.delete(user_to_delete) + + except Exception as e: + print(f"Error processing user {user_id}: {e}") + errors.append(f"Error processing user ID {user_id}") + continue + + # Commit all changes if we have deletions + if deleted_users: + db.session.commit() + + # Log the bulk deletion + deleted_names = [user['name'] for user in deleted_users] + print(f"Admin {current_user.username} permanently deleted {len(deleted_users)} users: {', '.join(deleted_names)}") + + # Prepare response message + if deleted_users and not errors: + message = f'Successfully deleted {len(deleted_users)} users and {deleted_qr_count} associated QR codes' + elif deleted_users and errors: + message = f'Deleted {len(deleted_users)} users and {deleted_qr_count} QR codes. {len(errors)} operations failed' + elif not deleted_users and errors: + return jsonify({ + 'success': False, + 'error': 'No users could be deleted', + 'details': errors + }), 400 + else: + return jsonify({ + 'success': False, + 'error': 'No valid users to delete' + }), 400 + + return jsonify({ + 'success': True, + 'message': message, + 'deleted_count': len(deleted_users), + 'deleted_qr_count': deleted_qr_count, + 'errors': errors if errors else None + }) + + except Exception as e: + db.session.rollback() + print(f"Error in bulk permanently delete users: {e}") + return jsonify({ + 'success': False, + 'error': 'Failed to delete users. Please try again.' + }), 500 + +# ENHANCED LOGIN WITH BETTER SESSION MANAGEMENT +@app.route('/login', methods=['GET', 'POST']) +def login(): + """Enhanced user authentication with better error handling""" + if request.method == 'POST': + username = request.form.get('username', '').strip() + password = request.form.get('password', '') + + if not username or not password: + flash('Please enter both username and password.', 'error') + return render_template('login.html') + + try: + # Find user (case-insensitive username) + user = User.query.filter( + User.username.ilike(username), + User.active_status == True + ).first() + + if user and user.check_password(password): + # Successful login + session['user_id'] = user.id + session['username'] = user.username + session['role'] = user.role + session['full_name'] = user.full_name + + # Update last login date + user.last_login_date = datetime.utcnow() + db.session.commit() + + flash(f'Welcome back, {user.full_name}!', 'success') + print(f"User {user.username} logged in successfully") + + # Redirect to intended page or dashboard + next_page = request.args.get('next') + return redirect(next_page) if next_page else redirect(url_for('dashboard')) + + else: + # Invalid credentials + flash('Invalid username or password.', 'error') + print(f"Failed login attempt for username: {username}") + + except Exception as e: + print(f"Login error: {e}") + flash('Login error. Please try again.', 'error') + + return render_template('login.html') + +# Add this helper function to check admin requirements more safely +def is_admin_user(user_id): + """Helper function to safely check if user is admin""" + try: + user = User.Query.get(user_id) + return user and user.active_status and user.role == 'admin' + except: + return False + +# QR code management routes +@app.route('/qr-codes/create', methods=['GET', 'POST']) +@login_required +def create_qr_code(): + """Enhanced create QR code with address coordinates""" + if request.method == 'POST': + name = request.form['name'] + location = request.form['location'] + location_address = request.form['location_address'] + location_event = request.form['location_event'] + + # Get coordinates from hidden form fields (set by JavaScript) + address_latitude = request.form.get('address_latitude') + address_longitude = request.form.get('address_longitude') + coordinate_accuracy = request.form.get('coordinate_accuracy', 'geocoded') + + # Create QR code record first (without QR image and URL) + new_qr_code = QRCode( + name=name, + location=location, + location_address=location_address, + location_event=location_event, + qr_code_image='', # Temporary empty value + qr_url='', # Temporary empty value + created_by=session['user_id'] + ) + + # Add coordinates if available + if address_latitude and address_longitude: + try: + lat = float(address_latitude) + lng = float(address_longitude) + new_qr_code.address_latitude = lat + new_qr_code.address_longitude = lng + new_qr_code.coordinate_accuracy = coordinate_accuracy + new_qr_code.coordinates_updated_date = datetime.utcnow() + print(f"✅ Added coordinates to QR code: {lat:.10f}, {lng:.10f}") + except (ValueError, TypeError) as e: + print(f"⚠️ Invalid coordinates provided: {e}") + + # Add to session and flush to get the ID + db.session.add(new_qr_code) + db.session.flush() # This assigns the ID without committing + + # Now we can use the ID to generate the URL + qr_url = generate_qr_url(name, new_qr_code.id) + + # Generate QR code data with the destination URL + qr_data = f"{request.url_root}qr/{qr_url}" + qr_image = generate_qr_code(qr_data) + + # Update the QR code with the URL and image + new_qr_code.qr_url = qr_url + new_qr_code.qr_code_image = qr_image + + # Now commit all changes + db.session.commit() + + coord_msg = "" + if new_qr_code.has_coordinates: + coord_msg = f" with coordinates ({new_qr_code.coordinates_display})" + + flash(f'QR code created successfully{coord_msg}!', 'success') + return redirect(url_for('dashboard')) + + return render_template('create_qr_code.html') + +@app.route('/qr-codes//edit', methods=['GET', 'POST']) +@login_required +def edit_qr_code(qr_id): + """Enhanced edit QR code with address coordinates""" + qr_code = QRCode.query.get_or_404(qr_id) + + if request.method == 'POST': + # Store original values for comparison + original_name = qr_code.name + original_address = qr_code.location_address + + # Update QR code fields + new_name = request.form['name'] + new_address = request.form['location_address'] + + qr_code.name = new_name + qr_code.location = request.form['location'] + qr_code.location_address = new_address + qr_code.location_event = request.form['location_event'] + + # Handle address coordinates + address_latitude = request.form.get('address_latitude') + address_longitude = request.form.get('address_longitude') + coordinate_accuracy = request.form.get('coordinate_accuracy', 'geocoded') + + # Update coordinates if provided + if address_latitude and address_longitude: + try: + lat = float(address_latitude) + lng = float(address_longitude) + qr_code.address_latitude = lat + qr_code.address_longitude = lng + qr_code.coordinate_accuracy = coordinate_accuracy + qr_code.coordinates_updated_date = datetime.utcnow() + print(f"✅ Updated coordinates for QR code: {lat:.10f}, {lng:.10f}") + except (ValueError, TypeError) as e: + print(f"⚠️ Invalid coordinates provided during edit: {e}") + + # Check if name changed and handle URL regeneration + if original_name != new_name: + # Name changed, regenerate URL + new_qr_url = generate_qr_url(new_name, qr_code.id) + qr_code.qr_url = new_qr_url + + # Update QR code data with new URL + qr_data = f"{request.url_root}qr/{new_qr_url}" + else: + # Name didn't change, use existing URL (if it exists) + if qr_code.qr_url: + qr_data = f"{request.url_root}qr/{qr_code.qr_url}" + else: + # Fallback: generate URL if it doesn't exist (for legacy QR codes) + new_qr_url = generate_qr_url(new_name, qr_code.id) + qr_code.qr_url = new_qr_url + qr_data = f"{request.url_root}qr/{new_qr_url}" + + # Regenerate QR code with updated data (destination URL) + qr_code.qr_code_image = generate_qr_code(qr_data) + + db.session.commit() + + coord_msg = "" + if qr_code.has_coordinates: + coord_msg = f" Coordinates: ({qr_code.coordinates_display})" + + flash(f'QR code updated successfully!{coord_msg}', 'success') + return redirect(url_for('dashboard')) + + return render_template('edit_qr_code.html', qr_code=qr_code) + +@app.route('/qr-codes//delete', methods=['GET', 'POST']) +@admin_required +def delete_qr_code(qr_id): + """Permanently delete QR code (Admin only) - Hard delete""" + + # OBVIOUS DEBUGGING - You MUST see this in console + print("\n" + "="*60) + print("🔥 DELETE ROUTE WAS CALLED! 🔥") + print(f"🔥 QR ID: {qr_id}") + print(f"🔥 Method: {request.method}") + print(f"🔥 User: {session.get('username', 'NO_USER')}") + print(f"🔥 Role: {session.get('role', 'NO_ROLE')}") + print("="*60 + "\n") + + try: + qr_code = QRCode.query.get_or_404(qr_id) + print(f"✅ Found QR Code: {qr_code.name}") + + if request.method == 'POST': + qr_name = qr_code.name + print(f"🗑️ ATTEMPTING TO DELETE: {qr_name}") + + # Check if QR exists before delete + before_count = QRCode.query.count() + print(f"📊 QR count before delete: {before_count}") + + # Delete the QR code + db.session.delete(qr_code) + print("💾 Called db.session.delete()") + + db.session.commit() + print("💾 Called db.session.commit()") + + # Check count after delete + after_count = QRCode.query.count() + print(f"📊 QR count after delete: {after_count}") + print(f"✅ DELETE SUCCESS! Removed {before_count - after_count} records") + + flash(f'QR code "{qr_name}" has been permanently deleted!', 'success') + return redirect(url_for('dashboard')) + + # GET request - show confirmation page + print("📄 Showing confirmation page") + return render_template('confirm_delete_qr.html', qr_code=qr_code) + + except Exception as e: + db.session.rollback() + print(f"❌ ERROR in delete route: {e}") + print(f"❌ Exception type: {type(e)}") + import traceback + print(f"❌ Traceback: {traceback.format_exc()}") + flash('Error deleting QR code. Please try again.', 'error') + return redirect(url_for('dashboard')) + +@app.route('/qr/') +def qr_destination(qr_url): + """QR code destination page where staff check in""" + try: + # Find QR code by URL + qr_code = QRCode.query.filter_by(qr_url=qr_url, active_status=True).first() + + if not qr_code: + flash('QR code not found or inactive.', 'error') + return render_template('qr_not_found.html'), 404 + + # Log the scan + print(f"QR Code scanned: {qr_code.name} at {datetime.now()}") + + return render_template('qr_destination.html', qr_code=qr_code) + + except Exception as e: + print(f"Error loading QR destination: {e}") + flash('Error loading QR code destination.', 'error') + return render_template('qr_not_found.html'), 500 + +@app.route('/qr//checkin', methods=['POST']) +def qr_checkin(qr_url): + """ + Enhanced staff check-in with location accuracy calculation + Allows multiple check-ins with minimum interval between them + PRESERVES coordinate-to-address conversion functionality + """ + try: + print(f"\n🚀 STARTING ENHANCED CHECK-IN PROCESS") + print(f" QR URL: {qr_url}") + print(f" Timestamp: {datetime.now()}") + + # Find QR code by URL + qr_code = QRCode.query.filter_by(qr_url=qr_url, active_status=True).first() + + if not qr_code: + print(f"❌ QR code not found or inactive: {qr_url}") + return jsonify({ + 'success': False, + 'message': 'QR code not found or inactive.' + }), 404 + + print(f"✅ Found QR code: {qr_code.name} (ID: {qr_code.id})") + print(f" Location: {qr_code.location}") + print(f" QR Address: {qr_code.location_address}") + + # Get and validate employee ID + employee_id = request.form.get('employee_id', '').strip() + + if not employee_id: + return jsonify({ + 'success': False, + 'message': 'Employee ID is required.' + }), 400 + + # Check for recent check-ins with 30-minute interval validation + today = date.today() + current_time = datetime.now() + time_interval = int(os.environ.get('TIME_INTERVAL')) + the_last_checkin_time = current_time - timedelta(minutes=time_interval) + + # Find the most recent check-in for this employee at this location today + recent_checkin = AttendanceData.query.filter_by( + qr_code_id=qr_code.id, + employee_id=employee_id.upper(), + check_in_date=today + ).order_by(AttendanceData.check_in_time.desc()).first() + + if recent_checkin: + # Convert check_in_time (time) to datetime for comparison + recent_checkin_datetime = datetime.combine(today, recent_checkin.check_in_time) + + # Check if 30 minutes have passed since the last check-in + if recent_checkin_datetime > the_last_checkin_time: + minutes_remaining = time_interval - int((current_time - recent_checkin_datetime).total_seconds() / 60) + print(f"⚠️ Too soon for another check-in for {employee_id}") + print(f" Last check-in: {recent_checkin.check_in_time.strftime('%H:%M')}") + print(f" Minutes remaining: {minutes_remaining}") + + return jsonify({ + 'success': False, + 'message': f'You can check in again in {minutes_remaining} minutes. Last check-in was at {recent_checkin.check_in_time.strftime("%H:%M")}.' + }), 400 + else: + print(f"✅ 30-minute interval satisfied. Allowing new check-in for {employee_id}") + else: + print(f"✅ First check-in today for {employee_id}") + + # PRESERVED: Process location data with coordinate-to-address conversion + location_data = process_location_data_enhanced(request.form) + + # PRESERVED: Get device and network info + user_agent_string = request.headers.get('User-Agent', '') + device_info = detect_device_info(user_agent_string) + client_ip = get_client_ip() + + print(f"📱 Device Info: {device_info}") + print(f"🌐 IP Address: {client_ip}") + print(f"📍 Location Data: {location_data}") + + # PRESERVED: Create attendance record + print(f"\n💾 CREATING ATTENDANCE RECORD:") + + attendance = AttendanceData( + qr_code_id=qr_code.id, + employee_id=employee_id.upper(), + check_in_date=today, + check_in_time=datetime.now().time(), + device_info=device_info, + user_agent=user_agent_string, + ip_address=client_ip, + location_name=qr_code.location, + latitude=location_data['latitude'], + longitude=location_data['longitude'], + accuracy=location_data['accuracy'], + altitude=location_data['altitude'], + location_source=location_data['source'], + address=location_data['address'], # This now includes converted address + status='present' + ) + + print(f"✅ Created base attendance record") + + # PRESERVED: Calculate location accuracy + print(f"\n🎯 CALCULATING LOCATION ACCURACY...") + location_accuracy = None + + try: + location_accuracy = calculate_location_accuracy_enhanced( + qr_address=qr_code.location_address, + checkin_address=location_data['address'], + checkin_lat=location_data['latitude'], + checkin_lng=location_data['longitude'] + ) + + if location_accuracy is not None: + attendance.location_accuracy = location_accuracy + accuracy_level = get_location_accuracy_level_enhanced(location_accuracy) + print(f"✅ Location accuracy set: {location_accuracy:.4f} miles ({accuracy_level})") + else: + print(f"⚠️ Could not calculate location accuracy") + + except Exception as e: + print(f"❌ Error in location accuracy calculation: {e}") + + # PRESERVED: Save to database + try: + db.session.add(attendance) + db.session.commit() + print(f"✅ Successfully saved attendance record with ID: {attendance.id}") + + # NEW: Count total check-ins for today for this employee at this location + today_checkin_count = AttendanceData.query.filter_by( + qr_code_id=qr_code.id, + employee_id=employee_id.upper(), + check_in_date=today + ).count() + + checkin_sequence_text = f"{qr_code.location_event} details" + + except Exception as e: + print(f"❌ Database error: {e}") + db.session.rollback() + return jsonify({ + 'success': False, + 'message': 'Database error occurred.' + }), 500 + + # ENHANCED: Return success response with sequence information + response_data = { + 'success': True, + 'message': f'Check-in successful! {checkin_sequence_text} for today.', + 'data': { + 'employee_id': attendance.employee_id, + 'location': attendance.location_name, + 'location_event': qr_code.location_event, + 'check_in_time': attendance.check_in_time.strftime('%H:%M:%S'), + 'check_in_date': attendance.check_in_date.strftime('%m/%d/%Y'), + 'device_info': attendance.device_info, + 'ip_address': attendance.ip_address, + 'location_accuracy': location_accuracy, + 'checkin_count_today': today_checkin_count, + 'checkin_sequence': checkin_sequence_text + } + } + + if location_data['address']: + response_data['data']['address'] = location_data['address'] + + if location_data['latitude'] and location_data['longitude']: + response_data['data']['coordinates'] = f"{location_data['latitude']:.10f}, {location_data['longitude']:.10f}" + + print(f"✅ Check-in completed successfully") + print(f" Employee: {attendance.employee_id}") + print(f" Time: {attendance.check_in_time}") + print(f" Location: {attendance.location_name}") + print(f" Address: {attendance.address}") + print(f" Today's count: {today_checkin_count}") + + return jsonify(response_data), 200 + + except Exception as e: + print(f"❌ Unexpected error in check-in process: {e}") + import traceback + print(f"❌ Traceback: {traceback.format_exc()}") + + return jsonify({ + 'success': False, + 'message': 'An unexpected error occurred during check-in.' + }), 500 + +@app.route('/qr-codes//toggle-status', methods=['POST']) +@login_required +def toggle_qr_status(qr_id): + """Toggle QR code active/inactive status""" + try: + qr_code = QRCode.query.get_or_404(qr_id) + + # Toggle the status + qr_code.active_status = not qr_code.active_status + db.session.commit() + + status_text = "activated" if qr_code.active_status else "deactivated" + flash(f'QR code "{qr_code.name}" has been {status_text} successfully!', 'success') + + return jsonify({ + 'success': True, + 'new_status': qr_code.active_status, + 'status_text': 'Active' if qr_code.active_status else 'Inactive', + 'message': f'QR code {status_text} successfully!' + }) + + except Exception as e: + db.session.rollback() + print(f"Error toggling QR status: {e}") + return jsonify({ + 'success': False, + 'message': 'Error updating QR code status. Please try again.' + }), 500 + +@app.route('/qr-codes//activate', methods=['POST']) +@login_required +def activate_qr_code(qr_id): + """Activate a QR code""" + try: + qr_code = QRCode.query.get_or_404(qr_id) + qr_code.active_status = True + db.session.commit() + + flash(f'QR code "{qr_code.name}" has been activated successfully!', 'success') + return jsonify({ + 'success': True, + 'new_status': True, + 'status_text': 'Active', + 'message': 'QR code activated successfully!' + }) + + except Exception as e: + db.session.rollback() + print(f"Error activating QR code: {e}") + return jsonify({ + 'success': False, + 'message': 'Error activating QR code. Please try again.' + }), 500 + +@app.route('/qr-codes//deactivate', methods=['POST']) +@login_required +def deactivate_qr_code(qr_id): + """Deactivate a QR code""" + try: + qr_code = QRCode.query.get_or_404(qr_id) + qr_code.active_status = False + db.session.commit() + + flash(f'QR code "{qr_code.name}" has been deactivated successfully!', 'success') + return jsonify({ + 'success': True, + 'new_status': False, + 'status_text': 'Inactive', + 'message': 'QR code deactivated successfully!' + }) + + except Exception as e: + db.session.rollback() + print(f"Error deactivating QR code: {e}") + return jsonify({ + 'success': False, + 'message': 'Error deactivating QR code. Please try again.' + }), 500 + +@app.route('/qr-codes//toggle-status', methods=['POST']) +@admin_required +def toggle_qr_status_api(qr_id): + """Toggle QR code active/inactive status - Enhanced JSON API""" + try: + qr_code = QRCode.query.get_or_404(qr_id) + qr_code.active_status = not qr_code.active_status + db.session.commit() + + status_text = "activated" if qr_code.active_status else "deactivated" + + if request.headers.get('X-Requested-With') == 'XMLHttpRequest': + return jsonify({ + 'success': True, + 'new_status': qr_code.active_status, + 'status_text': 'Active' if qr_code.active_status else 'Inactive', + 'message': f'QR code "{qr_code.name}" has been {status_text} successfully!' + }) + else: + flash(f'QR code "{qr_code.name}" has been {status_text} successfully!', 'success') + return redirect(url_for('dashboard')) + + except Exception as e: + db.session.rollback() + + if request.headers.get('X-Requested-With') == 'XMLHttpRequest': + return jsonify({ + 'success': False, + 'message': 'Error updating QR code status. Please try again.' + }), 500 + else: + flash('Error updating QR code status. Please try again.', 'error') + return redirect(url_for('dashboard')) + +@app.route('/attendance') +# @admin_required +def attendance_report(): + """Safe attendance report with backward compatibility for location_accuracy""" + try: + print("📊 Loading attendance report...") + + # Check if location_accuracy column exists + has_location_accuracy = check_location_accuracy_column_exists() + print(f"🔍 Location accuracy column exists: {has_location_accuracy}") + + # Get filter parameters + date_from = request.args.get('date_from', '') + date_to = request.args.get('date_to', '') + location_filter = request.args.get('location', '') + employee_filter = request.args.get('employee', '') + + # Build base query - conditional based on column existence + if has_location_accuracy: + # New query with location accuracy + base_query = """ + SELECT + ad.id, + ad.employee_id, + ad.check_in_date, + ad.check_in_time, + ad.location_name, + qc.location_event, + qc.location_address as qr_address, + ad.address as checked_in_address, + ad.latitude, + ad.longitude, + ad.location_accuracy, + ad.accuracy as gps_accuracy, + ad.device_info + FROM attendance_data ad + LEFT JOIN qr_codes qc ON ad.qr_code_id = qc.id + WHERE 1=1 + """ + else: + # Fallback query without location accuracy + base_query = """ + SELECT + ad.id, + ad.employee_id, + ad.check_in_date, + ad.check_in_time, + ad.location_name, + qc.location_event, + qc.location_address as qr_address, + ad.address as checked_in_address, + ad.latitude, + ad.longitude, + NULL as location_accuracy, + ad.accuracy as gps_accuracy, + ad.device_info + FROM attendance_data ad + LEFT JOIN qr_codes qc ON ad.qr_code_id = qc.id + WHERE 1=1 + """ + + conditions = [] + params = {} + + # Apply date range filter + if date_from: + conditions.append("ad.check_in_date >= :date_from") + params['date_from'] = date_from + + if date_to: + conditions.append("ad.check_in_date <= :date_to") + params['date_to'] = date_to + + # Apply location filter + if location_filter: + conditions.append("ad.location_name ILIKE :location") + params['location'] = f"%{location_filter}%" + + # Apply employee filter + if employee_filter: + conditions.append("ad.employee_id ILIKE :employee") + params['employee'] = f"%{employee_filter}%" + + # Add conditions to query + if conditions: + base_query += " AND " + " AND ".join(conditions) + + # Add ordering + base_query += " ORDER BY ad.check_in_date DESC, ad.check_in_time DESC" + + print(f"🔍 Executing query with {len(params)} parameters") + + # Execute query + query_result = db.session.execute(text(base_query), params) + attendance_records = query_result.fetchall() + + print(f"✅ Found {len(attendance_records)} attendance records") + + # Process records to add calculated fields + processed_records = [] + for record in attendance_records: + # Safe attribute access with fallbacks + location_accuracy = getattr(record, 'location_accuracy', None) + gps_accuracy = getattr(record, 'gps_accuracy', None) + qr_address = getattr(record, 'qr_address', None) + if location_accuracy is not None and location_accuracy != "None": + try: + accuracy_value = float(location_accuracy) if isinstance(location_accuracy, str) else location_accuracy + checked_in_address = qr_address if (accuracy_value <= 0.5) else getattr(record, 'checked_in_address', None) + except (ValueError, TypeError): + checked_in_address = getattr(record, 'checked_in_address', None) + else: + checked_in_address = getattr(record, 'checked_in_address', None) + record_dict = { + 'id': record.id, + 'employee_id': record.employee_id, + 'check_in_date': record.check_in_date, + 'check_in_time': record.check_in_time, + 'location_name': record.location_name, + 'location_event': getattr(record, 'location_event', ''), + 'qr_address': qr_address or 'Not available', + 'checked_in_address': checked_in_address or 'Location not captured', + 'device_info': getattr(record, 'device_info', ''), + 'location_accuracy': location_accuracy, + 'gps_accuracy': gps_accuracy, + 'accuracy_level': get_location_accuracy_level(location_accuracy) if location_accuracy else 'unknown', + 'has_location_data': record.latitude is not None and record.longitude is not None, + 'coordinates': f"{record.latitude:.10f}, {record.longitude:.10f}" if record.latitude and record.longitude else "No GPS data", + 'has_location_accuracy_feature': has_location_accuracy + } + processed_records.append(record_dict) + + print(f"✅ Processed {len(processed_records)} records") + + # Get unique locations for filter dropdown + try: + locations_query = db.session.execute(text(""" + SELECT DISTINCT location_name + FROM attendance_data + WHERE location_name IS NOT NULL + ORDER BY location_name + """)) + locations = [row[0] for row in locations_query.fetchall()] + print(f"✅ Found {len(locations)} unique locations") + except Exception as e: + print(f"⚠️ Error loading locations: {e}") + locations = [] + + # Get attendance statistics + try: + if has_location_accuracy: + stats_query = db.session.execute(text(""" + SELECT + COUNT(*) as total_checkins, + COUNT(DISTINCT employee_id) as unique_employees, + COUNT(DISTINCT qr_code_id) as active_locations, + COUNT(CASE WHEN check_in_date = CURRENT_DATE THEN 1 END) as today_checkins, + COUNT(CASE WHEN latitude IS NOT NULL AND longitude IS NOT NULL THEN 1 END) as records_with_gps, + COUNT(CASE WHEN location_accuracy IS NOT NULL THEN 1 END) as records_with_accuracy, + AVG(location_accuracy) as avg_location_accuracy + FROM attendance_data + """)) + else: + stats_query = db.session.execute(text(""" + SELECT + COUNT(*) as total_checkins, + COUNT(DISTINCT employee_id) as unique_employees, + COUNT(DISTINCT qr_code_id) as active_locations, + COUNT(CASE WHEN check_in_date = CURRENT_DATE THEN 1 END) as today_checkins, + COUNT(CASE WHEN latitude IS NOT NULL AND longitude IS NOT NULL THEN 1 END) as records_with_gps, + 0 as records_with_accuracy, + 0 as avg_location_accuracy + FROM attendance_data + """)) + + stats = stats_query.fetchone() + print(f"✅ Loaded statistics: {stats.total_checkins} total check-ins") + except Exception as e: + print(f"⚠️ Error loading statistics: {e}") + # Fallback stats + stats = type('Stats', (), { + 'total_checkins': 0, + 'unique_employees': 0, + 'active_locations': 0, + 'today_checkins': 0, + 'records_with_gps': 0, + 'records_with_accuracy': 0, + 'avg_location_accuracy': 0 + })() + + # Add today's date for template + today_date = datetime.now().strftime('%m/%d/%Y') + current_date_formatted = datetime.now().strftime('%B %d') + + print("✅ Rendering attendance report template") + + return render_template('attendance_report.html', + attendance_records=processed_records, + locations=locations, + stats=stats, + date_from=date_from, + date_to=date_to, + location_filter=location_filter, + employee_filter=employee_filter, + today_date=today_date, + current_date_formatted=current_date_formatted, + has_location_accuracy_feature=has_location_accuracy) + + except Exception as e: + print(f"❌ Error loading attendance report: {e}") + print(f"❌ Exception type: {type(e)}") + import traceback + print(f"❌ Traceback: {traceback.format_exc()}") + + flash('Error loading attendance report. Please check the server logs for details.', 'error') + return redirect(url_for('dashboard')) + +@app.route('/api/attendance/stats') +@admin_required +def attendance_stats_api(): + """API endpoint for attendance statistics""" + try: + # Daily stats for the last 7 days + daily_stats = db.session.execute(text(""" + SELECT + check_in_date, + COUNT(*) as checkins, + COUNT(DISTINCT employee_id) as unique_employees + FROM attendance_data + WHERE check_in_date >= CURRENT_DATE - INTERVAL '7 days' + GROUP BY check_in_date + ORDER BY check_in_date DESC + """)).fetchall() + + # Location stats + location_stats = db.session.execute(text(""" + SELECT + location_name, + COUNT(*) as total_checkins, + COUNT(DISTINCT employee_id) as unique_employees + FROM attendance_data + GROUP BY location_name + ORDER BY total_checkins DESC + LIMIT 10 + """)).fetchall() + + # Peak hours + hourly_stats = db.session.execute(text(""" + SELECT + EXTRACT(hour FROM check_in_time) as hour, + COUNT(*) as checkins + FROM attendance_data + WHERE check_in_date >= CURRENT_DATE - INTERVAL '30 days' + GROUP BY EXTRACT(hour FROM check_in_time) + ORDER BY hour + """)).fetchall() + + return jsonify({ + 'daily_stats': [{'date': str(row[0]), 'checkins': row[1], 'employees': row[2]} for row in daily_stats], + 'location_stats': [{'location': row[0], 'checkins': row[1], 'employees': row[2]} for row in location_stats], + 'hourly_stats': [{'hour': int(row[0]), 'checkins': row[1]} for row in hourly_stats] + }) + + except Exception as e: + print(f"Error fetching attendance stats: {e}") + return jsonify({'error': 'Failed to fetch attendance statistics'}), 500 + +# Jinja2 filters for better template functionality +@app.template_filter('days_since') +def days_since_filter(date): + """Calculate days since a given date""" + if not date: + return 0 + from datetime import datetime + now = datetime.utcnow() + return (now - date).days + +@app.template_filter('time_ago') +def time_ago_filter(date): + """Human readable time ago""" + if not date: + return 'Never' + from datetime import datetime + now = datetime.utcnow() + diff = now - date + + if diff.days > 365: + years = diff.days // 365 + return f"{years} year{'s' if years != 1 else ''} ago" + elif diff.days > 30: + months = diff.days // 30 + return f"{months} month{'s' if months != 1 else ''} ago" + elif diff.days > 0: + return f"{diff.days} day{'s' if diff.days != 1 else ''} ago" + elif diff.seconds > 3600: + hours = diff.seconds // 3600 + return f"{hours} hour{'s' if hours != 1 else ''} ago" + elif diff.seconds > 60: + minutes = diff.seconds // 60 + return f"{minutes} minute{'s' if minutes != 1 else ''} ago" + else: + return "Just now" + +# Error handlers +@app.errorhandler(500) +def internal_error(error): + """Handle internal server errors with user-friendly page""" + if app.debug: + # Let Flask handle debug errors naturally + return None + + return ''' + + + Server Error + +

🔧 Something went wrong

+

We're working to fix this issue. Please try again later.

+ ← Back to Home + + + ''', 500 + +@app.errorhandler(404) +def not_found(error): + """Handle page not found errors""" + return ''' + + + Page Not Found + +

🔍 Page Not Found

+

The page you're looking for doesn't exist.

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