From 9f42276fc4ad2c1713e9a6852b62ec931330ab91 Mon Sep 17 00:00:00 2001 From: Nguyen Ngo Date: Fri, 22 Aug 2025 13:11:26 -0400 Subject: [PATCH] Add payroll functionalities --- app.py | 391 +++++++++++++++++++++- payroll_excel_exporter.py | 463 ++++++++++++++++++++++++++ single_checkin_calculator.py | 320 ++++++++++++++++++ templates/base_authenticated.html | 16 +- templates/payroll_dashboard.html | 530 ++++++++++++++++++++++++++++++ working_hours_calculator.py | 445 +++++++++++++++++++++++++ 6 files changed, 2151 insertions(+), 14 deletions(-) create mode 100644 payroll_excel_exporter.py create mode 100644 single_checkin_calculator.py create mode 100644 templates/payroll_dashboard.html create mode 100644 working_hours_calculator.py diff --git a/app.py b/app.py index a84806d..e6cb83a 100644 --- a/app.py +++ b/app.py @@ -5,13 +5,15 @@ from functools import wraps from datetime import datetime, date, time, timedelta from sqlalchemy import text from user_agents import parse -import io, os, base64, re, uuid, requests, json, qrcode, math +import io, os, base64, re, uuid, requests, json, qrcode, math, traceback from PIL import Image, ImageDraw from math import radians, sin, cos, asin, sqrt from dotenv import load_dotenv # Import the logging handler from logger_handler import AppLogger, log_user_activity, log_database_operations +from single_checkin_calculator import SingleCheckInCalculator +from payroll_excel_exporter import PayrollExcelExporter # Load environment variables in .env load_dotenv() @@ -465,7 +467,6 @@ def calculate_location_accuracy_enhanced(qr_address, checkin_address, checkin_la except Exception as e: print(f"❌ Error calculating distance: {e}") - import traceback print(f"❌ Distance calculation traceback: {traceback.format_exc()}") return None @@ -2815,7 +2816,6 @@ def delete_qr_code(qr_id): logger_handler.log_database_error('qr_code_deletion', e) print(f"❌ ERROR in delete route: {e}") print(f"❌ Exception type: {type(e)}") - import traceback print(f"❌ Traceback: {traceback.format_exc()}") flash('Error deleting QR code. Please try again.', 'error') return redirect(url_for('dashboard')) @@ -3006,7 +3006,6 @@ def qr_checkin(qr_url): except Exception as e: print(f"❌ Error in location accuracy calculation: {e}") - import traceback print(f"❌ Full traceback: {traceback.format_exc()}") # ENHANCED DEBUG: Save to database with verification @@ -3050,7 +3049,6 @@ def qr_checkin(qr_url): except Exception as e: print(f"❌ Database error: {e}") - import traceback print(f"❌ Full traceback: {traceback.format_exc()}") db.session.rollback() logger_handler.log_database_error('checkin_save', e) @@ -3094,7 +3092,6 @@ def qr_checkin(qr_url): except Exception as e: print(f"❌ Unexpected error in check-in process: {e}") - import traceback print(f"❌ Traceback: {traceback.format_exc()}") return jsonify({ @@ -3505,7 +3502,6 @@ def attendance_report(): except Exception as e: print(f"❌ Error loading attendance report: {e}") print(f"❌ Exception type: {type(e)}") - import traceback print(f"❌ Traceback: {traceback.format_exc()}") # Log the error @@ -3784,7 +3780,6 @@ def export_configuration(): except Exception as e: print(f"❌ Error in export_configuration route: {e}") - import traceback print(f"❌ Traceback: {traceback.format_exc()}") # Use your existing logger error method with correct parameters @@ -3902,7 +3897,6 @@ def generate_excel_export(): except Exception as e: print(f"❌ Error in generate_excel_export route: {e}") - import traceback print(f"❌ Traceback: {traceback.format_exc()}") # Use your existing logger error method with correct parameters @@ -4077,7 +4071,6 @@ def create_excel_export(selected_columns, column_names, filters): except Exception as e: print(f"❌ Error creating Excel export: {e}") - import traceback print(f"❌ Traceback: {traceback.format_exc()}") return None @@ -4276,10 +4269,384 @@ def create_excel_export_ordered(selected_columns, column_names, filters): except Exception as e: print(f"❌ Error creating Excel export: {e}") - import traceback print(f"❌ Traceback: {traceback.format_exc()}") return None - + +@app.route('/payroll') +@login_required +def payroll_dashboard(): + """Payroll dashboard for calculating and exporting working hours""" + try: + # Check if user has payroll access + user_role = session.get('role') + if user_role not in ['admin', 'payroll']: + logger_handler.logger.warning(f"User {session.get('username', 'unknown')} (role: {user_role}) attempted to access payroll dashboard without permissions") + flash('Access denied. Only administrators and payroll staff can access payroll features.', 'error') + return redirect(url_for('dashboard')) + + print("📊 Loading payroll dashboard") + + # Log payroll dashboard access + logger_handler.logger.info(f"User {session.get('username', 'unknown')} accessed payroll dashboard") + + # Get filter parameters with defaults + date_from = request.args.get('date_from', '') + date_to = request.args.get('date_to', '') + project_filter = request.args.get('project_filter', '') + include_travel_time = request.args.get('include_travel_time', 'true').lower() == 'true' + + # Set default date range if not provided (last 2 weeks) + if not date_from or not date_to: + end_date = datetime.now().date() + start_date = end_date - timedelta(days=13) # 2 weeks (14 days) + date_from = start_date.strftime('%Y-%m-%d') + date_to = end_date.strftime('%Y-%m-%d') + + # Get list of projects for dropdown + projects = [] + try: + projects = Project.query.filter_by(active_status=True).order_by(Project.name).all() + print(f"📊 Found {len(projects)} active projects for filter") + except Exception as e: + print(f"⚠️ Error loading projects: {e}") + + # Get attendance records for the period + attendance_records = [] + working_hours_data = None + + if date_from and date_to: + try: + start_date = datetime.strptime(date_from, '%Y-%m-%d') + end_date = datetime.strptime(date_to, '%Y-%m-%d') + + # Query attendance records with optional project filter + query = db.session.query(AttendanceData).join(QRCode, AttendanceData.qr_code_id == QRCode.id) + + # Apply date filter + query = query.filter( + AttendanceData.check_in_date >= start_date.date(), + AttendanceData.check_in_date <= end_date.date() + ) + + # 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}") + + 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") + + # Calculate working hours if we have records + if attendance_records: + calculator = SingleCheckInCalculator() + 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") + + except ValueError as e: + print(f"⚠️ Invalid date format: {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.log_database_error('payroll_calculation', e) + flash('Error calculating working hours. Please check the server logs.', 'error') + + # Get employee names for display + employee_names = {} + if working_hours_data: + try: + # Try to get employee names from your employee table if it exists + employee_ids = list(working_hours_data['employees'].keys()) + # This assumes you have an employee table - modify as needed + employee_query = db.session.execute(text(""" + SELECT id, CONCAT(firstName, ' ', lastName) as full_name + FROM employee + WHERE id IN :employee_ids + """), {'employee_ids': tuple(employee_ids)}) + + for row in employee_query: + employee_names[str(row[0])] = row[1] + + except Exception as e: + print(f"⚠️ Could not load employee names: {e}") + # 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)) + if selected_project: + selected_project_name = selected_project.name + except Exception as e: + print(f"⚠️ Error getting selected project name: {e}") + + return render_template('payroll_dashboard.html', + working_hours_data=working_hours_data, + employee_names=employee_names, + projects=projects, + date_from=date_from, + date_to=date_to, + project_filter=project_filter, + selected_project_name=selected_project_name, + include_travel_time=include_travel_time, + 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.log_flask_error( + error_type="payroll_dashboard_error", + error_message=str(e), + stack_trace=traceback.format_exc() + ) + + flash('Error loading payroll dashboard. Please check the server logs.', 'error') + return redirect(url_for('dashboard')) + + +@app.route('/payroll/export-excel', methods=['POST']) +@login_required +@log_database_operations('payroll_excel_export') +def export_payroll_excel(): + """Export payroll report to Excel with working hours calculations""" + try: + # Check permissions + user_role = session.get('role') + if user_role not in ['admin', 'payroll']: + logger_handler.logger.warning(f"User {session.get('username', 'unknown')} (role: {user_role}) attempted unauthorized payroll Excel export") + flash('Access denied. Only administrators and payroll staff can export payroll data.', 'error') + return redirect(url_for('payroll_dashboard')) + + print("📊 Payroll Excel export started") + + # Get parameters from form + date_from = request.form.get('date_from') + date_to = request.form.get('date_to') + project_filter = request.form.get('project_filter', '') + include_travel_time = request.form.get('include_travel_time', 'false').lower() == 'true' + report_type = request.form.get('report_type', 'payroll') # 'payroll' or 'detailed' + + if not date_from or not date_to: + flash('Please provide both start and end dates for the export.', 'error') + return redirect(url_for('payroll_dashboard')) + + try: + start_date = datetime.strptime(date_from, '%Y-%m-%d') + end_date = datetime.strptime(date_to, '%Y-%m-%d') + except ValueError: + flash('Invalid date format. Please use YYYY-MM-DD format.', 'error') + return redirect(url_for('payroll_dashboard')) + + # Get attendance records with project filter + query = db.session.query(AttendanceData).join(QRCode, AttendanceData.qr_code_id == QRCode.id) + + # Apply date filter + query = query.filter( + AttendanceData.check_in_date >= start_date.date(), + AttendanceData.check_in_date <= end_date.date() + ) + + # 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}") + + query = query.order_by(AttendanceData.employee_id, AttendanceData.check_in_date, AttendanceData.check_in_time) + + attendance_records = query.all() + + if not attendance_records: + flash('No attendance records found for the selected date range and project.', 'warning') + return redirect(url_for('payroll_dashboard')) + + print(f"📊 Exporting {len(attendance_records)} attendance records to Excel") + + # Get employee names + employee_names = {} + try: + employee_ids = list(set(str(record.employee_id) for record in attendance_records)) + employee_query = db.session.execute(text(""" + SELECT id, CONCAT(firstName, ' ', lastName) as full_name + FROM employee + WHERE id IN :employee_ids + """), {'employee_ids': tuple(employee_ids)}) + + for row in employee_query: + employee_names[str(row[0])] = row[1] + + except Exception as e: + print(f"⚠️ Could not load employee names for export: {e}") + + # Get project name for filename + project_name = '' + if project_filter: + try: + project = Project.query.get(int(project_filter)) + if project: + project_name = f"_{project.name.replace(' ', '_')}" + except Exception as e: + print(f"⚠️ Error getting project name for filename: {e}") + + # Create Excel exporter + exporter = PayrollExcelExporter( + company_name=os.environ.get('COMPANY_NAME', 'Your Company'), + contract_name=os.environ.get('CONTRACT_NAME', 'Default Contract') + ) + + # Generate Excel file + if report_type == 'detailed': + excel_file = exporter.create_detailed_hours_report( + start_date, end_date, attendance_records, employee_names, include_travel_time + ) + filename_prefix = 'detailed_hours_report' + else: + excel_file = exporter.create_payroll_report( + start_date, end_date, attendance_records, employee_names, include_travel_time + ) + filename_prefix = 'payroll_report' + + if excel_file: + # Generate filename with timestamp and project name + timestamp = datetime.now().strftime('%Y%m%d_%H%M%S') + travel_suffix = '_with_travel' if include_travel_time else '_no_travel' + filename = f'{filename_prefix}_{date_from}_to_{date_to}{project_name}{travel_suffix}_{timestamp}.xlsx' + + print(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}") + + return send_file( + excel_file, + as_attachment=True, + download_name=filename, + mimetype='application/vnd.openxmlformats-officedocument.spreadsheetml.sheet' + ) + else: + flash('Error generating payroll Excel file.', 'error') + return redirect(url_for('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.log_flask_error( + error_type="payroll_excel_export_error", + error_message=str(e), + stack_trace=traceback.format_exc() + ) + + flash('Error generating payroll Excel export. Please check the server logs.', 'error') + return redirect(url_for('payroll_dashboard')) + +@app.route('/api/working-hours/calculate', methods=['POST']) +@login_required +@log_database_operations('working_hours_api_calculation') +def calculate_working_hours_api(): + """API endpoint for calculating working hours""" + try: + # Check permissions + user_role = session.get('role') + if user_role not in ['admin', 'payroll']: + return jsonify({ + 'success': False, + 'message': 'Access denied. Insufficient permissions.' + }), 403 + + # Get parameters from JSON request + data = request.get_json() + if not data: + return jsonify({ + 'success': False, + 'message': 'No data provided' + }), 400 + + employee_id = data.get('employee_id') + date_from = data.get('date_from') + date_to = data.get('date_to') + include_travel_time = data.get('include_travel_time', True) + + if not all([employee_id, date_from, date_to]): + return jsonify({ + 'success': False, + 'message': 'Missing required parameters: employee_id, date_from, date_to' + }), 400 + + try: + start_date = datetime.strptime(date_from, '%Y-%m-%d') + end_date = datetime.strptime(date_to, '%Y-%m-%d') + except ValueError: + return jsonify({ + 'success': False, + 'message': 'Invalid date format. Use YYYY-MM-DD.' + }), 400 + + # Get attendance records for the employee + query = db.session.query(AttendanceData).filter( + AttendanceData.employee_id == str(employee_id), + AttendanceData.check_in_date >= start_date.date(), + AttendanceData.check_in_date <= end_date.date() + ).order_by(AttendanceData.check_in_date, AttendanceData.check_in_time) + + attendance_records = query.all() + + # Calculate working hours + calculator = SingleCheckInCalculator() + hours_data = calculator.calculate_employee_hours( + str(employee_id), start_date, end_date, attendance_records + ) + + # Log API usage + logger_handler.logger.info(f"Working hours API used by {session.get('username', 'unknown')} for employee {employee_id}") + + return jsonify({ + 'success': True, + 'data': hours_data + }) + + except Exception as e: + print(f"❌ Error in calculate_working_hours_api: {e}") + logger_handler.log_flask_error( + error_type="working_hours_api_error", + error_message=str(e), + stack_trace=traceback.format_exc() + ) + + return jsonify({ + 'success': False, + 'message': 'Internal server error. Please check the server logs.' + }), 500 + +def get_employee_name(employee_id): + """Helper function to get employee full name by ID""" + try: + result = db.session.execute(text(""" + SELECT CONCAT(firstName, ' ', lastName) as full_name + FROM employee + WHERE id = :employee_id + """), {'employee_id': employee_id}) + + row = result.fetchone() + return row[0] if row else f"Employee {employee_id}" + + except Exception as e: + print(f"⚠️ Error getting employee name for ID {employee_id}: {e}") + return f"Employee {employee_id}" + +@app.context_processor +def inject_payroll_utils(): + """Inject payroll utility functions into templates""" + return { + 'get_employee_name': get_employee_name, + 'format_hours': lambda hours: f"{hours:.2f}" if hours else "0.00" + } # Jinja2 filters for better template functionality @app.template_filter('days_since') diff --git a/payroll_excel_exporter.py b/payroll_excel_exporter.py new file mode 100644 index 0000000..926d664 --- /dev/null +++ b/payroll_excel_exporter.py @@ -0,0 +1,463 @@ +#!/usr/bin/env python3 +""" +Payroll Excel Exporter +===================== + +Enhanced Excel exporter for payroll reports with working hours calculations. +Based on the Java PayrollReport.java implementation. + +Features: +- Employee working hours with daily/weekly breakdown +- Regular and overtime hour calculations +- Travel time inclusion options +- Professional Excel formatting +- Multiple report formats +""" + +import io +from datetime import datetime, timedelta +from typing import List, Dict, Any, Optional +from openpyxl import Workbook +from openpyxl.styles import Font, Alignment, PatternFill, Border, Side +from openpyxl.utils import get_column_letter +from single_checkin_calculator import SingleCheckInCalculator +from logger_handler import log_database_operations + + +class PayrollExcelExporter: + """Excel exporter for payroll reports with working hours""" + + def __init__(self, company_name: str = "Company Name", contract_name: str = "Default Contract"): + self.company_name = company_name + self.contract_name = contract_name + + # Excel styling + self.header1_style = None + self.header2_style = None + self.header3_style = None + self.table_header_style = None + self.data_style = None + self.total_style = None + + def _setup_styles(self, workbook: Workbook): + """Setup Excel cell styles""" + + # Header 1 style (Company name) + self.header1_style = { + 'font': Font(name="Arial", size=16, bold=True, color="FFFFFF"), + 'fill': PatternFill(start_color="2F4F4F", end_color="2F4F4F", fill_type="solid"), + 'alignment': Alignment(horizontal="center", vertical="center") + } + + # Header 2 style (Report title) + self.header2_style = { + 'font': Font(name="Arial", size=14, bold=True, color="FFFFFF"), + 'fill': PatternFill(start_color="4682B4", end_color="4682B4", fill_type="solid"), + 'alignment': Alignment(horizontal="center", vertical="center") + } + + # Header 3 style (Contract and date range) + self.header3_style = { + 'font': Font(name="Arial", size=12, bold=True), + 'alignment': Alignment(horizontal="center", vertical="center") + } + + # Table header style + self.table_header_style = { + 'font': Font(name="Arial", size=11, bold=True, color="FFFFFF"), + 'fill': PatternFill(start_color="366092", end_color="366092", fill_type="solid"), + 'alignment': Alignment(horizontal="center", vertical="center"), + 'border': Border( + left=Side(style="thin"), + right=Side(style="thin"), + top=Side(style="thin"), + bottom=Side(style="thin") + ) + } + + # Data style + self.data_style = { + 'font': Font(name="Arial", size=10), + 'alignment': Alignment(horizontal="center", vertical="center"), + 'border': Border( + left=Side(style="thin"), + right=Side(style="thin"), + top=Side(style="thin"), + bottom=Side(style="thin") + ) + } + + # Total style + self.total_style = { + 'font': Font(name="Arial", size=11, bold=True), + 'fill': PatternFill(start_color="E6E6FA", end_color="E6E6FA", fill_type="solid"), + 'alignment': Alignment(horizontal="center", vertical="center"), + 'border': Border( + left=Side(style="thick"), + right=Side(style="thick"), + top=Side(style="thick"), + bottom=Side(style="thick") + ) + } + + def _apply_style(self, cell, style_dict): + """Apply a style dictionary to a cell""" + for attr, value in style_dict.items(): + setattr(cell, attr, value) + + @log_database_operations('payroll_excel_export') + def create_payroll_report(self, start_date: datetime, end_date: datetime, + attendance_records: List[Dict], employee_names: Dict[str, str] = None, + include_travel_time: bool = True) -> io.BytesIO: + """ + Create a comprehensive payroll report with working hours + + Args: + start_date: Report start date + end_date: Report end date + attendance_records: List of attendance records + employee_names: Dictionary mapping employee_id to full name + include_travel_time: Whether to include travel time in calculations + + Returns: + BytesIO buffer containing the Excel file + """ + try: + print(f"📊 Creating payroll report from {start_date.strftime('%Y-%m-%d')} to {end_date.strftime('%Y-%m-%d')}") + + # Calculate working hours + calculator = SingleCheckInCalculator() + hours_data = calculator.calculate_all_employees_hours(start_date, end_date, attendance_records) + + # Create workbook + workbook = Workbook() + worksheet = workbook.active + worksheet.title = "Payroll Report" + + # Setup styles + self._setup_styles(workbook) + + # Write report headers + current_row = self._write_report_headers(worksheet, start_date, end_date) + + # Write employee data + current_row = self._write_employee_payroll_data(worksheet, hours_data, employee_names, current_row) + + # Write summary totals + self._write_summary_totals(worksheet, hours_data, current_row) + + # Auto-adjust column widths + self._auto_adjust_columns(worksheet) + + # Save to BytesIO + excel_buffer = io.BytesIO() + workbook.save(excel_buffer) + excel_buffer.seek(0) + + print("✅ Payroll report Excel file created successfully") + return excel_buffer + + except Exception as e: + print(f"❌ Error creating payroll report: {e}") + raise e + + def _write_report_headers(self, worksheet, start_date: datetime, end_date: datetime) -> int: + """Write report headers and return next row number""" + current_row = 1 + + # Company name + cell = worksheet.cell(row=current_row, column=1, value=self.company_name) + self._apply_style(cell, self.header1_style) + worksheet.merge_cells(f'A{current_row}:M{current_row}') + current_row += 1 + + # Report title + cell = worksheet.cell(row=current_row, column=1, value="Payroll Report - Working Hours Summary") + self._apply_style(cell, self.header2_style) + worksheet.merge_cells(f'A{current_row}:M{current_row}') + current_row += 1 + + # Contract name + cell = worksheet.cell(row=current_row, column=1, value=self.contract_name) + self._apply_style(cell, self.header3_style) + worksheet.merge_cells(f'A{current_row}:M{current_row}') + current_row += 1 + + # Date range + date_range = f"Date range: {start_date.strftime('%Y-%m-%d')} to {end_date.strftime('%Y-%m-%d')}" + cell = worksheet.cell(row=current_row, column=1, value=date_range) + self._apply_style(cell, self.header3_style) + worksheet.merge_cells(f'A{current_row}:M{current_row}') + current_row += 2 # Add extra space + + return current_row + + def _write_employee_payroll_data(self, worksheet, hours_data: Dict, employee_names: Dict[str, str], start_row: int) -> int: + """Write employee payroll data and return next row number""" + current_row = start_row + + # Table headers + headers = [ + "#", "Employee ID", "Employee Name", + "Week 1 Mon", "Week 1 Tue", "Week 1 Wed", "Week 1 Thu", "Week 1 Fri", "Week 1 Sat", "Week 1 Sun", + "Week 1 Regular", "Week 1 OT", + "Week 2 Mon", "Week 2 Tue", "Week 2 Wed", "Week 2 Thu", "Week 2 Fri", "Week 2 Sat", "Week 2 Sun", + "Week 2 Regular", "Week 2 OT", + "Total Regular", "Total OT", "Grand Total" + ] + + for col, header in enumerate(headers, 1): + cell = worksheet.cell(row=current_row, column=col, value=header) + self._apply_style(cell, self.table_header_style) + current_row += 1 + + # Employee data + counter = 1 + for employee_id, emp_data in hours_data['employees'].items(): + # Employee basic info + employee_name = employee_names.get(employee_id, f"Employee {employee_id}") if employee_names else f"Employee {employee_id}" + + row_data = [ + counter, + employee_id, + employee_name + ] + + # Calculate daily hours for two weeks (14 days) + daily_hours = self._calculate_daily_hours_for_payroll(emp_data, hours_data['period_start']) + + # Week 1 daily hours (7 days) + week1_total = 0 + for day_idx in range(7): + hours = daily_hours.get(day_idx, 0) + row_data.append(round(hours, 2)) + week1_total += hours + + # Week 1 regular and OT + week1_regular = min(week1_total, 40.0) + week1_ot = max(0, week1_total - 40.0) + row_data.extend([round(week1_regular, 2), round(week1_ot, 2)]) + + # Week 2 daily hours (7 days) + week2_total = 0 + for day_idx in range(7, 14): + hours = daily_hours.get(day_idx, 0) + row_data.append(round(hours, 2)) + week2_total += hours + + # Week 2 regular and OT + week2_regular = min(week2_total, 40.0) + week2_ot = max(0, week2_total - 40.0) + row_data.extend([round(week2_regular, 2), round(week2_ot, 2)]) + + # Totals + total_regular = week1_regular + week2_regular + total_ot = week1_ot + week2_ot + grand_total = total_regular + total_ot + row_data.extend([round(total_regular, 2), round(total_ot, 2), round(grand_total, 2)]) + + # Write row data + for col, value in enumerate(row_data, 1): + cell = worksheet.cell(row=current_row, column=col, value=value) + self._apply_style(cell, self.data_style) + + current_row += 1 + counter += 1 + + return current_row + 1 # Add space before summary + + def _calculate_daily_hours_for_payroll(self, emp_data: Dict, start_date_str: str) -> Dict[int, float]: + """Calculate daily hours for payroll format (14 days)""" + daily_hours = {} + start_date = datetime.strptime(start_date_str, '%Y-%m-%d') + + for day_idx in range(14): + current_date = start_date + timedelta(days=day_idx) + date_key = current_date.strftime('%Y-%m-%d') + + if date_key in emp_data['daily_hours']: + day_data = emp_data['daily_hours'][date_key] + if not day_data['is_miss_punch']: + daily_hours[day_idx] = day_data['total_hours'] + else: + daily_hours[day_idx] = 0 # Miss punch = 0 hours + else: + daily_hours[day_idx] = 0 # No records = 0 hours + + return daily_hours + + def _write_summary_totals(self, worksheet, hours_data: Dict, start_row: int): + """Write summary totals section""" + current_row = start_row + 1 + + # Summary header + cell = worksheet.cell(row=current_row, column=1, value="PAYROLL SUMMARY") + self._apply_style(cell, self.header2_style) + worksheet.merge_cells(f'A{current_row}:F{current_row}') + current_row += 2 + + # Calculate totals across all employees + total_employees = len(hours_data['employees']) + total_regular_hours = 0 + total_overtime_hours = 0 + total_hours = 0 + + for emp_data in hours_data['employees'].values(): + total_regular_hours += emp_data['grand_totals']['regular_hours'] + total_overtime_hours += emp_data['grand_totals']['overtime_hours'] + total_hours += emp_data['grand_totals']['total_hours'] + + # Summary data + summary_data = [ + ("Total Employees:", total_employees), + ("Total Regular Hours:", round(total_regular_hours, 2)), + ("Total Overtime Hours:", round(total_overtime_hours, 2)), + ("Grand Total Hours:", round(total_hours, 2)), + ("Travel Time Included:", "Yes" if hours_data['include_travel_time'] else "No"), + ("Report Generated:", datetime.now().strftime('%Y-%m-%d %H:%M:%S')) + ] + + for label, value in summary_data: + cell1 = worksheet.cell(row=current_row, column=1, value=label) + self._apply_style(cell1, self.total_style) + cell2 = worksheet.cell(row=current_row, column=2, value=value) + self._apply_style(cell2, self.total_style) + current_row += 1 + + def _auto_adjust_columns(self, worksheet): + """Auto-adjust column widths""" + for column in worksheet.columns: + max_length = 0 + column_letter = get_column_letter(column[0].column) + + for cell in column: + try: + if len(str(cell.value)) > max_length: + max_length = len(str(cell.value)) + except: + pass + + adjusted_width = min(max_length + 2, 20) + worksheet.column_dimensions[column_letter].width = adjusted_width + + @log_database_operations('detailed_hours_export') + def create_detailed_hours_report(self, start_date: datetime, end_date: datetime, + attendance_records: List[Dict], employee_names: Dict[str, str] = None, + include_travel_time: bool = True) -> io.BytesIO: + """ + Create a detailed daily hours report for all employees + + Args: + start_date: Report start date + end_date: Report end date + attendance_records: List of attendance records + employee_names: Dictionary mapping employee_id to full name + include_travel_time: Whether to include travel time in calculations + + Returns: + BytesIO buffer containing the Excel file + """ + try: + print(f"📊 Creating detailed hours report from {start_date.strftime('%Y-%m-%d')} to {end_date.strftime('%Y-%m-%d')}") + + # Calculate working hours + calculator = SingleCheckInCalculator() + hours_data = calculator.calculate_all_employees_hours(start_date, end_date, attendance_records) + + # Create workbook + workbook = Workbook() + worksheet = workbook.active + worksheet.title = "Detailed Hours Report" + + # Setup styles + self._setup_styles(workbook) + + # Write headers + current_row = 1 + cell = worksheet.cell(row=current_row, column=1, value=self.company_name) + self._apply_style(cell, self.header1_style) + worksheet.merge_cells(f'A{current_row}:J{current_row}') + current_row += 1 + + cell = worksheet.cell(row=current_row, column=1, value="Detailed Daily Hours Report") + self._apply_style(cell, self.header2_style) + worksheet.merge_cells(f'A{current_row}:J{current_row}') + current_row += 2 + + # Table headers + headers = ["Employee ID", "Employee Name", "Date", "Day of Week", "Total Hours", + "Status", "Records Count", "Regular Hours", "Overtime Hours", "Notes"] + + for col, header in enumerate(headers, 1): + cell = worksheet.cell(row=current_row, column=col, value=header) + self._apply_style(cell, self.table_header_style) + current_row += 1 + + # Write detailed data + for employee_id, emp_data in hours_data['employees'].items(): + employee_name = employee_names.get(employee_id, f"Employee {employee_id}") if employee_names else f"Employee {employee_id}" + + # Sort daily hours by date + daily_items = sorted(emp_data['daily_hours'].items()) + + for date_str, day_data in daily_items: + date_obj = datetime.strptime(date_str, '%Y-%m-%d') + day_name = date_obj.strftime('%A') + + # Determine status and notes + if day_data['is_miss_punch']: + status = "Miss Punch" + notes = "Missing check-in or check-out" + hours = 0 + elif day_data['records_count'] == 0: + status = "No Records" + notes = "No attendance records" + hours = 0 + else: + status = "Complete" + notes = f"{day_data['records_count']} record(s)" + hours = day_data['total_hours'] + + # Calculate regular/overtime for this day + regular_hours = min(hours, 8.0) # Daily limit + overtime_hours = max(0, hours - 8.0) + + row_data = [ + employee_id, + employee_name, + date_str, + day_name, + round(hours, 2), + status, + day_data['records_count'], + round(regular_hours, 2), + round(overtime_hours, 2), + notes + ] + + for col, value in enumerate(row_data, 1): + cell = worksheet.cell(row=current_row, column=col, value=value) + self._apply_style(cell, self.data_style) + + # Color code miss punches + if status == "Miss Punch": + cell.fill = PatternFill(start_color="FFE6E6", end_color="FFE6E6", fill_type="solid") + elif status == "No Records": + cell.fill = PatternFill(start_color="F0F0F0", end_color="F0F0F0", fill_type="solid") + + current_row += 1 + + # Auto-adjust columns + self._auto_adjust_columns(worksheet) + + # Save to BytesIO + excel_buffer = io.BytesIO() + workbook.save(excel_buffer) + excel_buffer.seek(0) + + print("✅ Detailed hours report Excel file created successfully") + return excel_buffer + + except Exception as e: + print(f"❌ Error creating detailed hours report: {e}") + raise e \ No newline at end of file diff --git a/single_checkin_calculator.py b/single_checkin_calculator.py new file mode 100644 index 0000000..348903f --- /dev/null +++ b/single_checkin_calculator.py @@ -0,0 +1,320 @@ +#!/usr/bin/env python3 +""" +Single Check-in Working Hours Calculator +======================================= + +Calculator specifically designed for single check-in systems where each +attendance record represents a check-in only (not check-in/check-out pairs). + +This calculator interprets consecutive check-ins as work periods: +- 1st check-in = start work +- 2nd check-in = end work (or start of next period) +- 3rd check-in = end previous period, start new period +- etc. + +Based on your attendance system structure. +""" + +from datetime import datetime, timedelta, time +from typing import List, Dict, Optional, Tuple, Any +from dataclasses import dataclass +import math +from logger_handler import log_database_operations + + +@dataclass +class CheckInRecord: + """Represents a single check-in record""" + id: int + employee_id: str + check_in_date: datetime + check_in_time: time + location_name: str + timestamp: datetime = None + + def __post_init__(self): + if self.timestamp is None: + # Combine date and time for timestamp + self.timestamp = datetime.combine(self.check_in_date, self.check_in_time) + + +@dataclass +class WorkPeriod: + """Represents a calculated work period from consecutive check-ins""" + start_record: CheckInRecord + end_record: Optional[CheckInRecord] + duration_minutes: int = 0 + is_complete: bool = False + + def __post_init__(self): + if self.end_record and self.start_record: + duration = self.end_record.timestamp - self.start_record.timestamp + self.duration_minutes = int(duration.total_seconds() / 60) + self.is_complete = True + else: + self.duration_minutes = 0 + self.is_complete = False + + +class SingleCheckInCalculator: + """Calculator for single check-in attendance systems""" + + def __init__(self, max_work_period_hours: float = 12.0, min_break_minutes: int = 30): + self.max_work_period_hours = max_work_period_hours # Maximum reasonable work period + self.min_break_minutes = min_break_minutes # Minimum break between work periods + + @log_database_operations('single_checkin_hours_calculation') + def calculate_employee_hours(self, employee_id: str, start_date: datetime, end_date: datetime, + attendance_records: List[Dict]) -> Dict[str, Any]: + """ + Calculate working hours for an employee using single check-in records + + Logic: + - Convert consecutive check-ins into work periods + - 1st check-in = start work, 2nd check-in = end work + - Handle multiple periods per day + - Round to nearest quarter hour + """ + try: + print(f"🔍 Calculating hours for employee {employee_id}") + + # Convert database records to CheckInRecord objects + records = [] + for record in attendance_records: + if hasattr(record, '__dict__'): + emp_id = str(record.employee_id) + date_val = record.check_in_date + time_val = record.check_in_time + location = record.location_name + record_id = record.id + else: + emp_id = str(record['employee_id']) + date_val = record['check_in_date'] + time_val = record['check_in_time'] + location = record['location_name'] + record_id = record['id'] + + if emp_id == employee_id: + checkin_record = CheckInRecord( + id=record_id, + employee_id=emp_id, + check_in_date=date_val, + check_in_time=time_val, + location_name=location + ) + records.append(checkin_record) + + if not records: + print(f"⚠️ No records found for employee {employee_id}") + return self._empty_result(employee_id, start_date, end_date) + + print(f"📊 Found {len(records)} check-in records for employee {employee_id}") + + # Group records by date and calculate daily hours + daily_records = {} + for record in records: + date_key = record.check_in_date.strftime('%Y-%m-%d') + if date_key not in daily_records: + daily_records[date_key] = [] + daily_records[date_key].append(record) + + # Calculate daily hours + daily_hours = {} + weekly_totals = [] + + current_date = start_date + current_week_hours = [] + + while current_date <= end_date: + date_key = current_date.strftime('%Y-%m-%d') + day_records = daily_records.get(date_key, []) + + # Calculate hours for this day + day_hours, is_miss_punch = self._calculate_daily_hours_from_checkins(day_records) + + daily_hours[date_key] = { + 'total_minutes': int(day_hours * 60) if day_hours > 0 else 0, + 'total_hours': day_hours if day_hours > 0 else 0, + 'is_miss_punch': is_miss_punch, + 'records_count': len(day_records) + } + + print(f"📅 {date_key}: {len(day_records)} records, {day_hours:.2f} hours, miss_punch: {is_miss_punch}") + + # Add to weekly calculation (only positive hours) + current_week_hours.append(max(0, day_hours)) + + # Check if end of week (Sunday) or end of period + if current_date.weekday() == 6 or current_date == end_date: # Sunday or last day + week_total = sum(current_week_hours) + week_regular = min(week_total, 40.0) + week_overtime = max(0, week_total - 40.0) + + weekly_totals.append({ + 'total_hours': week_total, + 'regular_hours': week_regular, + 'overtime_hours': week_overtime, + 'total_minutes': int(week_total * 60), + 'regular_minutes': int(week_regular * 60), + 'overtime_minutes': int(week_overtime * 60) + }) + + current_week_hours = [] + + current_date += timedelta(days=1) + + # Calculate grand totals + grand_total_hours = sum(week['total_hours'] for week in weekly_totals) + grand_regular_hours = sum(week['regular_hours'] for week in weekly_totals) + grand_overtime_hours = sum(week['overtime_hours'] for week in weekly_totals) + + print(f"✅ Employee {employee_id}: {grand_total_hours:.2f} total hours, {grand_regular_hours:.2f} regular, {grand_overtime_hours:.2f} OT") + + return { + 'employee_id': employee_id, + 'start_date': start_date.strftime('%Y-%m-%d'), + 'end_date': end_date.strftime('%Y-%m-%d'), + 'include_travel_time': True, # Not applicable for single check-in system + 'daily_hours': daily_hours, + 'weekly_hours': weekly_totals, + 'grand_totals': { + 'total_hours': grand_total_hours, + 'regular_hours': grand_regular_hours, + 'overtime_hours': grand_overtime_hours, + 'total_minutes': int(grand_total_hours * 60), + 'regular_minutes': int(grand_regular_hours * 60), + 'overtime_minutes': int(grand_overtime_hours * 60) + } + } + + except Exception as e: + print(f"❌ Error calculating working hours for employee {employee_id}: {e}") + import traceback + print(f"❌ Traceback: {traceback.format_exc()}") + return self._empty_result(employee_id, start_date, end_date) + + def _calculate_daily_hours_from_checkins(self, day_records: List[CheckInRecord]) -> Tuple[float, bool]: + """ + Calculate hours for a single day from check-in records + + Logic: + - Pair consecutive check-ins as work periods + - 1st = start, 2nd = end, 3rd = start new period, 4th = end, etc. + - Single record = miss punch + - Validate reasonable work periods + + Returns: (hours, is_miss_punch) + """ + if not day_records: + return 0.0, False + + if len(day_records) == 1: + print(f"⚠️ Single check-in found - miss punch") + return 0.0, True # Single check-in = miss punch + + # Sort records by time + sorted_records = sorted(day_records, key=lambda r: r.timestamp) + print(f"📝 Processing {len(sorted_records)} records for the day") + + # Create work periods from consecutive check-ins + work_periods = [] + for i in range(0, len(sorted_records) - 1, 2): + start_record = sorted_records[i] + end_record = sorted_records[i + 1] if i + 1 < len(sorted_records) else None + + if end_record: + period = WorkPeriod(start_record, end_record) + + # Validate work period duration + if self._is_valid_work_period(period): + work_periods.append(period) + print(f"✅ Valid work period: {start_record.check_in_time} - {end_record.check_in_time} = {period.duration_minutes/60:.2f} hours") + else: + print(f"⚠️ Invalid work period: {period.duration_minutes/60:.2f} hours (too long or negative)") + return 0.0, True # Invalid period = miss punch + else: + print(f"⚠️ Unpaired check-in at {start_record.check_in_time}") + return 0.0, True # Unpaired record = miss punch + + # Calculate total hours + total_minutes = sum(period.duration_minutes for period in work_periods) + total_hours = total_minutes / 60.0 + + # Round to nearest quarter hour + rounded_hours = round(total_hours * 4) / 4 + + print(f"📊 Daily total: {rounded_hours:.2f} hours from {len(work_periods)} work periods") + + return rounded_hours, False + + def _is_valid_work_period(self, period: WorkPeriod) -> bool: + """Check if a work period is valid (reasonable duration)""" + if not period.is_complete: + return False + + hours = period.duration_minutes / 60.0 + + # Must be positive and less than max work period + if hours <= 0 or hours > self.max_work_period_hours: + return False + + # Must be at least 15 minutes + if period.duration_minutes < 15: + return False + + return True + + def _empty_result(self, employee_id: str, start_date: datetime, end_date: datetime) -> Dict[str, Any]: + """Return empty result structure""" + return { + 'employee_id': employee_id, + 'start_date': start_date.strftime('%Y-%m-%d'), + 'end_date': end_date.strftime('%Y-%m-%d'), + 'include_travel_time': True, + 'daily_hours': {}, + 'weekly_hours': [], + 'grand_totals': { + 'total_hours': 0.0, + 'regular_hours': 0.0, + 'overtime_hours': 0.0, + 'total_minutes': 0, + 'regular_minutes': 0, + 'overtime_minutes': 0 + } + } + + def calculate_all_employees_hours(self, start_date: datetime, end_date: datetime, + attendance_records: List[Dict]) -> Dict[str, Any]: + """Calculate working hours for all employees in the given period""" + try: + print(f"🚀 Starting calculation for all employees") + + # Get unique employee IDs + employee_ids = set() + for record in attendance_records: + if hasattr(record, '__dict__'): + employee_ids.add(str(record.employee_id)) + else: + employee_ids.add(str(record['employee_id'])) + + print(f"👥 Found {len(employee_ids)} unique employees") + + results = {} + for emp_id in sorted(employee_ids): + print(f"\n🔄 Processing employee {emp_id}") + results[emp_id] = self.calculate_employee_hours(emp_id, start_date, end_date, attendance_records) + + return { + 'calculation_date': datetime.now().strftime('%Y-%m-%d %H:%M:%S'), + 'period_start': start_date.strftime('%Y-%m-%d'), + 'period_end': end_date.strftime('%Y-%m-%d'), + 'include_travel_time': True, # Not applicable for single check-in + 'employee_count': len(employee_ids), + 'employees': results + } + + except Exception as e: + print(f"❌ Error calculating hours for all employees: {e}") + import traceback + print(f"❌ Traceback: {traceback.format_exc()}") + raise e \ No newline at end of file diff --git a/templates/base_authenticated.html b/templates/base_authenticated.html index a7bf3e9..c1527be 100644 --- a/templates/base_authenticated.html +++ b/templates/base_authenticated.html @@ -68,18 +68,30 @@ Reports + + + Payroll + System Logs - {% elif session.role in ['payroll', 'project_manager'] %} + {% elif session.role in ['payroll'] %} + + + Reports + + + + Payroll + + {% elif session.role in ['project_manager'] %} Reports {% endif %} - Profile diff --git a/templates/payroll_dashboard.html b/templates/payroll_dashboard.html new file mode 100644 index 0000000..6061658 --- /dev/null +++ b/templates/payroll_dashboard.html @@ -0,0 +1,530 @@ +{% extends "base_authenticated.html" %} +{% block title %}Payroll Dashboard - Working Hours{% endblock %} + +{% block extra_head %} + +{% endblock %} + +{% block content %} +
+
+ + + + Back to Dashboard + + + +
+

+ + Payroll Dashboard +

+

Calculate employee working hours and generate payroll reports

+
+ + +
+

+ + Calculation Parameters +

+ +
+
+ + +
+ +
+ + +
+ +
+ + +
+ +
+ +
+ +
+ + +
+
+
+ + + {% if working_hours_data %} +
+

+ + Working Hours Summary +

+ + +
+
+

{{ working_hours_data.employee_count }}

+

Total Employees

+
+
+

{{ "%.1f"|format(working_hours_data.employees.values() | map(attribute='grand_totals.total_hours') | sum) }}

+

Total Hours

+
+
+

{{ "%.1f"|format(working_hours_data.employees.values() | map(attribute='grand_totals.regular_hours') | sum) }}

+

Regular Hours

+
+
+

{{ "%.1f"|format(working_hours_data.employees.values() | map(attribute='grand_totals.overtime_hours') | sum) }}

+

Overtime Hours

+
+
+ + +
+ + + + + + + + + + + + + + {% for employee_id, emp_data in working_hours_data.employees.items() %} + + + + + + + + + + {% endfor %} + +
Employee IDEmployee NameTotal HoursRegular HoursOvertime HoursWorking DaysStatus
{{ employee_id }}{{ employee_names.get(employee_id, 'Employee ' + employee_id) }} + + {{ "%.2f"|format(emp_data.grand_totals.total_hours) }} hrs + + + + {{ "%.2f"|format(emp_data.grand_totals.regular_hours) }} hrs + + + {% if emp_data.grand_totals.overtime_hours > 0 %} + + {{ "%.2f"|format(emp_data.grand_totals.overtime_hours) }} hrs + + {% else %} + + 0.00 hrs + + {% endif %} + + {% set working_days = emp_data.daily_hours.values() | selectattr('total_hours', '>', 0) | list | length %} + {{ working_days }} days + + {% set miss_punches = emp_data.daily_hours.values() | selectattr('is_miss_punch', 'equalto', true) | list | length %} + {% if miss_punches > 0 %} + + {{ miss_punches }} miss punch(es) + + {% else %} + + Complete + + {% endif %} +
+
+ + +
+
+ + + + + +
+ +
+ + + + + +
+
+ +
+ Report Details:
+ Period: {{ working_hours_data.period_start }} to {{ working_hours_data.period_end }}
+ Travel Time: {{ "Included" if working_hours_data.include_travel_time else "Excluded" }}
+ Generated: {{ working_hours_data.calculation_date }} +
+
+ {% else %} +
+
+
+ {% if date_from and date_to %} + No attendance records found for the selected date range.
+ Please try a different date range or check if attendance data has been recorded. + {% else %} + Please select a date range to calculate working hours. + {% endif %} +
+
+ {% endif %} +
+
+ + +{% endblock %} \ No newline at end of file diff --git a/working_hours_calculator.py b/working_hours_calculator.py new file mode 100644 index 0000000..41b79b6 --- /dev/null +++ b/working_hours_calculator.py @@ -0,0 +1,445 @@ +#!/usr/bin/env python3 +""" +Working Hours Calculator for Employee Payroll +============================================ + +This module implements the working hours calculation logic +based on the Java files provided. It handles: +- Daily time calculations with travel time options +- Weekly regular and overtime hours +- Record pairing (check-in/check-out) +- Missing punch detection +- Quarter-hour rounding + +Based on the Java classes: +- DailyTimeCalculator.java +- WeeklyTimeCalculator.java +- PayrollReport.java +""" + +from datetime import datetime, timedelta, time +from typing import List, Dict, Optional, Tuple, Any +from dataclasses import dataclass +import math +from logger_handler import log_database_operations + +# Constants from Java implementation +TRAVEL_TIME_MAX_MINUTES = 60 +RECORD_GROUPING_MAX_MINUTES = 60 * 6 # 6 hours +MAX_REGULAR_TIME_MINUTES = 60 * 40 # 40 hours per week + + +@dataclass +class AttendanceRecord: + """Represents a single attendance record""" + id: int + employee_id: str + check_in_date: datetime + check_in_time: time + location_name: str + record_type: str = 'check_in' # 'check_in' or 'check_out' + timestamp: datetime = None + + def __post_init__(self): + if self.timestamp is None: + # Combine date and time for timestamp + self.timestamp = datetime.combine(self.check_in_date, self.check_in_time) + + +@dataclass +class RecordPair: + """Represents a paired check-in/check-out record""" + check_in: Optional[AttendanceRecord] + check_out: Optional[AttendanceRecord] + is_miss_punch: bool = False + date: datetime = None + location: str = "" + + def __post_init__(self): + if self.check_in: + self.date = self.check_in.check_in_date + self.location = self.check_in.location_name + elif self.check_out: + self.date = self.check_out.check_in_date + self.location = self.check_out.location_name + + @property + def duration_minutes(self) -> int: + """Calculate duration in minutes between check-in and check-out""" + if self.is_miss_punch or not self.check_in or not self.check_out: + return -1 + + duration = self.check_out.timestamp - self.check_in.timestamp + return int(duration.total_seconds() / 60) + + +class TimeCalculator: + """Base time calculator with rounding functionality""" + + @staticmethod + def round_time_to_nearest_quarter_hour(minutes: int) -> int: + """Round time to nearest quarter hour (15 minutes)""" + if minutes < 0: + return minutes # Keep negative values for miss punches + + # Round to nearest 15-minute interval + return round(minutes / 15) * 15 + + +class DailyTimeCalculator(TimeCalculator): + """Calculate daily working hours with travel time options""" + + def __init__(self): + self.record_pairs: List[RecordPair] = [] + + def add_record_pair(self, pair: RecordPair): + """Add a record pair to the daily calculation""" + self.record_pairs.append(pair) + + def get_minutes_total_include_travel_time(self) -> int: + """Calculate total minutes including travel time between locations""" + minute_total = 0 + grouped_pairs = self._group_record_pairs(self.record_pairs) + + for group in grouped_pairs: + group_minutes = self._get_minutes_per_record_pair_group(group) + if group_minutes < 0: + return -1 # Miss punch detected + minute_total += group_minutes + + return self.round_time_to_nearest_quarter_hour(minute_total) + + def get_minutes_total_exclude_travel_time(self) -> int: + """Calculate total minutes excluding travel time""" + minute_total = 0 + + for pair in self.record_pairs: + if not pair.is_miss_punch: + minute_total += pair.duration_minutes + else: + return -1 # Miss punch detected + + return self.round_time_to_nearest_quarter_hour(minute_total) + + def _group_record_pairs(self, record_pairs: List[RecordPair]) -> List[List[RecordPair]]: + """Group record pairs based on time gaps (similar to Java implementation)""" + if not record_pairs: + return [] + + # Sort pairs by time + sorted_pairs = sorted(record_pairs, key=lambda p: p.check_in.timestamp if p.check_in else p.check_out.timestamp) + + all_groups = [] + current_group = [sorted_pairs[0]] + + for i in range(1, len(sorted_pairs)): + prev_pair = sorted_pairs[i-1] + current_pair = sorted_pairs[i] + + # Calculate time gap + if prev_pair.check_out and current_pair.check_in: + gap_minutes = (current_pair.check_in.timestamp - prev_pair.check_out.timestamp).total_seconds() / 60 + + if gap_minutes > TRAVEL_TIME_MAX_MINUTES: + # Start new group + all_groups.append(current_group) + current_group = [current_pair] + else: + current_group.append(current_pair) + else: + current_group.append(current_pair) + + all_groups.append(current_group) + return all_groups + + def _get_minutes_per_record_pair_group(self, group: List[RecordPair]) -> int: + """Calculate minutes for a group of record pairs with travel time""" + if not group: + return 0 + + # Check for miss punches + for pair in group: + if pair.is_miss_punch: + return -1 + + # Find overall start and end times for the group + start_time = min(pair.check_in.timestamp for pair in group if pair.check_in) + end_time = max(pair.check_out.timestamp for pair in group if pair.check_out) + + duration = end_time - start_time + return int(duration.total_seconds() / 60) + + +class WeeklyTimeCalculator(TimeCalculator): + """Calculate weekly regular and overtime hours""" + + def __init__(self): + self.daily_calculators: List[DailyTimeCalculator] = [] + self.total_minutes = 0 + self.regular_minutes = 0 + self.overtime_minutes = 0 + self.include_travel_time = True # Default setting + + def add_daily_calculator(self, daily_calc: DailyTimeCalculator): + """Add a daily time calculator to the weekly calculation""" + self.daily_calculators.append(daily_calc) + + def set_include_travel_time(self, include: bool): + """Set whether to include travel time in calculations""" + self.include_travel_time = include + + def calculate_time(self): + """Calculate weekly totals with regular and overtime split""" + self.total_minutes = 0 + + for daily_calc in self.daily_calculators: + if self.include_travel_time: + daily_minutes = daily_calc.get_minutes_total_include_travel_time() + else: + daily_minutes = daily_calc.get_minutes_total_exclude_travel_time() + + if daily_minutes > 0: + self.total_minutes += daily_minutes + + # Calculate regular and overtime + if self.total_minutes > MAX_REGULAR_TIME_MINUTES: + self.regular_minutes = MAX_REGULAR_TIME_MINUTES + else: + self.regular_minutes = self.total_minutes + + self.overtime_minutes = self.total_minutes - self.regular_minutes + + @property + def total_hours(self) -> float: + """Get total hours as decimal""" + return self.total_minutes / 60.0 + + @property + def regular_hours(self) -> float: + """Get regular hours as decimal""" + return self.regular_minutes / 60.0 + + @property + def overtime_hours(self) -> float: + """Get overtime hours as decimal""" + return self.overtime_minutes / 60.0 + + +class RecordPairBuilder: + """Build record pairs from attendance records""" + + @staticmethod + def build_pairs_from_records(records: List[AttendanceRecord]) -> List[RecordPair]: + """Build check-in/check-out pairs from attendance records""" + if not records: + return [] + + # Sort records by timestamp + sorted_records = sorted(records, key=lambda r: r.timestamp) + pairs = [] + + i = 0 + while i < len(sorted_records): + current_record = sorted_records[i] + + # Look for matching check-out record + check_out_record = None + if i + 1 < len(sorted_records): + next_record = sorted_records[i + 1] + # Simple pairing: assume alternating check-in/check-out + if current_record.record_type == 'check_in' and next_record.record_type == 'check_out': + check_out_record = next_record + i += 2 # Skip both records + else: + i += 1 + else: + i += 1 + + # Create pair + if current_record.record_type == 'check_in': + pair = RecordPair( + check_in=current_record, + check_out=check_out_record, + is_miss_punch=(check_out_record is None) + ) + else: + # Orphaned check-out + pair = RecordPair( + check_in=None, + check_out=current_record, + is_miss_punch=True + ) + + pairs.append(pair) + + return pairs + + +class WorkingHoursCalculator: + """Main calculator for employee working hours""" + + def __init__(self, include_travel_time: bool = True): + self.include_travel_time = include_travel_time + + @log_database_operations('working_hours_calculation') + def calculate_employee_hours(self, employee_id: str, start_date: datetime, end_date: datetime, + attendance_records: List[Dict]) -> Dict[str, Any]: + """ + Calculate working hours for an employee over a date range + + Args: + employee_id: Employee ID + start_date: Start date for calculation + end_date: End date for calculation + attendance_records: List of attendance records from database + + Returns: + Dictionary containing daily and weekly hour calculations + """ + try: + # Convert database records to AttendanceRecord objects + records = [] + for record in attendance_records: + # Handle both dictionary and SQLAlchemy object formats + if hasattr(record, '__dict__'): + # SQLAlchemy object + emp_id = str(record.employee_id) + date_val = record.check_in_date + time_val = record.check_in_time + location = record.location_name + record_id = record.id + else: + # Dictionary + emp_id = str(record['employee_id']) + date_val = record['check_in_date'] + time_val = record['check_in_time'] + location = record['location_name'] + record_id = record['id'] + + if emp_id == employee_id: + att_record = AttendanceRecord( + id=record_id, + employee_id=emp_id, + check_in_date=date_val, + check_in_time=time_val, + location_name=location, + record_type='check_in' # Default, could be enhanced + ) + records.append(att_record) + + # Group records by date + daily_records = {} + for record in records: + date_key = record.check_in_date.strftime('%Y-%m-%d') + if date_key not in daily_records: + daily_records[date_key] = [] + daily_records[date_key].append(record) + + # Calculate daily hours + daily_hours = {} + weekly_calculators = [] + + current_date = start_date + while current_date <= end_date: + date_key = current_date.strftime('%Y-%m-%d') + day_records = daily_records.get(date_key, []) + + daily_calc = DailyTimeCalculator() + + if day_records: + # Build record pairs + pairs = RecordPairBuilder.build_pairs_from_records(day_records) + for pair in pairs: + daily_calc.add_record_pair(pair) + + # Calculate daily totals + if self.include_travel_time: + total_minutes = daily_calc.get_minutes_total_include_travel_time() + else: + total_minutes = daily_calc.get_minutes_total_exclude_travel_time() + + daily_hours[date_key] = { + 'total_minutes': total_minutes, + 'total_hours': total_minutes / 60.0 if total_minutes > 0 else 0, + 'is_miss_punch': total_minutes < 0, + 'records_count': len(day_records) + } + + # Add to weekly calculator (group by week) + if current_date.weekday() == 0: # Monday - start new week + weekly_calc = WeeklyTimeCalculator() + weekly_calc.set_include_travel_time(self.include_travel_time) + weekly_calculators.append(weekly_calc) + + if weekly_calculators: + weekly_calculators[-1].add_daily_calculator(daily_calc) + + current_date += timedelta(days=1) + + # Calculate weekly totals + weekly_hours = [] + for week_calc in weekly_calculators: + week_calc.calculate_time() + weekly_hours.append({ + 'total_hours': week_calc.total_hours, + 'regular_hours': week_calc.regular_hours, + 'overtime_hours': week_calc.overtime_hours, + 'total_minutes': week_calc.total_minutes, + 'regular_minutes': week_calc.regular_minutes, + 'overtime_minutes': week_calc.overtime_minutes + }) + + # Calculate grand totals + grand_total_hours = sum(week['total_hours'] for week in weekly_hours) + grand_regular_hours = sum(week['regular_hours'] for week in weekly_hours) + grand_overtime_hours = sum(week['overtime_hours'] for week in weekly_hours) + + return { + 'employee_id': employee_id, + 'start_date': start_date.strftime('%Y-%m-%d'), + 'end_date': end_date.strftime('%Y-%m-%d'), + 'include_travel_time': self.include_travel_time, + 'daily_hours': daily_hours, + 'weekly_hours': weekly_hours, + 'grand_totals': { + 'total_hours': grand_total_hours, + 'regular_hours': grand_regular_hours, + 'overtime_hours': grand_overtime_hours, + 'total_minutes': int(grand_total_hours * 60), + 'regular_minutes': int(grand_regular_hours * 60), + 'overtime_minutes': int(grand_overtime_hours * 60) + } + } + + except Exception as e: + print(f"❌ Error calculating working hours for employee {employee_id}: {e}") + raise e + + def calculate_all_employees_hours(self, start_date: datetime, end_date: datetime, + attendance_records: List[Dict]) -> Dict[str, Any]: + """Calculate working hours for all employees in the given period""" + try: + # Get unique employee IDs + employee_ids = set() + for record in attendance_records: + if hasattr(record, '__dict__'): + employee_ids.add(str(record.employee_id)) + else: + employee_ids.add(str(record['employee_id'])) + + results = {} + for emp_id in employee_ids: + results[emp_id] = self.calculate_employee_hours(emp_id, start_date, end_date, attendance_records) + + return { + 'calculation_date': datetime.now().strftime('%Y-%m-%d %H:%M:%S'), + 'period_start': start_date.strftime('%Y-%m-%d'), + 'period_end': end_date.strftime('%Y-%m-%d'), + 'include_travel_time': self.include_travel_time, + 'employee_count': len(employee_ids), + 'employees': results + } + + except Exception as e: + print(f"❌ Error calculating hours for all employees: {e}") + raise e \ No newline at end of file