from datetime import date from calendar import monthrange from flask import Blueprint, render_template, request, redirect, url_for, flash, jsonify from flask_login import login_required, current_user from sqlalchemy import or_, func from extensions import db from models import ( Employee, Department, Loan, LoanEMISchedule, LoanTransaction, LoanSkipRequest, LoanExtraPaymentRequest,LoanType, LoanSetting, Payroll, ) from typing import Optional from sqlalchemy.orm import joinedload import math from datetime import datetime from flask import current_app import uuid import tempfile import os from datetime import date import pandas as pd from werkzeug.utils import secure_filename from flask import send_file loan_bp = Blueprint("loan", __name__, url_prefix="/loan") def _build_emi_schedule_view(loan: Loan): """Return a list of objects suitable for the loan detail template.""" today = date.today() items = [] for emi in sorted(loan.emi_schedule, key=lambda e: e.due_month): status = (emi.status or "pending").lower() # Treat EMI due date as last calendar day of due_month y, m = emi.due_month.year, emi.due_month.month last_day = monthrange(y, m)[1] due_date = date(y, m, last_day) # Is payroll frozen for this EMI month? month_str = emi.due_month.strftime("%Y-%m") frozen_payroll = Payroll.query.filter_by( employee_id=loan.employee_id, month=month_str, ).first() can_skip = (status == "pending") and (frozen_payroll is None) items.append( { "id": emi.id, "month_label": emi.due_month.strftime("%b %Y"), "amount": float(emi.amount or 0), "status": status, "paid": status == "paid", "is_overdue": status == "pending" and emi.due_month < today, "due_date": due_date, "due_date_label": due_date.strftime("%d-%b-%Y"), "can_skip": can_skip, } ) return items def _last_paid_emi_date(loan: Loan): paid_dates = [ e.due_month for e in loan.emi_schedule if (e.status or "").lower() == "paid" ] return max(paid_dates) if paid_dates else None def _require_loan_admin(): """Allow only Admin / HR / HR Admin / HR Manager (same as admin_detail).""" raw_role = getattr(current_user, "role", None) if isinstance(raw_role, str): role_name = raw_role.lower() else: role_name = (getattr(raw_role, "name", "") or "").lower() allowed_roles = {"admin", "system administrator", "hr", "hr admin", "hr manager"} if role_name not in allowed_roles and not getattr(current_user, "is_admin", False): flash("Only HR/Admin can manage loan requests.", "danger") return False return True def _loan_display_type(loan: Loan) -> str: """ Return user-facing loan type using LoanType.label. Works even if Loan.loan_type currently stores either name or label. """ if not loan.loan_type: return "" lt = ( LoanType.query .filter( or_( LoanType.name == loan.loan_type, LoanType.label == loan.loan_type, ) ) .first() ) return lt.label if lt else loan.loan_type def _create_emi_schedule(loan: Loan): """ Build (or rebuild) EMI schedule rows for a loan. Assumes loan.start_month, loan.tenure_months and loan.emi_amount are set. """ # Remove any old schedule rows for this loan for emi in list(loan.emi_schedule): db.session.delete(emi) if not loan.start_month or not loan.tenure_months or not loan.emi_amount: return base = loan.start_month emi_amount = loan.emi_amount tenure = loan.tenure_months for i in range(tenure): month_index = (base.month - 1) + i year = base.year + month_index // 12 month = (month_index % 12) + 1 due_month = date(year, month, 1) row = LoanEMISchedule( loan_id=loan.id, due_month=due_month, amount=emi_amount, status="pending", ) db.session.add(row) def _ensure_disbursement_txn(loan: Loan, created_by_id: int = None): """ Create a single DR transaction for the sanctioned amount if not present. Used when loan is approved. """ existing = LoanTransaction.query.filter_by( loan_id=loan.id, txn_type="disbursement", ).first() if existing: return amt = loan.sanctioned_amount or loan.requested_amount if not amt: return txn = LoanTransaction( loan_id=loan.id, txn_date=date.today(), txn_type="disbursement", direction="DR", # company gave money to employee amount=amt, reference=f"Loan #{loan.id} sanction", source="manual", created_by_id=created_by_id, remarks="Initial loan disbursement", ) db.session.add(txn) # routes/loan.py (only the two views changed parts) def _loan_summary_numbers(loan: Loan): """Return (amount, tenure, emi, total_paid, outstanding) for display.""" # Base amount – prefer sanctioned, else requested amount = float(loan.sanctioned_amount or loan.requested_amount or 0) # Tenure – prefer final tenure, else requested tenure = loan.tenure_months or loan.requested_tenure_months # EMI – if not stored, do simple (principal / tenure) as placeholder emi = float(loan.emi_amount or 0) if emi == 0 and amount and tenure: emi = amount / tenure # Total paid so far = sum of CR transactions total_paid = sum( float(t.amount or 0) for t in loan.transactions if (t.direction or "").upper() == "CR" ) # True outstanding from transactions outstanding = loan.outstanding_amount() # If there is no DR/CR yet, outstanding_amount() returns 0. # For Approved / Active loans, show "amount - total_paid" instead. if outstanding == 0 and loan.status in ("approved", "active"): outstanding = amount - total_paid return amount, tenure, emi, total_paid, outstanding # ------------------------------------------------------------------- # Employee: My Loans # ------------------------------------------------------------------- @loan_bp.route("/my") @login_required def my_loans(): emp: Employee = current_user q = Loan.query.filter(Loan.employee_id == emp.id) search = (request.args.get("q") or "").strip() if search: like = f"%{search}%" q = q.filter(or_( Loan.loan_type.ilike(like), Loan.status.ilike(like), )) loans = q.order_by(Loan.requested_on.desc()).all() # Attach display label for each loan for l in loans: l.display_loan_type = _loan_display_type(l) active_statuses = ("approved", "active") pending_statuses = ("requested", "under_review") active_loans = [l for l in loans if l.status in active_statuses] pending_loans = [l for l in loans if l.status in pending_statuses] # --- Total active amount (with fallback when no transactions yet) --- def _display_outstanding(l: Loan) -> float: amount = float(l.sanctioned_amount or l.requested_amount or 0) total_paid = sum( float(t.amount or 0) for t in l.transactions if (t.direction or "").upper() == "CR" ) bal = l.outstanding_amount() if bal == 0 and l.status in active_statuses: bal = amount - total_paid return bal total_active_amount = sum((_display_outstanding(l) for l in active_loans), 0.0) # --- Monthly EMI (fallback = simple principal/tenure) --- def _display_emi(l: Loan) -> float: amount = float(l.sanctioned_amount or l.requested_amount or 0) tenure = l.tenure_months or l.requested_tenure_months emi = float(l.emi_amount or 0) if emi == 0 and amount and tenure: emi = amount / tenure return emi total_monthly_emi = sum((_display_emi(l) for l in active_loans), 0.0) today = date.today() overdue_emi_count = ( LoanEMISchedule.query .join(Loan, LoanEMISchedule.loan_id == Loan.id) .filter( Loan.employee_id == emp.id, Loan.status.in_(active_statuses), LoanEMISchedule.status == "pending", LoanEMISchedule.due_month < today, ) .count() ) # --- Current month EMI reminder (shown on/after 25th) --- current_month_emis = [] if today.day >= 25: for l in active_loans: emi_row = next( ( e for e in l.emi_schedule if e.due_month.year == today.year and e.due_month.month == today.month and (e.status or "pending").lower() == "pending" ), None, ) if emi_row: y, m = emi_row.due_month.year, emi_row.due_month.month last_day = monthrange(y, m)[1] due_date = date(y, m, last_day) current_month_emis.append( { "loan": l, "emi_amount": float(emi_row.amount or 0), "due_date": due_date, } ) summary = { "total_active_amount": total_active_amount, "pending_count": len(pending_loans), "total_monthly_emi": total_monthly_emi, "overdue_emi_count": overdue_emi_count, "current_month_emis": current_month_emis, # 👈 for UI banner / list } return render_template( "loan/my_loans.html", loans=loans, summary=summary, ) # ------------------------------------------------------------------- # Eligibility check helper # ------------------------------------------------------------------- def _check_loan_eligibility(employee: Employee,loan_settings: Optional["LoanSetting"] = None,): msgs = [] min_months = loan_settings.min_months_service if loan_settings else 12 # you can also use loan_settings.min_amount / max_amount later if employee.date_of_joining: delta_days = (date.today() - employee.date_of_joining).days months_completed = delta_days // 30 if months_completed < min_months: msgs.append(f"Minimum {min_months} months of service required.") else: msgs.append("Date of joining not set for your profile.") active_exists = ( Loan.query.filter( Loan.employee_id == employee.id, Loan.status.in_(["requested", "under_review", "approved", "active"]), ) .count() > 0 ) if active_exists: msgs.append("You already have an active or in-approval loan.") return (len(msgs) == 0), msgs # ------------------------------------------------------------------- # Employee: Apply for Loan # ------------------------------------------------------------------- @loan_bp.route("/apply", methods=["GET", "POST"]) @login_required def apply_loan(): emp: Employee = current_user loan_types = ( LoanType.query .filter_by(is_active=True) .order_by(LoanType.display_order, LoanType.label) .all() ) loan_settings = LoanSetting.query.get(1) # singleton row, can be None eligible, reasons = _check_loan_eligibility(emp, loan_settings) if request.method == "POST": form_data = request.form if not eligible: flash("You are not eligible for a loan: " + "; ".join(reasons), "danger") return render_template( "loan/apply_loan.html", employee=emp, eligible=eligible, ineligible_reasons=reasons, form_data=form_data, loan_types=loan_types, loan_settings=loan_settings, ) amount = request.form.get("amount", type=float) tenure = request.form.get("tenure_months", type=int) start_month_str = request.form.get("start_month") loan_type_name = request.form.get("loan_type") # string from dropdown if not amount or not tenure or not loan_type_name: flash("Loan type, amount and tenure are required.", "danger") return render_template( "loan/apply_loan.html", employee=emp, eligible=eligible, ineligible_reasons=reasons, form_data=form_data, loan_types=loan_types, loan_settings=loan_settings, ) # parse / default start month (same as before) if start_month_str: year, month = map(int, start_month_str.split("-")) start_month = date(year, month, 1) else: today = date.today() month = today.month + 1 year = today.year if month > 12: month = 1 year += 1 start_month = date(year, month, 1) loan = Loan( employee_id=emp.id, loan_type=loan_type_name, # still store as string, no schema change requested_amount=amount, requested_tenure_months=tenure, start_month=start_month, status="requested", created_by_id=emp.id, ) db.session.add(loan) db.session.commit() flash("Loan application submitted for approval.", "success") return redirect(url_for("loan.my_loans")) # GET return render_template( "loan/apply_loan.html", employee=emp, eligible=eligible, ineligible_reasons=reasons, form_data={}, loan_types=loan_types, loan_settings=loan_settings, ) # ------------------------------------------------------------------- # Loan detail (employee view) # ------------------------------------------------------------------- @loan_bp.route("/") @login_required def loan_detail(loan_id): loan = Loan.query.get_or_404(loan_id) # Employee can see own loan; admin can see any if loan.employee_id != current_user.id and not getattr(current_user, "is_admin", False): flash("You are not allowed to view this loan.", "danger") return redirect(url_for("loan.my_loans")) loan.display_loan_type = _loan_display_type(loan) emi_schedule = _build_emi_schedule_view(loan) loan_amount, tenure_months, emi_amount, total_paid, outstanding = _loan_summary_numbers(loan) # Allow employee to cancel only while in early workflow can_cancel = loan.status in ("requested", "pending", "under_review") # --- Histories for this loan --- skip_requests = ( LoanSkipRequest.query .filter_by(loan_id=loan.id) .order_by(LoanSkipRequest.request_month.desc()) .all() ) extra_payment_requests = ( LoanExtraPaymentRequest.query .filter_by(loan_id=loan.id) .order_by(LoanExtraPaymentRequest.id.desc()) .all() ) transactions = ( LoanTransaction.query .filter_by(loan_id=loan.id) .order_by(LoanTransaction.txn_date.desc()) .all() ) return render_template( "loan/loan_detail.html", loan=loan, emi_schedule=emi_schedule, can_cancel=can_cancel, admin_view=False, loan_amount=loan_amount, tenure_months=tenure_months, emi_amount=emi_amount, total_paid=total_paid, outstanding=outstanding, skip_requests=skip_requests, # ✔ matches template extra_payment_requests=extra_payment_requests, # ✔ matches template transactions=transactions, # optional – for txn history block ) # ------------------------------------------------------------------- # Skip EMI request (employee) # ------------------------------------------------------------------- @loan_bp.route("//skip-emi", methods=["POST"]) @login_required def request_skip_emi(loan_id): loan = Loan.query.get_or_404(loan_id) emp = current_user if loan.employee_id != emp.id: flash("Not allowed.", "danger") return redirect(url_for("loan.my_loans")) emi_id = request.form.get("emi_id", type=int) reason = request.form.get("reason") emi = LoanEMISchedule.query.get_or_404(emi_id) if emi.loan_id != loan.id: flash("Invalid EMI selected.", "danger") return redirect(url_for("loan.loan_detail", loan_id=loan.id)) # 🔒 Block skip after payroll is frozen month_str = emi.due_month.strftime("%Y-%m") frozen_payroll = Payroll.query.filter_by( employee_id=emp.id, month=month_str, ).first() if frozen_payroll: flash( f"Payroll for {month_str} is already frozen. EMI for this month cannot be skipped.", "warning", ) return redirect(url_for("loan.loan_detail", loan_id=loan.id)) if emi.status != "pending": flash("This EMI is not pending, cannot request skip.", "warning") return redirect(url_for("loan.loan_detail", loan_id=loan.id)) existing_req = LoanSkipRequest.query.filter_by(emi_schedule_id=emi.id).first() if existing_req: flash("Skip request already submitted for this EMI.", "warning") return redirect(url_for("loan.loan_detail", loan_id=loan.id)) req = LoanSkipRequest( loan_id=loan.id, emi_schedule_id=emi.id, request_month=emi.due_month, reason=reason, status="pending", ) db.session.add(req) db.session.commit() flash("Skip EMI request submitted for approval.", "success") return redirect(url_for("loan.loan_detail", loan_id=loan.id)) # ------------------------------------------------------------------- # Extra payment request (employee) # ------------------------------------------------------------------- @loan_bp.route("//extra-payment", methods=["POST"]) @login_required def request_extra_payment(loan_id): loan = Loan.query.get_or_404(loan_id) emp = current_user if loan.employee_id != emp.id: flash("Not allowed.", "danger") return redirect(url_for("loan.my_loans")) amount = request.form.get("amount", type=float) payment_mode = request.form.get("payment_mode") payment_ref = request.form.get("payment_ref") if not amount or amount <= 0: flash("Enter a valid extra payment amount.", "danger") return redirect(url_for("loan.loan_detail", loan_id=loan.id)) req = LoanExtraPaymentRequest( loan_id=loan.id, amount=amount, payment_mode=payment_mode, payment_ref=payment_ref, status="pending", ) db.session.add(req) db.session.commit() flash("Extra payment request submitted for approval.", "success") return redirect(url_for("loan.loan_detail", loan_id=loan.id)) # ------------------------------------------------------------------- # Admin: Loan detail (review view) # ------------------------------------------------------------------- @loan_bp.route("/admin/", methods=["GET", "POST"]) @login_required def admin_detail(loan_id): loan = Loan.query.get_or_404(loan_id) # ---- Role check: allow only Admin / HR roles ---- if not _is_loan_admin(current_user): flash("Only HR/Admin can access loan approvals.", "danger") return redirect(url_for("loan.my_loans")) # ✅ always return a real response if request.method == "POST": action = request.form.get("action") remarks = request.form.get("approver_remarks") or None if action == "approve": # ---- Read editable fields from form ---- sanctioned_amount_form = request.form.get("sanctioned_amount", type=float) tenure_months_form = request.form.get("tenure_months", type=int) emi_amount_form = request.form.get("emi_amount", type=float) start_month_str = request.form.get("start_month") # Amount if sanctioned_amount_form and sanctioned_amount_form > 0: loan.sanctioned_amount = sanctioned_amount_form elif loan.sanctioned_amount is None: loan.sanctioned_amount = loan.requested_amount # Tenure if tenure_months_form and tenure_months_form > 0: loan.tenure_months = tenure_months_form elif loan.tenure_months is None: loan.tenure_months = loan.requested_tenure_months # Start month if start_month_str: try: y, m = map(int, start_month_str.split("-")) loan.start_month = date(y, m, 1) except ValueError: flash("Invalid start month format. Use YYYY-MM.", "danger") return redirect(url_for("loan.admin_detail", loan_id=loan.id)) elif not loan.start_month: today = date.today() m = today.month + 1 y = today.year if m > 12: m = 1 y += 1 loan.start_month = date(y, m, 1) # ---- EMI amount: must be multiple of 500 ---- if emi_amount_form and emi_amount_form > 0: # Validate manual EMI if emi_amount_form % 500 != 0: flash("EMI amount must be a multiple of 500.", "danger") return redirect(url_for("loan.admin_detail", loan_id=loan.id)) loan.emi_amount = emi_amount_form elif loan.sanctioned_amount and loan.tenure_months: # Auto-calc EMI and round **down** to nearest 500 auto_emi = loan.sanctioned_amount / loan.tenure_months rounded_emi = (auto_emi // 500) * 500 # e.g. 3571 -> 3500 if rounded_emi <= 0: flash( "Calculated EMI is too low. Please increase loan amount or tenure.", "danger", ) return redirect(url_for("loan.admin_detail", loan_id=loan.id)) loan.emi_amount = rounded_emi # If we still don't have a valid EMI, block approval if not loan.emi_amount or float(loan.emi_amount or 0) <= 0: flash("Unable to determine a valid EMI amount.", "danger") return redirect(url_for("loan.admin_detail", loan_id=loan.id)) # Final approval fields loan.status = "approved" loan.approved_on = date.today() loan.approved_by_id = current_user.id loan.remarks = remarks # Rebuild EMI schedule and create disbursement txn _create_emi_schedule(loan) _ensure_disbursement_txn(loan, created_by_id=current_user.id) db.session.commit() flash("Loan approved.", "success") return redirect(url_for("loan.admin_dashboard")) elif action == "reject": loan.status = "rejected" loan.approved_on = date.today() loan.approved_by_id = current_user.id loan.remarks = remarks db.session.commit() flash("Loan rejected.", "info") return redirect(url_for("loan.admin_dashboard")) else: # Safety: if some unexpected action value comes in flash("Invalid action.", "danger") return redirect(url_for("loan.admin_detail", loan_id=loan.id)) # ----------------- GET (or fall-through after POST) ----------------- emi_schedule = _build_emi_schedule_view(loan) loan_amount, tenure_months, emi_amount, total_paid, outstanding = _loan_summary_numbers(loan) # --- Histories for this loan --- skip_requests = ( LoanSkipRequest.query .filter_by(loan_id=loan.id) .order_by(LoanSkipRequest.request_month.desc()) .all() ) extra_payment_requests = ( LoanExtraPaymentRequest.query .filter_by(loan_id=loan.id) .order_by(LoanExtraPaymentRequest.id.desc()) .all() ) transactions = ( LoanTransaction.query .filter_by(loan_id=loan.id) .order_by(LoanTransaction.txn_date.desc()) .all() ) # --- Previous loans for this employee (excluding current) --- previous_loans_raw = ( Loan.query .filter(Loan.employee_id == loan.employee_id, Loan.id != loan.id) .order_by(Loan.requested_on.desc()) .all() ) previous_loans = [] for pl in previous_loans_raw: amt, ten, emi_pl, total_paid_pl, outstanding_pl = _loan_summary_numbers(pl) last_paid_emi = _last_paid_emi_date(pl) previous_loans.append( { "loan": pl, "loan_amount": amt, "tenure_months": ten, "paid_amount": total_paid_pl, "outstanding_amount": outstanding_pl, "last_paid_emi_date": last_paid_emi, } ) # Attach display label for current and previous loans loan.display_loan_type = _loan_display_type(loan) for row in previous_loans: row["loan"].display_loan_type = _loan_display_type(row["loan"]) return render_template( "loan/admin_loan_detail.html", loan=loan, emi_schedule=emi_schedule, loan_amount=loan_amount, tenure_months=tenure_months, emi_amount=emi_amount, total_paid=total_paid, outstanding=outstanding, admin_view=True, skip_requests=skip_requests, extra_payment_requests=extra_payment_requests, transactions=transactions, previous_loans=previous_loans, ) # ------------------------------------------------------------------- # Admin: Dashboard # ------------------------------------------------------------------- from datetime import date from calendar import monthrange from sqlalchemy import func, case, and_ @loan_bp.route("/admin") @login_required def admin_dashboard(): if not _is_loan_admin(current_user): flash("Only HR/Admin can access loan dashboard.", "danger") return redirect(url_for("loan.my_loans")) today = date.today() pending_status = (request.args.get("pending_status") or "").strip() # ---- Pending list ---- pending_q = Loan.query if pending_status: pending_q = pending_q.filter(Loan.status == pending_status) else: pending_q = pending_q.filter(Loan.status.in_(["requested", "under_review"])) pending_loans = pending_q.order_by(Loan.requested_on.desc()).all() active_statuses = ("approved", "active") # ✅ ALWAYS define these before using them active_loans = ( Loan.query .options(joinedload(Loan.employee), joinedload(Loan.transactions), joinedload(Loan.emi_schedule)) .filter(Loan.status.in_(active_statuses)) .order_by(Loan.requested_on.desc()) .all() ) closed_loans = ( Loan.query .options(joinedload(Loan.employee)) .filter(Loan.status == "closed") .order_by(Loan.requested_on.desc()) .all() ) # ✅ Auto-close any loans that are fully recovered (status correction) any_closed = False for l in active_loans: if _maybe_close_loan(l, note="Loan auto-closed (fully recovered)."): any_closed = True if any_closed: db.session.commit() # refresh lists after closing active_loans = ( Loan.query .options(joinedload(Loan.employee), joinedload(Loan.transactions), joinedload(Loan.emi_schedule)) .filter(Loan.status.in_(active_statuses)) .order_by(Loan.requested_on.desc()) .all() ) closed_loans = ( Loan.query .options(joinedload(Loan.employee)) .filter(Loan.status == "closed") .order_by(Loan.requested_on.desc()) .all() ) skip_requests = LoanSkipRequest.query.filter_by(status="pending").all() extra_pay_requests = LoanExtraPaymentRequest.query.filter_by(status="pending").all() # ---- Metrics helpers (as you already have) ---- def _display_outstanding(l: Loan) -> float: amount = float(l.sanctioned_amount or l.requested_amount or 0) total_paid = sum(float(t.amount or 0) for t in l.transactions if (t.direction or "").upper() == "CR") bal = l.outstanding_amount() if bal == 0 and l.status in active_statuses: bal = amount - total_paid return bal def _display_emi(l: Loan) -> float: amount = float(l.sanctioned_amount or l.requested_amount or 0) tenure = l.tenure_months or l.requested_tenure_months emi = float(l.emi_amount or 0) if emi == 0 and amount and tenure: emi = amount / tenure return emi total_outstanding = sum((_display_outstanding(l) for l in active_loans), 0.0) month_emi_total = sum((_display_emi(l) for l in active_loans), 0.0) overdue_emi_count = ( LoanEMISchedule.query .filter( LoanEMISchedule.due_month < date(today.year, today.month, 1), # compare to month start LoanEMISchedule.status == "pending", ) .count() ) metrics = { "pending_count": len(pending_loans), "total_outstanding": float(total_outstanding or 0), "month_emi_total": float(month_emi_total or 0), "overdue_emi_count": overdue_emi_count, } # ---- Right pane filters (your existing) ---- emp_q = (request.args.get("emp") or "").strip() dept_id = request.args.get("department_id", type=int) status_filter = (request.args.get("status_filter") or "").strip() recent_q = Loan.query.join(Employee, Loan.employee_id == Employee.id) if emp_q: like = f"%{emp_q}%" recent_q = recent_q.filter(or_(Employee.full_name.ilike(like), Employee.employee_code.ilike(like))) if dept_id: recent_q = recent_q.filter(Employee.department_id == dept_id) if status_filter: recent_q = recent_q.filter(Loan.status == status_filter) recent_loans = recent_q.order_by(Loan.requested_on.desc()).limit(20).all() departments = Department.query.order_by(Department.name.asc()).all() # ✅ label everywhere for l in (pending_loans + active_loans + closed_loans + recent_loans): l.display_loan_type = _loan_display_type(l) # ------------------------------------------------------------------ # ✅ NEW: Summary table rows (Employee-wise loan snapshot) # ------------------------------------------------------------------ # subquery per-loan paid + overdue + next_due_month paid_sq = ( db.session.query( LoanTransaction.loan_id.label("loan_id"), func.coalesce(func.sum(case((LoanTransaction.direction == "CR", LoanTransaction.amount), else_=0)), 0).label("total_paid"), ) .group_by(LoanTransaction.loan_id) .subquery() ) next_due_sq = ( db.session.query( LoanEMISchedule.loan_id.label("loan_id"), func.min( case( (LoanEMISchedule.status == "pending", LoanEMISchedule.due_month), else_=None ) ).label("next_due_month"), func.coalesce( func.sum( case( (and_(LoanEMISchedule.status == "pending", LoanEMISchedule.due_month < date(today.year, today.month, 1)), 1), else_=0 ) ), 0 ).label("overdue_count"), ) .group_by(LoanEMISchedule.loan_id) .subquery() ) summary_q = ( db.session.query( Loan.id.label("loan_id"), Loan.status.label("status"), Loan.loan_type.label("loan_type"), Loan.sanctioned_amount.label("sanctioned_amount"), Loan.requested_amount.label("requested_amount"), Loan.tenure_months.label("tenure_months"), Loan.requested_tenure_months.label("requested_tenure_months"), Loan.emi_amount.label("emi_amount"), Employee.full_name.label("employee_name"), Employee.employee_code.label("employee_code"), Department.name.label("department_name"), func.coalesce(paid_sq.c.total_paid, 0).label("total_paid"), func.coalesce(next_due_sq.c.next_due_month, None).label("next_due_month"), func.coalesce(next_due_sq.c.overdue_count, 0).label("overdue_count"), ) .join(Employee, Employee.id == Loan.employee_id) .outerjoin(Department, Department.id == Employee.department_id) .outerjoin(paid_sq, paid_sq.c.loan_id == Loan.id) .outerjoin(next_due_sq, next_due_sq.c.loan_id == Loan.id) .filter(Loan.status.in_(["approved", "active"])) # summary focuses on running loans .order_by(Employee.full_name.asc()) ) # reuse your filters if emp_q: like = f"%{emp_q}%" summary_q = summary_q.filter(or_(Employee.full_name.ilike(like), Employee.employee_code.ilike(like))) if dept_id: summary_q = summary_q.filter(Employee.department_id == dept_id) summary_rows_raw = summary_q.all() # compute outstanding with same fallback logic summary_rows = [] for r in summary_rows_raw: amount = float(r.sanctioned_amount or r.requested_amount or 0) tenure = r.tenure_months or r.requested_tenure_months emi = float(r.emi_amount or 0) or (amount / tenure if amount and tenure else 0) total_paid = float(r.total_paid or 0) outstanding = max(amount - total_paid, 0) # pretty next due date = last day of that due_month next_due_date = None if r.next_due_month: y, m = r.next_due_month.year, r.next_due_month.month last_day = monthrange(y, m)[1] next_due_date = date(y, m, last_day) summary_rows.append({ "loan_id": r.loan_id, "employee_name": r.employee_name, "employee_code": r.employee_code, "department_name": r.department_name, "loan_type": _loan_display_type(type("X",(object,),{"loan_type":r.loan_type})()), # safe label lookup "status": r.status, "amount": amount, "emi": emi, "total_paid": total_paid, "outstanding": outstanding, "next_due_date": next_due_date, "overdue_count": int(r.overdue_count or 0), }) return render_template( "loan/admin_dashboard.html", metrics=metrics, pending_loans=pending_loans, active_loans=active_loans, closed_loans=closed_loans, skip_requests=skip_requests, extra_pay_requests=extra_pay_requests, recent_loans=recent_loans, departments=departments, summary_rows=summary_rows, # ✅ NEW ) # ------------------------------------------------------------------- # Admin: Manage Loan Types # ------------------------------------------------------------------- @loan_bp.route("/admin/types", methods=["GET", "POST"]) @login_required def manage_loan_types(): # TODO: restrict to HR / Admin roles if request.method == "POST": lt_id = request.form.get("id", type=int) name = (request.form.get("name") or "").strip() label = (request.form.get("label") or "").strip() description = (request.form.get("description") or "").strip() max_amount = request.form.get("max_amount", type=float) max_tenure = request.form.get("max_tenure_months", type=int) is_active = bool(request.form.get("is_active")) if not name or not label: flash("Name and label are required.", "danger") else: if lt_id: # edit existing lt = LoanType.query.get_or_404(lt_id) lt.name = name lt.label = label lt.description = description or None lt.max_amount = max_amount if max_amount else None lt.max_tenure_months = max_tenure if max_tenure else None lt.is_active = is_active else: # new lt = LoanType( name=name, label=label, description=description or None, max_amount=max_amount if max_amount else None, max_tenure_months=max_tenure if max_tenure else None, is_active=is_active, ) db.session.add(lt) db.session.commit() flash("Loan type saved.", "success") return redirect(url_for("loan.manage_loan_types")) types = LoanType.query.order_by(LoanType.display_order, LoanType.label).all() return render_template("loan/loan_types.html", types=types) @loan_bp.route("/admin/types//toggle", methods=["POST"]) @login_required def toggle_loan_type(type_id): lt = LoanType.query.get_or_404(type_id) lt.is_active = not lt.is_active db.session.commit() flash(f"Loan type '{lt.label}' is now {'active' if lt.is_active else 'inactive'}.", "success") return redirect(url_for("loan.manage_loan_types")) @loan_bp.route("/admin/settings", methods=["GET", "POST"]) @login_required def loan_settings_admin(): # TODO: restrict to HR / Admin settings = LoanSetting.query.get(1) if not settings: settings = LoanSetting(id=1) db.session.add(settings) db.session.flush() # no commit yet if request.method == "POST": settings.policy_html = request.form.get("policy_html") or "" settings.min_months_service = request.form.get( "min_months_service", type=int ) or settings.min_months_service db.session.commit() flash("Loan settings updated.", "success") return redirect(url_for("loan.loan_settings_admin")) return render_template("loan/loan_settings.html", settings=settings) @loan_bp.route("//cancel", methods=["POST"]) @login_required def cancel(loan_id): loan = Loan.query.get_or_404(loan_id) emp: Employee = current_user # Only the owner can cancel if loan.employee_id != emp.id: flash("You are not allowed to cancel this loan.", "danger") return redirect(url_for("loan.my_loans")) # Allow cancel only while still in workflow if loan.status not in ("requested", "pending", "under_review"): flash("This loan can no longer be cancelled.", "warning") return redirect(url_for("loan.loan_detail", loan_id=loan.id)) # Mark as cancelled loan.status = "cancelled" loan.closed_on = date.today() # optional: append remark if loan.remarks: loan.remarks += "\nCancelled by employee." else: loan.remarks = "Cancelled by employee." db.session.commit() flash("Loan request cancelled.", "success") return redirect(url_for("loan.my_loans")) from datetime import datetime def _is_loan_admin(user) -> bool: raw_role = getattr(user, "role", None) if isinstance(raw_role, str): role_name = raw_role.lower() else: role_name = (getattr(raw_role, "name", "") or "").lower() allowed_roles = {"admin", "system administrator", "hr", "hr admin", "hr manager"} return role_name in allowed_roles or getattr(user, "is_admin", False) @loan_bp.route("/admin/skip-request/", methods=["POST"]) @login_required def handle_skip_request(req_id): if not _is_loan_admin(current_user): flash("Only HR/Admin can act on skip requests.", "danger") return redirect(url_for("loan.admin_dashboard")) req = LoanSkipRequest.query.get_or_404(req_id) loan = req.loan # ✅ Load EMI row using FK emi = LoanEMISchedule.query.get(req.emi_schedule_id) if req.emi_schedule_id else None if req.status != "pending": flash("This skip request is already processed.", "warning") return redirect(url_for("loan.admin_dashboard")) decision = request.form.get("decision") remarks = request.form.get("remarks") or None if decision == "approve": req.status = "approved" # these fields assume your model has them; if not, just drop them req.approved_by_id = current_user.id req.approved_on = date.today() req.remarks = remarks # Mark this EMI as skipped if emi and emi.status == "pending": emi.status = "skipped" # Push same EMI amount to the end of schedule all_emis = list(loan.emi_schedule) if all_emis: last_due = max(e.due_month for e in all_emis) else: last_due = loan.start_month or date.today() next_month_index = last_due.month + 1 next_year = last_due.year if next_month_index > 12: next_month_index = 1 next_year += 1 new_due = date(next_year, next_month_index, 1) extra_emi = LoanEMISchedule( loan_id=loan.id, due_month=new_due, amount=emi.amount if emi else req.loan.emi_amount, status="pending", ) db.session.add(extra_emi) db.session.commit() flash("EMI skip approved and schedule updated.", "success") elif decision == "reject": req.status = "rejected" req.approved_by_id = current_user.id req.approved_on = date.today() req.remarks = remarks db.session.commit() flash("Skip request rejected.", "info") else: flash("Invalid decision.", "danger") return redirect(url_for("loan.admin_dashboard")) from datetime import date from datetime import date def _maybe_close_loan(loan: Loan, note: str = None) -> bool: """ Close the loan if fully recovered. Criteria (as per your requirement): - If due/outstanding is cleared (<= 0), mark loan CLOSED. Also: when closing, mark any remaining pending EMIs as paid with amount=0 so the schedule doesn't keep showing pending dues. """ if (loan.status or "").lower() in ("closed", "cancelled", "rejected"): return False # Prefer true ledger outstanding (DR - CR) outstanding = float(loan.outstanding_amount() or 0) # Fallback: if ledger is not built properly yet, compute from sanctioned/requested - total_paid if outstanding == 0 and (loan.status or "").lower() in ("approved", "active"): amount = float(loan.sanctioned_amount or loan.requested_amount or 0) total_paid = sum( float(t.amount or 0) for t in loan.transactions if (t.direction or "").upper() == "CR" ) outstanding = amount - total_paid if outstanding <= 0.0001: loan.status = "closed" loan.closed_on = date.today() # ✅ IMPORTANT: clear EMI schedule so it doesn't remain pending forever for emi in loan.emi_schedule: if (emi.status or "").lower() == "pending": emi.status = "paid" emi.paid_on = date.today() # reduce remaining due to zero emi.amount = 0 emi.paid_amount = (emi.paid_amount or 0) if note: loan.remarks = ((loan.remarks or "").strip() + ("\n" if loan.remarks else "") + note).strip() return True return False @loan_bp.route("/admin/extra-payment/", methods=["POST"]) @login_required def handle_extra_payment_request(req_id): if not _is_loan_admin(current_user): flash("Only HR/Admin can act on extra payment requests.", "danger") return redirect(url_for("loan.admin_dashboard")) req = LoanExtraPaymentRequest.query.get_or_404(req_id) loan = req.loan if req.status != "pending": flash("This extra payment request is already processed.", "warning") return redirect(url_for("loan.admin_dashboard")) decision = request.form.get("decision") admin_remarks = request.form.get("remarks") or None # from form textarea if decision == "approve": txn_date = req.txn_date or date.today() pay_amount = float(req.amount or 0) if pay_amount <= 0: flash("Invalid extra payment amount.", "danger") return redirect(url_for("loan.admin_dashboard")) # 1) Create CR transaction (repayment) txn = LoanTransaction( loan_id=loan.id, txn_date=txn_date, txn_type="extra_payment", direction="CR", amount=req.amount, reference=f"Extra payment ({req.payment_mode or 'manual'} {req.payment_ref or ''})".strip(), source="manual", created_by_id=current_user.id, remarks=admin_remarks, ) db.session.add(txn) # 2) Apply amount to EMI schedule from END (latest pending EMI first) remaining = pay_amount pending_emis = ( LoanEMISchedule.query .filter_by(loan_id=loan.id, status="pending") .order_by(LoanEMISchedule.due_month.desc()) # ✅ end-first .all() ) for emi in pending_emis: if remaining <= 0: break emi_amt = float(emi.amount or 0) if emi_amt <= 0: continue if remaining >= emi_amt: # fully cover EMI emi.status = "paid" emi.paid_on = txn_date emi.paid_amount = emi.amount # store full EMI paid emi.remarks = (f"Paid via extra payment request #{req.id}." + (f" {admin_remarks}" if admin_remarks else "")).strip() remaining -= emi_amt else: # partial cover => reduce EMI amount emi.status = "pending" emi.paid_on = txn_date # optional: indicates partial payment date emi.paid_amount = (emi.paid_amount or 0) + remaining emi.amount = round(emi_amt - remaining, 2) emi.remarks = (f"Adjusted by extra payment request #{req.id}: prepaid {remaining}." + (f" {admin_remarks}" if admin_remarks else "")).strip() remaining = 0 break # 3) Mark request approved req.status = "approved" req.approved_by_id = current_user.id req.approved_on = txn_date req.admin_remarks = admin_remarks # 4) Auto-close loan if fully recovered closed = _maybe_close_loan( loan, note=f"Loan auto-closed on {date.today().strftime('%d-%b-%Y')} (fully recovered)." ) db.session.commit() if closed: flash("Extra payment approved, EMIs adjusted, and loan CLOSED.", "success") else: flash("Extra payment approved and applied to EMIs.", "success") return redirect(url_for("loan.admin_dashboard")) elif decision == "reject": req.status = "rejected" req.approved_by_id = current_user.id req.approved_on = date.today() req.admin_remarks = admin_remarks db.session.commit() flash("Extra payment request rejected.", "info") return redirect(url_for("loan.admin_dashboard")) flash("Invalid decision.", "danger") return redirect(url_for("loan.admin_dashboard")) @loan_bp.route("/admin/loans") @login_required def admin_loans_list(): # Role guard if not _is_loan_admin(current_user): flash("Only HR/Admin can view all loans.", "danger") return redirect(url_for("loan.my_loans")) # ---- Filters ---- q_text = (request.args.get("q") or "").strip() status_filter = (request.args.get("status") or "").strip() dept_id = request.args.get("department_id", type=int) loan_type_filter = (request.args.get("loan_type") or "").strip() base_q = ( Loan.query .join(Employee, Loan.employee_id == Employee.id) .options( joinedload(Loan.employee).joinedload(Employee.department) ) ) if q_text: like = f"%{q_text}%" # allow search by employee name, code or loan id base_q = base_q.filter( or_( Employee.full_name.ilike(like), Employee.employee_code.ilike(like), func.cast(Loan.id, db.String).ilike(like), ) ) if status_filter: base_q = base_q.filter(Loan.status == status_filter) if dept_id: base_q = base_q.filter(Employee.department_id == dept_id) if loan_type_filter: base_q = base_q.filter(Loan.loan_type == loan_type_filter) loans = ( base_q .order_by(Loan.requested_on.desc()) .all() ) # Precompute summary numbers for each loan (for table) rows = [] for loan in loans: loan.display_loan_type = _loan_display_type(loan) loan_amount, tenure_months, emi_amount, total_paid, outstanding = _loan_summary_numbers(loan) rows.append( { "loan": loan, "loan_amount": loan_amount, "tenure_months": tenure_months, "emi_amount": emi_amount, "total_paid": total_paid, "outstanding": outstanding, } ) departments = Department.query.order_by(Department.name.asc()).all() loan_types = ( LoanType.query .order_by(LoanType.label) .all() ) return render_template( "loan/admin_loans_list.html", rows=rows, departments=departments, loan_types=loan_types, ) def _parse_month_yyyy_mm(v) -> date: if v is None or (isinstance(v, float) and math.isnan(v)): raise ValueError("empty month") # pandas Timestamp / datetime / date if isinstance(v, (datetime, date)): return date(v.year, v.month, 1) s = str(v).strip() # if excel gives full date-like text: "2025-01-01 00:00:00" try: dt = datetime.fromisoformat(s.replace("Z", "")) return date(dt.year, dt.month, 1) except Exception: pass # strict "YYYY-MM" dt = datetime.strptime(s, "%Y-%m") return date(dt.year, dt.month, 1) def _month_add(d: date, n: int) -> date: # add n months to 1st-of-month date y = d.year + (d.month - 1 + n) // 12 m = (d.month - 1 + n) % 12 + 1 return date(y, m, 1) def _coerce_float(v, default=None): if v is None or (isinstance(v, float) and math.isnan(v)): return default s = str(v).strip() if s == "": return default return float(s) def _coerce_int(v, default=None): if v is None or (isinstance(v, float) and math.isnan(v)): return default s = str(v).strip() if s == "": return default return int(float(s)) def _normalize_status(s: str) -> str: s = (s or "").strip().lower() if s in ("active", "running", "open"): return "active" if s in ("closed", "complete", "completed", "settled"): return "closed" return s # let validator catch def _safe_str(v) -> str: if v is None or (isinstance(v, float) and math.isnan(v)): return "" return str(v).strip() def _validate_historical_loans_excel(path: str): """ Returns: preview_rows: list[dict] - normalized, safe values errors: list[dict] with {row, field, error} """ errors = [] preview = [] try: df = pd.read_excel(path, sheet_name="historical_loans") except Exception: # fallback: first sheet df = pd.read_excel(path) # normalize columns df.columns = [str(c).strip() for c in df.columns] required = [ "employee_code", "loan_type", "sanctioned_amount", "tenure_months", "emi_amount", "start_month", "total_paid", "status" ] for col in required: if col not in df.columns: errors.append({"row": "-", "field": col, "error": "Missing required column"}) if errors: return [], errors for idx, r in df.iterrows(): row_no = int(idx) + 2 # excel row number (header=1) employee_code = _safe_str(r.get("employee_code")) loan_type = _safe_str(r.get("loan_type")) sanctioned = _coerce_float(r.get("sanctioned_amount"), default=None) tenure = _coerce_int(r.get("tenure_months"), default=None) emi_amount = _coerce_float(r.get("emi_amount"), default=None) start_month_s = _safe_str(r.get("start_month")) total_paid = _coerce_float(r.get("total_paid"), default=0.0) status = _normalize_status(_safe_str(r.get("status"))) last_payment_date_s = _safe_str(r.get("last_payment_date")) remarks = _safe_str(r.get("remarks")) # ---- validations ---- if not employee_code: errors.append({"row": row_no, "field": "employee_code", "error": "Required"}) else: emp = Employee.query.filter(Employee.employee_code == employee_code).first() if not emp: errors.append({"row": row_no, "field": "employee_code", "error": f"Employee not found: {employee_code}"}) if not loan_type: errors.append({"row": row_no, "field": "loan_type", "error": "Required"}) if sanctioned is None or sanctioned <= 0: errors.append({"row": row_no, "field": "sanctioned_amount", "error": "Must be > 0"}) if tenure is None or tenure <= 0: errors.append({"row": row_no, "field": "tenure_months", "error": "Must be > 0"}) if emi_amount is None or emi_amount <= 0: errors.append({"row": row_no, "field": "emi_amount", "error": "Must be > 0"}) # start_month must be YYYY-MM try: start_month = _parse_month_yyyy_mm(start_month_s) except Exception: errors.append({"row": row_no, "field": "start_month", "error": "Invalid format. Use YYYY-MM"}) start_month = None if total_paid is None or total_paid < 0: errors.append({"row": row_no, "field": "total_paid", "error": "Must be >= 0"}) if status not in ("active", "closed"): errors.append({"row": row_no, "field": "status", "error": "Must be 'active' or 'closed'"}) # optional date last_payment_date = None if last_payment_date_s: try: # try ISO first last_payment_date = datetime.fromisoformat(last_payment_date_s.replace("Z","")).date() except Exception: try: last_payment_date = datetime.strptime(last_payment_date_s, "%Y-%m-%d").date() except Exception: errors.append({"row": row_no, "field": "last_payment_date", "error": "Use YYYY-MM-DD"}) # logical constraints if sanctioned is not None and total_paid is not None and total_paid > sanctioned + 0.01: errors.append({"row": row_no, "field": "total_paid", "error": "Cannot exceed sanctioned_amount"}) # if closed, total_paid should be >= sanctioned (we allow small rounding) if status == "closed" and sanctioned and total_paid < sanctioned - 0.01: errors.append({"row": row_no, "field": "status", "error": "Closed loan must have total_paid >= sanctioned_amount"}) # build preview row preview.append({ "row_no": row_no, "employee_code": employee_code, "loan_type": loan_type, "sanctioned_amount": float(sanctioned or 0.0), "tenure_months": int(tenure or 0), "emi_amount": float(emi_amount or 0.0), "start_month": start_month_s, "total_paid": float(total_paid or 0.0), "status": status, "last_payment_date": last_payment_date_s, "remarks": remarks, }) # remove fully empty rows (optional) preview = [p for p in preview if any(str(p.get(k, "")).strip() for k in ("employee_code","loan_type","sanctioned_amount"))] return preview, errors def _import_historical_loans(preview_rows: list[dict]): created = {"loans": 0, "transactions": 0, "emis": 0, "skipped": 0} for r in preview_rows: employee_code = r["employee_code"] emp = Employee.query.filter(Employee.employee_code == employee_code).first() if not emp: created["skipped"] += 1 continue loan_type = r["loan_type"] sanctioned_amount = float(r["sanctioned_amount"]) tenure_months = int(r["tenure_months"]) emi_amount = float(r["emi_amount"]) start_month = _parse_month_yyyy_mm(r["start_month"]) total_paid = float(r["total_paid"]) status = r["status"] last_payment_date = None if r.get("last_payment_date"): try: last_payment_date = datetime.strptime(r["last_payment_date"], "%Y-%m-%d").date() except Exception: last_payment_date = None remarks = (r.get("remarks") or "").strip() # ---- Create Loan ---- loan = Loan( employee_id=emp.id, loan_type=loan_type, # adjust if your model uses loan_type_id requested_amount=sanctioned_amount, requested_tenure_months=tenure_months, sanctioned_amount=sanctioned_amount, tenure_months=tenure_months, emi_amount=emi_amount, repayment_mode="salary", status="active" if status == "active" else "closed", start_month=start_month, # adjust if your Loan has different field name remarks=remarks or "Imported historical", ) db.session.add(loan) db.session.flush() # get loan.id created["loans"] += 1 # ---- DR transaction (disbursal) ---- disbursal_date = start_month db.session.add(LoanTransaction( loan_id=loan.id, txn_date=disbursal_date, txn_type="disbursement", direction="DR", amount=sanctioned_amount, reference=start_month.strftime("%Y-%m"), source="historical_import", created_by_id=getattr(current_user, "id", None), remarks=f"Historical loan disbursal import ({loan_type})", )) created["transactions"] += 1 # ---- CR transaction (total recovery till date) ---- if total_paid > 0: cr_date = last_payment_date or date.today() db.session.add(LoanTransaction( loan_id=loan.id, txn_date=cr_date, txn_type="recovery_import", direction="CR", amount=total_paid, reference=(cr_date.strftime("%Y-%m-%d")), source="historical_import", created_by_id=getattr(current_user, "id", None), remarks="Historical recovery import (total_paid)", )) created["transactions"] += 1 # ---- EMI schedule ---- # Create schedule months: start_month .. start_month+tenure-1 # Mark as paid for months covered by total_paid (full EMI blocks) remaining_paid = total_paid for i in range(tenure_months): due_m = _month_add(start_month, i) sched = LoanEMISchedule( loan_id=loan.id, due_month=due_m, amount=emi_amount, status="pending", ) # consume paid into schedule status (for nicer UI; payroll uses CR sum anyway) if remaining_paid >= emi_amount - 0.01: sched.status = "paid" sched.paid_on = last_payment_date or date.today() sched.paid_amount = emi_amount remaining_paid = max(0.0, remaining_paid - emi_amount) else: # keep pending even if partial paid; your _emi_payable_amount() will clamp last EMI correctly sched.status = "pending" db.session.add(sched) created["emis"] += 1 # If loan is closed but rounding oddities exist, keep status closed anyway (already validated) return created @loan_bp.route("/loans/historical-template.xlsx") @login_required def download_historical_loans_template(): import pandas as pd import io # columns as agreed cols = [ "employee_code", "loan_type", "sanctioned_amount", "tenure_months", "emi_amount", "start_month", # YYYY-MM "total_paid", "status", # active / closed "last_payment_date", # optional YYYY-MM-DD "remarks", ] # provide 2 sample rows sample = [ { "employee_code": "ABC-1001", "loan_type": "Salary Advance", "sanctioned_amount": 50000, "tenure_months": 10, "emi_amount": 5000, "start_month": "2025-01", "total_paid": 20000, "status": "active", "last_payment_date": "2025-04-30", "remarks": "Imported historical", }, { "employee_code": "ABC-1002", "loan_type": "Personal Loan", "sanctioned_amount": 30000, "tenure_months": 6, "emi_amount": 5000, "start_month": "2024-10", "total_paid": 30000, "status": "closed", "last_payment_date": "2025-03-31", "remarks": "", }, ] df = pd.DataFrame(sample, columns=cols) out = io.BytesIO() with pd.ExcelWriter(out, engine="openpyxl") as writer: df.to_excel(writer, sheet_name="historical_loans", index=False) out.seek(0) return send_file( out, as_attachment=True, download_name="historical_loans_template.xlsx", mimetype="application/vnd.openxmlformats-officedocument.spreadsheetml.sheet", ) @loan_bp.route("/loans/historical-import", methods=["GET", "POST"]) @login_required def historical_loans_import(): """ Upload excel -> validate -> preview -> confirm import. """ if request.method == "GET": return render_template("loan/historical_import.html") f = request.files.get("file") if not f or not f.filename: flash("Please upload an Excel file.", "danger") return redirect(url_for("loan.historical_loans_import")) filename = secure_filename(f.filename) if not filename.lower().endswith((".xlsx", ".xls")): flash("Only Excel files (.xlsx/.xls) are supported.", "danger") return redirect(url_for("loan.historical_loans_import")) tmp_dir = tempfile.gettempdir() token = str(uuid.uuid4()) tmp_path = os.path.join(tmp_dir, f"historical_loans_{token}.xlsx") f.save(tmp_path) # parse + validate preview_rows, errors = _validate_historical_loans_excel(tmp_path) # If everything is empty -> treat as error if not preview_rows and not errors: errors.append({"row": "-", "field": "-", "error": "No rows found in sheet 'historical_loans'."}) return render_template( "loan/historical_import_preview.html", token=token, preview_rows=preview_rows, errors=errors, ok_to_import=(len(errors) == 0 and len(preview_rows) > 0), ) @loan_bp.route("/loans/historical-import/confirm", methods=["POST"]) @login_required def historical_loans_import_confirm(): token = (request.form.get("token") or "").strip() if not token: flash("Import token missing. Please upload again.", "danger") return redirect(url_for("loan.historical_loans_import")) tmp_path = os.path.join(tempfile.gettempdir(), f"historical_loans_{token}.xlsx") if not os.path.exists(tmp_path): flash("Import file expired/missing. Please upload again.", "danger") return redirect(url_for("loan.historical_loans_import")) # validate again (safety) preview_rows, errors = _validate_historical_loans_excel(tmp_path) if errors: flash("Cannot import: validation errors exist.", "danger") return render_template( "loan/historical_import_preview.html", token=token, preview_rows=preview_rows, errors=errors, ok_to_import=False, ) # import in a DB transaction created = {"loans": 0, "transactions": 0, "emis": 0, "skipped": 0} try: created = _import_historical_loans(preview_rows) db.session.commit() except Exception as e: db.session.rollback() current_app.logger.exception("Historical loan import failed: %s", e) flash(f"Import failed: {e}", "danger") return redirect(url_for("loan.historical_loans_import")) finally: # cleanup temp file try: os.remove(tmp_path) except Exception: pass flash( f"✅ Imported successfully. Loans={created['loans']}, EMIs={created['emis']}, " f"Transactions={created['transactions']}, Skipped={created['skipped']}", "success", ) return redirect(url_for("loan.historical_loans_import")) # ------------------------------------------------------------------- # Admin/HR: Initiate loan request for any employee # ------------------------------------------------------------------- @loan_bp.route("/admin/initiate", methods=["GET", "POST"]) @login_required def admin_initiate_loan(): if not _is_loan_admin(current_user): flash("Only HR/Admin can initiate loans for employees.", "danger") return redirect(url_for("loan.my_loans")) # dropdowns employees = ( Employee.query .options(joinedload(Employee.department)) .order_by(Employee.full_name.asc()) .all() ) loan_types = ( LoanType.query .filter_by(is_active=True) .order_by(LoanType.display_order, LoanType.label) .all() ) loan_settings = LoanSetting.query.get(1) # can be None # defaults for UI today = date.today() default_next_month_m = today.month + 1 default_next_month_y = today.year if default_next_month_m > 12: default_next_month_m = 1 default_next_month_y += 1 default_start_month = f"{default_next_month_y:04d}-{default_next_month_m:02d}" if request.method == "POST": employee_id = request.form.get("employee_id", type=int) amount = request.form.get("amount", type=float) tenure = request.form.get("tenure_months", type=int) start_month_str = (request.form.get("start_month") or "").strip() loan_type_name = (request.form.get("loan_type") or "").strip() remarks = (request.form.get("remarks") or "").strip() ignore_eligibility = bool(request.form.get("ignore_eligibility")) if not employee_id: flash("Please select an employee.", "danger") return render_template( "loan/admin_initiate_loan.html", employees=employees, loan_types=loan_types, loan_settings=loan_settings, default_start_month=default_start_month, form_data=request.form, ) emp = Employee.query.get(employee_id) if not emp: flash("Invalid employee selected.", "danger") return redirect(url_for("loan.admin_initiate_loan")) if not amount or not tenure or not loan_type_name: flash("Loan type, amount and tenure are required.", "danger") return render_template( "loan/admin_initiate_loan.html", employees=employees, loan_types=loan_types, loan_settings=loan_settings, default_start_month=default_start_month, form_data=request.form, ) # Eligibility check (optional bypass) eligible, reasons = _check_loan_eligibility(emp, loan_settings) if (not eligible) and (not ignore_eligibility): flash("Employee is not eligible: " + "; ".join(reasons), "danger") return render_template( "loan/admin_initiate_loan.html", employees=employees, loan_types=loan_types, loan_settings=loan_settings, default_start_month=default_start_month, form_data=request.form, ineligible_reasons=reasons, ) # parse start month (YYYY-MM) OR default next month if start_month_str: try: y, m = map(int, start_month_str.split("-")) start_month = date(y, m, 1) except Exception: flash("Invalid start month. Use YYYY-MM.", "danger") return render_template( "loan/admin_initiate_loan.html", employees=employees, loan_types=loan_types, loan_settings=loan_settings, default_start_month=default_start_month, form_data=request.form, ) else: y, m = map(int, default_start_month.split("-")) start_month = date(y, m, 1) loan = Loan( employee_id=emp.id, loan_type=loan_type_name, # stored as string (name/label) like your current flow requested_amount=amount, requested_tenure_months=tenure, start_month=start_month, status="requested", # stays in admin dashboard pending list created_by_id=current_user.id, remarks=(remarks or "Initiated by HR/Admin."), ) db.session.add(loan) db.session.commit() flash(f"Loan request created for {emp.full_name} ({emp.employee_code}).", "success") return redirect(url_for("loan.admin_detail", loan_id=loan.id)) # GET return render_template( "loan/admin_initiate_loan.html", employees=employees, loan_types=loan_types, loan_settings=loan_settings, default_start_month=default_start_month, form_data={}, ineligible_reasons=[], )