from flask import Blueprint, request, render_template, redirect, url_for, flash, current_app from flask_login import login_required, current_user from datetime import datetime, timedelta, date from sqlalchemy.exc import IntegrityError from extensions import db,mail from models import Employee, LeaveType, LeaveBalance, LeaveRequest, LeaveApprovalLog, RHConfig, LeavePolicy, Role, Department, Holiday, LeaveSettings, CompOffRequest, LeaveAllocationBatch, LeaveAllocationEntry,LeaveInformee, Attendance from utils.leave_utils import calculate_sandwich_days from utils.notify import notify_employee, notify_many from typing import Optional from sqlalchemy import and_, or_ # --- Notification helpers --- from urllib.parse import urljoin def _days_overlap_in_month(lr, month_start, month_end): s = max(lr.start_date, month_start) e = min(lr.end_date, month_end) if e < s: return 0.0 if bool(getattr(lr, "half_day", False)) or float(getattr(lr, "total_days", 0) or 0) == 0.5: if lr.start_date == lr.end_date and month_start <= lr.start_date <= month_end: return 0.5 return float((e - s).days + 1) def _get_applicable_leave_policy(employee, leave_type_id: int): """ Pick the most specific policy for the employee + leave type. Priority: 1) role + department 2) department only 3) role only 4) global policy """ role_id = getattr(employee, "role_id", None) department_id = getattr(employee, "department_id", None) base = LeavePolicy.query.filter_by(leave_type_id=leave_type_id) candidates = [ and_(LeavePolicy.role_id == role_id, LeavePolicy.department_id == department_id), and_(LeavePolicy.role_id.is_(None), LeavePolicy.department_id == department_id), and_(LeavePolicy.role_id == role_id, LeavePolicy.department_id.is_(None)), and_(LeavePolicy.role_id.is_(None), LeavePolicy.department_id.is_(None)), ] for cond in candidates: policy = base.filter(cond).order_by(LeavePolicy.id.desc()).first() if policy: return policy return None def _link(endpoint: str, **kwargs) -> str: """ Build public, absolute links for notifications. Prefer EXTERNAL_BASE_URL if configured; else fall back to _external=True (which relies on ProxyFix + forwarded headers). """ base = current_app.config.get("EXTERNAL_BASE_URL") if base: path = url_for(endpoint, _external=False, **kwargs) return urljoin(base, path) return url_for(endpoint, _external=True, **kwargs) def _format_range(d1, d2): if d1 == d2: return d1.strftime('%d-%b-%Y') # use ASCII "to" instead of the Unicode arrow return f"{d1.strftime('%d-%b-%Y')} to {d2.strftime('%d-%b-%Y')}" def _hr_ids(): """ Try to find HR recipients. - First by role.name == 'HR' - Fallback by department.name contains 'HR' - If none, return [] (no HR ping) """ ids = [] try: # Role-based ids = [e.id for e in Employee.query.join(Role).filter(Role.name.ilike('hr')).all()] if ids: return ids except Exception: pass try: # Department-based fallback ids = [e.id for e in Employee.query.join(Department).filter(Department.name.ilike('%hr%')).all()] return ids except Exception: return [] leave_bp = Blueprint('leave', __name__, url_prefix='/leave') from sqlalchemy import and_ def get_leave_year_range(): settings = LeaveSettings.query.first() if settings and settings.leave_year_start and settings.leave_year_end: return settings.leave_year_start, settings.leave_year_end # fallback: financial year Apr-Mar if not configured today = datetime.today().date() fy_start = date(today.year if today.month >= 4 else today.year - 1, 4, 1) fy_end = date(fy_start.year + 1, 3, 31) return fy_start, fy_end def get_optional_holiday_dates(start_dt, end_dt, employee=None): # fetch all holidays in window once hols = (Holiday.query .filter(Holiday.date >= start_dt, Holiday.date <= end_dt) .all()) by_date = {} for h in hols: by_date.setdefault(h.date, []).append(h) whitelist = set() emp_loc = getattr(employee, 'location', None) for dt, items in by_date.items(): # Prefer specific location if it exists specific = next((x for x in items if emp_loc and x.location == emp_loc), None) if specific: # specific decides RH eligibility if (specific.type or '').lower() == 'optional': whitelist.add(dt) continue # if specific is mandatory, NOT eligible; skip # No specific entry -> fall back to ALL all_row = next((x for x in items if x.location == 'ALL'), None) if all_row and (all_row.type or '').lower() == 'optional': whitelist.add(dt) return whitelist def get_applicable_holiday_for(employee, dt): q = Holiday.query.filter_by(date=dt) emp_loc = getattr(employee, 'location', None) if emp_loc: h = q.filter_by(location=emp_loc).first() if h: return h return q.filter_by(location='ALL').first() from typing import Optional def _normalize_holiday_type(raw: str) -> Optional[str]: """ Map various inputs to canonical {'mandatory','optional'}. Treat 'National Holiday' as mandatory (non-RH). """ if not raw: return None t = raw.strip().lower() mandatory_aliases = { 'mandatory', 'national', 'national holiday', 'gazetted', 'public', 'public holiday', 'closed' } optional_aliases = {'optional', 'restricted', 'restricted holiday', 'rh'} if t in mandatory_aliases: return 'mandatory' if t in optional_aliases: return 'optional' return None # ----------------------------- # Allocation helpers (freeze + delta-based) # ----------------------------- from datetime import datetime, date from sqlalchemy import func from typing import Optional from models import LeaveAllocationBatch, LeaveAllocationEntry def _make_batch_key(fy_label:int, leave_type_code:str, strategy:str) -> str: ts = datetime.now().strftime('%Y%m%d_%H%M%S') return f"FY{fy_label}_{leave_type_code}_{strategy}_{ts}" def _latest_non_revoked_batch(leave_type_id:int, fy_label:int, strategy:str): return (LeaveAllocationBatch.query .filter_by(leave_type_id=leave_type_id, fy_label=fy_label, strategy=strategy, is_revoked=False) .order_by(LeaveAllocationBatch.run_at.desc()) .first()) def fy_bounds_for_today(today: date): """Return (fy_start_date, fy_end_date) where FY is Apr–Mar containing 'today'.""" fy_start_year = today.year if today.month >= 4 else today.year - 1 fy_start = date(fy_start_year, 4, 1) fy_end = date(fy_start_year + 1, 3, 31) return fy_start, fy_end def months_in_fy_until(today: date): """List of (year, month) from Apr..today’s month within the same FY.""" fy_start, _ = fy_bounds_for_today(today) months = [(fy_start.year, m) for m in range(4, 13)] # Apr..Dec months += [(fy_start.year + 1, m) for m in range(1, 4)] # Jan..Mar cut = [] for (y, m) in months: cut.append((y, m)) if y == today.year and m == today.month: break return cut def months_after_freeze_until(today: date, freeze_year:int, freeze_month:int): """Months strictly AFTER (freeze_year, freeze_month) up to today's month within same FY.""" all_months = months_in_fy_until(today) return [(y, m) for (y, m) in all_months if (y, m) > (freeze_year, freeze_month)] def calc_chl_earned_leave_total_after_freeze(doj: Optional[date], today: date, freeze_year:int, freeze_month:int) -> float: """ CHL EL rule, but only counting months strictly AFTER the freeze month. - If DOJ <= FY start: Apr=3, May=2, Jun–current=1 each (but only those months > freeze cut) - If DOJ > FY start: 1 per month from max(DOJ month, freeze+1) up to current """ fy_start, _ = fy_bounds_for_today(today) months = months_after_freeze_until(today, freeze_year, freeze_month) if doj is None: doj = date(1900, 1, 1) total = 0.0 if doj <= fy_start: for (_, m) in months: if m == 4: total += 3 elif m == 5: total += 2 else: total += 1 return total # New in FY: start from max(DOJ, freeze+1) for (y, m) in months: if (y, m) >= (doj.year, doj.month): total += 1 return total def _sum_credited_since(emp_id:int, lt_id:int, since_dt:datetime) -> float: """Sum of non-revoked credit deltas for this employee/type credited on/after since_dt.""" q = (db.session.query(db.func.coalesce(db.func.sum(LeaveAllocationEntry.credit_delta), 0.0)) .join(LeaveAllocationBatch, LeaveAllocationBatch.id == LeaveAllocationEntry.batch_id) .filter( LeaveAllocationEntry.employee_id == emp_id, LeaveAllocationEntry.leave_type_id == lt_id, LeaveAllocationBatch.is_revoked == False, LeaveAllocationBatch.run_at >= since_dt )) return float(q.scalar() or 0.0) # -------------------------------------------------------------------------------------------------------------------- _UNPAID_LEAVE_CODES = {"LWP", "LOP", "UL"} # allow common aliases def sync_leave_to_attendance( db, employee_id: int, start_date: date, end_date: date, leave_type: LeaveType, half_day: bool = False, half_day_session: str = "FH", ): """ Attendance sync that correctly marks LWP/LOP/UL as unpaid (lop + remarks='LWP'), and other leaves as paid_leave. """ code = (getattr(leave_type, "code", "") or "").strip().upper() name = (getattr(leave_type, "name", "") or "").strip() d = start_date while d <= end_date: att = Attendance.query.filter_by(employee_id=employee_id, date=d).first() if not att: att = Attendance(employee_id=employee_id, date=d) db.session.add(att) # Override attendance to reflect approved leave att.source = "Leave" # Clear timings so it doesn't look like "present by hours" att.check_in_time = None att.check_out_time = None att.working_hours = 0 # --- fix: unpaid vs paid --- if code in _UNPAID_LEAVE_CODES: att.status = "lop" att.remarks = "LWP" else: att.status = "paid_leave" # keep remarks informative (code preferred, else name) att.remarks = code or name or "Leave" # Optional: half-day marker (doesn't change paid/unpaid decision) if half_day: # keep your payroll-safe convention att.remarks = "Half Day" if code not in _UNPAID_LEAVE_CODES else "LWP" if hasattr(att, "note"): tag = f"HD_LEAVE:{(half_day_session or 'FH').upper()}" att.note = tag d += timedelta(days=1) # ----------------------------- # LEAVE TYPE MASTER # ----------------------------- @leave_bp.route('/leave-types') @login_required def view_leave_types(): leave_types = LeaveType.query.order_by(LeaveType.id).all() return render_template('leave/leave_types.html', leave_types=leave_types) @leave_bp.route('/leave-types/add', methods=['POST']) @login_required def add_leave_type(): form = request.form lt = LeaveType( name=form['name'], code=form['code'].upper(), default_quota=float(form.get('default_quota', 0)), allow_half_day='allow_half_day' in form, requires_approval='requires_approval' in form, carry_forward='carry_forward' in form, encashable='encashable' in form, sandwich_applicable='sandwich_applicable' in form, probation_restricted='probation_restricted' in form, remarks=form.get('remarks', '') ) db.session.add(lt) db.session.commit() flash('✅ Leave type added.', 'success') return redirect(url_for('leave.view_leave_types')) @leave_bp.route('/leave-types/edit/', methods=['GET', 'POST']) @login_required def edit_leave_type(type_id): lt = LeaveType.query.get_or_404(type_id) if request.method == 'POST': form = request.form lt.name = form['name'] lt.code = form['code'].upper() lt.default_quota = float(form.get('default_quota', 0)) lt.allow_half_day = 'allow_half_day' in form lt.requires_approval = 'requires_approval' in form lt.carry_forward = 'carry_forward' in form lt.encashable = 'encashable' in form lt.sandwich_applicable = 'sandwich_applicable' in form lt.probation_restricted = 'probation_restricted' in form lt.remarks = form.get('remarks', '') db.session.commit() flash('✅ Leave type updated.', 'success') return redirect(url_for('leave.view_leave_types')) return render_template('leave/edit_leave_type.html', lt=lt) @leave_bp.route('/leave-types/delete/', methods=['POST']) @login_required def delete_leave_type(type_id): lt = LeaveType.query.get_or_404(type_id) db.session.delete(lt) db.session.commit() flash('✅ Leave type deleted.', 'success') return redirect(url_for('leave.view_leave_types')) @leave_bp.route('/settings', methods=['GET', 'POST']) @login_required def leave_settings(): settings = LeaveSettings.query.first() if not settings: settings = LeaveSettings(sandwich_policy_enabled=True, rh_max_per_year=5) db.session.add(settings) db.session.commit() if request.method == 'POST': settings.sandwich_policy_enabled = 'sandwich_policy_enabled' in request.form settings.rh_max_per_year = int(request.form['rh_max_per_year']) db.session.commit() flash("✅ Settings updated successfully.", "success") return redirect(url_for('leave.leave_settings')) return render_template('leave/leave_settings.html', settings=settings) # ----------------------------- # LEAVE POLICY CONFIGURATION # ----------------------------- @leave_bp.route('/leave-policies') @login_required def view_leave_policies(): leave_types = LeaveType.query.order_by(LeaveType.name).all() policies = LeavePolicy.query.all() roles = Role.query.order_by(Role.name).all() departments = Department.query.order_by(Department.name).all() return render_template('leave/leave_policies.html', policies=policies, leave_types=leave_types, roles=roles, departments=departments ) @leave_bp.route('/leave-policies/add', methods=['POST']) @login_required def add_leave_policy(): form = request.form lp = LeavePolicy( company_id=None, leave_type_id=int(form['leave_type_id']), role_id=int(form.get('role_id')) if form.get('role_id') else None, department_id=int(form.get('department_id')) if form.get('department_id') else None, annual_quota=float(form.get('annual_quota') or 0), monthly_credit='monthly_credit' in form, max_leaves_per_month=float(form.get('max_leaves_per_month')) if form.get('max_leaves_per_month') else None, max_consecutive_days=int(form.get('max_consecutive_days')) if form.get('max_consecutive_days') else 30, join_after_cutoff_month=int(form.get('join_after_cutoff_month')) if form.get('join_after_cutoff_month') else None, rh_consecutive_limit=int(form.get('rh_consecutive_limit')) if form.get('rh_consecutive_limit') else None ) db.session.add(lp) db.session.commit() flash('✅ Leave policy added.', 'success') return redirect(url_for('leave.view_leave_policies')) @leave_bp.route('/leave-policies/delete/', methods=['POST']) @login_required def delete_leave_policy(policy_id): policy = LeavePolicy.query.get_or_404(policy_id) db.session.delete(policy) db.session.commit() flash('✅ Policy deleted.', 'success') return redirect(url_for('leave.view_leave_policies')) # ----------------------------- # HOLIDAY MASTER + RH CONFIG # ----------------------------- @leave_bp.route('/holidays') @login_required def view_holidays(): holidays = Holiday.query.order_by(Holiday.date).all() settings = LeaveSettings.query.all() return render_template('leave/holidays.html', holidays=holidays, settings=settings) @leave_bp.route('/holidays/add', methods=['POST']) @login_required def add_holiday(): form = request.form # Validate and normalize inputs try: date_ = datetime.strptime(form['date'], '%Y-%m-%d').date() except Exception: flash('❌ Invalid date format.', 'danger') return redirect(url_for('leave.view_holidays')) name = (form.get('name') or '').strip() raw_type = (form.get('type') or '').strip() canonical_type = _normalize_holiday_type(raw_type) # <— normalize here # Location handling (keep your existing logic) raw_loc = (form.get('location') or '').strip() if not raw_loc or raw_loc.lower() in {'other', 'others', 'other location', 'others location', '-', 'all'}: location = 'ALL' else: location = raw_loc if not name: flash('❌ Holiday name is required.', 'danger') return redirect(url_for('leave.view_holidays')) if not canonical_type: flash('❌ Invalid holiday type.', 'danger') return redirect(url_for('leave.view_holidays')) # Friendly pre-check (matches composite unique key) exists = Holiday.query.filter( and_(Holiday.date == date_, Holiday.location == location) ).first() if exists: flash(f"⚠️ A holiday already exists on {date_.isoformat()} for '{location}'.", 'warning') return redirect(url_for('leave.view_holidays')) # Save canonical type: 'mandatory' or 'optional' h = Holiday(date=date_, name=name, type=canonical_type, location=location) db.session.add(h) try: db.session.commit() flash('✅ Holiday added.', 'success') except IntegrityError: db.session.rollback() clash_locs = [x.location for x in Holiday.query.filter(Holiday.date == date_).all()] where = ", ".join(sorted(set(clash_locs))) or "unknown" flash(f"❌ Conflict: a holiday already exists on {date_.isoformat()} for [{where}].", 'danger') return redirect(url_for('leave.view_holidays')) @leave_bp.route('/holidays/delete/', methods=['POST']) @login_required def delete_holiday(holiday_id): h = Holiday.query.get_or_404(holiday_id) db.session.delete(h) db.session.commit() flash('✅ Holiday deleted.', 'success') return redirect(url_for('leave.view_holidays')) @leave_bp.route('/rh-config/update', methods=['POST']) @login_required def update_rh_config(): rh_max = int(request.form['rh_max_per_year']) sandwich_enabled = 'sandwich_policy_enabled' in request.form setting = LeaveSettings.query.filter_by(company_id=None).first() if not setting: setting = LeaveSettings(company_id=None) setting.rh_max_per_year = rh_max setting.sandwich_policy_enabled = sandwich_enabled db.session.add(setting) db.session.commit() flash('✅ RH config and sandwich policy updated!', 'success') return redirect(url_for('leave.view_holidays')) @leave_bp.route('/apply', methods=['GET', 'POST']) @login_required def apply_leave(): today = datetime.today().date() fy_start, fy_end = get_leave_year_range() fy_label = fy_start.year # <-- IMPORTANT (FY label, not calendar year) leave_types = LeaveType.query.order_by(LeaveType.name).all() balances = LeaveBalance.query.filter_by(employee_id=current_user.id, year=fy_label).all() balance_map = {b.leave_type_id: float(b.balance or 0.0) for b in balances} # holidays for sandwich warning only (doesn't affect final calc) holidays = Holiday.query.with_entities(Holiday.date).all() holiday_dates = {h.date for h in holidays} settings = LeaveSettings.query.first() max_rh = settings.rh_max_per_year if settings else 5 # Use leave-year window for RH counters ly_start, ly_end = get_leave_year_range() # Count approved RH as DAYS rh_approved_days = ( db.session.query(func.coalesce(func.sum(func.coalesce(LeaveRequest.total_days, 1.0)), 0.0)) .filter( LeaveRequest.employee_id == current_user.id, LeaveRequest.leave_type.has(code='RH'), LeaveRequest.start_date >= ly_start, LeaveRequest.end_date <= ly_end, LeaveRequest.status == 'final_approved' ) .scalar() ) # In-flight RH (pending + manager_approved + final_approved) as DAYS in_flight_statuses = ['pending', 'manager_approved', 'final_approved'] rh_in_flight_days = ( db.session.query(func.coalesce(func.sum(func.coalesce(LeaveRequest.total_days, 1.0)), 0.0)) .filter( LeaveRequest.employee_id == current_user.id, LeaveRequest.leave_type.has(code='RH'), LeaveRequest.start_date >= ly_start, LeaveRequest.end_date <= ly_end, LeaveRequest.status.in_(in_flight_statuses) ) .scalar() ) rh_remaining = max(0.0, float(max_rh) - float(rh_in_flight_days or 0.0)) if request.method == 'POST': form = request.form try: leave_type_id = int(form['leave_type_id']) start_date = datetime.strptime(form['start_date'], '%Y-%m-%d').date() end_date = datetime.strptime(form['end_date'], '%Y-%m-%d').date() except Exception: flash("Invalid form input. Please check dates and leave type.", "danger") return redirect(url_for('leave.apply_leave')) if end_date < start_date: flash("End date cannot be before start date.", "danger") return redirect(url_for('leave.apply_leave')) reason = (form.get('reason') or '').strip() if not reason: flash("Reason is required.", "danger") return redirect(url_for('leave.apply_leave')) half_day = ('half_day' in form) # ✅ NEW: Session wiring from Apply Leave half_day_session = (request.form.get("half_day_session") or "FH").upper() if half_day_session not in ("FH", "SH"): half_day_session = "FH" leave_type = LeaveType.query.get(leave_type_id) if not leave_type: flash("Invalid leave type selected.", "danger") return redirect(url_for('leave.apply_leave')) # ---------------- RH / NON-RH TOTAL DAYS ---------------- if leave_type.code == 'RH': # RH must be single-day if end_date != start_date: flash("❌ RH must be a single-day leave. Please select exactly one Optional Holiday date.", "danger") return redirect(url_for('leave.apply_leave')) # Eligible date must be Optional for employee location rh_whitelist = get_optional_holiday_dates(ly_start, ly_end, employee=current_user) if start_date not in rh_whitelist: flash("❌ Only Optional (RH) holidays can be selected for RH.", "danger") return redirect(url_for('leave.apply_leave')) requested_units = 0.5 if half_day else 1.0 if (float(rh_in_flight_days or 0.0) + requested_units) > float(max_rh): remaining = max(0.0, float(max_rh) - float(rh_in_flight_days or 0.0)) flash(f"❌ RH limit exceeded. Remaining RH this year: {remaining} day(s).", "danger") return redirect(url_for('leave.apply_leave')) total_days = requested_units else: # Non-RH total days if half_day: if end_date != start_date: flash("❌ Half-day leave can only be applied for a single day.", "danger") return redirect(url_for('leave.apply_leave')) total_days = 0.5 else: total_days = (end_date - start_date).days + 1 force_split_due_to_policy = False policy_paid_days = 0 policy_lwp_days = 0 # ---------------- POLICY VALIDATION ---------------- policy = _get_applicable_leave_policy(current_user, leave_type_id) if policy: # Max consecutive days # Max consecutive days if policy.max_consecutive_days and float(total_days) > float(policy.max_consecutive_days): if leave_type.code == 'RH': flash( f"❌ Policy allows maximum {policy.max_consecutive_days} consecutive day(s) for {leave_type.name}.", "danger" ) return redirect(url_for('leave.apply_leave')) max_allowed = float(policy.max_consecutive_days or 0) excess_days = float(total_days) - max_allowed # for half day cases, do not auto-split here if float(total_days) == 0.5: flash( f"❌ Policy allows maximum {policy.max_consecutive_days} consecutive day(s) for {leave_type.name}.", "danger" ) return redirect(url_for('leave.apply_leave')) user_confirmed_excess = (form.get('confirm_excess_as_lwp') or '').strip().lower() == 'yes' if not user_confirmed_excess: flash( f"❌ Policy allows maximum {int(max_allowed)} consecutive day(s) for {leave_type.name}. " f"Extra {int(excess_days)} day(s) will be marked as LWP if you continue.", "danger" ) return redirect(url_for('leave.apply_leave')) # Store these for later split logic force_split_due_to_policy = True policy_paid_days = int(max_allowed) policy_lwp_days = int(excess_days) else: force_split_due_to_policy = False policy_paid_days = 0 policy_lwp_days = 0 # Join-after-cutoff restriction if policy.join_after_cutoff_month: doj = getattr(current_user, 'date_of_joining', None) if doj and doj.year == start_date.year and doj.month >= int(policy.join_after_cutoff_month): flash( f"❌ As per policy, employees joining on/after month {policy.join_after_cutoff_month} cannot apply for {leave_type.name} in this cycle.", "danger" ) return redirect(url_for('leave.apply_leave')) # Max leaves per month monthly_cap = getattr(policy, 'max_leaves_per_month', None) if monthly_cap is not None: month_start = start_date.replace(day=1) if start_date.month == 12: month_end = start_date.replace(year=start_date.year + 1, month=1, day=1) - timedelta(days=1) else: month_end = start_date.replace(month=start_date.month + 1, day=1) - timedelta(days=1) overlapping_reqs = LeaveRequest.query.filter( LeaveRequest.employee_id == current_user.id, LeaveRequest.leave_type_id == leave_type_id, LeaveRequest.status.in_(['pending', 'manager_approved', 'final_approved']), LeaveRequest.start_date <= month_end, LeaveRequest.end_date >= month_start ).all() existing_month_days = sum( _days_overlap_in_month(r, month_start, month_end) for r in overlapping_reqs ) if float(existing_month_days) + float(total_days) > float(monthly_cap): remaining = max(0.0, float(monthly_cap) - float(existing_month_days)) flash( f"❌ Monthly limit exceeded for {leave_type.name}. Remaining allowed this month: {remaining} day(s).", "danger" ) return redirect(url_for('leave.apply_leave')) # Sandwich warning (display only) span_days = 1 if total_days == 0.5 else (end_date - start_date).days + 1 date_set = {start_date + timedelta(days=i) for i in range(span_days)} sandwich_days = {d for d in date_set if d.weekday() in (5, 6) or d in holiday_dates} if sandwich_days: flash(f"⚠️ Sandwich leave: includes {len(sandwich_days)} weekend/holiday day(s).", "warning") # ------------------ Create LeaveRequest(s), split to LWP if needed ------------------ def _create_lr(lt_id, d1, d2, days, half=False): lr = LeaveRequest( employee_id=current_user.id, leave_type_id=lt_id, start_date=d1, end_date=d2, total_days=days, reason=reason, half_day=bool(half), half_day_session=half_day_session if bool(half) else None, status='pending', ) db.session.add(lr) return lr created_reqs = [] # RH or LWP: no balance validation if leave_type.code in ('LWP', 'RH'): lr = _create_lr(leave_type_id, start_date, end_date, total_days, (total_days == 0.5)) created_reqs.append(lr) else: current_balance = float(balance_map.get(leave_type_id, 0.0) or 0.0) # HALF-DAY request (0.5) if total_days == 0.5: if current_balance >= 0.5: lr = _create_lr(leave_type_id, start_date, end_date, 0.5, True) created_reqs.append(lr) else: lwp = (LeaveType.query .filter(db.func.upper(LeaveType.code).in_(['LWP', 'LOP', 'UL'])) .first()) if not lwp: flash(f"Insufficient balance: {current_balance} day(s) left and no LWP fallback configured.", "danger") return redirect(url_for('leave.apply_leave')) lr = _create_lr(lwp.id, start_date, end_date, 0.5, True) created_reqs.append(lr) flash("⚠️ No balance left. Applied as LWP (0.5 day).", "warning") else: requested = int(total_days) lwp = (LeaveType.query .filter(db.func.upper(LeaveType.code).in_(['LWP', 'LOP', 'UL'])) .first()) # Case 1: policy max-consecutive exceeded and user agreed to split if force_split_due_to_policy: if not lwp: flash("❌ LWP leave type is not configured, so excess days cannot be split.", "danger") return redirect(url_for('leave.apply_leave')) allowed_paid_days = min(policy_paid_days, int(current_balance)) balance_shortfall_days = max(0, policy_paid_days - allowed_paid_days) total_lwp_days = int(policy_lwp_days + balance_shortfall_days) if allowed_paid_days > 0: partA_start = start_date partA_end = start_date + timedelta(days=allowed_paid_days - 1) lrA = _create_lr(leave_type_id, partA_start, partA_end, allowed_paid_days, False) created_reqs.append(lrA) if total_lwp_days > 0: partB_start = start_date + timedelta(days=allowed_paid_days) partB_end = end_date lrB = _create_lr(lwp.id, partB_start, partB_end, total_lwp_days, False) created_reqs.append(lrB) flash( f"⚠️ Policy max exceeded. Split into {allowed_paid_days} day(s) {leave_type.code} " f"+ {total_lwp_days} day(s) LWP.", "warning" ) # Case 2: normal balance-based processing elif current_balance >= requested: lr = _create_lr(leave_type_id, start_date, end_date, requested, False) created_reqs.append(lr) else: if not lwp: flash(f"Insufficient balance: {current_balance} day(s) left and no LWP fallback configured.", "danger") return redirect(url_for('leave.apply_leave')) paid_days = int(max(0, min(requested, int(current_balance)))) lwp_days = int(requested - paid_days) if paid_days > 0: partA_start = start_date partA_end = start_date + timedelta(days=paid_days - 1) lrA = _create_lr(leave_type_id, partA_start, partA_end, paid_days, False) created_reqs.append(lrA) if lwp_days > 0: partB_start = start_date + timedelta(days=paid_days) partB_end = end_date lrB = _create_lr(lwp.id, partB_start, partB_end, lwp_days, False) created_reqs.append(lrB) flash(f"⚠️ Not enough balance. Split into {paid_days} day(s) {leave_type.code} + {lwp_days} day(s) LWP.", "warning") # Persist db.session.flush() # Inform to inform_ids = [int(i) for i in form.getlist('inform_to_ids') if i] if inform_ids: for lr in created_reqs: for eid in inform_ids: db.session.add(LeaveInformee( leave_request_id=lr.id, employee_id=eid )) db.session.commit() # email reporting manager (one summary mail) _send_leave_submission_email_to_manager(created_reqs, current_user) # Notifications def _rng(d1, d2): return d1.strftime('%d-%b-%Y') if d1 == d2 else f"{d1:%d-%b-%Y} → {d2:%d-%b-%Y}" if len(created_reqs) == 1: lr = created_reqs[0] rng = _rng(lr.start_date, lr.end_date) lt = LeaveType.query.get(lr.leave_type_id) notify_employee( employee_id=current_user.id, title="Leave request submitted", message=f"Your {lt.name} request ({rng}) has been submitted and is pending manager approval.", link=_link('leave.view_history') ) mgr_id = getattr(current_user, 'reporting_manager_id', None) if mgr_id: notify_employee( employee_id=mgr_id, title="Leave approval needed", message=f"{current_user.full_name} applied for {lt.name} ({rng}). Please review.", link=_link('leave.manager_pending') ) hr_list = _hr_ids() if hr_list: notify_many( employee_ids=hr_list, title="New leave request (FYI)", message=f"{current_user.full_name} applied for {lt.name} ({rng}). Awaiting manager action.", link=_link('leave.hr_pending') ) else: parts = [] for r in created_reqs: lt = LeaveType.query.get(r.leave_type_id) parts.append(f"{lt.code} ({_rng(r.start_date, r.end_date)})") summary = " + ".join(parts) notify_employee( employee_id=current_user.id, title="Leave requests submitted (split)", message=f"Submitted: {summary}. Pending manager approval.", link=_link('leave.view_history') ) mgr_id = getattr(current_user, 'reporting_manager_id', None) if mgr_id: notify_employee( employee_id=mgr_id, title="Leave approval needed", message=f"{current_user.full_name} applied: {summary}. Please review.", link=_link('leave.manager_pending') ) hr_list = _hr_ids() if hr_list: notify_many( employee_ids=hr_list, title="New leave requests (FYI)", message=f"{current_user.full_name} applied: {summary}. Awaiting manager action.", link=_link('leave.hr_pending') ) flash("✅ Leave request submitted.", "success") return redirect(url_for('leave.view_history')) # GET data rh_type = LeaveType.query.filter_by(code='RH').first() rh_type_id = rh_type.id if rh_type else None informable_emps = ( Employee.query .filter(Employee.id != current_user.id) .order_by(Employee.full_name) .all() ) leave_policy_map = {} for lt in leave_types: pol = _get_applicable_leave_policy(current_user, lt.id) leave_policy_map[lt.id] = { "max_consecutive_days": int(pol.max_consecutive_days) if pol and pol.max_consecutive_days else None } return render_template( 'leave/apply_leave.html', leave_types=leave_types, balances=balances, max_rh=max_rh, rh_remaining=rh_remaining, current_year=fy_label, informable_emps=informable_emps, rh_type_id=rh_type_id, leave_policy_map=leave_policy_map, ) @leave_bp.route('/history') @login_required def view_history(): leave_requests = LeaveRequest.query.filter_by(employee_id=current_user.id).order_by(LeaveRequest.applied_on.desc()).all() return render_template('leave/view_history.html', leave_requests=leave_requests) # ----------------------------- # MANAGER PENDING + APPROVE # ----------------------------- from calendar import monthrange @leave_bp.route('/manager/pending') @login_required def manager_pending(): from calendar import monthrange from datetime import date # ---- filters from query params ---- emp_id = request.args.get('employee_id', type=int) month_from = request.args.get('month_from', type=str) # 'YYYY-MM' month_to = request.args.get('month_to', type=str) # 'YYYY-MM' remark_q = (request.args.get('remark', '') or '').strip() def month_bounds(yyyy_mm: str): y, m = map(int, yyyy_mm.split('-')) start = date(y, m, 1) end = date(y, m, monthrange(y, m)[1]) return start, end # restrict requests to manager’s team team_ids = [e.id for e in Employee.query.filter_by(reporting_manager_id=current_user.id).all()] # if you ALSO want the manager to see their own leaves here, uncomment: # if current_user.id not in team_ids: # team_ids.append(current_user.id) if not team_ids: team_requests = [] else: q = (LeaveRequest.query .join(Employee, LeaveRequest.employee_id == Employee.id) .filter(LeaveRequest.employee_id.in_(team_ids)) .order_by(LeaveRequest.applied_on.desc())) # Employee filter if emp_id: q = q.filter(LeaveRequest.employee_id == emp_id) # Month range filter (on start_date) if month_from and month_to: d1 = month_bounds(month_from)[0] d2 = month_bounds(month_to)[1] q = q.filter(LeaveRequest.start_date.between(d1, d2)) elif month_from: d1, d2 = month_bounds(month_from) q = q.filter(LeaveRequest.start_date.between(d1, d2)) elif month_to: d2 = month_bounds(month_to)[1] q = q.filter(LeaveRequest.start_date <= d2) team_requests = q.all() # ---- split by status ONLY ---- pending_requests = [r for r in team_requests if (r.status or '').lower() == 'pending'] processed_requests = [r for r in team_requests if (r.status or '').lower() != 'pending'] # Manager remark filter applies to processed (logs exist there) if remark_q: rq = remark_q.lower() processed_requests = [ r for r in processed_requests if any( (log.approver_role == 'manager') and (log.remark or '').lower().find(rq) >= 0 for log in getattr(r, 'approval_logs', []) ) ] # if you also want to filter pending by the employee's reason, you can: # pending_requests = [r for r in pending_requests if (r.reason or '').lower().find(rq) >= 0] # Team employees for the filter dropdown team_employees = Employee.query.filter(Employee.id.in_(team_ids)).order_by(Employee.full_name).all() if team_ids else [] return render_template( 'leave/manager_pending.html', pending_requests=pending_requests, processed_requests=processed_requests, employees=team_employees, # echo filter values so the template can keep selections emp_id=emp_id, month_from=month_from, month_to=month_to, remark_q=remark_q ) @leave_bp.route('/manager/approve/', methods=['POST']) @login_required def manager_approve(req_id): req = LeaveRequest.query.get_or_404(req_id) if req.status != 'pending': flash('⚠️ Already processed.', 'warning') return redirect(url_for('leave.manager_pending')) req.status = 'manager_approved' db.session.add(LeaveApprovalLog( leave_request_id=req.id, approver_id=current_user.id, approver_role='manager', status='approved', remark=request.form['remark'] )) db.session.commit() # Notifications rng = _format_range(req.start_date, req.end_date) emp = Employee.query.get(req.employee_id) lt = LeaveType.query.get(req.leave_type_id) # 4.1 Notify employee notify_employee( employee_id=req.employee_id, title="Leave approved by Manager", message=f"Your {lt.name} request ({rng}) was approved by your manager and sent to HR.", link=_link('leave.view_history') ) # 4.2 Notify HR to act hr_list = _hr_ids() if hr_list: notify_many( employee_ids=hr_list, title="Leave request awaiting HR action", message=f"{emp.full_name}'s {lt.name} ({rng}) needs HR approval.", link=_link('leave.hr_pending') ) _send_leave_inform_emails(req) flash('✅ Approved and forwarded to HR.', 'success') return redirect(url_for('leave.manager_pending')) # ----------------------------- # HR PENDING + APPROVE # ----------------------------- from calendar import monthrange from datetime import date from flask import request, render_template @leave_bp.route('/hr/pending') @login_required def hr_pending(): # ---- filters ---- emp_id = request.args.get('employee_id', type=int) month_from = request.args.get('month_from', type=str) # 'YYYY-MM' month_to = request.args.get('month_to', type=str) # 'YYYY-MM' remark_q = (request.args.get('remark', '') or '').strip() # HR remark contains def month_bounds(yyyy_mm: str): y, m = map(int, yyyy_mm.split('-')) start = date(y, m, 1) end = date(y, m, monthrange(y, m)[1]) return start, end # Base queries q_pending = (LeaveRequest.query .filter(LeaveRequest.status == 'manager_approved') .order_by(LeaveRequest.applied_on.desc())) q_processed = (LeaveRequest.query .filter(LeaveRequest.id.in_( db.session.query(LeaveApprovalLog.leave_request_id) .filter(LeaveApprovalLog.approver_role == 'hr') )) .order_by(LeaveRequest.applied_on.desc())) # Apply Employee filter if emp_id: q_pending = q_pending.filter(LeaveRequest.employee_id == emp_id) q_processed = q_processed.filter(LeaveRequest.employee_id == emp_id) # Apply Month range on start_date if month_from and month_to: d1 = month_bounds(month_from)[0] d2 = month_bounds(month_to)[1] q_pending = q_pending.filter(LeaveRequest.start_date.between(d1, d2)) q_processed = q_processed.filter(LeaveRequest.start_date.between(d1, d2)) elif month_from: d1, d2 = month_bounds(month_from) q_pending = q_pending.filter(LeaveRequest.start_date.between(d1, d2)) q_processed = q_processed.filter(LeaveRequest.start_date.between(d1, d2)) elif month_to: d2 = month_bounds(month_to)[1] q_pending = q_pending.filter(LeaveRequest.start_date <= d2) q_processed = q_processed.filter(LeaveRequest.start_date <= d2) pending_requests = q_pending.all() processed_requests = q_processed.all() # HR remark filter applies to processed (based on HR logs) if remark_q: rq = remark_q.lower() processed_requests = [ r for r in processed_requests if any( (log.approver_role == 'hr') and (log.remark or '').lower().find(rq) >= 0 for log in getattr(r, 'approval_logs', []) ) ] # If you also want to filter PENDING by the employee's reason, uncomment: # pending_requests = [r for r in pending_requests if (r.reason or '').lower().find(rq) >= 0] # Dropdown data (all employees) employees = Employee.query.order_by(Employee.full_name).all() return render_template( 'leave/hr_pending.html', pending_requests=pending_requests, processed_requests=processed_requests, employees=employees, emp_id=emp_id, month_from=month_from, month_to=month_to, remark_q=remark_q ) @leave_bp.route('/hr/approve/', methods=['POST']) @login_required def hr_approve(req_id): req = LeaveRequest.query.get_or_404(req_id) leave_type = LeaveType.query.get(req.leave_type_id) fy_start, fy_end = get_leave_year_range() fy_label = fy_start.year if req.status != 'manager_approved': flash('⚠️ Not eligible for HR approval.', 'warning') return redirect(url_for('leave.hr_pending')) if not leave_type: flash('❌ Leave type not found for this request.', 'danger') return redirect(url_for('leave.hr_pending')) # Deduct balance only for non-RH, non-LWP if leave_type.code not in ('RH', 'LWP'): balance = LeaveBalance.query.filter_by( employee_id=req.employee_id, leave_type_id=req.leave_type_id, year=fy_label ).first() if not balance: balance = LeaveBalance( employee_id=req.employee_id, leave_type_id=req.leave_type_id, year=fy_label, balance=0 ) db.session.add(balance) available = float(balance.balance or 0.0) if available < float(req.total_days or 0.0): flash(f'❌ Insufficient balance. Available: {available}', 'danger') return redirect(url_for('leave.hr_pending')) balance.balance = available - float(req.total_days or 0.0) # Finalise req.status = 'final_approved' db.session.add(LeaveApprovalLog( leave_request_id=req.id, approver_id=current_user.id, approver_role='hr', status='approved', remark=request.form.get('remark', '').strip() )) db.session.commit() # Sync to attendance sync_leave_to_attendance( db, req.employee_id, req.start_date, req.end_date, leave_type, half_day=bool(req.half_day or req.total_days == 0.5), half_day_session=getattr(req, "half_day_session", "FH"), ) db.session.commit() rng = _format_range(req.start_date, req.end_date) emp = Employee.query.get(req.employee_id) lt = LeaveType.query.get(req.leave_type_id) notify_employee( employee_id=req.employee_id, title="Leave fully approved", message=f"Your {lt.name} request ({rng}) has been approved by HR.", link=_link('leave.view_history') ) if emp and emp.reporting_manager_id: notify_employee( employee_id=emp.reporting_manager_id, title="Leave approved (HR)", message=f"{emp.full_name}'s {lt.name} ({rng}) is fully approved.", link=_link('leave.manager_pending') ) flash('✅ Final approval done & attendance updated.', 'success') return redirect(url_for('leave.hr_pending')) # ----------------------------- # Comp-Off Application # ----------------------------- @leave_bp.route('/comp-off/apply', methods=['GET', 'POST']) @login_required def apply_comp_off(): if request.method == 'POST': work_date = datetime.strptime(request.form['work_date'], '%Y-%m-%d').date() reason = (request.form['reason'] or '').strip() valid_until = CompOffRequest.default_valid_until(work_date, days=30) if CompOffRequest.query.filter_by(employee_id=current_user.id, work_date=work_date).first(): flash('⚠️ Comp-off already requested for this date.', 'warning') return redirect(url_for('leave.apply_comp_off')) db.session.add(CompOffRequest( employee_id=current_user.id, work_date=work_date, reason=reason, valid_until=valid_until, status='pending' )) db.session.commit() flash('✅ Comp-off request submitted.', 'success') return redirect(url_for('leave.apply_comp_off')) # --- Filters for "my requests" --- d_from = request.args.get('date_from') d_to = request.args.get('date_to') status_q = (request.args.get('status') or '').lower() # pending|approved|rejected|all q_text = (request.args.get('q') or '').strip().lower() q = (CompOffRequest.query .filter(CompOffRequest.employee_id == current_user.id) .order_by(CompOffRequest.work_date.desc())) if d_from: try: q = q.filter(CompOffRequest.work_date >= datetime.strptime(d_from, "%Y-%m-%d").date()) except Exception: pass if d_to: try: q = q.filter(CompOffRequest.work_date <= datetime.strptime(d_to, "%Y-%m-%d").date()) except Exception: pass if status_q and status_q != 'all': q = q.filter(CompOffRequest.status == status_q) if q_text: q = q.filter(db.func.lower(CompOffRequest.reason).contains(q_text)) existing = q.all() return render_template( 'leave/comp_off_apply.html', requests=existing, date_from=d_from, date_to=d_to, status_q=status_q, q_text=q_text ) from datetime import datetime, date from flask import request, render_template, redirect, url_for, flash from flask_login import login_required, current_user from sqlalchemy.orm import joinedload # ----------------------------- # Helpers # ----------------------------- 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", "demo admin") # ----------------------------- # Comp-Off Approval (Admin/HR) # ----------------------------- @leave_bp.route('/comp-off/pending') @login_required def pending_comp_off(): if not _is_hr_admin_user(): flash("Access denied.", "danger") return redirect(url_for("employee.dashboard")) emp_id = request.args.get('employee_id', type=int) d_from = request.args.get('date_from') d_to = request.args.get('date_to') q_text = (request.args.get('q') or '').strip().lower() qp = CompOffRequest.query.options(joinedload(CompOffRequest.employee)).order_by(CompOffRequest.work_date.desc()) qr = CompOffRequest.query.options(joinedload(CompOffRequest.employee)).order_by(CompOffRequest.work_date.desc()) if emp_id: qp = qp.filter(CompOffRequest.employee_id == emp_id) qr = qr.filter(CompOffRequest.employee_id == emp_id) if d_from: try: d1 = datetime.strptime(d_from, "%Y-%m-%d").date() qp = qp.filter(CompOffRequest.work_date >= d1) qr = qr.filter(CompOffRequest.work_date >= d1) except Exception: pass if d_to: try: d2 = datetime.strptime(d_to, "%Y-%m-%d").date() qp = qp.filter(CompOffRequest.work_date <= d2) qr = qr.filter(CompOffRequest.work_date <= d2) except Exception: pass if q_text: qp = qp.filter(db.func.lower(CompOffRequest.reason).contains(q_text)) qr = qr.filter(db.func.lower(CompOffRequest.reason).contains(q_text)) pending_requests = qp.filter(CompOffRequest.status == 'pending').all() processed_requests = qr.filter(CompOffRequest.status.in_(['approved', 'rejected'])).all() employees = Employee.query.order_by(Employee.full_name).all() return render_template( 'leave/comp_off_pending.html', pending_requests=pending_requests, processed_requests=processed_requests, employees=employees, emp_id=emp_id, date_from=d_from, date_to=d_to, q_text=q_text, ) @leave_bp.route('/comp-off/approve/', methods=['POST']) @login_required def approve_comp_off(req_id): if not _is_hr_admin_user(): flash("Access denied.", "danger") return redirect(url_for("employee.dashboard")) req = CompOffRequest.query.get_or_404(req_id) if (req.status or "").lower() != 'pending': flash('⚠️ Already processed.', 'warning') return redirect(url_for('leave.pending_comp_off')) # Expiry guard today = date.today() if getattr(req, "valid_until", None) and req.valid_until < today: req.status = "rejected" if hasattr(req, "rejection_reason"): req.rejection_reason = "Expired comp-off request" if hasattr(req, "approved_by_id"): req.approved_by_id = current_user.id if hasattr(req, "approved_on"): req.approved_on = datetime.utcnow() db.session.commit() flash("❌ Cannot approve: comp-off request expired.", "danger") return redirect(url_for('leave.pending_comp_off')) # Find Comp Off leave type co_type = (LeaveType.query .filter(db.func.upper(LeaveType.code).in_(['CO', 'COMP_OFF'])) .first()) if not co_type: flash('❌ Comp Off Leave Type missing (code CO or COMP_OFF). Create it first.', 'danger') return redirect(url_for('leave.pending_comp_off')) # FY label fy_start, fy_end = get_leave_year_range() fy_label = fy_start.year try: # 1) Mark approved + approver fields if present req.status = 'approved' if hasattr(req, "approved_by_id"): req.approved_by_id = current_user.id if hasattr(req, "approved_on"): req.approved_on = datetime.utcnow() # 2) Credit LeaveBalance (IMPORTANT: use year, not credit_year) bal = (LeaveBalance.query .filter_by(employee_id=req.employee_id, leave_type_id=co_type.id, year=fy_label) .first()) if not bal: bal = LeaveBalance( employee_id=req.employee_id, leave_type_id=co_type.id, year=fy_label, balance=0.0 ) db.session.add(bal) db.session.flush() bal.balance = float(bal.balance or 0.0) + 1.0 db.session.commit() flash('✅ Comp-off approved and 1 day credited to CO balance.', 'success') except Exception: db.session.rollback() current_app.logger.exception("Comp-off approval failed") flash("❌ Approval failed due to an internal error.", "danger") return redirect(url_for('leave.pending_comp_off')) @leave_bp.route('/comp-off/reject/', methods=['POST']) @login_required def reject_comp_off(req_id): if not _is_hr_admin_user(): flash("Access denied.", "danger") return redirect(url_for("employee.dashboard")) req = CompOffRequest.query.get_or_404(req_id) if (req.status or "").lower() != 'pending': flash('⚠️ Already processed.', 'warning') return redirect(url_for('leave.pending_comp_off')) reason = (request.form.get("reason") or "").strip() req.status = 'rejected' if hasattr(req, "rejection_reason"): req.rejection_reason = reason or None if hasattr(req, "approved_by_id"): req.approved_by_id = current_user.id if hasattr(req, "approved_on"): req.approved_on = datetime.utcnow() db.session.commit() flash('❌ Comp-off rejected.', 'warning') return redirect(url_for('leave.pending_comp_off')) @leave_bp.route('/allocation', methods=['GET', 'POST']) @login_required def leave_allocation(): leave_types = LeaveType.query.order_by(LeaveType.name).all() results = [] today = date.today() fy_start, fy_end = fy_bounds_for_today(today) selected_year = fy_start.year # FY label (your LeaveBalance.year) if request.method == 'POST': form = request.form leave_type_id = int(form['leave_type_id']) strategy = form['strategy'] force = ('force' in form) # allow override of guard leave_type = LeaveType.query.get_or_404(leave_type_id) # ----- Freeze cutoff (from form or default to Aug of this FY) ----- # Input comes as YYYY-MM (type="month") freeze_till = (form.get('freeze_till') or '').strip() # e.g., "2025-08" if freeze_till: f_year, f_month = map(int, freeze_till.split('-')) else: # Default: freeze till August of the FY label (Apr-Mar) f_year, f_month = selected_year, 8 # ‘since’ moment is the first day of the month after the freeze # e.g., freeze 2025-08 -> since_dt = 2025-09-01 00:00:00 if f_month == 12: since_dt = datetime(f_year + 1, 1, 1) else: since_dt = datetime(f_year, f_month + 1, 1) # Guard against accidental re-run for same FY + LT + strategy (unless forced) existing_batch = _latest_non_revoked_batch(leave_type_id, selected_year, strategy) if existing_batch and not force: flash(f"⚠️ An allocation batch already exists for FY {selected_year}, " f"{leave_type.code}, strategy {strategy}. Tick 'Force apply' to proceed or Revoke the batch.", "warning") return render_template( 'leave/leave_allocation.html', leave_types=leave_types, selected_year=selected_year, batch_result=None, guard_exists=True, guard_batch=existing_batch, freeze_default=f"{f_year:04d}-{f_month:02d}" ) # ---- Build a new batch audit shell ---- batch_key = _make_batch_key(selected_year, leave_type.code, strategy) batch = LeaveAllocationBatch( batch_key=batch_key, leave_type_id=leave_type.id, fy_label=selected_year, strategy=strategy, run_by_id=getattr(current_user, 'id', None), note=f"Allocation post-freeze (frozen till {f_year}-{f_month:02d})" ) db.session.add(batch) db.session.flush() # get batch.id employees = Employee.query.filter(db.func.lower(Employee.status) == 'active').all() for emp in employees: doj = getattr(emp, 'date_of_joining', None) # If DOJ is after FY end, skip if doj and doj > fy_end: results.append({"employee_name": emp.full_name, "allocated": 0, "remark": "Joined after FY, skipped"}) continue # Ensure there is a balance row for (emp, lt, FY) bal = (LeaveBalance.query .filter_by(employee_id=emp.id, leave_type_id=leave_type.id, year=selected_year) .with_for_update(read=True) .first()) if not bal: bal = LeaveBalance(employee_id=emp.id, leave_type_id=leave_type.id, year=selected_year, balance=0.0) db.session.add(bal) db.session.flush() prev = float(bal.balance or 0.0) # ---- Compute target_total for months strictly AFTER freeze ---- if strategy == 'monthly_credit': months = months_after_freeze_until(today, f_year, f_month) monthly = round((leave_type.default_quota or 0) / 12.0, 2) # cap at default_quota, but only for counted months target_total = round(min(len(months) * monthly, (leave_type.default_quota or 0)), 2) remark_core = f"{len(months)} month(s) × {monthly} post-freeze" elif strategy == 'chl_earned_leave': target_total = calc_chl_earned_leave_total_after_freeze(doj, today, f_year, f_month) remark_core = "CHL EL rule (post-freeze)" elif strategy == 'annual': # If annual and freeze is set, annual post-freeze is ambiguous. # Common approach: allocate 0 until after freeze, then allocate full once (if you want). # Here we credit nothing pre-freeze and full once after freeze month: months = months_after_freeze_until(today, f_year, f_month) target_total = float(leave_type.default_quota or 0) if months else 0.0 remark_core = "Annual post-freeze" else: results.append({"employee_name": emp.full_name, "allocated": 0, "remark": "❌ Unknown strategy"}) continue # ---- Delta = target_total - already_credited_since_freeze ---- already_since = _sum_credited_since(emp.id, leave_type.id, since_dt) delta = max(0.0, round(target_total - already_since, 2)) # Audit entry BEFORE changing balance entry = LeaveAllocationEntry( batch_id=batch.id, employee_id=emp.id, leave_type_id=leave_type.id, fy_label=selected_year, previous_balance=prev, new_balance=round(prev + delta, 2), credit_delta=delta ) db.session.add(entry) # Apply only the delta; never overwrite usage if delta > 0: bal.balance = round(prev + delta, 2) results.append({ "employee_name": emp.full_name, "allocated": delta, "remark": f"{strategy.replace('_',' ').title()} ({remark_core}); " f"freeze_till={f_year}-{f_month:02d}, target_post_freeze={target_total}, already_post_freeze={already_since}, was={prev}" }) db.session.commit() flash(f"✅ Allocation (post-freeze) applied in batch {batch.batch_key} for {len(results)} active employee(s).", "success") return render_template( 'leave/leave_allocation.html', leave_types=leave_types, selected_year=selected_year, batch_result={"leave_type": leave_type.name, "entries": results, "batch_key": batch.batch_key}, freeze_default=f"{f_year:04d}-{f_month:02d}" ) # GET # default freeze shown as Aug of current FY (editable in UI) return render_template( 'leave/leave_allocation.html', leave_types=leave_types, selected_year=selected_year, batch_result=None, freeze_default=f"{selected_year}-08" ) @leave_bp.route('/hr/reject/', methods=['POST']) @login_required def hr_reject(req_id): req = LeaveRequest.query.get_or_404(req_id) # Check if already acted on existing_log = LeaveApprovalLog.query.filter_by( leave_request_id=req.id, approver_role='hr' ).first() if existing_log: flash('⚠️ HR has already acted on this request.', 'warning') return redirect(url_for('leave.hr_pending')) req.status = 'rejected' db.session.add(LeaveApprovalLog( leave_request_id=req.id, approver_id=current_user.id, approver_role='hr', status='rejected', remark=request.form['remark'] )) db.session.commit() # Notifications rng = _format_range(req.start_date, req.end_date) emp = Employee.query.get(req.employee_id) lt = LeaveType.query.get(req.leave_type_id) # 7.1 Employee notify_employee( employee_id=req.employee_id, title="Leave rejected by HR", message=f"Your {lt.name} request ({rng}) was rejected by HR.", link=_link('leave.view_history') ) # 7.2 Manager (FYI) if emp and emp.reporting_manager_id: notify_employee( employee_id=emp.reporting_manager_id, title="Leave rejected (HR)", message=f"{emp.full_name}'s {lt.name} ({rng}) was rejected by HR.", link=_link('leave.manager_pending') ) flash('❌ Leave rejected by HR.', 'danger') return redirect(url_for('leave.hr_pending')) @leave_bp.route('/manager/reject/', methods=['POST']) @login_required def manager_reject(req_id): req = LeaveRequest.query.get_or_404(req_id) if req.status != 'pending': flash('⚠️ Already processed.', 'warning') return redirect(url_for('leave.manager_pending')) # Update the request status req.status = 'rejected' # Log the manager's rejection db.session.add(LeaveApprovalLog( leave_request_id=req.id, approver_id=current_user.id, approver_role='manager', status='rejected', remark=request.form['remark'] )) db.session.commit() # Notifications rng = _format_range(req.start_date, req.end_date) lt = LeaveType.query.get(req.leave_type_id) notify_employee( employee_id=req.employee_id, title="Leave rejected by Manager", message=f"Your {lt.name} request ({rng}) was rejected by your manager.", link=_link('leave.view_history') ) flash('❌ Leave rejected by Manager.', 'danger') return redirect(url_for('leave.manager_pending')) @leave_bp.route('/cancel/', methods=['POST']) @login_required def cancel_leave(req_id): req = LeaveRequest.query.get_or_404(req_id) if req.employee_id != current_user.id: flash('❌ You are not authorized to cancel this request.', 'danger') return redirect(url_for('leave.view_history')) if req.status != 'pending': flash('⚠️ Only pending requests can be cancelled.', 'warning') return redirect(url_for('leave.view_history')) req.status = 'cancelled' db.session.commit() # Notifications rng = _format_range(req.start_date, req.end_date) lt = LeaveType.query.get(req.leave_type_id) # 8.1 Manager mgr_id = getattr(current_user, 'reporting_manager_id', None) if mgr_id: notify_employee( employee_id=mgr_id, title="Leave request cancelled", message=f"{current_user.full_name} cancelled their {lt.name} request ({rng}).", link=_link('leave.manager_pending') ) # 8.2 HR hr_list = _hr_ids() if hr_list: notify_many( employee_ids=hr_list, title="Leave request cancelled", message=f"{current_user.full_name} cancelled {lt.name} ({rng}).", link=_link('leave.hr_pending') ) flash('✅ Leave request cancelled successfully.', 'success') return redirect(url_for('leave.view_history')) @leave_bp.route('/eligible-dates', methods=['GET']) @login_required def eligible_dates(): q_type = request.args.get('type', '').lower() ly_start, ly_end = get_leave_year_range() if q_type == 'rh': all_opt = get_optional_holiday_dates(ly_start, ly_end, employee=current_user) # Exclude in-flight RH dates in_flight = set() in_flight_statuses = ['pending', 'manager_approved', 'final_approved'] existing = LeaveRequest.query.filter( LeaveRequest.employee_id == current_user.id, LeaveRequest.leave_type.has(code='RH'), LeaveRequest.start_date >= ly_start, LeaveRequest.end_date <= ly_end, LeaveRequest.status.in_(in_flight_statuses) ).all() for r in existing: dt = r.start_date while dt <= r.end_date: in_flight.add(dt) dt += timedelta(days=1) allowed = sorted(d for d in all_opt - in_flight) # Build {date, name} items = [] for d in allowed: h = get_applicable_holiday_for(current_user, d) # uses location-aware fallback items.append({ "date": d.strftime("%Y-%m-%d"), "name": (h.name if h else "Optional Holiday") }) return {"dates": items} return {"dates": []} # ----------------------------- # BULK ACTIONS: MANAGER + HR # ----------------------------- from sqlalchemy import and_ def _ensure_manager_scope(qs, manager_id: int): """Limit queryset to the manager's team.""" team_ids = [e.id for e in Employee.query.filter_by(reporting_manager_id=manager_id).all()] return [r for r in qs if r.employee_id in team_ids] @leave_bp.post('/manager/bulk-action') @login_required def manager_bulk_action(): """Approve/Reject multiple PENDING requests belonging to current manager's team.""" action = request.form.get('action') # 'approve' | 'reject' ids = request.form.getlist('req_ids') # list of request IDs (strings) remark = (request.form.get('remark') or '').strip() if action not in ('approve', 'reject') or not ids: flash('❌ Select at least one request and an action.', 'danger') return redirect(url_for('leave.manager_pending')) # Fetch requests and scope to manager's team reqs = LeaveRequest.query.filter( LeaveRequest.id.in_(ids), LeaveRequest.status == 'pending' ).all() reqs = _ensure_manager_scope(reqs, current_user.id) approved, rejected, skipped = 0, 0, 0 for r in reqs: if action == 'approve': # Already processed by someone? if r.status != 'pending': skipped += 1 continue r.status = 'manager_approved' db.session.add(LeaveApprovalLog( leave_request_id=r.id, approver_id=current_user.id, approver_role='manager', status='approved', remark=remark )) approved += 1 # Notify HR (batching later is OK; keep simple) hr_list = _hr_ids() if hr_list: emp = Employee.query.get(r.employee_id) lt = LeaveType.query.get(r.leave_type_id) rng = _format_range(r.start_date, r.end_date) notify_many( employee_ids=hr_list, title="Leave request awaiting HR action", message=f"{emp.full_name}'s {lt.name} ({rng}) needs HR approval.", link=_link('leave.hr_pending') ) # Notify employee lt = LeaveType.query.get(r.leave_type_id) notify_employee( employee_id=r.employee_id, title="Leave approved by Manager", message=f"Your {lt.name} request ({_format_range(r.start_date, r.end_date)}) was approved by your manager and sent to HR.", link=_link('leave.view_history') ) _send_leave_inform_emails(r) else: if r.status != 'pending': skipped += 1 continue r.status = 'rejected' db.session.add(LeaveApprovalLog( leave_request_id=r.id, approver_id=current_user.id, approver_role='manager', status='rejected', remark=remark )) rejected += 1 # Notify employee lt = LeaveType.query.get(r.leave_type_id) notify_employee( employee_id=r.employee_id, title="Leave rejected by Manager", message=f"Your {lt.name} request ({_format_range(r.start_date, r.end_date)}) was rejected by your manager.", link=_link('leave.view_history') ) db.session.commit() flash(f"✅ Manager bulk: approved={approved}, rejected={rejected}, skipped={skipped}.", "success") return redirect(url_for('leave.manager_pending')) @leave_bp.post('/hr/bulk-action') @login_required def hr_bulk_action(): """Approve/Reject multiple requests pending for HR (status='manager_approved').""" action = request.form.get('action') # 'approve' | 'reject' ids = request.form.getlist('req_ids') remark = (request.form.get('remark') or '').strip() # Simple HR guard if not _hr_ids() or current_user.id not in _hr_ids(): # If you have a better is_hr() function, use it. # Or replace with your existing is_hr() helper if available. pass # Optional: enforce role if action not in ('approve', 'reject') or not ids: flash('❌ Select at least one request and an action.', 'danger') return redirect(url_for('leave.hr_pending')) reqs = LeaveRequest.query.filter( LeaveRequest.id.in_(ids), LeaveRequest.status == 'manager_approved' ).all() approved, rejected, skipped, balance_errors = 0, 0, 0, 0 # Common fy_start, fy_end = get_leave_year_range() fy_label = fy_start.year for r in reqs: # Re-fetch leave type lt = LeaveType.query.get(r.leave_type_id) if not lt: skipped += 1 continue if action == 'approve': # Balance deduction (non-RH, non-LWP) if lt.code not in ('RH', 'LWP'): bal = LeaveBalance.query.filter_by( employee_id=r.employee_id, leave_type_id=r.leave_type_id, year=fy_label ).first() if not bal: bal = LeaveBalance( employee_id=r.employee_id, leave_type_id=r.leave_type_id, year=fy_label, balance=0 ) db.session.add(bal) db.session.flush() need = 0.5 if r.half_day or (r.total_days == 0.5) else (r.total_days or 0) if (bal.balance or 0) < need: # Not enough balance -> skip this item balance_errors += 1 continue bal.balance = (bal.balance or 0) - need # Finalise approval r.status = 'final_approved' db.session.add(LeaveApprovalLog( leave_request_id=r.id, approver_id=current_user.id, approver_role='hr', status='approved', remark=remark )) approved += 1 # Attendance sync try: sync_leave_to_attendance( db, r.employee_id, r.start_date, r.end_date, lt, half_day=bool(r.half_day or (r.total_days == 0.5)), half_day_session=getattr(r, "half_day_session", "FH"), ) except Exception: # Don’t fail the whole batch on sync error; you can add logging if needed pass # Notify employee + manager FYI emp = Employee.query.get(r.employee_id) rng = _format_range(r.start_date, r.end_date) notify_employee( employee_id=r.employee_id, title="Leave fully approved", message=f"Your {lt.name} request ({rng}) has been approved by HR.", link=_link('leave.view_history') ) if emp and emp.reporting_manager_id: notify_employee( employee_id=emp.reporting_manager_id, title="Leave approved (HR)", message=f"{emp.full_name}'s {lt.name} ({rng}) is fully approved.", link=_link('leave.manager_pending') ) else: # reject r.status = 'rejected' db.session.add(LeaveApprovalLog( leave_request_id=r.id, approver_id=current_user.id, approver_role='hr', status='rejected', remark=remark )) rejected += 1 # Notify employee + manager FYI emp = Employee.query.get(r.employee_id) rng = _format_range(r.start_date, r.end_date) notify_employee( employee_id=r.employee_id, title="Leave rejected by HR", message=f"Your {lt.name} request ({rng}) was rejected by HR.", link=_link('leave.view_history') ) if emp and emp.reporting_manager_id: notify_employee( employee_id=emp.reporting_manager_id, title="Leave rejected (HR)", message=f"{emp.full_name}'s {lt.name} ({rng}) was rejected by HR.", link=_link('leave.manager_pending') ) db.session.commit() msg = f"✅ HR bulk: approved={approved}, rejected={rejected}, skipped={skipped}" if balance_errors: msg += f", balance_insufficient={balance_errors}" flash(msg + ".", "success") return redirect(url_for('leave.hr_pending')) # ----------------------------- # LEAVE BALANCE REPORT + EXPORT # ----------------------------- from io import StringIO, BytesIO import csv def _fy_range_for_report(): """Use the leave-year window (Apr–Mar) you already compute elsewhere.""" start, end = get_leave_year_range() # Your LeaveBalance.year stores FY start year (as in your allocation screen) fy_label = start.year return start, end, fy_label def _sum_used_days(employee_id: int, leave_type_id: int, d1: date, d2: date) -> float: """Sum approved (final_approved) leave days in the FY for a type.""" q = (LeaveRequest.query .filter( LeaveRequest.employee_id == employee_id, LeaveRequest.leave_type_id == leave_type_id, LeaveRequest.status == 'final_approved', LeaveRequest.start_date >= d1, LeaveRequest.end_date <= d2 ) ) used = 0.0 for lr in q: if lr.total_days and lr.total_days > 0: used += float(lr.total_days) else: # Fallback if total_days was not set span = (lr.end_date - lr.start_date).days + 1 used += 0.5 if lr.half_day else span return round(used, 2) @leave_bp.get('/balance-report') @login_required def balance_report(): # Filters dept_id = request.args.get('department_id', type=int) emp_id = request.args.get('employee_id', type=int) lt_id = request.args.get('leave_type_id', type=int) fy_start, fy_end, fy_label = _fy_range_for_report() # Base query of balances for FY qb = (db.session.query(LeaveBalance, Employee, LeaveType) .join(Employee, LeaveBalance.employee_id == Employee.id) .join(LeaveType, LeaveBalance.leave_type_id == LeaveType.id) .filter(LeaveBalance.year == fy_label)) if dept_id: qb = qb.join(Department, Department.id == Employee.department_id).filter(Department.id == dept_id) if emp_id: qb = qb.filter(LeaveBalance.employee_id == emp_id) if lt_id: qb = qb.filter(LeaveBalance.leave_type_id == lt_id) rows = qb.order_by(Employee.full_name, LeaveType.name).all() # Build report rows report = [] totals = {"credited": 0.0, "used": 0.0, "remaining": 0.0} for bal, emp, lt in rows: remaining = float(bal.balance or 0.0) used = _sum_used_days(emp.id, lt.id, fy_start, fy_end) credited = round(remaining + used, 2) # since remaining = credited - used in your flow report.append({ "emp_code": emp.employee_code, "employee": emp.full_name, "department": getattr(emp.department, "name", "-") if hasattr(emp, "department") else "-", "leave_type": lt.name, "leave_code": lt.code, "credited": credited, "used": used, "remaining": remaining, }) totals["credited"] += credited totals["used"] += used totals["remaining"] += remaining # dropdown data departments = Department.query.order_by(Department.name).all() employees = Employee.query.order_by(Employee.full_name).all() leave_types = LeaveType.query.order_by(LeaveType.name).all() return render_template( 'leave/balance_report.html', report=report, totals=totals, fy_label=fy_label, fy_start=fy_start, fy_end=fy_end, departments=departments, employees=employees, leave_types=leave_types, dept_id=dept_id, emp_id=emp_id, lt_id=lt_id ) # ----------------------------- # MONTH-WISE LEAVE REPORT (HR/Admin) # ----------------------------- from calendar import monthrange from datetime import date, datetime, timedelta from sqlalchemy.orm import joinedload @leave_bp.get('/report/monthly') @login_required def monthly_leave_report(): # Guard: only HR/Admin (uses your helper from comp-off section) if not _is_hr_admin_user(): flash("Access denied.", "danger") return redirect(url_for("employee.dashboard")) # --- Filters --- month = (request.args.get("month") or datetime.today().strftime("%Y-%m")).strip() # YYYY-MM dept_id = request.args.get("department_id", type=int) emp_id = request.args.get("employee_id", type=int) lt_id = request.args.get("leave_type_id", type=int) status = (request.args.get("status") or "final_approved").strip() # default: final_approved q_name = (request.args.get("q") or "").strip().lower() # Month bounds try: y, m = map(int, month.split("-")) month_start = date(y, m, 1) month_end = date(y, m, monthrange(y, m)[1]) except Exception: flash("Invalid month format. Use YYYY-MM.", "danger") return redirect(url_for("leave.monthly_leave_report")) # We need leaves that overlap this month: # overlap if start_date <= month_end AND end_date >= month_start q = (LeaveRequest.query .options( joinedload(LeaveRequest.employee).joinedload(Employee.department), joinedload(LeaveRequest.leave_type), joinedload(LeaveRequest.approval_logs) ) .filter(LeaveRequest.start_date <= month_end, LeaveRequest.end_date >= month_start) .order_by(LeaveRequest.start_date.asc(), LeaveRequest.applied_on.desc()) ) # Status filter if status and status.lower() != "all": q = q.filter(LeaveRequest.status == status) # Employee filter if emp_id: q = q.filter(LeaveRequest.employee_id == emp_id) # Leave type filter if lt_id: q = q.filter(LeaveRequest.leave_type_id == lt_id) # Department filter if dept_id: q = (q.join(Employee, LeaveRequest.employee_id == Employee.id) .filter(Employee.department_id == dept_id)) # Name search if q_name: q = (q.join(Employee, LeaveRequest.employee_id == Employee.id) .filter(db.func.lower(Employee.full_name).like(f"%{q_name}%"))) reqs = q.all() # Helpers _UNPAID_CODES = {"LWP", "LOP", "UL"} def _overlap_days(lr: LeaveRequest) -> float: """Count days of this leave that fall inside the selected month. Handles half-day (0.5) if single-day + half_day. """ s = max(lr.start_date, month_start) e = min(lr.end_date, month_end) if e < s: return 0.0 # half-day leaves should be single-day in your apply logic if (lr.half_day or (lr.total_days == 0.5)) and lr.start_date == lr.end_date: # Only count if that day is inside this month window return 0.5 if (month_start <= lr.start_date <= month_end) else 0.0 # otherwise normal day span return float((e - s).days + 1) def _manager_remark(lr: LeaveRequest) -> str: for log in (lr.approval_logs or []): if (log.approver_role or "").lower() == "manager": return (log.remark or "").strip() return "" def _hr_remark(lr: LeaveRequest) -> str: for log in (lr.approval_logs or []): if (log.approver_role or "").lower() == "hr": return (log.remark or "").strip() return "" rows = [] totals = { "count_requests": 0, "days_total": 0.0, "days_paid": 0.0, "days_unpaid": 0.0, "days_rh": 0.0, "days_co": 0.0, } # Summary by employee summary_by_emp = {} # emp_id -> dict for lr in reqs: days = _overlap_days(lr) if days <= 0: continue totals["count_requests"] += 1 totals["days_total"] += days lt_code = ((lr.leave_type.code if lr.leave_type else "") or "").strip().upper() is_unpaid = lt_code in _UNPAID_CODES is_rh = lt_code == "RH" is_co = lt_code in {"CO", "COMP_OFF"} if is_unpaid: totals["days_unpaid"] += days else: totals["days_paid"] += days if is_rh: totals["days_rh"] += days if is_co: totals["days_co"] += days emp = lr.employee dept_name = getattr(getattr(emp, "department", None), "name", "-") if emp else "-" rows.append({ "id": lr.id, "emp_code": getattr(emp, "employee_code", "") if emp else "", "employee": getattr(emp, "full_name", "-") if emp else "-", "department": dept_name, "leave_type": (lr.leave_type.name if lr.leave_type else "-"), "leave_code": lt_code or "-", "start_date": lr.start_date, "end_date": lr.end_date, "days_in_month": round(days, 2), "status": lr.status, "reason": (lr.reason or ""), "half_day": bool(lr.half_day or (lr.total_days == 0.5)), "half_day_session": getattr(lr, "half_day_session", None), "paid_or_lwp": ("LWP" if is_unpaid else "Paid"), "manager_remark": _manager_remark(lr), "hr_remark": _hr_remark(lr), }) # Employee summary if emp: s = summary_by_emp.setdefault(emp.id, { "emp_code": emp.employee_code, "employee": emp.full_name, "department": dept_name, "paid_days": 0.0, "unpaid_days": 0.0, "rh_days": 0.0, "co_days": 0.0, "total": 0.0, }) s["total"] += days if is_unpaid: s["unpaid_days"] += days else: s["paid_days"] += days if is_rh: s["rh_days"] += days if is_co: s["co_days"] += days # Round totals for k in ["days_total", "days_paid", "days_unpaid", "days_rh", "days_co"]: totals[k] = round(float(totals[k] or 0.0), 2) emp_summary_list = sorted(summary_by_emp.values(), key=lambda x: (x["department"] or "", x["employee"] or "")) # Dropdowns departments = Department.query.order_by(Department.name).all() employees = Employee.query.order_by(Employee.full_name).all() leave_types = LeaveType.query.order_by(LeaveType.name).all() return render_template( "leave/monthly_leave_report.html", month=month, month_start=month_start, month_end=month_end, rows=rows, totals=totals, emp_summary=emp_summary_list, departments=departments, employees=employees, leave_types=leave_types, dept_id=dept_id, emp_id=emp_id, lt_id=lt_id, status=status, q_name=q_name ) @leave_bp.get('/report/monthly/export') @login_required def monthly_leave_report_export(): # Same guard if not _is_hr_admin_user(): flash("Access denied.", "danger") return redirect(url_for("employee.dashboard")) # Reuse the same filter logic by calling the report and building CSV quickly # (Simple approach: duplicate minimal parsing + query) month = (request.args.get("month") or datetime.today().strftime("%Y-%m")).strip() dept_id = request.args.get("department_id", type=int) emp_id = request.args.get("employee_id", type=int) lt_id = request.args.get("leave_type_id", type=int) status = (request.args.get("status") or "final_approved").strip() q_name = (request.args.get("q") or "").strip().lower() try: y, m = map(int, month.split("-")) month_start = date(y, m, 1) month_end = date(y, m, monthrange(y, m)[1]) except Exception: flash("Invalid month format. Use YYYY-MM.", "danger") return redirect(url_for("leave.monthly_leave_report")) q = (LeaveRequest.query .options(joinedload(LeaveRequest.employee).joinedload(Employee.department), joinedload(LeaveRequest.leave_type)) .filter(LeaveRequest.start_date <= month_end, LeaveRequest.end_date >= month_start) .order_by(LeaveRequest.start_date.asc()) ) if status and status.lower() != "all": q = q.filter(LeaveRequest.status == status) if emp_id: q = q.filter(LeaveRequest.employee_id == emp_id) if lt_id: q = q.filter(LeaveRequest.leave_type_id == lt_id) if dept_id: q = q.join(Employee, LeaveRequest.employee_id == Employee.id).filter(Employee.department_id == dept_id) if q_name: q = q.join(Employee, LeaveRequest.employee_id == Employee.id).filter(db.func.lower(Employee.full_name).like(f"%{q_name}%")) reqs = q.all() def overlap_days(lr): s = max(lr.start_date, month_start) e = min(lr.end_date, month_end) if e < s: return 0.0 if (lr.half_day or (lr.total_days == 0.5)) and lr.start_date == lr.end_date: return 0.5 if (month_start <= lr.start_date <= month_end) else 0.0 return float((e - s).days + 1) # CSV import csv from io import StringIO from flask import Response si = StringIO() w = csv.writer(si) w.writerow([ "Emp Code", "Employee", "Department", "Leave Code", "Leave Type", "Start Date", "End Date", "Days In Month", "Status", "Reason" ]) for lr in reqs: days = overlap_days(lr) if days <= 0: continue emp = lr.employee dept_name = getattr(getattr(emp, "department", None), "name", "-") if emp else "-" lt_code = ((lr.leave_type.code if lr.leave_type else "") or "").strip().upper() w.writerow([ getattr(emp, "employee_code", "") if emp else "", getattr(emp, "full_name", "-") if emp else "-", dept_name, lt_code, (lr.leave_type.name if lr.leave_type else "-"), lr.start_date.isoformat() if lr.start_date else "", lr.end_date.isoformat() if lr.end_date else "", round(days, 2), lr.status, (lr.reason or "").replace("\n", " ").strip() ]) out = si.getvalue() filename = f"monthly_leave_report_{month}.csv" return Response(out, mimetype="text/csv", headers={"Content-Disposition": f"attachment; filename={filename}"}) @leave_bp.post('/allocation/preview') @login_required def leave_allocation_preview(): form = request.form leave_type_id = int(form['leave_type_id']) strategy = form['strategy'] leave_type = LeaveType.query.get_or_404(leave_type_id) today = date.today() fy_start, fy_end = fy_bounds_for_today(today) fy_label = fy_start.year # Freeze cutoff freeze_till = (form.get('freeze_till') or '').strip() if freeze_till: f_year, f_month = map(int, freeze_till.split('-')) else: f_year, f_month = fy_label, 8 if f_month == 12: since_dt = datetime(f_year + 1, 1, 1) else: since_dt = datetime(f_year, f_month + 1, 1) employees = Employee.query.filter(func.lower(Employee.status) == 'active').all() rows = [] for emp in employees: bal = LeaveBalance.query.filter_by(employee_id=emp.id, leave_type_id=leave_type.id, year=fy_label).first() prev = float(bal.balance if bal else 0.0) doj = getattr(emp, 'date_of_joining', None) if doj and doj > fy_end: rows.append({ "emp_code": emp.employee_code, "employee": emp.full_name, "previous": prev, "new": prev, "delta": 0.0, "remark": "Joined after FY, skipped" }) continue if strategy == 'monthly_credit': months = months_after_freeze_until(today, f_year, f_month) monthly = round((leave_type.default_quota or 0) / 12.0, 2) target_total = round(min(len(months) * monthly, (leave_type.default_quota or 0)), 2) remark = f"{len(months)} × {monthly} post-freeze" elif strategy == 'chl_earned_leave': target_total = calc_chl_earned_leave_total_after_freeze(doj, today, f_year, f_month) remark = "CHL EL rule (post-freeze)" elif strategy == 'annual': months = months_after_freeze_until(today, f_year, f_month) target_total = float(leave_type.default_quota or 0) if months else 0.0 remark = "Annual post-freeze" else: target_total = 0.0 remark = "Unknown strategy" already_since = _sum_credited_since(emp.id, leave_type.id, since_dt) delta = max(0.0, round(target_total - already_since, 2)) rows.append({ "emp_code": emp.employee_code, "employee": emp.full_name, "previous": prev, "new": round(prev + delta, 2), "delta": delta, "remark": f"{remark}; freeze_till={f_year}-{f_month:02d}, already_post_freeze={already_since}, target_post_freeze={target_total}" }) return render_template( 'leave/leave_allocation_preview.html', leave_type=leave_type, fy_label=fy_label, strategy=strategy, freeze_till=f"{f_year:04d}-{f_month:02d}", rows=rows ) @leave_bp.post('/allocation/revoke/') @login_required def revoke_allocation_batch(batch_id:int): batch = LeaveAllocationBatch.query.get_or_404(batch_id) if batch.is_revoked: flash('⚠️ This batch is already revoked.', 'warning') return redirect(url_for('leave.leave_allocation')) adjusted = 0 if (batch.strategy or '').lower() == 'manual_set_balance': # Use the previous_balance restoration path restored, created = _revoke_set_balance_batch(batch) adjusted = restored note_tail = f" (manual_set_balance: restored={restored}, created_missing_rows={created})" else: # Your original delta-based revoke flow entries = LeaveAllocationEntry.query.filter_by(batch_id=batch.id).all() for ent in entries: if not ent.credit_delta or ent.credit_delta <= 0: continue bal = (LeaveBalance.query .filter_by(employee_id=ent.employee_id, leave_type_id=ent.leave_type_id, year=ent.fy_label) .with_for_update() .first()) if not bal: continue bal.balance = round(max(0.0, (bal.balance or 0.0) - ent.credit_delta), 2) adjusted += 1 note_tail = "" batch.is_revoked = True batch.revoked_at = datetime.utcnow() batch.revoked_by_id = getattr(current_user, 'id', None) db.session.commit() flash(f"🧹 Reverted batch {batch.batch_key}. Adjusted {adjusted} balances.{note_tail}", "success") return redirect(url_for('leave.leave_allocation')) # --- Bulk correction via CSV (desired TOTAL credit-to-date) --- # POST /leave/allocation/correct (multipart/form-data with file=csv) @leave_bp.post('/allocation/correct') @login_required def leave_allocation_correct(): from werkzeug.utils import secure_filename import csv, io file = request.files.get('file') if not file: flash('Upload a CSV with: emp_code, lt_code, fy_label, desired_total_credit', 'danger') return redirect(url_for('leave.leave_allocation')) # Read CSV into memory data = file.read().decode('utf-8', errors='ignore') rows = list(csv.DictReader(io.StringIO(data))) required = {'emp_code','lt_code','fy_label','desired_total_credit'} if not required.issubset({c.strip() for c in rows[0].keys()}): flash('CSV must have columns: emp_code, lt_code, fy_label, desired_total_credit', 'danger') return redirect(url_for('leave.leave_allocation')) # Build a correction batch batch_key = f"CORRECT_{datetime.utcnow().strftime('%Y%m%d_%H%M%S')}" batch = LeaveAllocationBatch( batch_key=batch_key, leave_type_id=None, # not a single type; per-row below fy_label=None, # per-row below strategy='manual_correction', run_by_id=getattr(current_user, 'id', None), note='CSV corrections to total credit-to-date' ) db.session.add(batch) db.session.flush() # Maps for lookups lt_by_code = {lt.code.upper(): lt for lt in LeaveType.query.all()} emp_by_code = {e.employee_code: e for e in Employee.query.all()} adjusted, skipped = 0, 0 for r in rows: emp_code = r['emp_code'].strip() lt_code = r['lt_code'].strip().upper() try: fy_label = int(r['fy_label']) desired_total = float(r['desired_total_credit']) except Exception: skipped += 1 continue emp = emp_by_code.get(emp_code) lt = lt_by_code.get(lt_code) if not emp or not lt: skipped += 1 continue # How much credit has already been granted (via batches) this FY? since_dt = datetime(fy_label, 4, 1) already = _sum_credited_since(emp.id, lt.id, since_dt) delta = round(desired_total - already, 2) # can be negative # Get current balance row bal = (LeaveBalance.query .filter_by(employee_id=emp.id, leave_type_id=lt.id, year=fy_label) .with_for_update() .first()) if not bal: bal = LeaveBalance(employee_id=emp.id, leave_type_id=lt.id, year=fy_label, balance=0.0) db.session.add(bal) db.session.flush() prev = float(bal.balance or 0.0) new_bal = round(max(0.0, prev + delta), 2) # Audit entry with delta (can be negative) db.session.add(LeaveAllocationEntry( batch_id=batch.id, employee_id=emp.id, leave_type_id=lt.id, fy_label=fy_label, previous_balance=prev, new_balance=new_bal, credit_delta=delta )) # Apply delta (preserve usage) bal.balance = new_bal adjusted += 1 db.session.commit() flash(f'✅ Correction batch {batch.batch_key} applied: adjusted={adjusted}, skipped={skipped}', 'success') return redirect(url_for('leave.leave_allocation')) # ============================= # Allocation Batches: list, detail, export # ============================= from flask import Response from sqlalchemy import desc @leave_bp.get('/allocation/batches') @login_required def allocation_batches(): """ Filters via query params (all optional): status=active|revoked|all (default: active) leave_type_id= fy_label= strategy= q= page= (default 1) per_page= (default 20) """ status = (request.args.get('status') or 'active').lower() leave_type_id = request.args.get('leave_type_id', type=int) fy_label = request.args.get('fy_label', type=int) strategy = (request.args.get('strategy') or '').strip() q = (request.args.get('q') or '').strip() page = request.args.get('page', default=1, type=int) per_page = request.args.get('per_page', default=20, type=int) qb = LeaveAllocationBatch.query if status in ('active', 'revoked'): qb = qb.filter(LeaveAllocationBatch.is_revoked == (status == 'revoked')) if leave_type_id: qb = qb.filter(LeaveAllocationBatch.leave_type_id == leave_type_id) if fy_label: qb = qb.filter(LeaveAllocationBatch.fy_label == fy_label) if strategy: qb = qb.filter(LeaveAllocationBatch.strategy == strategy) if q: qb = qb.filter(LeaveAllocationBatch.batch_key.ilike(f"%{q}%")) qb = qb.order_by(desc(LeaveAllocationBatch.run_at)) # Pagination pagination = qb.paginate(page=page, per_page=per_page, error_out=False) batches = pagination.items # Pre-aggregate totals for each batch (sum of credit_delta and count of entries) # Avoid n+1: pull ids and do group sums batch_ids = [b.id for b in batches] totals_by_batch = {} if batch_ids: sums = (db.session.query( LeaveAllocationEntry.batch_id, db.func.coalesce(db.func.sum(LeaveAllocationEntry.credit_delta), 0.0), db.func.count(LeaveAllocationEntry.id) ) .filter(LeaveAllocationEntry.batch_id.in_(batch_ids)) .group_by(LeaveAllocationEntry.batch_id) .all()) totals_by_batch = {bid: {"sum_delta": float(s), "count": int(c)} for (bid, s, c) in sums} # Dropdown data leave_types = LeaveType.query.order_by(LeaveType.name).all() strategies = ['monthly_credit', 'chl_earned_leave', 'annual', 'manual_correction', 'manual_set_balance', 'manual_adjust_one'] return render_template( 'leave/allocation_batches.html', batches=batches, totals_by_batch=totals_by_batch, leave_types=leave_types, strategies=strategies, status=status, leave_type_id=leave_type_id, fy_label=fy_label, strategy=strategy, q=q, pagination=pagination ) @leave_bp.get('/allocation/batches/') @login_required def allocation_batch_detail(batch_id: int): """ Show all entries for a batch with employee details. """ batch = LeaveAllocationBatch.query.get_or_404(batch_id) # Join entries with employees & leave types for display entries = (db.session.query( LeaveAllocationEntry, Employee.employee_code, Employee.full_name, LeaveType.code.label('lt_code'), LeaveType.name.label('lt_name') ) .join(Employee, Employee.id == LeaveAllocationEntry.employee_id) .join(LeaveType, LeaveType.id == LeaveAllocationEntry.leave_type_id) .filter(LeaveAllocationEntry.batch_id == batch.id) .order_by(Employee.full_name.asc()) .all()) # Totals total_delta = sum(float(e.LeaveAllocationEntry.credit_delta) for e in entries) if entries else 0.0 count = len(entries) return render_template( 'leave/allocation_batch_detail.html', batch=batch, entries=entries, total_delta=round(total_delta, 2), count=count ) @leave_bp.get('/allocation/batches//export') @login_required def allocation_batch_export(batch_id: int): """ Export a batch to CSV: emp_code, employee, leave_type, fy_label, prev, delta, new """ batch = LeaveAllocationBatch.query.get_or_404(batch_id) rows = (db.session.query( Employee.employee_code, Employee.full_name, LeaveType.code.label('lt_code'), LeaveAllocationEntry.fy_label, LeaveAllocationEntry.previous_balance, LeaveAllocationEntry.credit_delta, LeaveAllocationEntry.new_balance ) .join(LeaveAllocationEntry, LeaveAllocationEntry.employee_id == Employee.id) .join(LeaveType, LeaveType.id == LeaveAllocationEntry.leave_type_id) .filter(LeaveAllocationEntry.batch_id == batch.id) .order_by(Employee.full_name.asc()) .all()) def _gen(): yield "emp_code,employee,leave_type,fy_label,previous,delta,new\n" for r in rows: yield f"{r.employee_code},{r.full_name},{r.lt_code},{r.fy_label},{r.previous_balance},{r.credit_delta},{r.new_balance}\n" filename = f"allocation_batch_{batch.batch_key}.csv" return Response(_gen(), mimetype='text/csv', headers={"Content-Disposition": f"attachment; filename={filename}"}) # ============================ # MANUAL "SET BALANCE (AS-OF TODAY)" – overrides current balance # ============================ from io import StringIO import csv @leave_bp.get('/balance/set') @login_required def balance_set_form(): """ UI to set balances: - Option A: choose a leave type and a single value -> apply to all active employees - Option B: upload CSV with columns: emp_code, lt_code, fy_label, set_balance """ leave_types = LeaveType.query.order_by(LeaveType.name).all() # Default FY label = Apr–Mar containing today _, _, fy_label = _fy_range_for_report() return render_template( 'leave/balance_set.html', leave_types=leave_types, fy_label=fy_label ) @leave_bp.post('/balance/set/apply') @login_required def balance_set_apply(): """ Apply a single value for one leave type to all active employees (override). form: leave_type_id, fy_label, set_value, include_notice, include_resigned """ try: leave_type_id = int(request.form['leave_type_id']) fy_label = int(request.form['fy_label']) set_value = float(request.form['set_value']) except Exception: flash("Invalid inputs. Please verify leave type, FY, and value.", "danger") return redirect(url_for('leave.balance_set_form')) include_notice = ('include_notice' in request.form) include_resigned = ('include_resigned' in request.form) # Employee scope statuses = ['active'] if include_notice: statuses.append('notice_period') if include_resigned: statuses.append('resigned') employees = (Employee.query .filter(func.lower(Employee.status).in_([s.lower() for s in statuses])) .order_by(Employee.full_name).all()) if not employees: flash("No employees matched the selected statuses.", "warning") return redirect(url_for('leave.balance_set_form')) lt = LeaveType.query.get_or_404(leave_type_id) # Create an audit batch – strategy ensures we don't count this as a credit batch = LeaveAllocationBatch( batch_key=f"SETBAL_{datetime.utcnow().strftime('%Y%m%d_%H%M%S')}", leave_type_id=leave_type_id, fy_label=fy_label, strategy='manual_set_balance', run_by_id=getattr(current_user, 'id', None), note='Manual set balance (override) for all employees' ) db.session.add(batch) db.session.flush() adjusted = 0 for emp in employees: bal = (LeaveBalance.query .filter_by(employee_id=emp.id, leave_type_id=leave_type_id, year=fy_label) .with_for_update(read=True) .first()) if not bal: bal = LeaveBalance(employee_id=emp.id, leave_type_id=leave_type_id, year=fy_label, balance=0.0) db.session.add(bal) db.session.flush() prev = float(bal.balance or 0.0) new_bal = round(max(0.0, set_value), 2) # guard against negative # Audit entry – credit_delta=0 so future allocations don't treat this as "credited" db.session.add(LeaveAllocationEntry( batch_id=batch.id, employee_id=emp.id, leave_type_id=leave_type_id, fy_label=fy_label, previous_balance=prev, new_balance=new_bal, credit_delta=0.0 )) bal.balance = new_bal adjusted += 1 db.session.commit() flash(f"✅ Set balance to {set_value} for {adjusted} employee(s) [{lt.code}, FY {fy_label}].", "success") return redirect(url_for('leave.allocation_batches')) @leave_bp.post('/balance/set/csv') @login_required def balance_set_csv(): """ CSV override (per-employee): Columns: emp_code, lt_code, fy_label, set_balance Behaviour: set LeaveBalance.balance = set_balance (absolute), audit with credit_delta=0 Assumption: one leave type + one FY per CSV file. (Rows with different lt_code/fy_label will be skipped.) """ from werkzeug.utils import secure_filename import csv, io, codecs from sqlalchemy import func file = request.files.get('file') if not file: flash('Upload a CSV with: emp_code, lt_code, fy_label, set_balance', 'danger') return redirect(url_for('leave.balance_set_form')) # Read raw bytes raw = file.read() # Try to detect UTF-16 (common when saving CSV from Excel on Windows) if raw.startswith(codecs.BOM_UTF16_LE) or raw.startswith(codecs.BOM_UTF16_BE): try: data = raw.decode('utf-16') except Exception: flash('Could not decode file as UTF-16 CSV. Please save as UTF-8 CSV and re-upload.', 'danger') return redirect(url_for('leave.balance_set_form')) else: # Fallback: assume UTF-8 try: data = raw.decode('utf-8', errors='ignore') except Exception: flash('Could not decode file. Please upload a standard CSV (UTF-8).', 'danger') return redirect(url_for('leave.balance_set_form')) # Remove any NUL characters that break the csv module if '\x00' in data: data = data.replace('\x00', '') # Now parse CSV safely try: rows = list(csv.DictReader(io.StringIO(data))) except Exception as e: flash(f'Error reading CSV: {e}. Please check the file format.', 'danger') return redirect(url_for('leave.balance_set_form')) required = {'emp_code', 'lt_code', 'fy_label', 'set_balance'} if not rows or not required.issubset({(c or '').strip() for c in rows[0].keys()}): flash('CSV must have columns: emp_code, lt_code, fy_label, set_balance', 'danger') return redirect(url_for('leave.balance_set_form')) # ---- Determine batch-level leave_type + FY from first row ---- first = rows[0] first_lt_code = (first.get('lt_code') or '').strip().upper() try: first_fy_label = int(first.get('fy_label')) except Exception: flash("Invalid fy_label in first row of CSV.", "danger") return redirect(url_for('leave.balance_set_form')) # Lookups lt_by_code = {lt.code.upper(): lt for lt in LeaveType.query.all()} emp_by_code = {e.employee_code: e for e in Employee.query.all()} leave_type = lt_by_code.get(first_lt_code) if not leave_type: flash(f"Leave type with code '{first_lt_code}' not found. Please check the CSV.", "danger") return redirect(url_for('leave.balance_set_form')) # Create one audit batch for this file – now with NON-NULL leave_type_id & fy_label batch = LeaveAllocationBatch( batch_key=f"SETBALCSV_{datetime.utcnow().strftime('%Y%m%d_%H%M%S')}", leave_type_id=leave_type.id, fy_label=first_fy_label, strategy='manual_set_balance', run_by_id=getattr(current_user, 'id', None), note='CSV manual set balance (override)' ) db.session.add(batch) db.session.flush() adjusted = skipped = mismatched = 0 for r in rows: emp_code = (r.get('emp_code') or '').strip() lt_code = (r.get('lt_code') or '').strip().upper() try: fy_label = int(r.get('fy_label')) set_value = float(r.get('set_balance')) except Exception: skipped += 1 continue # Enforce single-type & single-FY assumption for this batch if lt_code != first_lt_code or fy_label != first_fy_label: mismatched += 1 continue emp = emp_by_code.get(emp_code) lt = lt_by_code.get(lt_code) if not emp or not lt: skipped += 1 continue bal = (LeaveBalance.query .filter_by(employee_id=emp.id, leave_type_id=lt.id, year=fy_label) .with_for_update(read=True) .first()) if not bal: bal = LeaveBalance(employee_id=emp.id, leave_type_id=lt.id, year=fy_label, balance=0.0) db.session.add(bal) db.session.flush() prev = float(bal.balance or 0.0) new_bal = round(max(0.0, set_value), 2) # Audit entry – credit_delta=0 so future allocations add on top db.session.add(LeaveAllocationEntry( batch_id=batch.id, employee_id=emp.id, leave_type_id=lt.id, fy_label=fy_label, previous_balance=prev, new_balance=new_bal, credit_delta=0.0 )) bal.balance = new_bal adjusted += 1 db.session.commit() msg = f'✅ Manual set via CSV done: adjusted={adjusted}, skipped={skipped}' if mismatched: msg += f', mismatched_lt_or_fy={mismatched}' flash(msg, 'success') return redirect(url_for('leave.allocation_batches')) from sqlalchemy import and_ def _revoke_set_balance_batch(batch): """ For strategy='manual_set_balance' batches: Restore LeaveBalance.balance to the Entry.previous_balance for every entry. Returns (restored_count, created_missing_rows). """ entries = LeaveAllocationEntry.query.filter_by(batch_id=batch.id).all() restored = 0 created = 0 for ent in entries: # Fetch (or create) the balance row for this (emp, lt, FY) bal = (LeaveBalance.query .filter_by(employee_id=ent.employee_id, leave_type_id=ent.leave_type_id, year=ent.fy_label) .with_for_update() .first()) if not bal: bal = LeaveBalance( employee_id=ent.employee_id, leave_type_id=ent.leave_type_id, year=ent.fy_label, balance=0.0 ) db.session.add(bal) db.session.flush() created += 1 # Restore to the exact previous balance we captured when setting bal.balance = float(ent.previous_balance or 0.0) restored += 1 return restored, created @leave_bp.post('/allocation/revoke-set/') @login_required def revoke_set_balance_batch(batch_id: int): """ Revoke a 'manual_set_balance' batch by restoring the balances to the per-entry previous_balance. """ batch = LeaveAllocationBatch.query.get_or_404(batch_id) if batch.is_revoked: flash('⚠️ This batch is already revoked.', 'warning') return redirect(url_for('leave.allocation_batches')) if (batch.strategy or '').lower() != 'manual_set_balance': flash('❌ This revoke endpoint is only for "manual_set_balance" batches.', 'danger') return redirect(url_for('leave.allocation_batches')) restored, created = _revoke_set_balance_batch(batch) batch.is_revoked = True batch.revoked_at = datetime.utcnow() batch.revoked_by_id = getattr(current_user, 'id', None) db.session.commit() flash(f"🧹 Reverted manual set-balance batch {batch.batch_key}. " f"Restored={restored}, created_missing_rows={created}.", "success") return redirect(url_for('leave.allocation_batches')) # put this near your other Set Balance routes from flask import Response from io import StringIO import csv @leave_bp.get('/balance/set/template') @login_required def balance_set_template(): """ Downloadable CSV template for manual set balance. Columns: emp_code, lt_code, fy_label, set_balance Includes a few sample rows for convenience. """ # FY label (Apr–Mar) using your helper _, _, fy_label = _fy_range_for_report() si = StringIO() w = csv.writer(si) w.writerow(['emp_code', 'lt_code', 'fy_label', 'set_balance']) # Optional sample rows (first 5 employees) sample_emps = Employee.query.order_by(Employee.employee_code).limit(5).all() for e in sample_emps: w.writerow([e.employee_code or '', 'EL', fy_label, 0]) out = si.getvalue() filename = f"set_balance_template_{datetime.utcnow().strftime('%Y%m%d')}.csv" return Response(out, mimetype='text/csv', headers={"Content-Disposition": f"attachment; filename={filename}"}) from flask_mail import Message from flask import current_app def _employee_email(emp: Employee) -> Optional[str]: """Pick a reasonable email field from Employee.""" return ( getattr(emp, "email", None) or getattr(emp, "work_email", None) or getattr(emp, "personal_email", None) or getattr(emp, "email", None) ) def _get_mail_sender(): """ Resolve sender based on app.config: 1. Prefer MAIL_DEFAULT_SENDER (which may have been set from DB by apply_email_settings_from_db). - Can be a string or a (name, email) tuple. 2. If not present, fall back to MAIL_FROM or MAIL_USERNAME from config. 3. If still nothing, return None and let Flask-Mail complain (config is incomplete). """ sender = current_app.config.get("MAIL_DEFAULT_SENDER") # MAIL_DEFAULT_SENDER can be string or (name, email) if isinstance(sender, (list, tuple)) and len(sender) >= 2: return (sender[0], sender[1]) if isinstance(sender, str) and sender: return sender # Flask-Mail accepts a plain string email too # Fallbacks (from config.py), used only if no DB sender is set fallback = current_app.config.get("MAIL_FROM") or current_app.config.get("MAIL_USERNAME") if fallback: # Use same value for name+email by default return fallback # No sender at all → misconfigured environment return None def _send_leave_submission_email_to_manager(created_reqs, employee: Employee): """ Send a single email to the employee's reporting manager when leave is submitted. Handles split requests (e.g. EL + LWP) by summarising all parts in one mail. """ if not created_reqs or not employee: return mgr_id = getattr(employee, "reporting_manager_id", None) if not mgr_id: return mgr = Employee.query.get(mgr_id) if not mgr: return recipient = _employee_email(mgr) if not recipient: return sender = _get_mail_sender() if not sender: return # misconfigured mail sender; silently skip lines = [] lines.append(f"Dear {mgr.full_name},") lines.append("") lines.append(f"{employee.full_name} has submitted the following leave request(s):") lines.append("") for lr in created_reqs: lt = LeaveType.query.get(lr.leave_type_id) rng = _format_range(lr.start_date, lr.end_date) # Fallback days calculation if total_days is missing if lr.total_days: days = lr.total_days else: span = (lr.end_date - lr.start_date).days + 1 days = 0.5 if lr.half_day else span lines.append(f"- {lt.name if lt else 'Leave'}: {rng} ({days} day(s))") if getattr(employee, "employee_code", None): lines.append(f"Employee Code: {employee.employee_code}") # Use the first request's reason (all split parts share the same reason) first_reason = created_reqs[0].reason if created_reqs[0].reason else None if first_reason: lines.append(f"Reason: {first_reason}") lines.append("") lines.append(f"You can review and take action here: {_link('leave.manager_pending')}") lines.append("") lines.append("This is an automated notification from the HR system.") subject = f"[Leave Request] {employee.full_name} has applied for leave" msg = Message( subject=subject, recipients=[recipient], sender=sender, body="\n".join(lines), ) mail.send(msg) def _send_leave_inform_emails(req: LeaveRequest): informees = getattr(req, "informees", []) # will be [] if no relation or no rows if not informees: return emp = Employee.query.get(req.employee_id) if not emp: return lt = LeaveType.query.get(req.leave_type_id) mgr = Employee.query.get(emp.reporting_manager_id) if emp.reporting_manager_id else None rng = _format_range(req.start_date, req.end_date) subject = f"[Leave Info] {emp.full_name} on {lt.name} from {rng}" sender = _get_mail_sender() for inf in informees: inf_emp = inf.employee if not inf_emp: continue recipient = _employee_email(inf_emp) if not recipient: continue lines = [ f"Dear {inf_emp.full_name},", "", f"This is to inform you that {emp.full_name}" f"{f' ({emp.employee_code})' if getattr(emp, 'employee_code', None) else ''}" f" will be on {lt.name} from {rng}.", ] if req.reason: lines.append(f"Reason: {req.reason}") if mgr: lines.append(f"Approved by: {mgr.full_name}") lines += [ "", "This is an automated notification from the HR system." ] msg = Message( subject=subject, recipients=[recipient], sender=sender, # <-- uses DB if present, else config body="\n".join(lines), ) mail.send(msg)