867 lines
41 KiB
Python
867 lines
41 KiB
Python
"""
|
|
routes/attendance.py
|
|
====================
|
|
Attendance check-in records, manual entry, verification review,
|
|
export configuration, and Excel export routes.
|
|
|
|
Routes: /attendance, /attendance/<id>/edit, /attendance/add,
|
|
/attendance/save_manual, /api/attendance/*, /api/search_employees,
|
|
/api/get_project_locations, /verification-review/*,
|
|
/export-configuration, /generate-excel-export
|
|
"""
|
|
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,
|
|
expand_employee_id_filter,
|
|
get_base_employee_id,
|
|
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
|
|
|
|
bp = Blueprint('attendance', __name__)
|
|
|
|
# Case-folded employee_id, matched against the REGEXP patterns from
|
|
# build_employee_id_regex() so any separator style ("1234 SP", "1234.PW",
|
|
# "1234-PT", "SP1234") is found regardless of the column's collation.
|
|
UPPER_EMPLOYEE_ID_SQL = "UPPER(ad.employee_id)"
|
|
|
|
|
|
|
|
@bp.route('/attendance', endpoint='attendance_report')
|
|
@login_required
|
|
def attendance_report():
|
|
"""Safe attendance report with backward compatibility for location_accuracy and fixed datetime handling"""
|
|
try:
|
|
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:
|
|
pass
|
|
|
|
# Check if location_accuracy column exists
|
|
has_location_accuracy = check_location_accuracy_column_exists()
|
|
logger_handler.logger.debug(f"Location accuracy column exists: {has_location_accuracy}")
|
|
|
|
# Get filter parameters
|
|
date_from = request.args.get('date_from', '')
|
|
date_to = request.args.get('date_to', '')
|
|
location_filter = request.args.get('location', '')
|
|
# employee param is now a comma-separated list of IDs (multi-employee filter)
|
|
employee_filter = request.args.get('employee', '')
|
|
project_filter = request.args.get('project', '')
|
|
|
|
# Build the list of selected employee IDs (strip blanks)
|
|
employee_ids = [e.strip() for e in employee_filter.split(',') if e.strip()] if employee_filter else []
|
|
|
|
# Build display names for each selected employee
|
|
employee_display_names = []
|
|
for eid in employee_ids:
|
|
try:
|
|
# Use the base ID so work-type IDs ("1234SP") still resolve a name
|
|
emp = Employee.query.filter_by(id=int(get_base_employee_id(eid))).first()
|
|
if emp:
|
|
employee_display_names.append({
|
|
'id': eid,
|
|
'name': f"{emp.lastName}, {emp.firstName}"
|
|
})
|
|
else:
|
|
employee_display_names.append({'id': eid, 'name': f"ID: {eid}"})
|
|
except (ValueError, TypeError):
|
|
employee_display_names.append({'id': eid, 'name': eid})
|
|
|
|
# Legacy single-value display name (kept for backward compat in template)
|
|
employee_display_name = ', '.join([e['name'] for e in employee_display_names])
|
|
|
|
# ============================================================
|
|
# PROJECT MANAGER ACCESS CONTROL
|
|
# ============================================================
|
|
user_role = session.get('role')
|
|
user_id = session.get('user_id')
|
|
|
|
# Initialize permission filters
|
|
allowed_project_ids = []
|
|
allowed_location_names = []
|
|
|
|
# Check if user is Project Manager and get their permissions
|
|
if user_role == 'project_manager':
|
|
logger_handler.logger.debug(f"Project Manager access control enabled for user {session.get('username')}")
|
|
|
|
try:
|
|
# Get assigned projects
|
|
assigned_projects = UserProjectPermission.query.filter_by(user_id=user_id).all()
|
|
allowed_project_ids = [p.project_id for p in assigned_projects]
|
|
|
|
# Get assigned locations
|
|
assigned_locations = UserLocationPermission.query.filter_by(user_id=user_id).all()
|
|
allowed_location_names = [l.location_name for l in assigned_locations]
|
|
|
|
# Log the permissions
|
|
logger_handler.logger.info(
|
|
f"🔒 Project Manager {session.get('username')} restricted to: "
|
|
f"Projects: {allowed_project_ids}, Locations: {allowed_location_names}"
|
|
)
|
|
|
|
logger_handler.logger.debug(f"PM allowed projects: {allowed_project_ids}, locations: {allowed_location_names}")
|
|
except Exception as 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
|
|
if not allowed_project_ids and not allowed_location_names:
|
|
logger_handler.logger.warning(
|
|
f"Project Manager {session.get('username')} has no assigned projects or locations"
|
|
)
|
|
flash('You do not have access to any projects or locations. Please contact an administrator.', 'warning')
|
|
|
|
# Create empty stats object using named tuple style
|
|
from collections import namedtuple
|
|
Stats = namedtuple('Stats', ['total_checkins', 'unique_employees', 'active_locations',
|
|
'today_checkins', 'records_with_gps', 'records_with_accuracy',
|
|
'avg_location_accuracy'])
|
|
empty_stats = Stats(0, 0, 0, 0, 0, 0, 0)
|
|
|
|
# Return empty template
|
|
return render_template('attendance_report.html',
|
|
attendance_records=[],
|
|
locations=[],
|
|
projects=[],
|
|
stats=empty_stats,
|
|
date_from=date_from,
|
|
date_to=date_to,
|
|
location_filter=location_filter,
|
|
employee_filter=employee_filter,
|
|
employee_ids=employee_ids,
|
|
employee_display_names=employee_display_names,
|
|
employee_display_name=employee_display_name,
|
|
project_filter=project_filter,
|
|
today_date=datetime.now().strftime('%Y-%m-%d'),
|
|
current_date_formatted=datetime.now().strftime('%B %d'),
|
|
has_location_accuracy_feature=has_location_accuracy,
|
|
user_role=user_role)
|
|
|
|
# ============================================================
|
|
# END: PROJECT MANAGER ACCESS CONTROL
|
|
# ============================================================
|
|
|
|
# Build base query - conditional based on column existence
|
|
if has_location_accuracy:
|
|
# New query with location accuracy
|
|
base_query = """
|
|
SELECT
|
|
ad.id,
|
|
ad.employee_id,
|
|
ad.check_in_date,
|
|
ad.check_in_time,
|
|
ad.location_name,
|
|
qc.location_event,
|
|
COALESCE(ad.qr_address, qc.location_address) as qr_address,
|
|
ad.address as checked_in_address,
|
|
ad.latitude,
|
|
ad.longitude,
|
|
ad.location_accuracy,
|
|
ad.accuracy as gps_accuracy,
|
|
ad.device_info,
|
|
ad.created_timestamp,
|
|
ad.updated_timestamp,
|
|
CONCAT(e.firstName, ' ', e.lastName) as employee_name,
|
|
ad.verification_required,
|
|
ad.verification_status,
|
|
ad.verification_photo,
|
|
COALESCE(ad.is_dynamic_qr, 0) as is_dynamic_qr
|
|
FROM attendance_data ad
|
|
LEFT JOIN qr_codes qc ON ad.qr_code_id = qc.id
|
|
LEFT JOIN employee e ON CAST(ad.employee_id AS UNSIGNED) = e.id
|
|
WHERE 1=1
|
|
"""
|
|
else:
|
|
# Fallback query without location accuracy
|
|
base_query = """
|
|
SELECT
|
|
ad.id,
|
|
ad.employee_id,
|
|
ad.check_in_date,
|
|
ad.check_in_time,
|
|
ad.location_name,
|
|
qc.location_event,
|
|
COALESCE(ad.qr_address, qc.location_address) as qr_address,
|
|
ad.address as checked_in_address,
|
|
ad.latitude,
|
|
ad.longitude,
|
|
NULL as location_accuracy,
|
|
ad.accuracy as gps_accuracy,
|
|
ad.device_info,
|
|
ad.created_timestamp,
|
|
ad.updated_timestamp,
|
|
CONCAT(e.firstName, ' ', e.lastName) as employee_name,
|
|
ad.verification_required,
|
|
ad.verification_status,
|
|
ad.verification_photo,
|
|
COALESCE(ad.is_dynamic_qr, 0) as is_dynamic_qr
|
|
FROM attendance_data ad
|
|
LEFT JOIN qr_codes qc ON ad.qr_code_id = qc.id
|
|
LEFT JOIN employee e ON CAST(ad.employee_id AS UNSIGNED) = e.id
|
|
WHERE 1=1
|
|
"""
|
|
|
|
# Prepare filter conditions and parameters
|
|
filter_conditions = []
|
|
query_params = {}
|
|
|
|
# ============================================================
|
|
# APPLY PROJECT MANAGER FILTERS TO SQL QUERY
|
|
# ============================================================
|
|
if user_role == 'project_manager':
|
|
# Filter by allowed projects
|
|
if allowed_project_ids:
|
|
project_placeholders = ','.join([f':project_{i}' for i in range(len(allowed_project_ids))])
|
|
filter_conditions.append(f"qc.project_id IN ({project_placeholders})")
|
|
for i, pid in enumerate(allowed_project_ids):
|
|
query_params[f'project_{i}'] = pid
|
|
|
|
# Filter by allowed locations
|
|
if allowed_location_names:
|
|
location_placeholders = ','.join([f':location_{i}' for i in range(len(allowed_location_names))])
|
|
filter_conditions.append(f"ad.location_name IN ({location_placeholders})")
|
|
for i, loc in enumerate(allowed_location_names):
|
|
query_params[f'location_{i}'] = loc
|
|
# ============================================================
|
|
# END: APPLY PROJECT MANAGER FILTERS
|
|
# ============================================================
|
|
|
|
# Apply user-selected filters
|
|
if date_from:
|
|
filter_conditions.append("ad.check_in_date >= :date_from")
|
|
query_params['date_from'] = date_from
|
|
|
|
if date_to:
|
|
filter_conditions.append("ad.check_in_date <= :date_to")
|
|
query_params['date_to'] = date_to
|
|
|
|
if location_filter:
|
|
# Exact match — dropdown value IS the exact location_name string
|
|
filter_conditions.append("ad.location_name = :location")
|
|
query_params['location'] = location_filter
|
|
|
|
if employee_ids:
|
|
# Expand each selected ID into every stored spelling so extra-work
|
|
# check-ins (SP / PW / PT) are included alongside regular records.
|
|
# Two branches: exact match on the raw column (uses the index), plus a
|
|
# REGEXP match that catches any separator style ("1234.PW", "1234-SP").
|
|
exact_variants, regex_patterns = expand_employee_id_filter(employee_ids)
|
|
# Never emit an empty IN () — fall back to the raw selection if expansion
|
|
# somehow produced nothing, so the filter can't degrade into "match all".
|
|
if not exact_variants:
|
|
exact_variants = list(employee_ids)
|
|
|
|
exact_placeholders = ', '.join([f':employee_{i}' for i in range(len(exact_variants))])
|
|
exact_condition = f"ad.employee_id IN ({exact_placeholders})"
|
|
for i, variant in enumerate(exact_variants):
|
|
query_params[f'employee_{i}'] = variant
|
|
|
|
if regex_patterns:
|
|
regex_clauses = ' OR '.join([
|
|
f"{UPPER_EMPLOYEE_ID_SQL} REGEXP :employee_re_{i}"
|
|
for i in range(len(regex_patterns))
|
|
])
|
|
filter_conditions.append(f"({exact_condition} OR {regex_clauses})")
|
|
for i, pattern in enumerate(regex_patterns):
|
|
query_params[f'employee_re_{i}'] = pattern
|
|
else:
|
|
filter_conditions.append(exact_condition)
|
|
logger_handler.logger.info(
|
|
f"Attendance report filtered by employee IDs: {employee_ids} "
|
|
f"(matching {len(exact_variants)} ID variants incl. SP/PW/PT) "
|
|
f"by user {session.get('username', 'unknown')}"
|
|
)
|
|
|
|
if project_filter:
|
|
# For standard QR records: match by the QR code's project_id directly.
|
|
# For dynamic QR records: the dynamic QR itself may not be in any project,
|
|
# but the employee-selected location corresponds to a standard QR in that
|
|
# project. Match those by checking if attendance_data.location_name
|
|
# appears in the locations of QR codes belonging to the selected project.
|
|
filter_conditions.append(
|
|
"(qc.project_id = :project OR "
|
|
"(ad.is_dynamic_qr = 1 AND ad.location_name IN ("
|
|
" SELECT DISTINCT qc2.location FROM qr_codes qc2 "
|
|
" WHERE qc2.project_id = :project AND qc2.qr_type = 'standard' "
|
|
" AND qc2.location IS NOT NULL AND qc2.location != ''"
|
|
")))"
|
|
)
|
|
query_params['project'] = project_filter
|
|
|
|
# Combine query with filters
|
|
if filter_conditions:
|
|
base_query += " AND " + " AND ".join(filter_conditions)
|
|
|
|
# Fetch one extra record to detect truncation without a separate COUNT query
|
|
ATTENDANCE_PAGE_LIMIT = 1000
|
|
base_query += f" ORDER BY ad.check_in_date DESC, ad.check_in_time DESC LIMIT {ATTENDANCE_PAGE_LIMIT + 1}"
|
|
|
|
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()
|
|
# If we got more than the limit, the result set is truncated
|
|
records_truncated = len(records) > ATTENDANCE_PAGE_LIMIT
|
|
if records_truncated:
|
|
records = records[:ATTENDANCE_PAGE_LIMIT]
|
|
logger_handler.logger.debug(f"Loaded {len(records)} attendance records (truncated={records_truncated})")
|
|
|
|
# Process records
|
|
processed_records = []
|
|
for record in records:
|
|
try:
|
|
record_dict = {
|
|
'id': record[0],
|
|
'employee_id': record[1],
|
|
'check_in_date': record[2],
|
|
'check_in_time': record[3],
|
|
'location_name': record[4],
|
|
'location_event': record[5],
|
|
'qr_address': record[6],
|
|
'checked_in_address': record[7],
|
|
'latitude': record[8],
|
|
'longitude': record[9],
|
|
'location_accuracy': record[10] if has_location_accuracy else None,
|
|
'gps_accuracy': record[11],
|
|
'device_info': record[12],
|
|
'created_timestamp': record[13],
|
|
'updated_timestamp': record[14],
|
|
'employee_name': record[15] or 'Unknown Employee',
|
|
'verification_required': record[16] if len(record) > 16 else False,
|
|
'verification_status': record[17] if len(record) > 17 else None,
|
|
'verification_photo': record[18] if len(record) > 18 else None,
|
|
'is_dynamic_qr': bool(record[19]) if len(record) > 19 else False
|
|
}
|
|
|
|
# Calculate accuracy_level for template display
|
|
if record_dict['location_accuracy'] is not None:
|
|
accuracy_value = float(record_dict['location_accuracy'])
|
|
if accuracy_value <= 0.3:
|
|
record_dict['accuracy_level'] = 'accurate'
|
|
else:
|
|
record_dict['accuracy_level'] = 'inaccurate'
|
|
else:
|
|
record_dict['accuracy_level'] = 'unknown'
|
|
processed_records.append(record_dict)
|
|
except Exception as rec_error:
|
|
logger_handler.logger.warning(f"Error processing attendance record: {rec_error}")
|
|
continue
|
|
|
|
# Get unique locations for filter dropdown
|
|
try:
|
|
# ============================================================
|
|
# FILTER LOCATIONS FOR PROJECT MANAGER
|
|
# ============================================================
|
|
if user_role == 'project_manager' and allowed_location_names:
|
|
# Only show locations the PM has access to
|
|
locations = sorted(allowed_location_names)
|
|
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("""
|
|
SELECT DISTINCT location_name
|
|
FROM attendance_data
|
|
WHERE location_name IS NOT NULL
|
|
AND location_name != 'Dynamic'
|
|
AND location_name != ''
|
|
ORDER BY location_name
|
|
"""))
|
|
locations = [row[0] for row in locations_query.fetchall()]
|
|
logger_handler.logger.debug(f"Found {len(locations)} unique locations")
|
|
# ============================================================
|
|
# END: FILTER LOCATIONS FOR PROJECT MANAGER
|
|
# ============================================================
|
|
except Exception as e:
|
|
logger_handler.logger.warning(f"Error loading locations filter: {e}")
|
|
locations = []
|
|
|
|
# Get projects for filter dropdown
|
|
try:
|
|
# ============================================================
|
|
# FILTER PROJECTS FOR PROJECT MANAGER
|
|
# ============================================================
|
|
if user_role == 'project_manager' and allowed_project_ids:
|
|
# Only show projects the PM has access to
|
|
project_placeholders = ','.join([str(pid) for pid in allowed_project_ids])
|
|
projects_query = db.session.execute(text(f"""
|
|
SELECT p.id, p.name, COUNT(DISTINCT ad.id) as attendance_count
|
|
FROM projects p
|
|
LEFT JOIN qr_codes qc ON qc.project_id = p.id
|
|
LEFT JOIN attendance_data ad ON ad.qr_code_id = qc.id
|
|
WHERE p.active_status = true AND p.id IN ({project_placeholders})
|
|
GROUP BY p.id, p.name
|
|
ORDER BY p.name
|
|
"""))
|
|
projects = projects_query.fetchall()
|
|
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("""
|
|
SELECT p.id, p.name, COUNT(DISTINCT ad.id) as attendance_count
|
|
FROM projects p
|
|
LEFT JOIN qr_codes qc ON qc.project_id = p.id
|
|
LEFT JOIN attendance_data ad ON ad.qr_code_id = qc.id
|
|
WHERE p.active_status = true
|
|
GROUP BY p.id, p.name
|
|
HAVING COUNT(DISTINCT ad.id) > 0
|
|
ORDER BY p.name
|
|
""")).fetchall()
|
|
logger_handler.logger.debug(f"Loaded {len(projects)} projects with attendance data")
|
|
# ============================================================
|
|
# END: FILTER PROJECTS FOR PROJECT MANAGER
|
|
# ============================================================
|
|
except Exception as e:
|
|
logger_handler.logger.warning(f"Error loading projects filter: {e}")
|
|
projects = []
|
|
|
|
# ============================================================
|
|
# STATISTICS - COMPLETELY REWRITTEN FOR SAFETY
|
|
# ============================================================
|
|
logger_handler.logger.debug("Loading attendance statistics")
|
|
|
|
# Create simple dict for stats (most compatible approach)
|
|
stats_dict = {
|
|
'total_checkins': 0,
|
|
'unique_employees': 0,
|
|
'active_locations': 0,
|
|
'today_checkins': 0,
|
|
'records_with_gps': 0,
|
|
'records_with_accuracy': 0,
|
|
'avg_location_accuracy': 0.0
|
|
}
|
|
|
|
try:
|
|
# Build stats query
|
|
if has_location_accuracy:
|
|
stats_select = """
|
|
SELECT
|
|
COALESCE(COUNT(*), 0) as total_checkins,
|
|
COALESCE(COUNT(DISTINCT employee_id), 0) as unique_employees,
|
|
COALESCE(COUNT(DISTINCT qr_code_id), 0) as active_locations,
|
|
COALESCE(COUNT(CASE WHEN check_in_date = CURRENT_DATE THEN 1 END), 0) as today_checkins,
|
|
COALESCE(COUNT(CASE WHEN latitude IS NOT NULL AND longitude IS NOT NULL THEN 1 END), 0) as records_with_gps,
|
|
COALESCE(COUNT(CASE WHEN location_accuracy IS NOT NULL THEN 1 END), 0) as records_with_accuracy,
|
|
COALESCE(AVG(location_accuracy), 0) as avg_location_accuracy
|
|
"""
|
|
else:
|
|
stats_select = """
|
|
SELECT
|
|
COALESCE(COUNT(*), 0) as total_checkins,
|
|
COALESCE(COUNT(DISTINCT employee_id), 0) as unique_employees,
|
|
COALESCE(COUNT(DISTINCT qr_code_id), 0) as active_locations,
|
|
COALESCE(COUNT(CASE WHEN check_in_date = CURRENT_DATE THEN 1 END), 0) as today_checkins,
|
|
COALESCE(COUNT(CASE WHEN latitude IS NOT NULL AND longitude IS NOT NULL THEN 1 END), 0) as records_with_gps,
|
|
0 as records_with_accuracy,
|
|
0 as avg_location_accuracy
|
|
"""
|
|
|
|
stats_query_text = stats_select + " FROM attendance_data ad"
|
|
stats_params = {}
|
|
|
|
# Add filters for Project Manager
|
|
if user_role == 'project_manager':
|
|
stats_query_text += " LEFT JOIN qr_codes qc ON ad.qr_code_id = qc.id WHERE 1=1"
|
|
|
|
stats_conditions = []
|
|
|
|
if allowed_project_ids:
|
|
project_placeholders = ','.join([f':stat_project_{i}' for i in range(len(allowed_project_ids))])
|
|
stats_conditions.append(f"qc.project_id IN ({project_placeholders})")
|
|
for i, pid in enumerate(allowed_project_ids):
|
|
stats_params[f'stat_project_{i}'] = pid
|
|
|
|
if allowed_location_names:
|
|
location_placeholders = ','.join([f':stat_location_{i}' for i in range(len(allowed_location_names))])
|
|
stats_conditions.append(f"ad.location_name IN ({location_placeholders})")
|
|
for i, loc in enumerate(allowed_location_names):
|
|
stats_params[f'stat_location_{i}'] = loc
|
|
|
|
if stats_conditions:
|
|
stats_query_text += " AND " + " AND ".join(stats_conditions)
|
|
|
|
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()
|
|
|
|
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:
|
|
try:
|
|
stats_dict['total_checkins'] = int(stats_row[0]) if stats_row[0] is not None else 0
|
|
stats_dict['unique_employees'] = int(stats_row[1]) if stats_row[1] is not None else 0
|
|
stats_dict['active_locations'] = int(stats_row[2]) if stats_row[2] is not None else 0
|
|
stats_dict['today_checkins'] = int(stats_row[3]) if stats_row[3] is not None else 0
|
|
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
|
|
logger_handler.logger.debug(f"Loaded statistics: {stats_dict['total_checkins']} total check-ins")
|
|
except (IndexError, TypeError, ValueError) as extract_error:
|
|
logger_handler.logger.warning(f"Error extracting stats values: {extract_error}")
|
|
# stats_dict already has default values
|
|
else:
|
|
logger_handler.logger.warning("Stats query returned None or insufficient columns, using default stats")
|
|
|
|
except Exception as stats_error:
|
|
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
|
|
class StatsObject:
|
|
def __init__(self, stats_dict):
|
|
for key, value in stats_dict.items():
|
|
setattr(self, key, value)
|
|
|
|
stats = StatsObject(stats_dict)
|
|
logger_handler.logger.debug(f"Stats object created: total_checkins={stats.total_checkins}")
|
|
|
|
# ============================================================
|
|
# END: STATISTICS
|
|
# ============================================================
|
|
|
|
# Add today's date for template
|
|
today_date = datetime.now().strftime('%Y-%m-%d')
|
|
current_date_formatted = datetime.now().strftime('%B %d')
|
|
|
|
logger_handler.logger.debug("Rendering attendance report template")
|
|
|
|
return render_template('attendance_report.html',
|
|
attendance_records=processed_records,
|
|
records_truncated=records_truncated,
|
|
records_limit=ATTENDANCE_PAGE_LIMIT,
|
|
locations=locations,
|
|
projects=projects,
|
|
stats=stats,
|
|
date_from=date_from,
|
|
date_to=date_to,
|
|
location_filter=location_filter,
|
|
employee_filter=employee_filter,
|
|
employee_ids=employee_ids,
|
|
employee_display_names=employee_display_names,
|
|
employee_display_name=employee_display_name,
|
|
project_filter=project_filter,
|
|
today_date=datetime.now().strftime('%Y-%m-%d'),
|
|
current_date_formatted=datetime.now().strftime('%B %d'),
|
|
has_location_accuracy_feature=has_location_accuracy,
|
|
user_role=user_role)
|
|
|
|
except Exception as e:
|
|
logger_handler.logger.error(f"Error loading attendance report: {e}", exc_info=True)
|
|
|
|
error_traceback = traceback.format_exc()
|
|
|
|
|
|
# Log the error
|
|
try:
|
|
logger_handler.log_database_error('attendance_report', e)
|
|
except Exception as 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'))
|
|
|
|
@bp.route('/api/time-attendance/locations', endpoint='time_attendance_locations_api')
|
|
@login_required
|
|
def time_attendance_locations_api():
|
|
"""Return distinct location_name values from time_attendance, optionally filtered by project_id.
|
|
Used by the time attendance records page to dynamically scope the location dropdown."""
|
|
try:
|
|
project_id = request.args.get('project_id', '').strip()
|
|
|
|
if project_id:
|
|
try:
|
|
project_id_int = int(project_id)
|
|
except (ValueError, TypeError):
|
|
return jsonify({'success': False, 'error': 'Invalid project_id'}), 400
|
|
|
|
result = db.session.execute(text("""
|
|
SELECT DISTINCT location_name
|
|
FROM time_attendance
|
|
WHERE project_id = :project_id
|
|
AND location_name IS NOT NULL
|
|
ORDER BY location_name
|
|
"""), {'project_id': project_id_int})
|
|
else:
|
|
result = db.session.execute(text("""
|
|
SELECT DISTINCT location_name
|
|
FROM time_attendance
|
|
WHERE location_name IS NOT NULL
|
|
ORDER BY location_name
|
|
"""))
|
|
|
|
locations = [row[0] for row in result.fetchall()]
|
|
logger_handler.logger.info(
|
|
f"User {session.get('username', 'unknown')} fetched time attendance locations"
|
|
+ (f" for project_id={project_id}" if project_id else " (all projects)")
|
|
)
|
|
return jsonify({'success': True, 'locations': locations})
|
|
|
|
except Exception as e:
|
|
logger_handler.logger.error(f"Error in time_attendance_locations_api: {e}")
|
|
return jsonify({'success': False, 'error': str(e)}), 500
|
|
|
|
|
|
@bp.route('/api/attendance/locations', endpoint='attendance_locations_api')
|
|
@login_required
|
|
def attendance_locations_api():
|
|
"""Return distinct location_name values from attendance_data, optionally filtered by project_id.
|
|
Used by the attendance report page to dynamically scope the location dropdown when a project is selected."""
|
|
try:
|
|
project_id = request.args.get('project_id', '').strip()
|
|
|
|
if project_id:
|
|
try:
|
|
project_id_int = int(project_id)
|
|
except (ValueError, TypeError):
|
|
return jsonify({'success': False, 'error': 'Invalid project_id'}), 400
|
|
|
|
result = db.session.execute(text("""
|
|
SELECT DISTINCT ad.location_name
|
|
FROM attendance_data ad
|
|
INNER JOIN qr_codes qc ON ad.qr_code_id = qc.id
|
|
WHERE qc.project_id = :project_id
|
|
AND ad.location_name IS NOT NULL
|
|
ORDER BY ad.location_name
|
|
"""), {'project_id': project_id_int})
|
|
else:
|
|
result = db.session.execute(text("""
|
|
SELECT DISTINCT location_name
|
|
FROM attendance_data
|
|
WHERE location_name IS NOT NULL
|
|
ORDER BY location_name
|
|
"""))
|
|
|
|
locations = [row[0] for row in result.fetchall()]
|
|
logger_handler.logger.info(
|
|
f"User {session.get('username', 'unknown')} fetched attendance locations"
|
|
+ (f" for project_id={project_id}" if project_id else " (all projects)")
|
|
)
|
|
return jsonify({'success': True, 'locations': locations})
|
|
|
|
except Exception as e:
|
|
logger_handler.logger.error(f"Error in attendance_locations_api: {e}")
|
|
return jsonify({'success': False, 'error': str(e)}), 500
|
|
|
|
|
|
@bp.route('/api/search_employees', endpoint='search_employees_api')
|
|
@login_required
|
|
def search_employees_api():
|
|
"""
|
|
API endpoint to search employees by name or ID.
|
|
Returns matches from the Employee table first, then appends any IDs found
|
|
in attendance_data that have no Employee record — so unregistered IDs
|
|
that have attendance records can still be filtered on the attendance page.
|
|
"""
|
|
try:
|
|
search_query = request.args.get('q', '').strip()
|
|
|
|
if not search_query or len(search_query) < 2:
|
|
return jsonify({'employees': []})
|
|
|
|
search_pattern = f"%{search_query}%"
|
|
|
|
# 1. Registered employees — search by ID or name
|
|
employees = Employee.query.filter(
|
|
db.or_(
|
|
Employee.id.like(search_pattern),
|
|
Employee.firstName.like(search_pattern),
|
|
Employee.lastName.like(search_pattern),
|
|
db.func.concat(Employee.firstName, ' ', Employee.lastName).like(search_pattern)
|
|
)
|
|
).limit(10).all()
|
|
|
|
employee_list = [{
|
|
'id': emp.id,
|
|
'firstName': emp.firstName,
|
|
'lastName': emp.lastName,
|
|
'full_name': f"{emp.firstName} {emp.lastName}"
|
|
} for emp in employees]
|
|
|
|
registered_ids = {str(emp.id) for emp in employees}
|
|
|
|
# 2. Unregistered IDs — present in attendance_data but not in Employee table.
|
|
# Only add when the search term looks like (part of) a numeric ID and we
|
|
# still have room in the result list.
|
|
if len(employee_list) < 10:
|
|
remaining_slots = 10 - len(employee_list)
|
|
try:
|
|
unregistered_rows = db.session.execute(
|
|
text("""
|
|
SELECT DISTINCT ad.employee_id
|
|
FROM attendance_data ad
|
|
LEFT JOIN employee e ON CAST(ad.employee_id AS UNSIGNED) = e.id
|
|
WHERE e.id IS NULL
|
|
AND ad.employee_id LIKE :pattern
|
|
ORDER BY ad.employee_id
|
|
LIMIT :lim
|
|
"""),
|
|
{'pattern': search_pattern, 'lim': remaining_slots}
|
|
).fetchall()
|
|
|
|
for row in unregistered_rows:
|
|
emp_id = str(row[0])
|
|
if emp_id not in registered_ids:
|
|
employee_list.append({
|
|
'id': emp_id,
|
|
'firstName': f'ID: {emp_id}',
|
|
'lastName': '(no record)',
|
|
'full_name': f'ID: {emp_id} (no record)'
|
|
})
|
|
except Exception as unreg_err:
|
|
logger_handler.logger.warning(f"Could not search unregistered employee IDs: {unreg_err}")
|
|
|
|
return jsonify({'employees': employee_list})
|
|
|
|
except Exception as e:
|
|
logger_handler.logger.error(f"Error searching employees: {e}")
|
|
return jsonify({'employees': [], 'error': str(e)}), 500
|
|
|
|
|
|
@bp.route('/api/get_project_locations', endpoint='get_project_locations_api')
|
|
@login_required
|
|
def get_project_locations_api():
|
|
"""
|
|
API endpoint to get locations for a specific project
|
|
Returns JSON with location list
|
|
"""
|
|
try:
|
|
project_id = request.args.get('project_id', '').strip()
|
|
|
|
if not project_id:
|
|
return jsonify({'success': False, 'locations': [], 'error': 'Project ID required'})
|
|
|
|
# Get active QR codes for this project
|
|
qr_codes = QRCode.query.filter_by(
|
|
project_id=int(project_id),
|
|
active_status=True
|
|
).order_by(QRCode.location).all()
|
|
|
|
# Group QR codes by location to get unique locations
|
|
locations_dict = {}
|
|
for qr in qr_codes:
|
|
location_key = f"{qr.location}||{qr.location_address}"
|
|
|
|
if location_key not in locations_dict:
|
|
locations_dict[location_key] = {
|
|
'location': qr.location,
|
|
'location_address': qr.location_address,
|
|
'qr_codes': {}
|
|
}
|
|
|
|
# Store QR code ID for each event type
|
|
locations_dict[location_key]['qr_codes'][qr.location_event] = qr.id
|
|
|
|
# Convert to list format
|
|
location_list = [{
|
|
'location': loc_data['location'],
|
|
'location_address': loc_data['location_address'],
|
|
'qr_codes': loc_data['qr_codes']
|
|
} for loc_data in locations_dict.values()]
|
|
|
|
return jsonify({'success': True, 'locations': location_list})
|
|
|
|
except Exception as e:
|
|
logger_handler.logger.error(f"Error getting project locations: {e}")
|
|
return jsonify({'success': False, 'locations': [], 'error': str(e)}), 500
|
|
|
|
@bp.route('/attendance/<int:record_id>/delete', methods=['POST'], endpoint='delete_attendance')
|
|
@login_required
|
|
@log_database_operations('attendance_delete')
|
|
def delete_attendance(record_id):
|
|
"""Delete attendance record (Admin and Payroll only)"""
|
|
# Check if user has permission to delete attendance records
|
|
if session.get('role') not in ['admin', 'payroll', 'accounting']:
|
|
if request.headers.get('X-Requested-With') == 'XMLHttpRequest':
|
|
return jsonify({
|
|
'success': False,
|
|
'message': 'Access denied. Only administrators and payroll staff can delete attendance records.'
|
|
}), 403
|
|
else:
|
|
flash('Access denied. Only administrators and payroll staff can delete attendance records.', 'error')
|
|
return redirect(url_for('attendance.attendance_report'))
|
|
|
|
try:
|
|
attendance_record = db.session.get(AttendanceData, record_id)
|
|
if attendance_record is None:
|
|
abort(404)
|
|
|
|
# Store record info for logging before deletion
|
|
employee_id = attendance_record.employee_id
|
|
location_name = attendance_record.location_name
|
|
check_in_date = attendance_record.check_in_date
|
|
|
|
# Log the deletion
|
|
logger_handler.log_security_event(
|
|
event_type="attendance_record_deletion",
|
|
description=f"{session.get('role', 'unknown').title()} {session.get('username')} deleted attendance record {record_id}",
|
|
severity="HIGH",
|
|
additional_data={
|
|
'record_id': record_id,
|
|
'employee_id': employee_id,
|
|
'location_name': location_name,
|
|
'check_in_date': str(check_in_date),
|
|
'user_role': session.get('role')
|
|
}
|
|
)
|
|
|
|
# Delete the record
|
|
db.session.delete(attendance_record)
|
|
db.session.commit()
|
|
|
|
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':
|
|
return jsonify({
|
|
'success': True,
|
|
'message': f'Attendance record for {employee_id} deleted successfully!'
|
|
})
|
|
else:
|
|
flash(f'Attendance record for {employee_id} deleted successfully!', 'success')
|
|
return redirect(url_for('attendance.attendance_report'))
|
|
|
|
except Exception as e:
|
|
db.session.rollback()
|
|
logger_handler.log_database_error('attendance_delete', 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({
|
|
'success': False,
|
|
'message': 'Error deleting attendance record. Please try again.'
|
|
}), 500
|
|
else:
|
|
flash('Error deleting attendance record. Please try again.', 'error')
|
|
return redirect(url_for('attendance.attendance_report'))
|
|
|