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

Mar. 23 2026 patch code 2

Browse Source
This commit is contained in:
nngo
2026-03-23 12:46:35 -04:00
parent 3c36dbc15f
commit 28bcd13c7f
15 changed files with 783 additions and 760 deletions
Download Patch File Download Diff File Expand all files Collapse all files
app.py
+3 -4
View File
@@ -326,7 +326,6 @@ def update_existing_qr_codes():
lh.logger.info(f"Updated {updated_count} existing QR codes with missing URLs/images")
except Exception as e:
lh.log_database_error('update_existing_qr_codes', e)
print(f"Error updating existing QR codes: {e}")
def log_slow_query_performance(app_instance):
@@ -368,15 +367,15 @@ if __name__ == '__main__':
create_tables()
log_slow_query_performance(app)
print("🚀 Initializing performance optimizations...")
from extensions import logger_handler
logger_handler.logger.info("Initializing performance optimizations")
cached_query = initialize_performance_optimizations(app, db, logger_handler)
performance_monitor = PerformanceMonitor(app, db, logger_handler)
if cached_query:
print("✅ Performance optimizations completed successfully")
logger_handler.logger.info("Performance optimizations completed successfully")
else:
print("⚠️ Performance optimizations completed with warnings")
logger_handler.logger.warning("Performance optimizations completed with warnings")
logger_handler.logger.info("QR Attendance Management System started successfully")
logger_handler.py
+59 -49
View File
@@ -25,7 +25,7 @@ import os
import traceback
from datetime import datetime, date, timedelta
from functools import wraps
from flask import request, session, g, render_template, has_request_context
from flask import request, session, g, render_template, has_request_context, current_app
from sqlalchemy import text
from sqlalchemy.exc import SQLAlchemyError
import uuid
@@ -169,7 +169,7 @@ class AppLogger:
self.db.session.commit()
except Exception as e:
print(f"Warning: Could not create log_events table: {e}")
logging.getLogger('qr_attendance_app').warning(f"Could not create log_events table: {e}")
def _register_error_handlers(self):
"""Register Flask error handlers for automatic logging"""
@@ -258,7 +258,7 @@ class AppLogger:
except Exception as e:
# Don't let logging errors break the application
print(f"Database logging error: {e}")
logging.getLogger('qr_attendance_app').warning(f"Database logging error (non-fatal): {e}")
try:
self.db.session.rollback()
except:
@@ -618,10 +618,10 @@ class AppLogger:
try:
table_check = self.db.session.execute(text("SHOW TABLES LIKE 'log_events'")).fetchone()
if not table_check:
print("⚠️ log_events table does not exist")
self.logger.warning("get_log_statistics: log_events table does not exist")
return stats
except Exception as table_error:
print(f"⚠️ Cannot check if log_events table exists: {table_error}")
self.logger.warning(f"get_log_statistics: cannot check table existence: {table_error}")
return stats
# Get total events count
@@ -635,9 +635,9 @@ class AppLogger:
total_result = self.db.session.execute(text(total_sql), {'cutoff_date': cutoff_date}).fetchone()
if total_result:
stats['total_events'] = total_result.total_events
print(f"✅ Found {stats['total_events']} total events in last {days} days")
self.logger.debug(f"get_log_statistics: {stats['total_events']} total events in last {days} days")
except Exception as total_error:
print(f"⚠️ Error getting total events: {total_error}")
self.logger.warning(f"get_log_statistics: error getting total events: {total_error}")
# Get events by category
try:
@@ -655,7 +655,7 @@ class AppLogger:
for row in category_result:
category = row.event_category
count = row.event_count
print(f"✅ Found {count} events in category: {category}")
self.logger.debug(f"get_log_statistics: {count} events in category: {category}")
# Map categories to stats keys
if category == 'security':
@@ -672,13 +672,13 @@ class AppLogger:
stats['system_events'] = count
except Exception as category_error:
print(f"⚠️ Error getting category stats: {category_error}")
self.logger.warning(f"get_log_statistics: error getting category stats: {category_error}")
print(f"📊 Final stats: {stats}")
self.logger.debug(f"get_log_statistics result: {stats}")
return stats
except Exception as e:
print(f"❌ Error in get_log_statistics: {e}")
self.logger.error(f"Error in get_log_statistics: {e}", exc_info=True)
self.log_database_error('get_log_statistics', e)
return {
'total_events': 0,
@@ -695,16 +695,16 @@ class AppLogger:
try:
from datetime import datetime, timedelta # Import here as backup
cutoff_date = datetime.now() - timedelta(days=days_to_keep)
print(f"🧹 Starting log cleanup: removing entries older than {cutoff_date}")
self.logger.info(f"Starting log cleanup: removing entries older than {cutoff_date}")
# Check if table exists first
try:
table_check = self.db.session.execute(text("SHOW TABLES LIKE 'log_events'")).fetchone()
if not table_check:
print("⚠️ log_events table does not exist")
self.logger.warning("cleanup_old_logs: log_events table does not exist")
return 0
except Exception as table_error:
print(f"⚠️ Cannot check if log_events table exists: {table_error}")
self.logger.warning(f"cleanup_old_logs: cannot check table existence: {table_error}")
return 0
# First, count how many records will be deleted
@@ -719,14 +719,14 @@ class AppLogger:
count_result = self.db.session.execute(text(count_sql), {'cutoff_date': cutoff_date}).fetchone()
count_to_delete = count_result.count_to_delete if count_result else 0
print(f"📊 Found {count_to_delete} records to delete")
self.logger.debug(f"cleanup_old_logs: {count_to_delete} records to delete")
if count_to_delete == 0:
print("✅ No old records found to cleanup")
self.logger.info("cleanup_old_logs: no old records found to cleanup")
return 0
except Exception as count_error:
print(f"⚠️ Error counting records to delete: {count_error}")
self.logger.warning(f"cleanup_old_logs: error counting records: {count_error}")
return 0
# Perform the cleanup - exclude critical logs
@@ -741,7 +741,7 @@ class AppLogger:
deleted_count = result.rowcount
self.db.session.commit()
print(f"🗑️ Successfully deleted {deleted_count} old log entries")
self.logger.info(f"cleanup_old_logs: deleted {deleted_count} old log entries")
# Log the cleanup operation
self.logger.info(f"Log cleanup completed: {deleted_count} entries removed (keeping entries newer than {days_to_keep} days)")
@@ -749,12 +749,12 @@ class AppLogger:
return deleted_count
except Exception as delete_error:
print(f"❌ Error during deletion: {delete_error}")
self.logger.error(f"cleanup_old_logs: error during deletion: {delete_error}", exc_info=True)
self.db.session.rollback()
return 0
except Exception as e:
print(f"❌ Error in cleanup_old_logs: {e}")
self.logger.error(f"Error in cleanup_old_logs: {e}", exc_info=True)
self.db.session.rollback()
self.log_database_error('cleanup_old_logs', e)
return 0
@@ -819,7 +819,7 @@ class AppLogger:
except Exception as e:
self.log_database_error('get_recent_logs', e)
print(f"Error in get_recent_logs: {e}")
self.logger.error(f"Error in get_recent_logs: {e}", exc_info=True)
return []
def log_system_event(self, event_type, description, severity='INFO', additional_data=None):
@@ -873,15 +873,15 @@ class AppLogger:
"""
result = self.db.session.execute(text(check_table_sql)).fetchone()
if result.table_exists == 0:
print("⚠️ log_events table does not exist. Creating it now...")
self.logger.warning("log_events table does not exist — creating it now")
self._create_log_table()
return True
count_sql = "SELECT COUNT(*) as record_count FROM log_events"
count_result = self.db.session.execute(text(count_sql)).fetchone()
print(f"✅ log_events table exists with {count_result.record_count} records")
self.logger.debug(f"log_events table exists with {count_result.record_count} records")
return True
except Exception as e:
print(f"❌ Error verifying log table: {e}")
self.logger.error(f"Error verifying log table: {e}", exc_info=True)
return False
def log_modal_interaction(self, event_type, description, additional_data=None):
@@ -904,64 +904,74 @@ class AppLogger:
severity='INFO'
)
except Exception as e:
print(f"Error logging modal interaction: {e}")
self.logger.warning(f"Error logging modal interaction: {e}")
# DECORATOR FUNCTIONS FOR AUTOMATIC LOGGING
def log_user_activity(activity_type):
"""Decorator to automatically log user activities"""
"""Decorator to automatically log user activities to file and database."""
def decorator(f):
@wraps(f)
def decorated_function(*args, **kwargs):
try:
result = f(*args, **kwargs)
# Log successful activity
if hasattr(g, 'app') and hasattr(g.app, 'logger_handler'):
logger = g.app.logger_handler
logger.logger.info(json.dumps({
'event': f'user_activity_{activity_type}',
'data': {
'user_id': session.get('user_id'),
'username': session.get('username'),
'activity': activity_type,
'timestamp': datetime.now().isoformat()
}
}))
# Log successful activity via the AppLogger instance on current_app
try:
lh = current_app.logger_handler
lh.log_user_activity(
activity_type=activity_type,
description=(
f"User '{session.get('username', 'anonymous')}' "
f"completed activity: {activity_type}"
)
)
except Exception as log_error:
# Logging must never break the decorated route
logging.getLogger('qr_attendance_app').warning(
f"log_user_activity decorator failed for '{activity_type}': {log_error}"
)
return result
except Exception as e:
# Log error
if hasattr(g, 'app') and hasattr(g.app, 'logger_handler'):
logger = g.app.logger_handler
logger.log_flask_error(
# Log the error, then re-raise so Flask handles it normally
try:
lh = current_app.logger_handler
lh.log_flask_error(
error_type=f"activity_error_{activity_type}",
error_message=str(e),
stack_trace=traceback.format_exc()
)
except Exception as log_error:
logging.getLogger('qr_attendance_app').warning(
f"log_user_activity error-branch failed for '{activity_type}': {log_error}"
)
raise
return decorated_function
return decorator
def log_database_operations(operation_name):
"""Decorator to automatically log database operations"""
"""Decorator to automatically log database operation errors."""
def decorator(f):
@wraps(f)
def decorated_function(*args, **kwargs):
try:
result = f(*args, **kwargs)
return result
return f(*args, **kwargs)
except Exception as e:
# Log database error
if hasattr(g, 'app') and hasattr(g.app, 'logger_handler'):
logger = g.app.logger_handler
logger.log_database_error(
# Log the database error via the AppLogger instance on current_app
try:
lh = current_app.logger_handler
lh.log_database_error(
operation=operation_name,
error=e
)
except Exception as log_error:
logging.getLogger('qr_attendance_app').warning(
f"log_database_operations decorator failed for '{operation_name}': {log_error}"
)
raise
return decorated_function
routes/admin.py
+26 -33
View File
@@ -77,8 +77,10 @@ def api_recent_logs():
severity = request.args.get('severity', '')
search = request.args.get('search', '')
print(f"📊 API request - Days: {days}, Limit: {limit}, Page: {page}")
print(f"📊 Filters - Category: {category}, Severity: {severity}, Search: {search}")
logger_handler.logger.debug(
f"api_recent_logs: days={days}, limit={limit}, page={page}, "
f"category={category!r}, severity={severity!r}, search={search!r}"
)
cutoff_date = datetime.now() - timedelta(days=days)
@@ -164,7 +166,7 @@ def api_recent_logs():
'ip_address': row.ip_address or '-'
})
print(f"📊 Returning {len(logs)} logs out of {total_count} total")
logger_handler.logger.debug(f"api_recent_logs: returning {len(logs)} of {total_count} total records")
return jsonify({
'success': True,
@@ -179,7 +181,6 @@ def api_recent_logs():
except Exception as e:
logger_handler.log_database_error('api_recent_logs', e)
print(f"Error in api_recent_logs: {e}")
return jsonify({
'success': False,
'error': f'Failed to fetch recent logs: {str(e)}'
@@ -191,11 +192,11 @@ def api_log_stats():
"""API endpoint to get logging statistics"""
try:
days = request.args.get('days', 7, type=int)
print(f"📊 Getting log statistics for last {days} days")
logger_handler.logger.debug(f"api_log_stats: fetching statistics for last {days} days")
# Get statistics from logger handler
stats = logger_handler.get_log_statistics(days=days)
print(f"📈 Retrieved stats: {stats}")
# Ensure all expected keys exist with updated categories
expected_stats = {
@@ -217,7 +218,6 @@ def api_log_stats():
except Exception as e:
logger_handler.log_database_error('api_log_stats', e)
print(f"❌ Error in api_log_stats: {e}")
return jsonify({
'success': False,
'error': f'Failed to fetch log statistics: {str(e)}',
@@ -240,25 +240,21 @@ def api_cleanup_logs():
# Get JSON data
data = request.get_json()
if not data:
print("❌ No JSON data provided")
return jsonify({
'success': False,
'error': 'No JSON data provided'
}), 400
days_to_keep = data.get('days_to_keep', 90)
print(f"🧹 Cleanup request: keep last {days_to_keep} days")
# Validate input
if not isinstance(days_to_keep, int) or days_to_keep < 7:
print(f"❌ Invalid days_to_keep: {days_to_keep}")
return jsonify({
'success': False,
'error': 'days_to_keep must be an integer >= 7'
}), 400
if days_to_keep > 365:
print(f"❌ days_to_keep too large: {days_to_keep}")
return jsonify({
'success': False,
'error': 'days_to_keep cannot exceed 365 days'
@@ -268,7 +264,10 @@ def api_cleanup_logs():
deleted_count = logger_handler.cleanup_old_logs(days_to_keep=days_to_keep)
admin_username = session.get('username', 'unknown')
print(f"✅ Cleanup completed by {admin_username}: {deleted_count} records deleted")
logger_handler.logger.info(
f"Admin {admin_username} performed log cleanup: {deleted_count} records deleted "
f"(keeping last {days_to_keep} days)"
)
# Log the admin action
logger_handler.log_security_event(
@@ -294,7 +293,6 @@ def api_cleanup_logs():
except Exception as e:
logger_handler.log_database_error('api_cleanup_logs', e)
print(f"❌ Error in api_cleanup_logs: {e}")
return jsonify({
'success': False,
'error': f'Failed to cleanup old logs: {str(e)}'
@@ -306,7 +304,7 @@ def api_clear_logs():
"""API endpoint to clear ALL log entries"""
try:
admin_username = session.get('username', 'unknown')
print(f"🧹 Clear logs request by admin: {admin_username}")
logger_handler.logger.info(f"Admin {admin_username} initiated full log clear")
# Count existing logs before deletion
try:
@@ -314,10 +312,7 @@ def api_clear_logs():
count_result = db.session.execute(text(count_sql)).fetchone()
total_logs = count_result.total_logs if count_result else 0
print(f"📊 Total logs to be cleared: {total_logs}")
if total_logs == 0:
print("✅ No logs found to clear")
return jsonify({
'success': True,
'deleted_count': 0,
@@ -325,7 +320,7 @@ def api_clear_logs():
})
except Exception as count_error:
print(f"⚠️ Error counting logs: {count_error}")
logger_handler.logger.warning(f"Error counting logs before clear: {count_error}")
total_logs = 0
# Perform the clear operation
@@ -335,7 +330,9 @@ def api_clear_logs():
deleted_count = result.rowcount
db.session.commit()
print(f"🗑️ Successfully cleared {deleted_count} log entries")
logger_handler.logger.info(
f"Admin {admin_username} cleared all log entries: {deleted_count} records deleted"
)
# Log the clear operation (this will be the first entry in the new log)
logger_handler.log_security_event(
@@ -358,7 +355,7 @@ def api_clear_logs():
})
except Exception as delete_error:
print(f"❌ Error during log clearing: {delete_error}")
logger_handler.log_database_error('api_clear_logs_delete', delete_error)
db.session.rollback()
return jsonify({
'success': False,
@@ -367,7 +364,6 @@ def api_clear_logs():
except Exception as e:
logger_handler.log_database_error('api_clear_logs', e)
print(f"❌ Error in api_clear_logs: {e}")
return jsonify({
'success': False,
'error': f'Failed to clear logs: {str(e)}'
@@ -381,7 +377,6 @@ def api_clear_old_logs():
# Get JSON data
data = request.get_json()
if not data:
print("❌ No JSON data provided")
return jsonify({
'success': False,
'error': 'No JSON data provided'
@@ -389,11 +384,9 @@ def api_clear_old_logs():
days_threshold = data.get('days_threshold', 90)
admin_username = session.get('username', 'unknown')
print(f"🧹 Clear old logs request by admin: {admin_username}, threshold: {days_threshold} days")
# Validate input
if not isinstance(days_threshold, int) or days_threshold not in [30, 60, 90]:
print(f"❌ Invalid days_threshold: {days_threshold}")
return jsonify({
'success': False,
'error': 'days_threshold must be 30, 60, or 90'
@@ -408,10 +401,7 @@ def api_clear_old_logs():
count_result = db.session.execute(text(count_sql), {'cutoff_date': cutoff_date}).fetchone()
total_logs = count_result.total_logs if count_result else 0
print(f"📊 Total logs older than {days_threshold} days to be cleared: {total_logs}")
if total_logs == 0:
print("✅ No old logs found to clear")
return jsonify({
'success': True,
'deleted_count': 0,
@@ -419,7 +409,7 @@ def api_clear_old_logs():
})
except Exception as count_error:
print(f"⚠️ Error counting old logs: {count_error}")
logger_handler.logger.warning(f"Error counting old logs before clear: {count_error}")
total_logs = 0
# Perform the clear operation
@@ -429,7 +419,9 @@ def api_clear_old_logs():
deleted_count = result.rowcount
db.session.commit()
print(f"🗑️ Successfully cleared {deleted_count} log entries older than {days_threshold} days")
logger_handler.logger.info(
f"Admin {admin_username} cleared {deleted_count} log entries older than {days_threshold} days"
)
# Log the clear operation
logger_handler.log_security_event(
@@ -455,7 +447,7 @@ def api_clear_old_logs():
})
except Exception as delete_error:
print(f"❌ Error during old log clearing: {delete_error}")
logger_handler.log_database_error('api_clear_old_logs_delete', delete_error)
db.session.rollback()
return jsonify({
'success': False,
@@ -464,7 +456,6 @@ def api_clear_old_logs():
except Exception as e:
logger_handler.log_database_error('api_clear_old_logs', e)
print(f"❌ Error in api_clear_old_logs: {e}")
return jsonify({
'success': False,
'error': f'Failed to clear old logs: {str(e)}'
@@ -481,7 +472,10 @@ def api_export_logs():
search = request.args.get('search', '')
admin_username = session.get('username', 'unknown')
print(f"📊 Export logs request by admin: {admin_username}")
logger_handler.logger.info(
f"Admin {admin_username} initiated log export: last {days} days, "
f"category={category!r}, severity={severity!r}"
)
cutoff_date = datetime.now() - timedelta(days=days)
@@ -578,7 +572,6 @@ def api_export_logs():
except Exception as e:
logger_handler.log_database_error('api_export_logs', e)
print(f"❌ Error in api_export_logs: {e}")
return jsonify({
'success': False,
'error': f'Failed to export logs: {str(e)}'
routes/attendance.py
+129 -145
View File
@@ -44,18 +44,18 @@ bp = Blueprint('attendance', __name__)
def attendance_report():
"""Safe attendance report with backward compatibility for location_accuracy and fixed datetime handling"""
try:
print("📊 Loading attendance report...")
logger_handler.logger.debug("Loading attendance report")
# Log attendance report access
try:
user_role = session.get('role', 'unknown')
logger_handler.logger.info(f"User {session.get('username', 'unknown')} accessed attendance report")
except Exception as log_error:
print(f"⚠️ Logging error (non-critical): {log_error}")
except Exception:
pass
# Check if location_accuracy column exists
has_location_accuracy = check_location_accuracy_column_exists()
print(f"🔍 Location accuracy column exists: {has_location_accuracy}")
logger_handler.logger.debug(f"Location accuracy column exists: {has_location_accuracy}")
# Get filter parameters
date_from = request.args.get('date_from', '')
@@ -98,7 +98,7 @@ def attendance_report():
# Check if user is Project Manager and get their permissions
if user_role == 'project_manager':
print(f"🔒 Project Manager access control enabled for user {session.get('username')}")
logger_handler.logger.debug(f"Project Manager access control enabled for user {session.get('username')}")
try:
# Get assigned projects
@@ -115,10 +115,9 @@ def attendance_report():
f"Projects: {allowed_project_ids}, Locations: {allowed_location_names}"
)
print(f"🔒 Allowed projects: {allowed_project_ids}")
print(f"🔒 Allowed locations: {allowed_location_names}")
logger_handler.logger.debug(f"PM allowed projects: {allowed_project_ids}, locations: {allowed_location_names}")
except Exception as perm_error:
print(f"⚠️ Error loading permissions: {perm_error}")
logger_handler.logger.warning(f"Error loading PM permissions: {perm_error}")
logger_handler.logger.error(f"Error loading Project Manager permissions: {perm_error}")
# If no permissions assigned, user cannot view anything
@@ -278,12 +277,12 @@ def attendance_report():
base_query += " ORDER BY ad.check_in_date DESC, ad.check_in_time DESC LIMIT 1000"
print(f"🔍 Executing attendance query with filters: {list(query_params.keys())}")
logger_handler.logger.debug(f"Executing attendance query with filters: {list(query_params.keys())}")
# Execute query
result = db.session.execute(text(base_query), query_params)
records = result.fetchall()
print(f"✅ Loaded {len(records)} attendance records")
logger_handler.logger.debug(f"Loaded {len(records)} attendance records")
# Process records
processed_records = []
@@ -322,7 +321,7 @@ def attendance_report():
record_dict['accuracy_level'] = 'unknown'
processed_records.append(record_dict)
except Exception as rec_error:
print(f"⚠️ Error processing record: {rec_error}")
logger_handler.logger.warning(f"Error processing attendance record: {rec_error}")
continue
# Get unique locations for filter dropdown
@@ -333,7 +332,7 @@ def attendance_report():
if user_role == 'project_manager' and allowed_location_names:
# Only show locations the PM has access to
locations = sorted(allowed_location_names)
print(f"✅ Filtered to {len(locations)} locations for Project Manager")
logger_handler.logger.debug(f"Filtered to {len(locations)} locations for Project Manager")
else:
# Show all locations for Admin/Staff/Payroll
locations_query = db.session.execute(text("""
@@ -343,12 +342,12 @@ def attendance_report():
ORDER BY location_name
"""))
locations = [row[0] for row in locations_query.fetchall()]
print(f"✅ Found {len(locations)} unique locations")
logger_handler.logger.debug(f"Found {len(locations)} unique locations")
# ============================================================
# END: FILTER LOCATIONS FOR PROJECT MANAGER
# ============================================================
except Exception as e:
print(f"⚠️ Error loading locations: {e}")
logger_handler.logger.warning(f"Error loading locations filter: {e}")
locations = []
# Get projects for filter dropdown
@@ -369,7 +368,7 @@ def attendance_report():
ORDER BY p.name
"""))
projects = projects_query.fetchall()
print(f"✅ Filtered to {len(projects)} projects for Project Manager")
logger_handler.logger.debug(f"Filtered to {len(projects)} projects for Project Manager")
else:
# Show all projects for Admin/Staff/Payroll
projects = db.session.execute(text("""
@@ -382,18 +381,18 @@ def attendance_report():
HAVING COUNT(DISTINCT ad.id) > 0
ORDER BY p.name
""")).fetchall()
print(f"✅ Loaded {len(projects)} projects with attendance data")
logger_handler.logger.debug(f"Loaded {len(projects)} projects with attendance data")
# ============================================================
# END: FILTER PROJECTS FOR PROJECT MANAGER
# ============================================================
except Exception as e:
print(f"⚠️ Error loading projects: {e}")
logger_handler.logger.warning(f"Error loading projects filter: {e}")
projects = []
# ============================================================
# STATISTICS - COMPLETELY REWRITTEN FOR SAFETY
# ============================================================
print("📊 Loading statistics...")
logger_handler.logger.debug("Loading attendance statistics")
# Create simple dict for stats (most compatible approach)
stats_dict = {
@@ -455,15 +454,13 @@ def attendance_report():
if stats_conditions:
stats_query_text += " AND " + " AND ".join(stats_conditions)
print(f"📊 Executing stats query...")
print(f"📊 Stats params: {list(stats_params.keys())}")
logger_handler.logger.debug(f"Executing stats query with params: {list(stats_params.keys())}")
# Execute stats query
stats_result = db.session.execute(text(stats_query_text), stats_params)
stats_row = stats_result.fetchone()
print(f"📊 Stats row type: {type(stats_row)}")
print(f"📊 Stats row value: {stats_row}")
logger_handler.logger.debug(f"Stats row type: {type(stats_row).__name__}")
# Safely extract stats from row
if stats_row is not None and len(stats_row) >= 7:
@@ -475,17 +472,15 @@ def attendance_report():
stats_dict['records_with_gps'] = int(stats_row[4]) if stats_row[4] is not None else 0
stats_dict['records_with_accuracy'] = int(stats_row[5]) if stats_row[5] is not None else 0
stats_dict['avg_location_accuracy'] = float(stats_row[6]) if stats_row[6] is not None else 0.0
print(f"✅ Loaded statistics: {stats_dict['total_checkins']} total check-ins")
logger_handler.logger.debug(f"Loaded statistics: {stats_dict['total_checkins']} total check-ins")
except (IndexError, TypeError, ValueError) as extract_error:
print(f"⚠️ Error extracting stats values: {extract_error}")
logger_handler.logger.warning(f"Error extracting stats values: {extract_error}")
# stats_dict already has default values
else:
print("⚠️ Stats query returned None or insufficient columns, using default stats")
logger_handler.logger.warning("Stats query returned None or insufficient columns, using default stats")
except Exception as stats_error:
print(f"❌ Error loading statistics: {stats_error}")
import traceback
print(f"❌ Stats error traceback: {traceback.format_exc()}")
logger_handler.logger.error(f"Error loading statistics: {stats_error}", exc_info=True)
# stats_dict already has default values
# Convert dict to object-like for template compatibility
@@ -495,7 +490,7 @@ def attendance_report():
setattr(self, key, value)
stats = StatsObject(stats_dict)
print(f"✅ Stats object created: total_checkins={stats.total_checkins}")
logger_handler.logger.debug(f"Stats object created: total_checkins={stats.total_checkins}")
# ============================================================
# END: STATISTICS
@@ -505,8 +500,7 @@ def attendance_report():
today_date = datetime.now().strftime('%Y-%m-%d')
current_date_formatted = datetime.now().strftime('%B %d')
print("✅ Rendering attendance report template")
print(f"✅ Stats object: {stats}")
logger_handler.logger.debug("Rendering attendance report template")
return render_template('attendance_report.html',
attendance_records=processed_records,
@@ -527,18 +521,17 @@ def attendance_report():
user_role=user_role)
except Exception as e:
print(f"❌ Error loading attendance report: {e}")
print(f"❌ Exception type: {type(e)}")
logger_handler.logger.error(f"Error loading attendance report: {e}", exc_info=True)
import traceback
error_traceback = traceback.format_exc()
print(f"❌ Traceback: {error_traceback}")
# Log the error
try:
logger_handler.log_database_error('attendance_report', e)
except Exception as log_error:
print(f"⚠️ Additional logging error: {log_error}")
logger_handler.logger.warning(f"Additional logging error: {log_error}")
flash('Error loading attendance report. Please check the server logs for details.', 'error')
return redirect(url_for('dashboard.dashboard'))
@@ -595,7 +588,7 @@ def edit_attendance(record_id):
qr_codes=QRCode.query.filter_by(active_status=True).all())
# Validate the QR code exists
new_qr_code = QRCode.query.get(int(new_qr_code_id))
new_qr_code = db.session.get(QRCode, int(new_qr_code_id))
if not new_qr_code:
flash('Selected location event not found.', 'error')
projects = Project.query.filter_by(active_status=True).order_by(Project.name).all()
@@ -657,8 +650,10 @@ def edit_attendance(record_id):
'editor_username': session.get('username')
}
)
print(f"[LOG] {session.get('role', 'unknown').title()} {session.get('username')} updated attendance record {record_id}: {changes}")
print(f"[LOG] Edit reason: {edit_note}")
logger_handler.logger.info(
f"User {session.get('username')} ({session.get('role', 'unknown')}) "
f"updated attendance record {record_id}: {changes}, reason: {edit_note}"
)
else:
# Log even if no changes were made (for audit purposes)
logger_handler.log_security_event(
@@ -672,8 +667,10 @@ def edit_attendance(record_id):
'editor_username': session.get('username')
}
)
print(f"[LOG] {session.get('role', 'unknown').title()} {session.get('username')} edited record {record_id} with no changes")
print(f"[LOG] Edit reason: {edit_note}")
logger_handler.logger.info(
f"User {session.get('username')} ({session.get('role', 'unknown')}) "
f"edited attendance record {record_id} with no changes, reason: {edit_note}"
)
flash(f'Attendance record for {new_employee_id} updated successfully! Edit reason logged for audit.', 'success')
return redirect(url_for('attendance.attendance_report'))
@@ -693,7 +690,7 @@ def edit_attendance(record_id):
except Exception as e:
db.session.rollback()
logger_handler.log_database_error('attendance_update', e)
print(f"[LOG] Error updating attendance record {record_id}: {e}")
logger_handler.logger.error(f"Error updating attendance record {record_id}: {e}", exc_info=True)
flash('Error updating attendance record. Please try again.', 'error')
return redirect(url_for('attendance.attendance_report'))
@@ -770,7 +767,7 @@ def save_manual_attendance():
return redirect(url_for('attendance.add_manual_attendance'))
# Get QR code (location)
qr_code = QRCode.query.get(int(location_id))
qr_code = db.session.get(QRCode, int(location_id))
if not qr_code:
flash('Selected location not found.', 'error')
return redirect(url_for('attendance.add_manual_attendance'))
@@ -1093,7 +1090,10 @@ def delete_attendance(record_id):
db.session.delete(attendance_record)
db.session.commit()
print(f"[LOG] {session.get('role', 'unknown').title()} {session.get('username')} deleted attendance record {record_id} for employee {employee_id}")
logger_handler.logger.info(
f"User {session.get('username')} ({session.get('role', 'unknown')}) "
f"deleted attendance record {record_id} for employee {employee_id}"
)
# Return JSON response for AJAX requests
if request.headers.get('X-Requested-With') == 'XMLHttpRequest':
@@ -1108,7 +1108,7 @@ def delete_attendance(record_id):
except Exception as e:
db.session.rollback()
logger_handler.log_database_error('attendance_delete', e)
print(f"[LOG] Error deleting attendance record {record_id}: {e}")
logger_handler.logger.error(f"Error deleting attendance record {record_id}: {e}", exc_info=True)
if request.headers.get('X-Requested-With') == 'XMLHttpRequest':
return jsonify({
@@ -1191,7 +1191,7 @@ def verification_review():
project_id = record.qr_code.project_id
if project_id not in project_names:
try:
project = Project.query.get(project_id)
project = db.session.get(Project, project_id)
if project:
project_names[project_id] = project.name
else:
@@ -1307,16 +1307,11 @@ def get_verification_details(record_id):
record = AttendanceData.query.get_or_404(record_id)
# DEBUG: Log record details
print(f"=== VERIFICATION DETAILS DEBUG ===")
print(f"Record ID: {record.id}")
print(f"Employee: {record.employee_id}")
print(f"check_in_date type: {type(record.check_in_date)}")
print(f"check_in_date value: {record.check_in_date}")
print(f"check_in_time type: {type(record.check_in_time)}")
print(f"check_in_time value: {record.check_in_time}")
print(f"verification_photo exists: {record.verification_photo is not None}")
print(f"verification_status: {record.verification_status}")
print(f"==================================")
logger_handler.logger.debug(
f"Verification details: record={record.id}, employee={record.employee_id}, "
f"date={record.check_in_date}, time={record.check_in_time}, "
f"has_photo={record.verification_photo is not None}, status={record.verification_status}"
)
# Check if user has permission to view
# Allow admin and payroll staff to view verification details
@@ -1333,13 +1328,13 @@ def get_verification_details(record_id):
try:
check_in_date_str = record.check_in_date.strftime('%Y-%m-%d') if record.check_in_date else 'N/A'
except Exception as e:
print(f"Error formatting check_in_date: {e}")
logger_handler.logger.warning(f"Error formatting check_in_date for record {record_id}: {e}")
check_in_date_str = str(record.check_in_date) if record.check_in_date else 'N/A'
try:
check_in_time_str = record.check_in_time.strftime('%I:%M %p') if record.check_in_time else 'N/A'
except Exception as e:
print(f"Error formatting check_in_time: {e}")
logger_handler.logger.warning(f"Error formatting check_in_time for record {record_id}: {e}")
check_in_time_str = str(record.check_in_time) if record.check_in_time else 'N/A'
# Prepare record data with safe formatting
@@ -1373,10 +1368,7 @@ def get_verification_details(record_id):
})
except Exception as e:
logger_handler.logger.error(f"Error getting verification details for record {record_id}: {e}")
print(f"❌ Error in get_verification_details for record {record_id}: {e}")
import traceback
print(f"❌ Traceback: {traceback.format_exc()}")
logger_handler.logger.error(f"Error in get_verification_details for record {record_id}: {e}", exc_info=True)
return jsonify({
'success': False,
@@ -1402,7 +1394,7 @@ def verification_review_detail(record_id):
return redirect(url_for('attendance.attendance_report'))
# Get the QR code information for additional context
qr_code = QRCode.query.get(record.qr_code_id) if record.qr_code_id else None
qr_code = db.session.get(QRCode, record.qr_code_id) if record.qr_code_id else None
# Get employee name from Employee table
employee_name = None
@@ -1497,7 +1489,7 @@ def attendance_stats_api():
})
except Exception as e:
print(f"Error fetching attendance stats: {e}")
logger_handler.logger.error(f"Error fetching attendance stats: {e}", exc_info=True)
return jsonify({'error': 'Failed to fetch attendance statistics'}), 500
@bp.route('/export-configuration', endpoint='export_configuration')
@@ -1515,8 +1507,8 @@ def export_configuration():
try:
logger_handler.logger.info(f"User {session.get('username', 'unknown')} (role: {user_role}) accessed export configuration")
logger_handler.logger.info(f"User {session.get('username', 'unknown')} accessed export configuration page")
except Exception as log_error:
print(f"⚠️ Logging error (non-critical): {log_error}")
except Exception:
pass
# Get current filters from session or request args
filters = {
@@ -1527,24 +1519,24 @@ def export_configuration():
'project_filter': request.args.get('project', '')
}
print(f"📊 Filters: {filters}")
logger_handler.logger.debug(f"Export config filters: {filters}")
# Get project name if project filter is applied
project_name = None
if filters.get('project_filter'):
try:
project = Project.query.get(int(filters['project_filter']))
project = db.session.get(Project, int(filters['project_filter']))
if project:
project_name = project.name
print(f"📊 Project filter: ID={filters['project_filter']}, Name={project_name}")
logger_handler.logger.debug(f"Project filter: ID={filters['project_filter']}, Name={project_name}")
except Exception as e:
print(f"⚠️ Error fetching project name: {e}")
logger_handler.logger.warning(f"Error fetching project name for filter: {e}")
# Check if location accuracy feature exists
try:
has_location_accuracy = check_location_accuracy_column_exists()
except Exception as e:
print(f"⚠️ Error checking location accuracy column: {e}")
logger_handler.logger.warning(f"Error checking location accuracy column: {e}")
has_location_accuracy = False
# Define all available columns with their default settings
@@ -1574,7 +1566,7 @@ def export_configuration():
'enabled': True # Changed from False to True
})
print(f"📊 Rendering export configuration with {len(available_columns)} columns")
logger_handler.logger.debug(f"Rendering export configuration with {len(available_columns)} columns")
return render_template('export_configuration.html',
available_columns=available_columns,
@@ -1583,8 +1575,7 @@ def export_configuration():
has_location_accuracy_feature=has_location_accuracy)
except Exception as e:
print(f"❌ Error in export_configuration route: {e}")
print(f"❌ Traceback: {traceback.format_exc()}")
logger_handler.logger.error(f"Error in export_configuration route: {e}", exc_info=True)
# Use your existing logger error method with correct parameters
try:
@@ -1594,7 +1585,7 @@ def export_configuration():
stack_trace=traceback.format_exc()
)
except Exception as log_error:
print(f"⚠️ Could not log error: {log_error}")
logger_handler.logger.warning(f"Could not log error: {log_error}")
flash('Error loading export configuration page.', 'error')
return redirect(url_for('attendance.attendance_report'))
@@ -1610,17 +1601,17 @@ def generate_excel_export():
flash('Access denied. Only administrators and payroll staff can export data.', 'error')
return redirect(url_for('attendance.attendance_report'))
print("📊 Excel export generation started")
logger_handler.logger.info(f"Excel export started by user {session.get('username', 'unknown')}")
# Log export action using your existing logger
try:
logger_handler.logger.info(f"User {session.get('username', 'unknown')} generated Excel export")
except Exception as log_error:
print(f"⚠️ Logging error (non-critical): {log_error}")
except Exception:
pass
# Get selected columns and custom names from form
selected_columns_raw = request.form.getlist('selected_columns')
print(f"📊 Selected columns (raw): {selected_columns_raw}")
logger_handler.logger.debug(f"Selected columns (raw): {selected_columns_raw}")
# Get column order from form
column_order_json = request.form.get('column_order', '[]')
@@ -1629,7 +1620,7 @@ def generate_excel_export():
except (json.JSONDecodeError, TypeError):
column_order = []
print(f"📊 Column order from form: {column_order}")
logger_handler.logger.debug(f"Column order from form: {column_order}")
# Determine final column order
if column_order:
@@ -1643,7 +1634,7 @@ def generate_excel_export():
# Fallback to raw selection order
selected_columns = selected_columns_raw
print(f"📊 Final column order: {selected_columns}")
logger_handler.logger.debug(f"Final column order: {selected_columns}")
if not selected_columns:
flash('Please select at least one column to export.', 'error')
@@ -1662,8 +1653,7 @@ def generate_excel_export():
'project_filter': request.form.get('project_filter')
}
print(f"📊 Export filters: {filters}")
print(f"📊 Column names: {column_names}")
logger_handler.logger.debug(f"Export filters: {filters}")
# Save user preferences in session for next time
session['export_preferences'] = {
@@ -1680,13 +1670,13 @@ def generate_excel_export():
project_name_for_filename = ''
if filters.get('project_filter'):
try:
project = Project.query.get(int(filters['project_filter']))
project = db.session.get(Project, int(filters['project_filter']))
if project:
# Replace spaces and special characters with underscores
project_name_safe = project.name.replace(' ', '_').replace('/', '_').replace('\\', '_')
project_name_for_filename = f"{project_name_safe}_"
except Exception as e:
print(f"⚠️ Error getting project name for filename: {e}")
logger_handler.logger.warning(f"Error getting project name for filename: {e}")
# Format dates for filename (MMDDYYYY format)
date_from_formatted = ''
@@ -1717,14 +1707,13 @@ def generate_excel_export():
filename = f'{project_name_for_filename}attendance_report_{date_range_str}.xlsx'
print(f"📊 Excel file generated successfully: {filename}")
print(f"📊 Column order in export: {selected_columns}")
logger_handler.logger.info(f"Excel export generated successfully: {filename}")
# Log successful export using your existing logger
try:
logger_handler.logger.info(f"Excel export generated successfully with {len(selected_columns)} columns in custom order by user {session.get('username', 'unknown')}: {filename}")
except Exception as log_error:
print(f"⚠️ Logging error (non-critical): {log_error}")
except Exception:
pass
return send_file(
excel_file,
@@ -1737,8 +1726,7 @@ def generate_excel_export():
return redirect(url_for('attendance.export_configuration'))
except Exception as e:
print(f"❌ Error in generate_excel_export route: {e}")
print(f"❌ Traceback: {traceback.format_exc()}")
logger_handler.logger.error(f"Error in generate_excel_export route: {e}", exc_info=True)
# Use your existing logger error method with correct parameters
try:
@@ -1748,7 +1736,7 @@ def generate_excel_export():
stack_trace=traceback.format_exc()
)
except Exception as log_error:
print(f"⚠️ Could not log error: {log_error}")
logger_handler.logger.warning(f"Could not log error: {log_error}")
flash('Error generating Excel export.', 'error')
return redirect(url_for('attendance.export_configuration'))
@@ -1756,7 +1744,7 @@ def generate_excel_export():
def create_excel_export(selected_columns, column_names, filters):
"""Create Excel file with selected attendance data - Updated to include employee names"""
try:
print(f"📊 Creating Excel export with {len(selected_columns)} columns")
logger_handler.logger.info(f"Creating Excel export with {len(selected_columns)} columns")
# Import openpyxl modules
try:
@@ -1764,8 +1752,7 @@ def create_excel_export(selected_columns, column_names, filters):
from openpyxl.styles import Font, Alignment, PatternFill
from openpyxl.utils import get_column_letter
except ImportError as e:
print(f"❌ openpyxl import error: {e}")
print("💡 Install openpyxl: pip install openpyxl")
logger_handler.logger.error(f"openpyxl import error: {e}. Run: pip install openpyxl")
return None
# Build query based on filters - JOIN with QRCode to get location_event and location_address
@@ -1781,22 +1768,22 @@ def create_excel_export(selected_columns, column_names, filters):
try:
date_from = datetime.strptime(filters['date_from'], '%Y-%m-%d').date()
query = query.filter(AttendanceData.check_in_date >= date_from)
print(f"📊 Applied date_from filter: {date_from}")
logger_handler.logger.debug(f"Applied date_from filter: {date_from}")
except ValueError as e:
print(f"⚠️ Invalid date_from format: {e}")
logger_handler.logger.warning(f"Invalid date_from format: {e}")
if filters.get('date_to'):
try:
date_to = datetime.strptime(filters['date_to'], '%Y-%m-%d').date()
query = query.filter(AttendanceData.check_in_date <= date_to)
print(f"📊 Applied date_to filter: {date_to}")
logger_handler.logger.debug(f"Applied date_to filter: {date_to}")
except ValueError as e:
print(f"⚠️ Invalid date_to format: {e}")
logger_handler.logger.warning(f"Invalid date_to format: {e}")
# Apply location filter
if filters.get('location_filter'):
query = query.filter(AttendanceData.location_name.like(f"%{filters['location_filter']}%"))
print(f"📊 Applied location filter: {filters['location_filter']}")
logger_handler.logger.debug(f"Applied location filter: {filters['location_filter']}")
# Apply employee filter — supports comma-separated multi-employee values
if filters.get('employee_filter'):
@@ -1805,26 +1792,26 @@ def create_excel_export(selected_columns, column_names, filters):
query = query.filter(AttendanceData.employee_id == emp_ids[0])
elif len(emp_ids) > 1:
query = query.filter(AttendanceData.employee_id.in_(emp_ids))
print(f"📊 Applied employee filter: {emp_ids}")
logger_handler.logger.debug(f"Applied employee filter: {emp_ids}")
# Apply project filter
if filters.get('project_filter'):
try:
project_id = int(filters['project_filter'])
query = query.filter(QRCode.project_id == project_id)
print(f"📊 Applied project filter: {project_id}")
logger_handler.logger.debug(f"Applied project filter: {project_id}")
except (ValueError, TypeError) as e:
print(f"⚠️ Invalid project filter: {e}")
logger_handler.logger.warning(f"Invalid project filter: {e}")
# Order by date and time
query = query.order_by(AttendanceData.check_in_date.desc(), AttendanceData.check_in_time.desc())
# Execute query
results = query.all()
print(f"📊 Query returned {len(results)} records")
logger_handler.logger.debug(f"Query returned {len(results)} records for export")
if not results:
print("⚠️ No records found for export")
logger_handler.logger.warning("No records found for export")
return None
# Create workbook
@@ -1890,10 +1877,10 @@ def create_excel_export(selected_columns, column_names, filters):
lng_formatted = f"{float(qr_record.address_longitude):.10f}"
hyperlink_formula = f'=HYPERLINK("http://maps.google.com/maps?q={lat_formatted},{lng_formatted}","{address_text.strip()}")'
cell.value = hyperlink_formula
print(f"📍 Added QR address hyperlink for employee {attendance_record.employee_id}")
logger_handler.logger.debug(f"Added QR address hyperlink for employee {attendance_record.employee_id}")
else:
cell.value = address_text
print(f"📍 Using QR address for employee {attendance_record.employee_id} (accuracy: {accuracy_value:.3f} miles)")
logger_handler.logger.debug(f"Using QR address for employee {attendance_record.employee_id} (accuracy: {accuracy_value:.3f} miles)")
else:
# Lower accuracy - use actual check-in address with hyperlink
address_text = attendance_record.address or ''
@@ -1903,10 +1890,10 @@ def create_excel_export(selected_columns, column_names, filters):
lng_formatted = f"{float(attendance_record.longitude):.10f}"
hyperlink_formula = f'=HYPERLINK("http://maps.google.com/maps?q={lat_formatted},{lng_formatted}","{address_text.strip()}")'
cell.value = hyperlink_formula
print(f"📍 Added check-in address hyperlink for employee {attendance_record.employee_id}")
logger_handler.logger.debug(f"Added check-in address hyperlink for employee {attendance_record.employee_id}")
else:
cell.value = address_text
print(f"📍 Using check-in address for employee {attendance_record.employee_id} (accuracy: {accuracy_value:.3f} miles)")
logger_handler.logger.debug(f"Using check-in address for employee {attendance_record.employee_id} (accuracy: {accuracy_value:.3f} miles)")
except (ValueError, TypeError):
# If accuracy can't be converted to float, use check-in address with hyperlink
address_text = attendance_record.address or ''
@@ -1916,7 +1903,7 @@ def create_excel_export(selected_columns, column_names, filters):
lng_formatted = f"{float(attendance_record.longitude):.10f}"
hyperlink_formula = f'=HYPERLINK("http://maps.google.com/maps?q={lat_formatted},{lng_formatted}","{address_text.strip()}")'
cell.value = hyperlink_formula
print(f"📍 Added check-in address hyperlink for employee {attendance_record.employee_id} (fallback)")
logger_handler.logger.debug(f"Added check-in address hyperlink (fallback) for employee {attendance_record.employee_id}")
else:
cell.value = address_text
else:
@@ -1928,7 +1915,7 @@ def create_excel_export(selected_columns, column_names, filters):
lng_formatted = f"{float(attendance_record.longitude):.10f}"
hyperlink_formula = f'=HYPERLINK("http://maps.google.com/maps?q={lat_formatted},{lng_formatted}","{address_text.strip()}")'
cell.value = hyperlink_formula
print(f"📍 Added check-in address hyperlink for employee {attendance_record.employee_id} (no accuracy data)")
logger_handler.logger.debug(f"Added check-in address hyperlink (no accuracy data) for employee {attendance_record.employee_id}")
else:
cell.value = address_text
elif column_key == 'device_info':
@@ -1948,7 +1935,7 @@ def create_excel_export(selected_columns, column_names, filters):
else:
cell.value = ''
except Exception as cell_error:
print(f"⚠️ Error setting cell value for {column_key}: {cell_error}")
logger_handler.logger.warning(f"Error setting cell value for {column_key}: {cell_error}")
cell.value = ''
# Auto-adjust column widths based on content and header
@@ -2007,26 +1994,25 @@ def create_excel_export(selected_columns, column_names, filters):
ws.column_dimensions[column_letter].width = adjusted_width
print(f"📏 Column {column_letter} ({column_key}): set width to {adjusted_width} (content: {max_length} chars)")
logger_handler.logger.debug(f"Column {column_letter} ({column_key}): width={adjusted_width} (max_content={max_length})")
# Save to BytesIO
excel_buffer = io.BytesIO()
wb.save(excel_buffer)
excel_buffer.seek(0)
print("📊 Excel file created successfully with employee names")
logger_handler.logger.info("Excel file created successfully with employee names")
# Log export action with employee name column
try:
logger_handler.logger.info(f"Excel export with employee names generated by user {session.get('username', 'unknown')}")
except Exception as log_error:
print(f"⚠️ Logging error (non-critical): {log_error}")
except Exception:
pass
return excel_buffer
except Exception as e:
print(f"❌ Error creating Excel export: {e}")
print(f"❌ Traceback: {traceback.format_exc()}")
logger_handler.logger.error(f"Error creating Excel export: {e}", exc_info=True)
# Log error
try:
@@ -2036,7 +2022,7 @@ def create_excel_export(selected_columns, column_names, filters):
stack_trace=traceback.format_exc()
)
except Exception as log_error:
print(f"⚠️ Could not log error: {log_error}")
logger_handler.logger.warning(f"Could not log error: {log_error}")
return None
@@ -2052,7 +2038,7 @@ def format_employee_id_for_excel(employee_id):
def create_excel_export_ordered(selected_columns, column_names, filters):
"""Create Excel file with selected attendance data in specified column order"""
try:
print(f"📊 Creating Excel export with {len(selected_columns)} columns in order: {selected_columns}")
logger_handler.logger.info(f"Creating Excel export with {len(selected_columns)} columns")
# Import openpyxl modules
try:
@@ -2060,8 +2046,7 @@ def create_excel_export_ordered(selected_columns, column_names, filters):
from openpyxl.styles import Font, Alignment, PatternFill
from openpyxl.utils import get_column_letter
except ImportError as e:
print(f"❌ openpyxl import error: {e}")
print("💡 Install openpyxl: pip install openpyxl")
logger_handler.logger.error(f"openpyxl import error: {e}. Run: pip install openpyxl")
return None
# Build query based on filters - JOIN with QRCode to get location_event and location_address
@@ -2077,22 +2062,22 @@ def create_excel_export_ordered(selected_columns, column_names, filters):
try:
date_from = datetime.strptime(filters['date_from'], '%Y-%m-%d').date()
query = query.filter(AttendanceData.check_in_date >= date_from)
print(f"📊 Applied date_from filter: {date_from}")
logger_handler.logger.debug(f"Applied date_from filter: {date_from}")
except ValueError as e:
print(f"⚠️ Invalid date_from format: {e}")
logger_handler.logger.warning(f"Invalid date_from format: {e}")
if filters.get('date_to'):
try:
date_to = datetime.strptime(filters['date_to'], '%Y-%m-%d').date()
query = query.filter(AttendanceData.check_in_date <= date_to)
print(f"📊 Applied date_to filter: {date_to}")
logger_handler.logger.debug(f"Applied date_to filter: {date_to}")
except ValueError as e:
print(f"⚠️ Invalid date_to format: {e}")
logger_handler.logger.warning(f"Invalid date_to format: {e}")
# Apply location filter
if filters.get('location_filter'):
query = query.filter(AttendanceData.location_name.like(f"%{filters['location_filter']}%"))
print(f"📊 Applied location filter: {filters['location_filter']}")
logger_handler.logger.debug(f"Applied location filter: {filters['location_filter']}")
# Apply employee filter — supports comma-separated multi-employee values
if filters.get('employee_filter'):
@@ -2101,26 +2086,26 @@ def create_excel_export_ordered(selected_columns, column_names, filters):
query = query.filter(AttendanceData.employee_id == emp_ids[0])
elif len(emp_ids) > 1:
query = query.filter(AttendanceData.employee_id.in_(emp_ids))
print(f"📊 Applied employee filter: {emp_ids}")
logger_handler.logger.debug(f"Applied employee filter: {emp_ids}")
# Apply project filter
if filters.get('project_filter'):
try:
project_id = int(filters['project_filter'])
query = query.filter(QRCode.project_id == project_id)
print(f"📊 Applied project filter: {project_id}")
logger_handler.logger.debug(f"Applied project filter: {project_id}")
except (ValueError, TypeError) as e:
print(f"⚠️ Invalid project filter: {e}")
logger_handler.logger.warning(f"Invalid project filter: {e}")
# Order by date and time
query = query.order_by(AttendanceData.check_in_date.desc(), AttendanceData.check_in_time.desc())
# Execute query
results = query.all()
print(f"📊 Query returned {len(results)} records")
logger_handler.logger.debug(f"Query returned {len(results)} records for export")
if not results:
print("⚠️ No records found for export")
logger_handler.logger.warning("No records found for export")
return None
# Create workbook
@@ -2192,10 +2177,10 @@ def create_excel_export_ordered(selected_columns, column_names, filters):
lng_formatted = f"{float(qr_record.address_longitude):.10f}"
hyperlink_formula = f'=HYPERLINK("http://maps.google.com/maps?q={lat_formatted},{lng_formatted}","{address_text.strip()}")'
cell.value = hyperlink_formula
print(f"📍 Added QR address hyperlink for employee {attendance_record.employee_id}")
logger_handler.logger.debug(f"Added QR address hyperlink for employee {attendance_record.employee_id}")
else:
cell.value = address_text
print(f"📍 Using QR address for employee {attendance_record.employee_id} (accuracy: {accuracy_value:.3f} miles)")
logger_handler.logger.debug(f"Using QR address for employee {attendance_record.employee_id} (accuracy: {accuracy_value:.3f} miles)")
else:
# Lower accuracy - use actual check-in address with hyperlink
address_text = attendance_record.address or ''
@@ -2205,10 +2190,10 @@ def create_excel_export_ordered(selected_columns, column_names, filters):
lng_formatted = f"{float(attendance_record.longitude):.10f}"
hyperlink_formula = f'=HYPERLINK("http://maps.google.com/maps?q={lat_formatted},{lng_formatted}","{address_text.strip()}")'
cell.value = hyperlink_formula
print(f"📍 Added check-in address hyperlink for employee {attendance_record.employee_id}")
logger_handler.logger.debug(f"Added check-in address hyperlink for employee {attendance_record.employee_id}")
else:
cell.value = address_text
print(f"📍 Using check-in address for employee {attendance_record.employee_id} (accuracy: {accuracy_value:.3f} miles)")
logger_handler.logger.debug(f"Using check-in address for employee {attendance_record.employee_id} (accuracy: {accuracy_value:.3f} miles)")
except (ValueError, TypeError):
# If accuracy can't be converted to float, use check-in address with hyperlink
address_text = attendance_record.address or ''
@@ -2218,7 +2203,7 @@ def create_excel_export_ordered(selected_columns, column_names, filters):
lng_formatted = f"{float(attendance_record.longitude):.10f}"
hyperlink_formula = f'=HYPERLINK("http://maps.google.com/maps?q={lat_formatted},{lng_formatted}","{address_text.strip()}")'
cell.value = hyperlink_formula
print(f"📍 Added check-in address hyperlink for employee {attendance_record.employee_id} (fallback)")
logger_handler.logger.debug(f"Added check-in address hyperlink (fallback) for employee {attendance_record.employee_id}")
else:
cell.value = address_text
else:
@@ -2230,7 +2215,7 @@ def create_excel_export_ordered(selected_columns, column_names, filters):
lng_formatted = f"{float(attendance_record.longitude):.10f}"
hyperlink_formula = f'=HYPERLINK("http://maps.google.com/maps?q={lat_formatted},{lng_formatted}","{address_text.strip()}")'
cell.value = hyperlink_formula
print(f"📍 Added check-in address hyperlink for employee {attendance_record.employee_id} (no accuracy data)")
logger_handler.logger.debug(f"Added check-in address hyperlink (no accuracy data) for employee {attendance_record.employee_id}")
else:
cell.value = address_text
elif column_key == 'device_info':
@@ -2259,7 +2244,7 @@ def create_excel_export_ordered(selected_columns, column_names, filters):
else:
cell.value = ''
except Exception as cell_error:
print(f"⚠️ Error setting cell value for {column_key}: {cell_error}")
logger_handler.logger.warning(f"Error setting cell value for {column_key}: {cell_error}")
cell.value = ''
# Auto-adjust column widths based on content and header
@@ -2318,26 +2303,25 @@ def create_excel_export_ordered(selected_columns, column_names, filters):
ws.column_dimensions[column_letter].width = adjusted_width
print(f"📏 Column {column_letter} ({column_key}): set width to {adjusted_width} (content: {max_length} chars)")
logger_handler.logger.debug(f"Column {column_letter} ({column_key}): width={adjusted_width} (max_content={max_length})")
# Save to BytesIO
excel_buffer = io.BytesIO()
wb.save(excel_buffer)
excel_buffer.seek(0)
print("📊 Excel file created successfully with employee names and verification status coloring")
logger_handler.logger.info("Excel file created successfully with employee names and verification status coloring")
# Log export action with employee name column and verification status coloring
try:
logger_handler.logger.info(f"Excel export with employee names and verification status coloring generated by user {session.get('username', 'unknown')}")
except Exception as log_error:
print(f"⚠️ Logging error (non-critical): {log_error}")
except Exception:
pass
return excel_buffer
except Exception as e:
print(f"❌ Error creating Excel export: {e}")
print(f"❌ Traceback: {traceback.format_exc()}")
logger_handler.logger.error(f"Error creating Excel export: {e}", exc_info=True)
# Log error
try:
@@ -2347,6 +2331,6 @@ def create_excel_export_ordered(selected_columns, column_names, filters):
stack_trace=traceback.format_exc()
)
except Exception as log_error:
print(f"⚠️ Could not log error: {log_error}")
logger_handler.logger.warning(f"Could not log error: {log_error}")
return None
routes/auth.py
+3 -1
View File
@@ -161,6 +161,7 @@ def login():
logger_handler.logger.warning(f"Failed login attempt for username: {username}")
except Exception as e:
db.session.rollback()
logger_handler.log_database_error('user_login', e)
logger_handler.logger.error(f"Login error for username '{username}': {e}")
flash('Login error. Please try again.', 'error')
@@ -201,7 +202,7 @@ def logout():
def profile():
"""User profile management with logging"""
try:
user = User.query.get(session['user_id'])
user = db.session.get(User, session['user_id'])
if request.method == 'POST':
form_type = request.form.get('form_type')
@@ -259,6 +260,7 @@ def profile():
return render_template('profile.html', user=user)
except Exception as e:
db.session.rollback()
logger_handler.log_database_error('profile_update', e)
flash('Profile update failed. Please try again.', 'error')
return redirect(url_for('dashboard.dashboard'))
routes/dashboard.py
+1 -3
View File
@@ -26,7 +26,7 @@ bp = Blueprint('dashboard', __name__)
def dashboard():
"""Enhanced project-centric dashboard with search filters"""
try:
user = User.query.get(session['user_id'])
user = db.session.get(User, session['user_id'])
# Get search parameters from URL
search_name = request.args.get('search_name', '').strip()
@@ -71,7 +71,6 @@ def dashboard():
except Exception as e:
logger_handler.log_database_error('dashboard_load', e)
print(f"Error loading dashboard: {e}")
flash('Error loading dashboard. Please try again.', 'error')
return redirect(url_for('auth.login'))
@@ -127,7 +126,6 @@ def project_qr_codes(project_id):
except Exception as e:
logger_handler.log_database_error('project_qr_codes_view', e)
print(f"Error loading project QR codes: {e}")
flash('Error loading project QR codes. Please try again.', 'error')
return redirect(url_for('dashboard.dashboard'))
routes/employees.py
+2 -2
View File
@@ -134,7 +134,7 @@ def create_employee():
db.session.commit()
# Log employee creation with project info
project = Project.query.get(contract_id_int)
project = db.session.get(Project, contract_id_int)
project_name = project.name if project else f"Project {contract_id_int}"
logger_handler.logger.info(
f"User {session['username']} created new employee: "
@@ -213,7 +213,7 @@ def edit_employee(employee_index):
db.session.commit()
# Log employee update with project info
project = Project.query.get(contract_id_int)
project = db.session.get(Project, contract_id_int)
project_name = project.name if project else f"Project {contract_id_int}"
logger_handler.logger.info(
f"User {session['username']} updated employee: "
routes/payroll.py
+36 -48
View File
@@ -44,7 +44,7 @@ def payroll_dashboard():
flash('Access denied. Only administrators and payroll staff can access payroll features.', 'error')
return redirect(url_for('dashboard.dashboard'))
print("📊 Loading payroll dashboard")
logger_handler.logger.debug("Loading payroll dashboard")
# Log payroll dashboard access
logger_handler.logger.info(f"User {session.get('username', 'unknown')} accessed payroll dashboard")
@@ -65,9 +65,9 @@ def payroll_dashboard():
projects = []
try:
projects = Project.query.filter_by(active_status=True).order_by(Project.name).all()
print(f"📊 Found {len(projects)} active projects for filter")
logger_handler.logger.debug(f"Found {len(projects)} active projects for payroll filter")
except Exception as e:
print(f"⚠️ Error loading projects: {e}")
logger_handler.logger.warning(f"Error loading projects for payroll filter: {e}")
# Get attendance records for the period
attendance_records = []
@@ -90,12 +90,12 @@ def payroll_dashboard():
# Apply project filter if selected
if project_filter and project_filter != '':
query = query.filter(QRCode.project_id == int(project_filter))
print(f"📊 Applied project filter: {project_filter}")
logger_handler.logger.debug(f"Applied project filter: {project_filter}")
query = query.order_by(AttendanceData.employee_id, AttendanceData.check_in_date, AttendanceData.check_in_time)
attendance_records = query.all()
print(f"📊 Found {len(attendance_records)} attendance records for payroll calculation")
logger_handler.logger.debug(f"Found {len(attendance_records)} attendance records for payroll calculation")
# Calculate working hours if we have records
if attendance_records:
@@ -103,13 +103,13 @@ def payroll_dashboard():
working_hours_data = calculator.calculate_all_employees_hours(
start_date, end_date, attendance_records
)
print(f"📊 Calculated hours for {working_hours_data['employee_count']} employees")
logger_handler.logger.debug(f"Calculated hours for {working_hours_data['employee_count']} employees")
except ValueError as e:
print(f"⚠️ Invalid date format: {e}")
logger_handler.logger.warning(f"Invalid date format in payroll dashboard: {e}")
flash('Invalid date format. Please use YYYY-MM-DD format.', 'error')
except Exception as e:
print(f"❌ Error calculating working hours: {e}")
logger_handler.logger.error(f"Error calculating working hours: {e}", exc_info=True)
logger_handler.log_database_error('payroll_calculation', e)
flash('Error calculating working hours. Please check the server logs.', 'error')
@@ -136,23 +136,21 @@ def payroll_dashboard():
if row[1]: # Only add if we got a name
employee_names[str(row[0])] = row[1]
print(f"📊 Retrieved names for {len(employee_names)} employees using CAST method")
logger_handler.logger.debug(f"Retrieved names for {len(employee_names)} employees")
except Exception as e:
print(f"⚠️ Could not load employee names: {e}")
import traceback
print(f"⚠️ Traceback: {traceback.format_exc()}")
logger_handler.logger.warning(f"Could not load employee names: {e}", exc_info=True)
# Continue without names - will use employee IDs
# Get selected project name for display
selected_project_name = ''
if project_filter:
try:
selected_project = Project.query.get(int(project_filter))
selected_project = db.session.get(Project, int(project_filter))
if selected_project:
selected_project_name = selected_project.name
except Exception as e:
print(f"⚠️ Error getting selected project name: {e}")
logger_handler.logger.warning(f"Error getting selected project name: {e}")
return render_template('payroll_dashboard.html',
working_hours_data=working_hours_data,
@@ -165,9 +163,7 @@ def payroll_dashboard():
user_role=user_role)
except Exception as e:
print(f"❌ Error loading payroll dashboard: {e}")
import traceback
print(f"❌ Traceback: {traceback.format_exc()}")
logger_handler.logger.error(f"Error loading payroll dashboard: {e}", exc_info=True)
logger_handler.log_flask_error(
'payroll_dashboard_error',
@@ -191,7 +187,7 @@ def export_payroll_excel():
flash('Access denied. Only administrators and payroll staff can export payroll data.', 'error')
return redirect(url_for('payroll.payroll_dashboard'))
print("📊 Payroll Excel export started")
logger_handler.logger.info(f"Payroll Excel export started by user {session.get('username', 'unknown')}")
# Get parameters from form
date_from = request.form.get('date_from')
@@ -222,7 +218,7 @@ def export_payroll_excel():
# Apply project filter if selected
if project_filter and project_filter != '':
query = query.filter(QRCode.project_id == int(project_filter))
print(f"📊 Applied project filter to export: {project_filter}")
logger_handler.logger.debug(f"Applied project filter to export: {project_filter}")
query = query.order_by(AttendanceData.employee_id, AttendanceData.check_in_date, AttendanceData.check_in_time)
@@ -235,13 +231,13 @@ def export_payroll_excel():
attendance_data.qr_code = qr_code
attendance_records.append(attendance_data)
print(f"📊 Export: Found {len(attendance_records)} records with QR data")
logger_handler.logger.info(f"Payroll export: found {len(attendance_records)} records with QR data")
if not attendance_records:
flash('No attendance records found for the selected date range and project.', 'warning')
return redirect(url_for('payroll.payroll_dashboard'))
print(f"📊 Exporting {len(attendance_records)} attendance records to Excel")
logger_handler.logger.info(f"Exporting {len(attendance_records)} attendance records to payroll Excel")
# Get employee names using the same method as dashboard
employee_names = {}
@@ -264,24 +260,22 @@ def export_payroll_excel():
if row[1]: # Only add if we got a name
employee_names[str(row[0])] = row[1]
print(f"📊 Retrieved names for {len(employee_names)} employees for export using CAST method")
logger_handler.logger.debug(f"Retrieved names for {len(employee_names)} employees for export")
except Exception as e:
print(f"⚠️ Could not load employee names for export: {e}")
import traceback
print(f"⚠️ Traceback: {traceback.format_exc()}")
logger_handler.logger.warning(f"Could not load employee names for export: {e}", exc_info=True)
# Get project name for enhanced reports and filename
project_name = None
project_name_for_filename = ''
if project_filter:
try:
project = Project.query.get(int(project_filter))
project = db.session.get(Project, int(project_filter))
if project:
project_name = project.name
project_name_for_filename = f"_{project.name.replace(' ', '_')}"
except Exception as e:
print(f"⚠️ Error getting project name: {e}")
logger_handler.logger.warning(f"Error getting project name for export: {e}")
# Generate Excel file based on report type
excel_file = None
@@ -289,7 +283,7 @@ def export_payroll_excel():
if report_type == 'enhanced':
# Use enhanced exporter for SP/PW reports
print("📊 Creating enhanced payroll report with SP/PW support")
logger_handler.logger.debug("Creating enhanced payroll report with SP/PW support")
try:
from enhanced_payroll_excel_exporter import EnhancedPayrollExcelExporter
exporter = EnhancedPayrollExcelExporter(company_name=current_app.config.get('COMPANY_NAME', 'QR Code Management System'))
@@ -297,9 +291,9 @@ def export_payroll_excel():
start_date, end_date, attendance_records, employee_names, project_name
)
filename_prefix = 'enhanced_payroll_report'
print("✅ Enhanced payroll report created successfully")
logger_handler.logger.info("Enhanced payroll report created successfully")
except ImportError:
print("⚠️ Enhanced exporter not available, falling back to standard exporter")
logger_handler.logger.warning("Enhanced exporter not available, falling back to standard exporter")
# Fall back to standard exporter
exporter = PayrollExcelExporter(
company_name=current_app.config.get('COMPANY_NAME', 'QR Code Management System'),
@@ -310,7 +304,7 @@ def export_payroll_excel():
)
filename_prefix = 'payroll_report'
except Exception as e:
print(f"⚠️ Error with enhanced exporter: {e}, falling back to standard exporter")
logger_handler.logger.warning(f"Enhanced exporter error: {e} — falling back to standard exporter")
# Fall back to standard exporter
exporter = PayrollExcelExporter(
company_name=current_app.config.get('COMPANY_NAME', 'QR Code Management System'),
@@ -323,7 +317,7 @@ def export_payroll_excel():
elif report_type == 'detailed_sp_pw':
# Detailed daily SP/PW breakdown
print("📊 Creating detailed SP/PW daily breakdown report")
logger_handler.logger.debug("Creating detailed SP/PW daily breakdown report")
try:
from enhanced_payroll_excel_exporter import EnhancedPayrollExcelExporter
exporter = EnhancedPayrollExcelExporter(company_name=current_app.config.get('COMPANY_NAME', 'QR Code Management System'))
@@ -331,9 +325,9 @@ def export_payroll_excel():
start_date, end_date, attendance_records, employee_names
)
filename_prefix = 'detailed_sp_pw_report'
print("✅ Detailed SP/PW report created successfully")
logger_handler.logger.info("Detailed SP/PW report created successfully")
except ImportError:
print("⚠️ Enhanced exporter not available, falling back to detailed hours report")
logger_handler.logger.warning("Enhanced exporter not available, falling back to detailed hours report")
# Fall back to standard detailed report
exporter = PayrollExcelExporter(
company_name=current_app.config.get('COMPANY_NAME', 'QR Code Management System'),
@@ -344,7 +338,7 @@ def export_payroll_excel():
)
filename_prefix = 'detailed_hours_report'
except Exception as e:
print(f"⚠️ Error with enhanced exporter: {e}, falling back to detailed hours report")
logger_handler.logger.warning(f"Enhanced exporter error: {e} — falling back to detailed hours report")
# Fall back to standard detailed report
exporter = PayrollExcelExporter(
company_name=current_app.config.get('COMPANY_NAME', 'QR Code Management System'),
@@ -384,7 +378,7 @@ def export_payroll_excel():
timestamp = datetime.now().strftime('%Y%m%d_%H%M%S')
filename = f'{filename_prefix}_{date_from}_to_{date_to}{project_name_for_filename}_{timestamp}.xlsx'
print(f"📊 Payroll Excel file generated successfully: {filename}")
logger_handler.logger.info(f"Payroll Excel file generated successfully: {filename}")
# Log successful export
logger_handler.logger.info(f"Payroll Excel export generated by user {session.get('username', 'unknown')}: {filename}")
@@ -406,9 +400,7 @@ def export_payroll_excel():
return redirect(url_for('payroll.payroll_dashboard'))
except Exception as e:
print(f"❌ Error in export_payroll_excel route: {e}")
import traceback
print(f"❌ Traceback: {traceback.format_exc()}")
logger_handler.logger.error(f"Error in export_payroll_excel route: {e}", exc_info=True)
logger_handler.log_flask_error(
'payroll_excel_export_error',
@@ -484,7 +476,7 @@ def calculate_working_hours_api():
})
except Exception as e:
print(f"❌ Error in calculate_working_hours_api: {e}")
logger_handler.logger.error(f"Error in calculate_working_hours_api: {e}", exc_info=True)
logger_handler.log_flask_error(
'working_hours_api_error',
str(e),
@@ -541,11 +533,9 @@ def get_miss_punch_details(employee_id):
employee_row = employee_query.fetchone()
employee_name = employee_row.full_name if employee_row and employee_row.full_name else f"Employee {employee_id}"
print(f"📋 Retrieved employee name: {employee_name} for ID: {employee_id}")
logger_handler.logger.debug(f"Retrieved employee name for ID {employee_id}: {employee_name}")
except Exception as e:
print(f"⚠️ Could not load employee name for ID {employee_id}: {e}")
import traceback
print(f"⚠️ Traceback: {traceback.format_exc()}")
logger_handler.logger.warning(f"Could not load employee name for ID {employee_id}: {e}", exc_info=True)
employee_name = f"Employee {employee_id}"
# Get attendance records for the employee within the period
@@ -647,9 +637,7 @@ def get_miss_punch_details(employee_id):
})
except Exception as e:
print(f"❌ Error in get_miss_punch_details: {e}")
import traceback
print(f"❌ Traceback: {traceback.format_exc()}")
logger_handler.logger.error(f"Error in get_miss_punch_details: {e}", exc_info=True)
logger_handler.log_flask_error(
'miss_punch_details_api_error',
@@ -675,7 +663,7 @@ def get_employee_name(employee_id):
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}")
logger_handler.logger.warning(f"Error getting employee name for ID {employee_id}: {e}")
return f"Employee {employee_id}"
def get_qr_code_checkin_count(qr_code_id):
routes/qr_codes.py
+62 -99
View File
@@ -86,9 +86,9 @@ def create_qr_code():
address_latitude = float(latitude)
address_longitude = float(longitude)
has_coordinates = True
print(f"✓ Coordinates received: {address_latitude}, {address_longitude}")
logger_handler.logger.debug(f"Coordinates received: {address_latitude}, {address_longitude}")
except (ValueError, TypeError) as e:
print(f"⚠️ Invalid coordinates format: {e}")
logger_handler.logger.warning(f"Invalid coordinates format: {e}")
address_latitude = None
address_longitude = None
has_coordinates = False
@@ -98,7 +98,7 @@ def create_qr_code():
if project_id:
try:
project_id = int(project_id)
project = Project.query.get(project_id)
project = db.session.get(Project, project_id)
if not project or not project.active_status:
flash('Selected project is not valid or inactive.', 'error')
return render_template('create_qr_code.html',
@@ -192,7 +192,7 @@ def create_qr_code():
db.session.rollback()
logger_handler.log_database_error('qr_code_creation', e)
flash('QR Code creation failed. Please try again.', 'error')
print(f"❌ QR Code creation error: {e}")
logger_handler.logger.error(f"QR Code creation error: {e}", exc_info=True)
# Get active projects and styles for dropdown
projects = Project.query.filter_by(active_status=True).order_by(Project.name.asc()).all()
@@ -435,7 +435,7 @@ def edit_qr_code(qr_id):
if new_project_id and new_project_id.strip():
try:
new_project_id = int(new_project_id)
project = Project.query.get(new_project_id)
project = db.session.get(Project, new_project_id)
if project and project.active_status:
qr_code.project_id = new_project_id
else:
@@ -518,16 +518,16 @@ def delete_qr_code(qr_id):
"""Permanently delete QR code (Admin only) - Hard delete - PRESERVING EXACT ROUTE"""
try:
qr_code = QRCode.query.get_or_404(qr_id)
print(f"✅ Found QR Code: {qr_code.name}")
logger_handler.logger.debug(f"Found QR Code for delete: {qr_code.name} (ID: {qr_id})")
if request.method == 'POST':
qr_name = qr_code.name
qr_code_id = qr_code.id
print(f"🗑️ ATTEMPTING TO DELETE: {qr_name}")
logger_handler.logger.info(f"User {session.get('username', 'unknown')} attempting to delete QR code: {qr_name} (ID: {qr_id})")
# Check if QR exists before delete
before_count = QRCode.query.count()
print(f"📊 QR count before delete: {before_count}")
logger_handler.logger.debug(f"QR count before delete: {before_count}")
# Log QR code deletion before actual deletion
logger_handler.log_qr_code_deleted(
@@ -538,29 +538,27 @@ def delete_qr_code(qr_id):
# 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")
logger_handler.logger.debug(f"QR count after delete: {after_count}")
logger_handler.logger.info(f"QR code deleted successfully: {qr_name} (ID: {qr_id}), removed {before_count - after_count} records")
flash(f'QR code "{qr_name}" has been permanently deleted!', 'success')
return redirect(url_for('dashboard.dashboard'))
# GET request - show confirmation page
print("📄 Showing confirmation page")
logger_handler.logger.debug(f"Showing delete confirmation page for QR code ID: {qr_id}")
return render_template('confirm_delete_qr.html', qr_code=qr_code)
except Exception as e:
db.session.rollback()
logger_handler.log_database_error('qr_code_deletion', e)
print(f"❌ ERROR in delete route: {e}")
print(f"❌ Exception type: {type(e)}")
print(f"❌ Traceback: {traceback.format_exc()}")
logger_handler.logger.error(f"Error in QR code delete route (ID: {qr_id}): {e}", exc_info=True)
flash('Error deleting QR code. Please try again.', 'error')
return redirect(url_for('dashboard.dashboard'))
@@ -603,23 +601,19 @@ def qr_checkin(qr_url):
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()}")
logger_handler.logger.debug(f"Starting check-in process for QR URL: {qr_url}")
# 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}")
logger_handler.logger.warning(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}")
logger_handler.logger.debug(f"Found QR code: {qr_code.name} (ID: {qr_code.id}), location: {qr_code.location}")
# Get and validate employee ID
employee_id = request.form.get('employee_id', '').strip()
@@ -650,9 +644,7 @@ def qr_checkin(qr_url):
# 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 {qr_code.location_event} for {employee_id}")
print(f" Last {qr_code.location_event}: {recent_checkin.check_in_time.strftime('%H:%M')}")
print(f" Minutes remaining: {minutes_remaining}")
logger_handler.logger.info(f"Too soon for another {qr_code.location_event} for employee {employee_id}: {minutes_remaining} minutes remaining")
return jsonify({
'success': False,
@@ -660,9 +652,9 @@ def qr_checkin(qr_url):
f"Puedes volver a registrarte en {minutes_remaining} minutos. El último registro fue a las {recent_checkin.check_in_time.strftime("%H:%M")}."
}), 400
else:
print(f"✅ {time_interval}-minute interval satisfied. Allowing new {qr_code.location_event} for {employee_id}")
logger_handler.logger.debug(f"{time_interval}-minute interval satisfied for employee {employee_id}")
else:
print(f"✅ First {qr_code.location_event} today for {employee_id}")
logger_handler.logger.debug(f"First {qr_code.location_event} today for employee {employee_id}")
# Process location data with coordinate-to-address conversion
location_data = process_location_data_enhanced(request.form)
@@ -672,12 +664,10 @@ def qr_checkin(qr_url):
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}")
logger_handler.logger.debug(f"Check-in device: {device_info}, IP: {client_ip}")
# Create attendance record
print(f"\n💾 CREATING ATTENDANCE RECORD:")
logger_handler.logger.debug("Creating attendance record")
attendance = AttendanceData(
qr_code_id=qr_code.id,
@@ -699,39 +689,29 @@ def qr_checkin(qr_url):
verification_status=None
)
print(f"✅ Created base attendance record")
logger_handler.logger.debug("Created base attendance record")
# ENHANCED DEBUG: Calculate location accuracy with detailed logging
print(f"\n🎯 CALCULATING LOCATION ACCURACY WITH ENHANCED DEBUG...")
print(f" 📊 QR Code Details:")
print(f" ID: {qr_code.id}")
print(f" Name: {qr_code.name}")
print(f" Location: {qr_code.location}")
print(f" Location Address: {qr_code.location_address}")
print(f" Has location_address: {qr_code.location_address is not None}")
print(f" Location Address Length: {len(qr_code.location_address) if qr_code.location_address else 0}")
print(f" 📍 Check-in Data:")
print(f" Latitude: {location_data['latitude']}")
print(f" Longitude: {location_data['longitude']}")
print(f" GPS Accuracy: {location_data['accuracy']}")
print(f" Address: {location_data['address']}")
print(f" Address Length: {len(location_data['address']) if location_data['address'] else 0}")
print(f" Source: {location_data['source']}")
# Calculate location accuracy
logger_handler.logger.debug(
f"Location accuracy check: QR='{qr_code.name}' (ID={qr_code.id}), "
f"lat={location_data['latitude']}, lng={location_data['longitude']}, "
f"source={location_data['source']}"
)
location_accuracy = None
try:
# Check if we have the required data
if not qr_code.location_address:
print(f"❌ QR code location_address is empty or None")
print(f" QR Code location_address value: '{qr_code.location_address}'")
logger_handler.logger.warning(f"QR code location_address is empty or None for QR ID: {qr_code.id}")
elif not location_data['address'] and not (location_data['latitude'] and location_data['longitude']):
print(f"❌ No check-in address or coordinates available")
print(f" Check-in address: '{location_data['address']}'")
print(f" Check-in coords: {location_data['latitude']}, {location_data['longitude']}")
logger_handler.logger.warning(
f"No check-in address or coordinates available: "
f"address={location_data['address']!r}, "
f"coords={location_data['latitude']}, {location_data['longitude']}"
)
else:
print(f"✅ Required data available, proceeding with calculation...")
logger_handler.logger.debug("Required location data available, proceeding with accuracy calculation")
location_accuracy = calculate_location_accuracy_enhanced(
qr_address=qr_code.location_address,
@@ -740,30 +720,28 @@ def qr_checkin(qr_url):
checkin_lng=location_data['longitude']
)
print(f"📐 Location accuracy calculation result: {location_accuracy}")
logger_handler.logger.debug(f"Location accuracy calculation result: {location_accuracy}")
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 successfully: {location_accuracy:.4f} miles ({accuracy_level})")
print(f"📊 Final attendance.location_accuracy value: {attendance.location_accuracy}")
logger_handler.logger.debug(f"Location accuracy set: {location_accuracy:.4f} miles ({accuracy_level})")
else:
print(f"⚠️ Could not calculate location accuracy - calculation returned None")
logger_handler.logger.warning("Could not calculate location accuracy — calculation returned None")
# CHECK DISTANCE THRESHOLD FOR PHOTO VERIFICATION
print(f"\n📸 CHECKING PHOTO VERIFICATION REQUIREMENT:")
print(f" Photo Verification Enabled: {current_app.config.get('PHOTO_VERIFICATION_ENABLED', True)}")
logger_handler.logger.debug(f"Photo verification enabled: {current_app.config.get('PHOTO_VERIFICATION_ENABLED', True)}")
requires_verification = False
verification_photo_data = None
if current_app.config.get('PHOTO_VERIFICATION_ENABLED', True) and location_accuracy is not None and location_accuracy > current_app.config.get('DISTANCE_THRESHOLD_FOR_VERIFICATION', 0.3):
print(f"⚠️ Distance ({location_accuracy:.3f} mi) exceeds threshold ({current_app.config.get('DISTANCE_THRESHOLD_FOR_VERIFICATION', 0.3)} mi)")
logger_handler.logger.info(f"Distance ({location_accuracy:.3f} mi) exceeds verification threshold for employee {employee_id}")
# Check if photo was provided
verification_photo_data = request.form.get('verification_photo', None)
if verification_photo_data:
print(f"✅ Verification photo provided (size: {len(verification_photo_data)} chars)")
logger_handler.logger.debug(f"Verification photo provided (size: {len(verification_photo_data)} chars)")
# Validate photo data (basic validation)
if verification_photo_data.startswith('data:image/'):
@@ -771,16 +749,16 @@ def qr_checkin(qr_url):
attendance.verification_required = True
attendance.verification_status = 'pending'
attendance.verification_timestamp = datetime.now()
print(f"✅ Photo verification set to PENDING status")
logger_handler.logger.info(f"Photo verification set to PENDING for employee {employee_id}")
else:
print(f"⚠️ Invalid photo format provided")
logger_handler.logger.warning(f"Invalid photo format provided for employee {employee_id}")
return jsonify({
'success': False,
'message': 'Invalid photo format. Please try again.',
'requires_verification': True
}), 400
else:
print(f"❌ Photo verification REQUIRED but not provided")
logger_handler.logger.warning(f"Photo verification required but not provided for employee {employee_id}")
return jsonify({
'success': False,
'message': 'Photo verification required. Distance from location is too far.',
@@ -789,22 +767,14 @@ def qr_checkin(qr_url):
'threshold': current_app.config.get('DISTANCE_THRESHOLD_FOR_VERIFICATION', 0.3)
}), 400
else:
print(f"✅ Distance within threshold - no verification needed")
logger_handler.logger.debug(f"Distance within threshold for employee {employee_id} — no verification needed")
except Exception as e:
print(f"❌ Error in location accuracy calculation: {e}")
print(f"❌ Full traceback: {traceback.format_exc()}")
logger_handler.logger.error(f"Error in location accuracy calculation: {e}", exc_info=True)
# ENHANCED DEBUG: Save to database with verification
try:
print(f"\n💾 SAVING TO DATABASE...")
print(f" Attendance object before save:")
print(f" Employee ID: {attendance.employee_id}")
print(f" Location: {attendance.location_name}")
print(f" Latitude: {attendance.latitude}")
print(f" Longitude: {attendance.longitude}")
print(f" Address: {attendance.address}")
print(f" Location Accuracy: {attendance.location_accuracy}")
logger_handler.logger.debug(f"Saving attendance record: employee={attendance.employee_id}, location={attendance.location_name}, accuracy={attendance.location_accuracy}")
db.session.add(attendance)
db.session.commit()
@@ -819,14 +789,11 @@ def qr_checkin(qr_url):
)
# VERIFICATION: Read back from database
saved_record = AttendanceData.query.get(attendance.id)
print(f"✅ Successfully saved attendance record with ID: {attendance.id}")
print(f"📊 Verification - location accuracy in database: {saved_record.location_accuracy}")
saved_record = db.session.get(AttendanceData, attendance.id)
logger_handler.logger.info(f"Saved attendance record ID: {attendance.id}, db accuracy: {saved_record.location_accuracy}")
if saved_record.location_accuracy != attendance.location_accuracy:
print(f"⚠️ WARNING: Database value differs from object value!")
print(f" Object value: {attendance.location_accuracy}")
print(f" Database value: {saved_record.location_accuracy}")
logger_handler.logger.warning(f"DB accuracy mismatch: object={attendance.location_accuracy}, db={saved_record.location_accuracy}")
# Add enhanced logging for location accuracy save
if attendance.location_accuracy is not None:
@@ -844,8 +811,7 @@ def qr_checkin(qr_url):
checkin_sequence_text = f"{qr_code.location_event} details"
except Exception as e:
print(f"❌ Database error: {e}")
print(f"❌ Full traceback: {traceback.format_exc()}")
logger_handler.logger.error(f"Database error saving attendance record: {e}", exc_info=True)
db.session.rollback()
logger_handler.log_database_error('checkin_save', e)
return jsonify({
@@ -879,14 +845,11 @@ def qr_checkin(qr_url):
response_data['data']['coordinates'] = f"{location_data['latitude']:.10f}, {location_data['longitude']:.10f}"
# Enhanced logging for successful check-in with all details
print(f"✅ Check-in completed successfully")
print(f" Employee ID: {attendance.employee_id}")
print(f" Time: {attendance.check_in_time.strftime('%I:%M %p')}")
print(f" Date: {attendance.check_in_date.strftime('%B %d, %Y')}")
print(f" Location: {attendance.location_name}")
print(f" Action: {qr_code.location_event}")
print(f" Address: {attendance.address}")
print(f" Today's count: {today_checkin_count}")
logger_handler.logger.info(
f"Check-in completed: employee={attendance.employee_id}, "
f"action={qr_code.location_event}, location={attendance.location_name}, "
f"time={attendance.check_in_time.strftime('%H:%M')}, count_today={today_checkin_count}"
)
# Log to database for audit trail
logger_handler.logger.info(f"Check-in success - Employee: {attendance.employee_id}, Location: {attendance.location_name}, Time: {attendance.check_in_time}, Action: {qr_code.location_event}")
@@ -894,8 +857,8 @@ def qr_checkin(qr_url):
return jsonify(response_data), 200
except Exception as e:
print(f"❌ Unexpected error in check-in process: {e}")
print(f"❌ Traceback: {traceback.format_exc()}")
db.session.rollback()
logger_handler.logger.error(f"Unexpected error in check-in process (QR: {qr_url}): {e}", exc_info=True)
return jsonify({
'success': False,
@@ -925,7 +888,7 @@ def toggle_qr_status(qr_id):
except Exception as e:
db.session.rollback()
print(f"Error toggling QR status: {e}")
logger_handler.logger.error(f"Error toggling QR status (ID: {qr_id}): {e}", exc_info=True)
return jsonify({
'success': False,
'message': 'Error updating QR code status. Please try again.'
@@ -996,7 +959,7 @@ def activate_qr_code(qr_id):
except Exception as e:
db.session.rollback()
print(f"Error activating QR code: {e}")
logger_handler.logger.error(f"Error activating QR code (ID: {qr_id}): {e}", exc_info=True)
return jsonify({
'success': False,
'message': 'Error activating QR code. Please try again.'
@@ -1021,7 +984,7 @@ def deactivate_qr_code(qr_id):
except Exception as e:
db.session.rollback()
print(f"Error deactivating QR code: {e}")
logger_handler.logger.error(f"Error deactivating QR code (ID: {qr_id}): {e}", exc_info=True)
return jsonify({
'success': False,
'message': 'Error deactivating QR code. Please try again.'
routes/statistics.py
-7
View File
@@ -202,9 +202,6 @@ def qr_statistics():
except Exception as e:
# Log the error using the correct method
logger_handler.log_database_error('statistics_page_error', e)
print(f"❌ Error loading statistics: {e}")
print(f"❌ Traceback: {traceback.format_exc()}")
flash('Error loading statistics. Please try again.', 'error')
return redirect(url_for('dashboard.dashboard'))
@@ -297,15 +294,11 @@ def export_statistics():
except Exception as e:
logger_handler.log_database_error('statistics_export_error', e)
print(f"❌ Error exporting statistics: {e}")
return jsonify({'error': 'Export failed'}), 500
except Exception as e:
# Log the error
logger_handler.log_database_error('statistics_page_error', e)
print(f"❌ Error loading statistics: {e}")
print(f"❌ Traceback: {traceback.format_exc()}")
flash('Error loading statistics. Please try again.', 'error')
return redirect(url_for('dashboard.dashboard'))
routes/time_attendance.py
+82 -67
View File
@@ -104,7 +104,7 @@ def time_attendance_dashboard():
pass
except Exception as e:
# Database table doesn't exist yet or other error - use defaults
print(f"TimeAttendance query error: {e}")
logger_handler.logger.error(f"TimeAttendance query error: {e}", exc_info=True)
pass
return render_template('time_attendance_dashboard.html',
@@ -137,9 +137,10 @@ def import_time_attendance():
coming_from_invalid_review = request.form.get('from_invalid_review', 'false').lower() == 'true'
coming_from_duplicate_review = request.form.get('from_duplicate_review', 'false').lower() == 'true'
print(f"\n🔍 IMPORT FLOW DEBUG:")
print(f" Coming from invalid review: {coming_from_invalid_review}")
print(f" Coming from duplicate review: {coming_from_duplicate_review}")
logger_handler.logger.debug(
f"Import flow: coming_from_invalid={coming_from_invalid_review}, "
f"coming_from_duplicate={coming_from_duplicate_review}"
)
if coming_from_invalid_review or coming_from_duplicate_review:
# Retrieve file from session
@@ -157,8 +158,7 @@ def import_time_attendance():
session.pop('pending_import_filename', None)
return redirect(url_for('time_attendance.import_time_attendance'))
print(f"✅ Retrieved file from session: {filename}")
print(f"✅ Temp path exists: {os.path.exists(temp_path)}")
logger_handler.logger.debug(f"Retrieved file from session: {filename}, exists={os.path.exists(temp_path)}")
else:
# Normal file upload flow - now supports multiple files
@@ -199,8 +199,7 @@ def import_time_attendance():
temp_paths.append(temp_path)
filenames.append(filename)
print(f"✅ Uploaded file {len(temp_paths)}: {filename}")
print(f"✅ Saved to: {temp_path}")
logger_handler.logger.debug(f"Uploaded file {len(temp_paths)}: {filename} saved to {temp_path}")
if len(temp_paths) == 0:
flash('No valid files selected.', 'error')
@@ -214,7 +213,7 @@ def import_time_attendance():
temp_path = temp_paths[0] if len(temp_paths) == 1 else temp_paths
filename = filenames[0] if len(filenames) == 1 else ', '.join(filenames)
print(f"✅ Total files uploaded: {len(temp_paths)}")
logger_handler.logger.debug(f"Total files uploaded: {len(temp_paths)}")
# Determine if we're processing multiple files
is_multiple_files = session.get('pending_import_files_multiple', False)
@@ -230,7 +229,7 @@ def import_time_attendance():
# Single file mode (existing behavior)
files_to_process = [(temp_path, filename)]
print(f"📁 Processing {len(files_to_process)} file(s)")
logger_handler.logger.info(f"Processing {len(files_to_process)} file(s) for time attendance import")
try:
import_service = TimeAttendanceImportService(db, logger_handler)
@@ -241,12 +240,11 @@ def import_time_attendance():
analyze_duplicates = request.form.get('analyze_duplicates', 'false').lower() == 'true'
analyze_invalid = request.form.get('analyze_invalid', 'false').lower() == 'true'
print(f"📋 Import Options:")
print(f" Skip duplicates: {skip_duplicates}")
print(f" Validate only: {validate_only}")
print(f" Analyze duplicates: {analyze_duplicates}")
print(f" Analyze invalid: {analyze_invalid}")
print(f" Coming from invalid review: {coming_from_invalid_review}")
logger_handler.logger.debug(
f"Import options: skip_duplicates={skip_duplicates}, validate_only={validate_only}, "
f"analyze_duplicates={analyze_duplicates}, analyze_invalid={analyze_invalid}, "
f"coming_from_invalid={coming_from_invalid_review}"
)
# Store combined results for multiple files
all_results = {
@@ -263,14 +261,14 @@ def import_time_attendance():
# Process each file
for file_index, (current_temp_path, current_filename) in enumerate(files_to_process, 1):
print(f"\n📄 Processing file {file_index}/{len(files_to_process)}: {current_filename}")
logger_handler.logger.info(f"Processing file {file_index}/{len(files_to_process)}: {current_filename}")
import_result = None # Initialize to prevent reference errors
try:
# For multiple files, skip review screens and import directly
if is_multiple_files:
print(f" 📦 Batch mode: processing directly without review screens")
logger_handler.logger.debug("Batch mode: processing directly without review screens")
# Validate the file first
validation_result = import_service.validate_excel_file(current_temp_path)
@@ -309,7 +307,7 @@ def import_time_attendance():
'imported': import_result.get('imported_records', 0),
'batch_id': import_result.get('batch_id', '')
})
print(f" ✅ Imported {import_result.get('imported_records', 0)} records")
logger_handler.logger.info(f"Imported {import_result.get('imported_records', 0)} records from {current_filename}")
elif import_result:
# Import ran but failed
all_results['failed_files'] += 1
@@ -322,7 +320,7 @@ def import_time_attendance():
})
except Exception as file_error:
print(f"❌ Error processing file {current_filename}: {file_error}")
logger_handler.logger.error(f"Error processing file {current_filename}: {file_error}", exc_info=True)
logger_handler.logger.error(f"Error processing file {current_filename}: {file_error}")
all_results['failed_files'] += 1
all_results['errors'].append(f"{current_filename}: {str(file_error)}")
@@ -338,9 +336,9 @@ def import_time_attendance():
if is_multiple_files and os.path.exists(current_temp_path):
try:
os.remove(current_temp_path)
print(f" 🗑️ Cleaned up temp file")
logger_handler.logger.debug(f"Cleaned up temp file: {current_temp_path}")
except Exception as cleanup_error:
print(f" ⚠️ Failed to cleanup temp file: {cleanup_error}")
logger_handler.logger.warning(f"Failed to cleanup temp file: {cleanup_error}")
# After processing all files
if is_multiple_files:
@@ -374,12 +372,11 @@ def import_time_attendance():
session.pop('pending_import_filename', None)
session.pop('pending_import_files_multiple', None)
print(f"\n📊 Batch Import Summary:")
print(f" Total files: {all_results['total_files']}")
print(f" Successful: {all_results['successful_files']}")
print(f" Failed: {all_results['failed_files']}")
print(f" Total imported: {all_results['total_imported']}")
print(f" Total duplicates: {all_results['total_duplicates']}")
logger_handler.logger.info(
f"Batch import summary: files={all_results['total_files']}, "
f"successful={all_results['successful_files']}, failed={all_results['failed_files']}, "
f"imported={all_results['total_imported']}, duplicates={all_results['total_duplicates']}"
)
return redirect(url_for('time_attendance.time_attendance_dashboard'))
@@ -388,50 +385,66 @@ def import_time_attendance():
# If analyzing for duplicates, show review page (but not if coming from invalid/duplicate review)
if analyze_duplicates and not force_import_hashes and not coming_from_invalid_review and not coming_from_duplicate_review:
print("🔍 Analyzing for duplicates...")
duplicate_analysis = import_service.analyze_for_duplicates(temp_path)
logger_handler.logger.debug("Analyzing for duplicates")
project_id_for_analysis = request.form.get('project_id')
project_id_for_analysis = int(project_id_for_analysis) if project_id_for_analysis and project_id_for_analysis != '' else None
duplicate_analysis = import_service.analyze_for_duplicates(temp_path, project_id=project_id_for_analysis)
# Surface project-mismatch errors immediately
if duplicate_analysis.get('errors'):
for err in duplicate_analysis['errors']:
flash(err, 'error')
projects = Project.query.filter_by(active_status=True).order_by(Project.name).all()
return render_template('time_attendance_import.html', projects=projects)
if duplicate_analysis['duplicate_records'] > 0:
print(f"⚠️ Found {duplicate_analysis['duplicate_records']} duplicates")
# Get project_id from form
project_id = request.form.get('project_id')
logger_handler.logger.info(f"Found {duplicate_analysis['duplicate_records']} duplicates")
# Show duplicate review page
return render_template('time_attendance_duplicate_review.html',
analysis=duplicate_analysis,
filename=filename,
project_id=project_id)
project_id=project_id_for_analysis)
else:
print("✅ No duplicates found")
logger_handler.logger.debug("No duplicates found")
flash('No duplicates found. Proceeding with import.', 'info')
# Check for invalid rows and show review if any (but not if coming from invalid review)
if analyze_invalid and not coming_from_invalid_review:
print("🔍 Analyzing for invalid rows...")
invalid_analysis = import_service.analyze_for_invalid_rows(temp_path)
logger_handler.logger.debug("Analyzing for invalid rows")
project_id_for_analysis = request.form.get('project_id')
project_id_for_analysis = int(project_id_for_analysis) if project_id_for_analysis and project_id_for_analysis != '' else None
invalid_analysis = import_service.analyze_for_invalid_rows(temp_path, project_id=project_id_for_analysis)
# Surface project-mismatch errors immediately
if invalid_analysis.get('errors'):
for err in invalid_analysis['errors']:
flash(err, 'error')
projects = Project.query.filter_by(active_status=True).order_by(Project.name).all()
return render_template('time_attendance_import.html', projects=projects)
if invalid_analysis['invalid_rows'] > 0:
print(f"⚠️ Found {invalid_analysis['invalid_rows']} invalid rows")
# Get project_id from form
project_id = request.form.get('project_id')
logger_handler.logger.info(f"Found {invalid_analysis['invalid_rows']} invalid rows")
# Show invalid row review page
return render_template('time_attendance_invalid_review.html',
analysis=invalid_analysis,
filename=filename,
project_id=project_id)
project_id=project_id_for_analysis)
else:
print("✅ All rows are valid")
logger_handler.logger.debug("All rows are valid")
flash('All rows are valid. Proceeding with import.', 'info')
# If coming from invalid review, skip validation (already done)
if not coming_from_invalid_review:
print("🔍 Validating file...")
logger_handler.logger.debug("Validating file")
# Validate file
validation_result = import_service.validate_excel_file(temp_path)
if not validation_result['valid']:
print(f"❌ Validation failed: {validation_result['errors']}")
logger_handler.logger.warning(f"Validation failed: {validation_result['errors']}")
flash(f"File validation failed: {'; '.join(validation_result['errors'])}", 'error')
projects = Project.query.filter_by(active_status=True).order_by(Project.name).all()
return render_template('time_attendance_import.html',
projects=projects,
validation_result=validation_result)
if validation_result['warnings']:
@@ -439,15 +452,17 @@ def import_time_attendance():
flash(warning, 'warning')
if validate_only:
print(f"✅ Validation successful: {validation_result['valid_rows']} valid records")
logger_handler.logger.info(f"Validation successful: {validation_result['valid_rows']} valid records")
flash(f"File validation successful! Found {validation_result['valid_rows']} valid records.", 'success')
projects = Project.query.filter_by(active_status=True).order_by(Project.name).all()
return render_template('time_attendance_import.html',
projects=projects,
validation_result=validation_result)
else:
print("⏭️ Skipping validation (already validated)")
logger_handler.logger.debug("Skipping validation (already validated)")
# Proceed with import
print("🚀 Starting import process...")
logger_handler.logger.info("Starting import process")
import_source = request.form.get('import_source', f"Manual Import - {filename}")
project_id = request.form.get('project_id')
project_id = int(project_id) if project_id and project_id != '' else None
@@ -462,11 +477,11 @@ def import_time_attendance():
)
if import_result['success']:
print(f"✅ Import successful!")
print(f" Batch ID: {import_result['batch_id']}")
print(f" Imported: {import_result['imported_records']}/{import_result['total_records']}")
print(f" Duplicates: {import_result['duplicate_records']}")
print(f" Failed: {import_result['failed_records']}")
logger_handler.logger.info(
f"Import successful: batch_id={import_result['batch_id']}, "
f"imported={import_result['imported_records']}/{import_result['total_records']}, "
f"duplicates={import_result['duplicate_records']}, failed={import_result['failed_records']}"
)
logger_handler.logger.info(
f"User {session['username']} successfully imported time attendance data - "
@@ -496,36 +511,36 @@ def import_time_attendance():
os.remove(temp_path)
session.pop('pending_import_file', None)
session.pop('pending_import_filename', None)
print("🗑️ Cleaned up temp file")
logger_handler.logger.debug(f"Cleaned up temp file: {temp_path}")
except Exception as cleanup_error:
print(f"⚠️ Failed to cleanup temp file: {cleanup_error}")
logger_handler.logger.warning(f"Failed to cleanup temp file: {cleanup_error}")
return render_template('time_attendance_import_result.html',
import_result=import_result)
else:
print(f"❌ Import failed: {import_result['errors']}")
logger_handler.logger.error(f"Import failed: {import_result['errors']}")
flash(f"Import failed: {'; '.join(import_result['errors'][:3])}", 'error')
if len(import_result['errors']) > 3:
flash(f"...and {len(import_result['errors']) - 3} more errors", 'warning')
projects = Project.query.filter_by(active_status=True).order_by(Project.name).all()
return render_template('time_attendance_import.html',
projects=projects,
import_result=import_result)
except Exception as import_error:
print(f"❌ Import exception: {import_error}")
import traceback
print(f"❌ Traceback: {traceback.format_exc()}")
logger_handler.logger.error(f"Import exception: {import_error}", exc_info=True)
raise
except Exception as e:
logger_handler.log_database_error('time_attendance_import', e)
print(f"❌ Top-level exception: {e}")
import traceback
print(f"❌ Traceback: {traceback.format_exc()}")
logger_handler.logger.error(f"Top-level import exception: {e}", exc_info=True)
flash('Import failed due to an unexpected error.', 'error')
return render_template('time_attendance_import.html')
projects = Project.query.filter_by(active_status=True).order_by(Project.name).all()
return render_template('time_attendance_import.html', projects=projects)
# GET request
return render_template('time_attendance_import.html')
projects = Project.query.filter_by(active_status=True).order_by(Project.name).all()
return render_template('time_attendance_import.html', projects=projects)
@bp.route('/time-attendance/import/analyze-duplicates', methods=['POST'], endpoint='analyze_import_duplicates')
@@ -1198,13 +1213,13 @@ def export_time_attendance():
project_name_for_filename = ''
if project_filter:
try:
project = Project.query.get(int(project_filter))
project = db.session.get(Project, int(project_filter))
if project:
# Replace spaces and special characters with underscores
project_name_safe = project.name.replace(' ', '_').replace('/', '_').replace('\\', '_')
project_name_for_filename = f"{project_name_safe}_"
except Exception as e:
print(f"⚠️ Error getting project name for filename: {e}")
logger_handler.logger.warning(f"Error getting project name for filename: {e}")
# Log export
logger_handler.logger.info(
@@ -1343,13 +1358,13 @@ def export_time_attendance_by_building():
project_name_for_filename = ''
if project_filter:
try:
project = Project.query.get(int(project_filter))
project = db.session.get(Project, int(project_filter))
if project:
# Replace spaces and special characters with underscores
project_name_safe = project.name.replace(' ', '_').replace('/', '_').replace('\\', '_')
project_name_for_filename = f"{project_name_safe}_"
except Exception as e:
print(f"⚠️ Error getting project name for filename: {e}")
logger_handler.logger.warning(f"Error getting project name for filename: {e}")
# Log export
logger_handler.logger.info(
routes/time_attendance_export.py
+220 -250
View File
@@ -120,6 +120,180 @@ def _qtr(decimal_hours: float) -> float:
base100 = convert_minutes_to_base100(rounded_minutes)
return round_base100_hours(base100)
# ---------------------------------------------------------------------------
# Private export helpers — shared by both export functions below.
# ---------------------------------------------------------------------------
def _resolve_date_range(start_date_filter, end_date_filter, records, export_label='TA Excel export'):
"""
Resolve and validate the export date range.
Returns (start_date, end_date, filtered_records) where:
- start_date / end_date are date objects
- filtered_records is the input list capped to end_date + 1 day (overnight buffer)
- end_date is capped to a maximum 14-day window
Returns None when there are no records and no date filters.
"""
MAX_EXPORT_DAYS = 14
if start_date_filter and end_date_filter:
if isinstance(start_date_filter, str):
start_date = datetime.strptime(start_date_filter, '%Y-%m-%d').date()
else:
start_date = start_date_filter
if isinstance(end_date_filter, str):
end_date = datetime.strptime(end_date_filter, '%Y-%m-%d').date()
else:
end_date = end_date_filter
elif records:
start_date = min(r.attendance_date for r in records)
end_date = max(r.attendance_date for r in records)
else:
return None
if (end_date - start_date).days >= MAX_EXPORT_DAYS:
capped_end_date = start_date + timedelta(days=MAX_EXPORT_DAYS - 1)
logger_handler.logger.info(
f"{export_label}: date range [{start_date} \u2013 {end_date}] exceeds "
f"{MAX_EXPORT_DAYS} days; capping end_date to {capped_end_date}."
)
end_date = capped_end_date
# Preserve one extra calendar day so early-morning check-out records
# stored on Day N+1 remain available for overnight pairing detection.
# Display range is still controlled by dates_with_records (capped to end_date).
filtered_records = [r for r in records if r.attendance_date <= end_date + timedelta(days=1)]
return start_date, end_date, filtered_records
def _convert_ta_records(records):
"""
Convert TimeAttendance ORM records to the lightweight anonymous-class
format expected by WorkingHoursCalculator and the Excel rendering loops.
Returns a list of converted record objects.
"""
from working_hours_calculator import parse_employee_id_for_work_type
converted = []
for record in records:
distance_value = getattr(record, 'distance', None)
record_type = 'check_in'
if hasattr(record, 'action_description') and record.action_description:
action_lower = record.action_description.lower()
if 'out' in action_lower or 'checkout' in action_lower:
record_type = 'check_out'
_, work_type = parse_employee_id_for_work_type(str(record.employee_id))
base_location_name = record.location_name
if work_type and work_type in ('PT', 'SP', 'PW'):
display_location_name = f"{base_location_name} ({work_type})"
else:
display_location_name = base_location_name
converted_record = type('Record', (), {
'id': record.id,
'employee_id': str(record.employee_id),
'employee_name': getattr(record, 'employee_name', ''),
'check_in_date': record.attendance_date,
'check_in_time': record.attendance_time,
'location_name': display_location_name,
'original_location_name': base_location_name,
'work_type': work_type,
'latitude': None,
'longitude': None,
'distance': distance_value,
'record_type': record_type,
'action_description': record.action_description,
'event_description': record.event_description or '',
'recorded_address': record.recorded_address or '',
'qr_code': type('QRCode', (), {
'location': base_location_name,
'location_address': record.recorded_address or '',
'project': None
})()
})()
converted.append(converted_record)
return converted
def _build_employee_name_map(records):
"""
Build a {base_employee_id: "Lastname, Firstname"} map for export headers.
Looks up the Employee table by numeric base ID so work-type suffixes
(e.g. '3937SP') in the stored employee_name column do not pollute labels.
Falls back to the stored employee_name on lookup failure.
"""
from working_hours_calculator import parse_employee_id_for_work_type
employee_names = {}
for record in records:
base_id, _ = parse_employee_id_for_work_type(str(record.employee_id))
if base_id not in employee_names:
try:
emp = Employee.query.filter_by(id=int(base_id)).first()
if emp:
employee_names[base_id] = f"{emp.lastName}, {emp.firstName}"
else:
employee_names[base_id] = getattr(record, 'employee_name', f'Employee {base_id}')
logger_handler.logger.warning(
f"Employee ID {base_id} not found in employee table during export; "
f"using stored name."
)
except Exception as e:
employee_names[base_id] = getattr(record, 'employee_name', f'Employee {base_id}')
logger_handler.logger.warning(
f"Could not lookup employee name for ID {base_id} during export: {e}"
)
return employee_names
def _make_export_styles():
"""
Return a dict of openpyxl style objects shared by both export functions.
Keys: header_font, header_fill, data_font, bold_font, italic_bold_font,
border, missed_punch_fill, border_day_middle, border_day_last,
border_day_single, border_day_first, amber_fill
"""
header_font = Font(name='Aptos Narrow', size=11, bold=True, color='FFFFFF')
header_fill = PatternFill(start_color='000000', end_color='000000', fill_type='solid')
data_font = Font(name='Aptos Narrow', size=11)
bold_font = Font(name='Aptos Narrow', size=11, bold=True)
italic_bold_font = Font(name='Aptos Narrow', size=11, bold=True, italic=True)
border = Border(
left=Side(style='thin'),
right=Side(style='thin'),
top=Side(style='thin'),
bottom=Side(style='thin')
)
missed_punch_fill = PatternFill(start_color='FFC000', end_color='FFC000', fill_type='solid')
amber_fill = PatternFill(start_color='FFC000', end_color='FFC000', fill_type='solid')
border_day_middle = Border()
border_day_last = Border(bottom=Side(style='thin'))
border_day_single = Border(bottom=Side(style='thin'))
border_day_first = Border()
return {
'header_font': header_font,
'header_fill': header_fill,
'data_font': data_font,
'bold_font': bold_font,
'italic_bold_font': italic_bold_font,
'border': border,
'missed_punch_fill': missed_punch_fill,
'amber_fill': amber_fill,
'border_day_middle': border_day_middle,
'border_day_last': border_day_last,
'border_day_single': border_day_single,
'border_day_first': border_day_first,
}
def export_time_attendance_excel(records, project_name_for_filename, date_range_str, filter_str, start_date_filter=None, end_date_filter=None):
"""Generate Excel export with template format matching the provided template"""
from openpyxl import Workbook
@@ -132,105 +306,17 @@ def export_time_attendance_excel(records, project_name_for_filename, date_range_
ws = wb.active
ws.title = "Sheet0"
# Get date range for calculations
if start_date_filter and end_date_filter:
# Convert string dates to date objects if needed
if isinstance(start_date_filter, str):
start_date = datetime.strptime(start_date_filter, '%Y-%m-%d').date()
else:
start_date = start_date_filter
if isinstance(end_date_filter, str):
end_date = datetime.strptime(end_date_filter, '%Y-%m-%d').date()
else:
end_date = end_date_filter
elif records:
# Fallback to calculating from records if no filter dates provided
start_date = min(r.attendance_date for r in records)
end_date = max(r.attendance_date for r in records)
else:
# Resolve date range and cap to 14-day window
result = _resolve_date_range(start_date_filter, end_date_filter, records, 'TA Excel export')
if result is None:
return None
# Enforce maximum 2-week (14-day) export window.
# If the selected range exceeds 14 days, cap end_date to start_date + 13 days.
MAX_EXPORT_DAYS = 14
if (end_date - start_date).days >= MAX_EXPORT_DAYS:
capped_end_date = start_date + timedelta(days=MAX_EXPORT_DAYS - 1)
logger_handler.logger.info(
f"TA Excel export: date range [{start_date} – {end_date}] exceeds {MAX_EXPORT_DAYS} days; "
f"capping end_date to {capped_end_date}."
)
end_date = capped_end_date
# Drop records that fall outside the capped window
# Preserve one extra calendar day so early-morning check-out records
# stored on Day N+1 remain available for overnight pairing detection.
# Display range is still controlled by dates_with_records (capped to end_date).
records = [r for r in records if r.attendance_date <= end_date + timedelta(days=1)]
start_date, end_date, records = result
# Import parse function at the beginning for work type detection
from working_hours_calculator import parse_employee_id_for_work_type
# Convert TimeAttendance records to format expected by calculator
converted_records = []
for record in records:
# Get distance value from the record
distance_value = getattr(record, 'distance', None)
# CRITICAL: Determine record_type from action_description
record_type = 'check_in' # Default
if hasattr(record, 'action_description') and record.action_description:
action_lower = record.action_description.lower()
if 'out' in action_lower or 'checkout' in action_lower:
record_type = 'check_out'
# Extract work type (PT, SP, PW) from employee_id for location display in Excel
_, work_type = parse_employee_id_for_work_type(str(record.employee_id))
# Create display location name with work type suffix if applicable
base_location_name = record.location_name
if work_type and work_type in ('PT', 'SP', 'PW'):
display_location_name = f"{base_location_name} ({work_type})"
else:
display_location_name = base_location_name
converted_record = type('Record', (), {
'id': record.id,
'employee_id': str(record.employee_id),
'check_in_date': record.attendance_date,
'check_in_time': record.attendance_time,
'location_name': display_location_name, # Use display name with work type for Excel export
'original_location_name': base_location_name, # Keep original for internal grouping
'work_type': work_type, # Store work type for reference
'latitude': None,
'longitude': None,
'distance': distance_value,
'record_type': record_type,
'action_description': record.action_description,
'event_description': record.event_description or '',
'recorded_address': record.recorded_address or '',
'qr_code': type('QRCode', (), {
'location': base_location_name, # Keep original for QR code matching
'location_address': record.recorded_address or '',
'project': None
})()
})()
converted_records.append(converted_record)
# Log count of records with work types for audit trail
work_type_counts = {'PT': 0, 'SP': 0, 'PW': 0, 'Regular': 0}
for r in converted_records:
wt = getattr(r, 'work_type', None)
if wt in work_type_counts:
work_type_counts[wt] += 1
else:
work_type_counts['Regular'] += 1
if any(work_type_counts[wt] > 0 for wt in ['PT', 'SP', 'PW']):
logger_handler.logger.info(
f"Excel Export: Processing records with work types - "
f"Regular: {work_type_counts['Regular']}, PT: {work_type_counts['PT']}, "
f"SP: {work_type_counts['SP']}, PW: {work_type_counts['PW']}"
)
converted_records = _convert_ta_records(records)
# Calculate working hours using WorkingHoursCalculator
calculator = WorkingHoursCalculator()
@@ -240,53 +326,23 @@ def export_time_attendance_excel(records, project_name_for_filename, date_range_
converted_records
)
# Get employee names - map BASE employee IDs to names for consolidated display
# Look up from Employee table using the numeric base_id to get the correct name,
# regardless of what is stored in the employee_name column (which may contain
# work type characters such as 'Employee 3937SP' if imported with a decorated ID).
from working_hours_calculator import parse_employee_id_for_work_type
employee_names = {}
for record in records:
base_id, _ = parse_employee_id_for_work_type(str(record.employee_id))
if base_id not in employee_names:
try:
emp = Employee.query.filter_by(id=int(base_id)).first()
if emp:
employee_names[base_id] = f"{emp.lastName}, {emp.firstName}"
else:
# Fallback: use stored name if Employee table lookup fails
employee_names[base_id] = record.employee_name
logger_handler.logger.warning(f"Employee ID {base_id} not found in employee table during export; using stored name.")
except Exception as e:
employee_names[base_id] = record.employee_name
logger_handler.logger.warning(f"Could not lookup employee name for ID {base_id} during export: {e}")
# Build employee name map (Lastname, Firstname keyed by base employee ID)
employee_names = _build_employee_name_map(records)
# Setup styles
# White bold text on black background for column header row (no border)
header_font = Font(name='Aptos Narrow', size=11, bold=True, color='FFFFFF')
header_fill = PatternFill(start_color='000000', end_color='000000', fill_type='solid')
data_font = Font(name='Aptos Narrow', size=11)
bold_font = Font(name='Aptos Narrow', size=11, bold=True)
border = Border(
left=Side(style='thin'),
right=Side(style='thin'),
top=Side(style='thin'),
bottom=Side(style='thin')
)
# CHANGED: Sample format uses ONLY a bottom border on the last row of each day group.
# Intermediate rows and first rows have no borders at all (no left/right/top).
border_day_middle = Border() # No borders on intermediate rows
border_day_last = Border(
bottom=Side(style='thin') # Only bottom border on the last row of a day group
)
border_day_single = Border(
bottom=Side(style='thin') # Single-row days also get only bottom border
)
# border_day_first is same as middle (no borders) — kept for compatibility
border_day_first = Border()
# Setup styles (shared objects)
_styles = _make_export_styles()
header_font = _styles['header_font']
header_fill = _styles['header_fill']
data_font = _styles['data_font']
bold_font = _styles['bold_font']
italic_bold_font = _styles['italic_bold_font']
border = _styles['border']
missed_punch_fill = _styles['missed_punch_fill']
amber_fill = _styles['amber_fill']
border_day_middle = _styles['border_day_middle']
border_day_last = _styles['border_day_last']
border_day_single = _styles['border_day_single']
border_day_first = _styles['border_day_first']
def get_day_border(row_position, total_rows):
"""
@@ -511,8 +567,10 @@ def export_time_attendance_excel(records, project_name_for_filename, date_range_
del daily_location_data[_ndk][_co_loc]
if _ndk in daily_location_data and not daily_location_data[_ndk]:
del daily_location_data[_ndk]
print(f"\U0001f319 [TA Export] Overnight: moved checkout {_co.check_in_time} "
f"from {_ndk} to {_dk} for employee {employee_id}")
logger_handler.logger.info(
f"TA Export overnight shift: moved checkout {_co.check_in_time} "
f"from {_ndk} to {_dk} for employee {employee_id}"
)
# -------------------------------------------------------------------
# END OVERNIGHT SHIFT DETECTION
# -------------------------------------------------------------------
@@ -855,7 +913,7 @@ def export_time_attendance_excel(records, project_name_for_filename, date_range_
'work_type': wt, # None means regular
'used': False
})
print(f" Record at {record.check_in_time}: action='{record.action_description}', is_out={is_out}, work_type={wt}")
logger_handler.logger.debug(f"TA Export record: time={record.check_in_time}, action='{record.action_description}', is_out={is_out}, work_type={wt}")
ins = [ri for ri in record_info if not ri['is_out']]
outs = [ri for ri in record_info if ri['is_out']]
@@ -975,7 +1033,7 @@ def export_time_attendance_excel(records, project_name_for_filename, date_range_
'effective_work_type': ri['work_type']
})
print(f" Created {len(pairs_to_write)} pairs")
logger_handler.logger.debug(f"TA Export: created {len(pairs_to_write)} pairs for export")
# Sort pairs chronologically by the anchor record's time so that
# orphaned records (assembled last in Steps 2-3) appear in the
@@ -1300,82 +1358,16 @@ def export_time_attendance_by_building_excel(records, project_name_for_filename,
ws = wb.active
ws.title = "Sheet0"
# Get date range for calculations
if start_date_filter and end_date_filter:
if isinstance(start_date_filter, str):
start_date = datetime.strptime(start_date_filter, '%Y-%m-%d').date()
else:
start_date = start_date_filter
if isinstance(end_date_filter, str):
end_date = datetime.strptime(end_date_filter, '%Y-%m-%d').date()
else:
end_date = end_date_filter
elif records:
start_date = min(r.attendance_date for r in records)
end_date = max(r.attendance_date for r in records)
else:
# Resolve date range and cap to 14-day window
result = _resolve_date_range(start_date_filter, end_date_filter, records, 'TA by-building Excel export')
if result is None:
return None
start_date, end_date, records = result
# Enforce maximum 2-week (14-day) export window.
MAX_EXPORT_DAYS = 14
if (end_date - start_date).days >= MAX_EXPORT_DAYS:
capped_end_date = start_date + timedelta(days=MAX_EXPORT_DAYS - 1)
logger_handler.logger.info(
f"TA by-building Excel export: date range [{start_date} – {end_date}] exceeds {MAX_EXPORT_DAYS} days; "
f"capping end_date to {capped_end_date}."
)
end_date = capped_end_date
# Preserve one extra calendar day so early-morning check-out records
# stored on Day N+1 remain available for overnight pairing detection.
# Display range is still controlled by dates_with_records (capped to end_date).
records = [r for r in records if r.attendance_date <= end_date + timedelta(days=1)]
# Import parse function for work type detection
from working_hours_calculator import parse_employee_id_for_work_type
# Convert TimeAttendance records to format expected by calculator
converted_records = []
for record in records:
distance_value = getattr(record, 'distance', None)
record_type = 'check_in'
if hasattr(record, 'action_description') and record.action_description:
action_lower = record.action_description.lower()
if 'out' in action_lower or 'checkout' in action_lower:
record_type = 'check_out'
_, work_type = parse_employee_id_for_work_type(str(record.employee_id))
base_location_name = record.location_name
if work_type and work_type in ('PT', 'SP', 'PW'):
display_location_name = f"{base_location_name} ({work_type})"
else:
display_location_name = base_location_name
converted_record = type('Record', (), {
'id': record.id,
'employee_id': str(record.employee_id),
'employee_name': record.employee_name,
'check_in_date': record.attendance_date,
'check_in_time': record.attendance_time,
'location_name': display_location_name,
'original_location_name': base_location_name,
'work_type': work_type,
'latitude': None,
'longitude': None,
'distance': distance_value,
'record_type': record_type,
'action_description': record.action_description,
'event_description': record.event_description or '',
'recorded_address': record.recorded_address or '',
'qr_code': type('QRCode', (), {
'location': base_location_name,
'location_address': record.recorded_address or '',
'project': None
})()
})()
converted_records.append(converted_record)
converted_records = _convert_ta_records(records)
# Group records by location (building)
location_groups = {}
@@ -1401,43 +1393,21 @@ def export_time_attendance_by_building_excel(records, project_name_for_filename,
converted_records
)
# Get employee names map
# Look up from Employee table using the numeric base_id to get the correct name,
# regardless of what is stored in the employee_name column (which may contain
# work type characters such as 'Employee 3937SP' if imported with a decorated ID).
employee_names = {}
for record in records:
base_id, _ = parse_employee_id_for_work_type(str(record.employee_id))
if base_id not in employee_names:
try:
emp = Employee.query.filter_by(id=int(base_id)).first()
if emp:
employee_names[base_id] = f"{emp.lastName}, {emp.firstName}"
else:
# Fallback: use stored name if Employee table lookup fails
employee_names[base_id] = record.employee_name
logger_handler.logger.warning(f"Employee ID {base_id} not found in employee table during export (by-building); using stored name.")
except Exception as e:
employee_names[base_id] = record.employee_name
logger_handler.logger.warning(f"Could not lookup employee name for ID {base_id} during export (by-building): {e}")
# Build employee name map (Lastname, Firstname keyed by base employee ID)
employee_names = _build_employee_name_map(records)
# Setup styles
header_font = Font(name='Aptos Narrow', size=11, bold=True, color='FFFFFF')
header_fill = PatternFill(start_color='000000', end_color='000000', fill_type='solid')
data_font = Font(name='Aptos Narrow', size=11)
bold_font = Font(name='Aptos Narrow', size=11, bold=True)
italic_bold_font = Font(name='Aptos Narrow', size=11, bold=True, italic=True)
border = Border(
left=Side(style='thin'),
right=Side(style='thin'),
top=Side(style='thin'),
bottom=Side(style='thin')
)
missed_punch_fill = PatternFill(start_color='FFC000', end_color='FFC000', fill_type='solid')
# Bottom-only border on the last row of each day group (matches normal TA export).
# Intermediate rows within a day have no borders.
border_day_middle = Border() # No borders on intermediate rows
border_day_last = Border(bottom=Side(style='thin')) # Bottom border on last row of day
# Setup styles (shared objects)
_styles = _make_export_styles()
header_font = _styles['header_font']
header_fill = _styles['header_fill']
data_font = _styles['data_font']
bold_font = _styles['bold_font']
italic_bold_font = _styles['italic_bold_font']
border = _styles['border']
missed_punch_fill = _styles['missed_punch_fill']
amber_fill = _styles['amber_fill']
border_day_middle = _styles['border_day_middle']
border_day_last = _styles['border_day_last']
# Write main headers
current_row = 1
routes/users.py
+17 -17
View File
@@ -118,7 +118,7 @@ def create_user():
try:
project_id = int(pid)
# Verify project exists
if Project.query.get(project_id):
if db.session.get(Project, project_id):
valid_project_ids.append(project_id)
except (ValueError, TypeError):
logger_handler.logger.warning(f"Invalid project ID received: {pid}")
@@ -206,8 +206,8 @@ def create_user():
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'])
user_to_delete = db.session.get(User, user_id)
current_user = db.session.get(User, session['user_id'])
if not user_to_delete:
flash('User not found.', 'error')
@@ -247,8 +247,8 @@ def delete_user(user_id):
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'])
user_to_reactivate = db.session.get(User, user_id)
current_user = db.session.get(User, session['user_id'])
if not user_to_reactivate:
flash('User not found.', 'error')
@@ -277,8 +277,8 @@ def reactivate_user(user_id):
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'])
user_to_promote = db.session.get(User, user_id)
current_user = db.session.get(User, session['user_id'])
if not user_to_promote:
flash('User not found.', 'error')
@@ -307,8 +307,8 @@ def promote_user(user_id):
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'])
user_to_demote = db.session.get(User, user_id)
current_user = db.session.get(User, session['user_id'])
if not user_to_demote:
flash('User not found.', 'error')
@@ -437,7 +437,7 @@ def edit_user(user_id):
try:
project_id = int(pid)
# Verify project exists
if Project.query.get(project_id):
if db.session.get(Project, project_id):
valid_project_ids.append(project_id)
except (ValueError, TypeError):
logger_handler.logger.warning(f"Invalid project ID received: {pid}")
@@ -561,8 +561,8 @@ def edit_user(user_id):
def toggle_user_status(user_id):
"""Toggle user active status via AJAX (Admin only)"""
try:
user_to_toggle = User.query.get(user_id)
current_user = User.query.get(session['user_id'])
user_to_toggle = db.session.get(User, user_id)
current_user = db.session.get(User, session['user_id'])
if not user_to_toggle:
return jsonify({
@@ -619,8 +619,8 @@ def toggle_user_status(user_id):
def activate_user(user_id):
"""Activate a user (Admin only) - Alternative route"""
try:
user_to_activate = User.query.get(user_id)
current_user = User.query.get(session['user_id'])
user_to_activate = db.session.get(User, user_id)
current_user = db.session.get(User, session['user_id'])
if not user_to_activate:
flash('User not found.', 'error')
@@ -652,8 +652,8 @@ def activate_user(user_id):
def deactivate_user(user_id):
"""Deactivate a user (Admin only) - Alternative route"""
try:
user_to_deactivate = User.query.get(user_id)
current_user = User.query.get(session['user_id'])
user_to_deactivate = db.session.get(User, user_id)
current_user = db.session.get(User, session['user_id'])
if not user_to_deactivate:
flash('User not found.', 'error')
@@ -904,7 +904,7 @@ def permanently_delete_user(user_id):
"""Permanently delete user but preserve associated QR codes (Admin only)"""
try:
user_to_delete = User.query.get_or_404(user_id)
current_user = User.query.get(session['user_id'])
current_user = db.session.get(User, session['user_id'])
# Security checks
if user_to_delete.id == current_user.id:
templates/time_attendance_import.html
+11 -5
View File
@@ -927,11 +927,17 @@ importForm.addEventListener('submit', async (e) => {
const result = msg.result;
setProgress(result.total_records, result.total_records, 'Import finished.');
let summary = '✅ Imported ' + result.imported_records + ' of ' + result.total_records + ' records.';
if (result.duplicate_records > 0)
summary += ' Skipped ' + result.duplicate_records + ' duplicates.';
if (result.failed_records > 0)
summary += ' ⚠️ ' + result.failed_records + ' records failed.';
let summary;
if (!result.success && result.total_records === 0) {
// Validation failed before any records were processed (e.g. project mismatch)
summary = '❌ Import was rejected before processing any records.';
} else {
summary = (result.success ? '✅' : '❌') + ' Imported ' + result.imported_records + ' of ' + result.total_records + ' records.';
if (result.duplicate_records > 0)
summary += ' Skipped ' + result.duplicate_records + ' duplicates.';
if (result.failed_records > 0)
summary += ' ⚠️ ' + result.failed_records + ' records failed.';
}
const errors = (!result.success && result.errors && result.errors.length > 0)
? result.errors
time_attendance_import_service.py
+113 -11
View File
@@ -321,12 +321,15 @@ class TimeAttendanceImportService:
self.logger.logger.error(f"Failed to get existing record hashes with data: {e}")
return {}
def analyze_for_duplicates(self, file_path: str) -> Dict[str, Any]:
def analyze_for_duplicates(self, file_path: str, project_id: int = None) -> Dict[str, Any]:
"""
Analyze file for potential duplicates WITHOUT importing
Analyze file for potential duplicates WITHOUT importing.
Args:
file_path: Path to the Excel file
file_path: Path to the Excel file
project_id: Optional project ID — if supplied, every Location Name
in the file is validated against that project's QR code
locations before the duplicate analysis proceeds.
Returns:
Dictionary containing duplicate analysis
@@ -356,6 +359,46 @@ class TimeAttendanceImportService:
df = df.dropna(how='all')
analysis_result['total_records'] = len(df)
# ── Project-location validation ──────────────────────────────
if project_id:
try:
from models.qrcode import QRCode
from models.project import Project
file_locations = set(
str(loc).strip()
for loc in df['Location Name'].dropna().unique()
if str(loc).strip()
)
project_locations = set(
qr.location
for qr in QRCode.query.filter_by(project_id=project_id)
.with_entities(QRCode.location).all()
)
unmatched = next(
(loc for loc in sorted(file_locations) if loc not in project_locations),
None
)
if unmatched:
project_obj = self.db.session.get(Project, project_id)
project_name = project_obj.name if project_obj else f'ID {project_id}'
error_msg = (
f"The data in the file does not belong to the project '{project_name}'. "
f"Please verify the selected project or correct the file."
)
analysis_result['errors'].append(error_msg)
if self.logger:
self.logger.logger.warning(f"Project-location mismatch during duplicate analysis: {error_msg}")
return analysis_result
if self.logger:
self.logger.logger.info(
f"Project-location validation passed (duplicate analysis): "
f"all {len(file_locations)} location(s) belong to project ID {project_id}."
)
except Exception as e:
if self.logger:
self.logger.logger.warning(f"Could not perform project-location validation (duplicate analysis): {e}")
# ── End project-location validation ──────────────────────────
# Get existing record hashes
existing_hashes = self._get_existing_record_hashes_with_data()
@@ -445,12 +488,15 @@ class TimeAttendanceImportService:
return analysis_result
def analyze_for_invalid_rows(self, file_path: str) -> Dict[str, Any]:
def analyze_for_invalid_rows(self, file_path: str, project_id: int = None) -> Dict[str, Any]:
"""
Analyze file for invalid rows with detailed error information
Analyze file for invalid rows with detailed error information.
Args:
file_path: Path to the Excel file
file_path: Path to the Excel file
project_id: Optional project ID — if supplied, every Location Name
in the file is validated against that project's QR code
locations before the row analysis proceeds.
Returns:
Dictionary containing invalid row analysis
@@ -480,6 +526,46 @@ class TimeAttendanceImportService:
df = df.dropna(how='all')
analysis_result['total_rows'] = len(df)
# ── Project-location validation ──────────────────────────────
if project_id:
try:
from models.qrcode import QRCode
from models.project import Project
file_locations = set(
str(loc).strip()
for loc in df['Location Name'].dropna().unique()
if str(loc).strip()
)
project_locations = set(
qr.location
for qr in QRCode.query.filter_by(project_id=project_id)
.with_entities(QRCode.location).all()
)
unmatched = next(
(loc for loc in sorted(file_locations) if loc not in project_locations),
None
)
if unmatched:
project_obj = self.db.session.get(Project, project_id)
project_name = project_obj.name if project_obj else f'ID {project_id}'
error_msg = (
f"The data in the file does not belong to the project '{project_name}'. "
f"Please verify the selected project or correct the file."
)
analysis_result['errors'].append(error_msg)
if self.logger:
self.logger.logger.warning(f"Project-location mismatch during invalid-row analysis: {error_msg}")
return analysis_result
if self.logger:
self.logger.logger.info(
f"Project-location validation passed (invalid-row analysis): "
f"all {len(file_locations)} location(s) belong to project ID {project_id}."
)
except Exception as e:
if self.logger:
self.logger.logger.warning(f"Could not perform project-location validation (invalid-row analysis): {e}")
# ── End project-location validation ──────────────────────────
# Analyze each row
invalid_list = []
valid_count = 0
@@ -663,13 +749,22 @@ class TimeAttendanceImportService:
if str(loc).strip()
)
# Fetch all location names that belong to the selected project
# Fetch all location names that belong to the selected project.
# Strip each value — DB entries may have trailing whitespace.
project_locations = set(
qr.location
qr.location.strip()
for qr in QRCode.query.filter_by(project_id=project_id)
.with_entities(QRCode.location).all()
if qr.location
)
if self.logger:
self.logger.logger.info(
f"Project-location validation: project_id={project_id}, "
f"file_locations={sorted(file_locations)}, "
f"project_locations={sorted(project_locations)}"
)
# Find the first location in the file that is not in the project
unmatched = next(
(loc for loc in sorted(file_locations) if loc not in project_locations),
@@ -677,10 +772,11 @@ class TimeAttendanceImportService:
)
if unmatched:
project_obj = Project.query.get(project_id)
project_obj = self.db.session.get(Project, project_id)
project_name = project_obj.name if project_obj else f'ID {project_id}'
error_msg = (
f"The data in the file does not belong to the project '{project_name}'. "
f"Location '{unmatched}' was not found in this project. "
f"Please verify the selected project or correct the file."
)
import_results['errors'].append(error_msg)
@@ -697,10 +793,16 @@ class TimeAttendanceImportService:
)
except Exception as e:
# Fail closed: if validation cannot be performed, block the import.
# This prevents a DB or import error from silently bypassing the check.
error_msg = f"Project-location validation could not be completed: {e}"
import_results['errors'].append(error_msg)
if self.logger:
self.logger.logger.warning(
f"Could not perform project-location validation: {e}"
self.logger.logger.error(
f"Project-location validation error (failing closed): {e}",
exc_info=True
)
return import_results
# ── End project-location validation ──────────────────────────────────
# Track duplicates using hash