from datetime import date, datetime, timedelta from calendar import monthrange from flask import Blueprint, render_template, request, redirect, url_for, flash, jsonify from flask_login import login_required, current_user from sqlalchemy import func from extensions import db # import appraisal models (adjust import path as needed) from models import ( AppraisalQuestion, AppraisalScoreRule, DeptAppraisalBudget, GlobalAppraisalBudget, LocationAppraisalBudget, Appraisal, AppraisalResponse, AppraisalAnswer, AppraisalAuditLog, fy_for, Employee, Department ) appraisal_bp = Blueprint("appraisal", __name__, url_prefix="/appraisal") # --------------------------------------------------------------------- # Access helpers # --------------------------------------------------------------------- def fy_date_range(fy_start_year: int): fy_start = date(fy_start_year, 4, 1) fy_end = date(fy_start_year + 1, 3, 31) return fy_start, fy_end def _role_name() -> str: r = getattr(current_user, "role", None) if r is not None and not isinstance(r, str): role_name = getattr(r, "name", None) or getattr(r, "role_name", None) or getattr(r, "title", None) or "" else: role_name = r or "" return str(role_name).strip().lower() def _is_hr_admin() -> bool: return _role_name() in ("admin", "hr", "hr_admin", "superadmin") def is_manager() -> bool: # simple: anyone with reportees return Employee.query.filter(Employee.reporting_manager_id == current_user.id).count() > 0 def _audit(appraisal_id, action, old=None, new=None): db.session.add(AppraisalAuditLog( appraisal_id=appraisal_id, action=action, old_data=old, new_data=new, created_by_id=getattr(current_user, "id", None) )) def first_day_of_next_month(d: date) -> date: if d.month == 12: return date(d.year + 1, 1, 1) return date(d.year, d.month + 1, 1) def month_bounds(any_date: date): start = date(any_date.year, any_date.month, 1) end = date(any_date.year, any_date.month, monthrange(any_date.year, any_date.month)[1]) return start, end def due_date_for_employee(emp: Employee) -> date: base = emp.date_of_joining if getattr(emp, "last_appraisal_effective_month", None): base = max(base, emp.last_appraisal_effective_month) # + 1 year (safe method: add 365 days; if you want exact year rules, use dateutil.relativedelta) return base + timedelta(days=365) def _get_emp_name(emp: Employee) -> str: return getattr(emp, "full_name", None) or getattr(emp, "name", None) or getattr(emp, "employee_name", None) or f"Emp#{emp.id}" def _get_emp_code(emp: Employee) -> str: return getattr(emp, "employee_code", None) or "" from models import EmployeeSalaryStructures def _current_gross(emp: Employee) -> float: total = (db.session.query(func.coalesce(func.sum(EmployeeSalaryStructures.amount), 0.0)) .filter(EmployeeSalaryStructures.employee_id == emp.id) .filter(func.lower(EmployeeSalaryStructures.type) == "earning") .scalar()) return float(total or 0.0) # --------------------------------------------------------------------- # Scoring helpers (1..5 only) # --------------------------------------------------------------------- def calc_score_pct(appraisal_id: int, responder_type: str) -> float: resp = AppraisalResponse.query.filter_by(appraisal_id=appraisal_id, responder_type=responder_type).first() if not resp: return 0.0 # Join answers with questions to get weights rows = (db.session.query(AppraisalAnswer, AppraisalQuestion) .join(AppraisalQuestion, AppraisalQuestion.id == AppraisalAnswer.question_id) .filter(AppraisalAnswer.response_id == resp.id) .all()) total = 0.0 max_total = 0.0 for ans, q in rows: w = float(q.weight or 1.0) r = int(ans.rating) if r < 1: r = 1 if r > 5: r = 5 total += r * w max_total += 5 * w if max_total <= 0: return 0.0 return round((total / max_total) * 100.0, 2) def recommended_increment_pct(score_pct: float) -> float: rule = (AppraisalScoreRule.query .filter(AppraisalScoreRule.is_active == True) .filter(AppraisalScoreRule.min_score_pct <= score_pct) .filter(AppraisalScoreRule.max_score_pct >= score_pct) .order_by(AppraisalScoreRule.min_score_pct.desc()) .first()) if not rule: return 0.0 return round(float(rule.recommended_increment_pct or 0.0), 2) # --------------------------------------------------------------------- # Budget helpers (annualized impact) # --------------------------------------------------------------------- def _norm_loc(loc: str) -> str: return (loc or "").strip() def _fy_used_global(fy_start_year: int) -> float: fy_start, fy_end = fy_date_range(fy_start_year) used = (db.session.query(func.coalesce(func.sum(Appraisal.annualized_impact), 0.0)) .filter(Appraisal.status.in_(["hr_finalized", "applied"])) .filter(Appraisal.effective_month >= fy_start) .filter(Appraisal.effective_month <= fy_end) .scalar()) return float(used or 0.0) def _fy_used_location(fy_start_year: int, location: str) -> float: fy_start, fy_end = fy_date_range(fy_start_year) loc = _norm_loc(location).lower() used = (db.session.query(func.coalesce(func.sum(Appraisal.annualized_impact), 0.0)) .filter(Appraisal.status.in_(["hr_finalized", "applied"])) .filter(Appraisal.effective_month >= fy_start) .filter(Appraisal.effective_month <= fy_end) .filter(func.lower(func.coalesce(Appraisal.location, "")) == loc) .scalar()) return float(used or 0.0) def _fy_used_dept(fy_start_year: int, location: str, department_id: int) -> float: fy_start, fy_end = fy_date_range(fy_start_year) loc = _norm_loc(location).lower() used = (db.session.query(func.coalesce(func.sum(Appraisal.annualized_impact), 0.0)) .filter(Appraisal.status.in_(["hr_finalized", "applied"])) .filter(Appraisal.effective_month >= fy_start) .filter(Appraisal.effective_month <= fy_end) .filter(Appraisal.department_id == department_id) .filter(func.lower(func.coalesce(Appraisal.location, "")) == loc) .scalar()) return float(used or 0.0) def fy_remaining_global_budget(fy_start_year: int) -> dict: row = GlobalAppraisalBudget.query.filter_by(fy_start_year=fy_start_year).first() total = float(row.budget_amount) if row else 0.0 used = _fy_used_global(fy_start_year) return {"budget_total": round(total, 2), "used": round(used, 2), "remaining": round(total - used, 2)} def fy_remaining_location_budget(fy_start_year: int, location: str) -> dict: loc = _norm_loc(location) row = LocationAppraisalBudget.query.filter_by(fy_start_year=fy_start_year, location=loc).first() total = float(row.budget_amount) if row else 0.0 used = _fy_used_location(fy_start_year, loc) return {"budget_total": round(total, 2), "used": round(used, 2), "remaining": round(total - used, 2)} def fy_remaining_dept_budget(fy_start_year: int, location: str, department_id: int) -> dict: loc = _norm_loc(location) row = (DeptAppraisalBudget.query .filter_by(fy_start_year=fy_start_year, location=loc, department_id=department_id) .first()) total = float(row.budget_amount) if row else 0.0 used = _fy_used_dept(fy_start_year, loc, department_id) return {"budget_total": round(total, 2), "used": round(used, 2), "remaining": round(total - used, 2)} def fy_of_effective_month(eff: date) -> int: return fy_for(eff) # --------------------------------------------------------------------- # Dashboard: due this month + HR initiate anytime # --------------------------------------------------------------------- @appraisal_bp.route("/dashboard", methods=["GET"]) @login_required def dashboard(): ym = (request.args.get("month") or "").strip() # 'YYYY-MM' action = (request.args.get("action") or "all").strip().lower() # all/action/completed if ym: try: y, m = [int(x) for x in ym.split("-")] ref = date(y, m, 1) except Exception: today = date.today() ref = date(today.year, today.month, 1) else: today = date.today() ref = date(today.year, today.month, 1) start, end = month_bounds(ref) is_hr_admin = _is_hr_admin() mgr = is_manager() # ---------------------------- # HR Due list (month based) # ---------------------------- due_rows = [] if is_hr_admin: employees = Employee.query.all() for emp in employees: doj = getattr(emp, "date_of_joining", None) if not doj: continue due = due_date_for_employee(emp) if start <= due <= end: due_rows.append({ "emp": emp, "due_date": due, "dept": getattr(emp, "department", None) }) # ---------------------------- # Build appraisal query (role aware) # FIX: For HR/Admin, when action filter = "action" or "completed", # we do NOT restrict to selected month, so pending items are visible. # ---------------------------- q = Appraisal.query if is_hr_admin: if action == "action": # pending / needs action across ALL months q = q.filter(Appraisal.status.in_(["awaiting_employee", "awaiting_manager", "scored", "initiated"])) elif action == "completed": q = q.filter(Appraisal.status.in_(["hr_finalized", "applied", "cancelled"])) else: # All (month based) q = q.filter(Appraisal.effective_month >= start).filter(Appraisal.effective_month <= end) elif mgr: # manager view q = q.filter(Appraisal.manager_id == current_user.id) # keep month filter for manager dashboard (can be changed later if needed) q = q.filter(Appraisal.effective_month >= start).filter(Appraisal.effective_month <= end) else: # employee view q = q.filter(Appraisal.employee_id == current_user.id) # keep month filter for employee dashboard q = q.filter(Appraisal.effective_month >= start).filter(Appraisal.effective_month <= end) appraisals = (q.order_by(Appraisal.initiated_at.desc()) .limit(500) .all()) # ---------------------------- # employee map for display # (IMPORTANT: build from the actual list we display) # ---------------------------- emp_ids = list({a.employee_id for a in appraisals if a.employee_id}) emps = Employee.query.filter(Employee.id.in_(emp_ids)).all() if emp_ids else [] emp_map = {e.id: e for e in emps} return render_template( "appraisal/dashboard.html", ref_month=ref, due_rows=due_rows, appraisals=appraisals, emp_map=emp_map, is_hr_admin=is_hr_admin, ) # --------------------------------------------------------------------- # Budgets (HR/Admin) - set per FY and dept # --------------------------------------------------------------------- import re from sqlalchemy import func def _loc_key(loc: str) -> str: """ Convert location text into a safe HTML field key. Example: 'New Delhi' -> 'new_delhi' """ s = (loc or "").strip().lower() s = re.sub(r"[^a-z0-9]+", "_", s) s = re.sub(r"_+", "_", s).strip("_") return s or "loc" @appraisal_bp.route("/budgets", methods=["GET", "POST"]) @login_required def budgets(): if not _is_hr_admin(): flash("Not authorized.", "danger") return redirect(url_for("appraisal.dashboard")) fy = request.args.get("fy", type=int) or fy_for(date.today()) depts = Department.query.order_by(Department.name.asc()).all() # ✅ Auto locations from Employee.location locations = [r[0] for r in ( db.session.query(Employee.location) .filter(Employee.location.isnot(None)) .filter(func.trim(Employee.location) != "") .distinct() .order_by(Employee.location.asc()) .all() )] # If no locations exist yet, give one default bucket if not locations: locations = ["HO"] # --------------------------------------- # POST: Save Global + Location + Dept(Loc) # --------------------------------------- if request.method == "POST": # ----------------------------- # 0) Read ALL values first (no DB writes yet) # ----------------------------- global_val = float(request.form.get("global_budget", "0") or 0) # Location entered values loc_vals = {} for loc in locations: loc_clean = (loc or "").strip() lk = _loc_key(loc_clean) loc_vals[loc_clean] = float(request.form.get(f"loc_{lk}", "0") or 0) # Department entered values per location dept_vals = {} # (loc, dept_id) -> val dept_sums_by_loc = {loc: 0.0 for loc in loc_vals.keys()} for loc_clean in loc_vals.keys(): lk = _loc_key(loc_clean) for d in depts: v = float(request.form.get(f"dept_{lk}_{d.id}", "0") or 0) dept_vals[(loc_clean, d.id)] = v dept_sums_by_loc[loc_clean] += v # ----------------------------- # 1) VALIDATIONS (Hierarchy checks) # ----------------------------- # A) Sum of locations <= global sum_locations = sum(loc_vals.values()) if sum_locations > global_val + 1e-6: flash( f"❌ Invalid budgets: Sum of Location budgets ({sum_locations:.2f}) " f"cannot exceed Global budget ({global_val:.2f}).", "danger" ) return redirect(url_for("appraisal.budgets", fy=fy)) # B) For each location: sum of dept budgets <= location budget for loc_clean, dept_sum in dept_sums_by_loc.items(): loc_budget = float(loc_vals.get(loc_clean, 0.0)) if dept_sum > loc_budget + 1e-6: flash( f"❌ Invalid budgets for '{loc_clean}': Sum of Department budgets ({dept_sum:.2f}) " f"cannot exceed Location budget ({loc_budget:.2f}).", "danger" ) return redirect(url_for("appraisal.budgets", fy=fy)) # ----------------------------- # 2) SAVE (only after passing validations) # ----------------------------- # Global g = GlobalAppraisalBudget.query.filter_by(fy_start_year=fy).first() if not g: g = GlobalAppraisalBudget(fy_start_year=fy, created_by_id=current_user.id) db.session.add(g) g.budget_amount = global_val # Locations for loc_clean, val in loc_vals.items(): row = LocationAppraisalBudget.query.filter_by(fy_start_year=fy, location=loc_clean).first() if not row: row = LocationAppraisalBudget(fy_start_year=fy, location=loc_clean, created_by_id=current_user.id) db.session.add(row) row.budget_amount = val # Departments per location for (loc_clean, dept_id), val in dept_vals.items(): row = (DeptAppraisalBudget.query .filter_by(fy_start_year=fy, location=loc_clean, department_id=dept_id) .first()) if not row: row = DeptAppraisalBudget( fy_start_year=fy, location=loc_clean, department_id=dept_id, created_by_id=current_user.id ) db.session.add(row) row.budget_amount = val db.session.commit() flash("✅ Budgets saved (Global + Location + Department).", "success") return redirect(url_for("appraisal.budgets", fy=fy)) # --------------------------------------- # GET: Build maps for UI # --------------------------------------- global_budget = 0.0 g = GlobalAppraisalBudget.query.filter_by(fy_start_year=fy).first() if g: global_budget = float(g.budget_amount or 0.0) # Location budget map loc_bmap = {} for row in LocationAppraisalBudget.query.filter_by(fy_start_year=fy).all(): loc_bmap[(row.location or "").strip()] = float(row.budget_amount or 0.0) # Dept budget map (loc, dept_id) -> amount dept_bmap = {} for row in DeptAppraisalBudget.query.filter_by(fy_start_year=fy).all(): dept_bmap[((row.location or "").strip(), row.department_id)] = float(row.budget_amount or 0.0) # Usage (remaining) maps using your helpers you added earlier usage_global = fy_remaining_global_budget(fy) usage_loc = {} for loc in locations: usage_loc[loc] = fy_remaining_location_budget(fy, loc) usage_dept = {} for loc in locations: for d in depts: usage_dept[(loc, d.id)] = fy_remaining_dept_budget(fy, loc, d.id) return render_template( "appraisal/budgets.html", fy=fy, depts=depts, locations=locations, loc_key=_loc_key, # ✅ so template can build field names safely global_budget=global_budget, loc_bmap=loc_bmap, dept_bmap=dept_bmap, usage_global=usage_global, usage_loc=usage_loc, usage_dept=usage_dept, ) # ============================================================ try: from models import AppraisalScoreSnapshot # rule usage (optional) except Exception: AppraisalScoreSnapshot = None # ---------- Helpers ---------- def __is_hr_admin(): # reuse your existing role helper from flask_login import current_user r = getattr(current_user, "role", None) if r is not None and not isinstance(r, str): role_name = getattr(r, "name", None) or getattr(r, "role_name", None) or getattr(r, "title", None) or "" else: role_name = r or "" role_name = str(role_name).strip().lower() return role_name in ("admin", "hr", "hr_admin", "superadmin") def _safe_float(val, default=0.0): try: return float(val) except Exception: return default def _safe_int(val, default=0): try: return int(val) except Exception: return default def _question_is_used(question_id: int) -> bool: # If you have AppraisalAnswer(question_id=...), block delete if AppraisalAnswer is None: return False return db.session.query(func.count(AppraisalAnswer.id)).filter(AppraisalAnswer.question_id == question_id).scalar() > 0 def _rule_is_used(rule_id: int) -> bool: # Optional usage check; allow delete if table not present if AppraisalScoreSnapshot is None: return False return db.session.query(func.count(AppraisalScoreSnapshot.id)).filter(AppraisalScoreSnapshot.rule_id == rule_id).scalar() > 0 # ============================================================ # 1) QUESTION MASTER # ============================================================ @appraisal_bp.route("/masters/questions", methods=["GET", "POST"]) @login_required def masters_questions(): if not __is_hr_admin(): flash("Not authorized.", "danger") return redirect(url_for("employee_bp.dashboard")) if request.method == "POST": target = (request.form.get("target") or "employee").strip().lower() question_text = (request.form.get("question_text") or "").strip() weight = _safe_float(request.form.get("weight"), 1.0) sort_order = _safe_int(request.form.get("sort_order"), 0) if target not in ("employee", "manager"): flash("Invalid target.", "danger") return redirect(url_for("appraisal.masters_questions")) if not question_text: flash("Question text is required.", "danger") return redirect(url_for("appraisal.masters_questions")) q = AppraisalQuestion( target=target, question_text=question_text, weight=weight, sort_order=sort_order, is_active=True ) db.session.add(q) db.session.commit() flash("✅ Question added.", "success") return redirect(url_for("appraisal.masters_questions")) qs = (AppraisalQuestion.query .order_by(AppraisalQuestion.target.asc(), AppraisalQuestion.sort_order.asc(), AppraisalQuestion.id.desc()) .all()) return render_template("appraisal/masters_questions.html", qs=qs) @appraisal_bp.route("/masters/questions//toggle", methods=["POST"]) @login_required def toggle_question(qid): if not __is_hr_admin(): flash("Not authorized.", "danger") return redirect(url_for("appraisal.masters_questions")) q = AppraisalQuestion.query.get_or_404(qid) q.is_active = not bool(q.is_active) db.session.commit() flash(f"✅ Question {'enabled' if q.is_active else 'disabled'}.", "success") return redirect(url_for("appraisal.masters_questions")) @appraisal_bp.route("/masters/questions//edit", methods=["POST"]) @login_required def edit_question(qid): if not __is_hr_admin(): flash("Not authorized.", "danger") return redirect(url_for("appraisal.masters_questions")) q = AppraisalQuestion.query.get_or_404(qid) target = (request.form.get("target") or q.target or "employee").strip().lower() question_text = (request.form.get("question_text") or "").strip() weight = _safe_float(request.form.get("weight"), q.weight or 1.0) sort_order = _safe_int(request.form.get("sort_order"), q.sort_order or 0) if target not in ("employee", "manager"): flash("Invalid target.", "danger") return redirect(url_for("appraisal.masters_questions")) if not question_text: flash("Question text is required.", "danger") return redirect(url_for("appraisal.masters_questions")) q.target = target q.question_text = question_text q.weight = weight q.sort_order = sort_order db.session.commit() flash("✅ Question updated.", "success") return redirect(url_for("appraisal.masters_questions")) @appraisal_bp.route("/masters/questions//delete", methods=["POST"]) @login_required def delete_question(qid): if not __is_hr_admin(): flash("Not authorized.", "danger") return redirect(url_for("appraisal.masters_questions")) if _question_is_used(qid): flash("Cannot delete: this question is already used in an appraisal.", "warning") return redirect(url_for("appraisal.masters_questions")) q = AppraisalQuestion.query.get_or_404(qid) db.session.delete(q) db.session.commit() flash("🗑️ Question deleted.", "success") return redirect(url_for("appraisal.masters_questions")) # ============================================================ # 2) SCORE RULE MASTER # ============================================================ @appraisal_bp.route("/masters/score-rules", methods=["GET", "POST"]) @login_required def masters_score_rules(): if not __is_hr_admin(): flash("Not authorized.", "danger") return redirect(url_for("employee_bp.dashboard")) if request.method == "POST": min_score = _safe_float(request.form.get("min_score_pct")) max_score = _safe_float(request.form.get("max_score_pct")) rec_inc = _safe_float(request.form.get("recommended_increment_pct")) if min_score < 0 or max_score > 100 or min_score >= max_score: flash("Invalid range. Ensure 0 <= min < max <= 100.", "danger") return redirect(url_for("appraisal.masters_score_rules")) rule = AppraisalScoreRule( min_score_pct=min_score, max_score_pct=max_score, recommended_increment_pct=rec_inc, is_active=True ) db.session.add(rule) db.session.commit() flash("✅ Score rule added.", "success") return redirect(url_for("appraisal.masters_score_rules")) rules = (AppraisalScoreRule.query .order_by(AppraisalScoreRule.min_score_pct.asc(), AppraisalScoreRule.id.desc()) .all()) return render_template("appraisal/masters_score_rules.html", rules=rules) @appraisal_bp.route("/masters/score-rules//toggle", methods=["POST"]) @login_required def toggle_score_rule(rid): if not __is_hr_admin(): flash("Not authorized.", "danger") return redirect(url_for("appraisal.masters_score_rules")) r = AppraisalScoreRule.query.get_or_404(rid) r.is_active = not bool(r.is_active) db.session.commit() flash(f"✅ Rule {'enabled' if r.is_active else 'disabled'}.", "success") return redirect(url_for("appraisal.masters_score_rules")) @appraisal_bp.route("/masters/score-rules//edit", methods=["POST"]) @login_required def edit_score_rule(rid): if not __is_hr_admin(): flash("Not authorized.", "danger") return redirect(url_for("appraisal.masters_score_rules")) r = AppraisalScoreRule.query.get_or_404(rid) min_score = _safe_float(request.form.get("min_score_pct"), r.min_score_pct) max_score = _safe_float(request.form.get("max_score_pct"), r.max_score_pct) rec_inc = _safe_float(request.form.get("recommended_increment_pct"), r.recommended_increment_pct) if min_score < 0 or max_score > 100 or min_score >= max_score: flash("Invalid range. Ensure 0 <= min < max <= 100.", "danger") return redirect(url_for("appraisal.masters_score_rules")) r.min_score_pct = min_score r.max_score_pct = max_score r.recommended_increment_pct = rec_inc db.session.commit() flash("✅ Rule updated.", "success") return redirect(url_for("appraisal.masters_score_rules")) @appraisal_bp.route("/masters/score-rules//delete", methods=["POST"]) @login_required def delete_score_rule(rid): if not __is_hr_admin(): flash("Not authorized.", "danger") return redirect(url_for("appraisal.masters_score_rules")) if _rule_is_used(rid): flash("Cannot delete: this rule is already used in an appraisal snapshot.", "warning") return redirect(url_for("appraisal.masters_score_rules")) r = AppraisalScoreRule.query.get_or_404(rid) db.session.delete(r) db.session.commit() flash("🗑️ Rule deleted.", "success") return redirect(url_for("appraisal.masters_score_rules")) # --------------------------------------------------------------------- # HR Initiate appraisal (any employee any month) # --------------------------------------------------------------------- @appraisal_bp.route("/initiate", methods=["GET", "POST"]) @login_required def initiate(): if not _is_hr_admin(): flash("Not authorized.", "danger") return redirect(url_for("appraisal.dashboard")) emp_id = request.args.get("employee_id", type=int) emp = Employee.query.get(emp_id) if emp_id else None if request.method == "POST": employee_id = int(request.form.get("employee_id")) process_type = (request.form.get("process_type") or "").strip().upper() # P1/P2/P3 eff_month_str = (request.form.get("effective_month") or "").strip() # YYYY-MM if process_type not in ("P1", "P2", "P3"): flash("Invalid process type.", "danger") return redirect(url_for("appraisal.initiate")) y, m = [int(x) for x in eff_month_str.split("-")] effective_month = date(y, m, 1) emp2 = Employee.query.get(employee_id) if not emp2: flash("Employee not found.", "danger") return redirect(url_for("appraisal.initiate")) mgr_id = getattr(emp2, "reporting_manager_id", None) dept_id = getattr(emp2, "department_id", None) cur_gross = _current_gross(emp2) # ✅ Status handling: # P1 -> employee must fill first # P2 -> manager must fill # P3 -> HR must finalize (so keep it "initiated" until finalize is saved) status = ( "awaiting_employee" if process_type == "P1" else "awaiting_manager" if process_type == "P2" else "initiated" # ✅ IMPORTANT: NOT hr_finalized ) emp_loc = (getattr(emp2, "location", None) or "").strip() or None a = Appraisal( employee_id=emp2.id, manager_id=mgr_id, department_id=dept_id, location=emp_loc, # ✅ NEW (must exist in model + DB) process_type=process_type, current_gross=cur_gross, effective_month=effective_month, initiated_by_id=current_user.id, status=status ) db.session.add(a) db.session.flush() _audit(a.id, "initiate", new={ "employee_id": emp2.id, "process_type": process_type, "effective_month": effective_month.isoformat(), "current_gross": cur_gross }) db.session.commit() flash("Appraisal initiated.", "success") # ✅ For P3: take HR directly to finalize screen to enter % + new gross if process_type == "P3": return redirect(url_for("appraisal.hr_finalize", appraisal_id=a.id)) return redirect(url_for("appraisal.view", appraisal_id=a.id)) employees = Employee.query.order_by(Employee.full_name.asc()).all() today = date.today() default_eff = first_day_of_next_month(today) # default effective from next month return render_template("appraisal/initiate.html", emp=emp, employees=employees, default_eff=default_eff) # --------------------------------------------------------------------- # View appraisal (role-aware) # --------------------------------------------------------------------- @appraisal_bp.route("//view", methods=["GET"]) @login_required def view(appraisal_id): a = Appraisal.query.get_or_404(appraisal_id) # access rules if not _is_hr_admin(): if a.employee_id != current_user.id and a.manager_id != current_user.id: flash("Not authorized.", "danger") return redirect(url_for("appraisal.dashboard")) emp = Employee.query.get(a.employee_id) mgr = Employee.query.get(a.manager_id) if a.manager_id else None # responses er = AppraisalResponse.query.filter_by(appraisal_id=a.id, responder_type="employee").first() mr = AppraisalResponse.query.filter_by(appraisal_id=a.id, responder_type="manager").first() return render_template( "appraisal/view.html", a=a, emp=emp, mgr=mgr, er=er, mr=mr, is_hr_admin=_is_hr_admin(), # ✅ IMPORTANT ) # --------------------------------------------------------------------- # Employee form (P1 only) # --------------------------------------------------------------------- @appraisal_bp.route("//employee", methods=["GET", "POST"]) @login_required def employee_form(appraisal_id): a = Appraisal.query.get_or_404(appraisal_id) if a.process_type != "P1": flash("Employee form not applicable for this appraisal.", "warning") return redirect(url_for("appraisal.view", appraisal_id=appraisal_id)) if a.employee_id != current_user.id: flash("Not authorized.", "danger") return redirect(url_for("appraisal.dashboard")) if a.status not in ("awaiting_employee",): flash("Employee submission not allowed at this stage.", "warning") return redirect(url_for("appraisal.view", appraisal_id=appraisal_id)) questions = (AppraisalQuestion.query .filter_by(target="employee", is_active=True) .order_by(AppraisalQuestion.sort_order.asc()) .all()) resp = AppraisalResponse.query.filter_by(appraisal_id=a.id, responder_type="employee").first() if not resp: resp = AppraisalResponse(appraisal_id=a.id, responder_type="employee", responder_id=current_user.id) db.session.add(resp) db.session.flush() if request.method == "POST": # save answers (1..5) for q in questions: rating = int(request.form.get(f"q_{q.id}", "0") or 0) if rating < 1 or rating > 5: flash("All ratings must be between 1 and 5.", "danger") return redirect(url_for("appraisal.employee_form", appraisal_id=a.id)) # replace existing answers AppraisalAnswer.query.filter_by(response_id=resp.id).delete() db.session.flush() for q in questions: rating = int(request.form.get(f"q_{q.id}")) comment = (request.form.get(f"c_{q.id}") or "").strip() db.session.add(AppraisalAnswer(response_id=resp.id, question_id=q.id, rating=rating, comment=comment)) resp.submitted_at = datetime.utcnow() a.employee_submitted_at = resp.submitted_at a.status = "awaiting_manager" _audit(a.id, "employee_submitted", new={"submitted_at": resp.submitted_at.isoformat()}) db.session.commit() flash("Employee feedback submitted.", "success") return redirect(url_for("appraisal.view", appraisal_id=a.id)) # existing answers amap = {x.question_id: x for x in AppraisalAnswer.query.filter_by(response_id=resp.id).all()} return render_template("appraisal/employee_form.html", a=a, questions=questions, amap=amap) # --------------------------------------------------------------------- # Manager form (P1/P2) # --------------------------------------------------------------------- @appraisal_bp.route("//manager", methods=["GET", "POST"]) @login_required def manager_form(appraisal_id): a = Appraisal.query.get_or_404(appraisal_id) if a.process_type not in ("P1", "P2"): flash("Manager form not applicable for this appraisal.", "warning") return redirect(url_for("appraisal.view", appraisal_id=a.id)) if a.manager_id != current_user.id: flash("Not authorized (not reporting manager).", "danger") return redirect(url_for("appraisal.dashboard")) allowed_status = ("awaiting_manager",) if a.process_type == "P2" else ("awaiting_manager",) if a.status not in allowed_status: flash("Manager submission not allowed at this stage.", "warning") return redirect(url_for("appraisal.view", appraisal_id=a.id)) questions = (AppraisalQuestion.query .filter_by(target="manager", is_active=True) .order_by(AppraisalQuestion.sort_order.asc()) .all()) resp = AppraisalResponse.query.filter_by(appraisal_id=a.id, responder_type="manager").first() if not resp: resp = AppraisalResponse(appraisal_id=a.id, responder_type="manager", responder_id=current_user.id) db.session.add(resp) db.session.flush() if request.method == "POST": for q in questions: rating = int(request.form.get(f"q_{q.id}", "0") or 0) if rating < 1 or rating > 5: flash("All ratings must be between 1 and 5.", "danger") return redirect(url_for("appraisal.manager_form", appraisal_id=a.id)) AppraisalAnswer.query.filter_by(response_id=resp.id).delete() db.session.flush() for q in questions: rating = int(request.form.get(f"q_{q.id}")) comment = (request.form.get(f"c_{q.id}") or "").strip() db.session.add(AppraisalAnswer(response_id=resp.id, question_id=q.id, rating=rating, comment=comment)) resp.submitted_at = datetime.utcnow() a.manager_submitted_at = resp.submitted_at # scoring emp_score = calc_score_pct(a.id, "employee") if a.process_type == "P1" else None mgr_score = calc_score_pct(a.id, "manager") if a.process_type == "P1": total = round(((emp_score or 0.0) + (mgr_score or 0.0)) / 2.0, 2) # simple average a.employee_score_pct = emp_score a.manager_score_pct = mgr_score a.total_score_pct = total else: a.employee_score_pct = None a.manager_score_pct = mgr_score a.total_score_pct = mgr_score a.recommended_increment_pct = recommended_increment_pct(a.total_score_pct or 0.0) a.status = "scored" _audit(a.id, "manager_submitted_and_scored", new={ "employee_score_pct": a.employee_score_pct, "manager_score_pct": a.manager_score_pct, "total_score_pct": a.total_score_pct, "recommended_increment_pct": a.recommended_increment_pct }) db.session.commit() flash("Manager feedback submitted. Scoring complete.", "success") return redirect(url_for("appraisal.hr_finalize", appraisal_id=a.id)) amap = {x.question_id: x for x in AppraisalAnswer.query.filter_by(response_id=resp.id).all()} return render_template("appraisal/manager_form.html", a=a, questions=questions, amap=amap) # --------------------------------------------------------------------- # HR Finalize (P1/P2 after scored, P3 anytime) # --------------------------------------------------------------------- @appraisal_bp.route("//finalize", methods=["GET", "POST"]) @login_required def hr_finalize(appraisal_id): if not _is_hr_admin(): flash("Not authorized.", "danger") return redirect(url_for("appraisal.dashboard")) a = Appraisal.query.get_or_404(appraisal_id) emp = Employee.query.get(a.employee_id) # P1/P2 can finalize only after scoring if a.process_type in ("P1", "P2") and a.status not in ("scored", "hr_finalized"): flash("Finalize allowed only after scoring is complete.", "warning") return redirect(url_for("appraisal.view", appraisal_id=a.id)) # ✅ P3 can finalize when initiated (or if re-opening already finalized) if a.process_type == "P3" and a.status not in ("initiated", "hr_finalized"): flash("Finalize not allowed at this stage.", "warning") return redirect(url_for("appraisal.view", appraisal_id=a.id)) # P3: allow finalize anytime (it starts at hr_finalized but without numbers) if request.method == "POST": # read inputs safely final_pct_raw = (request.form.get("final_increment_pct") or "").strip() final_new_gross_raw = (request.form.get("final_new_gross") or "").strip() old_gross = float(a.current_gross or 0.0) # parse final_pct = float(final_pct_raw) if final_pct_raw else 0.0 final_new_gross = float(final_new_gross_raw) if final_new_gross_raw else 0.0 # ✅ If user entered % but gross not entered/updated, compute gross if final_new_gross <= 0 and final_pct > 0 and old_gross > 0: final_new_gross = round(old_gross * (1 + (final_pct / 100.0)), 2) # ✅ If user entered gross but % not entered, compute % if final_pct == 0 and final_new_gross > 0 and old_gross > 0: final_pct = round(((final_new_gross - old_gross) / old_gross) * 100.0, 2) if final_new_gross <= 0: flash("New Gross must be > 0.", "danger") return redirect(url_for("appraisal.hr_finalize", appraisal_id=a.id)) impact = round((final_new_gross - old_gross) * 12.0, 2) fy = fy_of_effective_month(a.effective_month) loc = (getattr(a, "location", None) or "").strip() dept_id = a.department_id existing = float(a.annualized_impact or 0.0) # allow re-finalize replacing previous impact # 1) Global check g = fy_remaining_global_budget(fy) g_rem_adjusted = float(g["remaining"]) + existing if impact > g_rem_adjusted + 1e-6: flash(f"Global budget exceeded. Required annual impact: {impact}. Remaining: {g_rem_adjusted}.", "danger") return redirect(url_for("appraisal.hr_finalize", appraisal_id=a.id)) # 2) Location check (only if location present) if loc: l = fy_remaining_location_budget(fy, loc) l_rem_adjusted = float(l["remaining"]) + existing if impact > l_rem_adjusted + 1e-6: flash(f"Location budget exceeded ({loc}). Required annual impact: {impact}. Remaining: {l_rem_adjusted}.", "danger") return redirect(url_for("appraisal.hr_finalize", appraisal_id=a.id)) # 3) Dept check (only if dept + location present) if dept_id and loc: d = fy_remaining_dept_budget(fy, loc, dept_id) d_rem_adjusted = float(d["remaining"]) + existing if impact > d_rem_adjusted + 1e-6: flash(f"Department budget exceeded ({loc}). Required annual impact: {impact}. Remaining: {d_rem_adjusted}.", "danger") return redirect(url_for("appraisal.hr_finalize", appraisal_id=a.id)) old = { "final_increment_pct": a.final_increment_pct, "final_new_gross": a.final_new_gross, "annualized_impact": a.annualized_impact, "status": a.status } a.final_increment_pct = round(final_pct, 2) a.final_new_gross = round(final_new_gross, 2) a.annualized_impact = impact a.hr_finalized_at = datetime.utcnow() a.hr_finalized_by_id = current_user.id a.status = "hr_finalized" _audit(a.id, "hr_finalized", old=old, new={ "final_increment_pct": a.final_increment_pct, "final_new_gross": a.final_new_gross, "annualized_impact": a.annualized_impact, "effective_month": a.effective_month.isoformat() }) db.session.commit() flash("HR finalized appraisal.", "success") return redirect(url_for("appraisal.view", appraisal_id=a.id)) # defaults recommended_pct = float(a.recommended_increment_pct or 0.0) default_new_gross = round(float(a.current_gross or 0.0) * (1 + (recommended_pct / 100.0)), 2) if recommended_pct else float(a.current_gross or 0.0) fy = fy_of_effective_month(a.effective_month) loc = (getattr(a, "location", None) or "").strip() budget_info = { "global": fy_remaining_global_budget(fy), "location": fy_remaining_location_budget(fy, loc) if loc else None, "department": fy_remaining_dept_budget(fy, loc, a.department_id) if (loc and a.department_id) else None, } return render_template( "appraisal/hr_finalize.html", a=a, emp=emp, default_new_gross=default_new_gross, budget_info=budget_info ) # --------------------------------------------------------------------- # Apply appraisal -> Redirect to Assign Salary with new gross # --------------------------------------------------------------------- @appraisal_bp.route("//apply", methods=["POST"]) @login_required def apply(appraisal_id): if not _is_hr_admin(): flash("Not authorized.", "danger") return redirect(url_for("appraisal.dashboard")) a = Appraisal.query.get_or_404(appraisal_id) if a.status != "hr_finalized": flash("Appraisal must be HR finalized before apply.", "warning") return redirect(url_for("appraisal.view", appraisal_id=a.id)) if not a.final_new_gross or a.final_new_gross <= 0: flash("Final new gross missing.", "danger") return redirect(url_for("appraisal.view", appraisal_id=a.id)) emp = Employee.query.get(a.employee_id) if not emp: flash("Employee not found.", "danger") return redirect(url_for("appraisal.dashboard")) # Update employee last appraisal effective month (month-based rule) emp.last_appraisal_effective_month = a.effective_month a.applied_at = datetime.utcnow() a.applied_by_id = current_user.id a.status = "applied" _audit(a.id, "applied", new={"employee_last_appraisal_effective_month": a.effective_month.isoformat()}) db.session.commit() # Redirect to your Assign Salary page with gross prefilled # Adjust endpoint as per your payroll blueprint. return redirect(url_for( "payroll.salary_structure", employee_id=emp.id, gross=a.final_new_gross, source="appraisal", appraisal_id=a.id )) # --------------------------------------------------------------------- # Audit view (HR/Admin) # --------------------------------------------------------------------- @appraisal_bp.route("//audit", methods=["GET"]) @login_required def audit(appraisal_id): if not _is_hr_admin(): flash("Not authorized.", "danger") return redirect(url_for("appraisal.dashboard")) a = Appraisal.query.get_or_404(appraisal_id) logs = (AppraisalAuditLog.query .filter_by(appraisal_id=a.id) .order_by(AppraisalAuditLog.created_at.desc()) .limit(500) .all()) return render_template("appraisal/audit.html", a=a, logs=logs) from sqlalchemy import or_ # --------------------------------------------------------------------- # List: all appraisals with status (role-aware + filters) # --------------------------------------------------------------------- @appraisal_bp.route("/list", methods=["GET"]) @login_required def list_appraisals(): role = _role_name() is_hr_admin = _is_hr_admin() mgr = is_manager() # Filters ym = (request.args.get("month") or "").strip() # YYYY-MM status = (request.args.get("status") or "").strip() # any status string dept_id = request.args.get("department_id", type=int) search = (request.args.get("search") or "").strip() # Month bounds (optional) start = end = None if ym: try: y, m = [int(x) for x in ym.split("-")] ref = date(y, m, 1) start, end = month_bounds(ref) except Exception: start = end = None # Base query q = (db.session.query(Appraisal, Employee, Department) .join(Employee, Employee.id == Appraisal.employee_id) .outerjoin(Department, Department.id == Employee.department_id)) # Role visibility if is_hr_admin: pass elif mgr: # manager sees appraisals where he is the manager OR employee is his direct report q = q.filter(or_(Appraisal.manager_id == current_user.id, Employee.reporting_manager_id == current_user.id)) else: q = q.filter(Appraisal.employee_id == current_user.id) # Apply filters if start and end: q = q.filter(Appraisal.effective_month >= start, Appraisal.effective_month <= end) if status: q = q.filter(Appraisal.status == status) if dept_id: q = q.filter(Employee.department_id == dept_id) if search: q = q.filter(or_( Employee.full_name.ilike(f"%{search}%"), Employee.employee_code.ilike(f"%{search}%") )) rows = (q.order_by(Appraisal.initiated_at.desc()) .limit(500) .all()) # For filter dropdown departments = Department.query.order_by(Department.name.asc()).all() # For status dropdown (keep in sync with your module statuses) status_choices = [ "initiated", "awaiting_employee", "awaiting_manager", "scored", "hr_finalized", "applied", "rejected", ] return render_template( "appraisal/appraisal_list.html", rows=rows, departments=departments, status_choices=status_choices, is_hr_admin=is_hr_admin, is_manager=mgr, filters={ "month": ym, "status": status, "department_id": dept_id, "search": search, } ) from sqlalchemy import or_, and_ # --------------------------------------------------------------------- # HOD / Management Summary View (Employee + Manager feedback) # --------------------------------------------------------------------- def _is_hod_or_management() -> bool: """ Adjust role names here based on what you use in your system. """ r = _role_name() return r in ( "hod", "management", "director", "vp", "cxo", "management_admin", "business_head" ) @appraisal_bp.route("/feedback-summary", methods=["GET"]) @login_required def feedback_summary(): """ Summary page to view both employee + manager feedback for appraisals. Visible to: HR/Admin + HOD/Management """ if not (_is_hr_admin() or _is_hod_or_management()): flash("Not authorized.", "danger") return redirect(url_for("appraisal.dashboard")) # ----------------------- # Filters # ----------------------- ym = (request.args.get("month") or "").strip() # YYYY-MM status = (request.args.get("status") or "").strip() # appraisal status dept_id = request.args.get("department_id", type=int) search = (request.args.get("search") or "").strip() # optional: show only finalized/scored, etc. view_mode = (request.args.get("view_mode") or "all").strip().lower() # view_mode: all | scored_only | finalized_only start = end = None if ym: try: y, m = [int(x) for x in ym.split("-")] ref = date(y, m, 1) start, end = month_bounds(ref) except Exception: start = end = None # ----------------------- # Base query (Appraisal + Employee + Department) # ----------------------- q = ( db.session.query(Appraisal, Employee, Department) .join(Employee, Employee.id == Appraisal.employee_id) .outerjoin(Department, Department.id == Employee.department_id) ) # ----------------------- # Role visibility: # HR/Admin: all # HOD/Management: (safe default) restrict to their department IF they have department_id # If you want Management to see all, remove this block for management roles. # ----------------------- if _is_hod_or_management() and not _is_hr_admin(): my_dept_id = getattr(current_user, "department_id", None) # If user has a dept, restrict to it; else allow all (or restrict as per your preference) if my_dept_id: q = q.filter(Employee.department_id == my_dept_id) # ----------------------- # Apply filters # ----------------------- if start and end: q = q.filter(Appraisal.effective_month >= start, Appraisal.effective_month <= end) if status: q = q.filter(Appraisal.status == status) if dept_id: q = q.filter(Employee.department_id == dept_id) if search: like = f"%{search}%" q = q.filter(or_( Employee.full_name.ilike(like), Employee.employee_code.ilike(like) )) if view_mode == "scored_only": q = q.filter(Appraisal.status.in_(["scored", "hr_finalized", "applied"])) elif view_mode == "finalized_only": q = q.filter(Appraisal.status.in_(["hr_finalized", "applied"])) rows = ( q.order_by(Appraisal.initiated_at.desc()) .limit(700) .all() ) # ----------------------- # Preload responses + answers in bulk (FAST) # ----------------------- appraisal_ids = [a.id for (a, e, d) in rows] response_map = {} # (appraisal_id, responder_type) -> AppraisalResponse answers_map = {} # response_id -> list[{q, rating, comment, weight}] if appraisal_ids: responses = ( AppraisalResponse.query .filter(AppraisalResponse.appraisal_id.in_(appraisal_ids)) .all() ) for r in responses: response_map[(r.appraisal_id, (r.responder_type or "").lower())] = r response_ids = [r.id for r in responses] if response_ids: # Join answers with questions so we can show question text + weight ans_rows = ( db.session.query(AppraisalAnswer, AppraisalQuestion) .join(AppraisalQuestion, AppraisalQuestion.id == AppraisalAnswer.question_id) .filter(AppraisalAnswer.response_id.in_(response_ids)) .order_by(AppraisalQuestion.sort_order.asc(), AppraisalQuestion.id.asc()) .all() ) for ans, q in ans_rows: answers_map.setdefault(ans.response_id, []).append({ "question_text": q.question_text, "weight": float(q.weight or 1.0), "rating": int(ans.rating or 0), "comment": (ans.comment or "").strip(), "target": (q.target or "").lower(), }) # ----------------------- # Score map (use your helper) # ----------------------- score_map = {} # appraisal_id -> {"employee": pct, "manager": pct, "total": pct} for (a, e, d) in rows: # these fields already exist on Appraisal, but we compute safely if missing emp_pct = a.employee_score_pct mgr_pct = a.manager_score_pct tot_pct = a.total_score_pct if emp_pct is None and a.process_type == "P1": emp_pct = calc_score_pct(a.id, "employee") if mgr_pct is None and a.process_type in ("P1", "P2"): mgr_pct = calc_score_pct(a.id, "manager") if tot_pct is None: if a.process_type == "P1": tot_pct = round(((emp_pct or 0.0) + (mgr_pct or 0.0)) / 2.0, 2) else: tot_pct = round((mgr_pct or 0.0), 2) score_map[a.id] = { "employee": emp_pct, "manager": mgr_pct, "total": tot_pct } departments = Department.query.order_by(Department.name.asc()).all() status_choices = [ "initiated", "awaiting_employee", "awaiting_manager", "scored", "hr_finalized", "applied", "rejected", "cancelled", ] return render_template( "appraisal/feedback_summary.html", rows=rows, response_map=response_map, answers_map=answers_map, score_map=score_map, departments=departments, status_choices=status_choices, filters={ "month": ym, "status": status, "department_id": dept_id, "search": search, "view_mode": view_mode, }, is_hr_admin=_is_hr_admin() )