# routes/payroll.py (For Client HRMS Logic) from sqlalchemy import and_ import io,os import re import math from datetime import datetime, date, timedelta from calendar import monthrange from admin import get_json_setting # or import from wherever you keep helpers 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,Designation,Loan,LoanEMISchedule,LoanTransaction, SalarySlipTemplate, SalarySlipFieldConfig, SalarySlipComponentConfig, BrandingSetting, ) from routes import employee # --------------------------------------------------------------------- # Blueprint # --------------------------------------------------------------------- payroll_bp = Blueprint("payroll", __name__, url_prefix="/payroll") # --------------------------------------------------------------------- # Helpers # --------------------------------------------------------------------- from sqlalchemy import func from io import BytesIO from reportlab.lib.pagesizes import landscape, A4 from reportlab.lib.units import mm from reportlab.pdfgen import canvas from io import BytesIO from reportlab.lib.pagesizes import landscape, A4 from reportlab.lib.units import mm from reportlab.lib import colors from reportlab.lib.utils import ImageReader from reportlab.platypus import Table, TableStyle def _hex_to_color(value, default="#f59e0b"): try: if not value: value = default value = value.strip() if not value.startswith("#"): value = f"#{value}" return colors.HexColor(value) except Exception: return colors.HexColor(default) def _money(v): try: return f"{float(v or 0):,.2f}" except Exception: return "0.00" def _safe_val(v): return "" if v is None else str(v) def _draw_table(c, data, x, y_top, col_widths, row_heights=None, style=None): tbl = Table(data, colWidths=col_widths, rowHeights=row_heights) if style: tbl.setStyle(style) w, h = tbl.wrapOn(c, 0, 0) tbl.drawOn(c, x, y_top - h) return h def _reportlab_salary_slip_pdf( employee, payroll, month, attendance_summary, earnings, deductions, ctc_components=None, slip_template=None, employee_fields=None, company_fields=None, attendance_fields=None, template_logo_url=None, ): from reportlab.lib import colors from reportlab.lib.pagesizes import landscape, A4 from reportlab.lib.units import mm from reportlab.lib.utils import ImageReader from reportlab.platypus import Table, TableStyle from io import BytesIO import os def hex_to_color(value, default="#f59e0b"): try: if not value: value = default value = value.strip() if not value.startswith("#"): value = f"#{value}" return colors.HexColor(value) except Exception: return colors.HexColor(default) def safe(v): return "" if v is None else str(v) def money(v): try: return f"{float(v or 0):,.2f}" except Exception: return "0.00" def draw_table(c, data, x, y_top, col_widths, style, row_heights=None): tbl = Table(data, colWidths=col_widths, rowHeights=row_heights, repeatRows=1) tbl.setStyle(style) w, h = tbl.wrapOn(c, 0, 0) tbl.drawOn(c, x, y_top - h) return h def ensure_space(c, y, needed_h, page_w, page_h, margin, draw_header_cb): if y - needed_h < 12 * mm: c.showPage() return draw_header_cb(c, page_w, page_h, margin) return y def batch_pairs(items): items = items or [] rows = [] for i in range(0, len(items), 2): left_item = items[i] if i < len(items) else {"label": "", "value": ""} right_item = items[i + 1] if i + 1 < len(items) else {"label": "", "value": ""} rows.append([ safe(left_item.get("label", "")), safe(left_item.get("value", "")), safe(right_item.get("label", "")), safe(right_item.get("value", "")), ]) return rows def normalize_logo_path(raw_logo): if not raw_logo: return None raw_logo = raw_logo.strip() if raw_logo.startswith("/static/"): return os.path.join(current_app.root_path, raw_logo.lstrip("/")) if raw_logo.startswith("static/"): return os.path.join(current_app.root_path, raw_logo) if raw_logo.startswith("http://") or raw_logo.startswith("https://"): return raw_logo return raw_logo brand = BrandingSetting.query.get(1) accent_hex = getattr(brand, "accent", "#f59e0b") if brand else "#f59e0b" accent = hex_to_color(accent_hex, "#f59e0b") accent_soft = colors.HexColor("#fdf2e2") border = colors.HexColor("#d7d7d7") white = colors.white text = colors.black buffer = BytesIO() c = canvas.Canvas(buffer, pagesize=landscape(A4)) page_w, page_h = landscape(A4) margin = 10 * mm content_w = page_w - 2 * margin employee_fields = employee_fields or [] company_fields = company_fields or [] attendance_fields = attendance_fields or [] earnings = earnings or [] deductions = deductions or [] ctc_components = ctc_components or [] def company_value(key): for field in company_fields: if field.get("key") == key and field.get("value"): return safe(field.get("value")) return "" def draw_page_header(c, page_w, page_h, margin): y = page_h - margin header_h = 24 * mm content_w = page_w - 2 * margin c.setFillColor(colors.white) c.rect(margin, y - header_h, content_w, header_h, fill=1, stroke=0) c.setStrokeColor(accent) c.setLineWidth(1.2) c.line(margin, y - header_h, margin + content_w, y - header_h) logo_path = normalize_logo_path(template_logo_url) if logo_path: try: c.drawImage( ImageReader(logo_path), margin, y - 20 * mm, width=42 * mm, height=18 * mm, preserveAspectRatio=True, mask="auto", ) except Exception: pass comp_name = company_value("company_name") or "Company" c.setFillColor(text) c.setFont("Helvetica-Bold", 15) c.drawString(margin + 48 * mm, y - 8 * mm, comp_name) company_meta_lines = [] for field in company_fields: if field.get("key") not in ("company_name", "footer_text") and field.get("value"): company_meta_lines.append(f"{field.get('label')}: {field.get('value')}") c.setFont("Helvetica", 9) meta_y = y - 14 * mm for line in company_meta_lines[:2]: c.drawString(margin + 48 * mm, meta_y, safe(line)) meta_y -= 5 * mm parts = month.split("-") year_txt = parts[0] month_num = int(parts[1]) month_names = ["", "January", "February", "March", "April", "May", "June", "July", "August", "September", "October", "November", "December"] c.setFont("Helvetica-Bold", 18) c.drawRightString(margin + content_w, y - 8 * mm, "Salary Slip –") c.drawRightString(margin + content_w, y - 15 * mm, f"{month_names[month_num]} {year_txt}") return y - header_h - 6 * mm y = draw_page_header(c, page_w, page_h, margin) # Employee details - use template-controlled fields and full width if (not slip_template) or getattr(slip_template, "show_employee_section", True): emp_rows = [["Employee Details", "", "", ""]] + batch_pairs(employee_fields) emp_style = TableStyle([ ("SPAN", (0, 0), (-1, 0)), ("BACKGROUND", (0, 0), (-1, 0), accent), ("TEXTCOLOR", (0, 0), (-1, 0), white), ("FONTNAME", (0, 0), (-1, 0), "Helvetica-Bold"), ("FONTSIZE", (0, 0), (-1, -1), 10), ("FONTNAME", (0, 1), (0, -1), "Helvetica-Bold"), ("FONTNAME", (2, 1), (2, -1), "Helvetica-Bold"), ("GRID", (0, 0), (-1, -1), 0.6, border), ("BACKGROUND", (0, 1), (0, -1), accent), ("BACKGROUND", (2, 1), (2, -1), accent), ("TEXTCOLOR", (0, 1), (0, -1), white), ("TEXTCOLOR", (2, 1), (2, -1), white), ("LEFTPADDING", (0, 0), (-1, -1), 8), ("RIGHTPADDING", (0, 0), (-1, -1), 8), ("TOPPADDING", (0, 0), (-1, -1), 7), ("BOTTOMPADDING", (0, 0), (-1, -1), 7), ]) col_q = content_w / 4.0 approx_emp_h = max(24 * mm, len(emp_rows) * 8 * mm) y = ensure_space(c, y, approx_emp_h, page_w, page_h, margin, draw_page_header) emp_h = draw_table(c, emp_rows, margin, y, [col_q, col_q, col_q, col_q], emp_style) y -= emp_h + 6 * mm # Earnings / deductions earnings_rows = [["Earnings", ""]] + [ [e.get("name") or e.get("head_name", ""), money(e.get("amount"))] for e in earnings ] + [["Total", money(getattr(payroll, "gross_pay", 0))]] deductions_rows = [["Deductions", ""]] + [ [d.get("name") or d.get("head_name", ""), money(d.get("amount"))] for d in deductions ] + [["Total", money(getattr(payroll, "deductions", 0))]] side_style = TableStyle([ ("SPAN", (0, 0), (-1, 0)), ("BACKGROUND", (0, 0), (-1, 0), accent), ("TEXTCOLOR", (0, 0), (-1, 0), white), ("FONTNAME", (0, 0), (-1, 0), "Helvetica-Bold"), ("GRID", (0, 0), (-1, -1), 0.6, border), ("ALIGN", (1, 1), (1, -1), "RIGHT"), ("FONTNAME", (0, -1), (-1, -1), "Helvetica-Bold"), ("BACKGROUND", (0, -1), (-1, -1), accent_soft), ("LEFTPADDING", (0, 0), (-1, -1), 8), ("RIGHTPADDING", (0, 0), (-1, -1), 8), ("TOPPADDING", (0, 0), (-1, -1), 7), ("BOTTOMPADDING", (0, 0), (-1, -1), 7), ]) left_w = (content_w - 6 * mm) / 2 rows_needed = max(len(earnings_rows), len(deductions_rows)) approx_side_h = max(35 * mm, rows_needed * 8 * mm) y = ensure_space(c, y, approx_side_h, page_w, page_h, margin, draw_page_header) earn_h = draw_table(c, earnings_rows, margin, y, [left_w * 0.62, left_w * 0.38], side_style) ded_h = draw_table(c, deductions_rows, margin + left_w + 6 * mm, y, [left_w * 0.62, left_w * 0.38], side_style) y -= max(earn_h, ded_h) + 6 * mm # Net Pay net_rows = [["Net Pay", money(getattr(payroll, "net_pay", 0))]] net_style = TableStyle([ ("BACKGROUND", (0, 0), (0, 0), accent), ("TEXTCOLOR", (0, 0), (0, 0), white), ("FONTNAME", (0, 0), (0, 0), "Helvetica-Bold"), ("FONTNAME", (1, 0), (1, 0), "Helvetica-Bold"), ("GRID", (0, 0), (-1, -1), 0.6, border), ("LEFTPADDING", (0, 0), (-1, -1), 8), ("RIGHTPADDING", (0, 0), (-1, -1), 8), ("TOPPADDING", (0, 0), (-1, -1), 8), ("BOTTOMPADDING", (0, 0), (-1, -1), 8), ]) y = ensure_space(c, y, 15 * mm, page_w, page_h, margin, draw_page_header) net_h = draw_table(c, net_rows, margin, y, [85 * mm, content_w - 85 * mm], net_style) y -= net_h + 6 * mm # CTC section - only if template allows if ((not slip_template) or getattr(slip_template, "show_ctc_section", False)) and ctc_components: ctc_rows = [["CTC Components", ""]] + [ [citem.get("name") or citem.get("head_name", ""), money(citem.get("amount"))] for citem in ctc_components ] ctc_style = TableStyle([ ("SPAN", (0, 0), (-1, 0)), ("BACKGROUND", (0, 0), (-1, 0), accent), ("TEXTCOLOR", (0, 0), (-1, 0), white), ("FONTNAME", (0, 0), (-1, 0), "Helvetica-Bold"), ("GRID", (0, 0), (-1, -1), 0.6, border), ("ALIGN", (1, 1), (1, -1), "RIGHT"), ("LEFTPADDING", (0, 0), (-1, -1), 8), ("RIGHTPADDING", (0, 0), (-1, -1), 8), ("TOPPADDING", (0, 0), (-1, -1), 7), ("BOTTOMPADDING", (0, 0), (-1, -1), 7), ]) approx_ctc_h = max(20 * mm, len(ctc_rows) * 8 * mm) y = ensure_space(c, y, approx_ctc_h, page_w, page_h, margin, draw_page_header) ctc_h = draw_table(c, ctc_rows, margin, y, [content_w * 0.63, content_w * 0.37], ctc_style) y -= ctc_h + 6 * mm # Attendance section - template-controlled if (not slip_template) or getattr(slip_template, "show_attendance_section", True): att_rows = [["Attendance Summary", "", "", ""]] + batch_pairs(attendance_fields) att_style = TableStyle([ ("SPAN", (0, 0), (-1, 0)), ("BACKGROUND", (0, 0), (-1, 0), accent), ("TEXTCOLOR", (0, 0), (-1, 0), white), ("FONTNAME", (0, 0), (-1, 0), "Helvetica-Bold"), ("GRID", (0, 0), (-1, -1), 0.6, border), ("BACKGROUND", (0, 1), (0, -1), accent), ("BACKGROUND", (2, 1), (2, -1), accent), ("TEXTCOLOR", (0, 1), (0, -1), white), ("TEXTCOLOR", (2, 1), (2, -1), white), ("FONTNAME", (0, 1), (0, -1), "Helvetica-Bold"), ("FONTNAME", (2, 1), (2, -1), "Helvetica-Bold"), ("LEFTPADDING", (0, 0), (-1, -1), 8), ("RIGHTPADDING", (0, 0), (-1, -1), 8), ("TOPPADDING", (0, 0), (-1, -1), 7), ("BOTTOMPADDING", (0, 0), (-1, -1), 7), ]) y = ensure_space(c, y, max(20 * mm, len(att_rows) * 8 * mm), page_w, page_h, margin, draw_page_header) draw_table(c, att_rows, margin, y, [content_w / 4.0] * 4, att_style) c.save() buffer.seek(0) return buffer.getvalue() def _salary_slip_pdf_response(employee_id, month, disposition="attachment"): payroll = Payroll.query.filter_by(employee_id=employee_id, month=month).first_or_404() employee = Employee.query.get_or_404(employee_id) earning_components, deduction_components, ctc_components, paid_days_from_payroll = _build_monthly_components_for_slip( employee_id=employee_id, month=month, payroll=payroll ) earnings = earning_components deductions = deduction_components 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 = _as_date(getattr(employee, "date_of_joining", None)) exit_d = _as_date(getattr(employee, "exit_date", None)) from_date = max(month_start, doj) if doj else month_start to_date = month_end if exit_d and exit_d < to_date: to_date = exit_d 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) 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: { "remarks": _safe_remark_text(a.remarks), "note": _safe_note_text(a.note), } for a in att_records } present = leave = holiday = weekend_count = lwp = 0.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 and cur not in att_by_date: leave += 1 else: att_info = att_by_date.get(cur, {}) rem = _safe_remark_text(att_info.get("remarks")) note = _safe_note_text(att_info.get("note")) if rem: p, lv, hol, wk, lw = _apply_attendance_remark_counts(rem, note) present += p leave += lv holiday += hol weekend_count += wk lwp += lw elif cur in leave_dates: leave += 1 else: lwp += 1 cur += timedelta(days=1) total_paid_days = present + leave + holiday + weekend_count attendance_summary = { "total_days": round(total_paid_days, 1), "present": present, "leave": leave, "lwp": lwp, "holidays": holiday, "weekends": weekend_count, } slip_template = get_salary_slip_template_for_employee(employee) earning_cfg = _build_component_config_map(slip_template, "earning") deduction_cfg = _build_component_config_map(slip_template, "deduction") ctc_cfg = _build_component_config_map(slip_template, "ctc") display_earnings = _apply_component_template_list(earnings, earning_cfg) display_deductions = _apply_component_template_list(deductions, deduction_cfg) display_ctc = _apply_component_template_list(ctc_components, ctc_cfg) display_fields = _build_salary_slip_display_fields(employee, attendance_summary, slip_template) template_logo_url = None if slip_template and slip_template.logo_path: template_logo_url = slip_template.logo_path else: theme = get_json_setting("theme", {}) if theme and theme.get("logo_url"): template_logo_url = theme.get("logo_url") else: template_logo_url = url_for("static", filename="images/logo.png") rendered = render_template( "payroll/salary_slip_pdf.html", employee=employee, payroll=payroll, earnings=display_earnings, deductions=display_deductions, ctc_components=display_ctc, month=month, current_time=datetime.now(), attendance_summary=attendance_summary, slip_template=slip_template, employee_fields=display_fields["employee_fields"], company_fields=display_fields["company_fields"], attendance_fields=display_fields["attendance_fields"], template_logo_url=template_logo_url, ) safe_name = f"SalarySlip_{employee.full_name.replace(' ', '_')}_{month}.pdf" try: pdf = _reportlab_salary_slip_pdf( employee=employee, payroll=payroll, month=month, attendance_summary=attendance_summary, earnings=display_earnings, deductions=display_deductions, ctc_components=display_ctc, slip_template=slip_template, employee_fields=display_fields["employee_fields"], company_fields=display_fields["company_fields"], attendance_fields=display_fields["attendance_fields"], template_logo_url=template_logo_url, ) resp = make_response(pdf) resp.headers["Content-Type"] = "application/pdf" resp.headers["Content-Disposition"] = f'{disposition}; filename={safe_name}' return resp except Exception as e: current_app.logger.exception("ReportLab PDF generation failed: %s", e) if request.path.startswith("/api/"): return jsonify({ "success": False, "message": "PDF generation failed on server" }), 500 flash("PDF generation is not available right now. Please use Print View.", "warning") return redirect(url_for("payroll.print_salary_slip", employee_id=employee_id, month=month)) def _safe_remark_text(rem): if rem is None: return "" if isinstance(rem, dict): rem = ( rem.get("remarks") or rem.get("remark") or rem.get("status") or rem.get("label") or "" ) return str(rem).strip() def _safe_note_text(note): if note is None: return "" if isinstance(note, dict): note = ( note.get("note") or note.get("notes") or note.get("remark") or "" ) return str(note).strip() def _apply_attendance_remark_counts(rem, note=""): present = half = leave = holiday = weekend = lwp = 0.0 rem = (rem or "").strip() note = (note or "").strip() if rem == "Present": present = 1.0 elif rem == "Half Day": if "HD_LEAVE:" in note: sh = "" if "SH:" in note: sh = note.split("SH:")[1].split("|")[0].strip() # first half leave always paid leave += 0.5 if sh == "Present": present += 0.5 elif sh == "Leave": leave += 0.5 elif sh == "LWP": lwp += 0.5 elif sh == "Half Day": # ✅ business rule: FH Leave + SH Half Day = 1.0 paid day present += 0.5 elif sh == "Holiday": holiday += 0.5 elif sh == "Weekend": weekend += 0.5 else: # fallback if other-half status missing lwp += 0.5 else: # normal half day present += 0.5 elif rem == "Leave": leave = 1.0 elif rem == "Holiday": holiday = 1.0 elif rem.lower() == "weekend": weekend = 1.0 elif rem == "LWP": lwp = 1.0 else: lwp = 1.0 return present, leave, holiday, weekend, lwp def _is_lwp_leave(lr) -> bool: """ Return True if this LeaveRequest represents Leave Without Pay (LWP). Tries multiple fields to be compatible with different schemas. """ # direct flag (if you have it) if hasattr(lr, "is_lwp") and lr.is_lwp is not None: return bool(lr.is_lwp) # relationship to LeaveType (common) lt = getattr(lr, "leave_type", None) lt_name = (getattr(lt, "name", None) or "").strip().lower() lt_code = (getattr(lt, "code", None) or "").strip().lower() # fallback fields (if stored directly on request) req_type = (getattr(lr, "leave_type_name", None) or getattr(lr, "type", None) or "").strip().lower() s = " ".join([lt_name, lt_code, req_type]).strip() return ("lwp" in s) or ("leave without pay" in s) or ("loss of pay" in s) or ("lop" in s) @payroll_bp.app_template_filter("dom_id") def jinja_dom_id_filter(value: str) -> str: return _normalize_dom_id(value or "") def _get_loan_total_payable(loan: "Loan") -> float: """ Total payable principal for the loan. Uses sanctioned_amount if present, else requested_amount. """ if getattr(loan, "sanctioned_amount", None) is not None: return float(loan.sanctioned_amount) return float(loan.requested_amount or 0.0) def _as_date(d): if not d: return None return d.date() if isinstance(d, datetime) else d def _build_monthly_components_for_slip(employee_id: int, month: str, payroll: Payroll): """ Returns: earning_components: list[{name, amount}] monthly/prorated deduction_components: list[{name, amount}] monthly/prorated ctc_components: list[{name, amount}] (master, not prorated) Breakup is forced to match payroll.gross_pay and payroll.deductions. """ comps = FrozenSalaryComponent.query.filter_by(employee_id=employee_id, month=month).all() e_map, d_map, ctc_map = {}, {}, {} for c in comps: name = (c.head_name or "").strip() amt = float(c.amount or 0.0) if not name: continue if c.component_type == "earning": e_map[name] = e_map.get(name, 0.0) + amt elif c.component_type == "deduction": d_map[name] = d_map.get(name, 0.0) + amt elif c.component_type == "ctc": ctc_map[name] = ctc_map.get(name, 0.0) + amt year_i, month_i = map(int, month.split("-")) month_days = monthrange(year_i, month_i)[1] # Paid days from frozen payroll row (MOST reliable) paid_days = float(payroll.present_days or 0) \ + float(payroll.approved_leaves or 0) \ + float(payroll.holidays or 0) \ + float(payroll.weekends or 0) # Monthly/prorated maps that MATCH payroll totals exactly e_month_map, _ = compute_monthly_earnings_from_frozen_map( e_map=e_map, paid_days=paid_days, month_days=month_days, payroll_total=float(payroll.gross_pay or 0.0), ) d_month_map, _ = compute_monthly_deductions_from_frozen_map( d_map=d_map, paid_days=paid_days, month_days=month_days, payroll_total=float(payroll.deductions or 0.0), ) # Convert to ordered lists (stable order: by name) earning_components = [ {"name": k, "amount": round(float(v or 0.0), 2)} for k, v in sorted(e_month_map.items(), key=lambda x: x[0].lower()) if round(float(v or 0.0), 2) != 0.0 ] deduction_components = [ {"name": k, "amount": round(float(v or 0.0), 2)} for k, v in sorted(d_month_map.items(), key=lambda x: x[0].lower()) if round(float(v or 0.0), 2) != 0.0 ] ctc_components = [ {"name": k, "amount": round(float(v or 0.0), 2)} for k, v in sorted(ctc_map.items(), key=lambda x: x[0].lower()) if round(float(v or 0.0), 2) != 0.0 ] return earning_components, deduction_components, ctc_components, paid_days from sqlalchemy import func def _emi_payable_amount(emi: "LoanEMISchedule") -> float: loan = emi.loan scheduled = float(emi.amount or 0.0) total_payable = _get_loan_total_payable(loan) if total_payable <= 0: return round(scheduled, 2) # ✅ Use loan transactions (CR repayments) as truth paid_sum = ( db.session.query(func.coalesce(func.sum(LoanTransaction.amount), 0)) .filter(LoanTransaction.loan_id == loan.id) .filter(LoanTransaction.direction == "CR") .scalar() ) or 0 paid_sum = float(paid_sum) remaining = round(total_payable - paid_sum, 2) if remaining <= 0: return 0.0 last_pending_due = ( db.session.query(func.max(LoanEMISchedule.due_month)) .filter(LoanEMISchedule.loan_id == loan.id) .filter(LoanEMISchedule.status != "paid") .scalar() ) # ✅ last EMI month => clamp to remaining (reduced EMI like 500) if last_pending_due and emi.due_month == last_pending_due: return float(min(scheduled, remaining)) return float(min(scheduled, remaining)) def _loan_due_for_month(employee_id: int, month_yyyy_mm: str): dt = datetime.strptime(month_yyyy_mm, "%Y-%m") month_first = date(dt.year, dt.month, 1) emi_rows = ( LoanEMISchedule.query .join(Loan, LoanEMISchedule.loan_id == Loan.id) .filter(Loan.employee_id == employee_id) .filter(Loan.repayment_mode == "salary") .filter(Loan.status.in_(["approved", "active"])) .filter(LoanEMISchedule.due_month == month_first) .filter(~LoanEMISchedule.status.in_(["paid", "skipped_approved"])) .all() ) total_due = 0.0 emi_pay_map = {} for emi in emi_rows: payable = round(_emi_payable_amount(emi), 2) emi_pay_map[emi.id] = payable total_due += payable return round(total_due, 2), emi_rows, emi_pay_map # --------------------------------------------------------------------- # 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, loan_emi """ 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": mapped.get("basic") or find(EarningHead, ["basic"]), "hra": mapped.get("hra") or find(EarningHead, ["hra", "house rent"]), "ta": mapped.get("ta") or find(EarningHead, ["conv", "conveyance", "transport", "ta"]), "ma": mapped.get("ma") or find(EarningHead, ["medical"]), "inc": mapped.get("inc") or find(EarningHead, ["incentive", "gc/tl/mr/dmr"]), "kpi": mapped.get("kpi") or find(EarningHead, ["kpi"]), "laptop": mapped.get("laptop") or find(EarningHead, ["laptop"]), "arrear": mapped.get("arrear") or find(EarningHead, ["arrear"]), # ---- Deductions ---- "pf": mapped.get("pf") or find(DeductionHead, ["provident", "pf"]), "esi": mapped.get("esi") or find(DeductionHead, ["esi"]), "tds": mapped.get("tds") or find(DeductionHead, ["tds"]), "pt": mapped.get("pt") or find(DeductionHead, ["prof. tax", "professional tax", "pt"]), "adv": mapped.get("adv") or find(DeductionHead, ["adv", "imprest"]), "target": mapped.get("target") or find(DeductionHead, ["other deduction", "target"]), "mobile": mapped.get("mobile") or find(DeductionHead, ["mobile"]), # Loan / EMI deduction (Loans & Advances module) "loan_emi": mapped.get("loan_emi") or find(DeductionHead, ["loan", "emi"]), } def _norm_head(s: str) -> str: return _normalize_dom_id((s or "").strip().lower()) def should_prorate_deduction(head_name: str, cfg: dict) -> bool: if not cfg or not cfg.get("enable_proration"): return False hn = _norm_head(head_name) force_no = {_norm_head(x) for x in (cfg.get("force_non_prorated_heads") or [])} if hn in force_no: return False mode = (cfg.get("prorate_mode") or "all_except").strip().lower() exceptions = {_norm_head(x) for x in (cfg.get("exception_heads") or [])} only_heads = {_norm_head(x) for x in (cfg.get("only_heads") or [])} if mode == "only": return hn in only_heads return hn not in exceptions def compute_monthly_deductions(deduction_rows, paid_days, month_days): cfg = get_json_setting("payroll_deduction_settings", { "enable_proration": True, "prorate_mode": "all_except", "exception_heads": ["pt"], "only_heads": [], "force_non_prorated_heads": ["pt", "loan emi", "loan_emi"], }) ratio = 0 if month_days and month_days > 0: ratio = float(paid_days) / float(month_days) computed = [] for row in deduction_rows: # row.head_name and row.amount assumed; change to your structure head = row.head_name amt = float(row.amount or 0) if should_prorate_deduction(head, cfg): amt = round(amt * ratio, 2) computed.append((head, amt)) return computed def _month_ratio(paid_days: float, year_i: int, month_i: int) -> float: mdays = monthrange(year_i, month_i)[1] or 0 return (float(paid_days) / float(mdays)) if mdays else 0.0 def compute_monthly_deductions_from_structures(emp_id: int, paid_days: float, year_i: int, month_i: int): """ Returns: - computed_list: list of (head_name, monthly_amount) - total: monthly total Uses your JSON setting rules: prorate all except PT (default). """ mdays = monthrange(year_i, month_i)[1] or 0 ratio = (float(paid_days) / float(mdays)) if mdays else 0.0 type_col, head_col = _ess_fields() d_rows = ( db.session.query( DeductionHead.name.label("head_name"), EmployeeSalaryStructures.amount.label("amount") ) .join(DeductionHead, DeductionHead.id == head_col) .filter(EmployeeSalaryStructures.employee_id == emp_id) .filter(type_col == "deduction") .all() ) class _Row: def __init__(self, head_name, amount): self.head_name = head_name self.amount = amount computed = compute_monthly_deductions([_Row(r.head_name, r.amount) for r in d_rows], paid_days, mdays) total = round(sum(a for _, a in computed), 2) return computed, total from typing import Optional def compute_monthly_deductions_from_frozen_map(d_map: dict, paid_days: float, month_days: int, payroll_total: Optional[float] = None): """ d_map: { head_name: master_amount } from FrozenSalaryComponent Returns: { head_name: monthly_amount }, total Applies should_prorate_deduction() rules. If payroll_total is provided, it will adjust prorated heads so the sum equals payroll_total while keeping non-prorated heads (e.g., PT) fixed. """ cfg = get_json_setting("payroll_deduction_settings", { "enable_proration": True, "prorate_mode": "all_except", "exception_heads": ["pt"], "only_heads": [], "force_non_prorated_heads": ["pt", "loan emi", "loan_emi"], }) ratio = (float(paid_days) / float(month_days)) if month_days else 0.0 fixed = {} # non-prorated (PT, etc.) var = {} # prorated heads for head, base in (d_map or {}).items(): base = float(base or 0.0) if should_prorate_deduction(head, cfg): var[head] = round(base * ratio, 2) else: fixed[head] = round(base, 2) # total before scaling out = {**fixed, **var} total = round(sum(out.values()), 2) # optional scaling so component sum matches Payroll.deductions exactly if payroll_total is not None: payroll_total = round(float(payroll_total or 0.0), 2) fixed_total = round(sum(fixed.values()), 2) remaining = round(payroll_total - fixed_total, 2) var_sum = round(sum(var.values()), 2) if var_sum > 0 and remaining >= 0: factor = remaining / var_sum for k in list(var.keys()): var[k] = round(var[k] * factor, 2) # handle rounding 0.01 drift by adjusting the last var head out = {**fixed, **var} drift = round(payroll_total - sum(out.values()), 2) if abs(drift) >= 0.01 and var: last = next(reversed(var.keys())) out[last] = round(out[last] + drift, 2) total = round(sum(out.values()), 2) return out, total # --------------------------------------------------------------------- # Earnings proration helpers (NEW) — mirrors deduction proration controls # --------------------------------------------------------------------- def should_prorate_earning(head_name: str, cfg: dict) -> bool: """ Earning proration rule (same structure as deduction rule). """ if not cfg or not cfg.get("enable_proration"): return False hn = _norm_head(head_name) force_no = {_norm_head(x) for x in (cfg.get("force_non_prorated_heads") or [])} if hn in force_no: return False mode = (cfg.get("prorate_mode") or "all_except").strip().lower() exceptions = {_norm_head(x) for x in (cfg.get("exception_heads") or [])} only_heads = {_norm_head(x) for x in (cfg.get("only_heads") or [])} if mode == "only": return hn in only_heads return hn not in exceptions def compute_monthly_earnings(earning_rows, paid_days, month_days): """ earning_rows: list-like with .head_name, .amount Returns: list of (head, monthly_amount) """ # ✅ Default keeps CURRENT behavior (everything prorated), # until you configure exceptions from Admin UI. cfg = get_json_setting("payroll_earning_settings", { "enable_proration": True, "prorate_mode": "all_except", "exception_heads": [], # e.g. ["incentive"] "only_heads": [], "force_non_prorated_heads": [], # e.g. ["incentive", "kpi incentive"] }) ratio = 0.0 if month_days and month_days > 0: ratio = float(paid_days) / float(month_days) computed = [] for row in earning_rows: head = row.head_name amt = float(row.amount or 0.0) if should_prorate_earning(head, cfg): amt = round(amt * ratio, 2) computed.append((head, amt)) return computed def compute_monthly_earnings_from_structures(emp_id: int, paid_days: float, year_i: int, month_i: int): """ Returns: - computed_list: list of (head_name, monthly_amount) - total: monthly gross total Uses your JSON setting rules for earnings proration. """ mdays = monthrange(year_i, month_i)[1] or 0 type_col, head_col = _ess_fields() e_rows = ( db.session.query( EarningHead.name.label("head_name"), EmployeeSalaryStructures.amount.label("amount") ) .join(EarningHead, EarningHead.id == head_col) .filter(EmployeeSalaryStructures.employee_id == emp_id) .filter(type_col == "earning") .all() ) class _Row: def __init__(self, head_name, amount): self.head_name = head_name self.amount = amount computed = compute_monthly_earnings([_Row(r.head_name, r.amount) for r in e_rows], paid_days, mdays) total = round(sum(a for _, a in computed), 2) return computed, total def compute_monthly_earnings_from_frozen_map(e_map: dict, paid_days: float, month_days: int, payroll_total: Optional[float] = None): """ e_map: { head_name: master_amount } from FrozenSalaryComponent Returns: { head_name: monthly_amount }, total Applies should_prorate_earning() rules. If payroll_total is provided, it will adjust prorated heads so the sum equals payroll_total while keeping non-prorated heads fixed. """ cfg = get_json_setting("payroll_earning_settings", { "enable_proration": True, "prorate_mode": "all_except", "exception_heads": [], "only_heads": [], "force_non_prorated_heads": [], }) ratio = (float(paid_days) / float(month_days)) if month_days else 0.0 fixed = {} # non-prorated heads var = {} # prorated heads for head, base in (e_map or {}).items(): base = float(base or 0.0) if should_prorate_earning(head, cfg): var[head] = round(base * ratio, 2) else: fixed[head] = round(base, 2) out = {**fixed, **var} total = round(sum(out.values()), 2) # optional scaling to match Payroll.gross_pay exactly (keeps non-prorated heads fixed) if payroll_total is not None: payroll_total = round(float(payroll_total or 0.0), 2) fixed_total = round(sum(fixed.values()), 2) remaining = round(payroll_total - fixed_total, 2) var_sum = round(sum(var.values()), 2) if var_sum > 0 and remaining >= 0: factor = remaining / var_sum for k in list(var.keys()): var[k] = round(var[k] * factor, 2) out = {**fixed, **var} drift = round(payroll_total - sum(out.values()), 2) if abs(drift) >= 0.01 and var: last = next(reversed(var.keys())) out[last] = round(out[last] + drift, 2) total = round(sum(out.values()), 2) return out, total 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 _ess_fields(): # type if hasattr(EmployeeSalaryStructures, "component_type"): type_col = EmployeeSalaryStructures.component_type elif hasattr(EmployeeSalaryStructures, "head_type"): type_col = EmployeeSalaryStructures.head_type else: type_col = EmployeeSalaryStructures.type # head id if hasattr(EmployeeSalaryStructures, "head_id"): head_col = EmployeeSalaryStructures.head_id elif hasattr(EmployeeSalaryStructures, "component_id"): head_col = EmployeeSalaryStructures.component_id else: head_col = EmployeeSalaryStructures.salary_head_id return type_col, head_col def _ess_get_type(row) -> str: if hasattr(row, "component_type"): return (row.component_type or "").lower() if hasattr(row, "head_type"): return (row.head_type or "").lower() return (row.type or "").lower() def _ess_get_head_id(row): if hasattr(row, "head_id"): return row.head_id if hasattr(row, "component_id"): return row.component_id return getattr(row, "salary_head_id", 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}" from calendar import monthrange from datetime import date, datetime def _emp_date_only(d): if d is None: return None return d.date() if isinstance(d, datetime) else d def _month_window(y: int, m: int): ms = date(y, m, 1) me = date(y, m, monthrange(y, m)[1]) return ms, me def _payroll_window_for_emp(emp: Employee, y: int, m: int): month_start, month_end = _month_window(y, m) doj = _emp_date_only(getattr(emp, "date_of_joining", None)) exit_d = _emp_date_only(getattr(emp, "exit_date", None)) start = month_start end = month_end if doj and doj > start: start = doj # ✅ Always clip to exit date if exists (even if exit month differs) if exit_d and exit_d < end: end = exit_d return start, end def _is_fnf_month(emp: Employee, y: int, m: int) -> bool: exit_d = _emp_date_only(getattr(emp, "exit_date", None)) return bool(exit_d and exit_d.year == y and exit_d.month == m) def _eligible_for_payroll(emp: Employee, y: int, m: int) -> bool: """ Eligible if employee was employed for ANY day in the month: - DOJ <= month_end - exit_date is None OR exit_date >= month_start This covers resigned employees even if status becomes 'inactive' later. """ month_start, month_end = _month_window(y, m) doj = _emp_date_only(getattr(emp, "date_of_joining", None)) if doj is None or doj > month_end: return False exit_d = _emp_date_only(getattr(emp, "exit_date", None)) if exit_d and exit_d < month_start: return False return True 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(): from sqlalchemy import asc # ---- Filters (GET) ---- inst_prefix = (request.args.get("inst") or "").strip() location = (request.args.get("location") or "").strip() dept_id = request.args.get("department_id", type=int) desig_id = request.args.get("designation_id", type=int) # employee_id can come from GET (when changing selection) or POST (when saving) employee_id = request.values.get("employee_id", type=int) # ---- Base employee query ---- emp_q = Employee.query.order_by(asc(Employee.full_name)) if inst_prefix: emp_q = emp_q.filter(Employee.employee_code.startswith(inst_prefix)) if location: emp_q = emp_q.filter(Employee.location == location) if dept_id: emp_q = emp_q.filter(Employee.department_id == dept_id) if desig_id: emp_q = emp_q.filter(Employee.designation_id == desig_id) employees = emp_q.all() # ---- Dropdown options: Institute (code prefix) & Location ---- codes = [ (row[0] or "") for row in db.session.query(Employee.employee_code) .filter(Employee.employee_code.isnot(None)) .all() ] prefixes = sorted({c.split("-")[0] for c in codes if c}) loc_rows = ( db.session.query(Employee.location) .filter(Employee.location.isnot(None), Employee.location != "") .distinct() .order_by(Employee.location) .all() ) locations = [r[0] for r in loc_rows] # ---- Salary heads ---- 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() selected_employee = Employee.query.get(employee_id) if employee_id else None # ---- Figure out field names on EmployeeSalaryStructures ---- # Type field: "component_type" | "head_type" | "type" if hasattr(EmployeeSalaryStructures, "component_type"): type_field = "component_type" elif hasattr(EmployeeSalaryStructures, "head_type"): type_field = "head_type" elif hasattr(EmployeeSalaryStructures, "type"): type_field = "type" else: type_field = None # in worst case, no type field # Head field: "head_id" | "component_id" | "salary_head_id" if hasattr(EmployeeSalaryStructures, "head_id"): head_field = "head_id" elif hasattr(EmployeeSalaryStructures, "component_id"): head_field = "component_id" elif hasattr(EmployeeSalaryStructures, "salary_head_id"): head_field = "salary_head_id" else: head_field = None # default structure container saved_data = {"earning": {}, "deduction": {}, "ctc": {}} # ---- Load existing structure, if any ---- if selected_employee and head_field: existing_rows = ( EmployeeSalaryStructures.query .filter(EmployeeSalaryStructures.employee_id == selected_employee.id) .all() ) for r in existing_rows: # type comp_type = "" if type_field: comp_type = (getattr(r, type_field, "") or "").lower() # head id head_id = getattr(r, head_field, None) if not head_id: continue if comp_type == "earning": saved_data["earning"][head_id] = r.amount elif comp_type == "deduction": saved_data["deduction"][head_id] = r.amount elif comp_type == "ctc": saved_data["ctc"][head_id] = r.amount # ---- Save on POST ---- if request.method == "POST" and selected_employee: action = request.form.get("action") if action == "save": # 🔹 Save payroll remarks from form selected_employee.payroll_remarks = (request.form.get("payroll_remarks") or "").strip() # wipe existing for that employee EmployeeSalaryStructures.query.filter_by( employee_id=selected_employee.id ).delete() # helper to create a row safely with dynamic field names def add_row(head_obj, kind: str, value_str: str): if not value_str.strip(): return if not head_field: return # can't store without knowing the head id column try: amt = float(value_str or 0) except ValueError: return kwargs = { "employee_id": selected_employee.id, "amount": amt, } # set head id field kwargs[head_field] = head_obj.id # set type field if it exists if type_field: kwargs[type_field] = kind # "earning" | "deduction" | "ctc" rec = EmployeeSalaryStructures(**kwargs) db.session.add(rec) # Earnings for head in earning_heads: val = request.form.get(f"earning_{head.id}", "") add_row(head, "earning", val) # Deductions for head in deduction_heads: val = request.form.get(f"deduction_{head.id}", "") add_row(head, "deduction", val) # CTC for head in ctc_heads: val = request.form.get(f"ctc_{head.id}", "") add_row(head, "ctc", val) db.session.commit() flash("✅ Salary structure saved.", "success") return redirect( url_for( "payroll.salary_structure", employee_id=selected_employee.id, inst=inst_prefix, location=location, department_id=dept_id, designation_id=desig_id, ) ) departments = Department.query.order_by(Department.name).all() designations = Designation.query.order_by(Designation.title).all() return render_template( "payroll/salary_structure.html", employees=employees, selected_employee=selected_employee, earning_heads=earning_heads, deduction_heads=deduction_heads, ctc_heads=ctc_heads, saved_data=saved_data, # filters + options code_prefixes=prefixes, selected_prefix=inst_prefix, locations=locations, selected_location=location, departments=departments, designations=designations, dept_id=dept_id, desig_id=desig_id, ) # --------------------------------------------------------------------- # Calculate (AJAX) — uses SalaryFormula with a simple two-pass eval # --------------------------------------------------------------------- def norm_key(name: str) -> str: return _normalize_dom_id(name or "") _ALLOWED_GLOBALS = { "__builtins__": {}, "min": min, "max": max, "abs": abs, "round": round, "int": int, "float": float, "math": math, } def safe_eval(expr: str, env: dict) -> float: if not expr or not expr.strip(): raise ValueError("Empty formula") blocked = ["import", "exec", "eval", "open", "os.", "sys.", "__", "lambda", "globals", "locals"] expr_l = expr.lower() if any(b in expr_l for b in blocked): raise ValueError("Unsafe expression") val = eval(expr, _ALLOWED_GLOBALS, env) return float(val) def calculate_all_components(gross: float) -> dict: gross = float(gross) if gross <= 0: raise ValueError("Gross must be > 0") earning_heads = EarningHead.query.order_by(EarningHead.id.asc()).all() deduction_heads = DeductionHead.query.order_by(DeductionHead.id.asc()).all() ctc_heads = CTCHead.query.order_by(CTCHead.id.asc()).all() # (component_type, component_id) -> formula f_map = {} for f in SalaryFormula.query.all(): ctype = (f.component_type or "").strip().lower() cid = int(f.component_id) f_map[(ctype, cid)] = (f.formula or "").strip() # ---- Build env ---- env = {"gross": gross} # seed all component keys so formulas can reference them safely for h in earning_heads: env[norm_key(h.name)] = 0.0 for h in deduction_heads: env[norm_key(h.name)] = 0.0 for h in ctc_heads: env[norm_key(h.name)] = 0.0 # ---- Add logical aliases (basic, hra, pf...) pointing to real head keys ---- aliases = _resolve_component_aliases() def _sync_alias_values(): """ Keep env['basic'] etc. in sync with env[actual_head_key]. Also, if formulas directly set env['basic'], reflect it back. """ for logical, head_name in (aliases or {}).items(): if not head_name: continue real_key = norm_key(head_name) # If real exists, push it into logical alias if real_key in env: env[logical] = float(env.get(real_key, 0.0) or 0.0) else: # still allow formulas to use logical keys even if head not found env.setdefault(logical, 0.0) _sync_alias_values() # all calculation targets targets = [] for h in earning_heads: targets.append(("earning", h.id, h.name)) for h in deduction_heads: targets.append(("deduction", h.id, h.name)) for h in ctc_heads: targets.append(("ctc", h.id, h.name)) pending = set((t[0], t[1]) for t in targets if (t[0], t[1]) in f_map) max_passes = 15 # multi-pass compute until dependencies settle for _ in range(max_passes): progressed = False still_pending = set() for (ctype, cid) in list(pending): formula_str = f_map.get((ctype, cid)) if not formula_str: continue # find head name to compute normalized key if ctype == "earning": head = next((h for h in earning_heads if h.id == cid), None) elif ctype == "deduction": head = next((h for h in deduction_heads if h.id == cid), None) else: head = next((h for h in ctc_heads if h.id == cid), None) if not head: continue key = norm_key(head.name) try: val = safe_eval(formula_str, env) if not math.isfinite(val) or val < 0: val = 0.0 env[key] = float(val) progressed = True # refresh logical aliases after every successful calculation _sync_alias_values() except NameError: still_pending.add((ctype, cid)) except Exception: still_pending.add((ctype, cid)) pending = still_pending if not progressed: break # return id->value mapping for JS (ids MUST match DOM ids) result = {} for h in earning_heads: k = norm_key(h.name) result[k] = float(env.get(k, 0.0)) for h in deduction_heads: k = norm_key(h.name) result[k] = float(env.get(k, 0.0)) for h in ctc_heads: k = norm_key(h.name) result[k] = float(env.get(k, 0.0)) return result @payroll_bp.route("/calculate-salary", methods=["POST"]) @login_required def calculate_salary(): payload = request.get_json(silent=True) or {} gross = payload.get("gross") try: if gross is None or str(gross).strip() == "": return jsonify({"success": False, "message": "Gross is required"}), 400 data = calculate_all_components(float(gross)) return jsonify({"success": True, "data": data}) except Exception as e: return jsonify({"success": False, "message": str(e)}), 400 # --------------------------------------------------------------------- # 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 type_col, head_col = _ess_fields() e_struct = dict( db.session.query(head_col, func.count()) .filter(type_col == "earning") .group_by(head_col) .all() ) d_struct = dict( db.session.query(head_col, func.count()) .filter(type_col == "deduction") .group_by(head_col) .all() ) c_struct = dict( db.session.query(head_col, func.count()) .filter(type_col == "ctc") .group_by(head_col) .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(): # ----- Read filters ----- search = (request.args.get("search") or "").strip() selected_institute = (request.args.get("institute") or "").strip() # Emp code prefix selected_location = (request.args.get("location") or "").strip() selected_department = request.args.get("department", type=int) # dept.id selected_designation = request.args.get("designation", type=int) # desig.id # ----- Base query: active employees ----- query = Employee.query.filter(Employee.status == "active") # Name search if search: query = query.filter(Employee.full_name.ilike(f"%{search}%")) # Company (emp code prefix) if selected_institute: query = query.filter(Employee.employee_code.ilike(f"{selected_institute}%")) # Location if selected_location: query = query.filter(Employee.location == selected_location) # Department if selected_department: query = query.filter(Employee.department_id == selected_department) # Designation if selected_designation: query = query.filter(Employee.designation_id == selected_designation) # Eager load relationships employees = ( query.options( joinedload(Employee.department), joinedload(Employee.designation), ) .order_by(Employee.full_name.asc()) .all() ) # ----- Masters for filters ----- departments = Department.query.order_by(Department.name).all() designations = Designation.query.order_by(Designation.title).all() # Institute list (emp code prefix) – reuse helper from routes.employee code_rows = db.session.query(Employee.employee_code).filter( Employee.employee_code.isnot(None) ).all() inst_set = set() for (code,) in code_rows: p = employee._extract_emp_prefix(code) # from routes.employee if p: inst_set.add(p) institutes = sorted(inst_set) # Locations loc_rows = db.session.query(Employee.location).filter( Employee.location.isnot(None), Employee.location != "" ).distinct().all() locations = sorted({loc for (loc,) in loc_rows}) # ----- Salary aggregation ----- data = [] for emp in employees: rows = EmployeeSalaryStructures.query.filter_by(employee_id=emp.id).all() gross = sum(float(s.amount or 0) for s in rows if _ess_get_type(s) == "earning") deductions = sum(float(s.amount or 0) for s in rows if _ess_get_type(s) == "deduction") ctc = sum(float(s.amount or 0) for s in rows if _ess_get_type(s) == "ctc") data.append({ "employee": emp, "gross": gross, "deductions": deductions, "net_pay": gross - deductions, "ctc": gross + ctc, "remarks": emp.payroll_remarks or "", }) return render_template( "payroll/view_salary_structures.html", data=data, departments=departments, designations=designations, institutes=institutes, locations=locations, # current filter values search=search, selected_department=selected_department, selected_designation=selected_designation, selected_institute=selected_institute, selected_location=selected_location, ) @payroll_bp.route("/salary-structures/remarks", methods=["POST"]) @login_required def update_salary_structure_remarks(): emp_id = request.form.get("employee_id", type=int) remarks = (request.form.get("payroll_remarks") or "").strip() if not emp_id: flash("Invalid employee.", "danger") return redirect(request.referrer or url_for("payroll.view_salary_structures")) emp = Employee.query.get_or_404(emp_id) emp.payroll_remarks = remarks db.session.commit() flash("Remarks updated.", "success") # Just go back to Overview with current filters preserved via referrer return redirect(request.referrer or url_for("payroll.view_salary_structures")) @payroll_bp.route("/view-salary-detail/") @login_required def view_salary_structure_detail(employee_id): emp = Employee.query.get_or_404(employee_id) rows = EmployeeSalaryStructures.query.filter_by(employee_id=employee_id).all() earnings, deductions, ctc_components = [], [], [] for r in rows: t = _ess_get_type(r) hid = _ess_get_head_id(r) amt = float(r.amount or 0.0) if t == "earning": h = EarningHead.query.get(hid) earnings.append({"name": (h.name if h else "Unknown"), "amount": amt}) elif t == "deduction": h = DeductionHead.query.get(hid) deductions.append({"name": (h.name if h else "Unknown"), "amount": amt}) elif t == "ctc": h = CTCHead.query.get(hid) ctc_components.append({"name": (h.name if h else "Unknown"), "amount": amt}) 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(): from sqlalchemy import asc # ---------- Month / Year ---------- 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 already frozen 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)) # ---------- Filters ---------- inst_prefix = (request.args.get("inst") or "").strip() location = (request.args.get("location") or "").strip() dept_id = request.args.get("department_id", type=int) desig_id = request.args.get("designation_id", type=int) # ---------- Month window ---------- total_days_in_month = monthrange(year_i, month_i)[1] month_start = date(year_i, month_i, 1) month_end = date(year_i, month_i, total_days_in_month) # ---------- Employee query ---------- emp_q = (Employee.query.filter(Employee.date_of_joining.isnot(None)).filter(Employee.date_of_joining <= month_end).filter(or_(Employee.exit_date.is_(None), Employee.exit_date >= month_start)).order_by(asc(Employee.full_name))) if inst_prefix: emp_q = emp_q.filter(Employee.employee_code.startswith(inst_prefix)) if location: emp_q = emp_q.filter(Employee.location == location) if dept_id: emp_q = emp_q.filter(Employee.department_id == dept_id) if desig_id: emp_q = emp_q.filter(Employee.designation_id == desig_id) employees = emp_q.all() emp_ids = [e.id for e in employees] # ---------- Dropdown options ---------- code_rows = db.session.query(Employee.employee_code).filter(Employee.employee_code.isnot(None)).all() inst_set = set() for (code,) in code_rows: if not code: continue p = employee._extract_emp_prefix(code) if p: inst_set.add(p) code_prefixes = sorted(inst_set) loc_rows = db.session.query(Employee.location).filter( Employee.location.isnot(None), Employee.location != "" ).distinct().all() locations = sorted({loc for (loc,) in loc_rows}) departments = Department.query.order_by(Department.name).all() designations = Designation.query.order_by(Designation.title).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(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" # ========================================================= # ✅ PREFETCH EVERYTHING (removes day-wise DB calls) # ========================================================= # 1) Holidays in month (single query) holiday_dates = { h.date for h in Holiday.query .filter(Holiday.date >= month_start, Holiday.date <= month_end) .all() } # 2) Attendance for all employees for month (single query) att_rows = ( Attendance.query .filter(Attendance.employee_id.in_(emp_ids)) .filter(Attendance.date >= month_start, Attendance.date <= month_end) .with_entities(Attendance.employee_id, Attendance.date, Attendance.remarks, Attendance.note) .all() ) att_map = { (eid, d): { "remarks": (rem or "").strip(), "note": (note or "").strip(), } for (eid, d, rem, note) in att_rows } # 3) Approved leaves overlapping the month (single query) leave_rows = ( LeaveRequest.query .filter(LeaveRequest.employee_id.in_(emp_ids)) .filter(LeaveRequest.status == "final_approved") .filter(LeaveRequest.start_date <= month_end, LeaveRequest.end_date >= month_start) .all() ) paid_leave_dates_by_emp = {eid: set() for eid in emp_ids} lwp_leave_dates_by_emp = {eid: set() for eid in emp_ids} for lr in leave_rows: eid = lr.employee_id cur = max(lr.start_date, month_start) stop = min(lr.end_date, month_end) target = lwp_leave_dates_by_emp if _is_lwp_leave(lr) else paid_leave_dates_by_emp while cur <= stop: target[eid].add(cur) cur += timedelta(days=1) # 4) Earnings/CTC totals grouped (avoid scalar per employee) type_col, _ = _ess_fields() earnings_rows = ( db.session.query(EmployeeSalaryStructures.employee_id, func.coalesce(func.sum(EmployeeSalaryStructures.amount), 0.0)) .filter(EmployeeSalaryStructures.employee_id.in_(emp_ids)) .filter(type_col == "earning") .group_by(EmployeeSalaryStructures.employee_id) .all() ) earnings_map = {eid: float(total or 0.0) for (eid, total) in earnings_rows} ctc_rows = ( db.session.query(EmployeeSalaryStructures.employee_id, func.coalesce(func.sum(EmployeeSalaryStructures.amount), 0.0)) .filter(EmployeeSalaryStructures.employee_id.in_(emp_ids)) .filter(type_col == "ctc") .group_by(EmployeeSalaryStructures.employee_id) .all() ) ctc_map = {eid: float(total or 0.0) for (eid, total) in ctc_rows} # 5) Last frozen month map (your code, already batched) last_month_subq = ( db.session.query( Payroll.employee_id.label("emp_id"), func.max(Payroll.month).label("last_month") ) .filter(Payroll.month < period) .group_by(Payroll.employee_id) .subquery() ) last_pay_rows = ( db.session.query(Payroll.employee_id, Payroll.month, Payroll.net_pay) .join(last_month_subq, and_( Payroll.employee_id == last_month_subq.c.emp_id, Payroll.month == last_month_subq.c.last_month )) .all() ) last_pay_map = {eid: {"month": m, "net_pay": float(net or 0.0)} for (eid, m, net) in last_pay_rows} # ---------- Build preview ---------- preview = [] mdays = total_days_in_month for emp in employees: if not _eligible_for_payroll(emp, year_i, month_i): continue # clip end day for FnF start_day, end_day = _payroll_window_for_emp(emp, year_i, month_i) if start_day > end_day: continue date_list = [start_day + timedelta(days=i) for i in range((end_day - start_day).days + 1)] present = holidays = weekends = lwp = 0.0 leave_count = 0.0 for day in date_list: weekday = day.weekday() is_weekend = (weekday == 6 and sun_off) or (weekday == 5 and is_saturday_off(day)) att_info = att_map.get((emp.id, day), {}) rem = (att_info.get("remarks") or "").strip() note = (att_info.get("note") or "").strip() is_holiday = day in holiday_dates is_paid_leave = day in paid_leave_dates_by_emp.get(emp.id, set()) is_lwp_leave = day in lwp_leave_dates_by_emp.get(emp.id, set()) # Priority: Holiday > Weekend > Manual Attendance Override > Approved Leave > LWP if is_holiday: holidays += 1 elif is_weekend: weekends += 1 elif rem: # ✅ manual/system attendance row should drive payroll if present p, lv, hol, wk, lw = _apply_attendance_remark_counts(rem, note) present += p leave_count += lv holidays += hol weekends += wk lwp += lw elif is_paid_leave: leave_count += 1 elif is_lwp_leave: lwp += 1 else: lwp += 1 present_total = present total_paid_days = present + leave_count + holidays + weekends earnings_master = float(earnings_map.get(emp.id, 0.0) or 0.0) # Keep your existing deduction policy function (still per-employee), # but overall load is now MUCH lower because attendance/holiday/leave are batched. ded_list, ded_total = compute_monthly_deductions_from_structures( emp.id, total_paid_days, year_i, month_i ) loan_due, _, _ = _loan_due_for_month(emp.id, period) deductions = round(float(ded_total or 0.0) + float(loan_due or 0.0), 2) ctc_total = float(ctc_map.get(emp.id, 0.0) or 0.0) last_info = last_pay_map.get(emp.id) # ✅ Earnings monthly using earning proration rules _, gross = compute_monthly_earnings_from_structures(emp.id, total_paid_days, year_i, month_i) net = gross - deductions preview.append({ "employee": emp, "present": round(present_total, 1), "leaves": leave_count, "holidays": holidays, "weekends": weekends, "lwp": lwp, "total_paid_days": round(total_paid_days, 2), "earnings": round(earnings_master, 2), "deductions": round(deductions, 2), "ctc": round(ctc_total, 2), "net_pay": round(net, 2), "last_frozen_month": (last_info["month"] if last_info else None), "last_frozen_net_pay": (round(last_info["net_pay"], 2) if last_info else None), "is_fnf": _is_fnf_month(emp, year_i, month_i), "remarks": emp.payroll_remarks or "", }) return render_template( "payroll/process_payroll_preview.html", payroll_preview=preview, month=month_i, year=year_i, period=period, code_prefixes=code_prefixes, selected_prefix=inst_prefix, locations=locations, selected_location=location, departments=departments, designations=designations, dept_id=dept_id, desig_id=desig_id, ) @payroll_bp.route("/enable-preview", methods=["POST"]) @login_required def enable_preview_for_month(): raw = (request.form.get("month") or "").strip() # can be 'YYYY-MM' or empty year = request.form.get("year") month = request.form.get("month_i") # if you used a separate field name # normalize to YYYY-MM for DB if raw and re.match(r"^\d{4}-\d{2}$", raw): month_yyyy_mm = raw y, m = raw.split("-") year_i, month_i = int(y), int(m) else: # fallback to separate ints year_i = int(year) month_i = int(month) month_yyyy_mm = f"{year_i:04d}-{month_i:02d}" rec = PayrollPreviewControl.query.filter_by(month=month_yyyy_mm).first() if not rec: rec = PayrollPreviewControl(month=month_yyyy_mm, 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_i, year=year_i)) # =========================== # Payroll Preview – Complete # =========================== # (Keep imports minimal here; db already imported above) # ---------- Helpers ---------- def _latest_preview_enabled_month() -> Optional[str]: """ 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: Optional[WeekendSettings], 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 = WeekendSettings.query.order_by(WeekendSettings.updated_on.desc()).first() sunday_off = bool(weekend and weekend.sunday == "off") present = leave = holiday = weekend_count = lwp = 0.0 for d in all_days: if doj and d < doj: continue weekday = d.weekday() is_weekend = (weekday == 6 and sunday_off) or (weekday == 5 and _wknd_is_saturday_off(weekend, 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() if holiday_entry: holiday += 1 continue if is_weekend: weekend_count += 1 continue if att and (att.remarks or "").strip(): p, lv, hol, wk, lw = _apply_attendance_remark_counts(att.remarks, getattr(att, "note", "")) present += p leave += lv holiday += hol weekend_count += wk lwp += lw continue if leave_entry: if _is_lwp_leave(leave_entry): lwp += 1 else: leave += 1 continue lwp += 1 full_month_days = monthrange(year_i, month_i)[1] paid_days = present + 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: t = _ess_get_type(row) hid = _ess_get_head_id(row) if t == "earning": head = EarningHead.query.get(hid) name = head.name if head else "Unknown" cfg_e = get_json_setting("payroll_earning_settings", { "enable_proration": True, "prorate_mode": "all_except", "exception_heads": [], "only_heads": [], "force_non_prorated_heads": [], }) amt0 = float(row.amount or 0.0) if should_prorate_earning(name, cfg_e): amt = round(amt0 * per_day_ratio, 2) else: amt = round(amt0, 2) total_earnings += amt earning_components.append({"name": name, "amount": amt}) elif t == "deduction": head = DeductionHead.query.get(hid) name = head.name if head else "Unknown" # ✅ apply proration rules (except PT by default) mdays = monthrange(year_i, month_i)[1] or 0 amt0 = float(row.amount or 0.0) cfg = get_json_setting("payroll_deduction_settings", { "enable_proration": True, "prorate_mode": "all_except", "exception_heads": ["pt"], "only_heads": [], "force_non_prorated_heads": ["pt", "loan emi", "loan_emi"], }) if should_prorate_deduction(name, cfg): amt = round(amt0 * per_day_ratio, 2) else: amt = round(amt0, 2) total_deductions += amt deduction_components[name] = amt elif t == "ctc": head = CTCHead.query.get(hid) name = head.name if head else "Unknown" amt = round((row.amount or 0.0), 2) total_ctc += amt ctc_components[name] = amt # ✅ Inject Loan EMI due for this month (same as Freeze) aliases = _resolve_component_aliases() loan_head_name = (aliases.get("loan_emi") or "Loan EMI").strip() loan_due, loan_emi_rows, emi_pay_map = _loan_due_for_month(emp.id, selected_month) if loan_due and loan_due > 0: deduction_components[loan_head_name] = round(float(loan_due), 2) total_deductions += float(loan_due) 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=0.0, # you can compute this if you have half-day logic lwp=lwp, ) # --------------------------------------------------------------------- # Freeze payroll # --------------------------------------------------------------------- @payroll_bp.route("/freeze-payroll", methods=["POST"]) @login_required def freeze_payroll(): selected_month = (request.form.get("month") or "").strip() # '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 aliases = _resolve_component_aliases() loan_head_name = (aliases.get("loan_emi") or "Loan EMI").strip() # ---- Block if already frozen ---- if Payroll.query.filter_by(month=selected_month).first(): flash(f"Payroll already frozen for {selected_month}.", "info") return redirect(url_for("payroll.payroll_summary", month=month_i, year=year_i)) # ---- 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] return getattr(weekend, f"{bucket}_saturday", None) == "off" # ---- Month window ---- month_start = date(year_i, month_i, 1) month_end = date(year_i, month_i, monthrange(year_i, month_i)[1]) # ---- Prefetch holidays once ---- holiday_dates = { h.date for h in Holiday.query .filter(Holiday.date >= month_start, Holiday.date <= month_end) .all() } # ---- Eligible employees ---- employees = (Employee.query.filter(Employee.date_of_joining.isnot(None)).filter(Employee.date_of_joining <= month_end).filter(or_(Employee.exit_date.is_(None), Employee.exit_date >= month_start)).all()) for emp in employees: if not _eligible_for_payroll(emp, year_i, month_i): continue doj = getattr(emp, "date_of_joining", None) if isinstance(doj, datetime): doj = doj.date() # Clip end date for F&F month start, end = _payroll_window_for_emp(emp, year_i, month_i) if start > end: continue # ---- Clean slate for this employee/month ---- FrozenSalaryComponent.query.filter_by(employee_id=emp.id, month=selected_month).delete(synchronize_session=False) Payroll.query.filter_by(employee_id=emp.id, month=selected_month).delete(synchronize_session=False) MonthlyPayroll.query.filter_by(employee_id=emp.id, month=month_i, year=year_i).delete(synchronize_session=False) # ---- Prefetch attendance for employee (map by date) ---- att_map = { a.date: { "remarks": (a.remarks or "").strip(), "note": (a.note or "").strip(), } for a in Attendance.query .filter(Attendance.employee_id == emp.id) .filter(Attendance.date >= start, Attendance.date <= end) .all() } # ---- Prefetch approved leave dates for employee ---- paid_leave_dates = set() lwp_leave_dates = set() approved_leaves = ( LeaveRequest.query .filter_by(employee_id=emp.id, status="final_approved") .filter(LeaveRequest.start_date <= end, LeaveRequest.end_date >= start) .all() ) for lr in approved_leaves: cur = max(lr.start_date, start) stop = min(lr.end_date, end) target_set = lwp_leave_dates if _is_lwp_leave(lr) else paid_leave_dates while cur <= stop: target_set.add(cur) cur += timedelta(days=1) # ---- Tally ---- present = leave = holiday = weekend_count = lwp = 0.0 cur = start while cur <= end: if cur < doj: cur += timedelta(days=1) continue weekday = cur.weekday() is_weekend = (weekday == 6 and sun_off) or (weekday == 5 and is_saturday_off(cur)) if cur in holiday_dates: holiday += 1 elif is_weekend: weekend_count += 1 else: att_info = att_map.get(cur, {}) rem = _safe_remark_text(att_info.get("remarks")) note = _safe_note_text(att_info.get("note")) if rem: p, lv, hol, wk, lw = _apply_attendance_remark_counts(rem, note) present += p leave += lv holiday += hol weekend_count += wk lwp += lw elif cur in paid_leave_dates: leave += 1 elif cur in lwp_leave_dates: lwp += 1 else: lwp += 1 cur += timedelta(days=1) paid_days = present + leave + holiday + weekend_count ratio = _month_ratio(paid_days, year_i, month_i) # ---- Freeze snapshot of master structure ---- structures = EmployeeSalaryStructures.query.filter_by(employee_id=emp.id).all() earnings_master = 0.0 ctc_master = 0.0 for item in structures: t = _ess_get_type(item) # earning|deduction|ctc hid = _ess_get_head_id(item) # id in the relevant head table amt = float(item.amount or 0.0) if t == "earning": head = EarningHead.query.get(hid) name = head.name if head else "Unknown" earnings_master += amt elif t == "deduction": head = DeductionHead.query.get(hid) name = head.name if head else "Unknown" # ✅ Skip manual Loan EMI from structures (we auto-inject it) if (name or "").strip().lower() == (loan_head_name or "").strip().lower(): continue elif t == "ctc": head = CTCHead.query.get(hid) name = head.name if head else "Unknown" ctc_master += amt else: continue db.session.add(FrozenSalaryComponent( employee_id=emp.id, month=selected_month, component_type=t, head_name=name, amount=amt, )) # ---- Earnings prorated by paid days ---- # ✅ Earnings monthly using earning proration rules (component-wise) _, gross = compute_monthly_earnings_from_structures(emp.id, paid_days, year_i, month_i) gross = round(float(gross or 0.0), 2) # ✅ Deductions prorated by paid days (except PT by default rule) _, deductions_monthly = compute_monthly_deductions_from_structures( emp.id, paid_days, year_i, month_i ) loan_due, loan_emi_rows, emi_pay_map = _loan_due_for_month(emp.id, selected_month) deductions_monthly = round(deductions_monthly + float(loan_due or 0.0),2) net = round(gross - deductions_monthly, 2) payroll_row = Payroll( employee_id=emp.id, month=selected_month, gross_pay=gross, deductions=deductions_monthly, net_pay=net, ctc=round(ctc_master, 2), present_days=present, approved_leaves=leave, holidays=holiday, weekends=weekend_count, lwp_days=lwp, ) db.session.add(payroll_row) db.session.flush() # 🔥 REQUIRED to get payroll_row.id # ------------------------------------------------------- # ✅ Freeze Loan EMI as deduction + mark EMI as paid # ------------------------------------------------------- if loan_due and loan_due > 0: # 1️⃣ Freeze as salary deduction head db.session.add(FrozenSalaryComponent( employee_id=emp.id, month=selected_month, component_type="deduction", head_name=loan_head_name, # alias-resolved "Loan EMI" amount=float(loan_due), )) # 2️⃣ Mark EMI schedule rows as paid + link payroll for emi in loan_emi_rows: payable = float(emi_pay_map.get(emi.id, 0.0) or 0.0) emi.status = "paid" emi.paid_on = date.today() emi.paid_amount = payable # could be partial payment emi.payroll_id = payroll_row.id # 3️⃣ Create Loan Transaction (CR) db.session.add(LoanTransaction( loan_id=emi.loan_id, txn_date=date.today(), txn_type="emi", direction="CR", amount=payable, # ✅ reduced EMI captured reference=selected_month, source="payroll", created_by_id=getattr(current_user, "id", None), remarks=f"EMI deducted via payroll {selected_month}", )) db.session.add(MonthlyPayroll( employee_id=emp.id, month=month_i, year=year_i, gross=gross, deductions=deductions_monthly, net_pay=net, ctc=round(ctc_master, 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(Attendance.employee_id == employee_id) .filter(Attendance.date >= start, Attendance.date <= end) .all() } dt_map = { l.date: l for l in DesktimeLog.query .filter(DesktimeLog.employee_id == employee_id) .filter(DesktimeLog.date >= start, DesktimeLog.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() } 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 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) split_status = None if att: status = (att.remarks or "").strip() or None note = (att.note or "").strip() 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" if status == "Half Day" and "HD_LEAVE:" in note: session = "SH" if "HD_LEAVE:SH" in note else "FH" other_half = "" if "SH:" in note: other_half = note.split("SH:")[1].split("|")[0].strip() if session == "FH": split_status = f"FH: Leave / SH: {other_half or 'LWP'}" else: split_status = f"FH: {other_half or 'LWP'} / SH: Leave" else: status = None note = "" check_in_time = None check_out_time = None working_hours = 0.0 source = "System" if not status: is_weekend = ( (d.weekday() == 6 and sun_off) or (d.weekday() == 5 and is_saturday_off(d)) ) if d in holiday_dates: status = "Holiday" elif is_weekend: status = "Weekend" source = "Weekend" 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, "split_status": split_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}" # ---------- NEW: Filters ---------- inst_prefix = (request.args.get("inst") or "").strip() # Company (Emp Code Prefix) location = (request.args.get("location") or "").strip() # Location / Workplace dept_id = request.args.get("department_id", type=int) # Department desig_id = request.args.get("designation_id", type=int) # Designation # ---------- Base query: join Employee so we can filter on it ---------- records_q = ( Payroll.query .options(joinedload(Payroll.employee)) .join(Employee, Payroll.employee_id == Employee.id) .filter(Payroll.month == period) ) # Company (emp code prefix) if inst_prefix: records_q = records_q.filter(Employee.employee_code.ilike(f"{inst_prefix}%")) # Location if location: records_q = records_q.filter(Employee.location == location) # Department if dept_id: records_q = records_q.filter(Employee.department_id == dept_id) # Designation if desig_id: records_q = records_q.filter(Employee.designation_id == desig_id) records = records_q.all() # ---------- Helpers ---------- 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) if records else 0.0 total_net = round(sum(r.net_pay for r in records), 2) if records else 0.0 # ---------- Dropdown options for filters ---------- # Company prefixes (emp code prefix) code_rows = db.session.query(Employee.employee_code).filter( Employee.employee_code.isnot(None) ).all() inst_set = set() for (code,) in code_rows: if not code: continue pfx = employee._extract_emp_prefix(code) # helper from routes.employee if pfx: inst_set.add(pfx) code_prefixes = sorted(inst_set) # Locations (Workplace) loc_rows = db.session.query(Employee.location).filter( Employee.location.isnot(None), Employee.location != "" ).distinct().all() locations = sorted({loc for (loc,) in loc_rows}) # Departments & Designations departments = Department.query.order_by(Department.name).all() designations = Designation.query.order_by(Designation.title).all() 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, # NEW: filter options + selected values code_prefixes=code_prefixes, selected_prefix=inst_prefix, locations=locations, selected_location=location, departments=departments, designations=designations, dept_id=dept_id, desig_id=desig_id, ) # --------------------------------------------------------------------- # 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", "hr admin", "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) earning_components, deduction_components, ctc_components, paid_days_from_payroll = _build_monthly_components_for_slip( employee_id=employee_id, month=month, payroll=payroll ) # If your template expects lists like FrozenSalaryComponent objects, # pass these lists instead (they are dicts with name/amount). earnings = earning_components deductions = deduction_components 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 = _as_date(getattr(employee, "date_of_joining", None)) exit_d = _as_date(getattr(employee, "exit_date", None)) # IMPORTANT from_date = max(month_start, doj) if doj else month_start to_date = month_end # ✅ clip to last working day / exit date (prevents weekends after exit from counting) if exit_d and exit_d < to_date: to_date = exit_d # ---- 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: 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: { "remarks": (a.remarks or "").strip(), "note": (a.note 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) rem_text = _safe_remark_text(rem).lower() if rem_text == "present": present_full += 1 elif rem_text == "half day": present_half += 1 elif rem_text == "leave": leave += 1 elif rem_text == "holiday": holiday += 1 elif rem_text == "weekend": weekend_count += 1 elif rem_text == "lwp" or rem_text is None: lwp += 1 else: lwp += 1 cur += timedelta(days=1) total_paid_days = present_full + (0.5 * present_half) + leave + holiday + weekend_count 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, } slip_template = get_salary_slip_template_for_employee(employee) earning_cfg = _build_component_config_map(slip_template, "earning") deduction_cfg = _build_component_config_map(slip_template, "deduction") ctc_cfg = _build_component_config_map(slip_template, "ctc") display_earnings = _apply_component_template_list(earnings, earning_cfg) display_deductions = _apply_component_template_list(deductions, deduction_cfg) display_ctc = _apply_component_template_list(ctc_components, ctc_cfg) display_fields = _build_salary_slip_display_fields(employee, attendance_summary, slip_template) template_logo_url = None if slip_template and slip_template.logo_path: template_logo_url = slip_template.logo_path else: theme = get_json_setting("theme", {}) if theme and theme.get("logo_url"): template_logo_url = theme.get("logo_url") else: template_logo_url = url_for("static", filename="images/logo.png") return render_template( "payroll/salary_slip.html", employee=employee, payroll=payroll, earnings=display_earnings, deductions=display_deductions, ctc_components=display_ctc, month=month, current_time=datetime.now(), attendance_summary=attendance_summary, slip_template=slip_template, employee_fields=display_fields["employee_fields"], company_fields=display_fields["company_fields"], attendance_fields=display_fields["attendance_fields"], template_logo_url=template_logo_url, ) @payroll_bp.route("/salary-slip///pdf") @login_required def download_salary_slip_pdf(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", "hr admin", "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) )) return _salary_slip_pdf_response(employee_id, month, disposition="attachment") # --------------------------------------------------------------------- # 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")) # ---------- Filters ---------- inst_prefix = (request.args.get("inst") or "").strip() location = (request.args.get("location") or "").strip() dept_id = request.args.get("department_id", type=int) desig_id = request.args.get("designation_id", type=int) records_q = ( Payroll.query .options(joinedload(Payroll.employee)) .join(Employee, Payroll.employee_id == Employee.id) .filter(Payroll.month == month) ) if inst_prefix: records_q = records_q.filter(Employee.employee_code.ilike(f"{inst_prefix}%")) if location: records_q = records_q.filter(Employee.location == location) if dept_id: records_q = records_q.filter(Employee.department_id == dept_id) if desig_id: records_q = records_q.filter(Employee.designation_id == desig_id) records = records_q.all() if not records: flash("No payroll records found for this month / filters.", "info") return redirect(url_for("payroll.payroll_summary")) # ---------- 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" loan_emi_name = aliases.get("loan_emi") or "Loan EMI" general_rows, master_rows, monthly_rows = [], [], [] 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 from frozen payroll row 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) paid_days = max(present_days + leave_days + holiday_days + weekend_days, 0.0) ratio = (paid_days / total_days_in_month) if total_days_in_month else 0.0 # Frozen master components (snapshot) frozen_components = FrozenSalaryComponent.query.filter_by(employee_id=emp.id, month=month).all() e_map, d_map = {}, {} for c in frozen_components: amt = float(c.amount or 0.0) if c.component_type == "earning": e_map[c.head_name] = e_map.get(c.head_name, 0.0) + amt elif c.component_type == "deduction": d_map[c.head_name] = d_map.get(c.head_name, 0.0) + amt # ---------------- 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_all_0 = gross_master - (basic_0 + hra_0 + ta_0 + ma_0 + inc_0) other_master = round(other_all_0, 2) residual_other_0 = other_all_0 - laptop_0 - kpi_0 - arrears_0 if residual_other_0 < 0: residual_other_0 = 0.0 # ✅ Monthly earnings map using earning proration rules (keeps non-prorated heads fixed) e_month_map, gross_m = compute_monthly_earnings_from_frozen_map( e_map=e_map, paid_days=paid_days, month_days=total_days_in_month, payroll_total=float(getattr(p, "gross_pay", 0.0) or 0.0) # force match Payroll.gross_pay ) basic_m = round(float(e_month_map.get(basic_name, 0.0) or 0.0), 2) hra_m = round(float(e_month_map.get(hra_name, 0.0) or 0.0), 2) ta_m = round(float(e_month_map.get(ta_name, 0.0) or 0.0), 2) ma_m = round(float(e_month_map.get(ma_name, 0.0) or 0.0), 2) inc_m = round(float(e_month_map.get(inc_name, 0.0) or 0.0), 2) laptop_m = round(float(e_month_map.get(laptop_name, 0.0) or 0.0), 2) kpi_m = round(float(e_month_map.get(kpi_name, 0.0) or 0.0), 2) arrears_m = round(float(e_month_map.get(arrear_name, 0.0) or 0.0), 2) # Other = whatever remains in gross after known buckets (same logic style as before) other_m = round( float(gross_m or 0.0) - (basic_m + hra_m + ta_m + ma_m + inc_m + laptop_m + kpi_m + arrears_m), 2 ) if other_m < 0: other_m = 0.0 gross_m = round(float(gross_m or 0.0), 2) # ---------------- ✅ Monthly deductions (prorated except PT) ---------------- # d_map is master amounts; compute component-wise monthly amounts using your rules. d_month_map, total_ded_m = compute_monthly_deductions_from_frozen_map( d_map=d_map, paid_days=paid_days, month_days=total_days_in_month, payroll_total=float(getattr(p, "deductions", 0.0) or 0.0) # force match Payroll.deductions ) pf_m = round(float(d_month_map.get(pf_name, 0.0) or 0.0), 2) esi_m = round(float(d_month_map.get(esi_name, 0.0) or 0.0), 2) tds_m = round(float(d_month_map.get(tds_name, 0.0) or 0.0), 2) pt_m = round(float(d_month_map.get(pt_name, 0.0) or 0.0), 2) adv_m = round(float(d_month_map.get(adv_name, 0.0) or 0.0), 2) target_m = round(float(d_month_map.get(target_name, 0.0) or 0.0), 2) mobile_m = round(float(d_month_map.get(mobile_name, 0.0) or 0.0), 2) loan_emi_m = round(float(d_month_map.get(loan_emi_name, 0.0) or 0.0), 2) total_ded_m = round(float(total_ded_m or 0.0), 2) # Net salary: use frozen payroll net net_salary_m = float(getattr(p, "net_pay", 0.0) or 0.0) employment_type_raw = (getattr(emp, "employment_type", None) or "payroll").strip().lower() is_stipend = employment_type_raw == "stipend" employment_type = "Stipend" if is_stipend else "Payroll" stipend_amount = net_salary_m if is_stipend else 0.0 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, "Employment Type": employment_type, "Stipend Amount": stipend_amount, "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, "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, "Loan EMI": loan_emi_m, "Net Salary": net_salary_m, }) # ---------- Filter dropdown data ---------- code_rows = db.session.query(Employee.employee_code).filter(Employee.employee_code.isnot(None)).all() inst_set = set() for (code,) in code_rows: if not code: continue pfx = employee._extract_emp_prefix(code) if pfx: inst_set.add(pfx) code_prefixes = sorted(inst_set) loc_rows = db.session.query(Employee.location).filter(Employee.location.isnot(None), Employee.location != "").distinct().all() locations = sorted({loc for (loc,) in loc_rows}) departments = Department.query.order_by(Department.name).all() designations = Designation.query.order_by(Designation.title).all() return render_template( "payroll/consolidated_salary_view.html", month=month, general_rows=general_rows, master_rows=master_rows, monthly_rows=monthly_rows, code_prefixes=code_prefixes, selected_prefix=inst_prefix, locations=locations, selected_location=location, departments=departments, designations=designations, dept_id=dept_id, desig_id=desig_id, ) @payroll_bp.route("/payroll/consolidated-download-excel") @login_required def download_consolidated_salary_excel(): import pandas as pd import io import math month = request.args.get("month") if not month: flash("Please provide a month", "danger") return redirect(url_for("payroll.consolidated_salary_view")) # ---------- Filters ---------- inst_prefix = (request.args.get("inst") or "").strip() location = (request.args.get("location") or "").strip() dept_id = request.args.get("department_id", type=int) desig_id = request.args.get("designation_id", type=int) records_q = ( Payroll.query .options(joinedload(Payroll.employee)) .join(Employee, Payroll.employee_id == Employee.id) .filter(Payroll.month == month) ) if inst_prefix: records_q = records_q.filter(Employee.employee_code.ilike(f"{inst_prefix}%")) if location: records_q = records_q.filter(Employee.location == location) if dept_id: records_q = records_q.filter(Employee.department_id == dept_id) if desig_id: records_q = records_q.filter(Employee.designation_id == desig_id) records = records_q.all() if not records: flash("No payroll records found for this month / filters.", "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" loan_emi_name = aliases.get("loan_emi") or "Loan EMI" year_i, month_i = map(int, month.split("-")) _, total_days_in_month = monthrange(year_i, month_i) consolidated_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) holiday_days = float(p.holidays or 0) weekend_days = float(p.weekends or 0) lwp_days = float(p.lwp_days or 0) paid_days = max(present_days + leave_days + holiday_days + weekend_days, 0.0) ratio = (paid_days / total_days_in_month) if total_days_in_month else 0.0 frozen_components = FrozenSalaryComponent.query.filter_by(employee_id=emp.id, month=month).all() e_map, d_map = {}, {} for c in frozen_components: amt = float(c.amount or 0.0) if c.component_type == "earning": e_map[c.head_name] = e_map.get(c.head_name, 0.0) + amt elif c.component_type == "deduction": d_map[c.head_name] = d_map.get(c.head_name, 0.0) + amt # 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_all_0 = gross_master - (basic_0 + hra_0 + ta_0 + ma_0 + inc_0) other_master = round(other_all_0, 2) residual_other_0 = other_all_0 - laptop_0 - kpi_0 - arrears_0 if residual_other_0 < 0: residual_other_0 = 0.0 # ✅ Monthly earnings map using earning proration rules (keeps non-prorated heads fixed) e_month_map, gross_m = compute_monthly_earnings_from_frozen_map( e_map=e_map, paid_days=paid_days, month_days=total_days_in_month, payroll_total=float(getattr(p, "gross_pay", 0.0) or 0.0) # force match Payroll.gross_pay ) basic_m = round(float(e_month_map.get(basic_name, 0.0) or 0.0), 2) hra_m = round(float(e_month_map.get(hra_name, 0.0) or 0.0), 2) ta_m = round(float(e_month_map.get(ta_name, 0.0) or 0.0), 2) ma_m = round(float(e_month_map.get(ma_name, 0.0) or 0.0), 2) inc_m = round(float(e_month_map.get(inc_name, 0.0) or 0.0), 2) laptop_m = round(float(e_month_map.get(laptop_name, 0.0) or 0.0), 2) kpi_m = round(float(e_month_map.get(kpi_name, 0.0) or 0.0), 2) arrears_m = round(float(e_month_map.get(arrear_name, 0.0) or 0.0), 2) # Other = whatever remains in gross after known buckets (same logic style as before) other_m = round( float(gross_m or 0.0) - (basic_m + hra_m + ta_m + ma_m + inc_m + laptop_m + kpi_m + arrears_m), 2 ) if other_m < 0: other_m = 0.0 gross_m = round(float(gross_m or 0.0), 2) # ✅ Monthly deductions (prorated except PT) — force match payroll.deductions d_month_map, total_ded_m = compute_monthly_deductions_from_frozen_map( d_map=d_map, paid_days=paid_days, month_days=total_days_in_month, payroll_total=float(getattr(p, "deductions", 0.0) or 0.0) ) pf_m = round(float(d_month_map.get(pf_name, 0.0) or 0.0), 2) esi_m = round(float(d_month_map.get(esi_name, 0.0) or 0.0), 2) tds_m = round(float(d_month_map.get(tds_name, 0.0) or 0.0), 2) pt_m = round(float(d_month_map.get(pt_name, 0.0) or 0.0), 2) adv_m = round(float(d_month_map.get(adv_name, 0.0) or 0.0), 2) target_m = round(float(d_month_map.get(target_name, 0.0) or 0.0), 2) mobile_m = round(float(d_month_map.get(mobile_name, 0.0) or 0.0), 2) loan_emi_m = round(float(d_month_map.get(loan_emi_name, 0.0) or 0.0), 2) base_net = float(getattr(p, "net_pay", 0.0) or 0.0) net_salary_m = int(math.ceil(base_net)) employment_type_raw = (getattr(emp, "employment_type", None) or "payroll").strip().lower() is_stipend = employment_type_raw == "stipend" employment_type = "Stipend" if is_stipend else "Payroll" stipend_amount = net_salary_m if is_stipend else 0.0 consolidated_rows.append({ "SN": i, "Employee Code": emp.employee_code or "", "Employee Name": emp.full_name or "", "Employment Type": employment_type, "Stipend Amount": stipend_amount, "Location": emp.location or "", "Department": emp.department.name if emp.department else "", "Institute": getattr(emp, "institute", "") or "", "UAN": emp.uan or "", "Present": present_days, "Leave": leave_days, "Holiday": holiday_days, "Weekend": weekend_days, "LWP": lwp_days, "Paid Days": paid_days, "Master Basic": basic_0, "Master HRA": hra_0, "Master TA": ta_0, "Master MA": ma_0, "Master Inc": inc_0, "Master Other": other_master, "Master Gross": gross_master, "Monthly Basic": basic_m, "Monthly HRA": hra_m, "Monthly TA": ta_m, "Monthly MA": ma_m, "Monthly Inc": inc_m, "Monthly Other": other_m, "Monthly Laptop": laptop_m, "Monthly KPI Incentive": kpi_m, "Monthly Arrear": arrears_m, "Monthly Gross": gross_m, "PF": pf_m, "ESI": esi_m, "TDS": tds_m, "PT": pt_m, "Advance": adv_m, "Target": target_m, "Mobile": mobile_m, "Loan EMI": loan_emi_m, "Net Salary": net_salary_m, }) df = pd.DataFrame(consolidated_rows) output = io.BytesIO() with pd.ExcelWriter(output, engine="openpyxl") as writer: df.to_excel(writer, sheet_name="Consolidated", 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 import io import math 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" loan_emi_name = aliases.get("loan_emi") or "Loan EMI" year_i, month_i = map(int, month.split("-")) _, total_days_in_month = monthrange(year_i, month_i) PRORATE_DEDUCTIONS = False # keep same as earlier 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) 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) # Gross from frozen (or ratio if needed) 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_m = ded_val(pf_name) 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) loan_emi_m = ded_val(loan_emi_name) total_ded_m = round(sum([pf_m, esi_m, tds_m, pt_m, adv_m, target_m, mobile_m, loan_emi_m]), 2) if getattr(p, "net_pay", None) is not None: net_salary_m = float(p.net_pay or 0.0) else: net_salary_m = round(gross_m - total_ded_m, 2) # Employment type + stipend amount employment_type_raw = (getattr(emp, "employment_type", None) or "payroll").strip().lower() is_stipend = employment_type_raw == "stipend" employment_type = "Stipend" if is_stipend else "Payroll" stipend_amount = net_salary_m if is_stipend else 0.0 rows.append({ "SN": i, "Employee Code": emp.employee_code or "", "Employee Name": getattr(emp, "full_name", None) or getattr(emp, "name", "") or "", "Employment Type": employment_type, # ✅ NEW "Stipend Amount": stipend_amount, # ✅ NEW "Account Holder Name": getattr(emp, "bank_account_name", None) or getattr(emp, "full_name", None) or getattr(emp, "name", "") or "", "Account No.": getattr(emp, "bank_account_number", "") or "", "Bank Name": getattr(emp, "bank_name", "") or "", "IFSC": getattr(emp, "ifsc_code", "") or "", "Branch Name": getattr(emp, "branch_name", "") or "", "Loan EMI": loan_emi_m, "Net Salary": net_salary_m, }) df = pd.DataFrame(rows) output = io.BytesIO() with pd.ExcelWriter(output, engine="openpyxl") 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", type=int) 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", "hr admin", "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") or "").strip() # '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")) year_i, month_i = map(int, month.split("-")) # Gather rows (before deletion for audit) payroll_q = Payroll.query.filter_by(month=month) mp_q = MonthlyPayroll.query.filter( MonthlyPayroll.month == month_i, MonthlyPayroll.year == year_i ) 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=month_i, year=year_i)) try: # --------------------------------------------------------- # ✅ FIX: Detach Loan EMI schedules linked to payroll rows # --------------------------------------------------------- payroll_ids = [pid for (pid,) in payroll_q.with_entities(Payroll.id).all()] if payroll_ids: emi_rows = ( LoanEMISchedule.query .filter(LoanEMISchedule.payroll_id.in_(payroll_ids)) .all() ) affected_loan_ids = {e.loan_id for e in emi_rows if e.loan_id} # Revert EMI schedule rows so they can be deducted again after re-freeze for emi in emi_rows: emi.payroll_id = None emi.status = "pending" # adjust if your system uses a different default emi.paid_on = None emi.paid_amount = None # Remove payroll-created loan transactions for this month if affected_loan_ids: LoanTransaction.query.filter( LoanTransaction.loan_id.in_(list(affected_loan_ids)), LoanTransaction.source == "payroll", LoanTransaction.reference == month, LoanTransaction.txn_type == "emi", LoanTransaction.direction == "CR", ).delete(synchronize_session=False) # --------------------------------------------------------- # ✅ Now it is safe to delete payroll artifacts # --------------------------------------------------------- fsc_q.delete(synchronize_session=False) payroll_q.delete(synchronize_session=False) mp_q.delete(synchronize_session=False) # Keep / re-enable preview for updated preview after fixes 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, proceed anyway pass db.session.commit() except Exception as e: db.session.rollback() flash(f"Unfreeze failed: {e}", "danger") return redirect(url_for("payroll.payroll_summary", month=month_i, year=year_i)) flash( f"Unfroze {month}: Payroll={count_payroll}, FrozenComponents={count_frozen}, Monthly={count_monthly}. " f"Loan EMIs detached + preview re-enabled.", "success" ) return redirect(url_for("payroll.payroll_summary", month=month_i, year=year_i)) @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") or "").strip() # '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")) year_i, month_i = map(int, month.split("-")) payroll_q = Payroll.query.filter_by(month=month, employee_id=employee_id) mp_q = MonthlyPayroll.query.filter_by( month=month_i, year=year_i, 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=month_i, year=year_i)) try: # --------------------------------------------------------- # ✅ FIX: Detach Loan EMI schedules linked to these payroll rows # --------------------------------------------------------- payroll_ids = [pid for (pid,) in payroll_q.with_entities(Payroll.id).all()] if payroll_ids: emi_rows = ( LoanEMISchedule.query .filter(LoanEMISchedule.payroll_id.in_(payroll_ids)) .all() ) affected_loan_ids = {e.loan_id for e in emi_rows if e.loan_id} for emi in emi_rows: emi.payroll_id = None emi.status = "pending" # adjust if your system uses a different default emi.paid_on = None emi.paid_amount = None if affected_loan_ids: LoanTransaction.query.filter( LoanTransaction.loan_id.in_(list(affected_loan_ids)), LoanTransaction.source == "payroll", LoanTransaction.reference == month, LoanTransaction.txn_type == "emi", LoanTransaction.direction == "CR", ).delete(synchronize_session=False) # --------------------------------------------------------- # ✅ Now delete payroll artifacts # --------------------------------------------------------- fsc_q.delete(synchronize_session=False) payroll_q.delete(synchronize_session=False) mp_q.delete(synchronize_session=False) # Re-enable preview for that month 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", month=month_i, year=year_i)) flash( f"Unfroze {month} for employee #{employee_id}: Payroll={count_payroll}, FrozenComponents={count_frozen}, " f"Monthly={count_monthly}. Loan EMIs detached + preview re-enabled.", "success" ) return redirect(url_for("payroll.payroll_summary", month=month_i, year=year_i)) # ---------- 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")) @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", "loan_emi" ] 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, ) @payroll_bp.route("/salary-structure/bulk-template") @login_required def bulk_salary_structure_template(): """ Download an Excel template with all employees and all salary components. Columns: Employee Code | Employee Name | | | Values for components are prefilled with existing salary structure amounts (if any). """ import pandas as pd import io from openpyxl.utils import get_column_letter from openpyxl.styles import Font, Alignment, PatternFill from openpyxl.worksheet.datavalidation import DataValidation # --- Get heads dynamically --- 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() # Column order columns = ["Employee Code", "Employee Name"] columns += [h.name for h in earning_heads] columns += [h.name for h in deduction_heads] columns += [h.name for h in ctc_heads] # (type, head_id) -> head_name head_name_by_key = {} for h in earning_heads: head_name_by_key[("earning", h.id)] = h.name for h in deduction_heads: head_name_by_key[("deduction", h.id)] = h.name for h in ctc_heads: head_name_by_key[("ctc", h.id)] = h.name rows = [] employees = ( Employee.query .filter(Employee.status.in_(["active", "notice_period", "resigned"])) .order_by(Employee.full_name) .all() ) for emp in employees: row = {col: "" for col in columns} row["Employee Code"] = emp.employee_code or "" row["Employee Name"] = emp.full_name or (getattr(emp, "name", "") or "") structs = EmployeeSalaryStructures.query.filter_by(employee_id=emp.id).all() for s in structs: key = (s.type, s.head_id) head_name = head_name_by_key.get(key) if head_name: row[head_name] = float(s.amount or 0.0) rows.append(row) df = pd.DataFrame(rows, columns=columns) output = io.BytesIO() with pd.ExcelWriter(output, engine="openpyxl") as writer: df.to_excel(writer, sheet_name="SalaryStructure", index=False) ws = writer.sheets["SalaryStructure"] # ✅ Freeze header row (openpyxl way) ws.freeze_panes = "A2" # ✅ Header styling (optional but nice) header_fill = PatternFill("solid", fgColor="F2F2F2") header_font = Font(bold=True) header_align = Alignment(horizontal="center", vertical="center", wrap_text=True) for cell in ws[1]: cell.fill = header_fill cell.font = header_font cell.alignment = header_align # ✅ Set widths (openpyxl way) def set_width(col_idx_1_based: int, width: float): ws.column_dimensions[get_column_letter(col_idx_1_based)].width = width set_width(1, 18) # Employee Code set_width(2, 30) # Employee Name # For component columns, use a reasonable default width for c in range(3, len(columns) + 1): set_width(c, 18) # ✅ Optional: number format for component cells # (all columns from 3rd onwards are amounts) for row_i in range(2, ws.max_row + 1): for col_i in range(3, len(columns) + 1): ws.cell(row=row_i, column=col_i).number_format = "0.00" # ✅ Optional: Add numeric validation (prevents text like "abc") # Applies to all component columns, rows 2..end last_col = get_column_letter(len(columns)) dv = DataValidation(type="decimal", operator="greaterThanOrEqual", formula1="0", allow_blank=True) dv.error = "Please enter a valid number (0 or more)." dv.errorTitle = "Invalid value" dv.prompt = "Enter numeric amount. Leave blank if not applicable." dv.promptTitle = "Amount" ws.add_data_validation(dv) dv.add(f"C2:{last_col}{ws.max_row}") output.seek(0) filename = "Salary_Structure_Bulk_Template.xlsx" resp = make_response(output.read()) resp.headers["Content-Disposition"] = f"attachment; filename={filename}" resp.headers["Content-Type"] = "application/vnd.openxmlformats-officedocument.spreadsheetml.sheet" return resp @payroll_bp.route("/salary-structure/bulk-upload", methods=["GET", "POST"]) @login_required def bulk_salary_structure_upload(): """ UI + processing for bulk assign salary via Excel. Expects the same column structure as /salary-structure/bulk-template. Matching is done by Employee Code and head names. """ import pandas as pd import io import math if request.method == "GET": # Simple upload page return render_template("payroll/salary_structure_bulk_upload.html") # --- POST: process uploaded file --- file = request.files.get("file") if not file or file.filename == "": flash("Please select an Excel file to upload.", "danger") return redirect(url_for("payroll.bulk_salary_structure_upload")) try: # Pandas can read file-like directly df = pd.read_excel(file) except Exception as e: current_app.logger.exception("Bulk salary upload: failed to read Excel") flash(f"Could not read Excel file: {e}", "danger") return redirect(url_for("payroll.bulk_salary_structure_upload")) # Normalize column names (strip spaces) df.columns = [str(c).strip() for c in df.columns] # Identify employee code column (support a couple of variants) emp_code_col = None for c in df.columns: lc = c.lower().replace(" ", "_") if lc in ("employee_code", "employee_code.", "emp_code", "employee_code*"): emp_code_col = c break if c.lower().strip() == "employee code": emp_code_col = c break if not emp_code_col: # fallback: exact "Employee Code" if "Employee Code" in df.columns: emp_code_col = "Employee Code" if not emp_code_col: flash("Template must have an 'Employee Code' column.", "danger") return redirect(url_for("payroll.bulk_salary_structure_upload")) # We ignore Employee Name for matching (just for HR reference) ignore_cols = {emp_code_col, "Employee Name", "employee name"} # --- Build head name index dynamically (multi-company-safe) --- 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() # Map: normalized name -> list of (type, head_id, display_name) head_index = {} def add_to_index(model_list, typ): for h in model_list: name = (h.name or "").strip() if not name: continue key = name.lower() head_index.setdefault(key, []).append((typ, h.id, name)) add_to_index(earning_heads, "earning") add_to_index(deduction_heads, "deduction") add_to_index(ctc_heads, "ctc") # Map Excel columns -> (type, head_id, display_name) column_mapping = {} unknown_columns = [] ambiguous_columns = [] for col in df.columns: if col in ignore_cols: continue key = col.strip().lower() matches = head_index.get(key, []) if not matches: unknown_columns.append(col) continue if len(matches) > 1: # Same name appears in more than one table ambiguous_columns.append(col) continue typ, head_id, display_name = matches[0] column_mapping[col] = (typ, head_id, display_name) if unknown_columns: flash(f"Skipped unknown component columns: {', '.join(unknown_columns)}", "warning") if ambiguous_columns: flash( "Skipped ambiguous component columns (same name in multiple tables): " + ", ".join(ambiguous_columns), "warning", ) if not column_mapping: flash("No valid component columns found in the file.", "danger") return redirect(url_for("payroll.bulk_salary_structure_upload")) # --- Process rows --- processed_employees = set() missing_codes = [] created_struct_rows = 0 # To avoid repeated deletes, track which employee_ids we already wiped wiped_employee_ids = set() for idx, row in df.iterrows(): raw_code = row.get(emp_code_col) if raw_code is None or (isinstance(raw_code, float) and math.isnan(raw_code)): continue emp_code = str(raw_code).strip() if not emp_code: continue emp = Employee.query.filter_by(employee_code=emp_code).first() if not emp: missing_codes.append(emp_code) continue emp_id = emp.id # Delete existing structures only once per employee if emp_id not in wiped_employee_ids: EmployeeSalaryStructures.query.filter_by(employee_id=emp_id).delete() wiped_employee_ids.add(emp_id) # For each component column, create EmployeeSalaryStructures row if value present for col, (typ, head_id, display_name) in column_mapping.items(): val = row.get(col) # Treat NaN/empty as skip if val is None: continue if isinstance(val, float) and math.isnan(val): continue try: amount = float(val) except (TypeError, ValueError): # If value is non-numeric, skip it for now continue # Skip zeros to keep data cleaner (optional) if abs(amount) < 0.000001: continue db.session.add(EmployeeSalaryStructures( employee_id=emp_id, head_id=head_id, type=typ, amount=amount, )) created_struct_rows += 1 processed_employees.add(emp_code) try: db.session.commit() except Exception as e: db.session.rollback() current_app.logger.exception("Bulk salary upload: DB error") flash(f"Database error while saving salary structures: {e}", "danger") return redirect(url_for("payroll.bulk_salary_structure_upload")) # --- Flash summary --- msg = ( f"Bulk salary upload completed. " f"Employees updated: {len(processed_employees)}; " f"Salary component rows saved: {created_struct_rows}." ) if missing_codes: unique_missing = sorted(set(missing_codes)) msg += f" Employee codes not found (skipped): {', '.join(unique_missing[:20])}" if len(unique_missing) > 20: msg += f" ... (+{len(unique_missing) - 20} more)" flash(msg, "success") return redirect(url_for("payroll.view_salary_structures")) @payroll_bp.route("/update-payroll-remarks/", methods=["POST"]) @login_required def update_payroll_remarks(employee_id): emp = Employee.query.get_or_404(employee_id) remarks = (request.form.get("payroll_remarks") or "").strip() emp.payroll_remarks = remarks db.session.commit() flash("Payroll remarks updated.", "success") return redirect(url_for("payroll.view_salary_structure_detail", employee_id=employee_id)) def get_salary_slip_template_for_employee(employee): if not employee: return None location = getattr(employee, "location", None) institute = getattr(employee, "institute", None) template = SalarySlipTemplate.query.filter_by( is_active=True, scope_type="location_institute", location=location, institute=institute ).first() if template: return template template = SalarySlipTemplate.query.filter_by( is_active=True, scope_type="institute", institute=institute ).first() if template: return template template = SalarySlipTemplate.query.filter_by( is_active=True, scope_type="location", location=location ).first() if template: return template template = SalarySlipTemplate.query.filter_by( is_active=True, scope_type="global" ).first() return template def _build_field_config_map(template, section_name): if not template: return {} rows = ( SalarySlipFieldConfig.query .filter_by(template_id=template.id, section_name=section_name) .order_by(SalarySlipFieldConfig.display_order.asc(), SalarySlipFieldConfig.id.asc()) .all() ) return { r.field_key: { "label": r.display_label or r.field_key.replace("_", " ").title(), "visible": bool(r.is_visible), "order": r.display_order or 0, } for r in rows } def _build_component_config_map(template, component_type): if not template: return {} rows = ( SalarySlipComponentConfig.query .filter_by(template_id=template.id, component_type=component_type) .order_by(SalarySlipComponentConfig.display_order.asc(), SalarySlipComponentConfig.id.asc()) .all() ) return { r.head_name: { "label": r.display_label or r.head_name, "visible": bool(r.is_visible), "order": r.display_order or 0, } for r in rows } def _apply_component_template_list(items, config_map): """ items = list of dicts like [{"name": "...", "amount": ...}] returns filtered/renamed/sorted list """ if not config_map: return items output = [] for item in items: name = item.get("name") amount = item.get("amount", 0) cfg = config_map.get(name) if cfg is None: output.append({"name": name, "amount": amount, "_order": 9999}) continue if not cfg.get("visible", True): continue output.append({ "name": cfg.get("label") or name, "amount": amount, "_order": cfg.get("order", 9999) }) output.sort(key=lambda x: (x.get("_order", 9999), (x.get("name") or "").lower())) for row in output: row.pop("_order", None) return output def _apply_component_template_dict(items_dict, config_map): """ items_dict = {"PF": 1000, "PT": 200} returns list of dicts like [{"name": "PF", "amount": 1000}] """ base_items = [{"name": k, "amount": v} for k, v in items_dict.items()] return _apply_component_template_list(base_items, config_map) def _apply_ctc_template_dict(items_dict, config_map): base_items = [{"name": k, "amount": v} for k, v in items_dict.items()] return _apply_component_template_list(base_items, config_map) def _build_salary_slip_display_fields(employee, attendance_summary, template): employee_cfg = _build_field_config_map(template, "employee") company_cfg = _build_field_config_map(template, "company") attendance_cfg = _build_field_config_map(template, "attendance") department_obj = getattr(employee, "department", None) designation_obj = getattr(employee, "designation", None) department_value = ( getattr(department_obj, "name", None) or getattr(department_obj, "title", None) or getattr(employee, "department_name", None) or "" ) designation_value = ( getattr(designation_obj, "name", None) or getattr(designation_obj, "title", None) or getattr(employee, "designation_name", None) or "" ) employee_source = { "employee_name": getattr(employee, "full_name", "") or "", "employee_code": getattr(employee, "employee_code", "") or "", "department": department_value, "designation": designation_value, "date_of_joining": employee.date_of_joining.strftime("%d-%m-%Y") if getattr(employee, "date_of_joining", None) else "", "uan": getattr(employee, "uan", "") or "", "pan_number": getattr(employee, "pan_number", "") or "", "bank_name": getattr(employee, "bank_name", "") or "", "bank_account_number": getattr(employee, "bank_account_number", "") or "", "ifsc_code": getattr(employee, "ifsc_code", "") or "", "location": getattr(employee, "location", "") or "", } company_source = { "company_name": getattr(template, "company_name", "") if template else "", "company_address": getattr(template, "company_address", "") if template else "", "company_contact": getattr(template, "company_contact", "") if template else "", "footer_text": getattr(template, "footer_text", "") if template else "", } attendance_source = { "total_days": attendance_summary.get("total_days", 0), "present": attendance_summary.get("present", 0), "leave": attendance_summary.get("leave", 0), "lwp": attendance_summary.get("lwp", 0), "holidays": attendance_summary.get("holidays", 0), "weekends": attendance_summary.get("weekends", 0), } def build_section(source, cfg_map): rows = [] if cfg_map: ordered = sorted(cfg_map.items(), key=lambda x: (x[1].get("order", 9999), x[0])) for key, cfg in ordered: if not cfg.get("visible", True): continue rows.append({ "key": key, "label": cfg.get("label", key.replace("_", " ").title()), "value": source.get(key, "") }) else: for key, value in source.items(): rows.append({ "key": key, "label": key.replace("_", " ").title(), "value": value }) return rows return { "employee_fields": build_section(employee_source, employee_cfg), "company_fields": build_section(company_source, company_cfg), "attendance_fields": build_section(attendance_source, attendance_cfg), } @payroll_bp.route("/salary-slip///print") @login_required def print_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", "hr admin", "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.print_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) earning_components, deduction_components, ctc_components, paid_days_from_payroll = _build_monthly_components_for_slip( employee_id=employee_id, month=month, payroll=payroll ) 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 = _as_date(getattr(employee, "date_of_joining", None)) exit_d = _as_date(getattr(employee, "exit_date", None)) from_date = max(month_start, doj) if doj else month_start to_date = month_end if exit_d and exit_d < to_date: to_date = exit_d 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) 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: { "remarks": (a.remarks or "").strip(), "note": (a.note 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) rem_text = _safe_remark_text(rem).lower() if rem_text == "present": present_full += 1 elif rem_text == "half day": present_half += 1 elif rem_text == "leave": leave += 1 elif rem_text == "holiday": holiday += 1 elif rem_text == "weekend": weekend_count += 1 elif rem_text == "lwp" or rem_text is None: lwp += 1 else: lwp += 1 cur += timedelta(days=1) total_paid_days = present_full + (0.5 * present_half) + leave + holiday + weekend_count 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, } slip_template = get_salary_slip_template_for_employee(employee) earning_cfg = _build_component_config_map(slip_template, "earning") deduction_cfg = _build_component_config_map(slip_template, "deduction") ctc_cfg = _build_component_config_map(slip_template, "ctc") display_earnings = _apply_component_template_list(earning_components, earning_cfg) display_deductions = _apply_component_template_list(deduction_components, deduction_cfg) display_ctc = _apply_component_template_list(ctc_components, ctc_cfg) display_fields = _build_salary_slip_display_fields(employee, attendance_summary, slip_template) template_logo_url = None if slip_template and slip_template.logo_path: template_logo_url = slip_template.logo_path else: theme = get_json_setting("theme", {}) if theme and theme.get("logo_url"): template_logo_url = theme.get("logo_url") else: template_logo_url = url_for("static", filename="images/logo.png") return render_template( "payroll/salary_slip_print.html", employee=employee, payroll=payroll, earnings=display_earnings, deductions=display_deductions, ctc_components=display_ctc, month=month, current_time=datetime.now(), attendance_summary=attendance_summary, slip_template=slip_template, employee_fields=display_fields["employee_fields"], company_fields=display_fields["company_fields"], attendance_fields=display_fields["attendance_fields"], template_logo_url=template_logo_url )