From 86d701a86fd9ab14c43eeab241c1b8accfe25188 Mon Sep 17 00:00:00 2001 From: Nguyen Ngo Date: Thu, 11 Sep 2025 11:47:55 -0400 Subject: [PATCH] Update working hours calculation with SP/PW --- app.py | 150 +++++++--- enhanced_payroll_excel_exporter.py | 391 ++++++++++++++++++++++++++ single_checkin_calculator.py | 435 +++++++++++++++-------------- templates/payroll_dashboard.html | 67 +++++ 4 files changed, 789 insertions(+), 254 deletions(-) create mode 100644 enhanced_payroll_excel_exporter.py diff --git a/app.py b/app.py index 88f6c4b..699211b 100644 --- a/app.py +++ b/app.py @@ -14,6 +14,7 @@ from logger_handler import AppLogger, log_user_activity, log_database_operations from single_checkin_calculator import SingleCheckInCalculator from payroll_excel_exporter import PayrollExcelExporter +from enhanced_payroll_excel_exporter import EnhancedPayrollExcelExporter # Load environment variables in .env load_dotenv() @@ -5182,7 +5183,7 @@ def payroll_dashboard(): @login_required @log_database_operations('payroll_excel_export') def export_payroll_excel(): - """Export payroll report to Excel with working hours calculations""" + """Export payroll report to Excel with working hours calculations including SP/PW support""" try: # Check permissions user_role = session.get('role') @@ -5197,7 +5198,7 @@ def export_payroll_excel(): date_from = request.form.get('date_from') date_to = request.form.get('date_to') project_filter = request.form.get('project_filter', '') - report_type = request.form.get('report_type', 'payroll') # 'payroll' or 'detailed' + report_type = request.form.get('report_type', 'payroll') # 'payroll', 'detailed', 'template', 'enhanced', 'detailed_sp_pw' if not date_from or not date_to: flash('Please provide both start and end dates for the export.', 'error') @@ -5243,12 +5244,6 @@ def export_payroll_excel(): print(f"šŸ“Š Exporting {len(attendance_records)} attendance records to Excel") - # Create Excel exporter - exporter = PayrollExcelExporter( - company_name=os.environ.get('COMPANY_NAME', 'Your Company'), - contract_name=os.environ.get('CONTRACT_NAME', 'Default Contract') - ) - # Get employee names using the same method as dashboard employee_names = {} try: @@ -5277,47 +5272,118 @@ def export_payroll_excel(): import traceback print(f"āš ļø Traceback: {traceback.format_exc()}") - # Generate Excel file with employee names - if report_type == 'detailed': - excel_file = exporter.create_detailed_hours_report( - start_date, end_date, attendance_records, employee_names - ) - filename_prefix = 'detailed_hours_report' - elif report_type == 'template': - # Get project name for the template - project_name = None - if project_filter: - try: - project = Project.query.get(int(project_filter)) - if project: - project_name = project.name - except Exception as e: - print(f"āš ļø Error getting project name for template: {e}") - - excel_file = exporter.create_template_format_report( - start_date, end_date, attendance_records, employee_names, project_name - ) - filename_prefix = 'time_attendance_report' - else: - excel_file = exporter.create_payroll_report( - start_date, end_date, attendance_records, employee_names - ) - filename_prefix = 'payroll_report' - - # Get project name for filename - project_name = '' + # Get project name for enhanced reports and filename + project_name = None + project_name_for_filename = '' if project_filter: try: project = Project.query.get(int(project_filter)) if project: - project_name = f"_{project.name.replace(' ', '_')}" + project_name = project.name + project_name_for_filename = f"_{project.name.replace(' ', '_')}" except Exception as e: - print(f"āš ļø Error getting project name for filename: {e}") + print(f"āš ļø Error getting project name: {e}") + + # Generate Excel file based on report type + excel_file = None + filename_prefix = 'payroll_report' + + if report_type == 'enhanced': + # Use enhanced exporter for SP/PW reports + print("šŸ“Š Creating enhanced payroll report with SP/PW support") + try: + from enhanced_payroll_excel_exporter import EnhancedPayrollExcelExporter + exporter = EnhancedPayrollExcelExporter(company_name=os.environ.get('COMPANY_NAME', 'Your Company')) + excel_file = exporter.create_enhanced_payroll_report( + start_date, end_date, attendance_records, employee_names, project_name + ) + filename_prefix = 'enhanced_payroll_report' + print("āœ… Enhanced payroll report created successfully") + except ImportError: + print("āš ļø Enhanced exporter not available, falling back to standard exporter") + # Fall back to standard exporter + exporter = PayrollExcelExporter( + company_name=os.environ.get('COMPANY_NAME', 'Your Company'), + contract_name=os.environ.get('CONTRACT_NAME', 'Default Contract') + ) + excel_file = exporter.create_payroll_report( + start_date, end_date, attendance_records, employee_names + ) + filename_prefix = 'payroll_report' + except Exception as e: + print(f"āš ļø Error with enhanced exporter: {e}, falling back to standard exporter") + # Fall back to standard exporter + exporter = PayrollExcelExporter( + company_name=os.environ.get('COMPANY_NAME', 'Your Company'), + contract_name=os.environ.get('CONTRACT_NAME', 'Default Contract') + ) + excel_file = exporter.create_payroll_report( + start_date, end_date, attendance_records, employee_names + ) + filename_prefix = 'payroll_report' + + elif report_type == 'detailed_sp_pw': + # Detailed daily SP/PW breakdown + print("šŸ“Š Creating detailed SP/PW daily breakdown report") + try: + from enhanced_payroll_excel_exporter import EnhancedPayrollExcelExporter + exporter = EnhancedPayrollExcelExporter(company_name=os.environ.get('COMPANY_NAME', 'Your Company')) + excel_file = exporter.create_detailed_sp_pw_report( + start_date, end_date, attendance_records, employee_names + ) + filename_prefix = 'detailed_sp_pw_report' + print("āœ… Detailed SP/PW report created successfully") + except ImportError: + print("āš ļø Enhanced exporter not available, falling back to detailed hours report") + # Fall back to standard detailed report + exporter = PayrollExcelExporter( + company_name=os.environ.get('COMPANY_NAME', 'Your Company'), + contract_name=os.environ.get('CONTRACT_NAME', 'Default Contract') + ) + excel_file = exporter.create_detailed_hours_report( + start_date, end_date, attendance_records, employee_names + ) + filename_prefix = 'detailed_hours_report' + except Exception as e: + print(f"āš ļø Error with enhanced exporter: {e}, falling back to detailed hours report") + # Fall back to standard detailed report + exporter = PayrollExcelExporter( + company_name=os.environ.get('COMPANY_NAME', 'Your Company'), + contract_name=os.environ.get('CONTRACT_NAME', 'Default Contract') + ) + excel_file = exporter.create_detailed_hours_report( + start_date, end_date, attendance_records, employee_names + ) + filename_prefix = 'detailed_hours_report' + + else: + # Use standard exporter for existing report types + exporter = PayrollExcelExporter( + company_name=os.environ.get('COMPANY_NAME', 'Your Company'), + contract_name=os.environ.get('CONTRACT_NAME', 'Default Contract') + ) + + if report_type == 'detailed': + excel_file = exporter.create_detailed_hours_report( + start_date, end_date, attendance_records, employee_names + ) + filename_prefix = 'detailed_hours_report' + elif report_type == 'template': + excel_file = exporter.create_template_format_report( + start_date, end_date, attendance_records, employee_names, project_name + ) + filename_prefix = 'time_attendance_report' + else: + # Default payroll report + excel_file = exporter.create_payroll_report( + start_date, end_date, attendance_records, employee_names + ) + filename_prefix = 'payroll_report' if excel_file: # Generate filename with timestamp and project name timestamp = datetime.now().strftime('%Y%m%d_%H%M%S') - filename = f'{filename_prefix}_{date_from}_to_{date_to}{project_name}_{timestamp}.xlsx' + filename = f'{filename_prefix}_{date_from}_to_{date_to}{project_name_for_filename}_{timestamp}.xlsx' print(f"šŸ“Š Payroll Excel file generated successfully: {filename}") @@ -5325,6 +5391,10 @@ def export_payroll_excel(): logger_handler.logger.info(f"Payroll Excel export generated by user {session.get('username', 'unknown')}: {filename}") if report_type == 'template': logger_handler.logger.info(f"Template format hours export generated by user {session.get('username', 'unknown')}: {filename}") + elif report_type == 'enhanced': + logger_handler.logger.info(f"Enhanced payroll export with SP/PW generated by user {session.get('username', 'unknown')}: {filename}") + elif report_type == 'detailed_sp_pw': + logger_handler.logger.info(f"Detailed SP/PW breakdown export generated by user {session.get('username', 'unknown')}: {filename}") return send_file( excel_file, diff --git a/enhanced_payroll_excel_exporter.py b/enhanced_payroll_excel_exporter.py new file mode 100644 index 0000000..4b3863a --- /dev/null +++ b/enhanced_payroll_excel_exporter.py @@ -0,0 +1,391 @@ +""" +Enhanced Payroll Excel Exporter with SP/PW Support +================================================= + +Extends the existing PayrollExcelExporter to support Special Project (SP) +and Periodic Work (PW) hours in Excel reports. +""" + +import io +from datetime import datetime, timedelta +from typing import List, Dict, Optional, Any +from openpyxl import Workbook +from openpyxl.styles import Font, PatternFill, Alignment, Border, Side +from openpyxl.utils import get_column_letter +from single_checkin_calculator import SingleCheckInCalculator +from logger_handler import log_database_operations + + +class EnhancedPayrollExcelExporter: + """Enhanced Excel exporter with SP/PW support""" + + def __init__(self, company_name: str = "Your Company Name"): + self.company_name = company_name + self._setup_styles() + + def _setup_styles(self): + """Setup Excel cell styles""" + # Header styles + self.header_style = { + 'font': Font(name="Arial", size=11, bold=True, color="FFFFFF"), + 'fill': PatternFill(start_color="4472C4", end_color="4472C4", 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 styles + 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") + ) + } + + # Summary styles + self.summary_style = { + 'font': Font(name="Arial", size=11, bold=True), + 'fill': PatternFill(start_color="E2EFDA", end_color="E2EFDA", 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") + ) + } + + def _apply_style(self, cell, style_dict): + """Apply style dictionary to a cell""" + for attr, value in style_dict.items(): + setattr(cell, attr, value) + + @log_database_operations('enhanced_payroll_export') + def create_enhanced_payroll_report(self, start_date: datetime, end_date: datetime, + attendance_records: List[Dict], employee_names: Dict[str, str] = None, + project_name: str = None) -> io.BytesIO: + """ + Create enhanced payroll report with SP/PW 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 + project_name: Project name for the report + + Returns: + BytesIO buffer containing the Excel file + """ + try: + print(f"šŸ“Š Creating enhanced payroll report with SP/PW support from {start_date.strftime('%Y-%m-%d')} to {end_date.strftime('%Y-%m-%d')}") + + # Get employee names using existing method + if employee_names is None: + employee_names = self._get_employee_names(attendance_records) + + # Calculate working hours using enhanced calculator + calculator = SingleCheckInCalculator() + working_hours_data = calculator.calculate_all_employees_hours( + start_date, end_date, attendance_records + ) + + # Create workbook + workbook = Workbook() + worksheet = workbook.active + worksheet.title = "Enhanced Payroll Report" + + # Write report header + current_row = self._write_enhanced_header(worksheet, start_date, end_date, project_name) + + # Write column headers + headers = [ + "Employee ID", "Employee Name", "Total Hours", "Regular Hours", + "Overtime Hours", "SP Hours", "PW Hours", "Working Days", "Status" + ] + + for col, header in enumerate(headers, 1): + cell = worksheet.cell(row=current_row, column=col, value=header) + self._apply_style(cell, self.header_style) + current_row += 1 + + # Write employee data + total_employees = 0 + totals = { + 'total_hours': 0.0, + 'regular_hours': 0.0, + 'overtime_hours': 0.0, + 'sp_hours': 0.0, + 'pw_hours': 0.0 + } + + for employee_id, emp_data in working_hours_data['employees'].items(): + employee_name = employee_names.get(employee_id, f'Employee {employee_id}') + + # Calculate working days and status + working_days = len([d for d in emp_data['daily_hours'].values() if d['total_hours'] > 0]) + miss_punches = len([d for d in emp_data['daily_hours'].values() if d['is_miss_punch']]) + status = f"{miss_punches} miss punch(es)" if miss_punches > 0 else "Complete" + + # Get grand totals + grand_totals = emp_data['grand_totals'] + + row_data = [ + employee_id, + employee_name, + round(grand_totals['total_hours'], 2), + round(grand_totals['regular_hours'], 2), + round(grand_totals['overtime_hours'], 2), + round(grand_totals['sp_hours'], 2), + round(grand_totals['pw_hours'], 2), + working_days, + status + ] + + 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) + + # Highlight miss punches + if miss_punches > 0 and col == 9: # Status column + cell.fill = PatternFill(start_color="FFE6E6", end_color="FFE6E6", fill_type="solid") + + # Add to totals + totals['total_hours'] += grand_totals['total_hours'] + totals['regular_hours'] += grand_totals['regular_hours'] + totals['overtime_hours'] += grand_totals['overtime_hours'] + totals['sp_hours'] += grand_totals['sp_hours'] + totals['pw_hours'] += grand_totals['pw_hours'] + total_employees += 1 + + current_row += 1 + + # Write summary section + current_row = self._write_enhanced_summary(worksheet, current_row, total_employees, totals) + + # Auto-adjust columns + self._auto_adjust_columns(worksheet) + + # Save to BytesIO + excel_buffer = io.BytesIO() + workbook.save(excel_buffer) + excel_buffer.seek(0) + + print("āœ… Enhanced payroll report Excel file created successfully") + return excel_buffer + + except Exception as e: + print(f"āŒ Error creating enhanced payroll report: {e}") + raise e + + def _write_enhanced_header(self, worksheet, start_date: datetime, end_date: datetime, project_name: str) -> int: + """Write enhanced report header""" + current_row = 1 + + # Company name + cell = worksheet.cell(row=current_row, column=1, value=self.company_name) + cell.font = Font(name="Arial", size=14, bold=True) + worksheet.merge_cells(f'A{current_row}:I{current_row}') + current_row += 1 + + # Report title + cell = worksheet.cell(row=current_row, column=1, value="Enhanced Payroll Report with SP/PW Hours") + cell.font = Font(name="Arial", size=12, bold=True) + worksheet.merge_cells(f'A{current_row}:I{current_row}') + current_row += 1 + + # Project name + if project_name: + cell = worksheet.cell(row=current_row, column=1, value=f"Project: {project_name}") + cell.font = Font(name="Arial", size=11) + worksheet.merge_cells(f'A{current_row}:I{current_row}') + current_row += 1 + + # Date range + date_range = f"Period: {start_date.strftime('%Y-%m-%d')} to {end_date.strftime('%Y-%m-%d')}" + cell = worksheet.cell(row=current_row, column=1, value=date_range) + cell.font = Font(name="Arial", size=11) + worksheet.merge_cells(f'A{current_row}:I{current_row}') + current_row += 2 # Add space + + return current_row + + def _write_enhanced_summary(self, worksheet, start_row: int, total_employees: int, totals: Dict) -> int: + """Write enhanced summary section""" + current_row = start_row + 1 + + # Summary header + cell = worksheet.cell(row=current_row, column=1, value="SUMMARY") + cell.font = Font(name="Arial", size=12, bold=True) + worksheet.merge_cells(f'A{current_row}:I{current_row}') + current_row += 1 + + # Summary data + summary_data = [ + ("Total Employees:", total_employees), + ("Total Hours:", round(totals['total_hours'], 2)), + ("Regular Hours:", round(totals['regular_hours'], 2)), + ("Overtime Hours:", round(totals['overtime_hours'], 2)), + ("Special Project (SP) Hours:", round(totals['sp_hours'], 2)), + ("Periodic Work (PW) Hours:", round(totals['pw_hours'], 2)), + ("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.summary_style) + cell2 = worksheet.cell(row=current_row, column=2, value=value) + self._apply_style(cell2, self.summary_style) + current_row += 1 + + return current_row + + def _get_employee_names(self, attendance_records: List[Dict]) -> Dict[str, str]: + """Get employee names from attendance records""" + # This method should be implemented based on your existing database structure + # For now, return a basic implementation + employee_names = {} + for record in attendance_records: + if hasattr(record, '__dict__'): + emp_id = str(record.employee_id) + if emp_id not in employee_names: + employee_names[emp_id] = f'Employee {emp_id}' + else: + emp_id = str(record['employee_id']) + if emp_id not in employee_names: + employee_names[emp_id] = f'Employee {emp_id}' + return employee_names + + 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, 25) + worksheet.column_dimensions[column_letter].width = adjusted_width + + @log_database_operations('detailed_sp_pw_export') + def create_detailed_sp_pw_report(self, start_date: datetime, end_date: datetime, + attendance_records: List[Dict], employee_names: Dict[str, str] = None) -> io.BytesIO: + """ + Create detailed daily report showing SP/PW breakdown by day + """ + try: + print(f"šŸ“Š Creating detailed SP/PW daily report") + + # Get employee names + if employee_names is None: + employee_names = self._get_employee_names(attendance_records) + + # Calculate working hours + calculator = SingleCheckInCalculator() + working_hours_data = calculator.calculate_all_employees_hours( + start_date, end_date, attendance_records + ) + + # Create workbook + workbook = Workbook() + worksheet = workbook.active + worksheet.title = "Detailed SP/PW Report" + + # Write header + current_row = 1 + cell = worksheet.cell(row=current_row, column=1, value="Detailed Daily Hours Report with SP/PW Breakdown") + cell.font = Font(name="Arial", size=14, bold=True) + worksheet.merge_cells(f'A{current_row}:K{current_row}') + current_row += 2 + + # Column headers + headers = [ + "Employee ID", "Employee Name", "Date", "Day", "Regular Hours", + "SP Hours", "PW Hours", "Total Hours", "Records", "Status", "Notes" + ] + + for col, header in enumerate(headers, 1): + cell = worksheet.cell(row=current_row, column=col, value=header) + self._apply_style(cell, self.header_style) + current_row += 1 + + # Write daily data for each employee + for employee_id, emp_data in working_hours_data['employees'].items(): + employee_name = employee_names.get(employee_id, f'Employee {employee_id}') + + for date_str, day_data in emp_data['daily_hours'].items(): + date_obj = datetime.strptime(date_str, '%Y-%m-%d') + day_name = date_obj.strftime('%A') + + # Get hours by type + regular_hours = day_data.get('regular_hours', 0.0) + sp_hours = day_data.get('sp_hours', 0.0) + pw_hours = day_data.get('pw_hours', 0.0) + total_hours = day_data.get('total_hours', 0.0) + + # Status and notes + if day_data.get('is_miss_punch', False): + status = "Miss Punch" + notes = "Incomplete records" + elif total_hours == 0: + status = "No Work" + notes = "No attendance records" + else: + status = "Complete" + notes = f"{day_data['records_count']} record(s)" + + row_data = [ + employee_id, + employee_name, + date_str, + day_name, + round(regular_hours, 2), + round(sp_hours, 2), + round(pw_hours, 2), + round(total_hours, 2), + day_data['records_count'], + status, + 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 coding + if status == "Miss Punch": + cell.fill = PatternFill(start_color="FFE6E6", end_color="FFE6E6", fill_type="solid") + elif sp_hours > 0 and col == 6: # SP Hours column + cell.fill = PatternFill(start_color="E6F3FF", end_color="E6F3FF", fill_type="solid") + elif pw_hours > 0 and col == 7: # PW Hours column + cell.fill = PatternFill(start_color="FFF2E6", end_color="FFF2E6", 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 SP/PW report Excel file created successfully") + return excel_buffer + + except Exception as e: + print(f"āŒ Error creating detailed SP/PW report: {e}") + raise e \ No newline at end of file diff --git a/single_checkin_calculator.py b/single_checkin_calculator.py index ce4d15c..6fc763b 100644 --- a/single_checkin_calculator.py +++ b/single_checkin_calculator.py @@ -1,165 +1,175 @@ #!/usr/bin/env python3 """ -Single Check-in Working Hours Calculator -======================================= +Safe Enhanced Single Check-in Working Hours Calculator with SP/PW Support +======================================================================== -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. +This version maintains full backward compatibility while adding SP/PW support. +It gracefully handles missing data and falls back to standard calculation when needed. """ from datetime import datetime, timedelta, time from typing import List, Dict, Optional, Tuple, Any from dataclasses import dataclass import math +import re 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 parse_employee_id_for_work_type(employee_id: str) -> Tuple[str, str]: + """Parse employee ID to extract base ID and work type""" + if not employee_id: + return str(employee_id), 'regular' - 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 + employee_id_clean = str(employee_id).strip().upper() - 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 + # Check for SP (Special Project) + sp_pattern = r'^(\d+)\s*SP$' + sp_match = re.match(sp_pattern, employee_id_clean) + if sp_match: + return sp_match.group(1), 'SP' + + # Check for PW (Periodic Work) + pw_pattern = r'^(\d+)\s*PW$' + pw_match = re.match(pw_pattern, employee_id_clean) + if pw_match: + return pw_match.group(1), 'PW' + + # Default to regular work + return employee_id_clean, 'regular' class SingleCheckInCalculator: - """Calculator for single check-in attendance systems""" + """Enhanced calculator for single check-in attendance systems with SP/PW support""" 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 + self.max_work_period_hours = max_work_period_hours + self.min_break_minutes = min_break_minutes - @log_database_operations('single_checkin_hours_calculation') + @log_database_operations('single_checkin_hours_calculation_sp_pw') 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 + Calculate working hours for an employee with SP/PW support and robust error handling """ try: - print(f"šŸ” Calculating hours for employee {employee_id}") + print(f"šŸ” Calculating hours for employee {employee_id} with SP/PW support") + + # Parse base employee ID and work type + base_employee_id, _ = parse_employee_id_for_work_type(employee_id) + + # Filter and categorize records + records_by_type = {'regular': [], 'SP': [], 'PW': []} - # 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) + try: + # Extract record data safely + if hasattr(record, '__dict__'): + record_emp_id = str(getattr(record, 'employee_id', '')).strip() + record_date = getattr(record, 'check_in_date', None) + record_time = getattr(record, 'check_in_time', None) + location = getattr(record, 'location_name', 'Unknown Location') + record_id = getattr(record, 'id', 0) + else: + record_emp_id = str(record.get('employee_id', '')).strip() + record_date = record.get('check_in_date') + record_time = record.get('check_in_time') + location = record.get('location_name', 'Unknown Location') + record_id = record.get('id', 0) + + # Skip invalid records + if not record_emp_id or record_date is None or record_time is None: + continue + + # Parse work type from this record + record_base_id, work_type = parse_employee_id_for_work_type(record_emp_id) + + # Only include records for this base employee + if record_base_id == base_employee_id: + # Create a simple record dict for processing + processed_record = { + 'id': record_id, + 'employee_id': record_emp_id, + 'check_in_date': record_date, + 'check_in_time': record_time, + 'location_name': location, + 'work_type': work_type, + 'timestamp': datetime.combine(record_date, record_time) if record_date and record_time else datetime.now() + } + records_by_type[work_type].append(processed_record) + + except Exception as record_error: + print(f"āš ļø Error processing record: {record_error}") + continue - if not records: - print(f"āš ļø No records found for employee {employee_id}") - return self._empty_result(employee_id, start_date, end_date) + total_records = sum(len(records_by_type[wt]) for wt in records_by_type) + print(f"šŸ“Š Found {total_records} records - Regular: {len(records_by_type['regular'])}, SP: {len(records_by_type['SP'])}, PW: {len(records_by_type['PW'])}") - 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 + # Calculate hours for each work type daily_hours = {} weekly_totals = [] - current_date = start_date - current_week_hours = [] + current_week_hours = {'regular': [], 'SP': [], 'PW': []} 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) + # Calculate hours for each work type on this day + hours_by_type = {} + is_miss_punch_by_type = {} + for work_type in ['regular', 'SP', 'PW']: + day_records = [r for r in records_by_type[work_type] + if r['check_in_date'] == current_date.date()] + hours, is_miss_punch = self._calculate_daily_hours_from_records(day_records) + hours_by_type[work_type] = hours + is_miss_punch_by_type[work_type] = is_miss_punch + + # Store daily data with SP/PW support + total_day_hours = sum(hours_by_type.values()) 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) + 'total_minutes': int(total_day_hours * 60), + 'total_hours': total_day_hours, + 'regular_hours': hours_by_type['regular'], + 'sp_hours': hours_by_type['SP'], + 'pw_hours': hours_by_type['PW'], + 'is_miss_punch': any(is_miss_punch_by_type.values()), + 'records_count': sum(len([r for r in records_by_type[wt] if r['check_in_date'] == current_date.date()]) + for wt in ['regular', 'SP', 'PW']), + 'miss_punch_details': { + 'regular': is_miss_punch_by_type['regular'], + 'SP': is_miss_punch_by_type['SP'], + 'PW': is_miss_punch_by_type['PW'] + } } - print(f"šŸ“… {date_key}: {len(day_records)} records, {day_hours:.2f} hours, miss_punch: {is_miss_punch}") + # Add to weekly calculation + for work_type in ['regular', 'SP', 'PW']: + current_week_hours[work_type].append(max(0, hours_by_type[work_type])) - # 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) + # Check if end of week or end of period + if current_date.weekday() == 6 or current_date == end_date: + week_regular_total = sum(current_week_hours['regular']) + week_sp_total = sum(current_week_hours['SP']) + week_pw_total = sum(current_week_hours['PW']) + + # Only regular hours count toward overtime + week_regular_hours = min(week_regular_total, 40.0) + week_overtime_hours = max(0, week_regular_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) + 'total_hours': week_regular_total + week_sp_total + week_pw_total, + 'regular_hours': week_regular_hours, + 'overtime_hours': week_overtime_hours, + 'sp_hours': week_sp_total, + 'pw_hours': week_pw_total, + 'total_minutes': int((week_regular_total + week_sp_total + week_pw_total) * 60), + 'regular_minutes': int(week_regular_hours * 60), + 'overtime_minutes': int(week_overtime_hours * 60), + 'sp_minutes': int(week_sp_total * 60), + 'pw_minutes': int(week_pw_total * 60) }) - current_week_hours = [] + current_week_hours = {'regular': [], 'SP': [], 'PW': []} current_date += timedelta(days=1) @@ -167,125 +177,92 @@ class SingleCheckInCalculator: 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) + grand_sp_hours = sum(week['sp_hours'] for week in weekly_totals) + grand_pw_hours = sum(week['pw_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") + print(f"āœ… Employee {employee_id}: Total: {grand_total_hours:.2f}h (Regular: {grand_regular_hours:.2f}h, OT: {grand_overtime_hours:.2f}h, SP: {grand_sp_hours:.2f}h, PW: {grand_pw_hours:.2f}h)") - return { + result = { 'employee_id': employee_id, + 'base_employee_id': base_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 + 'include_travel_time': True, '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, + 'sp_hours': grand_sp_hours, + 'pw_hours': grand_pw_hours, 'total_minutes': int(grand_total_hours * 60), 'regular_minutes': int(grand_regular_hours * 60), - 'overtime_minutes': int(grand_overtime_hours * 60) + 'overtime_minutes': int(grand_overtime_hours * 60), + 'sp_minutes': int(grand_sp_hours * 60), + 'pw_minutes': int(grand_pw_hours * 60) } } + return result + 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 - - CORRECTED Logic: - - 0 records = No work (0 hours, not miss punch) - - 1 record = Miss punch (incomplete pair) - - 2, 4, 6, 8... records = Complete pairs, calculate all - - 3, 5, 7, 9... records = Calculate complete pairs only, ignore last odd record - - Returns: (hours, is_miss_punch) - """ + def _calculate_daily_hours_from_records(self, day_records: List[Dict]) -> Tuple[float, bool]: + """Calculate hours for a single day from processed records""" if not day_records: - return 0.0, False # No records = no work + return 0.0, False - # Sort records by time - sorted_records = sorted(day_records, key=lambda r: r.timestamp) - print(f"šŸ“ Processing {len(sorted_records)} records for daily calculation") + # Sort records by timestamp + try: + sorted_records = sorted(day_records, key=lambda r: r['timestamp']) + except Exception as e: + print(f"āš ļø Error sorting records: {e}") + return 0.0, True - # SINGLE RECORD = MISS PUNCH (CORRECTED) + # Single record = miss punch if len(sorted_records) == 1: - print(f"āš ļø Single record found - Miss punch (incomplete pair)") - return 0.0, True # Single record is always miss punch + return 0.0, True - # CALCULATE COMPLETE PAIRS ONLY (CORRECTED FOR ODD NUMBERS) - # For odd numbers: process pairs and ignore the last unpaired record + # Calculate complete pairs only num_complete_pairs = len(sorted_records) // 2 - records_to_process = num_complete_pairs * 2 # Only process paired records - - print(f"šŸ“Š Processing {num_complete_pairs} complete pairs from {len(sorted_records)} total records") - - work_periods = [] total_hours = 0.0 - # Process complete pairs only - for i in range(0, records_to_process, 2): - start_record = sorted_records[i] - end_record = sorted_records[i + 1] - - period = WorkPeriod(start_record, end_record) - - # Validate work period duration - if self._is_valid_work_period(period): - work_periods.append(period) - pair_hours = period.duration_minutes / 60.0 - total_hours += pair_hours - print(f"āœ… Valid work period: {start_record.check_in_time} - {end_record.check_in_time} = {pair_hours:.2f} hours") - else: - print(f"āš ļø Invalid work period: {period.duration_minutes/60:.2f} hours - treating as miss punch") - return 0.0, True # Invalid period = miss punch - - # Check if we had unpaired records (odd number) - has_unpaired = len(sorted_records) % 2 != 0 - if has_unpaired: - unpaired_record = sorted_records[-1] - print(f"āš ļø Unpaired record found: {unpaired_record.check_in_time} (ignored in calculation)") + for i in range(0, num_complete_pairs * 2, 2): + try: + start_time = sorted_records[i]['timestamp'] + end_time = sorted_records[i + 1]['timestamp'] + + duration = end_time - start_time + period_hours = duration.total_seconds() / 3600.0 + + # Validate reasonable work period + if 0 < period_hours <= self.max_work_period_hours: + total_hours += period_hours + + except Exception as e: + print(f"āš ļø Error calculating work period: {e}") + continue # Round to nearest quarter hour - rounded_hours = round(total_hours * 4) / 4 + total_hours = round(total_hours * 4) / 4 - # Determine if this is a miss punch scenario - is_miss_punch = (num_complete_pairs == 0) # No valid pairs = miss punch + # Determine if miss punch + is_miss_punch = len(sorted_records) % 2 != 0 - if is_miss_punch: - print(f"āš ļø No valid work periods found - Miss punch") - return 0.0, True - else: - print(f"āœ… Daily total: {rounded_hours:.2f} hours from {num_complete_pairs} complete work periods") - # Log the calculation for tracking - print(f"šŸ“Š CORRECTED: Employee daily hours calculated - {rounded_hours:.2f} hours, Miss punch: {is_miss_punch}") - 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 + return total_hours, is_miss_punch def _empty_result(self, employee_id: str, start_date: datetime, end_date: datetime) -> Dict[str, Any]: - """Return empty result structure""" + """Return empty result structure with SP/PW support""" + base_employee_id, _ = parse_employee_id_for_work_type(employee_id) + return { 'employee_id': employee_id, + 'base_employee_id': base_employee_id, 'start_date': start_date.strftime('%Y-%m-%d'), 'end_date': end_date.strftime('%Y-%m-%d'), 'include_travel_time': True, @@ -295,39 +272,60 @@ class SingleCheckInCalculator: 'total_hours': 0.0, 'regular_hours': 0.0, 'overtime_hours': 0.0, + 'sp_hours': 0.0, + 'pw_hours': 0.0, 'total_minutes': 0, 'regular_minutes': 0, - 'overtime_minutes': 0 + 'overtime_minutes': 0, + 'sp_minutes': 0, + 'pw_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""" + """Calculate working hours for all employees with robust error handling""" try: - print(f"šŸš€ Starting calculation for all employees") + print(f"šŸš€ Starting calculation for all employees with SP/PW support") - # Get unique employee IDs - employee_ids = set() + # Get unique base employee IDs safely + base_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'])) + try: + if hasattr(record, '__dict__'): + employee_id = str(getattr(record, 'employee_id', '')).strip() + else: + employee_id = str(record.get('employee_id', '')).strip() + + if employee_id: + base_id, _ = parse_employee_id_for_work_type(employee_id) + if base_id: + base_employee_ids.add(base_id) + + except Exception as e: + print(f"āš ļø Error processing employee ID: {e}") + continue - print(f"šŸ‘„ Found {len(employee_ids)} unique employees") + print(f"šŸ‘„ Found {len(base_employee_ids)} unique base 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) + for base_emp_id in sorted(base_employee_ids): + try: + print(f"\nšŸ”„ Processing base employee {base_emp_id}") + results[base_emp_id] = self.calculate_employee_hours( + base_emp_id, start_date, end_date, attendance_records + ) + except Exception as e: + print(f"āŒ Error processing employee {base_emp_id}: {e}") + results[base_emp_id] = self._empty_result(base_emp_id, start_date, end_date) + continue 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), + 'include_travel_time': True, + 'employee_count': len(base_employee_ids), 'employees': results } @@ -335,4 +333,13 @@ class SingleCheckInCalculator: 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 + + # Return minimal safe result + 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, + 'employee_count': 0, + 'employees': {} + } \ No newline at end of file diff --git a/templates/payroll_dashboard.html b/templates/payroll_dashboard.html index c1a8570..765a7bf 100644 --- a/templates/payroll_dashboard.html +++ b/templates/payroll_dashboard.html @@ -141,6 +141,18 @@ box-shadow: 0 8px 32px rgba(0, 0, 0, 0.1); } + .hours-sp { + background: #e6f3ff !important; + color: #0066cc !important; + border: 1px solid #b3d9ff; + } + + .hours-pw { + background: #fff2e6 !important; + color: #cc6600 !important; + border: 1px solid #ffcc99; + } + .summary-cards { display: grid; grid-template-columns: repeat(auto-fit, minmax(200px, 1fr)); @@ -709,6 +721,14 @@

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

Overtime Hours

+
+

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

+

SP Hours

+
+
+

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

+

PW Hours

+
@@ -721,6 +741,8 @@ Total Hours Regular Hours Overtime Hours + SP Hours + PW Hours Working Days Status @@ -751,6 +773,28 @@ {% endif %} + + {% if emp_data.grand_totals.sp_hours > 0 %} + + {{ "%.2f"|format(emp_data.grand_totals.sp_hours) }} hrs + + {% else %} + + 0.00 hrs + + {% endif %} + + + {% if emp_data.grand_totals.pw_hours > 0 %} + + {{ "%.2f"|format(emp_data.grand_totals.pw_hours) }} hrs + + {% else %} + + 0.00 hrs + + {% endif %} + {% set working_days = emp_data.daily_hours.values() | selectattr('total_hours', '>', 0) | list | length %} {{ working_days }} days @@ -808,6 +852,29 @@ Export Time Attendance +