Fix bugs payroll export
This commit is contained in:
Nguyen Ngo
@@ -4358,20 +4358,31 @@ def payroll_dashboard():
|
||||
employee_names = {}
|
||||
if working_hours_data:
|
||||
try:
|
||||
# Try to get employee names from your employee table if it exists
|
||||
# Use the same SQL approach as attendance report - JOIN with CAST
|
||||
employee_ids = list(working_hours_data['employees'].keys())
|
||||
# This assumes you have an employee table - modify as needed
|
||||
employee_query = db.session.execute(text("""
|
||||
SELECT id, CONCAT(firstName, ' ', lastName) as full_name
|
||||
FROM employee
|
||||
WHERE id IN :employee_ids
|
||||
"""), {'employee_ids': tuple(employee_ids)})
|
||||
if employee_ids:
|
||||
# Build a query similar to attendance report
|
||||
placeholders = ','.join([f"'{emp_id}'" for emp_id in employee_ids])
|
||||
employee_query = db.session.execute(text(f"""
|
||||
SELECT
|
||||
ad.employee_id,
|
||||
CONCAT(e.firstName, ' ', e.lastName) as full_name
|
||||
FROM attendance_data ad
|
||||
LEFT JOIN employee e ON CAST(ad.employee_id AS UNSIGNED) = e.id
|
||||
WHERE ad.employee_id IN ({placeholders})
|
||||
GROUP BY ad.employee_id, e.firstName, e.lastName
|
||||
"""))
|
||||
|
||||
for row in employee_query:
|
||||
employee_names[str(row[0])] = row[1]
|
||||
for row in employee_query:
|
||||
if row[1]: # Only add if we got a name
|
||||
employee_names[str(row[0])] = row[1]
|
||||
|
||||
print(f"📊 Retrieved names for {len(employee_names)} employees using CAST method")
|
||||
|
||||
except Exception as e:
|
||||
print(f"⚠️ Could not load employee names: {e}")
|
||||
import traceback
|
||||
print(f"⚠️ Traceback: {traceback.format_exc()}")
|
||||
# Continue without names - will use employee IDs
|
||||
|
||||
# Get selected project name for display
|
||||
@@ -4467,39 +4478,41 @@ def export_payroll_excel():
|
||||
|
||||
print(f"📊 Exporting {len(attendance_records)} attendance records to Excel")
|
||||
|
||||
# Get employee names
|
||||
employee_names = {}
|
||||
try:
|
||||
employee_ids = list(set(str(record.employee_id) for record in attendance_records))
|
||||
employee_query = db.session.execute(text("""
|
||||
SELECT id, CONCAT(firstName, ' ', lastName) as full_name
|
||||
FROM employee
|
||||
WHERE id IN :employee_ids
|
||||
"""), {'employee_ids': tuple(employee_ids)})
|
||||
|
||||
for row in employee_query:
|
||||
employee_names[str(row[0])] = row[1]
|
||||
|
||||
except Exception as e:
|
||||
print(f"⚠️ Could not load employee names for export: {e}")
|
||||
|
||||
# Get project name for filename
|
||||
project_name = ''
|
||||
if project_filter:
|
||||
try:
|
||||
project = Project.query.get(int(project_filter))
|
||||
if project:
|
||||
project_name = f"_{project.name.replace(' ', '_')}"
|
||||
except Exception as e:
|
||||
print(f"⚠️ Error getting project name for filename: {e}")
|
||||
|
||||
# Create Excel exporter
|
||||
exporter = PayrollExcelExporter(
|
||||
company_name=os.environ.get('COMPANY_NAME', 'Your Company'),
|
||||
contract_name=os.environ.get('CONTRACT_NAME', 'Default Contract')
|
||||
)
|
||||
|
||||
# Generate Excel file
|
||||
# Get employee names using the same method as dashboard
|
||||
employee_names = {}
|
||||
try:
|
||||
employee_ids = list(set(str(record.employee_id) for record in attendance_records))
|
||||
if employee_ids:
|
||||
# Use the same SQL approach as attendance report - JOIN with CAST
|
||||
placeholders = ','.join([f"'{emp_id}'" for emp_id in employee_ids])
|
||||
employee_query = db.session.execute(text(f"""
|
||||
SELECT
|
||||
ad.employee_id,
|
||||
CONCAT(e.firstName, ' ', e.lastName) as full_name
|
||||
FROM attendance_data ad
|
||||
LEFT JOIN employee e ON CAST(ad.employee_id AS UNSIGNED) = e.id
|
||||
WHERE ad.employee_id IN ({placeholders})
|
||||
GROUP BY ad.employee_id, e.firstName, e.lastName
|
||||
"""))
|
||||
|
||||
for row in employee_query:
|
||||
if row[1]: # Only add if we got a name
|
||||
employee_names[str(row[0])] = row[1]
|
||||
|
||||
print(f"📊 Retrieved names for {len(employee_names)} employees for export using CAST method")
|
||||
|
||||
except Exception as e:
|
||||
print(f"⚠️ Could not load employee names for export: {e}")
|
||||
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, include_travel_time
|
||||
@@ -4511,6 +4524,16 @@ def export_payroll_excel():
|
||||
)
|
||||
filename_prefix = 'payroll_report'
|
||||
|
||||
# Get project name for filename
|
||||
project_name = ''
|
||||
if project_filter:
|
||||
try:
|
||||
project = Project.query.get(int(project_filter))
|
||||
if project:
|
||||
project_name = f"_{project.name.replace(' ', '_')}"
|
||||
except Exception as e:
|
||||
print(f"⚠️ Error getting project name for filename: {e}")
|
||||
|
||||
if excel_file:
|
||||
# Generate filename with timestamp and project name
|
||||
timestamp = datetime.now().strftime('%Y%m%d_%H%M%S')
|
||||
@@ -4571,7 +4594,6 @@ def calculate_working_hours_api():
|
||||
employee_id = data.get('employee_id')
|
||||
date_from = data.get('date_from')
|
||||
date_to = data.get('date_to')
|
||||
include_travel_time = data.get('include_travel_time', True)
|
||||
|
||||
if not all([employee_id, date_from, date_to]):
|
||||
return jsonify({
|
||||
@@ -4597,7 +4619,7 @@ def calculate_working_hours_api():
|
||||
|
||||
attendance_records = query.all()
|
||||
|
||||
# Calculate working hours
|
||||
# Calculate working hours using single check-in calculator
|
||||
calculator = SingleCheckInCalculator()
|
||||
hours_data = calculator.calculate_employee_hours(
|
||||
str(employee_id), start_date, end_date, attendance_records
|
||||
|
||||
@@ -105,6 +105,59 @@ class PayrollExcelExporter:
|
||||
for attr, value in style_dict.items():
|
||||
setattr(cell, attr, value)
|
||||
|
||||
def _get_employee_names(self, attendance_records: List[Dict]) -> Dict[str, str]:
|
||||
"""Get employee names - fallback method if names not provided"""
|
||||
print("⚠️ Excel exporter falling back to internal name lookup - this should not normally happen")
|
||||
return {} # Return empty dict - names should be provided by the route
|
||||
|
||||
def _get_employee_names(self, attendance_records: List[Dict]) -> Dict[str, str]:
|
||||
"""Get employee names using the same CAST method as attendance report"""
|
||||
employee_names = {}
|
||||
try:
|
||||
from flask import current_app
|
||||
from sqlalchemy import text
|
||||
|
||||
# Get unique employee IDs from attendance records
|
||||
employee_ids = []
|
||||
for record in attendance_records:
|
||||
if hasattr(record, '__dict__'):
|
||||
emp_id = str(record.employee_id)
|
||||
else:
|
||||
emp_id = str(record['employee_id'])
|
||||
|
||||
if emp_id not in employee_ids:
|
||||
employee_ids.append(emp_id)
|
||||
|
||||
if employee_ids:
|
||||
# Use the same SQL approach as attendance report - JOIN with CAST
|
||||
placeholders = ','.join([f"'{emp_id}'" for emp_id in employee_ids])
|
||||
|
||||
# Import db from current app context
|
||||
from app import db
|
||||
|
||||
employee_query = db.session.execute(text(f"""
|
||||
SELECT
|
||||
ad.employee_id,
|
||||
CONCAT(e.firstName, ' ', e.lastName) as full_name
|
||||
FROM attendance_data ad
|
||||
LEFT JOIN employee e ON CAST(ad.employee_id AS UNSIGNED) = e.id
|
||||
WHERE ad.employee_id IN ({placeholders})
|
||||
GROUP BY ad.employee_id, e.firstName, e.lastName
|
||||
"""))
|
||||
|
||||
for row in employee_query:
|
||||
if row[1]: # Only add if we got a name
|
||||
employee_names[str(row[0])] = row[1]
|
||||
|
||||
print(f"📊 Excel exporter retrieved names for {len(employee_names)} employees using CAST method")
|
||||
|
||||
except Exception as e:
|
||||
print(f"⚠️ Excel exporter could not load employee names: {e}")
|
||||
import traceback
|
||||
print(f"⚠️ Traceback: {traceback.format_exc()}")
|
||||
|
||||
return employee_names
|
||||
|
||||
@log_database_operations('payroll_excel_export')
|
||||
def create_payroll_report(self, start_date: datetime, end_date: datetime,
|
||||
attendance_records: List[Dict], employee_names: Dict[str, str] = None,
|
||||
@@ -125,6 +178,10 @@ class PayrollExcelExporter:
|
||||
try:
|
||||
print(f"📊 Creating payroll report from {start_date.strftime('%Y-%m-%d')} to {end_date.strftime('%Y-%m-%d')}")
|
||||
|
||||
# Get employee names using the same method as payroll routes
|
||||
if employee_names is None:
|
||||
employee_names = self._get_employee_names(attendance_records)
|
||||
|
||||
# Calculate working hours
|
||||
calculator = SingleCheckInCalculator()
|
||||
hours_data = calculator.calculate_all_employees_hours(start_date, end_date, attendance_records)
|
||||
@@ -360,6 +417,10 @@ class PayrollExcelExporter:
|
||||
try:
|
||||
print(f"📊 Creating detailed hours report from {start_date.strftime('%Y-%m-%d')} to {end_date.strftime('%Y-%m-%d')}")
|
||||
|
||||
# Get employee names using the same method as payroll routes
|
||||
if employee_names is None:
|
||||
employee_names = self._get_employee_names(attendance_records)
|
||||
|
||||
# Calculate working hours
|
||||
calculator = SingleCheckInCalculator()
|
||||
hours_data = calculator.calculate_all_employees_hours(start_date, end_date, attendance_records)
|
||||
|
||||
@@ -198,52 +198,49 @@ class SingleCheckInCalculator:
|
||||
Calculate hours for a single day from check-in records
|
||||
|
||||
Logic:
|
||||
- Pair consecutive check-ins as work periods
|
||||
- 1st = start, 2nd = end, 3rd = start new period, 4th = end, etc.
|
||||
- Single record = miss punch
|
||||
- Validate reasonable work periods
|
||||
- Must have even number of records (complete pairs)
|
||||
- 1st = check-in, 2nd = check-out, 3rd = check-in, 4th = check-out, etc.
|
||||
- Any odd number of records = miss punch
|
||||
- Any invalid work period = miss punch
|
||||
|
||||
Returns: (hours, is_miss_punch)
|
||||
"""
|
||||
if not day_records:
|
||||
return 0.0, False
|
||||
|
||||
if len(day_records) == 1:
|
||||
print(f"⚠️ Single check-in found - miss punch")
|
||||
return 0.0, True # Single check-in = miss punch
|
||||
# Must have even number of records for complete pairs
|
||||
if len(day_records) % 2 != 0:
|
||||
print(f"⚠️ Odd number of records ({len(day_records)}) - miss punch (incomplete pairs)")
|
||||
return 0.0, True # Odd number = miss punch
|
||||
|
||||
# Sort records by time
|
||||
sorted_records = sorted(day_records, key=lambda r: r.timestamp)
|
||||
print(f"📝 Processing {len(sorted_records)} records for the day")
|
||||
print(f"📝 Processing {len(sorted_records)} records for the day (must be complete pairs)")
|
||||
|
||||
# Create work periods from consecutive check-ins
|
||||
# Create work periods from consecutive check-ins (must be pairs)
|
||||
work_periods = []
|
||||
for i in range(0, len(sorted_records) - 1, 2):
|
||||
for i in range(0, len(sorted_records), 2):
|
||||
start_record = sorted_records[i]
|
||||
end_record = sorted_records[i + 1] if i + 1 < len(sorted_records) else None
|
||||
end_record = sorted_records[i + 1] # We know this exists because we checked even count
|
||||
|
||||
if end_record:
|
||||
period = WorkPeriod(start_record, end_record)
|
||||
period = WorkPeriod(start_record, end_record)
|
||||
|
||||
# Validate work period duration
|
||||
if self._is_valid_work_period(period):
|
||||
work_periods.append(period)
|
||||
print(f"✅ Valid work period: {start_record.check_in_time} - {end_record.check_in_time} = {period.duration_minutes/60:.2f} hours")
|
||||
else:
|
||||
print(f"⚠️ Invalid work period: {period.duration_minutes/60:.2f} hours (too long or negative)")
|
||||
return 0.0, True # Invalid period = miss punch
|
||||
# Validate work period duration
|
||||
if self._is_valid_work_period(period):
|
||||
work_periods.append(period)
|
||||
print(f"✅ Valid work period: {start_record.check_in_time} - {end_record.check_in_time} = {period.duration_minutes/60:.2f} hours")
|
||||
else:
|
||||
print(f"⚠️ Unpaired check-in at {start_record.check_in_time}")
|
||||
return 0.0, True # Unpaired record = miss punch
|
||||
print(f"⚠️ Invalid work period: {period.duration_minutes/60:.2f} hours - miss punch (invalid duration)")
|
||||
return 0.0, True # Invalid period = miss punch
|
||||
|
||||
# Calculate total hours
|
||||
# If we got here, all periods are valid
|
||||
total_minutes = sum(period.duration_minutes for period in work_periods)
|
||||
total_hours = total_minutes / 60.0
|
||||
|
||||
# Round to nearest quarter hour
|
||||
rounded_hours = round(total_hours * 4) / 4
|
||||
|
||||
print(f"📊 Daily total: {rounded_hours:.2f} hours from {len(work_periods)} work periods")
|
||||
print(f"📊 Daily total: {rounded_hours:.2f} hours from {len(work_periods)} complete work periods")
|
||||
|
||||
return rounded_hours, False
|
||||
|
||||
|
||||
Reference in New Issue
Block a user