# routes/payroll.py (For CHL-specific payroll features) # --------------------------------------------------------------------- # 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, SalaryHeadAlias,DesktimeLog, ) from routes import employee # --------------------------------------------------------------------- # Blueprint # --------------------------------------------------------------------- payroll_bp = Blueprint("payroll", __name__, url_prefix="/payroll") # --------------------------------------------------------------------- # Helpers # --------------------------------------------------------------------- # --------------------------------------------------------------------- # Component alias resolver – map logical keys -> actual head names # --------------------------------------------------------------------- def _resolve_component_aliases(): """ Dynamically map logical component keys like 'basic', 'hra', 'pf' to the actual head names in the master tables. Logical keys we use: basic, hra, ta, ma, inc, kpi, laptop, arrear, pf, esi, tds, pt, adv, target, mobile """ from sqlalchemy import func alias_rows = SalaryHeadAlias.query.all() mapped = {row.logical_key: row.head_name for row in alias_rows} def find(model, keywords): """ Find the first head whose name contains ANY of the keywords (case-insensitive). Returns the head.name or None. """ for kw in keywords: row = ( model.query .filter(func.lower(model.name).like(f"%{kw.lower()}%")) .order_by(model.id.asc()) .first() ) if row: return row.name return None return { # ---- Earnings ---- "basic": find(EarningHead, ["basic"]), "hra": find(EarningHead, ["hra", "house rent"]), "ta": find(EarningHead, ["conv", "conveyance", "transport", "ta"]), "ma": find(EarningHead, ["medical"]), "inc": find(EarningHead, ["incentive", "gc/tl/mr/dmr"]), "kpi": find(EarningHead, ["kpi"]), "laptop": find(EarningHead, ["laptop"]), "arrear": find(EarningHead, ["arrear"]), # ---- Deductions ---- "pf": find(DeductionHead, ["provident", "pf"]), "esi": find(DeductionHead, ["esi"]), "tds": find(DeductionHead, ["tds"]), "pt": find(DeductionHead, ["prof. tax", "professional tax", "pt"]), "adv": find(DeductionHead, ["adv", "imprest"]), "target": find(DeductionHead, ["other deduction", "target"]), "mobile": find(DeductionHead, ["mobile"]), } 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 doj = emp.date_of_joining if isinstance(doj, datetime): doj = doj.date() if doj is None or doj > 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 # --- Handle DOJ safely --- doj = getattr(emp, "date_of_joining", None) if isinstance(doj, datetime): doj = doj.date() start = date(year_i, month_i, 1) end = date(year_i, month_i, monthrange(year_i, month_i)[1]) # Clip to exit date for F&F month if _is_fnf_month(emp, year_i, month_i) and emp.exit_date and emp.exit_date < end: end = emp.exit_date # If DOJ is missing or after this period, skip payroll for this employee if doj is None or 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: # respect DOJ inside loop as well 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)] # --- Prefetch attendance rows --- att_map = { a.date: a for a in Attendance.query .filter(Attendance.employee_id == employee_id) .filter(Attendance.date >= start, Attendance.date <= end) .all() } # --- Prefetch DeskTime logs for month --- dt_map = { l.date: l for l in DesktimeLog.query .filter(DesktimeLog.employee_id == employee_id) .filter(DesktimeLog.date >= start, DesktimeLog.date <= end) .all() } # --- Prefetch approved leave dates --- 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) # --- Prefetch holidays --- holiday_dates = { h.date for h in Holiday.query.filter( Holiday.date >= start, Holiday.date <= end ).all() } # --- Weekend settings --- 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 or d.weekday() != 5: 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" rows = [] for d in dates: att = att_map.get(d) dt_log = dt_map.get(d) if att: status = (att.remarks or "").strip() or None check_in_time = getattr(att, "check_in_time", None) check_out_time = getattr(att, "check_out_time", None) working_hours = float(att.working_hours or 0.0) source = att.source or "System" else: status = None check_in_time = None check_out_time = None working_hours = 0.0 source = "System" # Derive status if missing if not status: is_weekend = ( (d.weekday() == 6 and sun_off) or (d.weekday() == 5 and is_saturday_off(d)) ) # Priority: Holiday > Weekend > Leave > LWP if d in holiday_dates: status = "Holiday" # keep source as System/Holiday elif is_weekend: status = "Holiday" source = "Weekend" # so template prints "Weekend Off" elif d in leave_dates: status = "Leave" source = "Leave" else: status = "LWP" rows.append({ "date": d, "check_in_time": check_in_time, "check_out_time": check_out_time, "working_hours": working_hours, "status": status, "source": source, "desktime_log": dt_log, }) 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) doj = getattr(employee, "date_of_joining", None) if isinstance(doj, datetime): doj = doj.date() from_date = max(month_start, doj) if doj else month_start 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) doj = getattr(employee, "date_of_joining", None) if isinstance(doj, datetime): doj = doj.date() from_date = max(month_start, doj) if doj else month_start 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) # ---------- Frozen component helpers for consolidated/exports ---------- def _frozen_component_maps(emp_id: int, month: str): """ Return four maps for a given employee + month: - e_by_name: earning components keyed by original head_name - d_by_name: deduction components keyed by original head_name - e_by_slug: earning components keyed by normalized slug (basic, hra, etc.) - d_by_slug: deduction components keyed by normalized slug """ comps = FrozenSalaryComponent.query.filter_by( employee_id=emp_id, month=month ).all() e_by_name, d_by_name = {}, {} e_by_slug, d_by_slug = {}, {} for c in comps: amt = float(c.amount or 0.0) name = c.head_name or "" slug = _normalize_dom_id(name) if c.component_type == "earning": e_by_name[name] = amt if slug: e_by_slug[slug] = amt elif c.component_type == "deduction": d_by_name[name] = amt if slug: d_by_slug[slug] = amt return e_by_name, d_by_name, e_by_slug, d_by_slug def _pick_from_slug(slug_map: dict, *candidate_slugs: str) -> float: """ Try multiple normalized keys (aliases) and return the first match. Example: _pick_from_slug(e_slug, 'basic', 'basic_salary') """ for key in candidate_slugs: if key in slug_map: return slug_map[key] return 0.0 # --------------------------------------------------------------------- # 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 .options(joinedload(Payroll.employee)) .filter_by(month=month) .all() ) if not records: flash("No payroll records found for this month.", "info") return redirect(url_for("payroll.payroll_summary")) # ---------- NEW: resolve component names ---------- aliases = _resolve_component_aliases() basic_name = aliases.get("basic") or "Basic" hra_name = aliases.get("hra") or "HRA" ta_name = aliases.get("ta") or "Conv Allowance TA" ma_name = aliases.get("ma") or "Medical Allowance" inc_name = aliases.get("inc") or "GC/TL/MR/DMR Incentive" kpi_name = aliases.get("kpi") or "KPI Incentive" laptop_name = aliases.get("laptop") or "Laptop Incentive" arrear_name = aliases.get("arrear") or "Arrear" pf_name = aliases.get("pf") or "Provident Fund" esi_name = aliases.get("esi") or "ESI" tds_name = aliases.get("tds") or "TDS" pt_name = aliases.get("pt") or "Prof. Tax" adv_name = aliases.get("adv") or "Adv/ Imprest" target_name = aliases.get("target") or "Other Deduction" mobile_name = aliases.get("mobile") or "Excess Mobile" # ---------- Build rows ---------- general_rows = [] master_rows = [] monthly_rows = [] # month days year_i, month_i = map(int, month.split("-")) _, total_days_in_month = monthrange(year_i, month_i) for i, p in enumerate(records, start=1): emp = p.employee if not emp: continue # Attendance & days (align with Payroll model fields) present_days = float(p.present_days or 0) leave_days = float(p.approved_leaves or 0) # <-- USE approved_leaves holiday_days = float(p.holidays or 0) weekend_days = float(p.weekends or 0) lwp_days = float(p.lwp_days or 0) # Paid days: present already includes half-day logic from freeze_payroll paid_days = present_days + leave_days + holiday_days + weekend_days paid_days = max(paid_days, 0) ratio = (paid_days / total_days_in_month) if total_days_in_month else 0 # ---------- components ---------- frozen_components = FrozenSalaryComponent.query.filter_by( employee_id=emp.id, month=month ).all() e_map = {} d_map = {} for c in frozen_components: if c.component_type == "earning": e_map[c.head_name] = e_map.get(c.head_name, 0.0) + (c.amount or 0.0) elif c.component_type == "deduction": d_map[c.head_name] = d_map.get(c.head_name, 0.0) + (c.amount or 0.0) # master earnings basic_0 = e_map.get(basic_name, 0.0) hra_0 = e_map.get(hra_name, 0.0) ta_0 = e_map.get(ta_name, 0.0) ma_0 = e_map.get(ma_name, 0.0) inc_0 = e_map.get(inc_name, 0.0) laptop_0 = e_map.get(laptop_name, 0.0) kpi_0 = e_map.get(kpi_name, 0.0) arrears_0 = e_map.get(arrear_name, 0.0) gross_master = round(sum(e_map.values()), 2) # "Other" for master tab = everything except Basic/HRA/TA/MA/Inc other_all_0 = gross_master - (basic_0 + hra_0 + ta_0 + ma_0 + inc_0) other_master = round(other_all_0, 2) # For monthly tab, we show Laptop/KPI/Arrear separately, # so "Other (Monthly)" must EXCLUDE those three heads residual_other_0 = other_all_0 - laptop_0 - kpi_0 - arrears_0 if residual_other_0 < 0: residual_other_0 = 0.0 # monthly earnings (Basic/HRA/Inc/Other are pro-rated; TA/MA fixed) basic_m = round(basic_0 * ratio, 2) hra_m = round(hra_0 * ratio, 2) ta_m = round(ta_0, 2) # 🔒 fixed, not pro-rated ma_m = round(ma_0, 2) # 🔒 fixed, not pro-rated inc_m = round(inc_0 * ratio, 2) other_m = round(residual_other_0 * ratio, 2) laptop_m = round(laptop_0 * ratio, 2) kpi_m = round(kpi_0 * ratio, 2) arrears_m = round(arrears_0, 2) gross_m = round(sum([ basic_m, hra_m, ta_m, ma_m, inc_m, other_m, laptop_m, kpi_m, arrears_m ]), 2) # master deductions sum_master_ded = sum(d_map.values()) or 0.0 def ded_val(head_name: str) -> float: """ For now, keep non-PF deductions as full monthly values (matching how freeze_payroll stores them). """ base = d_map.get(head_name, 0.0) return round(base, 2) # PF special: (Gross M − HRA M) * 12% capped at 1800 if PF is configured has_pf = pf_name in d_map if has_pf: pf_base_calc = max(gross_m - hra_m, 0) * 0.12 pf_m = round(min(pf_base_calc, 1800.0), 2) else: pf_m = 0.0 esi_m = ded_val(esi_name) tds_m = ded_val(tds_name) pt_m = ded_val(pt_name) adv_m = ded_val(adv_name) target_m = ded_val(target_name) mobile_m = ded_val(mobile_name) total_ded_m = round(sum([pf_m, esi_m, tds_m, pt_m, adv_m, target_m, mobile_m]), 2) net_salary_m = round(gross_m - total_ded_m, 2) # --------- General tab row --------- general_rows.append({ "SN": i, "Employee Code": emp.employee_code or "", "Employee Name": emp.full_name or "", "Department": emp.department.name if emp.department else "", "UAN": emp.uan or "", "Present": present_days, "Leave": leave_days, "Holiday": holiday_days, "Weekend": weekend_days, "LWP": lwp_days, "Paid Days": paid_days, "Net Salary": net_salary_m, }) # --------- Master tab row (assigned) --------- master_rows.append({ "SN": i, "Employee Code": emp.employee_code or "", "Employee Name": emp.full_name or "", "Department": emp.department.name if emp.department else "", "Basic": basic_0, "HRA": hra_0, "TA": ta_0, "MA": ma_0, "Inc": inc_0, "Other": other_master, "Gross": gross_master, }) # --------- Monthly tab row (paid) --------- monthly_rows.append({ "SN": i, "Employee Code": emp.employee_code or "", "Employee Name": emp.full_name or "", "Department": emp.department.name if emp.department else "", "Days": paid_days, "Basic": basic_m, "HRA": hra_m, "TA": ta_m, "MA": ma_m, "Inc": inc_m, "Other": other_m, # <-- NEW "Laptop": laptop_m, "KPI Incentive": kpi_m, "Arrear": arrears_m, "Gross": gross_m, "PF": pf_m, "ESI": esi_m, "TDS": tds_m, "PT": pt_m, "Advance": adv_m, "Target": target_m, "Mobile": mobile_m, "Net Salary": net_salary_m, }) return render_template( "payroll/consolidated_salary_view.html", month=month, general_rows=general_rows, master_rows=master_rows, monthly_rows=monthly_rows, ) @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 .options(joinedload(Payroll.employee)) .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", month=month)) aliases = _resolve_component_aliases() basic_name = aliases.get("basic") or "Basic" hra_name = aliases.get("hra") or "HRA" ta_name = aliases.get("ta") or "Conv Allowance TA" ma_name = aliases.get("ma") or "Medical Allowance" inc_name = aliases.get("inc") or "GC/TL/MR/DMR Incentive" kpi_name = aliases.get("kpi") or "KPI Incentive" laptop_name = aliases.get("laptop") or "Laptop Incentive" arrear_name = aliases.get("arrear") or "Arrear" pf_name = aliases.get("pf") or "Provident Fund" esi_name = aliases.get("esi") or "ESI" tds_name = aliases.get("tds") or "TDS" pt_name = aliases.get("pt") or "Prof. Tax" adv_name = aliases.get("adv") or "Adv/ Imprest" target_name = aliases.get("target") or "Other Deduction" mobile_name = aliases.get("mobile") or "Excess Mobile" year_i, month_i = map(int, month.split("-")) _, total_days_in_month = monthrange(year_i, month_i) general_rows = [] master_rows = [] monthly_rows = [] for i, p in enumerate(records, start=1): emp = p.employee if not emp: continue present_days = float(p.present_days or 0) leave_days = float(p.approved_leaves or 0) # <-- USE approved_leaves holiday_days = float(p.holidays or 0) weekend_days = float(p.weekends or 0) lwp_days = float(p.lwp_days or 0) paid_days = present_days + leave_days + holiday_days + weekend_days paid_days = max(paid_days, 0) ratio = (paid_days / total_days_in_month) if total_days_in_month else 0 frozen_components = FrozenSalaryComponent.query.filter_by( employee_id=emp.id, month=month ).all() e_map = {} d_map = {} for c in frozen_components: if c.component_type == "earning": e_map[c.head_name] = e_map.get(c.head_name, 0.0) + (c.amount or 0.0) elif c.component_type == "deduction": d_map[c.head_name] = d_map.get(c.head_name, 0.0) + (c.amount or 0.0) basic_0 = e_map.get(basic_name, 0.0) hra_0 = e_map.get(hra_name, 0.0) ta_0 = e_map.get(ta_name, 0.0) ma_0 = e_map.get(ma_name, 0.0) inc_0 = e_map.get(inc_name, 0.0) laptop_0 = e_map.get(laptop_name, 0.0) kpi_0 = e_map.get(kpi_name, 0.0) arrears_0 = e_map.get(arrear_name, 0.0) gross_master = round(sum(e_map.values()), 2) # Master "Other" (same as view) other_all_0 = gross_master - (basic_0 + hra_0 + ta_0 + ma_0 + inc_0) other_master = round(other_all_0, 2) # Monthly "Other" excludes Laptop/KPI/Arrear residual_other_0 = other_all_0 - laptop_0 - kpi_0 - arrears_0 if residual_other_0 < 0: residual_other_0 = 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) other_m = round(residual_other_0 * ratio, 2) laptop_m = round(laptop_0 * ratio, 2) kpi_m = round(kpi_0 * ratio, 2) arrears_m = round(arrears_0, 2) gross_m = round(sum([ basic_m, hra_m, ta_m, ma_m, inc_m, other_m, laptop_m, kpi_m, arrears_m ]), 2) sum_master_ded = sum(d_map.values()) or 0.0 def ded_val(head_name: str) -> float: """ Keep non-PF deductions as full monthly values. """ base = d_map.get(head_name, 0.0) return round(base, 2) has_pf = pf_name in d_map if has_pf: pf_base_calc = max(gross_m - hra_m, 0) * 0.12 pf_m = round(min(pf_base_calc, 1800.0), 2) else: pf_m = 0.0 esi_m = ded_val(esi_name) tds_m = ded_val(tds_name) pt_m = ded_val(pt_name) adv_m = ded_val(adv_name) target_m = ded_val(target_name) mobile_m = ded_val(mobile_name) total_ded_m = round(sum([pf_m, esi_m, tds_m, pt_m, adv_m, target_m, mobile_m]), 2) net_salary_m = round(gross_m - total_ded_m, 2) general_rows.append({ "SN": i, "Employee Code": emp.employee_code or "", "Employee Name": emp.full_name or "", "Department": emp.department.name if emp.department else "", "UAN": emp.uan or "", "Present": present_days, "Leave": leave_days, "Holiday": holiday_days, "Weekend": weekend_days, "LWP": lwp_days, "Paid Days": paid_days, "Net Salary": net_salary_m, }) master_rows.append({ "SN": i, "Employee Code": emp.employee_code or "", "Employee Name": emp.full_name or "", "Department": emp.department.name if emp.department else "", "Basic": basic_0, "HRA": hra_0, "TA": ta_0, "MA": ma_0, "Inc": inc_0, "Other": other_master, "Gross": gross_master, }) monthly_rows.append({ "SN": i, "Employee Code": emp.employee_code or "", "Employee Name": emp.full_name or "", "Department": emp.department.name if emp.department else "", "Days": paid_days, "Basic": basic_m, "HRA": hra_m, "TA": ta_m, "MA": ma_m, "Inc": inc_m, "Other": other_m, # <-- NEW "Laptop": laptop_m, "KPI Incentive": kpi_m, "Arrear": arrears_m, "Gross": gross_m, "PF": pf_m, "ESI": esi_m, "TDS": tds_m, "PT": pt_m, "Advance": adv_m, "Target": target_m, "Mobile": mobile_m, "Net Salary": net_salary_m, }) general_df = pd.DataFrame(general_rows) master_df = pd.DataFrame(master_rows) monthly_df = pd.DataFrame(monthly_rows) output = io.BytesIO() with pd.ExcelWriter(output, engine="xlsxwriter") as writer: general_df.to_excel(writer, sheet_name="General", index=False) master_df.to_excel(writer, sheet_name="Master", index=False) monthly_df.to_excel(writer, sheet_name="Monthly", index=False) output.seek(0) filename = f"Consolidated_Salary_{month}.xlsx" response = make_response(output.read()) response.headers["Content-Disposition"] = f"attachment; filename={filename}" response.headers["Content-type"] = "application/vnd.openxmlformats-officedocument.spreadsheetml.sheet" return response @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 .options(joinedload(Payroll.employee)) .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", month=month)) aliases = _resolve_component_aliases() pf_name = aliases.get("pf") or "Provident Fund" esi_name = aliases.get("esi") or "ESI" tds_name = aliases.get("tds") or "TDS" pt_name = aliases.get("pt") or "Prof. Tax" adv_name = aliases.get("adv") or "Adv/ Imprest" target_name = aliases.get("target") or "Other Deduction" mobile_name = aliases.get("mobile") or "Excess Mobile" year_i, month_i = map(int, month.split("-")) _, total_days_in_month = monthrange(year_i, month_i) PRORATE_DEDUCTIONS = False # keep same logic as earlier rows = [] for i, p in enumerate(records, start=1): emp = p.employee if not emp: continue # --- Attendance fields from Payroll model (correct names) --- present_days = float(p.present_days or 0) leave_days = float(p.approved_leaves or 0) holiday_days = float(p.holidays or 0) weekend_days = float(p.weekends or 0) lwp_days = float(p.lwp_days or 0) # We are not storing half days separately in Payroll; treat them as folded into present_days half_days = 0.0 paid_days = present_days + leave_days + holiday_days + weekend_days paid_days = max(paid_days, 0) ratio = (paid_days / total_days_in_month) if total_days_in_month else 0 # --- Frozen components --- frozen_components = FrozenSalaryComponent.query.filter_by( employee_id=emp.id, month=month ).all() e_map = {} d_map = {} for c in frozen_components: if c.component_type == "earning": e_map[c.head_name] = e_map.get(c.head_name, 0.0) + (c.amount or 0.0) elif c.component_type == "deduction": d_map[c.head_name] = d_map.get(c.head_name, 0.0) + (c.amount or 0.0) # gross & hra from monthly logic basic_0 = e_map.get(aliases.get("basic") or "Basic", 0.0) hra_0 = e_map.get(aliases.get("hra") or "HRA", 0.0) basic_m = round(basic_0 * ratio, 2) hra_m = round(hra_0 * ratio, 2) # Monthly earning total – if Payroll has gross_pay, prefer that if getattr(p, "gross_pay", None) is not None: gross_m = float(p.gross_pay or 0.0) else: gross_m = round(sum(e_map.values()) * ratio, 2) sum_master_ded = sum(d_map.values()) or 0.0 total_deductions_frozen = float(getattr(p, "deductions", 0.0) 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 – (Gross M − HRA M) × 12% capped at 1800 has_pf = pf_name in d_map if has_pf: pf_base_calc = max(gross_m - hra_m, 0) * 0.12 pf_m = round(min(pf_base_calc, 1800.0), 2) else: pf_m = 0.0 esi_m = ded_val(esi_name) tds_m = ded_val(tds_name) pt_m = ded_val(pt_name) adv_m = ded_val(adv_name) target_m = ded_val(target_name) mobile_m = ded_val(mobile_name) total_ded_m = round(sum([pf_m, esi_m, tds_m, pt_m, adv_m, target_m, mobile_m]), 2) net_salary_m = round(gross_m - total_ded_m, 2) rows.append({ "SN": i, "Employee Code": emp.employee_code or "", "Employee Name": getattr(emp, "full_name", None) or getattr(emp, "name", "") or "", "Account Holder Name": getattr(emp, "account_holder_name", None) or getattr(emp, "full_name", None) or getattr(emp, "name", "") or "", "Account No.": getattr(emp, "account_number", "") or "", "Bank Name": getattr(emp, "bank_name", "") or "", "IFSC": getattr(emp, "ifsc_code", "") or "", "Branch Name": getattr(emp, "branch_name", "") or "", "Net Salary": net_salary_m, }) df = pd.DataFrame(rows) output = io.BytesIO() with pd.ExcelWriter(output, engine="xlsxwriter") as writer: df.to_excel(writer, sheet_name="Disbursement", index=False) output.seek(0) filename = f"Disbursement_Sheet_{month}.xlsx" response = make_response(output.read()) response.headers["Content-Disposition"] = f"attachment; filename={filename}" response.headers["Content-type"] = "application/vnd.openxmlformats-officedocument.spreadsheetml.sheet" return response # --------------------------------------------------------------------- # 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 @payroll_bp.route("/component-aliases", methods=["GET", "POST"]) @login_required def component_aliases(): logical_keys = [ "basic", "hra", "ta", "ma", "inc", "kpi", "laptop", "arrear", "pf", "esi", "tds", "pt", "adv", "target", "mobile", ] if request.method == "POST": logical_key = request.form.get("logical_key") head_type = request.form.get("head_type") head_name = request.form.get("head_name") if logical_key and head_type and head_name: alias = SalaryHeadAlias.query.filter_by(logical_key=logical_key).first() if not alias: alias = SalaryHeadAlias(logical_key=logical_key) alias.head_type = head_type alias.head_name = head_name db.session.add(alias) db.session.commit() flash("Mapping saved.", "success") else: flash("Please select all fields.", "danger") return redirect(url_for("payroll.component_aliases")) earning_heads = EarningHead.query.order_by(EarningHead.name).all() deduction_heads = DeductionHead.query.order_by(DeductionHead.name).all() ctc_heads = CTCHead.query.order_by(CTCHead.name).all() existing_aliases = SalaryHeadAlias.query.all() return render_template( "payroll/component_aliases.html", logical_keys=logical_keys, earning_heads=earning_heads, deduction_heads=deduction_heads, ctc_heads=ctc_heads, existing_aliases=existing_aliases, )