Update working hours calculation with SP/PW

This commit is contained in:
Nguyen Ngo
2025-09-11 11:47:55 -04:00
parent 81da08ad58
commit 86d701a86f
4 changed files with 789 additions and 254 deletions
+110 -40
View File
@@ -14,6 +14,7 @@ from logger_handler import AppLogger, log_user_activity, log_database_operations
from single_checkin_calculator import SingleCheckInCalculator from single_checkin_calculator import SingleCheckInCalculator
from payroll_excel_exporter import PayrollExcelExporter from payroll_excel_exporter import PayrollExcelExporter
from enhanced_payroll_excel_exporter import EnhancedPayrollExcelExporter
# Load environment variables in .env # Load environment variables in .env
load_dotenv() load_dotenv()
@@ -5182,7 +5183,7 @@ def payroll_dashboard():
@login_required @login_required
@log_database_operations('payroll_excel_export') @log_database_operations('payroll_excel_export')
def export_payroll_excel(): 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: try:
# Check permissions # Check permissions
user_role = session.get('role') user_role = session.get('role')
@@ -5197,7 +5198,7 @@ def export_payroll_excel():
date_from = request.form.get('date_from') date_from = request.form.get('date_from')
date_to = request.form.get('date_to') date_to = request.form.get('date_to')
project_filter = request.form.get('project_filter', '') 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: if not date_from or not date_to:
flash('Please provide both start and end dates for the export.', 'error') 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") 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 # Get employee names using the same method as dashboard
employee_names = {} employee_names = {}
try: try:
@@ -5277,47 +5272,118 @@ def export_payroll_excel():
import traceback import traceback
print(f"⚠️ Traceback: {traceback.format_exc()}") print(f"⚠️ Traceback: {traceback.format_exc()}")
# Generate Excel file with employee names # Get project name for enhanced reports and filename
if report_type == 'detailed': project_name = None
excel_file = exporter.create_detailed_hours_report( project_name_for_filename = ''
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 = ''
if project_filter: if project_filter:
try: try:
project = Project.query.get(int(project_filter)) project = Project.query.get(int(project_filter))
if project: if project:
project_name = f"_{project.name.replace(' ', '_')}" project_name = project.name
project_name_for_filename = f"_{project.name.replace(' ', '_')}"
except Exception as e: 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: if excel_file:
# Generate filename with timestamp and project name # Generate filename with timestamp and project name
timestamp = datetime.now().strftime('%Y%m%d_%H%M%S') 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}") 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}") logger_handler.logger.info(f"Payroll Excel export generated by user {session.get('username', 'unknown')}: {filename}")
if report_type == 'template': if report_type == 'template':
logger_handler.logger.info(f"Template format hours export generated by user {session.get('username', 'unknown')}: {filename}") 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( return send_file(
excel_file, excel_file,
+391
View File
@@ -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
+212 -205
View File
@@ -1,165 +1,175 @@
#!/usr/bin/env python3 #!/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 This version maintains full backward compatibility while adding SP/PW support.
attendance record represents a check-in only (not check-in/check-out pairs). It gracefully handles missing data and falls back to standard calculation when needed.
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 datetime import datetime, timedelta, time
from typing import List, Dict, Optional, Tuple, Any from typing import List, Dict, Optional, Tuple, Any
from dataclasses import dataclass from dataclasses import dataclass
import math import math
import re
from logger_handler import log_database_operations from logger_handler import log_database_operations
@dataclass def parse_employee_id_for_work_type(employee_id: str) -> Tuple[str, str]:
class CheckInRecord: """Parse employee ID to extract base ID and work type"""
"""Represents a single check-in record""" if not employee_id:
id: int return str(employee_id), 'regular'
employee_id: str
check_in_date: datetime
check_in_time: time
location_name: str
timestamp: datetime = None
def __post_init__(self): employee_id_clean = str(employee_id).strip().upper()
if self.timestamp is None:
# Combine date and time for timestamp
self.timestamp = datetime.combine(self.check_in_date, self.check_in_time)
# 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'
@dataclass # Check for PW (Periodic Work)
class WorkPeriod: pw_pattern = r'^(\d+)\s*PW$'
"""Represents a calculated work period from consecutive check-ins""" pw_match = re.match(pw_pattern, employee_id_clean)
start_record: CheckInRecord if pw_match:
end_record: Optional[CheckInRecord] return pw_match.group(1), 'PW'
duration_minutes: int = 0
is_complete: bool = False
def __post_init__(self): # Default to regular work
if self.end_record and self.start_record: return employee_id_clean, 'regular'
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: 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): 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.max_work_period_hours = max_work_period_hours
self.min_break_minutes = min_break_minutes # Minimum break between work periods 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, def calculate_employee_hours(self, employee_id: str, start_date: datetime, end_date: datetime,
attendance_records: List[Dict]) -> Dict[str, Any]: attendance_records: List[Dict]) -> Dict[str, Any]:
""" """
Calculate working hours for an employee using single check-in records Calculate working hours for an employee with SP/PW support and robust error handling
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: 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: for record in attendance_records:
if hasattr(record, '__dict__'): try:
emp_id = str(record.employee_id) # Extract record data safely
date_val = record.check_in_date if hasattr(record, '__dict__'):
time_val = record.check_in_time record_emp_id = str(getattr(record, 'employee_id', '')).strip()
location = record.location_name record_date = getattr(record, 'check_in_date', None)
record_id = record.id record_time = getattr(record, 'check_in_time', None)
else: location = getattr(record, 'location_name', 'Unknown Location')
emp_id = str(record['employee_id']) record_id = getattr(record, 'id', 0)
date_val = record['check_in_date'] else:
time_val = record['check_in_time'] record_emp_id = str(record.get('employee_id', '')).strip()
location = record['location_name'] record_date = record.get('check_in_date')
record_id = record['id'] record_time = record.get('check_in_time')
location = record.get('location_name', 'Unknown Location')
record_id = record.get('id', 0)
if emp_id == employee_id: # Skip invalid records
checkin_record = CheckInRecord( if not record_emp_id or record_date is None or record_time is None:
id=record_id, continue
employee_id=emp_id,
check_in_date=date_val,
check_in_time=time_val,
location_name=location
)
records.append(checkin_record)
if not records: # Parse work type from this record
print(f"⚠️ No records found for employee {employee_id}") record_base_id, work_type = parse_employee_id_for_work_type(record_emp_id)
return self._empty_result(employee_id, start_date, end_date)
print(f"📊 Found {len(records)} check-in records for employee {employee_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)
# Group records by date and calculate daily hours except Exception as record_error:
daily_records = {} print(f"⚠️ Error processing record: {record_error}")
for record in records: continue
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 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'])}")
# Calculate hours for each work type
daily_hours = {} daily_hours = {}
weekly_totals = [] weekly_totals = []
current_date = start_date current_date = start_date
current_week_hours = [] current_week_hours = {'regular': [], 'SP': [], 'PW': []}
while current_date <= end_date: while current_date <= end_date:
date_key = current_date.strftime('%Y-%m-%d') date_key = current_date.strftime('%Y-%m-%d')
day_records = daily_records.get(date_key, [])
# Calculate hours for this day # Calculate hours for each work type on this day
day_hours, is_miss_punch = self._calculate_daily_hours_from_checkins(day_records) 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] = { daily_hours[date_key] = {
'total_minutes': int(day_hours * 60) if day_hours > 0 else 0, 'total_minutes': int(total_day_hours * 60),
'total_hours': day_hours if day_hours > 0 else 0, 'total_hours': total_day_hours,
'is_miss_punch': is_miss_punch, 'regular_hours': hours_by_type['regular'],
'records_count': len(day_records) '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) # Check if end of week or end of period
current_week_hours.append(max(0, day_hours)) 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'])
# Check if end of week (Sunday) or end of period # Only regular hours count toward overtime
if current_date.weekday() == 6 or current_date == end_date: # Sunday or last day week_regular_hours = min(week_regular_total, 40.0)
week_total = sum(current_week_hours) week_overtime_hours = max(0, week_regular_total - 40.0)
week_regular = min(week_total, 40.0)
week_overtime = max(0, week_total - 40.0)
weekly_totals.append({ weekly_totals.append({
'total_hours': week_total, 'total_hours': week_regular_total + week_sp_total + week_pw_total,
'regular_hours': week_regular, 'regular_hours': week_regular_hours,
'overtime_hours': week_overtime, 'overtime_hours': week_overtime_hours,
'total_minutes': int(week_total * 60), 'sp_hours': week_sp_total,
'regular_minutes': int(week_regular * 60), 'pw_hours': week_pw_total,
'overtime_minutes': int(week_overtime * 60) '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) current_date += timedelta(days=1)
@@ -167,125 +177,92 @@ class SingleCheckInCalculator:
grand_total_hours = sum(week['total_hours'] for week in weekly_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_regular_hours = sum(week['regular_hours'] for week in weekly_totals)
grand_overtime_hours = sum(week['overtime_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, 'employee_id': employee_id,
'base_employee_id': base_employee_id,
'start_date': start_date.strftime('%Y-%m-%d'), 'start_date': start_date.strftime('%Y-%m-%d'),
'end_date': end_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, 'daily_hours': daily_hours,
'weekly_hours': weekly_totals, 'weekly_hours': weekly_totals,
'grand_totals': { 'grand_totals': {
'total_hours': grand_total_hours, 'total_hours': grand_total_hours,
'regular_hours': grand_regular_hours, 'regular_hours': grand_regular_hours,
'overtime_hours': grand_overtime_hours, 'overtime_hours': grand_overtime_hours,
'sp_hours': grand_sp_hours,
'pw_hours': grand_pw_hours,
'total_minutes': int(grand_total_hours * 60), 'total_minutes': int(grand_total_hours * 60),
'regular_minutes': int(grand_regular_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: except Exception as e:
print(f"❌ Error calculating working hours for employee {employee_id}: {e}") print(f"❌ Error calculating working hours for employee {employee_id}: {e}")
import traceback import traceback
print(f"❌ Traceback: {traceback.format_exc()}") print(f"❌ Traceback: {traceback.format_exc()}")
return self._empty_result(employee_id, start_date, end_date) return self._empty_result(employee_id, start_date, end_date)
def _calculate_daily_hours_from_checkins(self, day_records: List[CheckInRecord]) -> Tuple[float, bool]: def _calculate_daily_hours_from_records(self, day_records: List[Dict]) -> Tuple[float, bool]:
""" """Calculate hours for a single day from processed records"""
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)
"""
if not day_records: if not day_records:
return 0.0, False # No records = no work return 0.0, False
# Sort records by time # Sort records by timestamp
sorted_records = sorted(day_records, key=lambda r: r.timestamp) try:
print(f"📝 Processing {len(sorted_records)} records for daily calculation") 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: if len(sorted_records) == 1:
print(f"⚠️ Single record found - Miss punch (incomplete pair)") return 0.0, True
return 0.0, True # Single record is always miss punch
# CALCULATE COMPLETE PAIRS ONLY (CORRECTED FOR ODD NUMBERS) # Calculate complete pairs only
# For odd numbers: process pairs and ignore the last unpaired record
num_complete_pairs = len(sorted_records) // 2 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 total_hours = 0.0
# Process complete pairs only for i in range(0, num_complete_pairs * 2, 2):
for i in range(0, records_to_process, 2): try:
start_record = sorted_records[i] start_time = sorted_records[i]['timestamp']
end_record = sorted_records[i + 1] end_time = sorted_records[i + 1]['timestamp']
period = WorkPeriod(start_record, end_record) duration = end_time - start_time
period_hours = duration.total_seconds() / 3600.0
# Validate work period duration # Validate reasonable work period
if self._is_valid_work_period(period): if 0 < period_hours <= self.max_work_period_hours:
work_periods.append(period) total_hours += period_hours
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) except Exception as e:
has_unpaired = len(sorted_records) % 2 != 0 print(f"⚠️ Error calculating work period: {e}")
if has_unpaired: continue
unpaired_record = sorted_records[-1]
print(f"⚠️ Unpaired record found: {unpaired_record.check_in_time} (ignored in calculation)")
# Round to nearest quarter hour # 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 # Determine if miss punch
is_miss_punch = (num_complete_pairs == 0) # No valid pairs = miss punch is_miss_punch = len(sorted_records) % 2 != 0
if is_miss_punch: return total_hours, 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
def _empty_result(self, employee_id: str, start_date: datetime, end_date: datetime) -> Dict[str, Any]: 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 { return {
'employee_id': employee_id, 'employee_id': employee_id,
'base_employee_id': base_employee_id,
'start_date': start_date.strftime('%Y-%m-%d'), 'start_date': start_date.strftime('%Y-%m-%d'),
'end_date': end_date.strftime('%Y-%m-%d'), 'end_date': end_date.strftime('%Y-%m-%d'),
'include_travel_time': True, 'include_travel_time': True,
@@ -295,39 +272,60 @@ class SingleCheckInCalculator:
'total_hours': 0.0, 'total_hours': 0.0,
'regular_hours': 0.0, 'regular_hours': 0.0,
'overtime_hours': 0.0, 'overtime_hours': 0.0,
'sp_hours': 0.0,
'pw_hours': 0.0,
'total_minutes': 0, 'total_minutes': 0,
'regular_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, def calculate_all_employees_hours(self, start_date: datetime, end_date: datetime,
attendance_records: List[Dict]) -> Dict[str, Any]: 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: try:
print(f"🚀 Starting calculation for all employees") print(f"🚀 Starting calculation for all employees with SP/PW support")
# Get unique employee IDs # Get unique base employee IDs safely
employee_ids = set() base_employee_ids = set()
for record in attendance_records: for record in attendance_records:
if hasattr(record, '__dict__'): try:
employee_ids.add(str(record.employee_id)) if hasattr(record, '__dict__'):
else: employee_id = str(getattr(record, 'employee_id', '')).strip()
employee_ids.add(str(record['employee_id'])) else:
employee_id = str(record.get('employee_id', '')).strip()
print(f"👥 Found {len(employee_ids)} unique employees") 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(base_employee_ids)} unique base employees")
results = {} results = {}
for emp_id in sorted(employee_ids): for base_emp_id in sorted(base_employee_ids):
print(f"\n🔄 Processing employee {emp_id}") try:
results[emp_id] = self.calculate_employee_hours(emp_id, start_date, end_date, attendance_records) 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 { return {
'calculation_date': datetime.now().strftime('%Y-%m-%d %H:%M:%S'), 'calculation_date': datetime.now().strftime('%Y-%m-%d %H:%M:%S'),
'period_start': start_date.strftime('%Y-%m-%d'), 'period_start': start_date.strftime('%Y-%m-%d'),
'period_end': end_date.strftime('%Y-%m-%d'), 'period_end': end_date.strftime('%Y-%m-%d'),
'include_travel_time': True, # Not applicable for single check-in 'include_travel_time': True,
'employee_count': len(employee_ids), 'employee_count': len(base_employee_ids),
'employees': results 'employees': results
} }
@@ -335,4 +333,13 @@ class SingleCheckInCalculator:
print(f"❌ Error calculating hours for all employees: {e}") print(f"❌ Error calculating hours for all employees: {e}")
import traceback import traceback
print(f"❌ Traceback: {traceback.format_exc()}") print(f"❌ Traceback: {traceback.format_exc()}")
raise e
# 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': {}
}
+67
View File
@@ -141,6 +141,18 @@
box-shadow: 0 8px 32px rgba(0, 0, 0, 0.1); 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 { .summary-cards {
display: grid; display: grid;
grid-template-columns: repeat(auto-fit, minmax(200px, 1fr)); grid-template-columns: repeat(auto-fit, minmax(200px, 1fr));
@@ -709,6 +721,14 @@
<h3>{{ "%.1f"|format(working_hours_data.employees.values() | map(attribute='grand_totals.overtime_hours') | sum) }}</h3> <h3>{{ "%.1f"|format(working_hours_data.employees.values() | map(attribute='grand_totals.overtime_hours') | sum) }}</h3>
<p>Overtime Hours</p> <p>Overtime Hours</p>
</div> </div>
<div class="summary-card">
<h3>{{ "%.1f"|format(working_hours_data.employees.values() | map(attribute='grand_totals.sp_hours') | sum) }}</h3>
<p>SP Hours</p>
</div>
<div class="summary-card">
<h3>{{ "%.1f"|format(working_hours_data.employees.values() | map(attribute='grand_totals.pw_hours') | sum) }}</h3>
<p>PW Hours</p>
</div>
</div> </div>
<!-- Employee Details Table --> <!-- Employee Details Table -->
@@ -721,6 +741,8 @@
<th>Total Hours</th> <th>Total Hours</th>
<th>Regular Hours</th> <th>Regular Hours</th>
<th>Overtime Hours</th> <th>Overtime Hours</th>
<th>SP Hours</th>
<th>PW Hours</th>
<th>Working Days</th> <th>Working Days</th>
<th>Status</th> <th>Status</th>
</tr> </tr>
@@ -751,6 +773,28 @@
</span> </span>
{% endif %} {% endif %}
</td> </td>
<td>
{% if emp_data.grand_totals.sp_hours > 0 %}
<span class="hours-badge hours-sp" style="background: #e6f3ff; color: #0066cc;">
{{ "%.2f"|format(emp_data.grand_totals.sp_hours) }} hrs
</span>
{% else %}
<span class="hours-badge" style="background: #f0f0f0; color: #666;">
0.00 hrs
</span>
{% endif %}
</td>
<td>
{% if emp_data.grand_totals.pw_hours > 0 %}
<span class="hours-badge hours-pw" style="background: #fff2e6; color: #cc6600;">
{{ "%.2f"|format(emp_data.grand_totals.pw_hours) }} hrs
</span>
{% else %}
<span class="hours-badge" style="background: #f0f0f0; color: #666;">
0.00 hrs
</span>
{% endif %}
</td>
<td> <td>
{% set working_days = emp_data.daily_hours.values() | selectattr('total_hours', '>', 0) | list | length %} {% set working_days = emp_data.daily_hours.values() | selectattr('total_hours', '>', 0) | list | length %}
{{ working_days }} days {{ working_days }} days
@@ -808,6 +852,29 @@
Export Time Attendance Export Time Attendance
</button> </button>
</form> </form>
<!--
<form method="POST" action="{{ url_for('export_payroll_excel') }}" style="display: inline;">
<input type="hidden" name="date_from" value="{{ date_from }}">
<input type="hidden" name="date_to" value="{{ date_to }}">
<input type="hidden" name="project_filter" value="{{ project_filter }}">
<input type="hidden" name="report_type" value="enhanced">
<button type="submit" class="btn btn-primary">
<i class="fas fa-file-excel"></i>
Export Enhanced Payroll (SP/PW)
</button>
</form>
<form method="POST" action="{{ url_for('export_payroll_excel') }}" style="display: inline;">
<input type="hidden" name="date_from" value="{{ date_from }}">
<input type="hidden" name="date_to" value="{{ date_to }}">
<input type="hidden" name="project_filter" value="{{ project_filter }}">
<input type="hidden" name="report_type" value="detailed_sp_pw">
<button type="submit" class="btn btn-info">
<i class="fas fa-list-alt"></i>
Export Daily SP/PW Breakdown
</button>
</form>
-->
</div> </div>
<div style="margin-top: 1rem; padding: 1rem; background: #f7fafc; border-radius: 8px; color: #4a5568; font-size: 0.9rem;"> <div style="margin-top: 1rem; padding: 1rem; background: #f7fafc; border-radius: 8px; color: #4a5568; font-size: 0.9rem;">