nngo/GOV_QR_Codes_Management
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity
Files
023b2fdf4685145cc041e65c4ac26b4dd71d28d5
GOV_QR_Codes_Management/routes/time_attendance_export.py
T

2025 lines
103 KiB
Python
Raw Blame History

This file contains ambiguous Unicode characters
This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.
"""
routes/time_attendance_export.py
=================================
Excel export logic for Time Attendance — helper functions called by
routes in time_attendance.py.
Contains:
- calculate_possible_violation()
- _overnight_aware_sort_key()
- _qtr()
- export_time_attendance_excel() (single-employee / all-employees)
- export_time_attendance_by_building_excel()
"""
from flask import send_file, g, current_app
from datetime import datetime, date, timedelta, time
import io, os, json, re
import time as _time
from extensions import db, logger_handler
from models.employee import Employee
from models.project import Project
from models.qrcode import QRCode
from models.time_attendance import TimeAttendance
from sqlalchemy import text
from working_hours_calculator import WorkingHoursCalculator, round_time_to_quarter_hour, convert_minutes_to_base100, round_base100_hours
import openpyxl
from openpyxl.styles import Font, PatternFill, Alignment, Border, Side, numbers
from openpyxl.utils import get_column_letter
import openpyxl.cell.cell
# Export helpers — no Blueprint needed, these are plain functions
# called from routes in time_attendance.py
def calculate_possible_violation(distance_value):
"""
Calculate possible violation status based on distance
Args:
distance_value: Distance in miles (float or None)
Returns:
'Yes' if distance > 0.3, 'No' otherwise
"""
if distance_value is None:
return 'No'
try:
distance_float = float(distance_value)
return 'Yes' if distance_float > 0.3 else 'No'
except (ValueError, TypeError):
return 'No'
def _overnight_aware_sort_key(record):
"""
Sort key for attendance records within a single calendar-date bucket.
Problem 1: when an overnight shift spans midnight, the check-out record's
check_in_time (e.g. 00:01 AM) sorts numerically BEFORE the check-in time
(e.g. 20:00 PM), producing an orphaned OUT followed by an orphaned IN.
Fix: push early-morning check-outs (hour <= 3) past midnight by adding
24 h worth of seconds so they sort after same-day evening check-ins.
Problem 2: two records in the same minute (e.g. IN 06:22:04, OUT 06:22:52)
had identical sort keys because seconds were not included, leaving the
database-delivery order intact (DESC → OUT first). The pairing loop then
encountered the OUT before the IN, emitting an orphaned-OUT row followed
by an orphaned-IN row — reversed from chronological order.
Fix: include seconds in the key so true chronological order is preserved.
"""
from datetime import time as _time
t = record.check_in_time
if isinstance(t, _time):
# Use fractional minutes (hours*60 + minutes + seconds/60) so that
# records sharing the same HH:MM still sort by their seconds component.
seconds_total = t.hour * 3600 + t.minute * 60 + t.second
else:
seconds_total = 0
action = (record.action_description or '').lower()
is_out = 'out' in action or 'checkout' in action
# Push early-morning check-outs past midnight to end of day order.
# Use seconds-based offset (24 h = 86400 s) to remain consistent with
# the seconds-granularity key above.
if is_out and t.hour <= 3:
seconds_total += 24 * 3600
return seconds_total
def _qtr(decimal_hours: float) -> float:
"""
Round a decimal-hours value to the nearest quarter hour (.00/.25/.50/.75).
Pipeline: decimal hours → minutes → quarter-hour rounding → base-100 → quarter rounding.
Examples: 4.03 → 4.0, 4.08 → 4.25, 3.87 → 4.0, 4.16 → 4.25
Returns 0.0 for negative or zero input.
"""
if decimal_hours <= 0:
return 0.0
minutes = decimal_hours * 60.0
rounded_minutes = round_time_to_quarter_hour(minutes)
base100 = convert_minutes_to_base100(rounded_minutes)
return round_base100_hours(base100)
# ---------------------------------------------------------------------------
# Private export helpers — shared by both export functions below.
# ---------------------------------------------------------------------------
def _resolve_date_range(start_date_filter, end_date_filter, records, export_label='TA Excel export', unlimited=False):
"""
Resolve and validate the export date range.
Returns (start_date, end_date, filtered_records) where:
- start_date / end_date are date objects
- filtered_records is the input list capped to end_date + 1 day (overnight buffer)
- end_date is capped to a maximum 14-day window unless unlimited=True
Returns None when there are no records and no date filters.
"""
MAX_EXPORT_DAYS = 14
if start_date_filter and end_date_filter:
if isinstance(start_date_filter, str):
start_date = datetime.strptime(start_date_filter, '%Y-%m-%d').date()
else:
start_date = start_date_filter
if isinstance(end_date_filter, str):
end_date = datetime.strptime(end_date_filter, '%Y-%m-%d').date()
else:
end_date = end_date_filter
elif records:
start_date = min(r.attendance_date for r in records)
end_date = max(r.attendance_date for r in records)
else:
return None
if not unlimited and (end_date - start_date).days >= MAX_EXPORT_DAYS:
capped_end_date = start_date + timedelta(days=MAX_EXPORT_DAYS - 1)
logger_handler.logger.info(
f"{export_label}: date range [{start_date} \u2013 {end_date}] exceeds "
f"{MAX_EXPORT_DAYS} days; capping end_date to {capped_end_date}."
)
end_date = capped_end_date
# Apply both bounds:
# - Lower bound: exclude records before start_date (prevents historical data leaking in).
# - Upper bound (+1 day buffer): early-morning check-out records stored on Day N+1
# must remain available for overnight pairing detection; the display range is
# still controlled by each export's sorted_dates (capped to end_date).
filtered_records = [
r for r in records
if start_date <= r.attendance_date <= end_date + timedelta(days=1)
]
return start_date, end_date, filtered_records
def _convert_ta_records(records):
"""
Convert TimeAttendance ORM records to the lightweight anonymous-class
format expected by WorkingHoursCalculator and the Excel rendering loops.
Returns a list of converted record objects.
"""
from working_hours_calculator import parse_employee_id_for_work_type
converted = []
for record in records:
distance_value = getattr(record, 'distance', None)
record_type = 'check_in'
if hasattr(record, 'action_description') and record.action_description:
action_lower = record.action_description.lower()
if 'out' in action_lower or 'checkout' in action_lower:
record_type = 'check_out'
_, work_type = parse_employee_id_for_work_type(str(record.employee_id))
base_location_name = record.location_name
if work_type and work_type in ('PT', 'SP', 'PW'):
display_location_name = f"{base_location_name} ({work_type})"
else:
display_location_name = base_location_name
converted_record = type('Record', (), {
'id': record.id,
'employee_id': str(record.employee_id),
'employee_name': getattr(record, 'employee_name', ''),
'check_in_date': record.attendance_date,
'check_in_time': record.attendance_time,
'location_name': display_location_name,
'original_location_name': base_location_name,
'work_type': work_type,
'latitude': None,
'longitude': None,
'distance': distance_value,
'record_type': record_type,
'action_description': record.action_description,
'event_description': record.event_description or '',
'recorded_address': record.recorded_address or '',
'qr_code': type('QRCode', (), {
'location': base_location_name,
'location_address': record.recorded_address or '',
'project': None
})()
})()
converted.append(converted_record)
return converted
def _build_employee_name_map(records):
"""
Build a {base_employee_id: "Lastname, Firstname"} map for export headers.
Looks up the Employee table by numeric base ID so work-type suffixes
(e.g. '3937SP') in the stored employee_name column do not pollute labels.
Falls back to the stored employee_name on lookup failure.
"""
from working_hours_calculator import parse_employee_id_for_work_type
employee_names = {}
for record in records:
base_id, _ = parse_employee_id_for_work_type(str(record.employee_id))
if base_id not in employee_names:
try:
emp = Employee.query.filter_by(id=int(base_id)).first()
if emp:
employee_names[base_id] = f"{emp.lastName}, {emp.firstName}"
else:
employee_names[base_id] = getattr(record, 'employee_name', f'Employee {base_id}')
logger_handler.logger.warning(
f"Employee ID {base_id} not found in employee table during export; "
f"using stored name."
)
except Exception as e:
employee_names[base_id] = getattr(record, 'employee_name', f'Employee {base_id}')
logger_handler.logger.warning(
f"Could not lookup employee name for ID {base_id} during export: {e}"
)
return employee_names
def _make_export_styles():
"""
Return a dict of openpyxl style objects shared by both export functions.
Keys: header_font, header_fill, data_font, bold_font, italic_bold_font,
border, missed_punch_fill, border_day_middle, border_day_last,
border_day_single, border_day_first, amber_fill
"""
header_font = Font(name='Aptos Narrow', size=11, bold=True, color='FFFFFF')
header_fill = PatternFill(start_color='000000', end_color='000000', fill_type='solid')
data_font = Font(name='Aptos Narrow', size=11)
bold_font = Font(name='Aptos Narrow', size=11, bold=True)
italic_bold_font = Font(name='Aptos Narrow', size=11, bold=True, italic=True)
border = Border(
left=Side(style='thin'),
right=Side(style='thin'),
top=Side(style='thin'),
bottom=Side(style='thin')
)
missed_punch_fill = PatternFill(start_color='FFC000', end_color='FFC000', fill_type='solid')
amber_fill = PatternFill(start_color='FFC000', end_color='FFC000', fill_type='solid')
border_day_middle = Border()
border_day_last = Border(bottom=Side(style='thin'))
border_day_single = Border(bottom=Side(style='thin'))
border_day_first = Border()
return {
'header_font': header_font,
'header_fill': header_fill,
'data_font': data_font,
'bold_font': bold_font,
'italic_bold_font': italic_bold_font,
'border': border,
'missed_punch_fill': missed_punch_fill,
'amber_fill': amber_fill,
'border_day_middle': border_day_middle,
'border_day_last': border_day_last,
'border_day_single': border_day_single,
'border_day_first': border_day_first,
}
def export_time_attendance_excel(records, project_name_for_filename, date_range_str, filter_str, start_date_filter=None, end_date_filter=None, unlimited=False):
"""Generate Excel export with template format matching the provided template"""
from openpyxl import Workbook
from openpyxl.styles import Font, PatternFill, Border, Side, Alignment
from openpyxl.utils import get_column_letter
import io
# Create workbook
wb = Workbook()
ws = wb.active
ws.title = "Sheet0"
# Resolve date range; skip 14-day cap when unlimited=True
result = _resolve_date_range(start_date_filter, end_date_filter, records, 'TA Excel export', unlimited=unlimited)
if result is None:
return None
start_date, end_date, records = result
# Import parse function at the beginning for work type detection
from working_hours_calculator import parse_employee_id_for_work_type
# Convert TimeAttendance records to format expected by calculator
converted_records = _convert_ta_records(records)
# Calculate working hours using WorkingHoursCalculator
calculator = WorkingHoursCalculator()
hours_data = calculator.calculate_all_employees_hours(
datetime.combine(start_date, datetime.min.time()),
datetime.combine(end_date, datetime.max.time()),
converted_records
)
# Build employee name map (Lastname, Firstname keyed by base employee ID)
employee_names = _build_employee_name_map(records)
# Setup styles (shared objects)
_styles = _make_export_styles()
header_font = _styles['header_font']
header_fill = _styles['header_fill']
data_font = _styles['data_font']
bold_font = _styles['bold_font']
italic_bold_font = _styles['italic_bold_font']
border = _styles['border']
missed_punch_fill = _styles['missed_punch_fill']
amber_fill = _styles['amber_fill']
border_day_middle = _styles['border_day_middle']
border_day_last = _styles['border_day_last']
border_day_single = _styles['border_day_single']
border_day_first = _styles['border_day_first']
def get_day_border(row_position, total_rows):
"""
Get appropriate border style based on row position within a day.
Matches sample.xlsx: only the LAST row of each day group has a bottom border.
Args:
row_position: Current row number (0-indexed) within the day
total_rows: Total number of rows for this day
Returns:
Border object
"""
if total_rows == 1:
return border_day_single
elif row_position == total_rows - 1:
return border_day_last
else:
return border_day_middle
# Orange background for Missed Punch
missed_punch_fill = PatternFill(start_color='FFC000', end_color='FFC000', fill_type='solid')
# Write main headers
current_row = 1
# Row 1: Company name
ws.merge_cells(f'A{current_row}:N{current_row}')
title_cell = ws.cell(row=current_row, column=1, value=current_app.config.get('COMPANY_NAME', 'QR Code Management System'))
title_cell.font = Font(name='Aptos Narrow', size=14, bold=True)
title_cell.alignment = Alignment(horizontal='left')
current_row += 1
# Row 2: Summary title
ws.merge_cells(f'A{current_row}:N{current_row}')
summary_cell = ws.cell(row=current_row, column=1, value='Summary report of Hours worked')
summary_cell.font = Font(name='Aptos Narrow', size=12, bold=True)
summary_cell.alignment = Alignment(horizontal='left')
current_row += 1
# Row 3: Project name
project_display = project_name_for_filename.replace('_', ' ').strip() if project_name_for_filename else "[Project Name]"
project_cell = ws.cell(row=current_row, column=1, value=project_display)
project_cell.font = Font(name='Aptos Narrow', size=11, bold=True)
project_cell.alignment = Alignment(horizontal='left')
current_row += 1
# Row 4: Date range
date_range_text = f"Date range: {start_date.strftime('%m/%d/%Y')} to {end_date.strftime('%m/%d/%Y')}"
ws.merge_cells(f'A{current_row}:N{current_row}')
date_cell = ws.cell(row=current_row, column=1, value=date_range_text)
date_cell.font = Font(name='Aptos Narrow', size=11)
date_cell.alignment = Alignment(horizontal='left')
current_row += 1
# Row 5: Empty row
current_row += 1
# Empty row before first employee
current_row += 1
# Sort employees by name for organized output
sorted_employees = sorted(
hours_data['employees'].items(),
key=lambda x: employee_names.get(x[0], f'Employee {x[0]}').lower()
)
# Write data for each employee (sorted by name)
for employee_id, emp_data in sorted_employees:
employee_name = employee_names.get(employee_id, f'Employee {employee_id}')
# Employee header row (merged A to O)
ws.merge_cells(f'A{current_row}:O{current_row}')
emp_header = ws.cell(row=current_row, column=1,
value=f'Employee ID {employee_id}: {employee_name}')
emp_header.font = Font(name='Aptos Narrow', size=11, bold=True)
emp_header.alignment = Alignment(horizontal='left')
current_row += 1
# Column headers
headers = ['Day', 'Date', 'In', 'Out', 'Location', 'Zone', 'Hours/Building',
'Daily Total', 'Regular Hours', 'OT Hours', 'Building Address',
'Recorded Location', 'Distance (Mile)', 'Possible Violation']
for col, header in enumerate(headers, 1):
cell = ws.cell(row=current_row, column=col, value=header)
# White bold text on black background; no border (matching sample.xlsx)
cell.font = header_font
cell.fill = header_fill
cell.alignment = Alignment(horizontal='center', vertical='center')
current_row += 1
# Group records by date AND location for separate rows per location
daily_location_data = {}
# Import parse function to match base employee ID with all variants (SP, PW, PT)
from working_hours_calculator import parse_employee_id_for_work_type
# Filter records where the BASE employee ID matches (includes 1234, 1234 SP, 1234 PW, 1234 PT)
employee_records = []
for r in converted_records:
record_base_id, _ = parse_employee_id_for_work_type(str(r.employee_id))
if record_base_id == employee_id:
employee_records.append(r)
for record in employee_records:
date_key = record.check_in_date.strftime('%Y-%m-%d')
location_key = record.location_name or 'Unknown Location'
# Create nested structure: date -> location -> records
if date_key not in daily_location_data:
daily_location_data[date_key] = {}
if location_key not in daily_location_data[date_key]:
daily_location_data[date_key][location_key] = {
'records': [],
'location_name': location_key
}
daily_location_data[date_key][location_key]['records'].append(record)
# -------------------------------------------------------------------
# OVERNIGHT SHIFT DETECTION
# The midnight check-out record is stored in the DB with the next
# calendar day's date (e.g. checkout at 12:18 AM on Thursday is
# stored as check_in_date = 2026-02-26). We need to move it into
# Wednesday's bucket so it pairs with the 8:18 PM check-in.
#
# Condition to move an early-morning checkout from Day N+1 -> Day N:
# Day N: has an unmatched late check-in (>= 18:00)
# Day N+1: has an early check-out (<= 06:00) that belongs to Day N,
# detected by the absence of a non-evening IN on Day N+1
# that could own the early OUT (or raw count imbalance).
# -------------------------------------------------------------------
def _is_out(r):
a = (r.action_description or '').lower()
return 'out' in a or 'checkout' in a
sorted_dk = sorted(daily_location_data.keys())
for _di, _dk in enumerate(sorted_dk):
if _di + 1 >= len(sorted_dk):
continue
# Guard: _dk or _ndk may have been deleted by a prior iteration
# when all its records were moved to the previous day's bucket.
# Without this check, iterating the stale sorted_dk snapshot raises KeyError.
if _dk not in daily_location_data:
continue
_ndk = sorted_dk[_di + 1]
if _ndk not in daily_location_data:
continue
# Must be consecutive calendar days
_dn = datetime.strptime(_dk, '%Y-%m-%d').date()
_dn1 = datetime.strptime(_ndk, '%Y-%m-%d').date()
if (_dn1 - _dn).days != 1:
continue
# Flatten all records for Day N and Day N+1 across locations
_day_recs = [r for loc in daily_location_data[_dk].values() for r in loc['records']]
_next_recs = [r for loc in daily_location_data[_ndk].values() for r in loc['records']]
_day_ins = [r for r in _day_recs if not _is_out(r)]
_day_outs = [r for r in _day_recs if _is_out(r)]
_nxt_ins = [r for r in _next_recs if not _is_out(r)]
_nxt_outs = [r for r in _next_recs if _is_out(r)]
# Early-morning OUTs on Day N (hour <= 3) are overnight orphans from
# Day N-1. Counting them as regular Day N outs inflates the out-count
# and makes the day appear balanced, which suppresses detection of an
# unmatched late IN that needs a next-day OUT. Exclude them.
_day_outs_non_early = [r for r in _day_outs if r.check_in_time.hour > 3]
# Day N must have an unmatched late check-in (more INs than non-early OUTs,
# with at least one IN at or after 12:00 PM)
if len(_day_ins) <= len(_day_outs_non_early):
continue
_late_ins = [r for r in _day_ins if r.check_in_time.hour >= 12]
if not _late_ins:
continue
# Find early-morning OUTs (<=03:00) on Day N+1
_early_outs = [r for r in _nxt_outs if r.check_in_time.hour <= 3]
if not _early_outs:
continue
# Determine whether the early OUT belongs to Day N or Day N+1.
# It belongs to Day N when Day N+1 has no morning (< 12:00) check-in
# that could own it, OR when OUTs outnumber INs on Day N+1.
# This handles both cases:
# Case A: Day N+1 has only afternoon/evening INs (all >= 12:00) -> early OUT is Day N's
# Case B: Day N+1 has more OUTs than INs overall -> early OUT is unmatched
# A morning IN on Day N+1 can only own an early OUT when that IN
# occurs STRICTLY BEFORE the early OUT's time (IN → OUT is time-ordered).
# An IN that starts AFTER the early OUT cannot own it and must NOT block
# the overnight move (e.g. 01:55 AM IN cannot own a 01:00 AM OUT).
_nxt_non_evening_ins = [
r for r in _nxt_ins
if r.check_in_time.hour < 12
and any(r.check_in_time < eo.check_in_time for eo in _early_outs)
]
if _nxt_non_evening_ins and len(_nxt_outs) <= len(_nxt_ins):
# Day N+1 has a morning IN that can own the early OUT, and counts
# are balanced -> do NOT move
continue
# Move up to as many early OUTs as there are unmatched late INs on Day N
_to_move = _early_outs[:len(_late_ins)]
for _co in _to_move:
_co_loc = _co.location_name or 'Unknown Location'
# Add to Day N bucket
if _co_loc not in daily_location_data[_dk]:
daily_location_data[_dk][_co_loc] = {'records': [], 'location_name': _co_loc}
daily_location_data[_dk][_co_loc]['records'].append(_co)
# Remove from Day N+1 bucket
if _ndk in daily_location_data and _co_loc in daily_location_data[_ndk]:
try:
daily_location_data[_ndk][_co_loc]['records'].remove(_co)
except ValueError:
pass
if not daily_location_data[_ndk][_co_loc]['records']:
del daily_location_data[_ndk][_co_loc]
if _ndk in daily_location_data and not daily_location_data[_ndk]:
del daily_location_data[_ndk]
logger_handler.logger.info(
f"TA Export overnight shift: moved checkout {_co.check_in_time} "
f"from {_ndk} to {_dk} for employee {employee_id}"
)
# -------------------------------------------------------------------
# END OVERNIGHT SHIFT DETECTION
# -------------------------------------------------------------------
# Track weekly hours for overtime calculation
weekly_total_hours = 0
current_week_start = None
grand_regular_hours = 0
grand_ot_hours = 0
# Accumulate SP/PW/PT hours from cross-type pairs (where the calculator
# could not detect them because it processes each work-type stream independently).
cross_type_sp_hours = 0.0
cross_type_pw_hours = 0.0
cross_type_pt_hours = 0.0
# Accumulate raw (uncapped) hours from regular (non-SP/PW/PT) pairs only.
# Used to populate the "Regular" summary row when an employee also has
# special work-type hours (SP/PW/PT).
regular_only_hours = 0.0
# Get all dates that have records (not all weekdays)
dates_with_records = sorted([
date_str for date_str, day_data in emp_data['daily_hours'].items()
if day_data.get('records_count', 0) > 0
])
# Write daily data (ONLY DAYS WITH RECORDS)
for date_str in dates_with_records:
date_obj = datetime.strptime(date_str, '%Y-%m-%d')
day_data = emp_data['daily_hours'][date_str]
# Check for week boundary anchored to start_date_filter (not calendar Monday)
_report_start = start_date if start_date_filter else date_obj.date()
week_start = (_report_start + timedelta(days=((date_obj.date() - _report_start).days // 7) * 7))
if current_week_start is not None and week_start != current_week_start:
# Write weekly total row
week_regular = min(weekly_total_hours, 40.0)
week_overtime = max(0, weekly_total_hours - 40.0)
ws.cell(row=current_row, column=7, value='Weekly Total: ').font = bold_font
ws.cell(row=current_row, column=8, value=_qtr(weekly_total_hours)).font = bold_font
ws.cell(row=current_row, column=9, value=_qtr(week_regular)).font = bold_font
ws.cell(row=current_row, column=10, value=_qtr(week_overtime)).font = bold_font
grand_regular_hours += week_regular
grand_ot_hours += week_overtime
current_row += 1
weekly_total_hours = 0
current_week_start = week_start
# Get all locations for this date
date_locations = daily_location_data.get(date_str, {})
total_locations = len(date_locations)
total_hours = day_data['total_hours']
is_miss_punch = day_data.get('is_miss_punch', False)
# Re-evaluate is_miss_punch from actual records in daily_location_data.
# The overnight detection may have moved a checkout into this day's bucket
# AFTER working_hours_calculator ran, so emp_data may still say
# is_miss_punch=True even though the records now form a valid IN/OUT pair.
if is_miss_punch and total_locations > 0:
_all_recs_check = [r for loc in date_locations.values() for r in loc['records']]
_ins_c = sum(1 for r in _all_recs_check if not _is_out(r))
_outs_c = sum(1 for r in _all_recs_check if _is_out(r))
if _ins_c > 0 and _outs_c > 0 and _ins_c == _outs_c:
# Balanced pairs — overnight fix resolved the miss punch
is_miss_punch = False
total_hours = 0.0 # will be recalculated below
# Calculate total hours for the day by mirroring the display pairing logic:
# group records by base location, apply the OUT-after-IN guard within each
# group, and sum only complete pairs. This ensures the daily total in
# column H matches exactly the pairs rendered in the export rows.
_day_total_hours = 0.0
# Track which records are consumed by same-building pairing so the
# cross-building pass only considers true orphans.
_same_building_used_ids = set()
for _loc_data in date_locations.values():
_loc_recs = sorted(_loc_data['records'], key=_overnight_aware_sort_key)
_loc_ins = [r for r in _loc_recs if not _is_out(r)]
_loc_outs = [r for r in _loc_recs if _is_out(r)]
_out_used = [False] * len(_loc_outs)
for _in_r in _loc_ins:
for _oi2, _out_r in enumerate(_loc_outs):
if _out_used[_oi2]:
continue
# Time-only pairing guard (mirrors Step 1/2 pairing logic).
_in_t_d = _in_r.check_in_time
_out_t_d = _out_r.check_in_time
if _out_t_d.hour <= 3:
if _in_t_d.hour < 12:
continue
# Orphan guard: an early-morning OUT whose check_in_date
# matches the current day is an orphan from the PREVIOUS
# overnight shift — it must NOT steal an evening IN.
# Only OUTs moved in by overnight detection (check_in_date
# is later than the current day) are valid partners.
_out_orig_date = _out_r.check_in_date
if hasattr(_out_orig_date, 'date'):
_out_orig_date = _out_orig_date.date()
if _out_orig_date <= date_obj.date():
continue
elif _out_t_d <= _in_t_d:
continue
_in_ts = datetime.combine(_in_r.check_in_date, _in_r.check_in_time)
_out_ts = datetime.combine(_out_r.check_in_date, _out_r.check_in_time)
if _out_ts < _in_ts:
_out_ts += timedelta(days=1)
_duration = (_out_ts - _in_ts).total_seconds() / 3600.0
if _duration > 24:
continue
_day_total_hours += _duration
_out_used[_oi2] = True
_same_building_used_ids.add(id(_in_r))
_same_building_used_ids.add(id(_out_r))
break
# -------------------------------------------------------------------
# CROSS-BUILDING PAIRING
# After same-building pairing, collect all orphaned INs and OUTs
# across every location group for this day. Pair them chronologically
# (earliest available OUT that is strictly after the IN). This handles
# employees who check in at one building and check out at another.
# -------------------------------------------------------------------
_all_day_recs_flat = []
for _loc_data in date_locations.values():
_all_day_recs_flat.extend(_loc_data['records'])
_orphan_ins = sorted(
[r for r in _all_day_recs_flat if not _is_out(r) and id(r) not in _same_building_used_ids],
key=_overnight_aware_sort_key
)
_orphan_outs = sorted(
[r for r in _all_day_recs_flat if _is_out(r) and id(r) not in _same_building_used_ids],
key=_overnight_aware_sort_key
)
# Pre-compute cross-building pairs for this day (used both for totals
# and for row writing after the location_groups loop).
cross_building_pairs = [] # list of {'check_in': r, 'check_out': r, 'hours': float}
_cb_out_used = [False] * len(_orphan_outs)
for _cb_in in _orphan_ins:
for _cb_oi, _cb_out in enumerate(_orphan_outs):
if _cb_out_used[_cb_oi]:
continue
# Same time-only guard as Steps 1–3
_cb_in_t = _cb_in.check_in_time
_cb_out_t = _cb_out.check_in_time
if _cb_out_t.hour <= 3:
if _cb_in_t.hour < 12:
continue
# Orphan guard: same-day early-morning OUT is from previous
# overnight shift — skip it. Only moved OUTs (check_in_date
# later than current day) are valid overnight partners.
_cb_out_orig = _cb_out.check_in_date
if hasattr(_cb_out_orig, 'date'):
_cb_out_orig = _cb_out_orig.date()
if _cb_out_orig <= date_obj.date():
continue
elif _cb_out_t <= _cb_in_t:
continue
_cb_in_ts = datetime.combine(_cb_in.check_in_date, _cb_in.check_in_time)
_cb_out_ts = datetime.combine(_cb_out.check_in_date, _cb_out.check_in_time)
if _cb_out_ts < _cb_in_ts:
_cb_out_ts += timedelta(days=1)
_cb_dur = (_cb_out_ts - _cb_in_ts).total_seconds() / 3600.0
if _cb_dur > 24:
continue
_cb_out_used[_cb_oi] = True
cross_building_pairs.append({
'check_in': _cb_in,
'check_out': _cb_out,
'hours': _cb_dur,
})
_day_total_hours += _cb_dur
logger_handler.logger.info(
f"[TA Export] Cross-building pair for employee {employee_id} on {date_str}: "
f"IN {_cb_in.location_name} @ {_cb_in.check_in_time} → "
f"OUT {_cb_out.location_name} @ {_cb_out.check_in_time} "
f"({_cb_dur:.2f} h)"
)
break
# Build a set of record ids that are part of a cross-building pair so
# the single-record group path can suppress its Missed Punch row.
_cross_building_record_ids = set()
for _cbp in cross_building_pairs:
_cross_building_record_ids.add(id(_cbp['check_in']))
_cross_building_record_ids.add(id(_cbp['check_out']))
# -------------------------------------------------------------------
# END CROSS-BUILDING PAIRING PRE-COMPUTATION
# -------------------------------------------------------------------
total_hours = _qtr(_day_total_hours)
weekly_total_hours += total_hours
# Daily total display (only shown on last location's last row)
daily_total_display = _qtr(total_hours) if total_hours > 0 else ''
# Get all records for the day and sort by time FIRST, then group by BASE location.
# Grouping by base location (original_location_name) ensures that records from the
# same building but different work types (e.g. regular IN + SP OUT) land in the
# same group so the cross-type pairing rule can resolve them.
all_day_records = []
for loc_data in date_locations.values():
all_day_records.extend(loc_data['records'])
# Sort all records by time chronologically, overnight-aware
all_day_records_sorted = sorted(all_day_records, key=_overnight_aware_sort_key)
# Group consecutive records by BASE location (original_location_name without work-type
# suffix) while maintaining time order.
def _base_loc(r):
return getattr(r, 'original_location_name', None) or r.location_name or 'Unknown Location'
location_groups = []
current_base_location = None
current_group = []
for record in all_day_records_sorted:
bloc = _base_loc(record)
if current_base_location is None or bloc == current_base_location:
current_base_location = bloc
current_group.append(record)
else:
if current_group:
location_groups.append({
'location': current_base_location,
'records': current_group
})
current_base_location = bloc
current_group = [record]
# Add the last group
if current_group:
location_groups.append({
'location': current_base_location,
'records': current_group
})
# Process each location group in chronological order
total_groups = len(location_groups)
for group_index, group_data in enumerate(location_groups):
location_count = group_index + 1
is_last_location = (location_count == total_groups)
location_name = group_data['location']
sorted_records = group_data['records']
if len(sorted_records) == 1:
# Single record for this location
single_record = sorted_records[0]
# If this record has been resolved by cross-building pairing,
# suppress the Missed Punch row here — it will be written after
# all location groups have been processed (Touch Point 3).
if id(single_record) in _cross_building_record_ids:
continue
# Get the original TimeAttendance record to check action_description
original_record = None
for rec in records:
if (rec.employee_id == single_record.employee_id and
rec.attendance_date == single_record.check_in_date and
rec.attendance_time == single_record.check_in_time):
original_record = rec
break
# Determine if this is a check-in or check-out
is_check_out = False
if original_record and original_record.action_description:
action_lower = original_record.action_description.lower()
is_check_out = 'out' in action_lower or 'checkout' in action_lower
# Show day name and date only for first group's first record
day_display = date_obj.strftime('%A').upper() if location_count == 1 else ''
date_display = date_obj.strftime('%m/%d/%Y') if location_count == 1 else ''
# Show daily total only if this is the last group
current_daily_total = daily_total_display if is_last_location else ''
if is_check_out:
# Orphaned check-out
row_data = [
day_display,
date_display,
'', # No check-in time
single_record.check_in_time.strftime('%I:%M:%S %p'), # Out
single_record.location_name,
'',
'Missed Punch',
current_daily_total,
'',
'',
single_record.event_description or '',
single_record.recorded_address or '',
getattr(single_record, 'distance', None) or '',
calculate_possible_violation(getattr(single_record, 'distance', None))
]
else:
# Orphaned check-in
row_data = [
day_display,
date_display,
single_record.check_in_time.strftime('%I:%M:%S %p'), # In
'', # No check-out time
single_record.location_name,
'',
'Missed Punch',
current_daily_total,
'',
'',
single_record.event_description or '',
single_record.recorded_address or '',
getattr(single_record, 'distance', None) or '',
calculate_possible_violation(getattr(single_record, 'distance', None))
]
day_border = border_day_last if is_last_location else border_day_middle
for col, value in enumerate(row_data, 1):
cell = ws.cell(row=current_row, column=col, value=value)
cell.font = data_font
cell.border = day_border
# Apply orange background to Missed Punch cell (column G)
if col == 7:
cell.fill = missed_punch_fill
current_row += 1
else:
# Multiple records for this location group.
# Build record_info with work_type included.
record_info = []
for record in sorted_records:
action_desc = record.action_description.lower() if record.action_description else ''
is_out = 'out' in action_desc or 'checkout' in action_desc
wt = getattr(record, 'work_type', None) # None = regular
record_info.append({
'record': record,
'is_out': is_out,
'work_type': wt, # None means regular
'used': False
})
logger_handler.logger.debug(f"TA Export record: time={record.check_in_time}, action='{record.action_description}', is_out={is_out}, work_type={wt}")
ins = [ri for ri in record_info if not ri['is_out']]
outs = [ri for ri in record_info if ri['is_out']]
pairs_to_write = []
# ── STEP 1: same-type pairing ──────────────────────────────────────
# Pair each IN with an OUT of the same work type first.
# Sort INs chronologically and OUTs with overnight-aware key so that
# an early-morning OUT (e.g. 00:30 moved in by overnight detection)
# sorts AFTER same-day evening OUTs and does not steal a daytime IN.
ins_sorted = sorted(ins, key=lambda ri: _overnight_aware_sort_key(ri['record']))
outs_sorted = sorted(outs, key=lambda ri: _overnight_aware_sort_key(ri['record']))
for in_ri in ins_sorted:
if in_ri['used']:
continue
for out_ri in outs_sorted:
if out_ri['used']:
continue
# Guard: time-only pairing rule.
# An early-morning OUT (hour<=3) is only valid for an afternoon/evening IN (hour>=12).
# For all other OUTs, the OUT time must be strictly after the IN time.
# Using time-only (not datetime) avoids false positives from moved overnight
# OUT records whose check_in_date is still a later date.
_in_t = in_ri['record'].check_in_time
_out_t = out_ri['record'].check_in_time
if _out_t.hour <= 3:
if _in_t.hour < 12:
continue # early-morning OUT cannot pair with morning IN
# Orphan guard: same-day early-morning OUT is from a
# previous overnight shift — not a valid partner for
# this evening IN. Only moved OUTs (check_in_date
# later than current day) should pair.
_out_orig_d = out_ri['record'].check_in_date
if hasattr(_out_orig_d, 'date'):
_out_orig_d = _out_orig_d.date()
if _out_orig_d <= date_obj.date():
continue
elif _out_t <= _in_t:
continue # same-day OUT must be strictly after IN
if out_ri['work_type'] == in_ri['work_type']:
# Matched same work type — standard pair
in_ri['used'] = True
out_ri['used'] = True
pairs_to_write.append({
'check_in': in_ri['record'],
'check_out': out_ri['record'],
'is_miss_punch': False,
'effective_work_type': in_ri['work_type']
})
break
# ── STEP 2: cross-type pairing (forgot the work code) ──────────────
# If any INs or OUTs remain unmatched after same-type pairing,
# attempt to pair an unmatched IN with an unmatched OUT of a
# *different* work type. Hours count as the special type's hours
# (if either side is special, the pair is treated as special;
# if both are different special types, use the OUT's type as
# the authoritative code — it's the scan that carries the code).
unmatched_ins = [ri for ri in ins_sorted if not ri['used']]
unmatched_outs = [ri for ri in outs_sorted if not ri['used']]
for in_ri in unmatched_ins:
if in_ri['used']:
continue
for out_ri in unmatched_outs:
if out_ri['used']:
continue
# Guard: same time-only rule as Step 1.
_in_t2 = in_ri['record'].check_in_time
_out_t2 = out_ri['record'].check_in_time
if _out_t2.hour <= 3:
if _in_t2.hour < 12:
continue
# Orphan guard: same-day early-morning OUT is from a
# previous overnight shift — not a valid partner for
# this evening IN. Only moved OUTs (check_in_date
# later than current day) should pair.
_out_orig_d2 = out_ri['record'].check_in_date
if hasattr(_out_orig_d2, 'date'):
_out_orig_d2 = _out_orig_d2.date()
if _out_orig_d2 <= date_obj.date():
continue
elif _out_t2 <= _in_t2:
continue
# Cross-type pair: one side is regular, other is special
# (or both special but different codes — treat OUT's type as definitive)
effective_wt = out_ri['work_type'] if out_ri['work_type'] else in_ri['work_type']
in_ri['used'] = True
out_ri['used'] = True
pairs_to_write.append({
'check_in': in_ri['record'],
'check_out': out_ri['record'],
'is_miss_punch': False,
'effective_work_type': effective_wt,
'is_cross_type': True
})
break
# ── STEP 3: remaining unmatched records → Missed Punch ─────────────
for ri in record_info:
if not ri['used']:
ri['used'] = True
if ri['is_out']:
pairs_to_write.append({
'check_in': None,
'check_out': ri['record'],
'is_miss_punch': True,
'effective_work_type': ri['work_type']
})
else:
pairs_to_write.append({
'check_in': ri['record'],
'check_out': None,
'is_miss_punch': True,
'effective_work_type': ri['work_type']
})
logger_handler.logger.debug(f"TA Export: created {len(pairs_to_write)} pairs for export")
# Sort pairs chronologically by the anchor record's time so that
# orphaned records (assembled last in Steps 2-3) appear in the
# correct time-order position relative to complete pairs.
def _pair_sort_key(pd):
anchor = pd['check_in'] or pd['check_out']
return _overnight_aware_sort_key(anchor) if anchor else 0
pairs_to_write.sort(key=_pair_sort_key)
# Write all pairs
for pair_idx, pair_data in enumerate(pairs_to_write):
check_in_record = pair_data['check_in']
check_out_record = pair_data['check_out']
is_miss_punch = pair_data['is_miss_punch']
# Show day name and date only for first pair of first location
day_display = date_obj.strftime('%A').upper() if (location_count == 1 and pair_idx == 0) else ''
date_display = date_obj.strftime('%m/%d/%Y') if (location_count == 1 and pair_idx == 0) else ''
# Calculate hours if complete pair
if check_in_record and check_out_record and not is_miss_punch:
pair_datetime_in = datetime.combine(check_in_record.check_in_date, check_in_record.check_in_time)
pair_datetime_out = datetime.combine(check_out_record.check_in_date, check_out_record.check_in_time)
# If check-out time is before check-in time (overnight shift),
# add one day to the check-out datetime so the duration is positive and correct.
if pair_datetime_out < pair_datetime_in:
pair_datetime_out += timedelta(days=1)
pair_hours = (pair_datetime_out - pair_datetime_in).total_seconds() / 3600.0
pair_hours = round(pair_hours, 2)
else:
pair_hours = 'Missed Punch'
# Accumulate SP/PW/PT hours for CROSS-TYPE pairs only.
# Same-type SP/PW/PT pairs are already captured in grand_totals
# by WorkingHoursCalculator; adding them again here would double-count.
if not is_miss_punch and isinstance(pair_hours, (int, float)) and pair_data.get('is_cross_type', False):
_ewt = pair_data.get('effective_work_type')
if _ewt == 'SP':
cross_type_sp_hours += pair_hours
elif _ewt == 'PW':
cross_type_pw_hours += pair_hours
elif _ewt == 'PT':
cross_type_pt_hours += pair_hours
# Accumulate raw regular-only hours (non-SP/PW/PT completed pairs).
# This feeds the "Regular" summary row shown when an employee also
# has SP/PW/PT hours. Cross-type pairs are excluded here because
# their effective_work_type is SP/PW/PT, not regular.
if not is_miss_punch and isinstance(pair_hours, (int, float)):
_pair_ewt = pair_data.get('effective_work_type')
if _pair_ewt not in ('SP', 'PW', 'PT'):
regular_only_hours += pair_hours
# Determine whether this is an overnight pair:
# check-in is late evening (>= 20:00) AND check-out is early morning (<= 03:00)
# Both records share the same check_in_date in the DB for this scenario.
_is_overnight_pair = (
check_in_record and check_out_record and
check_in_record.check_in_time.hour >= 20 and
check_out_record.check_in_time.hour <= 3
)
# Show daily total on last pair of last location
is_last_pair = (pair_idx == len(pairs_to_write) - 1) and is_last_location
current_daily_total = daily_total_display if is_last_pair else ''
# Build Out-time string (plain time only)
_out_time_str = check_out_record.check_in_time.strftime('%I:%M:%S %p') if check_out_record else ''
# Build Location string.
# For a complete pair, derive the display name from effective_work_type:
# - regular pair → base location name (no suffix)
# - SP/PW/PT pair → base location name + " (SP/PW/PT)"
# 'regular' is treated identically to None — no suffix is shown.
# For orphaned records keep their own location_name.
_effective_wt = pair_data.get('effective_work_type')
_is_special_wt = _effective_wt in ('SP', 'PW', 'PT')
_ref_record = check_in_record or check_out_record
if check_in_record and check_out_record:
_base = _base_loc(check_in_record)
if _is_special_wt:
_location_str = f"{_base} ({_effective_wt})"
else:
_location_str = _base
else:
_location_str = _ref_record.location_name if _ref_record else ''
if _is_overnight_pair:
_location_str = f"{_location_str} (midnight shift)"
# Build row data
if check_in_record and check_out_record:
row_data = [
day_display,
date_display,
check_in_record.check_in_time.strftime('%I:%M:%S %p'), # In
_out_time_str, # Out
_location_str, # Location (effective work type + optional midnight label)
'',
pair_hours,
current_daily_total,
'',
'',
check_in_record.event_description or '',
check_in_record.recorded_address or '',
getattr(check_in_record, 'distance', None) or '',
calculate_possible_violation(getattr(check_in_record, 'distance', None))
]
elif check_in_record: # IN without OUT
row_data = [
day_display,
date_display,
check_in_record.check_in_time.strftime('%I:%M:%S %p'), # In
'', # No OUT
_location_str,
'',
'Missed Punch',
current_daily_total,
'',
'',
check_in_record.event_description or '',
check_in_record.recorded_address or '',
getattr(check_in_record, 'distance', None) or '',
calculate_possible_violation(getattr(check_in_record, 'distance', None))
]
else: # OUT without IN
row_data = [
day_display,
date_display,
'', # No IN
check_out_record.check_in_time.strftime('%I:%M:%S %p'), # Out
_location_str,
'',
'Missed Punch',
current_daily_total,
'',
'',
check_out_record.event_description or '',
check_out_record.recorded_address or '',
getattr(check_out_record, 'distance', None) or '',
calculate_possible_violation(getattr(check_out_record, 'distance', None))
]
day_border = border_day_last if is_last_pair else border_day_middle
for col, value in enumerate(row_data, 1):
cell = ws.cell(row=current_row, column=col, value=value)
cell.font = data_font
cell.border = day_border
# Apply orange background to Missed Punch cell
if col == 7 and value == 'Missed Punch':
cell.fill = missed_punch_fill
current_row += 1
# -------------------------------------------------------------------
# CROSS-BUILDING PAIR ROW WRITING (Touch Point 3)
# Write one row per cross-building pair identified during pre-computation.
# The day-name and date columns are only shown for the very first row
# of this day that is actually rendered; we track that with a flag.
# -------------------------------------------------------------------
if cross_building_pairs:
# Determine whether any non-cross-building rows were already written
# for this day. We look at how many rows were consumed since the
# start of this date's block. The simplest proxy: check whether
# the first location group had at least one real (non-skipped) record.
# We use a dedicated flag instead to keep this clean.
_cb_first_row_of_day = not any(
id(r) not in _cross_building_record_ids
for loc_data in date_locations.values()
for r in loc_data['records']
)
for _cb_idx, _cbp in enumerate(cross_building_pairs):
_cb_in_rec = _cbp['check_in']
_cb_out_rec = _cbp['check_out']
_cb_hours = _cbp['hours']
_cb_pair_hours = round(_cb_hours, 2)
_is_last_cb = (_cb_idx == len(cross_building_pairs) - 1)
# Show day/date only on the very first row written for this date
# (either this is the first row overall, or prior groups had records)
if _cb_idx == 0 and _cb_first_row_of_day:
_cb_day_display = date_obj.strftime('%A').upper()
_cb_date_display = date_obj.strftime('%m/%d/%Y')
else:
_cb_day_display = ''
_cb_date_display = ''
# Show daily total on the last cross-building row if it is
# also the last row written for this day.
_cb_daily_total = daily_total_display if _is_last_cb else ''
# Location label: clearly identifies both buildings
_cb_in_loc = _base_loc(_cb_in_rec)
_cb_out_loc = _base_loc(_cb_out_rec)
_cb_loc_str = f"IN: {_cb_in_loc} → OUT: {_cb_out_loc}"
row_data = [
_cb_day_display,
_cb_date_display,
_cb_in_rec.check_in_time.strftime('%I:%M:%S %p'), # In
_cb_out_rec.check_in_time.strftime('%I:%M:%S %p'), # Out
_cb_loc_str,
'',
_cb_pair_hours,
_cb_daily_total,
'',
'',
_cb_in_rec.event_description or '',
_cb_in_rec.recorded_address or '',
getattr(_cb_in_rec, 'distance', None) or '',
calculate_possible_violation(getattr(_cb_in_rec, 'distance', None))
]
_cb_border = border_day_last if _is_last_cb else border_day_middle
for col, value in enumerate(row_data, 1):
cell = ws.cell(row=current_row, column=col, value=value)
cell.font = data_font
cell.border = _cb_border
current_row += 1
# -------------------------------------------------------------------
# END CROSS-BUILDING PAIR ROW WRITING
# -------------------------------------------------------------------
# Write final weekly total for this employee
if weekly_total_hours > 0:
week_regular = min(weekly_total_hours, 40.0)
week_overtime = max(0, weekly_total_hours - 40.0)
ws.cell(row=current_row, column=7, value='Weekly Total: ').font = bold_font
ws.cell(row=current_row, column=8, value=_qtr(weekly_total_hours)).font = bold_font
ws.cell(row=current_row, column=9, value=_qtr(week_regular)).font = bold_font
ws.cell(row=current_row, column=10, value=_qtr(week_overtime)).font = bold_font
grand_regular_hours += week_regular
grand_ot_hours += week_overtime
current_row += 1
# Write extra working hours rows (SP/PW/PT) if employee has any.
# Get extra hours from emp_data grand_totals, then add any cross-type hours
# accumulated during rendering (pairs the calculator could not detect).
grand_totals = emp_data.get('grand_totals', {})
sp_hours = grand_totals.get('sp_hours', 0.0) + cross_type_sp_hours
pw_hours = grand_totals.get('pw_hours', 0.0) + cross_type_pw_hours
pt_hours = grand_totals.get('pt_hours', 0.0) + cross_type_pt_hours
_has_special_hours = sp_hours > 0 or pw_hours > 0 or pt_hours > 0
_summary_font = Font(name='Aptos Narrow', size=11, bold=True, italic=True)
# Write SP row if hours > 0 (abbreviated label in col 8, hours in col 9)
if sp_hours > 0:
ws.cell(row=current_row, column=8, value='SP').font = _summary_font
ws.cell(row=current_row, column=9, value=_qtr(sp_hours)).font = _summary_font
logger_handler.logger.info(f"Export: Employee {employee_id} SP hours: {sp_hours:.2f}")
current_row += 1
# Write PW row if hours > 0
if pw_hours > 0:
ws.cell(row=current_row, column=8, value='PW').font = _summary_font
ws.cell(row=current_row, column=9, value=_qtr(pw_hours)).font = _summary_font
logger_handler.logger.info(f"Export: Employee {employee_id} PW hours: {pw_hours:.2f}")
current_row += 1
# Write PT row if hours > 0
if pt_hours > 0:
ws.cell(row=current_row, column=8, value='PT').font = _summary_font
ws.cell(row=current_row, column=9, value=_qtr(pt_hours)).font = _summary_font
logger_handler.logger.info(f"Export: Employee {employee_id} PT hours: {pt_hours:.2f}")
current_row += 1
# Write Regular row — only when the employee has at least one special
# work-type (SP/PW/PT). Shows raw accumulated hours from non-special pairs.
if _has_special_hours:
ws.cell(row=current_row, column=8, value='Regular').font = _summary_font
ws.cell(row=current_row, column=9, value=_qtr(regular_only_hours)).font = _summary_font
logger_handler.logger.info(f"Export: Employee {employee_id} Regular hours: {regular_only_hours:.2f}")
current_row += 1
# Write GRAND TOTAL row
ws.cell(row=current_row, column=7, value='GRAND TOTAL: ').font = Font(name='Aptos Narrow', size=11, bold=True)
ws.cell(row=current_row, column=9, value=_qtr(grand_regular_hours)).font = Font(name='Aptos Narrow', size=11, bold=True)
ws.cell(row=current_row, column=10, value=_qtr(grand_ot_hours)).font = Font(name='Aptos Narrow', size=11, bold=True)
current_row += 1
# Empty row after each employee
current_row += 1
# Auto-size columns - handle merged cells properly
for col_idx in range(1, 15):
column_letter = get_column_letter(col_idx)
# Set fixed width for Day column (column A)
if col_idx == 1:
ws.column_dimensions[column_letter].width = 18
continue
max_length = 0
for row in ws.iter_rows(min_col=col_idx, max_col=col_idx):
for cell in row:
if isinstance(cell, openpyxl.cell.cell.MergedCell):
continue
try:
if cell.value and len(str(cell.value)) > max_length:
max_length = len(str(cell.value))
except Exception:
pass # Non-string cell value — skip width measurement
adjusted_width = min(max_length + 2, 50)
ws.column_dimensions[column_letter].width = adjusted_width
# Save to BytesIO
output = io.BytesIO()
wb.save(output)
output.seek(0)
# Filename
if date_range_str:
filename = f'{project_name_for_filename}time_attendance_{date_range_str}.xlsx'
else:
filename = f'{project_name_for_filename}time_attendance.xlsx'
return send_file(
output,
mimetype='application/vnd.openxmlformats-officedocument.spreadsheetml.sheet',
as_attachment=True,
download_name=filename
)
def export_time_attendance_by_building_excel(records, project_name_for_filename, date_range_str, start_date_filter=None, end_date_filter=None, unlimited=False):
"""Generate Excel export grouped by building/location with template format"""
from openpyxl import Workbook
from openpyxl.styles import Font, PatternFill, Border, Side, Alignment
from openpyxl.utils import get_column_letter
import io
# Create workbook
wb = Workbook()
ws = wb.active
ws.title = "Sheet0"
# Resolve date range; skip 14-day cap when unlimited=True
result = _resolve_date_range(start_date_filter, end_date_filter, records, 'TA by-building Excel export', unlimited=unlimited)
if result is None:
return None
start_date, end_date, records = result
from working_hours_calculator import parse_employee_id_for_work_type
# Convert TimeAttendance records to format expected by calculator
converted_records = _convert_ta_records(records)
# Group records by location (building)
location_groups = {}
for record in converted_records:
loc_name = record.original_location_name or 'Unknown Location'
if loc_name not in location_groups:
location_groups[loc_name] = []
location_groups[loc_name].append(record)
# Sort locations alphabetically
sorted_locations = sorted(location_groups.keys())
# Log grouping info
logger_handler.logger.info(
f"Export by Building: Grouped {len(converted_records)} records into {len(sorted_locations)} locations"
)
# Calculate working hours using WorkingHoursCalculator for SP/PT/PW hours
calculator = WorkingHoursCalculator()
hours_data = calculator.calculate_all_employees_hours(
datetime.combine(start_date, datetime.min.time()),
datetime.combine(end_date, datetime.max.time()),
converted_records
)
# Build employee name map (Lastname, Firstname keyed by base employee ID)
employee_names = _build_employee_name_map(records)
# Setup styles (shared objects)
_styles = _make_export_styles()
header_font = _styles['header_font']
header_fill = _styles['header_fill']
data_font = _styles['data_font']
bold_font = _styles['bold_font']
italic_bold_font = _styles['italic_bold_font']
border = _styles['border']
missed_punch_fill = _styles['missed_punch_fill']
amber_fill = _styles['amber_fill']
border_day_middle = _styles['border_day_middle']
border_day_last = _styles['border_day_last']
# Write main headers
current_row = 1
# Row 1: Company name
ws.merge_cells(f'A{current_row}:N{current_row}')
title_cell = ws.cell(row=current_row, column=1, value=current_app.config.get('COMPANY_NAME', 'QR Code Management System'))
title_cell.font = Font(name='Aptos Narrow', size=14, bold=True)
title_cell.alignment = Alignment(horizontal='left')
current_row += 1
# Row 2: Summary title
ws.merge_cells(f'A{current_row}:N{current_row}')
summary_cell = ws.cell(row=current_row, column=1, value='Summary report of Hours worked')
summary_cell.font = Font(name='Aptos Narrow', size=12, bold=True)
summary_cell.alignment = Alignment(horizontal='left')
current_row += 1
# Row 3: Project name
project_display = project_name_for_filename.replace('_', ' ').strip() if project_name_for_filename else "[Project Name]"
project_cell = ws.cell(row=current_row, column=1, value=project_display)
project_cell.font = Font(name='Aptos Narrow', size=11, bold=True)
project_cell.alignment = Alignment(horizontal='left')
current_row += 1
# Row 4: Date range
date_range_text = f"Date range: {start_date.strftime('%m/%d/%Y')} to {end_date.strftime('%m/%d/%Y')}"
ws.merge_cells(f'A{current_row}:N{current_row}')
date_cell = ws.cell(row=current_row, column=1, value=date_range_text)
date_cell.font = Font(name='Aptos Narrow', size=11)
date_cell.alignment = Alignment(horizontal='left')
current_row += 1
# Empty rows before first building
current_row += 2
# Process each building/location
for location_index, location_name in enumerate(sorted_locations, 1):
location_records = location_groups[location_name]
# Get zone info from QR code if available
zone_info = ''
try:
qr_code = QRCode.query.filter_by(location=location_name).first()
if qr_code:
zone_info = getattr(qr_code, 'zone', '') or ''
except Exception as e:
logger_handler.logger.debug(f"Could not retrieve zone info for location '{location_name}': {e}")
# Building header row
building_header = f"{location_index}) {location_name} - Zone {zone_info}"
ws.merge_cells(f'A{current_row}:O{current_row}')
building_cell = ws.cell(row=current_row, column=1, value=building_header)
building_cell.font = Font(name='Aptos Narrow', size=11, bold=True)
building_cell.alignment = Alignment(horizontal='left')
current_row += 1
# Get unique employees for this location
employees_at_location = {}
for record in location_records:
base_id, _ = parse_employee_id_for_work_type(record.employee_id)
if base_id not in employees_at_location:
employees_at_location[base_id] = []
employees_at_location[base_id].append(record)
# Sort employees by name
sorted_employee_ids = sorted(
employees_at_location.keys(),
key=lambda emp_id: employee_names.get(emp_id, f'Employee {emp_id}').lower()
)
# Process each employee at this location
for employee_id in sorted_employee_ids:
emp_records = employees_at_location[employee_id]
emp_name = employee_names.get(employee_id, f'Employee {employee_id}')
# Compute SP/PW/PT hours from the records already scoped to this
# building and employee (emp_records). Using the calculator's
# grand_totals here would be incorrect: those totals are GLOBAL
# (across all buildings), so an employee with SP hours at Building A
# would incorrectly show an SP row at Building B where they have none.
#
# Strategy: pair same-building SP/PW/PT records the same way the
# main loop pairs regular records, and sum the durations.
def _building_special_hours(emp_recs, work_type_code):
"""Sum paired hours for a given work-type code at this building."""
from datetime import datetime as _dt, timedelta as _td
wt_recs = [r for r in emp_recs if getattr(r, 'work_type', None) == work_type_code]
if not wt_recs:
return 0.0
# Group by date
by_date = {}
for r in wt_recs:
dk = r.check_in_date.strftime('%Y-%m-%d') if hasattr(r.check_in_date, 'strftime') else str(r.check_in_date)
by_date.setdefault(dk, []).append(r)
total = 0.0
for dk, day_recs in by_date.items():
day_recs_s = sorted(day_recs, key=_overnight_aware_sort_key)
ins_r = [r for r in day_recs_s if not ('out' in (r.action_description or '').lower() or 'checkout' in (r.action_description or '').lower())]
outs_r = [r for r in day_recs_s if ('out' in (r.action_description or '').lower() or 'checkout' in (r.action_description or '').lower())]
used = [False] * len(outs_r)
d_obj = _dt.strptime(dk, '%Y-%m-%d')
for in_r in ins_r:
for oi, out_r in enumerate(outs_r):
if used[oi]:
continue
in_dt = _dt.combine(d_obj, in_r.check_in_time)
out_dt = _dt.combine(d_obj, out_r.check_in_time)
if out_dt < in_dt:
out_dt += _td(days=1)
dur = (out_dt - in_dt).total_seconds() / 3600.0
if 0 < dur < 24:
total += dur
used[oi] = True
break
return total
sp_hours = _building_special_hours(emp_records, 'SP')
pw_hours = _building_special_hours(emp_records, 'PW')
pt_hours = _building_special_hours(emp_records, 'PT')
# Employee header row
ws.merge_cells(f'A{current_row}:O{current_row}')
emp_header = ws.cell(row=current_row, column=1,
value=f'Employee ID {employee_id}: {emp_name}')
emp_header.font = Font(name='Aptos Narrow', size=11, bold=True)
emp_header.alignment = Alignment(horizontal='left')
current_row += 1
# Column headers
headers = ['Day', 'Date', 'In', 'Out', 'Location', 'Zone', 'Hours/Building',
'Daily Total', 'Regular Hours', 'OT Hours', 'Building Address',
'Recorded Location', 'Distance (Mile)', 'Possible Violation']
for col, header in enumerate(headers, 1):
cell = ws.cell(row=current_row, column=col, value=header)
cell.font = header_font
cell.fill = header_fill
cell.border = border
cell.alignment = Alignment(horizontal='center', vertical='center')
current_row += 1
# Group employee records by date
daily_records = {}
for record in emp_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)
# -----------------------------------------------------------
# OVERNIGHT SHIFT DETECTION (by-building export)
# The midnight check-out record is stored in the DB on the
# next calendar day's date (e.g. checkout at 01:00 AM on
# Thursday is stored as check_in_date = Thursday). Move it
# into Wednesday's bucket so it pairs with the 8 PM check-in.
#
# Mirrors the identical logic in export_time_attendance_excel.
# -----------------------------------------------------------
def _bb_is_out(r):
a = (r.action_description or '').lower()
return 'out' in a or 'checkout' in a
_bb_sorted_dk = sorted(daily_records.keys())
for _bb_di, _bb_dk in enumerate(_bb_sorted_dk):
if _bb_di + 1 >= len(_bb_sorted_dk):
continue
# Guard: bucket may have been emptied by a prior iteration
if _bb_dk not in daily_records:
continue
_bb_ndk = _bb_sorted_dk[_bb_di + 1]
if _bb_ndk not in daily_records:
continue
# Must be consecutive calendar days
_bb_dn = datetime.strptime(_bb_dk, '%Y-%m-%d').date()
_bb_dn1 = datetime.strptime(_bb_ndk, '%Y-%m-%d').date()
if (_bb_dn1 - _bb_dn).days != 1:
continue
# Collect INs/OUTs for Day N and Day N+1
_bb_day_recs = daily_records[_bb_dk]
_bb_next_recs = daily_records[_bb_ndk]
_bb_day_ins = [r for r in _bb_day_recs if not _bb_is_out(r)]
_bb_day_outs = [r for r in _bb_day_recs if _bb_is_out(r)]
_bb_nxt_ins = [r for r in _bb_next_recs if not _bb_is_out(r)]
_bb_nxt_outs = [r for r in _bb_next_recs if _bb_is_out(r)]
# Exclude early-morning OUTs on Day N from the balance check:
# they are overnight orphans from Day N-1, not Day N regulars.
_bb_day_outs_non_early = [r for r in _bb_day_outs if r.check_in_time.hour > 3]
# Day N must have an unmatched late check-in (>= 12:00 PM)
if len(_bb_day_ins) <= len(_bb_day_outs_non_early):
continue
_bb_late_ins = [r for r in _bb_day_ins if r.check_in_time.hour >= 12]
if not _bb_late_ins:
continue
# Find early-morning OUTs (<= 03:00) on Day N+1
_bb_early_outs = [r for r in _bb_nxt_outs if r.check_in_time.hour <= 3]
if not _bb_early_outs:
continue
# Non-morning INs guard: do NOT move if Day N+1 has a morning
# IN (< 12:00) that precedes the early OUT (i.e. it can own the early OUT)
# and the counts are balanced.
_bb_nxt_non_evening_ins = [
r for r in _bb_nxt_ins
if r.check_in_time.hour < 12
and any(r.check_in_time < eo.check_in_time for eo in _bb_early_outs)
]
if _bb_nxt_non_evening_ins and len(_bb_nxt_outs) <= len(_bb_nxt_ins):
continue
# Move up to as many early OUTs as there are unmatched late INs
_bb_to_move = _bb_early_outs[:len(_bb_late_ins)]
for _bb_co in _bb_to_move:
daily_records[_bb_dk].append(_bb_co)
daily_records[_bb_ndk].remove(_bb_co)
if not daily_records[_bb_ndk]:
del daily_records[_bb_ndk]
logger_handler.logger.info(
f"[TA by-building Export] Overnight: moved checkout "
f"{_bb_co.check_in_time} from {_bb_ndk} to {_bb_dk} "
f"for employee {employee_id} at {location_name}"
)
# -----------------------------------------------------------
# END OVERNIGHT SHIFT DETECTION
# -----------------------------------------------------------
# Track weekly hours for overtime calculation
weekly_total_hours = 0
current_week_start = None
grand_regular_hours = 0
grand_ot_hours = 0
# Accumulate raw regular-only (non-SP/PW/PT) pair hours.
# Used for the "Regular" summary row when the employee also has
# special work-type hours.
regular_only_hours = 0.0
# Sort dates (re-sort after overnight detection may have removed buckets).
# CRITICAL: cap to end_date — daily_records may contain the +1 buffer day
# (fetched so overnight checkout records are available for pairing) but
# that extra day must never be rendered, or it creates a spurious 3rd week.
sorted_dates = sorted(
dk for dk in daily_records.keys()
if datetime.strptime(dk, '%Y-%m-%d').date() <= end_date
)
for date_str in sorted_dates:
date_obj = datetime.strptime(date_str, '%Y-%m-%d')
# Sort records overnight-aware: early-morning OUTs (<=03:00) sort after
# evening records so they pair with the correct evening check-in.
day_records = sorted(daily_records[date_str], key=_overnight_aware_sort_key)
# Check for week boundary anchored to start_date_filter (not calendar Monday)
_report_start = start_date if start_date_filter else date_obj.date()
week_start = (_report_start + timedelta(days=((date_obj.date() - _report_start).days // 7) * 7))
if current_week_start is not None and week_start != current_week_start:
# Write weekly total row
week_regular = min(weekly_total_hours, 40.0)
week_overtime = max(0, weekly_total_hours - 40.0)
ws.cell(row=current_row, column=7, value='Weekly Total: ').font = bold_font
ws.cell(row=current_row, column=8, value=_qtr(weekly_total_hours)).font = bold_font
ws.cell(row=current_row, column=9, value=_qtr(week_regular)).font = bold_font
ws.cell(row=current_row, column=10, value=_qtr(week_overtime)).font = bold_font
grand_regular_hours += week_regular
grand_ot_hours += week_overtime
current_row += 1
weekly_total_hours = 0
current_week_start = week_start
# Re-evaluate miss-punch status after overnight detection may
# have moved a next-day checkout into this day's bucket.
# If INs and OUTs are now balanced, this day is no longer a
# miss punch (mirrors logic in export_time_attendance_excel).
_bb_all_day = day_records
_bb_ins_count = sum(1 for r in _bb_all_day if not _bb_is_out(r))
_bb_outs_count = sum(1 for r in _bb_all_day if _bb_is_out(r))
_bb_day_is_miss_punch = (_bb_ins_count != _bb_outs_count)
# Process day records - create IN/OUT pairs
record_info = []
for record in day_records:
action_desc = record.action_description.lower() if record.action_description else ''
is_out = 'out' in action_desc or 'checkout' in action_desc
record_info.append({
'record': record,
'is_out': is_out,
'used': False
})
# Create pairs
pairs = []
ins = [ri for ri in record_info if not ri['is_out']]
outs = [ri for ri in record_info if ri['is_out']]
if len(ins) > len(outs) and len(outs) > 0:
# Odd-IN rule: discard all but the LATEST IN; pair it with the earliest OUT.
# Use overnight-aware sort so early-morning OUTs sort after evening OUTs.
ins_sorted = sorted(ins, key=lambda ri: _overnight_aware_sort_key(ri['record']))
outs_sorted = sorted(outs, key=lambda ri: _overnight_aware_sort_key(ri['record']))
latest_in = ins_sorted[-1]
excess_ins = ins_sorted[:-1]
# Orphan guard: when the latest IN is afternoon/evening (>=12h), skip
# early-morning OUTs (<=3h) whose check_in_date matches the
# current day — they are orphans from a previous overnight shift.
_oi_in_hour = latest_in['record'].check_in_time.hour
earliest_out = None
_oi_skip = []
for _oi_ri in outs_sorted:
if (earliest_out is None
and _oi_in_hour >= 12
and _oi_ri['record'].check_in_time.hour <= 3):
_oi_out_d = _oi_ri['record'].check_in_date
if hasattr(_oi_out_d, 'date'):
_oi_out_d = _oi_out_d.date()
if _oi_out_d <= date_obj.date():
_oi_skip.append(_oi_ri)
continue
if earliest_out is None:
earliest_out = _oi_ri
break
for ri in excess_ins:
ri['used'] = True
pairs.append({'check_in': ri['record'], 'check_out': None, 'is_miss_punch': True})
if earliest_out is not None:
latest_in['used'] = True
earliest_out['used'] = True
pairs.append({'check_in': latest_in['record'], 'check_out': earliest_out['record'], 'is_miss_punch': False})
else:
latest_in['used'] = True
pairs.append({'check_in': latest_in['record'], 'check_out': None, 'is_miss_punch': True})
for ri in outs_sorted:
if not ri['used'] and ri not in _oi_skip:
ri['used'] = True
pairs.append({'check_in': None, 'check_out': ri['record'], 'is_miss_punch': True})
# Orphan OUTs that were skipped
for ri in _oi_skip:
ri['used'] = True
pairs.append({'check_in': None, 'check_out': ri['record'], 'is_miss_punch': True})
else:
# Standard pairing
i = 0
while i < len(record_info):
if record_info[i]['used']:
i += 1
continue
if not record_info[i]['is_out']: # IN
out_found = False
for j in range(i + 1, len(record_info)):
if record_info[j]['used']:
continue
if record_info[j]['is_out']:
# Orphan guard: when this IN is an afternoon/evening
# check-in (>=12h) and the candidate OUT is
# early-morning (<=3h), the OUT is only a
# valid partner if it was moved in by overnight
# detection (check_in_date > current day).
# Same-day early-morning OUTs are orphans from
# a previous overnight shift.
_in_rec = record_info[i]['record']
_out_rec = record_info[j]['record']
if (_in_rec.check_in_time.hour >= 12
and _out_rec.check_in_time.hour <= 3):
_out_bb_date = _out_rec.check_in_date
if hasattr(_out_bb_date, 'date'):
_out_bb_date = _out_bb_date.date()
if _out_bb_date <= date_obj.date():
continue # orphan — skip
pairs.append({
'check_in': record_info[i]['record'],
'check_out': record_info[j]['record'],
'is_miss_punch': False
})
record_info[i]['used'] = True
record_info[j]['used'] = True
out_found = True
break
if not out_found:
pairs.append({
'check_in': record_info[i]['record'],
'check_out': None,
'is_miss_punch': True
})
record_info[i]['used'] = True
else: # Orphaned OUT
pairs.append({
'check_in': None,
'check_out': record_info[i]['record'],
'is_miss_punch': True
})
record_info[i]['used'] = True
i += 1
# Calculate daily hours
daily_hours = 0
for pair in pairs:
if pair['check_in'] and pair['check_out'] and not pair['is_miss_punch']:
pair_in = datetime.combine(date_obj, pair['check_in'].check_in_time)
pair_out = datetime.combine(date_obj, pair['check_out'].check_in_time)
# Overnight shift correction: if OUT is before IN on the same
# calendar date, the employee worked past midnight — advance
# pair_out by one day so the duration is always positive.
if pair_out < pair_in:
pair_out += timedelta(days=1)
_bb_dur = (pair_out - pair_in).total_seconds() / 3600.0
# 24h guard: reject implausible durations (data errors)
if _bb_dur <= 24:
daily_hours += _bb_dur
# Accumulate regular-only hours: pairs where neither record
# carries a special work type (SP/PW/PT).
_bb_in_wt = getattr(pair['check_in'], 'work_type', None)
_bb_out_wt = getattr(pair['check_out'], 'work_type', None)
_bb_eff_wt = _bb_out_wt or _bb_in_wt # prefer OUT's type (mirrors main export)
if _bb_eff_wt not in ('SP', 'PW', 'PT'):
regular_only_hours += _bb_dur
daily_hours = _qtr(daily_hours)
weekly_total_hours += daily_hours
# Write pairs
for pair_idx, pair in enumerate(pairs):
check_in = pair['check_in']
check_out = pair['check_out']
is_miss_punch = pair['is_miss_punch']
# Day/date only on first row
day_display = date_obj.strftime('%A').upper() if pair_idx == 0 else ''
date_display = date_obj.strftime('%m/%d/%Y') if pair_idx == 0 else ''
# Calculate hours for this pair
if check_in and check_out and not is_miss_punch:
_pair_in_dt = datetime.combine(date_obj, check_in.check_in_time)
_pair_out_dt = datetime.combine(date_obj, check_out.check_in_time)
# Overnight shift correction: advance OUT by one day when it
# falls before IN (employee crossed midnight).
if _pair_out_dt < _pair_in_dt:
_pair_out_dt += timedelta(days=1)
pair_hours = round((_pair_out_dt - _pair_in_dt).total_seconds() / 3600.0, 2)
else:
pair_hours = 'Missed Punch'
# Daily total only on last row of day
daily_total_display = daily_hours if pair_idx == len(pairs) - 1 else ''
# Get record for address/distance info
ref_record = check_in or check_out
# Build row data
row_data = [
day_display,
date_display,
check_in.check_in_time.strftime('%I:%M:%S %p') if check_in else '',
check_out.check_in_time.strftime('%I:%M:%S %p') if check_out else '',
ref_record.location_name if ref_record else '',
zone_info,
pair_hours,
daily_total_display if daily_total_display else '',
'', # Regular Hours
'', # OT Hours
'', # Building Address (will be HYPERLINK)
'', # Recorded Location (will be HYPERLINK)
getattr(ref_record, 'distance', None) or '' if ref_record else '',
calculate_possible_violation(getattr(ref_record, 'distance', None)) if ref_record else ''
]
# Use bottom-only border on the last pair row of the day;
# no borders on intermediate rows (matches normal TA export).
_bb_is_last_pair = (pair_idx == len(pairs) - 1)
_bb_row_border = border_day_last if _bb_is_last_pair else border_day_middle
for col, value in enumerate(row_data, 1):
cell = ws.cell(row=current_row, column=col, value=value)
cell.font = data_font
cell.border = _bb_row_border
if col == 7 and value == 'Missed Punch':
cell.fill = missed_punch_fill
# Add HYPERLINK formulas for addresses
if ref_record:
building_address = ref_record.event_description or ''
if building_address:
encoded_addr = building_address.replace(' ', '+').replace(',', '%2C')
hyperlink_formula = f'=HYPERLINK("https://www.google.com/maps/place/{encoded_addr}","{building_address}")'
ws.cell(row=current_row, column=11, value=hyperlink_formula)
recorded_addr = ref_record.recorded_address or ''
if recorded_addr:
encoded_recorded = recorded_addr.replace(' ', '+').replace(',', '%2C')
recorded_hyperlink = f'=HYPERLINK("https://www.google.com/maps/place/{encoded_recorded}","{recorded_addr}")'
ws.cell(row=current_row, column=12, value=recorded_hyperlink)
current_row += 1
# Write final weekly total
if weekly_total_hours > 0:
week_regular = min(weekly_total_hours, 40.0)
week_overtime = max(0, weekly_total_hours - 40.0)
ws.cell(row=current_row, column=7, value='Weekly Total: ').font = bold_font
ws.cell(row=current_row, column=8, value=_qtr(weekly_total_hours)).font = bold_font
ws.cell(row=current_row, column=9, value=_qtr(week_regular)).font = bold_font
ws.cell(row=current_row, column=10, value=_qtr(week_overtime)).font = bold_font
grand_regular_hours += week_regular
grand_ot_hours += week_overtime
current_row += 1
# ================================================================
# Write extra working hours rows (SP/PW/PT) if employee has any.
# Abbreviated label in col 8, hours in col 9.
# A "Regular" row follows whenever at least one special type is present.
# ================================================================
_has_special_hours = sp_hours > 0 or pw_hours > 0 or pt_hours > 0
# Write SP row if hours > 0
if sp_hours > 0:
ws.cell(row=current_row, column=8, value='SP').font = italic_bold_font
ws.cell(row=current_row, column=9, value=_qtr(sp_hours)).font = italic_bold_font
logger_handler.logger.info(f"Export by Building: Employee {employee_id} SP hours: {sp_hours:.2f}")
current_row += 1
# Write PW row if hours > 0
if pw_hours > 0:
ws.cell(row=current_row, column=8, value='PW').font = italic_bold_font
ws.cell(row=current_row, column=9, value=_qtr(pw_hours)).font = italic_bold_font
logger_handler.logger.info(f"Export by Building: Employee {employee_id} PW hours: {pw_hours:.2f}")
current_row += 1
# Write PT row if hours > 0
if pt_hours > 0:
ws.cell(row=current_row, column=8, value='PT').font = italic_bold_font
ws.cell(row=current_row, column=9, value=_qtr(pt_hours)).font = italic_bold_font
logger_handler.logger.info(f"Export by Building: Employee {employee_id} PT hours: {pt_hours:.2f}")
current_row += 1
# Write Regular row — only when the employee has special work-type hours.
if _has_special_hours:
ws.cell(row=current_row, column=8, value='Regular').font = italic_bold_font
ws.cell(row=current_row, column=9, value=_qtr(regular_only_hours)).font = italic_bold_font
logger_handler.logger.info(f"Export by Building: Employee {employee_id} Regular hours: {regular_only_hours:.2f}")
current_row += 1
# ================================================================
# End of extra working hours section
# ================================================================
# Write GRAND TOTAL row
ws.cell(row=current_row, column=7, value='GRAND TOTAL: ').font = Font(name='Aptos Narrow', size=11, bold=True)
ws.cell(row=current_row, column=9, value=_qtr(grand_regular_hours)).font = Font(name='Aptos Narrow', size=11, bold=True)
ws.cell(row=current_row, column=10, value=_qtr(grand_ot_hours)).font = Font(name='Aptos Narrow', size=11, bold=True)
current_row += 1
# Empty row after each employee
current_row += 1
# Empty row after each building
current_row += 1
# Auto-size columns
for col_idx in range(1, 15):
column_letter = get_column_letter(col_idx)
if col_idx == 1:
ws.column_dimensions[column_letter].width = 18
continue
max_length = 0
for row in ws.iter_rows(min_col=col_idx, max_col=col_idx):
for cell in row:
if isinstance(cell, openpyxl.cell.cell.MergedCell):
continue
try:
if cell.value and len(str(cell.value)) > max_length:
max_length = len(str(cell.value))
except Exception:
pass # Non-string cell value — skip width measurement
adjusted_width = min(max_length + 2, 50)
ws.column_dimensions[column_letter].width = adjusted_width
# Save to BytesIO
output = io.BytesIO()
wb.save(output)
output.seek(0)
# Filename
if date_range_str:
filename = f'{project_name_for_filename}time_attendance_by_building_{date_range_str}.xlsx'
else:
filename = f'{project_name_for_filename}time_attendance_by_building.xlsx'
# Log successful export
logger_handler.logger.info(
f"Export by Building completed: {filename} with {len(sorted_locations)} buildings"
)
return send_file(
output,
mimetype='application/vnd.openxmlformats-officedocument.spreadsheetml.sheet',
as_attachment=True,
download_name=filename
)
Reference in New Issue View Git Blame Copy Permalink