Sep 11 - Reupload the code
This commit is contained in:
@@ -0,0 +1,824 @@
|
||||
"""
|
||||
utils/geocoding.py
|
||||
==================
|
||||
All geocoding, distance calculation, and location-accuracy helpers.
|
||||
|
||||
Extracted verbatim from app.py (lines 188–1317).
|
||||
No logic changes — only import paths updated.
|
||||
"""
|
||||
|
||||
import os
|
||||
import re
|
||||
import requests
|
||||
import traceback
|
||||
from datetime import datetime, timedelta
|
||||
from math import radians, sin, cos, asin, sqrt
|
||||
|
||||
import googlemaps
|
||||
|
||||
from extensions import db, logger_handler
|
||||
from address_normalization_fix import normalize_address, addresses_are_similar
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Google Maps client (initialized once at module import)
|
||||
# ---------------------------------------------------------------------------
|
||||
try:
|
||||
GOOGLE_MAPS_API_KEY = os.environ.get('GOOGLE_MAPS_API_KEY')
|
||||
if GOOGLE_MAPS_API_KEY:
|
||||
gmaps_client = googlemaps.Client(key=GOOGLE_MAPS_API_KEY)
|
||||
print("✅ Google Maps client initialized successfully")
|
||||
else:
|
||||
gmaps_client = None
|
||||
print("⚠️ Google Maps API key not found, falling back to OpenStreetMap")
|
||||
except Exception as e:
|
||||
gmaps_client = None
|
||||
print(f"❌ Error initializing Google Maps client: {e}")
|
||||
|
||||
|
||||
def is_gmaps_available():
|
||||
"""Return True if the Google Maps client is initialized and usable.
|
||||
|
||||
Always call this (or check ``if gmaps_client:``) before calling any
|
||||
method on ``gmaps_client`` to prevent AttributeError when the API key
|
||||
is absent.
|
||||
"""
|
||||
return gmaps_client is not None
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Geocoding cache
|
||||
# ---------------------------------------------------------------------------
|
||||
geocoding_cache = {}
|
||||
CACHE_MAX_SIZE = 1000
|
||||
CACHE_EXPIRY_HOURS = 24
|
||||
|
||||
|
||||
def get_cached_coordinates(address):
|
||||
"""Get coordinates from cache if available and not expired"""
|
||||
if address in geocoding_cache:
|
||||
cached_data = geocoding_cache[address]
|
||||
cache_time = cached_data.get('timestamp', datetime.min)
|
||||
if datetime.now() - cache_time < timedelta(hours=CACHE_EXPIRY_HOURS):
|
||||
print(f"📋 Using cached coordinates for: {address[:50]}...")
|
||||
return cached_data.get('lat'), cached_data.get('lng'), cached_data.get('accuracy')
|
||||
return None, None, None
|
||||
|
||||
|
||||
def cache_coordinates(address, lat, lng, accuracy):
|
||||
"""Cache coordinates to reduce future API calls"""
|
||||
try:
|
||||
if len(geocoding_cache) >= CACHE_MAX_SIZE:
|
||||
oldest_key = min(geocoding_cache.keys(), key=lambda k: geocoding_cache[k]['timestamp'])
|
||||
del geocoding_cache[oldest_key]
|
||||
geocoding_cache[address] = {
|
||||
'lat': lat,
|
||||
'lng': lng,
|
||||
'accuracy': accuracy,
|
||||
'timestamp': datetime.now()
|
||||
}
|
||||
print(f"💾 Cached coordinates for: {address[:50]}...")
|
||||
except Exception as e:
|
||||
print(f"⚠️ Error caching coordinates: {e}")
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Geocoding helpers
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
def log_google_maps_usage(operation_type):
|
||||
"""Log Google Maps API usage for monitoring"""
|
||||
try:
|
||||
logger_handler.log_user_activity('google_maps_api_usage', f'Google Maps API used: {operation_type}')
|
||||
except Exception as e:
|
||||
print(f"⚠️ Usage logging error: {e}")
|
||||
|
||||
|
||||
def get_coordinates_from_address(address):
|
||||
"""
|
||||
Get latitude and longitude from address using Google Maps Geocoding API.
|
||||
Falls back to OpenStreetMap if Google Maps is unavailable.
|
||||
Returns (lat, lng) tuple or (None, None) if failed.
|
||||
"""
|
||||
if not address or address.strip() == '':
|
||||
return None, None
|
||||
|
||||
address = address.strip()
|
||||
print(f"🌍 Geocoding address: {address}")
|
||||
|
||||
try:
|
||||
logger_handler.log_user_activity('geocoding', f'Geocoding address: {address[:50]}...')
|
||||
except Exception as log_error:
|
||||
print(f"⚠️ Logging error (non-critical): {log_error}")
|
||||
|
||||
try:
|
||||
if gmaps_client:
|
||||
print("🗺️ Using Google Maps Geocoding API")
|
||||
geocode_result = gmaps_client.geocode(address)
|
||||
if geocode_result:
|
||||
location = geocode_result[0]['geometry']['location']
|
||||
lat = location['lat']
|
||||
lng = location['lng']
|
||||
print(f"✅ Google Maps geocoded address '{address[:50]}...' to coordinates: {lat}, {lng}")
|
||||
try:
|
||||
logger_handler.log_user_activity('geocoding_success', f'Successfully geocoded: {address[:50]}... -> {lat}, {lng}')
|
||||
except Exception as log_error:
|
||||
print(f"⚠️ Logging error (non-critical): {log_error}")
|
||||
return lat, lng
|
||||
else:
|
||||
print(f"⚠️ Google Maps: No results found for address: {address}")
|
||||
|
||||
print("🌐 Falling back to OpenStreetMap Nominatim")
|
||||
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"✅ OSM geocoded address '{address[:50]}...' to coordinates: {lat}, {lng}")
|
||||
try:
|
||||
logger_handler.log_user_activity('geocoding_fallback', f'OSM fallback geocoded: {address[:50]}... -> {lat}, {lng}')
|
||||
except Exception as log_error:
|
||||
print(f"⚠️ Logging error (non-critical): {log_error}")
|
||||
return lat, lng
|
||||
|
||||
print(f"⚠️ Could not geocode address: {address}")
|
||||
try:
|
||||
logger_handler.log_user_activity('geocoding_failed', f'Failed to geocode: {address[:50]}...')
|
||||
except Exception as log_error:
|
||||
print(f"⚠️ Logging error (non-critical): {log_error}")
|
||||
return None, None
|
||||
|
||||
except Exception as e:
|
||||
print(f"❌ Error geocoding address '{address}': {e}")
|
||||
try:
|
||||
logger_handler.log_flask_error('geocoding_error', f'Error geocoding {address[:50]}...: {str(e)}')
|
||||
except Exception as log_error:
|
||||
print(f"⚠️ Logging error (non-critical): {log_error}")
|
||||
return None, None
|
||||
|
||||
|
||||
def get_coordinates_from_address_enhanced(address):
|
||||
"""
|
||||
Enhanced geocoding function using Google Maps with caching and better error handling.
|
||||
Returns (latitude, longitude, accuracy_level).
|
||||
"""
|
||||
if not address or address.strip() == "":
|
||||
return None, None, None
|
||||
|
||||
address = address.strip()
|
||||
print(f"🌍 Enhanced geocoding for: {address}")
|
||||
|
||||
normalized_address = normalize_address(address)
|
||||
cached_lat, cached_lng, cached_accuracy = get_cached_coordinates(normalized_address)
|
||||
if cached_lat is not None:
|
||||
print("✅ Using cached coordinates for normalized address")
|
||||
return cached_lat, cached_lng, cached_accuracy
|
||||
|
||||
try:
|
||||
logger_handler.log_user_activity('enhanced_geocoding', f'Enhanced geocoding: {address[:50]}...')
|
||||
except Exception as log_error:
|
||||
print(f"⚠️ Logging error (non-critical): {log_error}")
|
||||
|
||||
try:
|
||||
if gmaps_client:
|
||||
print("🗺️ Using Google Maps Geocoding API (Enhanced)")
|
||||
geocode_result = gmaps_client.geocode(address)
|
||||
if geocode_result:
|
||||
result = geocode_result[0]
|
||||
location = result['geometry']['location']
|
||||
lat = location['lat']
|
||||
lng = location['lng']
|
||||
location_type = result['geometry'].get('location_type', 'UNKNOWN')
|
||||
place_types = result.get('types', [])
|
||||
|
||||
if location_type == 'ROOFTOP':
|
||||
accuracy = 'excellent'
|
||||
elif location_type == 'RANGE_INTERPOLATED':
|
||||
accuracy = 'good'
|
||||
elif location_type == 'GEOMETRIC_CENTER':
|
||||
if any(ptype in place_types for ptype in ['premise', 'subpremise', 'street_address']):
|
||||
accuracy = 'good'
|
||||
elif any(ptype in place_types for ptype in ['neighborhood', 'sublocality']):
|
||||
accuracy = 'fair'
|
||||
else:
|
||||
accuracy = 'poor'
|
||||
elif location_type == 'APPROXIMATE':
|
||||
accuracy = 'poor'
|
||||
else:
|
||||
accuracy = 'fair'
|
||||
|
||||
print(f"✅ Google Maps enhanced geocoding successful:")
|
||||
print(f" Coordinates: {lat:.10f}, {lng:.10f}")
|
||||
print(f" Accuracy: {accuracy} (location_type: {location_type})")
|
||||
print(f" Place types: {place_types[:3]}")
|
||||
|
||||
cache_coordinates(normalized_address, lat, lng, accuracy)
|
||||
try:
|
||||
logger_handler.log_user_activity('enhanced_geocoding_success', f'Google Maps enhanced: {address[:50]}... -> {lat}, {lng} ({accuracy})')
|
||||
except Exception as log_error:
|
||||
print(f"⚠️ Logging error (non-critical): {log_error}")
|
||||
return lat, lng, accuracy
|
||||
else:
|
||||
print(f"⚠️ Google Maps: No results found for enhanced geocoding: {address}")
|
||||
|
||||
print("🌐 Falling back to OpenStreetMap Nominatim (Enhanced)")
|
||||
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'])
|
||||
place_type = result.get('type', 'unknown')
|
||||
osm_type = result.get('osm_type', 'unknown')
|
||||
|
||||
if place_type in ['house', 'building', 'shop', 'office'] or osm_type == 'way':
|
||||
accuracy = 'good'
|
||||
elif place_type in ['neighbourhood', 'suburb', 'quarter', 'residential']:
|
||||
accuracy = 'fair'
|
||||
elif place_type in ['city', 'town', 'village']:
|
||||
accuracy = 'poor'
|
||||
else:
|
||||
accuracy = 'poor'
|
||||
|
||||
print(f"✅ OSM enhanced geocoding successful:")
|
||||
print(f" Coordinates: {lat:.10f}, {lng:.10f}")
|
||||
print(f" Accuracy: {accuracy} (fallback)")
|
||||
cache_coordinates(normalized_address, lat, lng, accuracy)
|
||||
try:
|
||||
logger_handler.log_user_activity('enhanced_geocoding_fallback', f'OSM enhanced fallback: {address[:50]}... -> {lat}, {lng} ({accuracy})')
|
||||
except Exception as log_error:
|
||||
print(f"⚠️ Logging error (non-critical): {log_error}")
|
||||
return lat, lng, accuracy
|
||||
|
||||
print(f"⚠️ No results from enhanced geocoding for: {address}")
|
||||
try:
|
||||
logger_handler.log_user_activity('enhanced_geocoding_failed', f'Enhanced geocoding failed: {address[:50]}...')
|
||||
except Exception as log_error:
|
||||
print(f"⚠️ Logging error (non-critical): {log_error}")
|
||||
return None, None, None
|
||||
|
||||
except Exception as e:
|
||||
print(f"❌ Enhanced geocoding error: {e}")
|
||||
try:
|
||||
logger_handler.log_flask_error('enhanced_geocoding_error', f'Enhanced geocoding error {address[:50]}...: {str(e)}')
|
||||
except Exception as log_error:
|
||||
print(f"⚠️ Logging error (non-critical): {log_error}")
|
||||
return None, None, None
|
||||
|
||||
|
||||
def geocode_address_enhanced(address):
|
||||
"""
|
||||
Enhanced geocoding using Nominatim API with better accuracy classification.
|
||||
Returns: (latitude, longitude, accuracy_level)
|
||||
"""
|
||||
if not address or len(address.strip()) < 5:
|
||||
print("❌ Address too short for geocoding")
|
||||
return None, None, None
|
||||
|
||||
try:
|
||||
url = "https://nominatim.openstreetmap.org/search"
|
||||
params = {'q': address.strip(), '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'])
|
||||
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:
|
||||
logger_handler.log_flask_error('geocoding_error', str(e))
|
||||
print(f"❌ Geocoding error: {e}")
|
||||
return None, None, None
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Distance / accuracy
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
def calculate_distance_miles(lat1, lng1, lat2, lng2):
|
||||
"""
|
||||
Calculate DIRECT straight-line distance between two points using Haversine formula.
|
||||
Returns distance in miles (float) or None if calculation fails.
|
||||
"""
|
||||
if any(coord is None for coord in [lat1, lng1, lat2, lng2]):
|
||||
print("⚠️ Missing coordinates for distance calculation")
|
||||
return None
|
||||
|
||||
try:
|
||||
try:
|
||||
lat1_val = float(lat1)
|
||||
lng1_val = float(lng1)
|
||||
lat2_val = float(lat2)
|
||||
lng2_val = float(lng2)
|
||||
except (ValueError, TypeError) as e:
|
||||
print(f"⚠️ Invalid coordinate format: {e}")
|
||||
return None
|
||||
|
||||
if not (-90 <= lat1_val <= 90) or not (-90 <= lat2_val <= 90):
|
||||
print(f"⚠️ Invalid latitude values: {lat1_val}, {lat2_val}")
|
||||
return None
|
||||
if not (-180 <= lng1_val <= 180) or not (-180 <= lng2_val <= 180):
|
||||
print(f"⚠️ Invalid longitude values: {lng1_val}, {lng2_val}")
|
||||
return None
|
||||
|
||||
try:
|
||||
logger_handler.log_user_activity(
|
||||
'distance_calculation',
|
||||
f'Calculating direct distance: ({lat1_val:.6f}, {lng1_val:.6f}) to ({lat2_val:.6f}, {lng2_val:.6f})'
|
||||
)
|
||||
except Exception:
|
||||
pass
|
||||
|
||||
print("📐 Calculating direct straight-line distance using Haversine formula")
|
||||
|
||||
lat1_rad = radians(lat1_val)
|
||||
lng1_rad = radians(lng1_val)
|
||||
lat2_rad = radians(lat2_val)
|
||||
lng2_rad = radians(lng2_val)
|
||||
|
||||
dlat = lat2_rad - lat1_rad
|
||||
dlng = lng2_rad - lng1_rad
|
||||
|
||||
sin_dlat_half = sin(dlat / 2.0)
|
||||
sin_dlng_half = sin(dlng / 2.0)
|
||||
|
||||
a = (sin_dlat_half * sin_dlat_half +
|
||||
cos(lat1_rad) * cos(lat2_rad) * sin_dlng_half * sin_dlng_half)
|
||||
a = max(0.0, min(1.0, a))
|
||||
c = 2.0 * asin(sqrt(a))
|
||||
|
||||
# DO NOT CHANGE Earth's mean radius value
|
||||
EARTH_RADIUS_MILES = 3959.87433
|
||||
distance = round(c * EARTH_RADIUS_MILES, 4)
|
||||
|
||||
print(f"📏 Direct straight-line distance calculation:")
|
||||
print(f" Point 1: ({lat1_val:.10f}, {lng1_val:.10f})")
|
||||
print(f" Point 2: ({lat2_val:.10f}, {lng2_val:.10f})")
|
||||
print(f" Δlat: {abs(lat2_val - lat1_val):.10f}° = {dlat:.12f} radians")
|
||||
print(f" Δlng: {abs(lng2_val - lng1_val):.10f}° = {dlng:.12f} radians")
|
||||
print(f" a value: {a:.15f}")
|
||||
print(f" c value (central angle): {c:.15f} radians")
|
||||
print(f" 🎯 Distance: {distance:.4f} miles = {distance * 5280:.2f} feet = {distance * 1609.34:.2f} meters")
|
||||
|
||||
try:
|
||||
logger_handler.log_user_activity('distance_calculation_success', f'Direct distance: {distance:.4f} miles')
|
||||
except Exception:
|
||||
pass
|
||||
|
||||
return distance
|
||||
|
||||
except Exception as e:
|
||||
print(f"❌ Error in distance calculation: {e}")
|
||||
print(f" Traceback: {traceback.format_exc()}")
|
||||
try:
|
||||
logger_handler.log_flask_error('distance_calculation_error', f'Distance calculation error: {str(e)}')
|
||||
except Exception:
|
||||
pass
|
||||
return None
|
||||
|
||||
|
||||
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'
|
||||
if location_accuracy <= 0.05:
|
||||
return 'excellent'
|
||||
elif location_accuracy <= 0.1:
|
||||
return 'very_good'
|
||||
elif location_accuracy <= 0.25:
|
||||
return 'good'
|
||||
elif location_accuracy <= 0.5:
|
||||
return 'fair'
|
||||
elif location_accuracy <= 1.0:
|
||||
return 'poor'
|
||||
else:
|
||||
return 'very_poor'
|
||||
|
||||
|
||||
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}")
|
||||
|
||||
qr_lat, qr_lng = get_coordinates_from_address(qr_address)
|
||||
if qr_lat is None or qr_lng is None:
|
||||
print("⚠️ Could not geocode QR address, cannot calculate accuracy")
|
||||
return None
|
||||
|
||||
if checkin_lat is not None and checkin_lng is not None:
|
||||
checkin_coords_lat, checkin_coords_lng = checkin_lat, checkin_lng
|
||||
print("✅ 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("⚠️ Could not geocode check-in address, cannot calculate accuracy")
|
||||
return None
|
||||
print("✅ Using geocoded coordinates for check-in address")
|
||||
|
||||
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.
|
||||
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()}")
|
||||
|
||||
if not qr_address or qr_address.strip() == "":
|
||||
print("❌ QR address is empty or invalid")
|
||||
return None
|
||||
|
||||
print("\n📍 Step 1: Geocoding QR address...")
|
||||
try:
|
||||
if addresses_are_similar(qr_address, checkin_address, threshold=0.90):
|
||||
print("🎯 Addresses are essentially identical - returning near-zero distance")
|
||||
return 0.01
|
||||
|
||||
qr_lat, qr_lng, qr_accuracy = get_coordinates_from_address_enhanced(qr_address)
|
||||
print(f" Geocoding result: lat={qr_lat}, lng={qr_lng}, accuracy={qr_accuracy}")
|
||||
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})")
|
||||
except Exception as e:
|
||||
print(f"❌ Error geocoding QR address: {e}")
|
||||
return None
|
||||
|
||||
print("\n📱 Step 2: Determining check-in coordinates...")
|
||||
checkin_coords_lat = None
|
||||
checkin_coords_lng = None
|
||||
checkin_source = "unknown"
|
||||
|
||||
if checkin_lat is not None and checkin_lng is not None:
|
||||
try:
|
||||
lat_val = float(checkin_lat)
|
||||
lng_val = float(checkin_lng)
|
||||
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) as e:
|
||||
print(f"⚠️ Could not parse GPS coordinates: {e}")
|
||||
|
||||
if checkin_coords_lat is None and checkin_address:
|
||||
print("🌍 Falling back to geocoding check-in address...")
|
||||
try:
|
||||
checkin_coords_lat, checkin_coords_lng, checkin_accuracy = get_coordinates_from_address_enhanced(checkin_address)
|
||||
print(f" Checkin geocoding result: lat={checkin_coords_lat}, lng={checkin_coords_lng}, accuracy={checkin_accuracy}")
|
||||
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})")
|
||||
except Exception as e:
|
||||
print(f"❌ Error geocoding check-in address: {e}")
|
||||
|
||||
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
|
||||
|
||||
print("\n📏 Step 3: Calculating distance...")
|
||||
try:
|
||||
print(f" QR coordinates: {qr_lat:.10f}, {qr_lng:.10f}")
|
||||
print(f" Check-in coordinates: {checkin_coords_lat:.10f}, {checkin_coords_lng:.10f}")
|
||||
print(f" Source: {checkin_source}")
|
||||
|
||||
distance = calculate_distance_miles(qr_lat, qr_lng, checkin_coords_lat, checkin_coords_lng)
|
||||
print(f" Distance calculation result: {distance}")
|
||||
|
||||
if distance is not None:
|
||||
accuracy_level = get_location_accuracy_level_enhanced(distance)
|
||||
print(f"✅ Enhanced location accuracy calculated successfully!")
|
||||
print(f" Distance: {distance:.4f} miles")
|
||||
print(f" Accuracy Level: {accuracy_level}")
|
||||
return distance
|
||||
else:
|
||||
print("❌ Distance calculation returned None")
|
||||
return None
|
||||
except Exception as e:
|
||||
print(f"❌ Error calculating distance: {e}")
|
||||
print(f"❌ Distance calculation traceback: {traceback.format_exc()}")
|
||||
return None
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Reverse geocoding
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
def reverse_geocode_coordinates(latitude, longitude):
|
||||
"""
|
||||
Convert GPS coordinates to human-readable address.
|
||||
Falls back to OpenStreetMap if Google Maps is unavailable.
|
||||
Returns address string or None if failed.
|
||||
"""
|
||||
if not latitude or not longitude:
|
||||
return None
|
||||
|
||||
try:
|
||||
print(f"🌍 Reverse geocoding coordinates: {latitude}, {longitude}")
|
||||
try:
|
||||
logger_handler.log_user_activity('reverse_geocoding', f'Reverse geocoding: {latitude}, {longitude}')
|
||||
except Exception as log_error:
|
||||
print(f"⚠️ Logging error (non-critical): {log_error}")
|
||||
|
||||
if gmaps_client:
|
||||
print("🗺️ Using Google Maps Reverse Geocoding API")
|
||||
reverse_geocode_result = gmaps_client.reverse_geocode((latitude, longitude))
|
||||
if reverse_geocode_result:
|
||||
address = reverse_geocode_result[0]['formatted_address']
|
||||
print(f"✅ Google Maps reverse geocoded address: {address}")
|
||||
try:
|
||||
logger_handler.log_user_activity('reverse_geocoding_success', f'Google Maps reverse geocoded: {latitude}, {longitude} -> {address[:50]}...')
|
||||
except Exception as log_error:
|
||||
print(f"⚠️ Logging error (non-critical): {log_error}")
|
||||
return address
|
||||
else:
|
||||
print("⚠️ Google Maps: No address found for coordinates")
|
||||
|
||||
print("🌐 Falling back to OpenStreetMap Nominatim reverse geocoding")
|
||||
url = "https://nominatim.openstreetmap.org/reverse"
|
||||
params = {'lat': latitude, 'lon': longitude, 'format': 'json', 'addressdetails': 1, 'zoom': 18}
|
||||
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"✅ OSM reverse geocoded address: {address}")
|
||||
try:
|
||||
logger_handler.log_user_activity('reverse_geocoding_fallback', f'OSM reverse geocoded: {latitude}, {longitude} -> {address[:50]}...')
|
||||
except Exception as log_error:
|
||||
print(f"⚠️ Logging error (non-critical): {log_error}")
|
||||
return address
|
||||
else:
|
||||
print("⚠️ 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}")
|
||||
try:
|
||||
logger_handler.log_flask_error('reverse_geocoding_error', f'Reverse geocoding error {latitude}, {longitude}: {str(e)}')
|
||||
except Exception as log_error:
|
||||
print(f"⚠️ Logging error (non-critical): {log_error}")
|
||||
return None
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Location data processing
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
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:
|
||||
if location_data.get('latitude') and location_data['latitude'] not in ['null', '']:
|
||||
lat = float(location_data['latitude'])
|
||||
if -90 <= lat <= 90:
|
||||
processed['latitude'] = lat
|
||||
else:
|
||||
print(f"⚠️ Invalid latitude: {lat}")
|
||||
|
||||
if location_data.get('longitude') and location_data['longitude'] not in ['null', '']:
|
||||
lng = float(location_data['longitude'])
|
||||
if -180 <= lng <= 180:
|
||||
processed['longitude'] = lng
|
||||
else:
|
||||
print(f"⚠️ Invalid longitude: {lng}")
|
||||
|
||||
if location_data.get('accuracy') and location_data['accuracy'] not in ['null', '']:
|
||||
acc = float(location_data['accuracy'])
|
||||
if acc >= 0:
|
||||
processed['accuracy'] = acc
|
||||
else:
|
||||
print(f"⚠️ Invalid accuracy: {acc}")
|
||||
|
||||
if location_data.get('altitude') and location_data['altitude'] not in ['null', '']:
|
||||
alt = float(location_data['altitude'])
|
||||
processed['altitude'] = alt
|
||||
|
||||
except (ValueError, TypeError) as e:
|
||||
print(f"⚠️ Error processing location data: {e}")
|
||||
|
||||
return processed
|
||||
|
||||
|
||||
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:
|
||||
if form_data.get('latitude') and form_data['latitude'] not in ['null', '', 'undefined']:
|
||||
lat = float(form_data['latitude'])
|
||||
if -90 <= lat <= 90:
|
||||
processed['latitude'] = lat
|
||||
else:
|
||||
print(f"⚠️ Invalid latitude: {lat}")
|
||||
|
||||
if form_data.get('longitude') and form_data['longitude'] not in ['null', '', 'undefined']:
|
||||
lng = float(form_data['longitude'])
|
||||
if -180 <= lng <= 180:
|
||||
processed['longitude'] = lng
|
||||
else:
|
||||
print(f"⚠️ Invalid longitude: {lng}")
|
||||
|
||||
if form_data.get('accuracy') and form_data['accuracy'] not in ['null', '', 'undefined']:
|
||||
acc = float(form_data['accuracy'])
|
||||
if acc >= 0:
|
||||
processed['accuracy'] = acc
|
||||
else:
|
||||
print(f"⚠️ Invalid GPS accuracy: {acc}")
|
||||
|
||||
if form_data.get('altitude') and form_data['altitude'] not in ['null', '', 'undefined']:
|
||||
alt = float(form_data['altitude'])
|
||||
processed['altitude'] = alt
|
||||
|
||||
if form_data.get('address'):
|
||||
address = form_data['address'].strip()
|
||||
if address and address not in ['null', '', 'undefined']:
|
||||
if re.match(r'^-?\d+\.\d+,?\s*-?\d+\.\d+$', address.replace(' ', '')):
|
||||
print(f"🔍 Detected coordinate-format address: {address}")
|
||||
processed['address'] = None
|
||||
else:
|
||||
processed['address'] = address[:500]
|
||||
print(f"✅ Using provided address: {processed['address'][:100]}...")
|
||||
|
||||
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("⚠️ Could not reverse geocode coordinates, keeping coordinates as fallback")
|
||||
processed['address'] = f"{processed['latitude']:.10f}, {processed['longitude']:.10f}"
|
||||
|
||||
print("📍 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."""
|
||||
from sqlalchemy import text as sa_text
|
||||
try:
|
||||
print("🔄 Starting enhanced location accuracy migration...")
|
||||
records = db.session.execute(sa_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:
|
||||
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:
|
||||
db.session.execute(sa_text("""
|
||||
UPDATE attendance_data SET location_accuracy = :accuracy WHERE id = :record_id
|
||||
"""), {'accuracy': new_accuracy, 'record_id': record.id})
|
||||
updated_count += 1
|
||||
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}")
|
||||
|
||||
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 location_accuracy column exists in attendance_data table (MySQL compatible)."""
|
||||
from sqlalchemy import text as sa_text
|
||||
try:
|
||||
result = db.session.execute(sa_text("""
|
||||
SELECT COUNT(*) as count
|
||||
FROM INFORMATION_SCHEMA.COLUMNS
|
||||
WHERE TABLE_SCHEMA = DATABASE()
|
||||
AND TABLE_NAME = 'attendance_data'
|
||||
AND COLUMN_NAME = 'location_accuracy'
|
||||
"""))
|
||||
count = result.fetchone().count
|
||||
return count > 0
|
||||
except Exception as e:
|
||||
print(f"Error checking location_accuracy column: {e}")
|
||||
return False
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# QR code location helpers
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
def get_all_locations_from_qr_codes():
|
||||
"""Helper function to get all unique locations from QR codes"""
|
||||
from sqlalchemy import text as sa_text
|
||||
try:
|
||||
result = db.session.execute(sa_text("""
|
||||
SELECT DISTINCT location
|
||||
FROM qr_codes
|
||||
WHERE location IS NOT NULL
|
||||
AND active_status = 1
|
||||
ORDER BY location
|
||||
"""))
|
||||
return [row[0] for row in result.fetchall()]
|
||||
except Exception as e:
|
||||
logger_handler.logger.error(f"Error loading locations: {e}")
|
||||
return []
|
||||
@@ -0,0 +1,522 @@
|
||||
"""
|
||||
utils/helpers.py
|
||||
================
|
||||
Shared utility functions, decorators, QR-code generation helpers,
|
||||
and role/permission helpers.
|
||||
|
||||
Extracted verbatim from app.py (lines 234-329, 910-969, 1274-1467).
|
||||
No logic changes — only import paths updated.
|
||||
"""
|
||||
|
||||
import io
|
||||
import re
|
||||
import os
|
||||
import base64
|
||||
from datetime import datetime, date, time, timedelta
|
||||
from functools import wraps
|
||||
|
||||
import qrcode
|
||||
from flask import session, redirect, flash, request, url_for
|
||||
from user_agents import parse
|
||||
|
||||
from extensions import logger_handler
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Role constants
|
||||
# ---------------------------------------------------------------------------
|
||||
VALID_ROLES = ['admin', 'staff', 'payroll', 'project_manager', 'accounting']
|
||||
STAFF_LEVEL_ROLES = ['staff', 'payroll', 'project_manager', 'accounting']
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Employee ID filter helpers — SP / PW / PT aware
|
||||
# ---------------------------------------------------------------------------
|
||||
# Extra-work records are stored with a work-type code attached to the employee
|
||||
# ID ("1234SP", "1234 PW", "PT-1234", ...) in both attendance_data and
|
||||
# time_attendance. Filtering with a plain equality test on the numeric ID drops
|
||||
# every one of those records. These helpers expand a selected ID into all of its
|
||||
# spellings, and normalize IDs so separators/spacing differences still match.
|
||||
#
|
||||
# Used by: routes/attendance.py, routes/attendance_export.py,
|
||||
# routes/time_attendance.py
|
||||
|
||||
WORK_TYPE_CODES = ('SP', 'PW', 'PT', 'C')
|
||||
|
||||
def get_base_employee_id(raw_employee_id):
|
||||
"""Return the numeric base ID for a possibly work-type-suffixed employee ID."""
|
||||
from working_hours_calculator import parse_employee_id_for_work_type
|
||||
|
||||
base_id, _ = parse_employee_id_for_work_type(str(raw_employee_id or '').strip())
|
||||
return base_id
|
||||
|
||||
|
||||
def _work_type_codes_for(raw_employee_id):
|
||||
"""
|
||||
Return (base_id, codes) for one selected ID.
|
||||
|
||||
A plain ID ("1234") matches regular AND every work type. An ID that already
|
||||
carries a work type ("1234SP") matches only that work type — the user picked
|
||||
it deliberately, so don't broaden the result set.
|
||||
"""
|
||||
from working_hours_calculator import parse_employee_id_for_work_type
|
||||
|
||||
base_id, work_type = parse_employee_id_for_work_type(str(raw_employee_id or '').strip())
|
||||
codes = WORK_TYPE_CODES if work_type == 'regular' else (work_type,)
|
||||
return base_id, work_type, codes
|
||||
|
||||
|
||||
def build_employee_id_variants(raw_employee_id):
|
||||
"""
|
||||
Expand one selected employee ID into the exact ID spellings stored in the
|
||||
database — suffix and prefix forms, with and without a space.
|
||||
|
||||
"1234" -> 1234, 1234SP, "1234 SP", SP1234, "SP 1234", ... (PW / PT / C too)
|
||||
"1234SP" -> the four SP spellings only
|
||||
"""
|
||||
raw = str(raw_employee_id or '').strip()
|
||||
if not raw:
|
||||
return []
|
||||
|
||||
base_id, work_type, codes = _work_type_codes_for(raw)
|
||||
|
||||
variants = [raw]
|
||||
if work_type == 'regular':
|
||||
variants.append(base_id)
|
||||
|
||||
for code in codes:
|
||||
variants.extend([
|
||||
f"{base_id}{code}",
|
||||
f"{base_id} {code}",
|
||||
f"{code}{base_id}",
|
||||
f"{code} {base_id}",
|
||||
])
|
||||
|
||||
return _dedupe(variants)
|
||||
|
||||
|
||||
def build_employee_id_regex(raw_employee_id):
|
||||
"""
|
||||
Build a MySQL REGEXP pattern matching this employee's ID in ANY spelling
|
||||
stored in the database, whatever separator the source file used:
|
||||
|
||||
1759, "1759 SP", 1759SP, 1759.PW, 1759-PT, SP1759, "SP 1759", "01759 SP"
|
||||
|
||||
This is what a fixed variant list cannot do — imported IDs come straight from
|
||||
customer Excel files, so the separator is unpredictable.
|
||||
|
||||
Precision is preserved: the numeric part is anchored, so 17590, 11759 and
|
||||
1759.0 do NOT match a search for 1759.
|
||||
|
||||
Returns None when the base ID is not purely numeric. The pattern is built by
|
||||
string interpolation, and the ID is user-supplied text, so anything that could
|
||||
carry regex metacharacters is refused here — callers fall back to exact matching.
|
||||
"""
|
||||
raw = str(raw_employee_id or '').strip()
|
||||
if not raw:
|
||||
return None
|
||||
|
||||
base_id, work_type, codes = _work_type_codes_for(raw)
|
||||
if not base_id.isdigit():
|
||||
return None
|
||||
|
||||
codes_alt = '|'.join(codes)
|
||||
# 0* tolerates zero-padded IDs. The separator class excludes letters and digits,
|
||||
# so it matches any run of " ", ".", "-", "_" — or nothing at all. That covers
|
||||
# everything people actually type: "1759 PW", "1759.PW", "1759. PW", "1759 . PW".
|
||||
# It never runs over a word, so "1759 SPX" stays a different ID.
|
||||
sep = '[^0-9A-Z]*'
|
||||
number = f'0*{base_id}'
|
||||
|
||||
# Leading and trailing sep runs absorb stray spaces or a trailing dot
|
||||
# ("1759 PW ", "1759 PW.", " SP 1759").
|
||||
if work_type == 'regular':
|
||||
# Regular records AND every work type — code optional on either side
|
||||
return f'^{sep}({codes_alt})?{sep}{number}{sep}({codes_alt})?{sep}$'
|
||||
|
||||
# An explicitly picked work type ("1759SP") must NOT pull in regular records,
|
||||
# so the code is required — on one side or the other.
|
||||
return f'^{sep}(({codes_alt}){sep}{number}|{number}{sep}({codes_alt})){sep}$'
|
||||
|
||||
|
||||
def expand_employee_id_filter(employee_ids):
|
||||
"""
|
||||
Expand a list of selected employee IDs into (exact_variants, regex_patterns),
|
||||
both de-duplicated and order-preserving.
|
||||
|
||||
Callers OR the two together: the exact list is index-friendly and covers the
|
||||
common spellings, the regex list catches every other separator style.
|
||||
"""
|
||||
exact, patterns = [], []
|
||||
for raw in employee_ids or []:
|
||||
exact.extend(build_employee_id_variants(raw))
|
||||
pattern = build_employee_id_regex(raw)
|
||||
if pattern:
|
||||
patterns.append(pattern)
|
||||
return _dedupe(exact), _dedupe(patterns)
|
||||
|
||||
|
||||
def employee_id_regex_condition(column, patterns):
|
||||
"""
|
||||
SQLAlchemy condition: column matches any of the REGEXP patterns, upper-cased so
|
||||
the match does not depend on the column's collation. Patterns are bound as
|
||||
query parameters, never inlined into the SQL string.
|
||||
"""
|
||||
from sqlalchemy import func, or_
|
||||
|
||||
return or_(*[func.upper(column).op('REGEXP')(pattern) for pattern in patterns])
|
||||
|
||||
|
||||
def _dedupe(values):
|
||||
"""Order-preserving de-duplication, dropping empties."""
|
||||
seen, unique_values = set(), []
|
||||
for value in values:
|
||||
if value and value not in seen:
|
||||
seen.add(value)
|
||||
unique_values.append(value)
|
||||
return unique_values
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Role helpers
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
def is_valid_role(role):
|
||||
"""Check if role is valid"""
|
||||
return role in VALID_ROLES
|
||||
|
||||
|
||||
def has_admin_privileges(role):
|
||||
"""Check if role has admin privileges"""
|
||||
return role == 'admin'
|
||||
|
||||
|
||||
def has_staff_level_access(role):
|
||||
"""Check if role has staff-level access (includes new roles)"""
|
||||
return role in STAFF_LEVEL_ROLES
|
||||
|
||||
|
||||
def get_role_permissions(role):
|
||||
"""Get permissions description for a role"""
|
||||
permissions = {
|
||||
'admin': {
|
||||
'title': 'Administrator Permissions',
|
||||
'permissions': [
|
||||
'Full QR code management (create, edit, delete)',
|
||||
'Complete user management capabilities',
|
||||
'System configuration access',
|
||||
'View all system analytics',
|
||||
'Bulk operations and data export',
|
||||
'Access to all admin features'
|
||||
],
|
||||
'restrictions': ['With great power comes great responsibility!']
|
||||
},
|
||||
'staff': {
|
||||
'title': 'Staff User Permissions',
|
||||
'permissions': [
|
||||
'Create and edit QR codes',
|
||||
'View all QR codes in the system',
|
||||
'Download QR code images',
|
||||
'Update personal profile information',
|
||||
],
|
||||
'restrictions': [
|
||||
'Cannot delete QR codes',
|
||||
'Cannot manage other users',
|
||||
'Cannot access admin settings'
|
||||
]
|
||||
},
|
||||
'payroll': {
|
||||
'title': 'Payroll Specialist Permissions',
|
||||
'permissions': [
|
||||
'Create and edit QR codes',
|
||||
'View all QR codes in the system',
|
||||
'Download QR code images',
|
||||
'Update personal profile information',
|
||||
'Access dashboard and reports',
|
||||
'Same permissions as Staff (additional features coming soon)'
|
||||
],
|
||||
'restrictions': [
|
||||
'Cannot delete QR codes',
|
||||
'Cannot manage other users',
|
||||
'Cannot access admin settings'
|
||||
]
|
||||
},
|
||||
'project_manager': {
|
||||
'title': 'Project Manager Permissions',
|
||||
'permissions': [
|
||||
'Create and edit QR codes',
|
||||
'View all QR codes in the system',
|
||||
'Download QR code images',
|
||||
'Update personal profile information',
|
||||
'Access dashboard and reports',
|
||||
'Same permissions as Staff (additional features coming soon)'
|
||||
],
|
||||
'restrictions': [
|
||||
'Cannot delete QR codes',
|
||||
'Cannot manage other users',
|
||||
'Cannot access admin settings'
|
||||
]
|
||||
},
|
||||
'accounting': {
|
||||
'title': 'Accounting Specialist Permissions',
|
||||
'permissions': [
|
||||
'View and modify employee records',
|
||||
'Access attendance reports and analytics',
|
||||
'View and manage time attendance data',
|
||||
'Export payroll and attendance data',
|
||||
'Access financial reports and statistics',
|
||||
'Update personal profile information',
|
||||
'Delete attendance records (same as payroll)'
|
||||
],
|
||||
'restrictions': [
|
||||
'Cannot create or delete QR codes',
|
||||
'Cannot manage other users',
|
||||
'Cannot access admin settings',
|
||||
'Cannot manage projects'
|
||||
]
|
||||
}
|
||||
}
|
||||
return permissions.get(role, {})
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Auth decorators
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
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('auth.login'))
|
||||
return f(*args, **kwargs)
|
||||
return decorated_function
|
||||
|
||||
|
||||
def admin_required(f):
|
||||
"""Decorator to ensure user has admin privileges"""
|
||||
@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('auth.login'))
|
||||
user_role = session.get('role')
|
||||
if not has_admin_privileges(user_role):
|
||||
flash('Administrator privileges required for this action.', 'error')
|
||||
return redirect(url_for('dashboard.dashboard'))
|
||||
return f(*args, **kwargs)
|
||||
return decorated_function
|
||||
|
||||
|
||||
def staff_or_admin_required(f):
|
||||
"""Decorator to ensure user has staff-level or admin privileges"""
|
||||
@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('auth.login'))
|
||||
user_role = session.get('role')
|
||||
if not (has_admin_privileges(user_role) or has_staff_level_access(user_role)):
|
||||
flash('Insufficient privileges to access this page.', 'error')
|
||||
return redirect(url_for('dashboard.dashboard'))
|
||||
return f(*args, **kwargs)
|
||||
return decorated_function
|
||||
|
||||
|
||||
def is_admin_user(user_id):
|
||||
"""Helper function to safely check if user is admin"""
|
||||
from extensions import db
|
||||
from models import set_db
|
||||
try:
|
||||
# User model is available through the app context
|
||||
from flask import current_app
|
||||
with current_app.app_context():
|
||||
# Access via db session to avoid circular import
|
||||
from sqlalchemy import text
|
||||
result = db.session.execute(
|
||||
text("SELECT role, active_status FROM users WHERE id = :uid"),
|
||||
{'uid': user_id}
|
||||
).fetchone()
|
||||
return result and result.active_status and result.role == 'admin'
|
||||
except Exception:
|
||||
return False
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Request helpers
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
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]
|
||||
except Exception:
|
||||
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']
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# QR code generation
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
def generate_qr_url(name, qr_id):
|
||||
"""Generate a unique URL for QR code destination"""
|
||||
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()
|
||||
url_slug = f"qr-{qr_id}-{clean_name}"
|
||||
return url_slug[:200]
|
||||
|
||||
|
||||
def generate_qr_code(data, fill_color="black", back_color="white", box_size=10, border=4, error_correction='L'):
|
||||
"""Generate a QR code image and return as base64 string"""
|
||||
error_correction_map = {
|
||||
'L': qrcode.constants.ERROR_CORRECT_L,
|
||||
'M': qrcode.constants.ERROR_CORRECT_M,
|
||||
'Q': qrcode.constants.ERROR_CORRECT_Q,
|
||||
'H': qrcode.constants.ERROR_CORRECT_H
|
||||
}
|
||||
|
||||
try:
|
||||
qr = qrcode.QRCode(
|
||||
version=1,
|
||||
error_correction=error_correction_map.get(error_correction, qrcode.constants.ERROR_CORRECT_L),
|
||||
box_size=int(box_size),
|
||||
border=int(border),
|
||||
)
|
||||
qr.add_data(data)
|
||||
qr.make(fit=True)
|
||||
|
||||
img = qr.make_image(fill_color=fill_color, back_color=back_color)
|
||||
|
||||
buffer = io.BytesIO()
|
||||
img.save(buffer, format='PNG')
|
||||
img_str = base64.b64encode(buffer.getvalue()).decode()
|
||||
|
||||
try:
|
||||
logger_handler.log_qr_code_generated(
|
||||
data_length=len(data),
|
||||
fill_color=fill_color,
|
||||
back_color=back_color,
|
||||
box_size=box_size,
|
||||
border=border,
|
||||
error_correction=error_correction
|
||||
)
|
||||
except Exception:
|
||||
pass
|
||||
|
||||
return img_str
|
||||
|
||||
except Exception as e:
|
||||
logger_handler.log_database_error('qr_code_generation', e)
|
||||
return generate_default_qr_code(data)
|
||||
|
||||
|
||||
def generate_default_qr_code(data):
|
||||
"""Fallback function for basic QR code generation"""
|
||||
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")
|
||||
buffer = io.BytesIO()
|
||||
img.save(buffer, format='PNG')
|
||||
img_str = base64.b64encode(buffer.getvalue()).decode()
|
||||
return img_str
|
||||
|
||||
|
||||
def get_qr_styling(qr_code):
|
||||
"""Extract QR code styling parameters from database record"""
|
||||
return {
|
||||
'fill_color': getattr(qr_code, 'fill_color', '#000000') or '#000000',
|
||||
'back_color': getattr(qr_code, 'back_color', '#FFFFFF') or '#FFFFFF',
|
||||
'box_size': getattr(qr_code, 'box_size', 10) or 10,
|
||||
'border': getattr(qr_code, 'border', 4) or 4,
|
||||
'error_correction': getattr(qr_code, 'error_correction', 'L') or 'L'
|
||||
}
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Check-in history helpers
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
def get_employee_checkin_history(employee_id, qr_code_id, date_filter=None):
|
||||
"""Get check-in history for an employee at a specific location"""
|
||||
from extensions import db
|
||||
try:
|
||||
if date_filter is None:
|
||||
date_filter = date.today()
|
||||
# AttendanceData imported at call site to avoid circular import
|
||||
from flask import current_app
|
||||
from models.attendance import AttendanceData
|
||||
if AttendanceData:
|
||||
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
|
||||
return []
|
||||
except Exception as e:
|
||||
print(f"❌ Error retrieving checkin history: {e}")
|
||||
return []
|
||||
|
||||
|
||||
def format_checkin_intervals(checkins):
|
||||
"""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):
|
||||
"""Format minutes into human-readable time interval"""
|
||||
if minutes < 60:
|
||||
return f"{minutes} minutes"
|
||||
elif minutes < 1440:
|
||||
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"
|
||||
@@ -0,0 +1,75 @@
|
||||
"""
|
||||
utils/template_helpers.py
|
||||
=========================
|
||||
Template utility functions injected into Jinja2 via context processors.
|
||||
|
||||
Extracted from create_app() in app.py so they can be independently
|
||||
imported, tested, and reused.
|
||||
"""
|
||||
|
||||
from datetime import datetime
|
||||
from sqlalchemy import text as sa_text
|
||||
|
||||
from extensions import db, logger_handler
|
||||
|
||||
|
||||
def get_employee_name(employee_id):
|
||||
"""Return 'Lastname, Firstname' for a given employee ID.
|
||||
Falls back to 'Employee <id>' if not found or on error.
|
||||
"""
|
||||
try:
|
||||
result = db.session.execute(sa_text("""
|
||||
SELECT CONCAT(firstName, ' ', lastName) as full_name
|
||||
FROM employee
|
||||
WHERE id = :employee_id
|
||||
"""), {'employee_id': employee_id})
|
||||
row = result.fetchone()
|
||||
return row[0] if row else f"Employee {employee_id}"
|
||||
except Exception as e:
|
||||
print(f"⚠️ Error getting employee name for ID {employee_id}: {e}")
|
||||
return f"Employee {employee_id}"
|
||||
|
||||
|
||||
def get_qr_code_checkin_count(qr_code_id):
|
||||
"""Return total number of check-ins for a given QR code ID."""
|
||||
from models.attendance import AttendanceData
|
||||
try:
|
||||
return AttendanceData.query.filter_by(qr_code_id=qr_code_id).count()
|
||||
except Exception as e:
|
||||
logger_handler.logger.error(
|
||||
f"Error getting check-ins count for QR {qr_code_id}: {e}"
|
||||
)
|
||||
return 0
|
||||
|
||||
|
||||
def format_hours(hours):
|
||||
"""Format a decimal hours value to 2 decimal places."""
|
||||
return f"{hours:.2f}" if hours else "0.00"
|
||||
|
||||
|
||||
def register_template_helpers(app):
|
||||
"""
|
||||
Register all template helper context processors on the given Flask app.
|
||||
Call this once inside create_app() after the app is configured.
|
||||
"""
|
||||
from working_hours_calculator import (
|
||||
convert_minutes_to_base100, round_base100_hours
|
||||
)
|
||||
|
||||
@app.context_processor
|
||||
def inject_payroll_utils():
|
||||
"""Inject payroll utility functions into all templates."""
|
||||
return {
|
||||
'convert_minutes_to_base100': convert_minutes_to_base100,
|
||||
'round_base100_hours': round_base100_hours,
|
||||
'get_employee_name': get_employee_name,
|
||||
'format_hours': format_hours,
|
||||
}
|
||||
|
||||
@app.context_processor
|
||||
def inject_dashboard_utils():
|
||||
"""Inject dashboard utility functions into all templates."""
|
||||
return {
|
||||
'now': datetime.utcnow,
|
||||
'get_qr_code_checkin_count': get_qr_code_checkin_count,
|
||||
}
|
||||
Reference in New Issue
Block a user