399 lines
19 KiB
Python
399 lines
19 KiB
Python
"""
|
|
routes/attendance_edit.py
|
|
=========================
|
|
Attendance record edit, manual entry, and delete routes.
|
|
|
|
Routes: /attendance/<id>/edit, /attendance/add,
|
|
/attendance/save_manual, /attendance/<id>/delete
|
|
|
|
"""
|
|
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
|
|
|
|
# Employee IDs typed on the edit / manual-add forms follow the same rule as a
|
|
# check-in: 1 to 4 digits, optionally carrying a work-type code.
|
|
MANUAL_EMPLOYEE_ID_MAX_DIGITS = 4
|
|
|
|
|
|
def _normalize_manual_employee_id(raw_employee_id):
|
|
"""
|
|
(canonical_id, base_id, work_type) for an ID typed on a form, or
|
|
(None, None, None) when it is not a valid employee ID.
|
|
|
|
"1234" -> ("1234", "1234", "regular")
|
|
"1234sp" -> ("1234SP", "1234", "SP")
|
|
"1234 . PW"-> ("1234PW", "1234", "PW")
|
|
"12345", "abc", "" -> (None, None, None)
|
|
|
|
Manual entry used to do int(employee_id), so an extra-work record such as
|
|
"1234SP" could not be added by hand at all, and edit accepted any text,
|
|
which created pseudo-employees in the exports (§13).
|
|
"""
|
|
from working_hours_calculator import parse_employee_id_for_work_type
|
|
|
|
raw = str(raw_employee_id or '').strip()
|
|
if not raw:
|
|
return None, None, None
|
|
|
|
base_id, work_type = parse_employee_id_for_work_type(raw)
|
|
if not re.fullmatch(r'[0-9]+', base_id) or len(base_id) > MANUAL_EMPLOYEE_ID_MAX_DIGITS:
|
|
return None, None, None
|
|
|
|
canonical = base_id if work_type == 'regular' else f"{base_id}{work_type}"
|
|
return canonical, base_id, work_type
|
|
|
|
|
|
@bp.route('/attendance/<int:record_id>/edit', methods=['GET', 'POST'], endpoint='edit_attendance')
|
|
@login_required
|
|
@log_database_operations('attendance_update')
|
|
def edit_attendance(record_id):
|
|
"""Edit attendance record (Admin and Payroll only)"""
|
|
# Check if user has permission to edit attendance records
|
|
if session.get('role') not in ['admin', 'payroll', 'accounting']:
|
|
flash('Access denied. Only administrators and accounting staff can edit 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)
|
|
|
|
if request.method == 'POST':
|
|
# Get the audit note from form - REQUIRED
|
|
edit_note = request.form.get('edit_note', '').strip()
|
|
if not edit_note:
|
|
flash('Edit reason is required for audit purposes.', 'error')
|
|
projects = Project.query.filter_by(active_status=True).order_by(Project.name).all()
|
|
return render_template('edit_attendance.html',
|
|
attendance_record=attendance_record,
|
|
projects=projects,
|
|
qr_codes=QRCode.query.filter_by(active_status=True).all())
|
|
|
|
# Track changes for logging
|
|
changes = {}
|
|
old_values = {
|
|
'employee_id': attendance_record.employee_id,
|
|
'check_in_date': attendance_record.check_in_date,
|
|
'check_in_time': attendance_record.check_in_time,
|
|
'location_name': attendance_record.location_name,
|
|
'qr_code_id': attendance_record.qr_code_id,
|
|
'location_event': attendance_record.qr_code.location_event if attendance_record.qr_code else None
|
|
}
|
|
|
|
# Update attendance record fields
|
|
new_employee_id, _base_id, _work_type = _normalize_manual_employee_id(
|
|
request.form.get('employee_id', ''))
|
|
if not new_employee_id:
|
|
flash('Employee ID must be 1 to 4 digits, optionally with a work type '
|
|
'(for example 1234 or 1234SP).', 'error')
|
|
projects = Project.query.filter_by(active_status=True).order_by(Project.name).all()
|
|
return render_template('edit_attendance.html',
|
|
attendance_record=attendance_record,
|
|
projects=projects,
|
|
qr_codes=QRCode.query.filter_by(active_status=True).all())
|
|
new_check_in_date = datetime.strptime(request.form['check_in_date'], '%Y-%m-%d').date()
|
|
new_check_in_time = datetime.strptime(request.form['check_in_time'], '%H:%M').time()
|
|
new_location_name = request.form['location_name'].strip()
|
|
|
|
# Get the new QR code ID from the form (this determines the location event)
|
|
new_qr_code_id = request.form.get('qr_code_id', '').strip()
|
|
if not new_qr_code_id:
|
|
flash('Location event selection is required.', 'error')
|
|
projects = Project.query.filter_by(active_status=True).order_by(Project.name).all()
|
|
return render_template('edit_attendance.html',
|
|
attendance_record=attendance_record,
|
|
projects=projects,
|
|
qr_codes=QRCode.query.filter_by(active_status=True).all())
|
|
|
|
# Validate the QR code exists
|
|
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()
|
|
return render_template('edit_attendance.html',
|
|
attendance_record=attendance_record,
|
|
projects=projects,
|
|
qr_codes=QRCode.query.filter_by(active_status=True).all())
|
|
|
|
# Track what changed
|
|
if attendance_record.employee_id != new_employee_id:
|
|
changes['employee_id'] = f"{attendance_record.employee_id} → {new_employee_id}"
|
|
if attendance_record.check_in_date != new_check_in_date:
|
|
changes['check_in_date'] = f"{attendance_record.check_in_date} → {new_check_in_date}"
|
|
if attendance_record.check_in_time != new_check_in_time:
|
|
changes['check_in_time'] = f"{attendance_record.check_in_time} → {new_check_in_time}"
|
|
if attendance_record.location_name != new_location_name:
|
|
changes['location_name'] = f"{attendance_record.location_name} → {new_location_name}"
|
|
if attendance_record.qr_code_id != int(new_qr_code_id):
|
|
old_event = attendance_record.qr_code.location_event if attendance_record.qr_code else 'Unknown'
|
|
new_event = new_qr_code.location_event
|
|
changes['location_event'] = f"{old_event} → {new_event}"
|
|
changes['qr_code_id'] = f"{attendance_record.qr_code_id} → {new_qr_code_id}"
|
|
|
|
# Apply changes
|
|
attendance_record.employee_id = new_employee_id
|
|
attendance_record.check_in_date = new_check_in_date
|
|
attendance_record.check_in_time = new_check_in_time
|
|
attendance_record.location_name = new_location_name
|
|
attendance_record.qr_code_id = int(new_qr_code_id)
|
|
# Local time throughout (§18): check-in dates and times are local, so
|
|
# a UTC audit stamp read hours apart from the record it describes.
|
|
attendance_record.updated_timestamp = datetime.now()
|
|
|
|
# Store the audit note with timestamp and user info
|
|
timestamp = datetime.now().strftime('%Y-%m-%d %H:%M:%S')
|
|
username = session.get('username', 'Unknown')
|
|
role = session.get('role', 'unknown')
|
|
|
|
new_note_entry = f"[{timestamp}] {role.title()} '{username}': {edit_note}"
|
|
|
|
if attendance_record.edit_note:
|
|
# Append to existing notes
|
|
attendance_record.edit_note = f"{attendance_record.edit_note}\n\n{new_note_entry}"
|
|
else:
|
|
# First edit note
|
|
attendance_record.edit_note = new_note_entry
|
|
|
|
db.session.commit()
|
|
|
|
# Enhanced logging with audit note
|
|
if changes:
|
|
logger_handler.log_security_event(
|
|
event_type="attendance_record_update",
|
|
description=f"{session.get('role', 'unknown').title()} {session.get('username')} updated attendance record {record_id}",
|
|
severity="MEDIUM",
|
|
additional_data={
|
|
'record_id': record_id,
|
|
'changes': changes,
|
|
'user_role': session.get('role'),
|
|
'edit_reason': edit_note,
|
|
'editor_username': session.get('username')
|
|
}
|
|
)
|
|
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(
|
|
event_type="attendance_record_edit_no_changes",
|
|
description=f"{session.get('role', 'unknown').title()} {session.get('username')} accessed edit form for record {record_id} but made no changes",
|
|
severity="LOW",
|
|
additional_data={
|
|
'record_id': record_id,
|
|
'user_role': session.get('role'),
|
|
'edit_reason': edit_note,
|
|
'editor_username': session.get('username')
|
|
}
|
|
)
|
|
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'))
|
|
|
|
# GET request - show edit form
|
|
# Get available projects for the dropdown
|
|
projects = Project.query.filter_by(active_status=True).order_by(Project.name).all()
|
|
|
|
# Get available QR codes for location dropdown (for backward compatibility)
|
|
qr_codes = QRCode.query.filter_by(active_status=True).all()
|
|
|
|
return render_template('edit_attendance.html',
|
|
attendance_record=attendance_record,
|
|
projects=projects,
|
|
qr_codes=qr_codes)
|
|
|
|
except Exception as e:
|
|
db.session.rollback()
|
|
logger_handler.log_database_error('attendance_update', 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'))
|
|
|
|
@bp.route('/attendance/add', methods=['GET'], endpoint='add_manual_attendance')
|
|
@login_required
|
|
@log_user_activity('manual_attendance_access')
|
|
def add_manual_attendance():
|
|
"""
|
|
Display form to manually add attendance record
|
|
Only accessible by admin and accounting roles
|
|
"""
|
|
try:
|
|
user_role = session.get('role')
|
|
|
|
# Check authorization
|
|
if user_role not in ['admin', 'accounting']:
|
|
flash('You do not have permission to manually add attendance records.', 'error')
|
|
return redirect(url_for('attendance.attendance_report'))
|
|
|
|
# Get all active projects
|
|
projects = Project.query.filter_by(active_status=True).order_by(Project.name).all()
|
|
|
|
# Get today's date for form
|
|
today_date = datetime.now().strftime('%Y-%m-%d')
|
|
|
|
logger_handler.logger.info(
|
|
f"User {session.get('username')} ({user_role}) accessed manual attendance entry form"
|
|
)
|
|
|
|
return render_template('add_manual_attendance.html',
|
|
projects=projects,
|
|
today_date=today_date)
|
|
|
|
except Exception as e:
|
|
logger_handler.logger.error(f"Error loading manual attendance form: {e}")
|
|
flash('Error loading form. Please try again.', 'error')
|
|
return redirect(url_for('attendance.attendance_report'))
|
|
|
|
|
|
@bp.route('/attendance/save_manual', methods=['POST'], endpoint='save_manual_attendance')
|
|
@login_required
|
|
@log_user_activity('manual_attendance_creation')
|
|
@log_database_operations('manual_attendance_insert')
|
|
def save_manual_attendance():
|
|
"""
|
|
Save manually created attendance record
|
|
Only accessible by admin and accounting roles
|
|
"""
|
|
try:
|
|
user_role = session.get('role')
|
|
|
|
# Check authorization
|
|
if user_role not in ['admin', 'accounting']:
|
|
return jsonify({
|
|
'success': False,
|
|
'message': 'You do not have permission to manually add attendance records.'
|
|
}), 403
|
|
|
|
# Get form data
|
|
employee_id = request.form.get('employee_id', '').strip()
|
|
location_id = request.form.get('location_id', '').strip()
|
|
check_date = request.form.get('check_date', '').strip()
|
|
check_time = request.form.get('check_time', '').strip()
|
|
|
|
# Validate required fields
|
|
if not all([employee_id, location_id, check_date, check_time]):
|
|
flash('All fields are required.', 'error')
|
|
return redirect(url_for('attendance.add_manual_attendance'))
|
|
|
|
# Validate the ID, and keep any work-type code ("1234SP") on the stored value
|
|
canonical_employee_id, base_employee_id, work_type = _normalize_manual_employee_id(employee_id)
|
|
if not canonical_employee_id:
|
|
flash('Employee ID must be 1 to 4 digits, optionally with a work type '
|
|
'(for example 1234 or 1234SP).', 'error')
|
|
return redirect(url_for('attendance.add_manual_attendance'))
|
|
|
|
# Validate employee exists (by the numeric base ID)
|
|
employee = Employee.query.filter_by(id=int(base_employee_id)).first()
|
|
if not employee:
|
|
flash(f'Employee with ID {base_employee_id} not found.', 'error')
|
|
return redirect(url_for('attendance.add_manual_attendance'))
|
|
|
|
# Get QR code (location)
|
|
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'))
|
|
|
|
# Parse date and time
|
|
try:
|
|
check_date_obj = datetime.strptime(check_date, '%Y-%m-%d').date()
|
|
check_time_obj = datetime.strptime(check_time, '%H:%M').time()
|
|
except ValueError as e:
|
|
flash('Invalid date or time format.', 'error')
|
|
logger_handler.logger.error(f"Date/time parsing error: {e}")
|
|
return redirect(url_for('attendance.add_manual_attendance'))
|
|
|
|
# Check if record already exists for this employee, location, date, and time
|
|
existing_record = AttendanceData.query.filter_by(
|
|
employee_id=canonical_employee_id,
|
|
qr_code_id=qr_code.id,
|
|
check_in_date=check_date_obj,
|
|
check_in_time=check_time_obj
|
|
).first()
|
|
|
|
if existing_record:
|
|
flash('An attendance record already exists for this employee at this location, date, and time.', 'warning')
|
|
return redirect(url_for('attendance.add_manual_attendance'))
|
|
|
|
# Create new attendance record
|
|
# Use QR code's location address for both QR address and check-in address
|
|
# Set fixed distance of 0.010 miles
|
|
new_attendance = AttendanceData(
|
|
qr_code_id=qr_code.id,
|
|
employee_id=canonical_employee_id,
|
|
check_in_date=check_date_obj,
|
|
check_in_time=check_time_obj,
|
|
location_name=qr_code.location,
|
|
# Use QR code's coordinates
|
|
latitude=qr_code.address_latitude,
|
|
longitude=qr_code.address_longitude,
|
|
# Use QR code's address for both
|
|
address=qr_code.location_address,
|
|
# Set fixed distance
|
|
location_accuracy=0.010,
|
|
accuracy=0.010,
|
|
# Mark as manual entry
|
|
location_source='manual_entry',
|
|
device_info='Manual Entry by Admin/Accounting',
|
|
user_agent=f'Manual Entry - User: {session.get("username")}',
|
|
ip_address=get_client_ip(),
|
|
status='present',
|
|
verification_required=False,
|
|
verification_status='approved',
|
|
created_timestamp=datetime.now(),
|
|
updated_timestamp=datetime.now()
|
|
)
|
|
|
|
db.session.add(new_attendance)
|
|
db.session.commit()
|
|
|
|
# Log the manual entry
|
|
logger_handler.logger.info(
|
|
f"Manual attendance record created by {session.get('username')} ({user_role}): "
|
|
f"Employee {employee.firstName} {employee.lastName} (ID: {canonical_employee_id}), "
|
|
f"Location: {qr_code.location}, Event: {qr_code.location_event}, "
|
|
f"Date: {check_date}, Time: {check_time}"
|
|
)
|
|
|
|
flash(f'Attendance record successfully created for {employee.firstName} {employee.lastName}.', 'success')
|
|
return redirect(url_for('attendance.attendance_report'))
|
|
|
|
except Exception as e:
|
|
db.session.rollback()
|
|
logger_handler.logger.error(f"Error saving manual attendance record: {e}")
|
|
logger_handler.logger.error(f"Error saving manual attendance record: {e}", exc_info=True)
|
|
flash('Error saving attendance record. Please try again.', 'error')
|
|
return redirect(url_for('attendance.add_manual_attendance'))
|
|
|
|
|