# routes/attendance_summary.py import calendar from calendar import monthrange from flask import Blueprint, render_template, request, redirect, url_for, flash, make_response, current_app from flask_login import login_required,current_user from datetime import datetime, date, timedelta, time from sqlalchemy import and_, or_, func from sqlalchemy.orm import joinedload from sqlalchemy.exc import IntegrityError from extensions import db from models import ( Attendance, Employee, Payroll, WeekendSettings, Holiday, LeaveRequest, Department, DesktimeLog, SyncLog, BiometricPunch, SystemSetting,EmployeeInstituteAccess, ) attendance_summary_bp = Blueprint('attendance_summary', __name__) def _parse_hhmm(v: str): v = (v or "").strip() if not v: return None # supports "HH:MM" return datetime.strptime(v, "%H:%M").time() def _is_admin_or_hr_user(u) -> bool: rn = (getattr(getattr(u, "role", None), "name", "") or "").strip().lower() return rn in ("admin", "hr","hr admin") def _allowed_institutes_for(user_id: int) -> list[str]: rows = ( db.session.query(func.trim(EmployeeInstituteAccess.institute)) .filter(EmployeeInstituteAccess.viewer_employee_id == user_id) .all() ) return sorted({(r[0] or "").strip() for r in rows if r and (r[0] or "").strip()}) from typing import Optional, List def _visible_employee_ids_for(user) -> Optional[List[int]]: """ Returns: - None => can view ALL (admin/hr) - [ids...] => can view only these employees Rule: - if mapped institutes exist => employees in those institutes - else => only self """ if _is_admin_or_hr_user(user): return None # ✅ unlimited allowed = _allowed_institutes_for(user.id) if not allowed: return [user.id] # ✅ only self ids = [ r[0] for r in db.session.query(Employee.id) .filter(Employee.status == "active") .filter(func.trim(Employee.location).in_(allowed)) .all() ] # always include self even if institute blank if user.id not in ids: ids.append(user.id) return ids # --------------------------- # Monthly Attendance Summary # --------------------------- @attendance_summary_bp.route('/attendance/monthly-summary', methods=['GET']) @login_required def monthly_attendance_summary(): selected_month = request.args.get('month') or datetime.today().strftime('%Y-%m') employee_name = request.args.get('employee_name', '') try: month_year = datetime.strptime(selected_month, '%Y-%m') except ValueError: flash('Invalid month format', 'danger') return redirect(request.referrer) year = month_year.year month = month_year.month last_day = calendar.monthrange(year, month)[1] start_date = date(year, month, 1) end_date = date(year, month, last_day) visible_ids = _visible_employee_ids_for(current_user) # ✅ add employees_q = ( Employee.query .filter(Employee.date_of_joining.isnot(None)) .filter(Employee.date_of_joining <= end_date) .filter(or_(Employee.exit_date.is_(None), Employee.exit_date >= start_date)) ) # ✅ institute-based visibility if visible_ids is not None: employees_q = employees_q.filter(Employee.id.in_(visible_ids)) if employee_name: employees_q = employees_q.filter(Employee.full_name.ilike(f"%{employee_name}%")) employees = employees_q.order_by(Employee.full_name).all() frozen_employee_ids = { p.employee_id for p in Payroll.query.filter_by(month=selected_month).all() } emp_ids = [e.id for e in employees] # Attendance preload att_rows = [] if emp_ids: att_rows = ( Attendance.query .filter(Attendance.employee_id.in_(emp_ids)) .filter(and_(Attendance.date >= start_date, Attendance.date <= end_date)) .all() ) att_map = {(a.employee_id, a.date): a for a in att_rows} # ✅ Holiday preload for UI override holiday_dates = set() hol_rows = ( Holiday.query .filter(Holiday.date >= start_date, Holiday.date <= end_date) .all() ) for h in hol_rows: holiday_dates.add(h.date) # ✅ Weekend settings + weekend dates preload for UI override weekend_cfg = WeekendSettings.query.order_by(WeekendSettings.updated_on.desc()).first() def _norm(x): return (x or "").strip().lower() def saturday_mode(settings, day): """'off' | 'half-day' | 'working' """ if not settings: return 'working' week_number = (day.day - 1) // 7 + 1 key = ["first", "second", "third", "fourth", "fifth"][week_number - 1] raw = getattr(settings, f"{key}_saturday", None) v = _norm(raw) if v in ('off', 'holiday', 'weekend'): return 'off' if v in ('half-day', 'halfday', 'half'): return 'half-day' return 'working' def is_sunday_off(settings): return _norm(getattr(settings, "sunday", None)) in ('off', 'holiday', 'weekend') def is_weekend(d): if not weekend_cfg: return False # Sunday if d.weekday() == 6 and is_sunday_off(weekend_cfg): return True # Saturday off if d.weekday() == 5 and saturday_mode(weekend_cfg, d) == 'off': return True return False weekend_dates = set() for day in range(1, last_day + 1): d = date(year, month, day) if is_weekend(d): weekend_dates.add(d) # Half-day approved leaves for the month leave_rows = [] if emp_ids: leave_rows = ( LeaveRequest.query .filter(LeaveRequest.employee_id.in_(emp_ids)) .filter(LeaveRequest.status == 'final_approved') .filter(LeaveRequest.half_day == True) .filter(LeaveRequest.start_date <= end_date, LeaveRequest.end_date >= start_date) .all() ) # Convention: if your system doesn’t store session (FH/SH), we allow UI override via Attendance.note. # Default: First Half Leave. half_leave_map = {} # (emp_id, date) -> {"session": "FH"/"SH"} for lr in leave_rows: d0 = max(lr.start_date, start_date) d1 = min(lr.end_date, end_date) cur = d0 while cur <= d1: half_leave_map[(lr.employee_id, cur)] = {"session": "FH"} cur += timedelta(days=1) summary_data = [] for emp in employees: emp_data = {'employee': emp, 'days': []} for day in range(1, last_day + 1): date_obj = date(year, month, day) attendance = att_map.get((emp.id, date_obj)) hl = half_leave_map.get((emp.id, date_obj)) if attendance: src = (attendance.source or "").strip().lower() n = (attendance.note or "").strip() # manual/bulk must override approved leave unless explicitly saved as split leave if src in ("manual", "bulk"): hl = None if "HD_LEAVE:SH" in n: hl = {"session": "SH"} elif "HD_LEAVE:FH" in n: hl = {"session": "FH"} emp_data['days'].append({ 'date': date_obj, 'record': attendance, 'half_leave': hl, # None or {"session":"FH"/"SH"} }) emp_data['payroll_frozen'] = emp.id in frozen_employee_ids summary_data.append(emp_data) return render_template( 'attendance/attendance_monthly_summary.html', summary_data=summary_data, selected_month=selected_month, employee_name=employee_name, weekend_dates=weekend_dates, # ✅ added holiday_dates=holiday_dates, # ✅ added ) # --------------------------- # Manual Update Attendance # --------------------------- @attendance_summary_bp.route('/attendance/manual-update', methods=['POST']) @login_required def manual_update_attendance(): employee_id = request.form.get('employee_id', type=int) date_str = request.form.get('date') remarks = (request.form.get('remarks') or '').strip() note = (request.form.get('note') or '').strip() # new fields punch_in = request.form.get('check_in_time') # "HH:MM" punch_out = request.form.get('check_out_time') # "HH:MM" half_leave_session = (request.form.get('half_leave_session') or '').strip() # FH/SH/'' second_half_status = (request.form.get('second_half_status') or '').strip() # Present/LWP/etc. try: att_date = datetime.strptime(date_str, '%Y-%m-%d').date() except ValueError: flash('Invalid date format', 'danger') return redirect(request.referrer) visible_ids = _visible_employee_ids_for(current_user) if visible_ids is not None and employee_id not in visible_ids: flash("❌ You are not allowed to update attendance for this employee.", "danger") return redirect(request.referrer) # payroll freeze check month_str = att_date.strftime('%Y-%m') if Payroll.query.filter_by(employee_id=employee_id, month=month_str).first(): flash('❌ Cannot update. Payroll frozen for this employee.', 'danger') return redirect(request.referrer) attendance = Attendance.query.filter_by(employee_id=employee_id, date=att_date).first() if not attendance: attendance = Attendance(employee_id=employee_id, date=att_date) db.session.add(attendance) # If user selected Half-Day Leave (FH/SH), store it in note so UI can show split reliably # (without changing schema) # We keep remarks as "Half Day" so payroll logic remains safe. if remarks in ("HD Leave", "Half Day Leave"): attendance.remarks = "Half Day" # explicit manual split leave tag = f"HD_LEAVE:{half_leave_session or 'FH'}" other_half = f"SH:{second_half_status}" if second_half_status else "SH:" attendance.note = " | ".join([tag, other_half, note]).strip(" | ") else: # manual normal update must override any previous split/leave tag attendance.remarks = remarks clean_note = note or "" if "HD_LEAVE:" in clean_note: clean_note = "" attendance.note = clean_note attendance.source = 'Manual' # Save punch times if provided ci = _parse_hhmm(punch_in) co = _parse_hhmm(punch_out) attendance.check_in_time = ci attendance.check_out_time = co if attendance.check_in_time and attendance.check_out_time: dt_in = datetime.combine(att_date, attendance.check_in_time) dt_out = datetime.combine(att_date, attendance.check_out_time) if dt_out >= dt_in: attendance.working_hours = round((dt_out - dt_in).total_seconds() / 3600, 2) else: attendance.working_hours = None else: attendance.working_hours = None db.session.commit() flash('✅ Attendance updated.', 'success') return redirect(request.referrer) # --------------------------- # Bulk Month Update # --------------------------- @attendance_summary_bp.route('/attendance/bulk-month-update', methods=['POST']) @login_required def bulk_month_update(): """ Bulk mark attendance for selected employees for a month. Skips employees whose payroll is frozen for that month. Supports skip weekends + skip holidays. """ month = request.form.get('month') # "YYYY-MM" status = (request.form.get('status') or '').strip() # Present/Half Day/Leave/Holiday/Weekend/LWP/blank note = (request.form.get('note') or '').strip() skip_weekends = request.form.get('skip_weekends') == 'on' skip_holidays = request.form.get('skip_holidays') == 'on' employee_ids = request.form.getlist('employee_ids') # list[str] # validate try: y, m = map(int, (month or "").split('-')) except Exception: flash('Invalid month.', 'danger') return redirect(request.referrer) if not employee_ids: flash('Select at least one employee.', 'warning') return redirect(request.referrer) # month range last_day = monthrange(y, m)[1] start_date = date(y, m, 1) end_date = date(y, m, last_day) # payroll freeze map frozen_ids = {p.employee_id for p in Payroll.query.filter_by(month=month).all()} # preload holidays for the month (FAST: no per-day query) holiday_dates = { h.date for h in Holiday.query .filter(Holiday.date >= start_date, Holiday.date <= end_date) .all() } # weekend config weekend_cfg = WeekendSettings.query.order_by(WeekendSettings.updated_on.desc()).first() def _norm(x): return (x or "").strip().lower() def saturday_mode(settings, day): """'off' | 'half-day' | 'working' """ if not settings: return 'working' week_number = (day.day - 1) // 7 + 1 key = ["first", "second", "third", "fourth", "fifth"][week_number - 1] raw = getattr(settings, f"{key}_saturday", None) v = _norm(raw) if v in ('off', 'holiday', 'weekend'): return 'off' if v in ('half-day', 'halfday', 'half'): return 'half-day' return 'working' def is_sunday_off(settings): return _norm(getattr(settings, "sunday", None)) in ('off', 'holiday', 'weekend') def is_weekend(d): if not weekend_cfg: return False if d.weekday() == 6 and is_sunday_off(weekend_cfg): return True if d.weekday() == 5 and saturday_mode(weekend_cfg, d) == 'off': return True return False updated = 0 skipped_frozen = 0 # Convert IDs safely clean_ids = [] for s in employee_ids: try: clean_ids.append(int(s)) except Exception: pass visible_ids = _visible_employee_ids_for(current_user) if visible_ids is not None: allowed_set = set(visible_ids) clean_ids = [eid for eid in clean_ids if eid in allowed_set] if not clean_ids: flash("No eligible employees selected (you don't have access).", "danger") return redirect(request.referrer) # preload employees in one go employees = {e.id: e for e in Employee.query.filter(Employee.id.in_(clean_ids)).all()} for eid in clean_ids: if eid in frozen_ids: skipped_frozen += 1 continue emp = employees.get(eid) if not emp: continue d = start_date while d <= end_date: # skip pre-join days if emp.date_of_joining and d < emp.date_of_joining: d += timedelta(days=1) continue # skip weekends/holidays if skip_weekends and is_weekend(d): d += timedelta(days=1) continue if skip_holidays and d in holiday_dates: d += timedelta(days=1) continue att = Attendance.query.filter_by(employee_id=eid, date=d).first() if not att: att = Attendance(employee_id=eid, date=d) db.session.add(att) # ✅ DO NOT OVERWRITE manual updates (including manual Leave) if (att.source or "").strip().lower() == "manual": d += timedelta(days=1) continue att.remarks = status # allow '' to clear att.source = 'Bulk' if hasattr(att, 'note'): clean_note = note or "" if "HD_LEAVE:" in clean_note: clean_note = "" att.note = clean_note updated += 1 d += timedelta(days=1) db.session.commit() msg = f"✅ Bulk update done. {updated} day(s) updated." if skipped_frozen: msg += f" Skipped {skipped_frozen} employee(s) due to payroll freeze." flash(msg, 'success') return redirect(request.referrer or url_for('attendance_summary.monthly_attendance_summary')) # --------------------------- # Attendance Debug Source # --------------------------- @attendance_summary_bp.route("/attendance/debug-source", methods=["GET", "POST"]) @login_required def attendance_debug_source(): """ Debug helper: for a given employee + date, show which source was used and what data is available from Manual / Biometric / Self / Desktime. """ visible_ids = _visible_employee_ids_for(current_user) employees_q = Employee.query.order_by(Employee.full_name) if visible_ids is not None: employees_q = employees_q.filter(Employee.id.in_(visible_ids)) employees = employees_q.all() # defaults selected_emp_id = request.values.get("employee_id", type=int) date_str = request.values.get("date") or date.today().strftime("%Y-%m-%d") selected_date = None debug_info = None priority_list = [] # load configured priority setting = SystemSetting.query.filter_by(key="attendance_source_priority").first() if setting and setting.json_value: try: import json priority_list = json.loads(setting.json_value) except Exception: priority_list = [] if not priority_list: priority_list = ["manual", "biometric", "self", "desktime"] if selected_emp_id and date_str: try: selected_date = datetime.strptime(date_str, "%Y-%m-%d").date() except ValueError: flash("Invalid date format.", "danger") return redirect(url_for("attendance_summary.attendance_debug_source")) if visible_ids is not None and selected_emp_id not in set(visible_ids): flash("Not allowed to view debug for this employee.", "danger") return redirect(url_for("attendance_summary.attendance_debug_source")) emp = Employee.query.get(selected_emp_id) if not emp: flash("Employee not found.", "danger") return redirect(url_for("attendance_summary.attendance_debug_source")) # --- Attendance row (final result) --- att = Attendance.query.filter_by( employee_id=emp.id, date=selected_date ).first() # --- Biometric punches (including self) --- start_dt = datetime.combine(selected_date, datetime.min.time()) end_dt = datetime.combine(selected_date + timedelta(days=1), datetime.min.time()) biometric_q = BiometricPunch.query.filter( BiometricPunch.employee_id == emp.id, BiometricPunch.punch_time >= start_dt, BiometricPunch.punch_time < end_dt, ).order_by(BiometricPunch.punch_time.asc()) biometric_all = biometric_q.all() biometric_hw = [p for p in biometric_all if p.device_code != "WEB"] self_punches = [p for p in biometric_all if p.device_code == "WEB"] # --- Desktime log --- desk_log = DesktimeLog.query.filter_by( employee_id=emp.id, date=selected_date ).first() # derived flags has_manual = att is not None and (att.source or "").lower() in ("manual", "bulk") has_biometric = bool(biometric_hw) has_self = bool(self_punches) has_desktime = desk_log is not None chosen_source = (att.source if att and att.source else None) debug_info = { "employee": emp, "date": selected_date, "attendance": att, "priority_list": priority_list, "chosen_source": chosen_source, "sources": { "manual": {"has_data": has_manual}, "biometric": {"has_data": has_biometric, "punches": biometric_hw}, "self": {"has_data": has_self, "punches": self_punches}, "desktime": {"has_data": has_desktime, "log": desk_log}, }, } return render_template( "attendance/attendance_debug_source.html", employees=employees, selected_emp_id=selected_emp_id, date_str=date_str, debug_info=debug_info, priority_list=priority_list, ) # --------------------------- # Late Report Summary # --------------------------- def _time_to_minutes(t: time) -> int: return t.hour * 60 + t.minute @attendance_summary_bp.route("/attendance/late-summary", methods=["GET"]) @login_required def late_report_summary(): """ Employee-wise late report: Late = check_in_time - (shift_start + grace_minutes) ✅ Additional behavior: - For TODAY: if Attendance.check_in_time is missing, show earliest BiometricPunch as check_in_time (fallback). """ # ----------------------- # Filters # ----------------------- month = (request.args.get("month") or datetime.today().strftime("%Y-%m")).strip() from_date = (request.args.get("from_date") or "").strip() to_date = (request.args.get("to_date") or "").strip() q = (request.args.get("q") or "").strip() employee_id = request.args.get("employee_id") or "" department_id = request.args.get("department_id") or "" location = (request.args.get("location") or "").strip() manager_id = request.args.get("manager_id") or "" work_mode = (request.args.get("work_mode") or "").strip() status = (request.args.get("status") or "").strip() # present/absent/paid_leave/lop... source = (request.args.get("source") or "").strip() # Punch/Desktime/Manual... late_only = (request.args.get("late_only") == "1") grace_minutes = request.args.get("grace_minutes") or "0" min_late_minutes = request.args.get("min_late_minutes") or "1" page = int(request.args.get("page") or 1) per_page = int(request.args.get("per_page") or 50) try: grace_minutes = max(0, int(grace_minutes)) except Exception: grace_minutes = 0 try: min_late_minutes = max(0, int(min_late_minutes)) except Exception: min_late_minutes = 1 # ----------------------- # Date range # Priority: from/to else month # ----------------------- start_date = None end_date = None if from_date and to_date: try: start_date = datetime.strptime(from_date, "%Y-%m-%d").date() end_date = datetime.strptime(to_date, "%Y-%m-%d").date() except Exception: flash("Invalid From/To date format. Use YYYY-MM-DD.", "danger") return redirect(url_for("attendance_summary.late_report_summary")) if not start_date or not end_date: try: dt = datetime.strptime(month, "%Y-%m") y, m = dt.year, dt.month last_day = monthrange(y, m)[1] start_date = date(y, m, 1) end_date = date(y, m, last_day) except Exception: flash("Invalid month format. Use YYYY-MM.", "danger") return redirect(url_for("attendance_summary.late_report_summary")) # ----------------------- # Base query # ----------------------- qry = ( Attendance.query .join(Employee, Employee.id == Attendance.employee_id) .options(joinedload(Attendance.employee)) .filter(Attendance.date >= start_date, Attendance.date <= end_date) .filter(Employee.status == "active") .filter(or_(Employee.exit_date.is_(None), Employee.exit_date >= start_date)) ) visible_ids = _visible_employee_ids_for(current_user) if visible_ids is not None: qry = qry.filter(Employee.id.in_(visible_ids)) # Exclude pre-joining dates (only if doj exists) qry = qry.filter(or_(Employee.date_of_joining.is_(None), Employee.date_of_joining <= Attendance.date)) if q: like = f"%{q}%" qry = qry.filter(or_( Employee.full_name.ilike(like), Employee.employee_code.ilike(like), Employee.email.ilike(like) )) if employee_id: try: qry = qry.filter(Employee.id == int(employee_id)) except Exception: pass if department_id: try: qry = qry.filter(Employee.department_id == int(department_id)) except Exception: pass if location: qry = qry.filter(func.lower(Employee.location) == location.lower()) if manager_id: try: qry = qry.filter(Employee.reporting_manager_id == int(manager_id)) except Exception: pass if work_mode: qry = qry.filter(Employee.work_mode == work_mode) if status: qry = qry.filter(Attendance.status == status) if source: qry = qry.filter(Attendance.source == source) # Order qry = qry.order_by(Attendance.date.desc(), Employee.full_name.asc()) # ----------------------- # Pagination (DB) # ----------------------- pagination = qry.paginate(page=page, per_page=per_page, error_out=False) records = pagination.items # ----------------------- # TODAY fallback preload (avoid per-row query) # ----------------------- today = date.today() today_punch_map = {} today_emp_ids = [ att.employee_id for att in records if att.date == today and not att.check_in_time ] if today_emp_ids: day_start = datetime.combine(today, datetime.min.time()) day_end = datetime.combine(today + timedelta(days=1), datetime.min.time()) punches = ( BiometricPunch.query .filter(BiometricPunch.employee_id.in_(today_emp_ids)) .filter(BiometricPunch.punch_time >= day_start, BiometricPunch.punch_time < day_end) .order_by(BiometricPunch.employee_id.asc(), BiometricPunch.punch_time.asc()) .all() ) # first punch per employee for p in punches: if p.employee_id not in today_punch_map: today_punch_map[p.employee_id] = p.punch_time.time() # ----------------------- # Compute late minutes (Python) # ----------------------- rows = [] late_count = 0 late_minutes_sum = 0 for att in records: emp = att.employee shift_start = getattr(emp, "shift_start", None) shift_end = getattr(emp, "shift_end", None) # ✅ TODAY fallback: show punch-in if Attendance.check_in_time missing if att.date == today and not att.check_in_time: fallback_ci = today_punch_map.get(emp.id) if fallback_ci: att.check_in_time = fallback_ci late_minutes = None is_late = False if shift_start and att.check_in_time: base = _time_to_minutes(shift_start) + grace_minutes cin = _time_to_minutes(att.check_in_time) diff = cin - base late_minutes = max(0, diff) is_late = late_minutes >= min_late_minutes # late_only filter happens after computing if late_only and not is_late: continue if is_late: late_count += 1 late_minutes_sum += (late_minutes or 0) rows.append({ "attendance": att, "employee": emp, "shift_start": shift_start, "shift_end": shift_end, "late_minutes": late_minutes, "is_late": is_late, }) avg_late = round(late_minutes_sum / late_count, 2) if late_count else 0 # Dropdown data employees_q = Employee.query.filter_by(status="active") managers_q = Employee.query.filter_by(status="active") if visible_ids is not None: employees_q = employees_q.filter(Employee.id.in_(visible_ids)) managers_q = managers_q.filter(Employee.id.in_(visible_ids)) employees = employees_q.order_by(Employee.full_name.asc()).all() managers = managers_q.order_by(Employee.full_name.asc()).all() departments = Department.query.order_by(Department.name.asc()).all() locations = [ r[0] for r in db.session.query(Employee.location) .filter(Employee.location.isnot(None)) .distinct().order_by(Employee.location.asc()).all() if r and r[0] ] work_modes = ["Hybrid", "WFO", "WFH"] statuses = ["present", "absent", "paid_leave", "lop", "half_day"] sources = ["Punch", "Desktime", "Manual", "Self", "Biometric"] return render_template( "attendance/late_report_summary.html", # filters month=month, from_date=from_date, to_date=to_date, q=q, employee_id=employee_id, department_id=department_id, location=location, manager_id=manager_id, work_mode=work_mode, status=status, source=source, late_only=late_only, grace_minutes=grace_minutes, min_late_minutes=min_late_minutes, page=page, per_page=per_page, # data rows=rows, pagination=pagination, total_in_page=len(rows), late_count=late_count, avg_late=avg_late, start_date=start_date, end_date=end_date, # dropdowns employees=employees, departments=departments, managers=managers, locations=locations, work_modes=work_modes, statuses=statuses, sources=sources, ) @attendance_summary_bp.route("/attendance/late-summary/pdf", methods=["GET"]) @login_required def late_report_summary_pdf(): """ PDF export for filtered late report. ✅ Includes same filters as late_report_summary() ✅ TODAY fallback: - if Attendance.check_in_time missing -> earliest BiometricPunch - if Attendance.check_out_time missing -> latest BiometricPunch ✅ PDF columns include: Shift End, Check Out, Punch Approved By ✅ Late mins shown as HH:MM """ import io from typing import Optional, Dict, Tuple from reportlab.lib.pagesizes import A4, landscape from reportlab.lib import colors from reportlab.lib.styles import getSampleStyleSheet from reportlab.platypus import SimpleDocTemplate, Table, TableStyle, Paragraph, Spacer # ----------------------- # Filters (same as late_report_summary) # ----------------------- month = (request.args.get("month") or datetime.today().strftime("%Y-%m")).strip() from_date = (request.args.get("from_date") or "").strip() to_date = (request.args.get("to_date") or "").strip() q = (request.args.get("q") or "").strip() employee_id = request.args.get("employee_id") or "" department_id = request.args.get("department_id") or "" location = (request.args.get("location") or "").strip() manager_id = request.args.get("manager_id") or "" work_mode = (request.args.get("work_mode") or "").strip() status = (request.args.get("status") or "").strip() source = (request.args.get("source") or "").strip() late_only = (request.args.get("late_only") == "1") grace_minutes = request.args.get("grace_minutes") or "0" min_late_minutes = request.args.get("min_late_minutes") or "1" try: grace_minutes = max(0, int(grace_minutes)) except Exception: grace_minutes = 0 try: min_late_minutes = max(0, int(min_late_minutes)) except Exception: min_late_minutes = 1 # ----------------------- # Helpers # ----------------------- def _time_to_minutes(t) -> int: return (t.hour * 60) + t.minute def _mins_to_hhmm(m: Optional[int]) -> str: if m is None: return "" m = int(max(0, m)) hh = m // 60 mm = m % 60 return f"{hh:02d}:{mm:02d}" # ----------------------- # Date range (same logic) # ----------------------- start_date = None end_date = None if from_date and to_date: try: start_date = datetime.strptime(from_date, "%Y-%m-%d").date() end_date = datetime.strptime(to_date, "%Y-%m-%d").date() except Exception: flash("Invalid From/To date format. Use YYYY-MM-DD.", "danger") return redirect(url_for("attendance_summary.late_report_summary")) if not start_date or not end_date: try: dt = datetime.strptime(month, "%Y-%m") y, m = dt.year, dt.month last_day = monthrange(y, m)[1] start_date = date(y, m, 1) end_date = date(y, m, last_day) except Exception: flash("Invalid month format. Use YYYY-MM.", "danger") return redirect(url_for("attendance_summary.late_report_summary")) # ----------------------- # Base query # ----------------------- qry = ( Attendance.query .join(Employee, Employee.id == Attendance.employee_id) .options(joinedload(Attendance.employee)) .filter(Attendance.date >= start_date, Attendance.date <= end_date) .filter(Employee.status == "active") .filter(or_(Employee.exit_date.is_(None), Employee.exit_date >= start_date)) ) visible_ids = _visible_employee_ids_for(current_user) if visible_ids is not None: qry = qry.filter(Employee.id.in_(visible_ids)) # Exclude pre-joining dates (only if DOJ exists) qry = qry.filter(or_(Employee.date_of_joining.is_(None), Employee.date_of_joining <= Attendance.date)) if q: like = f"%{q}%" qry = qry.filter(or_( Employee.full_name.ilike(like), Employee.employee_code.ilike(like), Employee.email.ilike(like) )) if employee_id: try: qry = qry.filter(Employee.id == int(employee_id)) except Exception: pass if department_id: try: qry = qry.filter(Employee.department_id == int(department_id)) except Exception: pass if location: qry = qry.filter(func.lower(Employee.location) == location.lower()) if manager_id: try: qry = qry.filter(Employee.reporting_manager_id == int(manager_id)) except Exception: pass if work_mode: qry = qry.filter(Employee.work_mode == work_mode) if status: qry = qry.filter(Attendance.status == status) if source: qry = qry.filter(Attendance.source == source) qry = qry.order_by(Attendance.date.desc(), Employee.full_name.asc()) records = qry.all() # ----------------------- # TODAY fallback: preload first + last punch time # ----------------------- today = date.today() today_first_punch: Dict[int, object] = {} today_last_punch: Dict[int, object] = {} today_emp_ids = [ att.employee_id for att in records if att.date == today and (not att.check_in_time or not att.check_out_time) ] if today_emp_ids: day_start = datetime.combine(today, datetime.min.time()) day_end = datetime.combine(today + timedelta(days=1), datetime.min.time()) punches = ( BiometricPunch.query .filter(BiometricPunch.employee_id.in_(today_emp_ids)) .filter(BiometricPunch.punch_time >= day_start, BiometricPunch.punch_time < day_end) .order_by(BiometricPunch.employee_id.asc(), BiometricPunch.punch_time.asc()) .all() ) for p in punches: t = p.punch_time.time() if p.employee_id not in today_first_punch: today_first_punch[p.employee_id] = t today_last_punch[p.employee_id] = t # last punch # ----------------------- # Punch Approved By map (approved self punches) # ----------------------- approver_map: Dict[Tuple[int, date], str] = {} emp_ids_in_report = list({att.employee_id for att in records if att.employee_id}) if emp_ids_in_report: start_dt = datetime.combine(start_date, datetime.min.time()) end_dt = datetime.combine(end_date + timedelta(days=1), datetime.min.time()) approved_self_punches = ( BiometricPunch.query .options(joinedload(BiometricPunch.approved_by)) .filter(BiometricPunch.employee_id.in_(emp_ids_in_report)) .filter(BiometricPunch.source_name == "self") .filter(BiometricPunch.approval_status == "approved") .filter(BiometricPunch.approved_by_id.isnot(None)) .filter(BiometricPunch.punch_time >= start_dt, BiometricPunch.punch_time < end_dt) .order_by(BiometricPunch.employee_id.asc(), BiometricPunch.punch_time.desc()) .all() ) for p in approved_self_punches: key = (p.employee_id, p.punch_time.date()) if key not in approver_map: approver_map[key] = (p.approved_by.full_name if p.approved_by else "") or "" # ----------------------- # PDF table data # ----------------------- data = [[ "Date", "Emp Code", "Name", "Dept", "Shift Start", "Shift End", "Check In", "Check Out", "Late (HH:MM)", "Source", "Punch Approved By" ]] for att in records: emp = att.employee shift_start = getattr(emp, "shift_start", None) shift_end = getattr(emp, "shift_end", None) if att.date == today: if not att.check_in_time: fc = today_first_punch.get(emp.id) if fc: att.check_in_time = fc if not att.check_out_time: lc = today_last_punch.get(emp.id) if lc: att.check_out_time = lc late_minutes: Optional[int] = None is_late = False if shift_start and att.check_in_time: base = _time_to_minutes(shift_start) + grace_minutes cin = _time_to_minutes(att.check_in_time) diff = cin - base late_minutes = max(0, diff) is_late = late_minutes >= min_late_minutes if late_only and not is_late: continue dept_name = getattr(getattr(emp, "department", None), "name", "") or "" approved_by_name = approver_map.get((emp.id, att.date), "") data.append([ att.date.strftime("%Y-%m-%d"), emp.employee_code or "", emp.full_name or "", dept_name, shift_start.strftime("%H:%M") if shift_start else "", shift_end.strftime("%H:%M") if shift_end else "", att.check_in_time.strftime("%H:%M") if att.check_in_time else "", att.check_out_time.strftime("%H:%M") if att.check_out_time else "", _mins_to_hhmm(late_minutes) if late_minutes is not None else "", att.source or "", approved_by_name, ]) # ----------------------- # Generate PDF # ----------------------- buff = io.BytesIO() doc = SimpleDocTemplate( buff, pagesize=landscape(A4), leftMargin=14, rightMargin=14, topMargin=14, bottomMargin=14 ) styles = getSampleStyleSheet() story = [] story.append(Paragraph("Late Report Summary (Filtered Export)", styles["Title"])) story.append(Spacer(1, 8)) story.append(Paragraph(f"Range: {start_date} to {end_date}", styles["Normal"])) story.append(Spacer(1, 10)) table = Table(data, repeatRows=1) table.setStyle(TableStyle([ ("BACKGROUND", (0, 0), (-1, 0), colors.lightgrey), ("TEXTCOLOR", (0, 0), (-1, 0), colors.black), ("FONTNAME", (0, 0), (-1, 0), "Helvetica-Bold"), ("FONTSIZE", (0, 0), (-1, 0), 8), ("FONTSIZE", (0, 1), (-1, -1), 7), ("GRID", (0, 0), (-1, -1), 0.3, colors.grey), ("VALIGN", (0, 0), (-1, -1), "MIDDLE"), ("ROWBACKGROUNDS", (0, 1), (-1, -1), [colors.whitesmoke, colors.white]), ("LEFTPADDING", (0, 0), (-1, -1), 4), ("RIGHTPADDING", (0, 0), (-1, -1), 4), ])) story.append(table) doc.build(story) pdf = buff.getvalue() buff.close() filename = f"late_report_{start_date}_{end_date}.pdf" resp = make_response(pdf) resp.headers["Content-Type"] = "application/pdf" resp.headers["Content-Disposition"] = f'attachment; filename="{filename}"' return resp