479 lines
21 KiB
Python
479 lines
21 KiB
Python
#!/usr/bin/env python3
|
|
"""
|
|
Safe Enhanced Single Check-in Working Hours Calculator with SP/PW Support
|
|
========================================================================
|
|
|
|
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
|
|
|
|
|
|
def parse_employee_id_for_work_type(employee_id: str) -> Tuple[str, str]:
|
|
"""Parse employee ID to extract base ID and work type (supports SP, PW, PT)"""
|
|
if not employee_id:
|
|
return str(employee_id), 'regular'
|
|
|
|
employee_id_clean = str(employee_id).strip().upper()
|
|
|
|
# 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'
|
|
|
|
# Check for PT (Part-Time)
|
|
pt_pattern = r'^(\d+)\s*PT$'
|
|
pt_match = re.match(pt_pattern, employee_id_clean)
|
|
if pt_match:
|
|
return pt_match.group(1), 'PT'
|
|
|
|
# Default to regular work
|
|
return employee_id_clean, 'regular'
|
|
|
|
def round_time_to_quarter_hour(minutes: float) -> float:
|
|
"""
|
|
Round time to nearest quarter hour based on 7.5-minute increments
|
|
|
|
Rules:
|
|
- 0:00-0:07 → 0:00
|
|
- 0:08-0:22 → 0:15
|
|
- 0:23-0:37 → 0:30
|
|
- 0:38-0:52 → 0:45
|
|
- 0:53-1:07 → 1:00
|
|
"""
|
|
if minutes < 0:
|
|
return 0.0
|
|
|
|
# Get minutes within the hour
|
|
total_minutes = minutes
|
|
hours = int(total_minutes // 60)
|
|
minutes_in_hour = total_minutes % 60
|
|
|
|
# Determine rounding
|
|
if minutes_in_hour <= 7:
|
|
rounded_in_hour = 0
|
|
elif minutes_in_hour <= 22:
|
|
rounded_in_hour = 15
|
|
elif minutes_in_hour <= 37:
|
|
rounded_in_hour = 30
|
|
elif minutes_in_hour <= 52:
|
|
rounded_in_hour = 45
|
|
else: # 53-59
|
|
hours += 1
|
|
rounded_in_hour = 0
|
|
|
|
return hours * 60 + rounded_in_hour
|
|
|
|
|
|
def convert_minutes_to_base100(minutes: float) -> float:
|
|
"""Convert minutes to base-100 hours (each hour = 100 units)"""
|
|
if minutes < 0:
|
|
return 0.0
|
|
|
|
decimal_hours = minutes / 60.0
|
|
whole_hours = int(decimal_hours)
|
|
fractional_hours = decimal_hours - whole_hours
|
|
base100_fraction = fractional_hours * 100
|
|
|
|
return whole_hours + (base100_fraction / 100)
|
|
|
|
|
|
def round_base100_hours(base100_hours: float) -> float:
|
|
"""
|
|
Round base-100 hours to nearest quarter using 12.5 thresholds
|
|
|
|
Examples:
|
|
- 4.12 → 4.00
|
|
- 4.18 → 4.25
|
|
- 8.02 → 8.00
|
|
"""
|
|
if base100_hours < 0:
|
|
return 0.0
|
|
|
|
whole_hours = int(base100_hours)
|
|
fractional_part = (base100_hours - whole_hours) * 100
|
|
|
|
# Round to nearest quarter
|
|
if fractional_part < 12.5:
|
|
rounded_fraction = 0
|
|
elif fractional_part < 37.5:
|
|
rounded_fraction = 25
|
|
elif fractional_part < 62.5:
|
|
rounded_fraction = 50
|
|
elif fractional_part < 87.5:
|
|
rounded_fraction = 75
|
|
else:
|
|
whole_hours += 1
|
|
rounded_fraction = 0
|
|
|
|
return round(whole_hours + (rounded_fraction / 100), 2)
|
|
|
|
class SingleCheckInCalculator:
|
|
"""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
|
|
self.min_break_minutes = min_break_minutes
|
|
|
|
@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 with SP/PW support and robust error handling
|
|
"""
|
|
try:
|
|
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': [], 'PT': []}
|
|
|
|
for record in attendance_records:
|
|
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,
|
|
'action_description': getattr(record, 'action_description', '') if hasattr(record, '__dict__') else record.get('action_description', ''),
|
|
'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
|
|
|
|
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 = {}
|
|
weekly_totals = []
|
|
current_date = start_date
|
|
current_week_hours = {'regular': [], 'SP': [], 'PW': [], 'PT': []}
|
|
|
|
while current_date <= end_date:
|
|
date_key = current_date.strftime('%Y-%m-%d')
|
|
|
|
# Calculate hours for each work type on this day
|
|
hours_by_type = {}
|
|
is_miss_punch_by_type = {}
|
|
|
|
for work_type in ['regular', 'SP', 'PW', 'PT']:
|
|
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/PT support
|
|
total_day_hours = sum(hours_by_type.values())
|
|
daily_hours[date_key] = {
|
|
'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'],
|
|
'pt_hours': hours_by_type['PT'],
|
|
'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', 'PT']),
|
|
'miss_punch_details': {
|
|
'regular': is_miss_punch_by_type['regular'],
|
|
'SP': is_miss_punch_by_type['SP'],
|
|
'PW': is_miss_punch_by_type['PW'],
|
|
'PT': is_miss_punch_by_type['PT']
|
|
}
|
|
}
|
|
|
|
# Add to weekly calculation
|
|
for work_type in ['regular', 'SP', 'PW', 'PT']:
|
|
current_week_hours[work_type].append(max(0, hours_by_type[work_type]))
|
|
|
|
# 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'])
|
|
week_pt_total = sum(current_week_hours['PT'])
|
|
|
|
# Round the weekly total
|
|
week_total_raw = week_regular_total + week_sp_total + week_pw_total + week_pt_total
|
|
week_total_rounded = round_base100_hours(week_total_raw)
|
|
|
|
# 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)
|
|
|
|
# Round regular, overtime, and special hours
|
|
week_regular_hours_rounded = round_base100_hours(week_regular_hours)
|
|
week_overtime_hours_rounded = round_base100_hours(week_overtime_hours)
|
|
week_sp_hours_rounded = round_base100_hours(week_sp_total)
|
|
week_pw_hours_rounded = round_base100_hours(week_pw_total)
|
|
week_pt_hours_rounded = round_base100_hours(week_pt_total)
|
|
|
|
weekly_totals.append({
|
|
'total_hours': week_total_rounded,
|
|
'regular_hours': week_regular_hours_rounded,
|
|
'overtime_hours': week_overtime_hours_rounded,
|
|
'sp_hours': week_sp_hours_rounded,
|
|
'pw_hours': week_pw_hours_rounded,
|
|
'pt_hours': week_pt_hours_rounded,
|
|
'total_minutes': int(week_total_rounded * 60),
|
|
'regular_minutes': int(week_regular_hours_rounded * 60),
|
|
'overtime_minutes': int(week_overtime_hours_rounded * 60),
|
|
'sp_minutes': int(week_sp_hours_rounded * 60),
|
|
'pw_minutes': int(week_pw_hours_rounded * 60),
|
|
'pt_minutes': int(week_pt_hours_rounded * 60)
|
|
})
|
|
|
|
current_week_hours = {'regular': [], 'SP': [], 'PW': [], 'PT': []}
|
|
|
|
current_date += timedelta(days=1)
|
|
|
|
# Calculate grand totals
|
|
grand_total_raw = sum(week['total_hours'] for week in weekly_totals)
|
|
grand_regular_raw = sum(week['regular_hours'] for week in weekly_totals)
|
|
grand_overtime_raw = sum(week['overtime_hours'] for week in weekly_totals)
|
|
grand_sp_raw = sum(week['sp_hours'] for week in weekly_totals)
|
|
grand_pw_raw = sum(week['pw_hours'] for week in weekly_totals)
|
|
grand_pt_raw = sum(week.get('pt_hours', 0) for week in weekly_totals)
|
|
|
|
# Round grand totals
|
|
grand_total_hours = round_base100_hours(grand_total_raw)
|
|
grand_regular_hours = round_base100_hours(grand_regular_raw)
|
|
grand_overtime_hours = round_base100_hours(grand_overtime_raw)
|
|
grand_sp_hours = round_base100_hours(grand_sp_raw)
|
|
grand_pw_hours = round_base100_hours(grand_pw_raw)
|
|
grand_pt_hours = round_base100_hours(grand_pt_raw)
|
|
|
|
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, PT: {grand_pt_hours:.2f}h)")
|
|
|
|
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,
|
|
'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,
|
|
'pt_hours': grand_pt_hours,
|
|
'total_minutes': int(grand_total_hours * 60),
|
|
'regular_minutes': int(grand_regular_hours * 60),
|
|
'overtime_minutes': int(grand_overtime_hours * 60),
|
|
'sp_minutes': int(grand_sp_hours * 60),
|
|
'pw_minutes': int(grand_pw_hours * 60),
|
|
'pt_minutes': int(grand_pt_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_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
|
|
|
|
# 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
|
|
if len(sorted_records) == 1:
|
|
return 0.0, True
|
|
|
|
# NEW: Check if all records are IN or all OUT using action_description
|
|
if len(sorted_records) >= 2:
|
|
record_types = []
|
|
for record in sorted_records:
|
|
action_desc = record.get('action_description', '').lower()
|
|
if 'out' in action_desc or 'checkout' in action_desc:
|
|
record_types.append('OUT')
|
|
else:
|
|
record_types.append('IN')
|
|
|
|
# If all are IN or all are OUT, return 0 hours with miss punch
|
|
if len(set(record_types)) == 1:
|
|
print(f"⚠️ All {len(sorted_records)} records are {record_types[0]} - 0 working hours")
|
|
return 0.0, True
|
|
|
|
# Calculate complete pairs only
|
|
num_complete_pairs = len(sorted_records) // 2
|
|
total_hours = 0.0
|
|
total_rounded_minutes = 0.0 # NEW: Track rounded minutes for Daily Total
|
|
|
|
for i in range(0, num_complete_pairs * 2, 2):
|
|
try:
|
|
start_time = sorted_records[i]['timestamp']
|
|
end_time = sorted_records[i + 1]['timestamp']
|
|
|
|
# Calculate exact hours (for Hours/Building column)
|
|
duration = end_time - start_time
|
|
exact_minutes = duration.total_seconds() / 60.0
|
|
hours = exact_minutes / 60.0
|
|
total_hours += hours
|
|
|
|
# NEW: Round this pair's time and accumulate for Daily Total
|
|
pair_rounded_minutes = round_time_to_quarter_hour(exact_minutes)
|
|
total_rounded_minutes += pair_rounded_minutes
|
|
|
|
except Exception as e:
|
|
print(f"⚠️ Error calculating pair hours: {e}")
|
|
return 0.0, True
|
|
|
|
# NEW: Convert rounded minutes to base-100 and round
|
|
base100_hours = convert_minutes_to_base100(total_rounded_minutes)
|
|
rounded_total_hours = round_base100_hours(base100_hours)
|
|
|
|
# Odd number of records = miss punch
|
|
is_miss_punch = (len(sorted_records) % 2 != 0)
|
|
|
|
# NEW: Return rounded hours instead of exact hours for Daily Total
|
|
return rounded_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 with SP/PW/PT 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,
|
|
'daily_hours': {},
|
|
'weekly_hours': [],
|
|
'grand_totals': {
|
|
'total_hours': 0.0,
|
|
'regular_hours': 0.0,
|
|
'overtime_hours': 0.0,
|
|
'sp_hours': 0.0,
|
|
'pw_hours': 0.0,
|
|
'pt_hours': 0.0,
|
|
'total_minutes': 0,
|
|
'regular_minutes': 0,
|
|
'overtime_minutes': 0,
|
|
'sp_minutes': 0,
|
|
'pw_minutes': 0,
|
|
'pt_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 with robust error handling"""
|
|
try:
|
|
print(f"🚀 Starting calculation for all employees with SP/PW support")
|
|
|
|
# Get unique base employee IDs safely
|
|
base_employee_ids = set()
|
|
for record in attendance_records:
|
|
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(base_employee_ids)} unique base employees")
|
|
|
|
results = {}
|
|
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,
|
|
'employee_count': len(base_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()}")
|
|
|
|
# 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': {}
|
|
} |