04/28 Updated sprint 4 & 5
This commit is contained in:
@@ -0,0 +1,422 @@
|
||||
"""
|
||||
routes/verification.py
|
||||
======================
|
||||
Verification review routes for attendance records requiring photo verification.
|
||||
|
||||
Routes: /verification-review, /verification-review/<id>,
|
||||
/verification-review/<id>/update,
|
||||
/api/attendance/<id>/verification-details,
|
||||
/api/attendance/stats
|
||||
|
||||
"""
|
||||
from flask import abort, Blueprint, render_template, request, redirect, flash, session, jsonify, send_file, url_for
|
||||
from datetime import datetime, date, timedelta, time
|
||||
import io, os, json, re, traceback
|
||||
|
||||
from extensions import db, logger_handler
|
||||
from models.attendance import AttendanceData
|
||||
from models.employee import Employee
|
||||
from models.permissions import UserLocationPermission, UserProjectPermission
|
||||
from models.project import Project
|
||||
from models.qrcode import QRCode
|
||||
from models.user import User
|
||||
from sqlalchemy import text, or_, and_
|
||||
from logger_handler import log_user_activity, log_database_operations
|
||||
from utils.helpers import (
|
||||
admin_required,
|
||||
get_client_ip,
|
||||
has_admin_privileges,
|
||||
has_staff_level_access,
|
||||
login_required,
|
||||
staff_or_admin_required)
|
||||
from utils.geocoding import (calculate_location_accuracy_enhanced, process_location_data_enhanced,
|
||||
check_location_accuracy_column_exists)
|
||||
import openpyxl
|
||||
from openpyxl.styles import Font, PatternFill, Alignment, Border, Side
|
||||
from openpyxl.utils import get_column_letter
|
||||
|
||||
from routes.attendance import bp # shared blueprint — do not redefine
|
||||
|
||||
|
||||
@bp.route('/verification-review', endpoint='verification_review')
|
||||
@login_required
|
||||
def verification_review():
|
||||
"""Admin page to review pending photo verifications"""
|
||||
try:
|
||||
# Only admins can access
|
||||
if session.get('role') not in ['admin', 'payroll', 'accounting']:
|
||||
flash('Unauthorized access.', 'error')
|
||||
return redirect(url_for('dashboard.dashboard'))
|
||||
|
||||
# Get filter parameters
|
||||
status_filter = request.args.get('status', 'pending')
|
||||
date_from = request.args.get('date_from', '')
|
||||
date_to = request.args.get('date_to', '')
|
||||
project_filter = request.args.get('project', '')
|
||||
location_filter = request.args.get('location', '')
|
||||
employee_filter = request.args.get('employee', '')
|
||||
|
||||
# Build query - join with QRCode to access project_id
|
||||
query = AttendanceData.query.join(QRCode).filter(
|
||||
AttendanceData.verification_required == True
|
||||
)
|
||||
|
||||
if status_filter and status_filter != 'all':
|
||||
query = query.filter(AttendanceData.verification_status == status_filter)
|
||||
|
||||
if date_from:
|
||||
query = query.filter(AttendanceData.check_in_date >= date_from)
|
||||
|
||||
if date_to:
|
||||
query = query.filter(AttendanceData.check_in_date <= date_to)
|
||||
|
||||
# Apply project filter
|
||||
if project_filter:
|
||||
try:
|
||||
query = query.filter(QRCode.project_id == int(project_filter))
|
||||
except (ValueError, TypeError):
|
||||
pass
|
||||
|
||||
# Apply location filter
|
||||
if location_filter:
|
||||
query = query.filter(AttendanceData.location_name.ilike(f'%{location_filter}%'))
|
||||
|
||||
# Apply employee ID filter
|
||||
if employee_filter:
|
||||
query = query.filter(AttendanceData.employee_id.ilike(f'%{employee_filter}%'))
|
||||
|
||||
# Get records with QR code information
|
||||
verifications = query.order_by(
|
||||
AttendanceData.verification_timestamp.desc()
|
||||
).all()
|
||||
|
||||
# Build a dictionary for employee names lookup
|
||||
employee_names = {}
|
||||
for record in verifications:
|
||||
if record.employee_id and record.employee_id not in employee_names:
|
||||
try:
|
||||
employee = Employee.query.filter_by(id=int(record.employee_id)).first()
|
||||
if employee:
|
||||
employee_names[record.employee_id] = f"{employee.lastName}, {employee.firstName}"
|
||||
else:
|
||||
employee_names[record.employee_id] = None
|
||||
except (ValueError, TypeError):
|
||||
employee_names[record.employee_id] = None
|
||||
|
||||
# Build a dictionary for project names lookup
|
||||
project_names = {}
|
||||
for record in verifications:
|
||||
if record.qr_code and record.qr_code.project_id:
|
||||
project_id = record.qr_code.project_id
|
||||
if project_id not in project_names:
|
||||
try:
|
||||
project = db.session.get(Project, project_id)
|
||||
if project:
|
||||
project_names[project_id] = project.name
|
||||
else:
|
||||
project_names[project_id] = None
|
||||
except Exception:
|
||||
project_names[project_id] = None
|
||||
|
||||
# Get counts for status badges
|
||||
pending_count = AttendanceData.query.filter(
|
||||
AttendanceData.verification_status == 'pending'
|
||||
).count()
|
||||
|
||||
approved_count = AttendanceData.query.filter(
|
||||
AttendanceData.verification_status == 'approved'
|
||||
).count()
|
||||
|
||||
rejected_count = AttendanceData.query.filter(
|
||||
AttendanceData.verification_status == 'rejected'
|
||||
).count()
|
||||
|
||||
# Get all projects for filter dropdown
|
||||
projects = Project.query.filter_by(active_status=True).order_by(Project.name).all()
|
||||
|
||||
# Get unique locations for filter dropdown
|
||||
locations = db.session.query(AttendanceData.location_name).filter(
|
||||
AttendanceData.verification_required == True
|
||||
).distinct().order_by(AttendanceData.location_name).all()
|
||||
location_list = [loc[0] for loc in locations if loc[0]]
|
||||
|
||||
# Log access
|
||||
logger_handler.logger.info(
|
||||
f"User {session.get('username')} ({session.get('role')}) accessed verification review page"
|
||||
)
|
||||
|
||||
return render_template('verification_review.html',
|
||||
verifications=verifications,
|
||||
pending_count=pending_count,
|
||||
approved_count=approved_count,
|
||||
rejected_count=rejected_count,
|
||||
status_filter=status_filter,
|
||||
date_from=date_from,
|
||||
date_to=date_to,
|
||||
project_filter=project_filter,
|
||||
location_filter=location_filter,
|
||||
employee_filter=employee_filter,
|
||||
projects=projects,
|
||||
locations=location_list,
|
||||
employee_names=employee_names,
|
||||
project_names=project_names)
|
||||
|
||||
except Exception as e:
|
||||
logger_handler.logger.error(f"Error in verification review: {e}")
|
||||
flash('Error loading verification review.', 'error')
|
||||
return redirect(url_for('dashboard.dashboard'))
|
||||
|
||||
@bp.route('/verification-review/<int:record_id>/update', methods=['POST'], endpoint='update_verification_status')
|
||||
@login_required
|
||||
@log_database_operations('verification_update')
|
||||
def update_verification_status(record_id):
|
||||
"""Update verification status (approve/reject)"""
|
||||
try:
|
||||
# Only admins can update
|
||||
if session.get('role') not in ['admin', 'payroll', 'accounting']:
|
||||
return jsonify({
|
||||
'success': False,
|
||||
'message': 'Unauthorized access'
|
||||
}), 403
|
||||
|
||||
record = db.session.get(AttendanceData, record_id)
|
||||
if record is None:
|
||||
abort(404)
|
||||
|
||||
new_status = request.json.get('status')
|
||||
admin_note = request.json.get('note', '')
|
||||
|
||||
if new_status not in ['approved', 'rejected']:
|
||||
return jsonify({
|
||||
'success': False,
|
||||
'message': 'Invalid status'
|
||||
}), 400
|
||||
|
||||
# Update record
|
||||
record.verification_status = new_status
|
||||
record.edit_note = f"Verification {new_status} by {session.get('username')}. {admin_note}"
|
||||
|
||||
db.session.commit()
|
||||
|
||||
# Log the action
|
||||
logger_handler.log_photo_verification(
|
||||
employee_id=record.employee_id,
|
||||
qr_code_id=record.qr_code_id,
|
||||
distance=record.location_accuracy or 0,
|
||||
status=new_status
|
||||
)
|
||||
|
||||
return jsonify({
|
||||
'success': True,
|
||||
'message': f'Verification {new_status} successfully'
|
||||
})
|
||||
|
||||
except Exception as e:
|
||||
db.session.rollback()
|
||||
logger_handler.logger.error(f"Error updating verification: {e}")
|
||||
return jsonify({
|
||||
'success': False,
|
||||
'message': 'Error updating verification status'
|
||||
}), 500
|
||||
|
||||
@bp.route('/api/attendance/<int:record_id>/verification-details', endpoint='get_verification_details')
|
||||
@login_required
|
||||
def get_verification_details(record_id):
|
||||
"""API endpoint to get verification details for a specific record"""
|
||||
try:
|
||||
# Get the attendance record with verification data
|
||||
record = db.session.get(AttendanceData, record_id)
|
||||
if record is None:
|
||||
abort(404)
|
||||
|
||||
# DEBUG: Log record details
|
||||
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
|
||||
if session.get('role') not in ['admin', 'payroll', 'accounting']:
|
||||
return jsonify({
|
||||
'success': False,
|
||||
'message': 'Unauthorized access'
|
||||
}), 403
|
||||
|
||||
# Log the access for security audit
|
||||
logger_handler.logger.info(f"User {session.get('username')} ({session.get('role')}) accessed verification details for record {record_id}")
|
||||
|
||||
# Safely format dates/times with error handling
|
||||
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:
|
||||
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:
|
||||
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
|
||||
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:
|
||||
logger_handler.logger.debug(f"check_in_date strftime failed: {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:
|
||||
logger_handler.logger.debug(f"check_in_time strftime failed: {e}")
|
||||
check_in_time_str = str(record.check_in_time) if record.check_in_time else 'N/A'
|
||||
|
||||
record_data = {
|
||||
'id': record.id,
|
||||
'employee_id': record.employee_id,
|
||||
'location_name': record.location_name or 'Unknown',
|
||||
'check_in_date': check_in_date_str,
|
||||
'check_in_time': check_in_time_str,
|
||||
'location_accuracy': float(record.location_accuracy) if record.location_accuracy else None,
|
||||
'checked_in_address': record.address or 'No address',
|
||||
'verification_photo': record.verification_photo,
|
||||
'verification_status': record.verification_status,
|
||||
'verification_required': record.verification_required,
|
||||
'device_info': record.device_info or 'Unknown'
|
||||
}
|
||||
|
||||
return jsonify({
|
||||
'success': True,
|
||||
'record': record_data
|
||||
})
|
||||
|
||||
except Exception as e:
|
||||
logger_handler.logger.error(f"Error in get_verification_details for record {record_id}: {e}", exc_info=True)
|
||||
|
||||
return jsonify({
|
||||
'success': False,
|
||||
'message': 'Error loading verification details'
|
||||
}), 500
|
||||
|
||||
@bp.route('/verification-review/<int:record_id>', endpoint='verification_review_detail')
|
||||
@login_required
|
||||
def verification_review_detail(record_id):
|
||||
"""Review a single verification photo on a dedicated page"""
|
||||
try:
|
||||
# Check permissions
|
||||
if session.get('role') not in ['admin', 'payroll', 'accounting']:
|
||||
flash('Access denied. Only administrators, payroll, and accounting staff can review verification photos.', 'error')
|
||||
return redirect(url_for('attendance.attendance_report'))
|
||||
|
||||
# Get the attendance record
|
||||
record = db.session.get(AttendanceData, record_id)
|
||||
if record is None:
|
||||
abort(404)
|
||||
|
||||
# Check if this record has verification
|
||||
if not record.verification_required:
|
||||
flash('This record does not require verification.', 'warning')
|
||||
return redirect(url_for('attendance.attendance_report'))
|
||||
|
||||
# Get the QR code information for additional context
|
||||
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
|
||||
try:
|
||||
if record.employee_id:
|
||||
employee = Employee.query.filter_by(id=int(record.employee_id)).first()
|
||||
if employee:
|
||||
employee_name = f"{employee.lastName}, {employee.firstName}"
|
||||
else:
|
||||
employee_name = f"Unknown (ID: {record.employee_id})"
|
||||
except (ValueError, TypeError) as e:
|
||||
logger_handler.logger.warning(f"Could not lookup employee name for ID {record.employee_id}: {e}")
|
||||
employee_name = f"Unknown (ID: {record.employee_id})"
|
||||
|
||||
# Get event type from QR code (Check In/Check Out)
|
||||
location_event = qr_code.location_event if qr_code and qr_code.location_event else 'N/A'
|
||||
|
||||
# Log the access for audit trail
|
||||
logger_handler.logger.info(
|
||||
f"User {session.get('username')} ({session.get('role')}) "
|
||||
f"accessed verification review for record {record_id}"
|
||||
)
|
||||
|
||||
# Format date and time for display
|
||||
try:
|
||||
check_in_date = record.check_in_date.strftime('%m/%d/%Y') if record.check_in_date else 'N/A'
|
||||
except Exception as e:
|
||||
logger_handler.logger.debug(f"check_in_date strftime failed: {e}")
|
||||
check_in_date = str(record.check_in_date) if record.check_in_date else 'N/A'
|
||||
|
||||
try:
|
||||
check_in_time = record.check_in_time.strftime('%I:%M %p') if record.check_in_time else 'N/A'
|
||||
except Exception as e:
|
||||
logger_handler.logger.debug(f"check_in_time strftime failed: {e}")
|
||||
check_in_time = str(record.check_in_time) if record.check_in_time else 'N/A'
|
||||
|
||||
return render_template('verification_review_detail.html',
|
||||
record=record,
|
||||
qr_code=qr_code,
|
||||
check_in_date=check_in_date,
|
||||
check_in_time=check_in_time,
|
||||
employee_name=employee_name,
|
||||
location_event=location_event)
|
||||
|
||||
except Exception as e:
|
||||
logger_handler.logger.error(f"Error loading verification review detail: {e}")
|
||||
flash('Error loading verification details.', 'error')
|
||||
return redirect(url_for('attendance.attendance_report'))
|
||||
|
||||
@bp.route('/api/attendance/stats', endpoint='attendance_stats_api')
|
||||
@admin_required
|
||||
def attendance_stats_api():
|
||||
"""API endpoint for attendance statistics"""
|
||||
try:
|
||||
# Daily stats for the last 7 days
|
||||
daily_stats = db.session.execute(text("""
|
||||
SELECT
|
||||
check_in_date,
|
||||
COUNT(*) as checkins,
|
||||
COUNT(DISTINCT employee_id) as unique_employees
|
||||
FROM attendance_data
|
||||
WHERE check_in_date >= CURRENT_DATE - INTERVAL '7 days'
|
||||
GROUP BY check_in_date
|
||||
ORDER BY check_in_date DESC
|
||||
""")).fetchall()
|
||||
|
||||
# Location stats
|
||||
location_stats = db.session.execute(text("""
|
||||
SELECT
|
||||
location_name,
|
||||
COUNT(*) as total_checkins,
|
||||
COUNT(DISTINCT employee_id) as unique_employees
|
||||
FROM attendance_data
|
||||
GROUP BY location_name
|
||||
ORDER BY total_checkins DESC
|
||||
LIMIT 10
|
||||
""")).fetchall()
|
||||
|
||||
# Peak hours
|
||||
hourly_stats = db.session.execute(text("""
|
||||
SELECT
|
||||
EXTRACT(hour FROM check_in_time) as hour,
|
||||
COUNT(*) as checkins
|
||||
FROM attendance_data
|
||||
WHERE check_in_date >= CURRENT_DATE - INTERVAL '30 days'
|
||||
GROUP BY EXTRACT(hour FROM check_in_time)
|
||||
ORDER BY hour
|
||||
""")).fetchall()
|
||||
|
||||
return jsonify({
|
||||
'daily_stats': [{'date': str(row[0]), 'checkins': row[1], 'employees': row[2]} for row in daily_stats],
|
||||
'location_stats': [{'location': row[0], 'checkins': row[1], 'employees': row[2]} for row in location_stats],
|
||||
'hourly_stats': [{'hour': int(row[0]), 'checkins': row[1]} for row in hourly_stats]
|
||||
})
|
||||
|
||||
except Exception as e:
|
||||
logger_handler.logger.error(f"Error fetching attendance stats: {e}", exc_info=True)
|
||||
return jsonify({'error': 'Failed to fetch attendance statistics'}), 500
|
||||
Reference in New Issue
Block a user