# routes/payroll.py # --------------------------------------------------------------------- # Payroll: salary structure, formulas, preview/freeze, slips, exports # --------------------------------------------------------------------- from __future__ import annotations import io import re from datetime import datetime, date, timedelta from calendar import monthrange from flask import ( Blueprint, render_template, request, redirect, url_for, flash, jsonify, make_response,current_app ) from flask_login import login_required, current_user from sqlalchemy import func, or_ from sqlalchemy.orm import joinedload from extensions import db from models import ( # core Employee, Department, Attendance, WeekendSettings, Holiday, # salary structure EarningHead, DeductionHead, CTCHead, EmployeeSalaryStructures, SalaryFormula, # payroll artifacts Payroll, PayrollPreviewControl, FrozenSalaryComponent, MonthlyPayroll, # leaves/feedback LeaveRequest, FeedbackResponse, PayrollUnfreezeLog ) from routes import employee # --------------------------------------------------------------------- # Blueprint # --------------------------------------------------------------------- payroll_bp = Blueprint("payroll", __name__, url_prefix="/payroll") # --------------------------------------------------------------------- # Helpers # --------------------------------------------------------------------- def _normalize_dom_id(name: str) -> str: """Normalize head names to a stable DOM id (lowercase + underscores).""" name = name.lower() name = re.sub(r"[^a-z0-9]+", "_", name) return name.strip("_") def _as_float_or_none(raw): if raw is None: return None s = str(raw).strip() if s == "": return None try: return float(s) except ValueError: return None def _normalize_month(s: str) -> str: """Return 'YYYY-MM' (accepts 'YYYY-MM' or 'YYYY-MM-DD').""" parts = s.split("-") y, m = int(parts[0]), int(parts[1]) return f"{y:04d}-{m:02d}" def _is_fnf_month(emp: Employee, y: int, m: int) -> bool: return bool(getattr(emp, "exit_date", None)) and emp.exit_date.year == y and emp.exit_date.month == m def _eligible_for_payroll(emp: Employee, y: int, m: int) -> bool: s = (emp.status or "").lower() if s == "active": return True if s in ("notice_period", "resigned"): return _is_fnf_month(emp, y, m) return False def _has_filled_feedback(employee_id: int, month: str) -> bool: return db.session.query(FeedbackResponse.id).filter( FeedbackResponse.month == month, or_( FeedbackResponse.employee_id == employee_id, # non-anonymous FeedbackResponse.submitted_by_id == employee_id # anonymous by this user ) ).first() is not None # --------------------------------------------------------------------- # Salary Structure (no auto-save; only on explicit save) # --------------------------------------------------------------------- @payroll_bp.route("/salary-structure", methods=["GET", "POST"]) @login_required def salary_structure(): employees = Employee.query.filter_by(status="active").all() earning_heads = EarningHead.query.all() deduction_heads = DeductionHead.query.all() ctc_heads = CTCHead.query.all() formulas = SalaryFormula.query.all() formulas_map = {((f.component_type or "").lower(), f.component_id): f.formula for f in formulas} employee_id = request.args.get("employee_id") or request.form.get("employee_id") selected_employee = Employee.query.get(int(employee_id)) if employee_id else None # Prefill only saved_data = {"earning": {}, "deduction": {}, "ctc": {}} if selected_employee: for row in EmployeeSalaryStructures.query.filter_by(employee_id=selected_employee.id): saved_data[row.type][row.head_id] = row.amount # Save only when the form says action=save if request.method == "POST" and request.form.get("action") == "save" and selected_employee: EmployeeSalaryStructures.query.filter_by(employee_id=selected_employee.id).delete() for head in earning_heads: val = _as_float_or_none(request.form.get(f"earning_{head.id}")) if val is not None: db.session.add(EmployeeSalaryStructures( employee_id=selected_employee.id, head_id=head.id, type="earning", amount=val )) for head in deduction_heads: val = _as_float_or_none(request.form.get(f"deduction_{head.id}")) if val is not None: db.session.add(EmployeeSalaryStructures( employee_id=selected_employee.id, head_id=head.id, type="deduction", amount=val )) for head in ctc_heads: val = _as_float_or_none(request.form.get(f"ctc_{head.id}")) if val is not None: db.session.add(EmployeeSalaryStructures( employee_id=selected_employee.id, head_id=head.id, type="ctc", amount=val )) db.session.commit() flash("Salary structure updated.", "success") return redirect(url_for("payroll.salary_structure", employee_id=selected_employee.id)) return render_template( "payroll/salary_structure.html", employees=employees, earning_heads=earning_heads, deduction_heads=deduction_heads, ctc_heads=ctc_heads, selected_employee=selected_employee, saved_data=saved_data, formulas_map=formulas_map, ) # --------------------------------------------------------------------- # Calculate (AJAX) — uses SalaryFormula with a simple two-pass eval # --------------------------------------------------------------------- @payroll_bp.route("/calculate-salary", methods=["POST"]) @login_required def calculate_salary(): data = request.get_json(silent=True) or {} try: gross = float(data.get("gross", 0)) except (TypeError, ValueError): return jsonify(success=False, message="Invalid gross"), 400 context = {"gross": gross} results, errors = {}, [] earning_heads = EarningHead.query.all() deduction_heads = DeductionHead.query.all() ctc_heads = CTCHead.query.all() formulas = SalaryFormula.query.all() formula_map = {(f.component_type, f.component_id): f.formula for f in formulas} def safe_eval(formula_str: str, ctx: dict): try: # Expected to be simple arithmetic like: gross*0.5, basic*0.4, etc. return eval(formula_str, {}, ctx) except Exception as e: errors.append(f"{formula_str}: {e}") return None def evaluate_group(group_name: str, heads): pending = [] for head in heads: key = _normalize_dom_id(head.name) # must match HTML input IDs formula = formula_map.get((group_name, head.id)) if not formula: continue val = safe_eval(formula, context) if val is None: pending.append((key, formula)) else: context[key] = val results[key] = val # retry once for dependencies for key, formula in pending: if key in context: continue val = safe_eval(formula, context) if val is not None: context[key] = val results[key] = val evaluate_group("earning", earning_heads) evaluate_group("deduction", deduction_heads) evaluate_group("ctc", ctc_heads) if not results and not formulas: # tiny fallback when no formulas exist results = {"basic": round(gross * 0.50, 2), "hra": round(gross * 0.20, 2)} return jsonify(success=True, data=results) # --------------------------------------------------------------------- # Master head management (simple add) # --------------------------------------------------------------------- @payroll_bp.route("/manage-components", methods=["GET", "POST"]) @login_required def manage_components(): if request.method == "POST": head_type = request.form.get("head_type") name = request.form.get("name") description = request.form.get("description", "") if head_type == "earning": db.session.add(EarningHead(name=name, description=description)) elif head_type == "deduction": db.session.add(DeductionHead(name=name, description=description)) elif head_type == "ctc": db.session.add(CTCHead(name=name, description=description)) db.session.commit() flash(f"{head_type.capitalize()} component added.", "success") return redirect(url_for("payroll.manage_components")) # --- Usage counters for badges/disable --- # Structures: counts by head_id per type e_struct = dict( db.session.query(EmployeeSalaryStructures.head_id, func.count()) .filter(EmployeeSalaryStructures.type == "earning") .group_by(EmployeeSalaryStructures.head_id) .all() ) d_struct = dict( db.session.query(EmployeeSalaryStructures.head_id, func.count()) .filter(EmployeeSalaryStructures.type == "deduction") .group_by(EmployeeSalaryStructures.head_id) .all() ) c_struct = dict( db.session.query(EmployeeSalaryStructures.head_id, func.count()) .filter(EmployeeSalaryStructures.type == "ctc") .group_by(EmployeeSalaryStructures.head_id) .all() ) # Formulas: counts by component_id per component_type e_form = dict( db.session.query(SalaryFormula.component_id, func.count()) .filter(SalaryFormula.component_type == "earning") .group_by(SalaryFormula.component_id) .all() ) d_form = dict( db.session.query(SalaryFormula.component_id, func.count()) .filter(SalaryFormula.component_type == "deduction") .group_by(SalaryFormula.component_id) .all() ) c_form = dict( db.session.query(SalaryFormula.component_id, func.count()) .filter(SalaryFormula.component_type == "ctc") .group_by(SalaryFormula.component_id) .all() ) # Frozen components: counts by head_name per component_type # NOTE: FrozenSalaryComponent stores head_name (string). If a head was renamed later, # older frozen rows keep the old name. We show counts matching CURRENT name. e_frozen_by_name = dict( db.session.query(FrozenSalaryComponent.head_name, func.count()) .filter(FrozenSalaryComponent.component_type == "earning") .group_by(FrozenSalaryComponent.head_name) .all() ) d_frozen_by_name = dict( db.session.query(FrozenSalaryComponent.head_name, func.count()) .filter(FrozenSalaryComponent.component_type == "deduction") .group_by(FrozenSalaryComponent.head_name) .all() ) c_frozen_by_name = dict( db.session.query(FrozenSalaryComponent.head_name, func.count()) .filter(FrozenSalaryComponent.component_type == "ctc") .group_by(FrozenSalaryComponent.head_name) .all() ) # Map frozen counts to IDs using current names e_frozen = {h.id: e_frozen_by_name.get(h.name, 0) for h in EarningHead.query.all()} d_frozen = {h.id: d_frozen_by_name.get(h.name, 0) for h in DeductionHead.query.all()} c_frozen = {h.id: c_frozen_by_name.get(h.name, 0) for h in CTCHead.query.all()} return render_template( "payroll/manage_components.html", earning_heads=EarningHead.query.all(), deduction_heads=DeductionHead.query.all(), ctc_heads=CTCHead.query.all(), e_struct=e_struct, d_struct=d_struct, c_struct=c_struct, e_form=e_form, d_form=d_form, c_form=c_form, e_frozen=e_frozen, d_frozen=d_frozen, c_frozen=c_frozen, ) # --------------------------------------------------------------------- # Salary structure lists / details # --------------------------------------------------------------------- @payroll_bp.route("/salary-structures") @login_required def view_salary_structures(): search = (request.args.get("search") or "").strip() selected_department = request.args.get("department") or "" query = Employee.query.filter_by(status="active") if search: query = query.filter(Employee.full_name.ilike(f"%{search}%")) if selected_department: query = query.join(Employee.department).filter(Department.name == selected_department) employees = query.options(joinedload(Employee.department)).all() departments = Department.query.all() data = [] for emp in employees: rows = EmployeeSalaryStructures.query.filter_by(employee_id=emp.id).all() gross = sum(s.amount for s in rows if s.type == "earning") deductions = sum(s.amount for s in rows if s.type == "deduction") ctc = sum(s.amount for s in rows if s.type == "ctc") data.append({ "employee": emp, "gross": gross, "deductions": deductions, "net_pay": gross - deductions, "ctc": gross + ctc, }) return render_template( "payroll/view_salary_structures.html", data=data, departments=departments, search=search, selected_department=selected_department, ) @payroll_bp.route("/view-salary-detail/") @login_required def view_salary_structure_detail(employee_id): emp = Employee.query.get_or_404(employee_id) e_rows = EmployeeSalaryStructures.query.filter_by(employee_id=employee_id, type="earning").all() d_rows = EmployeeSalaryStructures.query.filter_by(employee_id=employee_id, type="deduction").all() c_rows = EmployeeSalaryStructures.query.filter_by(employee_id=employee_id, type="ctc").all() earnings = [{"name": EarningHead.query.get(e.head_id).name, "amount": e.amount} for e in e_rows] deductions = [{"name": DeductionHead.query.get(d.head_id).name, "amount": d.amount} for d in d_rows] ctc_components = [{"name": CTCHead.query.get(c.head_id).name, "amount": c.amount} for c in c_rows] gross = sum(e["amount"] for e in earnings) total_deductions = sum(d["amount"] for d in deductions) ctc_total = sum(c["amount"] for c in ctc_components) return render_template( "payroll/salary_structure_detail.html", employee=emp, earnings=earnings, deductions=deductions, ctc_components=ctc_components, gross=gross, total_deductions=total_deductions, net_pay=gross - total_deductions, ctc=gross + ctc_total, ) # --------------------------------------------------------------------- # Payroll preview (pre-freeze) # --------------------------------------------------------------------- @payroll_bp.route("/process-payroll-preview", methods=["GET", "POST"]) @login_required def process_payroll_preview(): month_arg = request.args.get("month") year_arg = request.args.get("year") if not month_arg or not year_arg: today = datetime.today() month_i, year_i = today.month, today.year else: month_i, year_i = int(month_arg), int(year_arg) period = f"{year_i}-{month_i:02d}" if Payroll.query.filter_by(month=period).first(): flash("Payroll already processed for this month.", "info") return redirect(url_for("payroll.payroll_summary", month=month_i, year=year_i)) weekend = WeekendSettings.query.order_by(WeekendSettings.updated_on.desc()).first() sun_off = bool(weekend and weekend.sunday == "off") def is_saturday_off(day: date) -> bool: if not weekend: return False week_number = (day.day - 1) // 7 + 1 attr = f"{['first','second','third','fourth','fifth'][week_number - 1]}_saturday" return getattr(weekend, attr) == "off" employees = Employee.query.filter(Employee.status.in_(["active", "notice_period", "resigned"])).all() preview = [] for emp in employees: if not _eligible_for_payroll(emp, year_i, month_i): continue total_days_in_month = monthrange(year_i, month_i)[1] start_day = date(year_i, month_i, 1) end_day = date(year_i, month_i, total_days_in_month) if _is_fnf_month(emp, year_i, month_i) and emp.exit_date and emp.exit_date < end_day: end_day = emp.exit_date if emp.date_of_joining > end_day: continue date_list = [start_day + timedelta(days=i) for i in range((end_day - start_day).days + 1)] present = present_half = holidays = weekends = lwp = 0 leave_dates = set() approved_leaves = LeaveRequest.query.filter_by(employee_id=emp.id, status="final_approved") \ .filter(LeaveRequest.start_date <= end_day, LeaveRequest.end_date >= start_day).all() for leave in approved_leaves: cur = max(leave.start_date, start_day) stop = min(leave.end_date, end_day) while cur <= stop: leave_dates.add(cur) cur += timedelta(days=1) for day in date_list: if day < emp.date_of_joining: continue weekday = day.weekday() is_weekend = (weekday == 6 and sun_off) or (weekday == 5 and is_saturday_off(day)) att = Attendance.query.filter_by(employee_id=emp.id, date=day).first() is_holiday = Holiday.query.filter_by(date=day).first() is not None is_leave = day in leave_dates if att and att.remarks == "Present": present += 1 elif att and att.remarks == "Half Day": present_half += 1 elif is_leave: pass elif is_holiday: holidays += 1 elif is_weekend: weekends += 1 else: lwp += 1 total_leaves = len(leave_dates) present_total = present + (0.5 * present_half) total_paid_days = present_total + total_leaves + holidays + weekends earnings = db.session.query(func.sum(EmployeeSalaryStructures.amount)) \ .filter_by(employee_id=emp.id, type="earning").scalar() or 0 deductions = db.session.query(func.sum(EmployeeSalaryStructures.amount)) \ .filter_by(employee_id=emp.id, type="deduction").scalar() or 0 ctc = db.session.query(func.sum(EmployeeSalaryStructures.amount)) \ .filter_by(employee_id=emp.id, type="ctc").scalar() or 0 per_day_salary = (earnings / monthrange(year_i, month_i)[1]) if monthrange(year_i, month_i)[1] > 0 else 0 gross = per_day_salary * total_paid_days net = gross - deductions preview.append({ "employee": emp, "present": round(present_total, 1), "leaves": total_leaves, "holidays": holidays, "weekends": weekends, "lwp": lwp, "total_paid_days": round(total_paid_days, 2), "earnings": round(earnings, 2), "deductions": round(deductions, 2), "ctc": round(ctc, 2), "net_pay": round(net, 2), "is_fnf": _is_fnf_month(emp, year_i, month_i), }) return render_template( "payroll/process_payroll_preview.html", payroll_preview=preview, month=month_i, year=year_i, period=period, ) @payroll_bp.route("/enable-preview", methods=["POST"]) @login_required def enable_preview_for_month(): month = request.form.get("month") rec = PayrollPreviewControl.query.filter_by(month=month).first() if not rec: rec = PayrollPreviewControl(month=month, preview_enabled=True, enabled_on=datetime.now()) db.session.add(rec) else: rec.preview_enabled = True rec.enabled_on = datetime.now() db.session.commit() flash("Preview enabled for employees.", "success") return redirect(url_for("payroll.process_payroll_preview", month=month)) # =========================== # Payroll Preview – Complete # =========================== # (Keep imports minimal here; db already imported above) # ---------- Helpers ---------- def _latest_preview_enabled_month() -> str | None: """ Return the newest YYYY-MM string where preview is enabled, or None. """ row = ( db.session.query(PayrollPreviewControl) .filter(PayrollPreviewControl.preview_enabled == True) .order_by(PayrollPreviewControl.month.desc()) .first() ) return row.month if row else None def _is_preview_enabled(month_yyyy_mm: str) -> bool: row = PayrollPreviewControl.query.filter_by(month=month_yyyy_mm).first() return bool(row and row.preview_enabled) def _wknd_is_saturday_off(weekend: WeekendSettings | None, d: date) -> bool: if not weekend: return False week_number = (d.day - 1) // 7 + 1 # 1..5 bucket = ["first", "second", "third", "fourth", "fifth"][week_number - 1] attr = f"{bucket}_saturday" return getattr(weekend, attr, None) == "off" # ---------- Route ---------- @payroll_bp.route("/preview-slip") @login_required def preview_salary_slip(): # Parse month/year (fallback to latest preview-enabled month, not "current") month_arg = request.args.get("month") year_arg = request.args.get("year") if not (month_arg and year_arg): latest = _latest_preview_enabled_month() # e.g., '2025-08' if latest: y, m = latest.split("-") return redirect(url_for("payroll.preview_salary_slip", year=int(y), month=int(m))) return ("403 Forbidden – Preview not enabled for this month", 403) try: month_i = int(month_arg) year_i = int(year_arg) except ValueError: flash("Invalid month/year.", "danger") return redirect(url_for("dashboard")) selected_month = f"{year_i}-{month_i:02d}" if not _is_preview_enabled(selected_month): return ("403 Forbidden – Preview not enabled for this month", 403) # Current user/employee emp = Employee.query.get(current_user.id) if not emp: flash("Invalid employee.", "danger") return redirect(url_for("dashboard")) # Allow active; or resigned/notice only for F&F month if not (emp.status == "active" or (emp.status in ("notice_period", "resigned") and _is_fnf_month(emp, year_i, month_i))): flash("Preview not available.", "warning") return redirect(url_for("dashboard")) # Date window for the month (clip to exit date if F&F month) start = date(year_i, month_i, 1) end = date(year_i, month_i, monthrange(year_i, month_i)[1]) if _is_fnf_month(emp, year_i, month_i) and emp.exit_date and emp.exit_date < end: end = emp.exit_date doj = getattr(emp, "date_of_joining", None) if doj and doj > end: flash("You were not employed during this month.", "warning") return redirect(url_for("dashboard")) total_days = (end - start).days + 1 all_days = [start + timedelta(days=i) for i in range(total_days)] # Weekend rules weekend = WeekendSettings.query.order_by(WeekendSettings.updated_on.desc()).first() sunday_off = bool(weekend and weekend.sunday == "off") # Tally counters present = half = leave = holiday = weekend_count = lwp = 0 for d in all_days: if doj and d < doj: continue weekday = d.weekday() # 0=Mon ... 5=Sat 6=Sun is_weekend = (weekday == 6 and sunday_off) or (weekday == 5 and _wknd_is_saturday_off(weekend, d)) # Fetch records for this day att = Attendance.query.filter_by(employee_id=emp.id, date=d).first() leave_entry = ( LeaveRequest.query .filter_by(employee_id=emp.id, status="final_approved") .filter(LeaveRequest.start_date <= d, LeaveRequest.end_date >= d) .first() ) holiday_entry = Holiday.query.filter_by(date=d).first() # Priority: Holiday > Weekend > Approved Leave > Attendance > LWP if holiday_entry: holiday += 1 continue if is_weekend: weekend_count += 1 continue if leave_entry: leave += 1 continue if att: rem = (att.remarks or "").strip() if rem == "Present": present += 1 elif rem == "Half Day": half += 1 elif rem == "Leave": leave += 1 elif rem == "Holiday": holiday += 1 elif rem.lower() == "weekend": weekend_count += 1 elif rem == "LWP": lwp += 1 else: # Unknown / empty -> count as LWP on a working day lwp += 1 else: # No records and it’s a working day lwp += 1 # Paid days formula (per your policy) full_month_days = monthrange(year_i, month_i)[1] paid_days = present + (0.5 * half) + leave + holiday + weekend_count per_day_ratio = (paid_days / full_month_days) if full_month_days > 0 else 0.0 # Salary structure → earnings/deductions/ctc structures = EmployeeSalaryStructures.query.filter_by(employee_id=emp.id).all() total_earnings = total_deductions = total_ctc = 0.0 earning_components = [] deduction_components = {} ctc_components = {} for row in structures: if row.type == "earning": head = EarningHead.query.get(row.head_id) name = head.name if head else "Unknown" amt = round((row.amount or 0.0) * per_day_ratio, 2) # earnings prorated by paid days total_earnings += amt earning_components.append({"name": name, "amount": amt}) elif row.type == "deduction": head = DeductionHead.query.get(row.head_id) name = head.name if head else "Unknown" amt = round((row.amount or 0.0), 2) # keep as-is unless your policy says prorate total_deductions += amt deduction_components[name] = amt elif row.type == "ctc": head = CTCHead.query.get(row.head_id) name = head.name if head else "Unknown" amt = round((row.amount or 0.0), 2) total_ctc += amt ctc_components[name] = amt gross = round(total_earnings, 2) net = round(total_earnings - total_deductions, 2) return render_template( "payroll/preview_salary_slip.html", emp=emp, month=month_i, year=year_i, earnings=gross, deductions=round(total_deductions, 2), gross=gross, net=net, ctc=round(total_ctc, 2), paid_days=round(paid_days, 2), components={ "earnings": earning_components, "deductions": deduction_components, "ctc": ctc_components, }, present=present, leave=leave, holiday=holiday, weekends=weekend_count, half=half, lwp=lwp, ) # --------------------------------------------------------------------- # Freeze payroll # --------------------------------------------------------------------- @payroll_bp.route("/freeze-payroll", methods=["POST"]) @login_required def freeze_payroll(): selected_month = request.form.get("month") # 'YYYY-MM' try: dt = datetime.strptime(selected_month, "%Y-%m") except Exception: flash("Invalid month format. Use YYYY-MM.", "danger") return redirect(url_for("payroll.process_payroll_preview")) year_i, month_i = dt.year, dt.month weekend = WeekendSettings.query.order_by(WeekendSettings.updated_on.desc()).first() sun_off = bool(weekend and weekend.sunday == "off") def is_saturday_off(d: date) -> bool: if not weekend: return False week_number = (d.day - 1) // 7 + 1 attr = f"{['first','second','third','fourth','fifth'][week_number - 1]}_saturday" return getattr(weekend, attr, "off") == "off" employees = Employee.query.filter(Employee.status.in_(["active", "notice_period", "resigned"])).all() for emp in employees: if not _eligible_for_payroll(emp, year_i, month_i): continue doj = emp.date_of_joining start, end = date(year_i, month_i, 1), date(year_i, month_i, monthrange(year_i, month_i)[1]) if _is_fnf_month(emp, year_i, month_i) and emp.exit_date and emp.exit_date < end: end = emp.exit_date if doj > end: continue span = (end - start).days + 1 days = [start + timedelta(days=i) for i in range(span)] # Clean slate for this employee/month FrozenSalaryComponent.query.filter_by(employee_id=emp.id, month=selected_month).delete() Payroll.query.filter_by(employee_id=emp.id, month=selected_month).delete() MonthlyPayroll.query.filter_by(employee_id=emp.id, month=month_i, year=year_i).delete() present = half = leave = holiday = weekend_count = lwp = 0 for d in days: if d < doj: continue weekday = d.weekday() is_weekend = (weekday == 6 and sun_off) or (weekday == 5 and is_saturday_off(d)) att = Attendance.query.filter_by(employee_id=emp.id, date=d).first() leave_entry = LeaveRequest.query.filter_by(employee_id=emp.id, status="final_approved") \ .filter(LeaveRequest.start_date <= d, LeaveRequest.end_date >= d).first() holiday_entry = Holiday.query.filter_by(date=d).first() # Priority: Holiday > Weekend > Approved Leave > Attendance > LWP if holiday_entry: holiday += 1 continue if is_weekend: weekend_count += 1 continue if leave_entry: leave += 1 continue if att: rem = (att.remarks or "").strip() if rem == "Present": present += 1 elif rem == "Half Day": half += 1 elif rem == "Leave": leave += 1 elif rem == "Holiday": holiday += 1 elif rem.lower() == "weekend": weekend_count += 1 elif rem == "LWP": lwp += 1 else: lwp += 1 else: # No attendance and it's a working day lwp += 1 paid_days = present + (0.5 * half) + leave + holiday + weekend_count # Freeze components (store master amounts; pro-rating happens via paid_days math) structures = EmployeeSalaryStructures.query.filter_by(employee_id=emp.id).all() earnings = deductions = ctc = 0.0 for item in structures: if item.type == "earning": head = EarningHead.query.get(item.head_id) name = head.name if head else "Unknown" earnings += float(item.amount or 0) elif item.type == "deduction": head = DeductionHead.query.get(item.head_id) name = head.name if head else "Unknown" deductions += float(item.amount or 0) elif item.type == "ctc": head = CTCHead.query.get(item.head_id) name = head.name if head else "Unknown" ctc += float(item.amount or 0) else: continue db.session.add(FrozenSalaryComponent( employee_id=emp.id, month=selected_month, component_type=item.type, head_name=name, amount=float(item.amount or 0), )) full_days = monthrange(year_i, month_i)[1] per_day_salary = (earnings / full_days) if full_days > 0 else 0.0 gross = per_day_salary * paid_days net = gross - deductions db.session.add(Payroll( employee_id=emp.id, month=selected_month, gross_pay=round(gross, 2), deductions=round(deductions, 2), net_pay=round(net, 2), ctc=round(ctc, 2), present_days=present + (0.5 * half), approved_leaves=leave, holidays=holiday, weekends=weekend_count, lwp_days=lwp, )) db.session.add(MonthlyPayroll( employee_id=emp.id, month=month_i, year=year_i, gross=round(gross, 2), deductions=round(deductions, 2), net_pay=round(net, 2), ctc=round(ctc, 2), )) db.session.commit() flash(f"✅ Payroll for {selected_month} has been frozen successfully.", "success") return redirect(url_for("payroll.payroll_summary", month=month_i, year=year_i)) # --------------------------------------------------------------------- # Attendance details (drill-down) # --------------------------------------------------------------------- @payroll_bp.route("/attendance-details/") @login_required def view_attendance_details(employee_id): emp = Employee.query.get_or_404(employee_id) year = int(request.args.get("year")) month_i = int(request.args.get("month")) start = date(year, month_i, 1) end = date(year, month_i, monthrange(year, month_i)[1]) total = (end - start).days + 1 dates = [start + timedelta(days=i) for i in range(total)] att_map = { a.date: a for a in Attendance.query .filter_by(employee_id=employee_id) .filter(Attendance.date >= start, Attendance.date <= end) .all() } leave_dates = set() for leave in LeaveRequest.query.filter_by(employee_id=employee_id, status="final_approved").all(): cur = leave.start_date while cur <= leave.end_date: if start <= cur <= end: leave_dates.add(cur) cur += timedelta(days=1) holiday_dates = {h.date for h in Holiday.query.filter(Holiday.date >= start, Holiday.date <= end).all()} rows = [] for d in dates: att = att_map.get(d) status = att.remarks if att and att.remarks else None check_in = getattr(att, "check_in", None) if att else None check_out = getattr(att, "check_out", None) if att else None hours = getattr(att, "hours", 0.0) if att else 0.0 source = getattr(att, "source", "System") if att else "System" if not status: if d in leave_dates: status = "Leave" elif d in holiday_dates: status = "Holiday" elif d.weekday() == 6: status = "Weekend" else: status = "LWP" rows.append({"date": d, "check_in": check_in, "check_out": check_out, "hours": round(hours, 2), "status": status, "source": source}) return render_template("payroll/attendance_details.html", employee=emp, records=rows, year=year, month=month_i) # --------------------------------------------------------------------- # Summary # --------------------------------------------------------------------- @payroll_bp.route("/payroll-summary") @login_required def payroll_summary(): def _safe_int(val, default): try: return int(val) except (ValueError, TypeError): return default today = datetime.today() month_i = _safe_int(request.args.get("month"), today.month) year_i = _safe_int(request.args.get("year"), today.year) period = f"{year_i}-{month_i:02d}" records = Payroll.query.filter(Payroll.month == period).all() def status_badge(emp: Employee): s = (emp.status or "").lower() if s == "active": return ("Active", "success") if s == "notice_period": return ("Notice Period", "warning") if s == "resigned": return ("Resigned", "danger") return (emp.status or "Unknown", "secondary") def is_fnf(emp: Employee): if not getattr(emp, "exit_date", None): return False if (emp.status or "").lower() not in ("notice_period", "resigned"): return False return emp.exit_date.year == year_i and emp.exit_date.month == month_i enriched = [] for r in records: emp = r.employee label, color = status_badge(emp) enriched.append({"record": r, "employee": emp, "status_label": label, "status_color": color, "is_fnf": is_fnf(emp)}) total_employees = len(records) total_gross = round(sum(r.gross_pay for r in records), 2) total_net = round(sum(r.net_pay for r in records), 2) return render_template( "payroll/payroll_summary.html", records=enriched, month=month_i, year=year_i, total_employees=total_employees, total_gross=total_gross, total_net=total_net, ) # --------------------------------------------------------------------- # Salary slip (view & pdf) — gated by feedback for self # --------------------------------------------------------------------- @payroll_bp.route("/salary-slip//") @login_required def view_salary_slip(employee_id, month): try: month = _normalize_month(month) except Exception: flash("Invalid month. Expected YYYY-MM.", "danger") return redirect(url_for("payroll.payroll_summary")) role = (getattr(current_user.role, "name", "") or "").lower() is_self = current_user.id == employee_id is_hr_admin = role in ("hr", "admin") if not is_hr_admin and not is_self: flash("You are not authorized to view this salary slip.", "danger") return redirect(url_for("payroll.payroll_summary")) if is_self and not _has_filled_feedback(employee_id, month): flash("Please fill the monthly feedback to view your salary slip.", "warning") return redirect(url_for("feedback_emp.form", month=month, next=url_for("payroll.view_salary_slip", employee_id=employee_id, month=month))) payroll = Payroll.query.filter_by(employee_id=employee_id, month=month).first_or_404() employee = Employee.query.get_or_404(employee_id) earnings = FrozenSalaryComponent.query.filter_by(employee_id=employee_id, month=month, component_type="earning").all() deductions = FrozenSalaryComponent.query.filter_by(employee_id=employee_id, month=month, component_type="deduction").all() year_i, mth = map(int, month.split("-")) total_days_in_month = monthrange(year_i, mth)[1] month_start, month_end = date(year_i, mth, 1), date(year_i, mth, total_days_in_month) from_date = max(month_start, employee.date_of_joining) to_date = month_end weekend = WeekendSettings.query.order_by(WeekendSettings.updated_on.desc()).first() sun_off = bool(weekend and weekend.sunday == "off") def is_saturday_off(d: date) -> bool: if not weekend: return False week_number = (d.day - 1) // 7 + 1 attr = f"{['first','second','third','fourth','fifth'][week_number - 1]}_saturday" return getattr(weekend, attr) == "off" weekend_days = set() for d in (from_date + timedelta(days=i) for i in range((to_date - from_date).days + 1)): if d.weekday() == 5 and is_saturday_off(d): weekend_days.add(d) elif d.weekday() == 6 and sun_off: weekend_days.add(d) # ---- Consistent attendance aggregation (Holiday > Weekend > Leave > Attendance > LWP) leave_dates = set() for lr in LeaveRequest.query.filter_by(employee_id=employee_id, status="final_approved") \ .filter(LeaveRequest.start_date <= to_date, LeaveRequest.end_date >= from_date).all(): cur = max(lr.start_date, from_date) stop = min(lr.end_date, to_date) while cur <= stop: leave_dates.add(cur) cur += timedelta(days=1) holiday_dates = {h.date for h in Holiday.query.filter( Holiday.date >= from_date, Holiday.date <= to_date ).all()} att_records = Attendance.query.filter( Attendance.employee_id == employee_id, Attendance.date >= from_date, Attendance.date <= to_date ).all() att_by_date = {a.date: (a.remarks or "").strip() for a in att_records} present_full = present_half = leave = holiday = weekend_count = lwp = 0 cur = from_date while cur <= to_date: if cur in holiday_dates: holiday += 1 elif cur in weekend_days: weekend_count += 1 elif cur in leave_dates: leave += 1 else: rem = att_by_date.get(cur) if rem == "Present": present_full += 1 elif rem == "Half Day": present_half += 1 elif rem == "Leave": leave += 1 elif rem == "Holiday": holiday += 1 elif (rem or "").lower() == "weekend": weekend_count += 1 elif rem == "LWP" or rem is None: lwp += 1 else: lwp += 1 cur += timedelta(days=1) total_paid_days = present_full + (0.5 * present_half) + leave + holiday + weekend_count return render_template( "payroll/salary_slip.html", employee=employee, payroll=payroll, earnings=earnings, deductions=deductions, month=month, current_time=datetime.now(), attendance_summary={ "total_days": round(total_paid_days, 1), "present": present_full + (0.5 * present_half), "leave": leave, "lwp": lwp, "holidays": holiday, "weekends": weekend_count, }, ) @payroll_bp.route("/salary-slip///pdf") @login_required def download_salary_slip_pdf(employee_id, month): # --- Validate and authorize --- try: month = _normalize_month(month) except Exception: flash("Invalid month. Expected YYYY-MM.", "danger") return redirect(url_for("payroll.payroll_summary")) role = (getattr(current_user.role, "name", "") or "").lower() is_self = current_user.id == employee_id is_hr_admin = role in ("hr", "admin") if not is_hr_admin and not is_self: flash("You are not authorized to download this salary slip.", "danger") return redirect(url_for("payroll.payroll_summary")) if is_self and not _has_filled_feedback(employee_id, month): flash("Please fill the monthly feedback to download your salary slip.", "warning") return redirect(url_for( "feedback_emp.form", month=month, next=url_for("payroll.download_salary_slip_pdf", employee_id=employee_id, month=month) )) # --- Data fetch --- payroll = Payroll.query.filter_by(employee_id=employee_id, month=month).first_or_404() employee = Employee.query.get_or_404(employee_id) earnings = FrozenSalaryComponent.query.filter_by( employee_id=employee_id, month=month, component_type="earning" ).all() deductions = FrozenSalaryComponent.query.filter_by( employee_id=employee_id, month=month, component_type="deduction" ).all() # --- Attendance window --- year_i, mth = map(int, month.split("-")) total_days_in_month = monthrange(year_i, mth)[1] month_start, month_end = date(year_i, mth, 1), date(year_i, mth, total_days_in_month) from_date = max(month_start, employee.date_of_joining) to_date = month_end # --- Weekend rules --- weekend = WeekendSettings.query.order_by(WeekendSettings.updated_on.desc()).first() sun_off = bool(weekend and weekend.sunday == "off") def is_saturday_off(d: date) -> bool: if not weekend: return False week_number = (d.day - 1) // 7 + 1 attr = f"{['first','second','third','fourth','fifth'][week_number - 1]}_saturday" return getattr(weekend, attr) == "off" weekend_days = set() for d in (from_date + timedelta(days=i) for i in range((to_date - from_date).days + 1)): if d.weekday() == 5 and is_saturday_off(d): weekend_days.add(d) elif d.weekday() == 6 and sun_off: weekend_days.add(d) # --- Consistent attendance aggregation (Holiday > Weekend > Leave > Attendance > LWP) --- leave_dates = set() for lr in LeaveRequest.query.filter_by(employee_id=employee_id, status="final_approved") \ .filter(LeaveRequest.start_date <= to_date, LeaveRequest.end_date >= from_date).all(): cur = max(lr.start_date, from_date) stop = min(lr.end_date, to_date) while cur <= stop: leave_dates.add(cur) cur += timedelta(days=1) holiday_dates = {h.date for h in Holiday.query.filter( Holiday.date >= from_date, Holiday.date <= to_date ).all()} att_records = Attendance.query.filter( Attendance.employee_id == employee_id, Attendance.date >= from_date, Attendance.date <= to_date ).all() att_by_date = {a.date: (a.remarks or "").strip() for a in att_records} present_full = present_half = leave = holiday = weekend_count = lwp = 0 cur = from_date while cur <= to_date: if cur in holiday_dates: holiday += 1 elif cur in weekend_days: weekend_count += 1 elif cur in leave_dates: leave += 1 else: rem = att_by_date.get(cur) if rem == "Present": present_full += 1 elif rem == "Half Day": present_half += 1 elif rem == "Leave": leave += 1 elif rem == "Holiday": holiday += 1 elif (rem or "").lower() == "weekend": weekend_count += 1 elif rem == "LWP" or rem is None: lwp += 1 else: lwp += 1 cur += timedelta(days=1) total_paid_days = present_full + (0.5 * present_half) + leave + holiday + weekend_count logo_url = url_for("static", filename="images/logo.png", _external=True) # --- Render HTML once for all backends --- rendered = render_template( "payroll/salary_slip_pdf.html", employee=employee, payroll=payroll, earnings=earnings, deductions=deductions, month=month, current_time=datetime.now(), attendance_summary={ "total_days": round(total_paid_days, 1), "present": present_full + (0.5 * present_half), "leave": leave, "lwp": lwp, "holidays": holiday, "weekends": weekend_count, }, ) safe_name = f"SalarySlip_{employee.full_name.replace(' ', '_')}_{month}.pdf" # --- Backend 1: WeasyPrint (preferred) --- try: from weasyprint import HTML pdf = HTML(string=rendered, base_url=request.host_url).write_pdf( stylesheets=["static/css/salary_pdf.css"], presentational_hints=True ) resp = make_response(pdf) resp.headers["Content-Type"] = "application/pdf" resp.headers["Content-Disposition"] = f"attachment; filename={safe_name}" return resp except Exception as e: # Common on shared hosts with mismatched libpango versions current_app.logger.exception("WeasyPrint failed; attempting wkhtmltopdf fallback: %s", e) # --- Backend 2: wkhtmltopdf via pdfkit (requires wkhtmltopdf in PATH) --- try: import pdfkit pdf = pdfkit.from_string(rendered, False) resp = make_response(pdf) resp.headers["Content-Type"] = "application/pdf" resp.headers["Content-Disposition"] = f"attachment; filename={safe_name}" return resp except Exception as e: current_app.logger.exception("pdfkit fallback failed; sending HTML instead: %s", e) # --- Backend 3: last resort — downloadable HTML --- html_name = safe_name.replace(".pdf", ".html") resp = make_response(rendered) resp.headers["Content-Type"] = "text/html; charset=utf-8" resp.headers["Content-Disposition"] = f"attachment; filename={html_name}" return resp # --------------------------------------------------------------------- # My slips list # --------------------------------------------------------------------- @payroll_bp.route("/my-salary-slips") @login_required def my_salary_slips(): if getattr(current_user.role, "name", "") == "hr": records = Payroll.query.order_by(Payroll.month.desc()).all() else: records = Payroll.query.filter_by(employee_id=current_user.id).order_by(Payroll.month.desc()).all() return render_template("payroll/my_salary_slips.html", records=records) # --------------------------------------------------------------------- # Consolidated views / exports # --------------------------------------------------------------------- @payroll_bp.route("/payroll/consolidated-view") @login_required def consolidated_salary_view(): import pandas as pd month = request.args.get("month") if not month: latest = db.session.query(Payroll.month).order_by(Payroll.month.desc()).first() if latest: return redirect(url_for("payroll.consolidated_salary_view", month=latest[0])) flash("Please provide a valid month (e.g., 2025-04)", "danger") return redirect(url_for("payroll.payroll_summary")) records = Payroll.query.filter_by(month=month).all() if not records: flash("No payroll records found for this month.", "info") return redirect(url_for("payroll.payroll_summary")) # Toggle: if True -> deductions prorated by days; if False -> distribute frozen total proportionally PRORATE_DEDUCTIONS = False rows = [] y_i, m_i = map(int, month.split("-")) full_days_in_month = monthrange(y_i, m_i)[1] or 0 for i, p in enumerate(records, start=1): emp = p.employee components = FrozenSalaryComponent.query.filter_by( employee_id=emp.id, month=month ).all() e_map = {c.head_name: float(c.amount or 0.0) for c in components if c.component_type == "earning"} d_map = {c.head_name: float(c.amount or 0.0) for c in components if c.component_type == "deduction"} # --- Master components --- basic_0 = e_map.get("Basic", 0.0) hra_0 = e_map.get("HRA", 0.0) ta_0 = e_map.get("Conv Allowance TA", 0.0) ma_0 = e_map.get("Medical Allowance", 0.0) inc_0 = e_map.get("GC/TL/MR/DMR Incentive", 0.0) gross_master = round(sum(e_map.values()), 2) other_master = round(gross_master - (basic_0 + hra_0 + ta_0 + ma_0 + inc_0), 2) # --- Paid days & frozen totals --- paid_days = round( (p.present_days or 0.0) + (p.approved_leaves or 0.0) + (p.holidays or 0.0) + (p.weekends or 0.0), 2 ) gross_m = round(p.gross_pay or 0.0, 2) # use frozen monthly gross total_deductions_frozen = float(p.deductions or 0.0) # --- Ratio for monthly earnings --- ratio = (paid_days / full_days_in_month) if full_days_in_month else 0.0 # --- Monthly earnings --- # Basic/HRA/Inc/Laptop/KPI/Arrears are prorated; TA & MA are FIXED; Other derived from Gross M basic_m = round(basic_0 * ratio, 2) hra_m = round(hra_0 * ratio, 2) ta_m = round(ta_0, 2) # FIXED ma_m = round(ma_0, 2) # FIXED inc_m = round(inc_0 * ratio, 2) laptop_m = round(e_map.get("Laptop Incentive", 0.0) * ratio, 2) kpi_m = round(e_map.get("KPI Incentive", 0.0) * ratio, 2) arrears_m = round(e_map.get("Arrear", 0.0) * ratio, 2) other_m = round(gross_m - (basic_m + hra_m + ta_m + ma_m + inc_m), 2) # --- Monthly deductions --- sum_master_ded = sum(d_map.values()) or 0.0 def ded_val(head_name: str) -> float: base = d_map.get(head_name, 0.0) if PRORATE_DEDUCTIONS: return round(base * ratio, 2) if sum_master_ded == 0.0 or total_deductions_frozen == 0.0: return 0.0 share = base / sum_master_ded return round(total_deductions_frozen * share, 2) pf_m = ded_val("Provident Fund") esi_m = ded_val("ESI") tds_m = ded_val("TDS") pt_m = ded_val("Prof. Tax") adv_m = ded_val("Adv/ Imprest") target_m = ded_val("Other Deduction") mobile_m = ded_val("Excess Mobile") # --- Net Salary = Gross M − Σ(monthly deductions) --- monthly_ded_total = round( (pf_m or 0) + (esi_m or 0) + (tds_m or 0) + (pt_m or 0) + (adv_m or 0) + (target_m or 0) + (mobile_m or 0), 2 ) net_m = round(gross_m - monthly_ded_total, 2) rows.append({ "SN": i, "Employee": emp.full_name, "UAN": getattr(emp, "uan", "") or "", # Master Salary [Earning] (Other derived) "Basic": basic_0, "HRA": hra_0, "TA": ta_0, "MA": ma_0, "Inc.": inc_0, "KPI Inc.": e_map.get("KPI Incentive", 0.0), "Other": other_master, "Gross": gross_master, # Monthly Salary [Earning] "Days": paid_days, "Basic M": basic_m, "HRA M": hra_m, "TA M": ta_m, "MA M": ma_m, "Inc. M": inc_m, "Other M": other_m, "Laptop": laptop_m, "KPI Incentive": kpi_m, "Arears": arrears_m, "Gross M": gross_m, # Monthly Salary [Deduction] "PF": pf_m, "ESI": esi_m, "TDS": tds_m, "PT": pt_m, "Advance": adv_m, "Target": target_m, "Mobile": mobile_m, # Summary "Net Salary": net_m, "Department Name": emp.department.name if emp.department else "", "Location": getattr(emp, "location", "") or "", }) df = pd.DataFrame(rows) master_heads = ["Basic", "HRA", "TA", "MA", "Inc.", "KPI Inc.", "Other", "Gross"] monthly_heads = { "earnings": ["Basic M", "HRA M", "TA M", "MA M", "Inc. M", "Other M", "Laptop", "KPI Incentive", "Arears", "Gross M"], "deductions": ["PF", "ESI", "TDS", "PT", "Advance", "Target", "Mobile"], } return render_template( "payroll/consolidated_salary_view.html", df=df, month=month, master_heads=master_heads, monthly_heads=monthly_heads ) @payroll_bp.route("/payroll/consolidated-download-excel") @login_required def download_consolidated_salary_excel(): import pandas as pd month = request.args.get("month") if not month: flash("Please provide a month", "danger") return redirect(url_for("payroll.consolidated_salary_view")) records = Payroll.query.filter_by(month=month).all() if not records: flash("No payroll records found for this month.", "info") return redirect(url_for("payroll.consolidated_salary_view")) PRORATE_DEDUCTIONS = False rows = [] y_i, m_i = map(int, month.split("-")) full_days_in_month = monthrange(y_i, m_i)[1] or 0 for i, p in enumerate(records, start=1): emp = p.employee comps = FrozenSalaryComponent.query.filter_by( employee_id=emp.id, month=month ).all() e_map = {c.head_name: float(c.amount or 0.0) for c in comps if c.component_type == "earning"} d_map = {c.head_name: float(c.amount or 0.0) for c in comps if c.component_type == "deduction"} # --- Master components --- basic_0 = e_map.get("Basic", 0.0) hra_0 = e_map.get("HRA", 0.0) ta_0 = e_map.get("Conv Allowance TA", 0.0) ma_0 = e_map.get("Medical Allowance", 0.0) inc_0 = e_map.get("GC/TL/MR/DMR Incentive", 0.0) gross_master = round(sum(e_map.values()), 2) other_master = round(gross_master - (basic_0 + hra_0 + ta_0 + ma_0 + inc_0), 2) # --- Paid days & frozen totals --- paid_days = round( (p.present_days or 0.0) + (p.approved_leaves or 0.0) + (p.holidays or 0.0) + (p.weekends or 0.0), 2 ) gross_m = round(p.gross_pay or 0.0, 2) # use frozen monthly gross total_deductions_frozen = float(p.deductions or 0.0) # --- Ratio for monthly earnings --- ratio = (paid_days / full_days_in_month) if full_days_in_month else 0.0 # --- Monthly earnings (TA & MA fixed; Other derived) --- basic_m = round(basic_0 * ratio, 2) hra_m = round(hra_0 * ratio, 2) ta_m = round(ta_0, 2) # FIXED ma_m = round(ma_0, 2) # FIXED inc_m = round(inc_0 * ratio, 2) laptop_m = round(e_map.get("Laptop Incentive", 0.0) * ratio, 2) kpi_m = round(e_map.get("KPI Incentive", 0.0) * ratio, 2) arrears_m = round(e_map.get("Arrear", 0.0) * ratio, 2) other_m = round(gross_m - (basic_m + hra_m + ta_m + ma_m + inc_m), 2) # --- Monthly deductions --- sum_master_ded = sum(d_map.values()) or 0.0 def ded_val(head_name: str) -> float: base = d_map.get(head_name, 0.0) if PRORATE_DEDUCTIONS: return round(base * ratio, 2) if sum_master_ded == 0.0 or total_deductions_frozen == 0.0: return 0.0 share = base / sum_master_ded return round(total_deductions_frozen * share, 2) pf_m = ded_val("Provident Fund") esi_m = ded_val("ESI") tds_m = ded_val("TDS") pt_m = ded_val("Prof. Tax") adv_m = ded_val("Adv/ Imprest") target_m = ded_val("Other Deduction") mobile_m = ded_val("Excess Mobile") # --- Net Salary = Gross M − Σ(monthly deductions) --- monthly_ded_total = round( (pf_m or 0) + (esi_m or 0) + (tds_m or 0) + (pt_m or 0) + (adv_m or 0) + (target_m or 0) + (mobile_m or 0), 2 ) net_m = round(gross_m - monthly_ded_total, 2) rows.append({ "SN": i, "Employee": emp.full_name, "UAN": getattr(emp, "uan", "") or "", # Master (assigned) — Other derived "Basic": basic_0, "HRA": hra_0, "TA": ta_0, "MA": ma_0, "Inc.": inc_0, "KPI Inc.": e_map.get("KPI Incentive", 0.0), "Other": other_master, "Gross": gross_master, # Monthly (paid) "Days": paid_days, "Basic M": basic_m, "HRA M": hra_m, "TA M": ta_m, "MA M": ma_m, "Inc. M": inc_m, "Other M": other_m, "Laptop": laptop_m, "KPI Incentive": kpi_m, "Arears": arrears_m, "Gross M": gross_m, # Deductions (monthly) "PF": pf_m, "ESI": esi_m, "TDS": tds_m, "PT": pt_m, "Advance": adv_m, "Target": target_m, "Mobile": mobile_m, # Summary "Net Salary": net_m, "Department Name": emp.department.name if emp.department else "", }) df = pd.DataFrame(rows) output = io.BytesIO() df.to_excel(output, index=False, engine="openpyxl") output.seek(0) resp = make_response(output.read()) resp.headers["Content-Disposition"] = f"attachment; filename=Consolidated_Salary_{month}.xlsx" resp.headers["Content-Type"] = "application/vnd.openxmlformats-officedocument.spreadsheetml.sheet" return resp @payroll_bp.route("/payroll/disbursement-sheet") @login_required def download_disbursement_sheet(): import pandas as pd month = request.args.get("month") if not month: flash("Please provide a month", "danger") return redirect(url_for("payroll.consolidated_salary_view")) records = Payroll.query.filter_by(month=month).all() if not records: flash("No payroll records found for this month.", "info") return redirect(url_for("payroll.consolidated_salary_view")) # Match consolidated routes: False = distribute frozen total proportionally; True = prorate by days PRORATE_DEDUCTIONS = False # Pre-calc month days for ratio when needed y_i, m_i = map(int, month.split("-")) full_days_in_month = monthrange(y_i, m_i)[1] or 0 rows = [] for i, p in enumerate(records, start=1): emp = p.employee # Frozen comps for this employee/month comps = FrozenSalaryComponent.query.filter_by( employee_id=emp.id, month=month ).all() d_map = {c.head_name: float(c.amount or 0.0) for c in comps if c.component_type == "deduction"} # From frozen payroll paid_days = round( (p.present_days or 0.0) + (p.approved_leaves or 0.0) + (p.holidays or 0.0) + (p.weekends or 0.0), 2 ) gross_m = round(p.gross_pay or 0.0, 2) total_deductions_frozen = float(p.deductions or 0.0) # Ratio for optional day-based deduction proration ratio = (paid_days / full_days_in_month) if full_days_in_month else 0.0 # Monthly deductions per head (same method as consolidated) sum_master_ded = sum(d_map.values()) or 0.0 def ded_val(head_name: str) -> float: base = d_map.get(head_name, 0.0) if PRORATE_DEDUCTIONS: return round(base * ratio, 2) if sum_master_ded == 0.0 or total_deductions_frozen == 0.0: return 0.0 share = base / sum_master_ded return round(total_deductions_frozen * share, 2) pf_m = ded_val("Provident Fund") esi_m = ded_val("ESI") tds_m = ded_val("TDS") pt_m = ded_val("Prof. Tax") adv_m = ded_val("Adv/ Imprest") target_m = ded_val("Other Deduction") mobile_m = ded_val("Excess Mobile") monthly_ded_total = round( (pf_m or 0) + (esi_m or 0) + (tds_m or 0) + (pt_m or 0) + (adv_m or 0) + (target_m or 0) + (mobile_m or 0), 2 ) net_m = round(gross_m - monthly_ded_total, 2) rows.append({ "SN": i, "Employee Code": emp.employee_code or "", "Employee Name": emp.full_name, "Acc Holder Name": (emp.bank_account_name or emp.full_name) if hasattr(emp, "bank_account_name") else emp.full_name, "Account No.": getattr(emp, "bank_account_number", "") or "", "Bank Name": getattr(emp, "bank_name", "") or "", "IFSC": getattr(emp, "ifsc_code", "") or "", # If your model has a branch field, this will populate; else stays blank gracefully "Branch Name": getattr(emp, "bank_branch_name", "") or "", "Net Salary": net_m, }) df = pd.DataFrame(rows) output = io.BytesIO() df.to_excel(output, index=False, engine="openpyxl") output.seek(0) resp = make_response(output.read()) resp.headers["Content-Disposition"] = f"attachment; filename=Disbursement_Sheet_{month}.xlsx" resp.headers["Content-Type"] = "application/vnd.openxmlformats-officedocument.spreadsheetml.sheet" return resp # --------------------------------------------------------------------- # Formula CRUD (basic) # --------------------------------------------------------------------- @payroll_bp.route("/salary-formula", methods=["GET", "POST"]) @login_required def salary_formula(): earning_heads = EarningHead.query.all() deduction_heads = DeductionHead.query.all() ctc_heads = CTCHead.query.all() all_components = ( [{"id": h.id, "name": h.name, "type": "earning"} for h in earning_heads] + [{"id": h.id, "name": h.name, "type": "deduction"} for h in deduction_heads] + [{"id": h.id, "name": h.name, "type": "ctc"} for h in ctc_heads] ) if request.method == "POST": component_id = request.form.get("component_id") component_type = (request.form.get("component_type") or "").lower().strip() formula = (request.form.get("formula") or "").strip() if not (component_id and component_type and formula): flash("All fields are required.", "danger") else: db.session.add(SalaryFormula( component_id=component_id, component_type=component_type, # always 'earning' / 'deduction' / 'ctc' formula=formula )) db.session.commit() flash("Formula saved successfully.", "success") return redirect(url_for("payroll.salary_formula")) return render_template("payroll/salary_formula.html", all_components=all_components, formulas=SalaryFormula.query.all()) @payroll_bp.route("/salary-formula/delete/", methods=["POST"]) @login_required def delete_salary_formula(formula_id): formula = SalaryFormula.query.get_or_404(formula_id) db.session.delete(formula) db.session.commit() flash("Formula deleted successfully.", "success") return redirect(url_for("payroll.salary_formula")) def _is_hr_admin_user() -> bool: role_name = (getattr(getattr(current_user, "role", None), "name", "") or "").lower() return role_name in ("hr", "admin") @payroll_bp.route("/unfreeze-month", methods=["POST"]) @login_required def unfreeze_month(): if not _is_hr_admin_user(): flash("Only HR/Admin can unfreeze payroll.", "danger") return redirect(url_for("payroll.payroll_summary")) month = request.form.get("month") # 'YYYY-MM' reason = (request.form.get("reason") or "").strip() # Defensive: verify month format try: _ = datetime.strptime(month, "%Y-%m") except Exception: flash("Invalid month format. Use YYYY-MM.", "danger") return redirect(url_for("payroll.payroll_summary")) # Gather rows (before deletion for audit) payroll_q = Payroll.query.filter_by(month=month) mp_q = MonthlyPayroll.query.filter( MonthlyPayroll.month == int(month.split("-")[1]), MonthlyPayroll.year == int(month.split("-")[0]) ) fsc_q = FrozenSalaryComponent.query.filter_by(month=month) count_payroll = payroll_q.count() count_monthly = mp_q.count() count_frozen = fsc_q.count() if count_payroll == 0 and count_frozen == 0 and count_monthly == 0: flash(f"No frozen payroll found for {month}.", "info") return redirect(url_for("payroll.payroll_summary", month=int(month.split("-")[1]), year=int(month.split("-")[0]))) # Delete in a transaction try: fsc_q.delete(synchronize_session=False) payroll_q.delete(synchronize_session=False) mp_q.delete(synchronize_session=False) # (Optional) keep preview enabled so employees can see updated preview after fixes # If you also want to auto-enable preview: upsert control row to True ctrl = PayrollPreviewControl.query.filter_by(month=month).first() if not ctrl: ctrl = PayrollPreviewControl(month=month, preview_enabled=True, enabled_on=datetime.now()) db.session.add(ctrl) else: ctrl.preview_enabled = True ctrl.enabled_on = datetime.now() # Audit log try: db.session.add(PayrollUnfreezeLog( month=month, employee_id=None, reason=reason or None, deleted_payroll_rows=count_payroll, deleted_frozen_components=count_frozen, deleted_monthly_rows=count_monthly, unfreezed_by_id=current_user.id )) except Exception: # If audit table not created, we still proceed pass db.session.commit() except Exception as e: db.session.rollback() flash(f"Unfreeze failed: {e}", "danger") return redirect(url_for("payroll.payroll_summary")) flash(f"Unfroze {month}: Payroll={count_payroll}, FrozenComponents={count_frozen}, Monthly={count_monthly}. Preview re-enabled.", "success") return redirect(url_for("payroll.payroll_summary", month=int(month.split("-")[1]), year=int(month.split("-")[0]))) @payroll_bp.route("/unfreeze-employee", methods=["POST"]) @login_required def unfreeze_employee(): if not _is_hr_admin_user(): flash("Only HR/Admin can unfreeze payroll.", "danger") return redirect(url_for("payroll.payroll_summary")) month = request.form.get("month") # 'YYYY-MM' employee_id = request.form.get("employee_id") reason = (request.form.get("reason") or "").strip() try: _ = datetime.strptime(month, "%Y-%m") employee_id = int(employee_id) except Exception: flash("Invalid input for unfreeze.", "danger") return redirect(url_for("payroll.payroll_summary")) payroll_q = Payroll.query.filter_by(month=month, employee_id=employee_id) mp_q = MonthlyPayroll.query.filter_by( month=int(month.split("-")[1]), year=int(month.split("-")[0]), employee_id=employee_id ) fsc_q = FrozenSalaryComponent.query.filter_by(month=month, employee_id=employee_id) count_payroll = payroll_q.count() count_monthly = mp_q.count() count_frozen = fsc_q.count() if count_payroll == 0 and count_frozen == 0 and count_monthly == 0: flash("No frozen data found for this employee/month.", "info") return redirect(url_for("payroll.payroll_summary", month=int(month.split("-")[1]), year=int(month.split("-")[0]))) try: fsc_q.delete(synchronize_session=False) payroll_q.delete(synchronize_session=False) mp_q.delete(synchronize_session=False) # Optional preview enable (same as above) ctrl = PayrollPreviewControl.query.filter_by(month=month).first() if not ctrl: ctrl = PayrollPreviewControl(month=month, preview_enabled=True, enabled_on=datetime.now()) db.session.add(ctrl) else: ctrl.preview_enabled = True ctrl.enabled_on = datetime.now() # Audit log try: db.session.add(PayrollUnfreezeLog( month=month, employee_id=employee_id, reason=reason or None, deleted_payroll_rows=count_payroll, deleted_frozen_components=count_frozen, deleted_monthly_rows=count_monthly, unfreezed_by_id=current_user.id )) except Exception: pass db.session.commit() except Exception as e: db.session.rollback() flash(f"Unfreeze failed: {e}", "danger") return redirect(url_for("payroll.payroll_summary")) flash(f"Unfroze {month} for employee #{employee_id}: Payroll={count_payroll}, FrozenComponents={count_frozen}, Monthly={count_monthly}. Preview re-enabled.", "success") return redirect(url_for("payroll.payroll_summary", month=int(month.split("-")[1]), year=int(month.split("-")[0]))) # ---------- Component helpers ---------- def _head_model(head_type: str): """Return (ModelClass, canonical_type_str) for earning|deduction|ctc.""" t = (head_type or "").strip().lower() if t == "earning": return EarningHead, "earning" if t == "deduction": return DeductionHead, "deduction" if t == "ctc": return CTCHead, "ctc" raise ValueError("Invalid head_type. Use earning|deduction|ctc") def _is_component_deletable(head_type: str, head_id: int) -> tuple[bool, str]: """ A component is deletable only if: - Not referenced in EmployeeSalaryStructures (by id) - Not referenced in SalaryFormula (by (type,id)) - Not frozen in any payroll (FrozenSalaryComponent) by matching type+name (Frozen stores head_name string; we check by current name) """ Model, canonical = _head_model(head_type) head = Model.query.get(head_id) if not head: return False, "Component not found." # 1) Used in salary structures? used_struct = ( EmployeeSalaryStructures.query .filter_by(type=canonical, head_id=head_id) .limit(1).count() > 0 ) if used_struct: return False, "This component is used in employee salary structures." # 2) Used in formulas? used_formula = ( SalaryFormula.query .filter_by(component_type=canonical, component_id=head_id) .limit(1).count() > 0 ) if used_formula: return False, "This component is referenced in a salary formula." # 3) Frozen in any payroll? # Since FrozenSalaryComponent keeps the name only, we check by current name. used_frozen = ( FrozenSalaryComponent.query .filter_by(component_type=canonical, head_name=head.name) .limit(1).count() > 0 ) if used_frozen: return False, "This component appears in frozen payroll (past months)." return True, "" @payroll_bp.post("/components///edit") @login_required def edit_component(head_type, head_id): try: Model, canonical = _head_model(head_type) except ValueError as e: flash(str(e), "danger") return redirect(url_for("payroll.manage_components")) head = Model.query.get_or_404(head_id) new_name = (request.form.get("name") or "").strip() new_desc = (request.form.get("description") or "").strip() if not new_name: flash("Name is required.", "danger") return redirect(url_for("payroll.manage_components")) # Optional: prevent duplicate names within the same table dup = Model.query.filter( func.lower(Model.name) == new_name.lower(), Model.id != head.id ).first() if dup: flash(f"A {canonical} component with this name already exists.", "warning") return redirect(url_for("payroll.manage_components")) # Update head.name = new_name head.description = new_desc db.session.commit() flash(f"{canonical.capitalize()} component updated.", "success") return redirect(url_for("payroll.manage_components")) @payroll_bp.post("/components///delete") @login_required def delete_component(head_type, head_id): try: Model, canonical = _head_model(head_type) except ValueError as e: flash(str(e), "danger") return redirect(url_for("payroll.manage_components")) head = Model.query.get_or_404(head_id) ok, reason = _is_component_deletable(head_type, head_id) if not ok: flash(f"Cannot delete: {reason}", "warning") return redirect(url_for("payroll.manage_components")) db.session.delete(head) db.session.commit() flash(f"{canonical.capitalize()} component deleted.", "success") return redirect(url_for("payroll.manage_components")) # --- Team salaries (manager's team) ----------------------------------- @payroll_bp.route("/team-salaries") @login_required def team_salaries(): """ Show my team's salaries for a selected month. mode=frozen (default) -> uses frozen Payroll + FrozenSalaryComponent mode=live -> computes from structures + attendance (even if not frozen) Also: - Resigned employees are hidden unless it is their F&F month. - Notice period employees are included. - In frozen mode, if a visible employee has no frozen row, we fallback to a live calculation for that person. """ month = request.args.get("month") manager_id = request.args.get("manager_id") mode = (request.args.get("mode") or "frozen").lower() role_name = (getattr(getattr(current_user, "role", None), "name", "") or "").lower() is_hr_admin = role_name in ("hr", "admin") # 1) Resolve month if not month: if mode == "frozen": latest = db.session.query(Payroll.month).order_by(Payroll.month.desc()).first() if latest: month = latest[0] else: today = datetime.today() month = f"{today.year:04d}-{today.month:02d}" else: today = datetime.today() month = f"{today.year:04d}-{today.month:02d}" # Parse month for later calculations y_i, m_i = map(int, month.split("-")) full_days_in_month = monthrange(y_i, m_i)[1] or 0 PRORATE_DEDUCTIONS = False # policy toggle # 2) Managers list (HR/Admin only) managers = [] if is_hr_admin: managers = ( Employee.query .filter( Employee.id.in_( db.session.query(Employee.reporting_manager_id) .filter(Employee.reporting_manager_id != None).distinct() ) ) .order_by(Employee.full_name.asc()) .all() ) # 3) Decide whose team if is_hr_admin: if manager_id: try: manager_id = int(manager_id) except Exception: manager_id = current_user.id else: manager_id = managers[0].id if managers else current_user.id else: manager_id = current_user.id # 4) Team (direct reports) team = ( Employee.query .options(joinedload(Employee.department)) .filter(Employee.reporting_manager_id == manager_id) .all() ) # ---- Filter visibility by status for this month ---- visible_team = [] excluded_team = [] # if you want to show who was hidden, keep for template for e in team: s = (e.status or "").lower() if s == "resigned" and not _is_fnf_month(e, y_i, m_i): excluded_team.append({"emp": e, "reason": "Resigned"}) continue visible_team.append(e) if not visible_team: flash("No eligible team members for the selected month.", "info") return render_template( "payroll/team_salary_view.html", month=month, rows=[], totals={}, team=[], managers=managers, selected_manager_id=manager_id, is_hr_admin=is_hr_admin, mode=mode, excluded_team=excluded_team ) # helpers shared by live/fallback def _wknd_is_saturday_off(weekend: WeekendSettings | None, d: date) -> bool: if not weekend: return False week_number = (d.day - 1) // 7 + 1 bucket = ["first","second","third","fourth","fifth"][week_number - 1] return getattr(weekend, f"{bucket}_saturday", None) == "off" def _live_attendance_tally(emp: Employee, year_i: int, month_i: int): start = date(year_i, month_i, 1) end = date(year_i, month_i, monthrange(year_i, month_i)[1]) # clip to employment window (respect F&F) if getattr(emp, "exit_date", None) and emp.exit_date.year == year_i and emp.exit_date.month == month_i and emp.exit_date < end: end = emp.exit_date if getattr(emp, "date_of_joining", None) and emp.date_of_joining > end: return dict(present=0, half=0, leave=0, holiday=0, weekend=0, lwp=0, paid_days=0) doj = getattr(emp, "date_of_joining", None) weekend = WeekendSettings.query.order_by(WeekendSettings.updated_on.desc()).first() sunday_off = bool(weekend and weekend.sunday == "off") # prefetch att_map = { a.date: (a.remarks or "").strip() for a in Attendance.query.filter( Attendance.employee_id == emp.id, Attendance.date >= start, Attendance.date <= end ).all() } holiday_dates = {h.date for h in Holiday.query.filter(Holiday.date >= start, Holiday.date <= end).all()} leave_dates = set() for lr in LeaveRequest.query.filter_by(employee_id=emp.id, status="final_approved") \ .filter(LeaveRequest.start_date <= end, LeaveRequest.end_date >= start).all(): cur = max(lr.start_date, start) stop = min(lr.end_date, end) while cur <= stop: leave_dates.add(cur) cur += timedelta(days=1) present = half = leave = holiday = wknd = lwp = 0 cur = start while cur <= end: if doj and cur < doj: cur += timedelta(days=1) continue weekday = cur.weekday() is_wknd = (weekday == 6 and sunday_off) or (weekday == 5 and _wknd_is_saturday_off(weekend, cur)) if cur in holiday_dates: holiday += 1 elif is_wknd: wknd += 1 elif cur in leave_dates: leave += 1 else: rem = att_map.get(cur) if rem == "Present": present += 1 elif rem == "Half Day": half += 1 elif rem == "Leave": leave += 1 elif rem == "Holiday": holiday += 1 elif (rem or "").lower() == "weekend": wknd += 1 else: # LWP or missing on a working day lwp += 1 cur += timedelta(days=1) paid = present + (0.5 * half) + leave + holiday + wknd return dict(present=present, half=half, leave=leave, holiday=holiday, weekend=wknd, lwp=lwp, paid_days=paid) rows = [] if mode == "frozen": # requires frozen payroll rows for visible team; fallback to live if not found visible_ids = [e.id for e in visible_team] records = Payroll.query.filter( Payroll.month == month, Payroll.employee_id.in_(visible_ids) ).all() rec_by_emp = {r.employee_id: r for r in records} for i, emp in enumerate(visible_team, start=1): p = rec_by_emp.get(emp.id) if p is None: # live fallback (projection) for this employee tally = _live_attendance_tally(emp, y_i, m_i) paid_days = round(tally["paid_days"], 2) ratio = (paid_days / full_days_in_month) if full_days_in_month else 0.0 # master from current structures structs = EmployeeSalaryStructures.query.filter_by(employee_id=emp.id).all() e_map, d_map = {}, {} for s in structs: if s.type == "earning": head = EarningHead.query.get(s.head_id) e_map[(head.name if head else "Unknown")] = float(s.amount or 0.0) elif s.type == "deduction": head = DeductionHead.query.get(s.head_id) d_map[(head.name if head else "Unknown")] = float(s.amount or 0.0) basic_0 = e_map.get("Basic", 0.0) hra_0 = e_map.get("HRA", 0.0) ta_0 = e_map.get("Conv Allowance TA", 0.0) ma_0 = e_map.get("Medical Allowance", 0.0) inc_0 = e_map.get("GC/TL/MR/DMR Incentive", 0.0) gross_master = round(sum(e_map.values()), 2) other_master = round(gross_master - (basic_0 + hra_0 + ta_0 + ma_0 + inc_0), 2) basic_m = round(basic_0 * ratio, 2) hra_m = round(hra_0 * ratio, 2) ta_m = round(ta_0, 2) # fixed ma_m = round(ma_0, 2) # fixed inc_m = round(inc_0 * ratio, 2) laptop_m = round(e_map.get("Laptop Incentive", 0.0) * ratio, 2) kpi_m = round(e_map.get("KPI Incentive", 0.0) * ratio, 2) arrears_m = round(e_map.get("Arrear", 0.0) * ratio, 2) gross_m = round(basic_m + hra_m + ta_m + ma_m + inc_m + laptop_m + kpi_m + arrears_m + (other_master * ratio), 2) other_m = round(gross_m - (basic_m + hra_m + ta_m + ma_m + inc_m), 2) def ded_val_live(head: str) -> float: base = d_map.get(head, 0.0) return round(base, 2) if not PRORATE_DEDUCTIONS else round(base * ratio, 2) pf_m = ded_val_live("Provident Fund") esi_m = ded_val_live("ESI") tds_m = ded_val_live("TDS") pt_m = ded_val_live("Prof. Tax") adv_m = ded_val_live("Adv/ Imprest") target_m = ded_val_live("Other Deduction") mobile_m = ded_val_live("Excess Mobile") monthly_ded_total = round((pf_m or 0)+(esi_m or 0)+(tds_m or 0)+(pt_m or 0)+(adv_m or 0)+(target_m or 0)+(mobile_m or 0), 2) net_m = round(gross_m - monthly_ded_total, 2) rows.append({ "sn": i, "emp": emp, "dept_name": emp.department.name if emp.department else "", "basic_0": basic_0, "hra_0": hra_0, "ta_0": ta_0, "ma_0": ma_0, "inc_0": inc_0, "other_0": other_master, "gross_0": gross_master, "days": paid_days, "basic_m": basic_m, "hra_m": hra_m, "ta_m": ta_m, "ma_m": ma_m, "inc_m": inc_m, "other_m": other_m, "laptop_m": laptop_m, "kpi_m": kpi_m, "arrears_m": arrears_m, "gross_m": gross_m, "pf_m": pf_m, "esi_m": esi_m, "tds_m": tds_m, "pt_m": pt_m, "adv_m": adv_m, "target_m": target_m, "mobile_m": mobile_m, "net_m": net_m, "note": "Live (no frozen row)" }) continue # frozen branch (as-is) comps = FrozenSalaryComponent.query.filter_by(employee_id=emp.id, month=month).all() e_map = {c.head_name: float(c.amount or 0.0) for c in comps if c.component_type == "earning"} d_map = {c.head_name: float(c.amount or 0.0) for c in comps if c.component_type == "deduction"} basic_0 = e_map.get("Basic", 0.0) hra_0 = e_map.get("HRA", 0.0) ta_0 = e_map.get("Conv Allowance TA", 0.0) ma_0 = e_map.get("Medical Allowance", 0.0) inc_0 = e_map.get("GC/TL/MR/DMR Incentive", 0.0) gross_master = round(sum(e_map.values()), 2) other_master = round(gross_master - (basic_0 + hra_0 + ta_0 + ma_0 + inc_0), 2) paid_days = round((p.present_days or 0.0) + (p.approved_leaves or 0.0) + (p.holidays or 0.0) + (p.weekends or 0.0), 2) gross_m = round(p.gross_pay or 0.0, 2) total_deductions_frozen = float(p.deductions or 0.0) ratio = (paid_days / full_days_in_month) if full_days_in_month else 0.0 basic_m = round(basic_0 * ratio, 2) hra_m = round(hra_0 * ratio, 2) ta_m = round(ta_0, 2) ma_m = round(ma_0, 2) inc_m = round(inc_0 * ratio, 2) laptop_m = round(e_map.get("Laptop Incentive", 0.0) * ratio, 2) kpi_m = round(e_map.get("KPI Incentive", 0.0) * ratio, 2) arrears_m = round(e_map.get("Arrear", 0.0) * ratio, 2) other_m = round(gross_m - (basic_m + hra_m + ta_m + ma_m + inc_m), 2) sum_master_ded = sum(d_map.values()) or 0.0 def ded_val(head: str) -> float: base = d_map.get(head, 0.0) if PRORATE_DEDUCTIONS: return round(base * ratio, 2) if sum_master_ded == 0.0 or total_deductions_frozen == 0.0: return 0.0 share = base / sum_master_ded return round(total_deductions_frozen * share, 2) pf_m = ded_val("Provident Fund") esi_m = ded_val("ESI") tds_m = ded_val("TDS") pt_m = ded_val("Prof. Tax") adv_m = ded_val("Adv/ Imprest") target_m = ded_val("Other Deduction") mobile_m = ded_val("Excess Mobile") monthly_ded_total = round((pf_m or 0)+(esi_m or 0)+(tds_m or 0)+(pt_m or 0)+(adv_m or 0)+(target_m or 0)+(mobile_m or 0), 2) net_m = round(gross_m - monthly_ded_total, 2) rows.append({ "sn": i, "emp": emp, "dept_name": emp.department.name if emp.department else "", "basic_0": basic_0, "hra_0": hra_0, "ta_0": ta_0, "ma_0": ma_0, "inc_0": inc_0, "other_0": other_master, "gross_0": gross_master, "days": paid_days, "basic_m": basic_m, "hra_m": hra_m, "ta_m": ta_m, "ma_m": ma_m, "inc_m": inc_m, "other_m": other_m, "laptop_m": laptop_m, "kpi_m": kpi_m, "arrears_m": arrears_m, "gross_m": gross_m, "pf_m": pf_m, "esi_m": esi_m, "tds_m": tds_m, "pt_m": pt_m, "adv_m": adv_m, "target_m": target_m, "mobile_m": mobile_m, "net_m": net_m, }) else: # LIVE mode — iterate only visible team for i, emp in enumerate(visible_team, start=1): tally = _live_attendance_tally(emp, y_i, m_i) paid_days = round(tally["paid_days"], 2) ratio = (paid_days / full_days_in_month) if full_days_in_month else 0.0 # Master components from current structures structs = EmployeeSalaryStructures.query.filter_by(employee_id=emp.id).all() e_map, d_map = {}, {} for s in structs: if s.type == "earning": name = (EarningHead.query.get(s.head_id).name if EarningHead.query.get(s.head_id) else "Unknown") e_map[name] = float(s.amount or 0.0) elif s.type == "deduction": name = (DeductionHead.query.get(s.head_id).name if DeductionHead.query.get(s.head_id) else "Unknown") d_map[name] = float(s.amount or 0.0) basic_0 = e_map.get("Basic", 0.0) hra_0 = e_map.get("HRA", 0.0) ta_0 = e_map.get("Conv Allowance TA", 0.0) ma_0 = e_map.get("Medical Allowance", 0.0) inc_0 = e_map.get("GC/TL/MR/DMR Incentive", 0.0) gross_master = round(sum(e_map.values()), 2) other_master = round(gross_master - (basic_0 + hra_0 + ta_0 + ma_0 + inc_0), 2) # Monthly earnings (no frozen gross -> compute from ratio; TA/MA fixed) basic_m = round(basic_0 * ratio, 2) hra_m = round(hra_0 * ratio, 2) ta_m = round(ta_0, 2) # FIXED ma_m = round(ma_0, 2) # FIXED inc_m = round(inc_0 * ratio, 2) laptop_m = round(e_map.get("Laptop Incentive", 0.0) * ratio, 2) kpi_m = round(e_map.get("KPI Incentive", 0.0) * ratio, 2) arrears_m = round(e_map.get("Arrear", 0.0) * ratio, 2) gross_m = round(basic_m + hra_m + ta_m + ma_m + inc_m + laptop_m + kpi_m + arrears_m + (other_master * ratio), 2) other_m = round(gross_m - (basic_m + hra_m + ta_m + ma_m + inc_m), 2) # Deductions policy in LIVE mode (keep as assigned by default) sum_master_ded = sum(d_map.values()) or 0.0 def ded_val_live(head: str) -> float: base = d_map.get(head, 0.0) if PRORATE_DEDUCTIONS: return round(base * ratio, 2) return round(base, 2) pf_m = ded_val_live("Provident Fund") esi_m = ded_val_live("ESI") tds_m = ded_val_live("TDS") pt_m = ded_val_live("Prof. Tax") adv_m = ded_val_live("Adv/ Imprest") target_m = ded_val_live("Other Deduction") mobile_m = ded_val_live("Excess Mobile") monthly_ded_total = round((pf_m or 0)+(esi_m or 0)+(tds_m or 0)+(pt_m or 0)+(adv_m or 0)+(target_m or 0)+(mobile_m or 0), 2) net_m = round(gross_m - monthly_ded_total, 2) rows.append({ "sn": i, "emp": emp, "dept_name": emp.department.name if emp.department else "", "basic_0": basic_0, "hra_0": hra_0, "ta_0": ta_0, "ma_0": ma_0, "inc_0": inc_0, "other_0": other_master, "gross_0": gross_master, "days": paid_days, "basic_m": basic_m, "hra_m": hra_m, "ta_m": ta_m, "ma_m": ma_m, "inc_m": inc_m, "other_m": other_m, "laptop_m": laptop_m, "kpi_m": kpi_m, "arrears_m": arrears_m, "gross_m": gross_m, "pf_m": pf_m, "esi_m": esi_m, "tds_m": tds_m, "pt_m": pt_m, "adv_m": adv_m, "target_m": target_m, "mobile_m": mobile_m, "net_m": net_m, }) totals = { "gross_0": round(sum(r["gross_0"] for r in rows), 2) if rows else 0.0, "gross_m": round(sum(r["gross_m"] for r in rows), 2) if rows else 0.0, "net_m": round(sum(r["net_m"] for r in rows), 2) if rows else 0.0, } return render_template( "payroll/team_salary_view.html", month=month, rows=rows, totals=totals, team=visible_team, # only eligible people managers=managers, selected_manager_id=manager_id, is_hr_admin=is_hr_admin, mode=mode, excluded_team=excluded_team ) @payroll_bp.route("/team-salaries.xlsx") @login_required def team_salaries_excel(): """Excel export for team salaries. Supports mode=frozen|live. Applies the same visibility rules as the HTML view. """ import pandas as pd, io month = request.args.get("month") manager_id = request.args.get("manager_id") mode = (request.args.get("mode") or "frozen").lower() role_name = (getattr(getattr(current_user, "role", None), "name", "") or "").lower() is_hr_admin = role_name in ("hr", "admin") # Month defaulting if not month: if mode == "frozen": latest = db.session.query(Payroll.month).order_by(Payroll.month.desc()).first() month = latest[0] if latest else f"{datetime.today().year:04d}-{datetime.today().month:02d}" else: today = datetime.today() month = f"{today.year:04d}-{today.month:02d}" # Manager resolve if is_hr_admin: try: manager_id = int(manager_id) if manager_id else current_user.id except Exception: manager_id = current_user.id else: manager_id = current_user.id # Team and visibility filtering team = Employee.query.filter(Employee.reporting_manager_id == manager_id).all() y_i, m_i = map(int, month.split("-")) full_days_in_month = monthrange(y_i, m_i)[1] or 0 visible_team = [] for e in team: s = (e.status or "").lower() if s == "resigned" and not _is_fnf_month(e, y_i, m_i): continue visible_team.append(e) if not visible_team: flash("No eligible team members found for export.", "info") return redirect(url_for("payroll.team_salaries", month=month, manager_id=manager_id, mode=mode)) # Build rows (keep export simple in 'frozen' mode) def build_rows_frozen(): out = [] recs = Payroll.query.filter(Payroll.month == month, Payroll.employee_id.in_([e.id for e in visible_team])).all() rec_by_emp = {r.employee_id: r for r in recs} for i, emp in enumerate(visible_team, start=1): p = rec_by_emp.get(emp.id) if not p: # skip non-frozen in export to keep sheet consistent; use on-screen view for projections continue comps = FrozenSalaryComponent.query.filter_by(employee_id=emp.id, month=month).all() e_map = {c.head_name: float(c.amount or 0.0) for c in comps if c.component_type == "earning"} basic_0, hra_0 = e_map.get("Basic", 0.0), e_map.get("HRA", 0.0) ta_0, ma_0 = e_map.get("Conv Allowance TA", 0.0), e_map.get("Medical Allowance", 0.0) inc_0 = e_map.get("GC/TL/MR/DMR Incentive", 0.0) gross_master = round(sum(e_map.values()), 2) paid_days = round((p.present_days or 0.0) + (p.approved_leaves or 0.0) + (p.holidays or 0.0) + (p.weekends or 0.0), 2) gross_m = round(p.gross_pay or 0.0, 2) ratio = (paid_days / full_days_in_month) if full_days_in_month else 0.0 basic_m, hra_m = round(basic_0 * ratio, 2), round(hra_0 * ratio, 2) ta_m, ma_m = round(ta_0, 2), round(ma_0, 2) inc_m = round(inc_0 * ratio, 2) other_m = round(gross_m - (basic_m + hra_m + ta_m + ma_m + inc_m), 2) out.append({ "SN": i, "Employee": emp.full_name, "Department": emp.department.name if emp.department else "", "Days": paid_days, "Basic": basic_0, "HRA": hra_0, "TA": ta_0, "MA": ma_0, "Inc.": inc_0, "Gross (Master)": gross_master, "Basic M": basic_m, "HRA M": hra_m, "TA M": ta_m, "MA M": ma_m, "Inc. M": inc_m, "Other M": other_m, "Gross M": gross_m, "Net Salary": round(p.net_pay or 0.0, 2) }) return out rows = build_rows_frozen() if mode == "frozen" else [] if mode == "live": flash("Live export is not enabled yet. Use the on-screen view or switch to frozen mode for Excel.", "info") return redirect(url_for("payroll.team_salaries", month=month, manager_id=manager_id, mode=mode)) if not rows: flash("Nothing to export for the selected criteria.", "info") return redirect(url_for("payroll.team_salaries", month=month, manager_id=manager_id, mode=mode)) df = pd.DataFrame(rows) output = io.BytesIO() df.to_excel(output, index=False, engine="openpyxl") output.seek(0) resp = make_response(output.read()) resp.headers["Content-Disposition"] = f"attachment; filename=Team_Salaries_{mode}_{month}.xlsx" resp.headers["Content-Type"] = "application/vnd.openxmlformats-officedocument.spreadsheetml.sheet" return resp