from __future__ import annotations from sqlalchemy import or_, func from models import Role from datetime import datetime import json from flask import ( Blueprint, render_template, request, redirect, url_for, flash, current_app ) from flask_login import login_required, current_user from extensions import db from models import ( Employee, Designation, Department, RecruitmentTemplate, TemplateKRA, TemplateKPI, TemplateSkill, TemplateSkillQuestion, Requisition, RequisitionApproval, RequisitionStage, Candidate, Application, Interview, InterviewFeedback, InterviewFeedbackItem,Role ) from sqlalchemy import or_ recruitment_bp = Blueprint("recruitment", __name__, url_prefix="/recruitment") # ----------------------------- # Helpers # ----------------------------- from sqlalchemy import or_ # add this import at the top (used later in applications_list) # ----------------------------- # Helpers # ----------------------------- def _get_current_employee() -> Employee | None: # If the logged-in user *is* an Employee instance try: if isinstance(current_user, Employee): return current_user except Exception: pass # direct field eid = getattr(current_user, "employee_id", None) if eid: emp = Employee.query.get(eid) if emp: return emp # relationship emp_rel = getattr(current_user, "employee", None) if emp_rel and getattr(emp_rel, "id", None): return emp_rel # fallback by email if getattr(current_user, "email", None): emp = Employee.query.filter_by(email=current_user.email).first() if emp: return emp return None def _is_hr_or_admin(emp: Employee | None) -> bool: if not emp: return False role = getattr(emp, "role", None) if not role: return False name = (role.name or "").strip().lower() return name in ("admin", "hr", "hr admin", "hr manager", "recruiter") # ================================================== # 1. TEMPLATE MANAGEMENT (HR / Admin only) # ================================================== @recruitment_bp.route("/templates") @login_required def template_list(): emp = _get_current_employee() if not _is_hr_or_admin(emp): flash("You are not allowed to manage recruitment templates.", "danger") return redirect(url_for("employee.dashboard")) templates = ( RecruitmentTemplate.query .join(Designation) .order_by(Designation.id) .all() ) designations = Designation.query.order_by(Designation.id).all() return render_template( "recruitment/template_list.html", templates=templates, designations=designations, ) @recruitment_bp.route("/templates/new", methods=["GET", "POST"]) @login_required def template_new(): emp = _get_current_employee() if not _is_hr_or_admin(emp): flash("You are not allowed to manage recruitment templates.", "danger") return redirect(url_for("employee.dashboard")) designations = Designation.query.order_by(Designation.id).all() if request.method == "POST": designation_id = request.form.get("designation_id", type=int) if not designation_id: flash("Please select a designation", "danger") return redirect(url_for("recruitment.template_new")) existing = RecruitmentTemplate.query.filter_by(designation_id=designation_id).first() if existing: flash("Template already exists for this designation.", "warning") return redirect(url_for("recruitment.template_edit", template_id=existing.id)) tpl = RecruitmentTemplate( designation_id=designation_id, min_education=request.form.get("min_education"), min_experience_years=request.form.get("min_experience_years") or None, age_bracket=request.form.get("age_bracket"), salary_bracket=request.form.get("salary_bracket"), accountability=request.form.get("accountability"), job_description=request.form.get("job_description"), ) db.session.add(tpl) db.session.commit() flash("Template created. Now add KRAs, KPIs & Skills.", "success") return redirect(url_for("recruitment.template_edit", template_id=tpl.id)) # Pass a *dict* for structure; JS will get it via |tojson empty_structure = {"kras": [], "skills": []} return render_template( "recruitment/template_form.html", template=None, designations=designations, structure=empty_structure, ) @recruitment_bp.route("/templates//edit", methods=["GET", "POST"]) @login_required def template_edit(template_id): emp = _get_current_employee() if not _is_hr_or_admin(emp): flash("You are not allowed to manage recruitment templates.", "danger") return redirect(url_for("employee.dashboard")) tpl = RecruitmentTemplate.query.get_or_404(template_id) designations = Designation.query.order_by(Designation.title).all() # ----------------- POST: save template + structure ----------------- if request.method == "POST": tpl.min_education = request.form.get("min_education") tpl.min_experience_years = request.form.get("min_experience_years") or None tpl.age_bracket = request.form.get("age_bracket") tpl.salary_bracket = request.form.get("salary_bracket") tpl.accountability = request.form.get("accountability") tpl.job_description = request.form.get("job_description") struct_raw = request.form.get("structure_json") or "" try: struct = json.loads(struct_raw) if struct_raw.strip() else {} except Exception: struct = {} flash("Could not parse KRA/KPI/Skill structure – please try again.", "danger") # Clear existing KRAs/KPIs existing_kras = TemplateKRA.query.filter_by(template_id=tpl.id).all() for kra in existing_kras: TemplateKPI.query.filter_by(kra_id=kra.id).delete() db.session.delete(kra) # Clear existing Skills/Questions existing_skills = TemplateSkill.query.filter_by(template_id=tpl.id).all() for sk in existing_skills: TemplateSkillQuestion.query.filter_by(skill_id=sk.id).delete() db.session.delete(sk) db.session.flush() # Recreate KRAs + KPIs for kra_dict in (struct.get("kras") or []): if not kra_dict.get("title"): continue kra = TemplateKRA( template_id=tpl.id, name=kra_dict.get("title"), description=kra_dict.get("description"), weightage=kra_dict.get("weightage") or None, ) db.session.add(kra) db.session.flush() for kpi_dict in (kra_dict.get("kpis") or []): if not kpi_dict.get("title"): continue db.session.add( TemplateKPI( template_id=tpl.id, kra_id=kra.id, name=kpi_dict.get("title"), target=kpi_dict.get("target"), weightage=kpi_dict.get("weightage") or None, ) ) # Recreate Skills + Questions for sk_dict in (struct.get("skills") or []): if not sk_dict.get("name"): continue sk = TemplateSkill( template_id=tpl.id, name=sk_dict.get("name"), basic_requirement=sk_dict.get("basic_requirement"), ) db.session.add(sk) db.session.flush() for q_dict in (sk_dict.get("questions") or []): text = (q_dict.get("text") or "").strip() if not text: continue db.session.add( TemplateSkillQuestion( skill_id=sk.id, question_text=text, ) ) db.session.commit() flash("Template updated with KRAs, KPIs & Skills.", "success") return redirect(url_for("recruitment.template_edit", template_id=tpl.id)) # ----------------- GET: load existing structure ----------------- kras = ( TemplateKRA.query .filter_by(template_id=tpl.id) .order_by(TemplateKRA.id) .all() ) skills = ( TemplateSkill.query .filter_by(template_id=tpl.id) .order_by(TemplateSkill.id) .all() ) structure = {"kras": [], "skills": []} for kra in kras: kpi_rows = ( TemplateKPI.query .filter_by(kra_id=kra.id) .order_by(TemplateKPI.id) .all() ) structure["kras"].append( { "id": kra.id, "title": kra.name, "description": kra.description, "weightage": kra.weightage, "kpis": [ { "id": kpi.id, "title": kpi.name, "target": kpi.target, "weightage": kpi.weightage, } for kpi in kpi_rows ], } ) for sk in skills: q_rows = ( TemplateSkillQuestion.query .filter_by(skill_id=sk.id) .order_by(TemplateSkillQuestion.id) .all() ) structure["skills"].append( { "id": sk.id, "name": sk.name, "basic_requirement": sk.basic_requirement, "questions": [ {"id": q.id, "text": q.question_text} for q in q_rows ], } ) # Give both the dict and JSON string to the template structure_json = json.dumps(structure, ensure_ascii=False) return render_template( "recruitment/template_form.html", template=tpl, designations=designations, structure=structure, structure_json=structure_json, ) # ================================================== # 2. REQUISITION – RAISE, LIST, APPROVE # ================================================== def _generate_req_code(): # Simple HRREQ-YYYYMM-XXXX generator now = datetime.now() prefix = f"HRREQ-{now:%Y%m}" last = ( Requisition.query.filter(Requisition.code.like(f"{prefix}-%")) .order_by(Requisition.code.desc()) .first() ) last_num = 0 if last and last.code: try: last_num = int(last.code.split("-")[-1]) except Exception: last_num = 0 return f"{prefix}-{last_num+1:03d}" @recruitment_bp.route("/requisitions/my") @login_required def my_requisitions(): emp = _get_current_employee() if not emp: flash("Employee profile not linked.", "danger") return redirect(url_for("employee.dashboard")) reqs = ( Requisition.query .filter_by(raised_by_id=emp.id) .order_by(Requisition.created_on.desc()) .all() ) return render_template("recruitment/requisition_list.html", requisitions=reqs) @recruitment_bp.route("/requisitions/new", methods=["GET", "POST"]) @login_required def requisition_new(): emp = _get_current_employee() if not emp: flash("Employee profile not linked.", "danger") return redirect(url_for("employee.dashboard")) designations = Designation.query.order_by(Designation.id).all() departments = Department.query.order_by(Department.name).all() selected_desig_id = request.args.get("designation_id", type=int) template = RecruitmentTemplate.query.filter_by(designation_id=selected_desig_id).first() if selected_desig_id else None if request.method == "POST": desig_id = request.form.get("designation_id", type=int) dept_id = request.form.get("department_id", type=int) positions = request.form.get("positions", type=int) or 1 if not desig_id: flash("Please select designation.", "danger") return redirect(url_for("recruitment.requisition_new")) template = RecruitmentTemplate.query.filter_by(designation_id=desig_id).first() req = Requisition( code=_generate_req_code(), raised_by_id=emp.id, manager_id=getattr(emp, "reporting_manager_id", None), hr_owner_id=None, # optional: you can auto-assign HR later designation_id=desig_id, department_id=dept_id, positions=positions, location=request.form.get("location"), work_mode=request.form.get("work_mode"), employment_type=request.form.get("employment_type"), expected_doj=request.form.get("expected_doj") or None, budget_ctc_min=request.form.get("budget_ctc_min") or None, budget_ctc_max=request.form.get("budget_ctc_max") or None, justification=request.form.get("justification"), status="pending_manager", ) # snapshot pre-filled template + user overrides snap = { "min_education": request.form.get("min_education") or (template.min_education if template else None), "min_experience_years": request.form.get("min_experience_years") or (template.min_experience_years if template else None), "age_bracket": request.form.get("age_bracket") or (template.age_bracket if template else None), "salary_bracket": request.form.get("salary_bracket") or (template.salary_bracket if template else None), "accountability": request.form.get("accountability") or (template.accountability if template else None), "job_description": request.form.get("job_description") or (template.job_description if template else None), } req.snapshot_json = snap db.session.add(req) db.session.flush() # req.id available # create approval rows: manager + HR if req.manager_id: db.session.add( RequisitionApproval( requisition_id=req.id, approver_id=req.manager_id, level="manager", status="pending", ) ) # Optionally assign HR owner based on role or config # For now we'll just leave HR approval to be created later or skip. db.session.commit() flash("Requisition created and sent for Manager approval.", "success") return redirect(url_for("recruitment.my_requisitions")) return render_template( "recruitment/requisition_new.html", designations=designations, departments=departments, template=template, selected_desig_id=selected_desig_id, ) @recruitment_bp.route("/requisitions/") @login_required def requisition_view(req_id): from sqlalchemy import func req = Requisition.query.get_or_404(req_id) emp = _get_current_employee() if not emp: flash("Employee profile not linked.", "danger") return redirect(url_for("employee.dashboard")) # basic visibility rule: raiser, manager, HR, admin can_view = ( req.raised_by_id == emp.id or req.manager_id == emp.id or req.hr_owner_id == emp.id or _is_hr_or_admin(emp) ) if not can_view: flash("You are not allowed to view this requisition.", "danger") return redirect(url_for("employee.dashboard")) # snapshot used in the template snap = req.snapshot_json or {} # -------- Stage stats (you already had this) -------- stage_stats = [] stages = RequisitionStage.query.filter_by(requisition_id=req.id).order_by(RequisitionStage.order).all() for st in stages: q = ( Interview.query .filter_by(stage_id=st.id) .filter(Interview.application.has(requisition_id=req.id)) ) total = q.count() scheduled = q.filter_by(status="scheduled").count() completed = q.filter_by(status="completed").count() cancelled = q.filter_by(status="cancelled").count() last_dt = q.with_entities(func.max(Interview.scheduled_at)).scalar() stage_stats.append( type("StageStat", (), { "stage": st, "total": total, "scheduled": scheduled, "completed": completed, "cancelled": cancelled, "last_dt": last_dt, }) ) # -------- HR finalisation data -------- is_hr_admin = _is_hr_or_admin(emp) final_list = [] if is_hr_admin: # Take only applications which have completed interviews + feedback rows = ( db.session.query( Application, Candidate, func.avg(InterviewFeedback.overall_rating).label("avg_rating"), func.count(InterviewFeedback.id).label("fb_count"), ) .join(Candidate, Candidate.id == Application.candidate_id) .join(Interview, Interview.application_id == Application.id) .join(InterviewFeedback, InterviewFeedback.interview_id == Interview.id) .filter(Application.requisition_id == req.id) .group_by(Application.id, Candidate.id) .having(func.count(InterviewFeedback.id) > 0) .order_by(func.avg(InterviewFeedback.overall_rating).desc()) .all() ) for app_obj, cand_obj, avg_rating, fb_count in rows: final_list.append({ "application": app_obj, "candidate": cand_obj, "avg_rating": round(avg_rating or 0, 1), "feedback_count": fb_count, }) return render_template( "recruitment/requisition_view.html", req=req, snap=snap, stage_stats=stage_stats, is_hr_admin=is_hr_admin, final_list=final_list, ) @recruitment_bp.route("/approvals/my") @login_required def my_requisition_approvals(): emp = _get_current_employee() if not emp: flash("Employee profile not linked.", "danger") return redirect(url_for("employee.dashboard")) # HR / Admin: see ALL pending approvals if _is_hr_or_admin(emp): pending = ( RequisitionApproval.query .filter_by(status="pending") .order_by(RequisitionApproval.id.desc()) .all() ) else: # Normal manager: only approvals assigned to them pending = ( RequisitionApproval.query .filter_by(approver_id=emp.id, status="pending") .order_by(RequisitionApproval.id.desc()) .all() ) return render_template("recruitment/approval_inbox.html", approvals=pending) @recruitment_bp.route("/approvals//decide", methods=["POST"]) @login_required def decide_requisition_approval(approval_id): emp = _get_current_employee() if not emp: flash("Employee profile not linked.", "danger") return redirect(url_for("employee.dashboard")) appr = RequisitionApproval.query.get_or_404(approval_id) # Allowed: the assigned approver OR any HR/Admin (including admin user) if appr.approver_id != emp.id and not _is_hr_or_admin(emp): flash("You cannot act on this approval.", "danger") return redirect(url_for("recruitment.my_requisition_approvals")) decision = request.form.get("decision") # approve / reject remarks = request.form.get("remarks") if decision not in ("approve", "reject"): flash("Invalid decision.", "danger") return redirect(url_for("recruitment.my_requisition_approvals")) appr.status = "approved" if decision == "approve" else "rejected" appr.remarks = remarks appr.decided_on = datetime.utcnow() req = appr.requisition # ---------------- Workflow logic ---------------- if appr.level == "manager": if appr.status == "approved": # Find an HR / Admin employee to own HR approval hr_emp = ( Employee.query .join(Role) .filter(func.lower(Role.name).in_(["hr", "hr admin", "hr manager", "admin"])) .first() ) if hr_emp: req.status = "pending_hr" req.hr_owner_id = hr_emp.id # Create HR approval row if it doesn't already exist existing_hr = RequisitionApproval.query.filter_by( requisition_id=req.id, level="hr", ).first() if not existing_hr: db.session.add( RequisitionApproval( requisition_id=req.id, approver_id=hr_emp.id, level="hr", status="pending", ) ) else: # No HR/Admin found – treat manager approval as final req.status = "approved" else: req.status = "rejected" elif appr.level == "hr": if appr.status == "approved": req.status = "approved" else: req.status = "rejected" db.session.commit() flash("Decision recorded.", "success") return redirect(url_for("recruitment.my_requisition_approvals")) # ================================================== # 3. CANDIDATE APPLICATIONS & INTERVIEWS # ================================================== from werkzeug.utils import secure_filename import os # Configure in Config: RECRUITMENT_UPLOAD_FOLDER def _resume_folder(): folder = current_app.config.get("RECRUITMENT_UPLOAD_FOLDER") or os.path.join( current_app.root_path, "uploads", "recruitment" ) os.makedirs(folder, exist_ok=True) return folder @recruitment_bp.route("/jobs//apply", methods=["GET", "POST"]) def public_apply(req_code): req = Requisition.query.filter_by(code=req_code, status="approved").first_or_404() if request.method == "POST": full_name = request.form.get("full_name") email = request.form.get("email") phone = request.form.get("phone") if not full_name or not email: flash("Name and Email are required.", "danger") return redirect(request.url) # save resume resume_file = request.files.get("resume") resume_path = None if resume_file and resume_file.filename: fname = secure_filename(resume_file.filename) folder = _resume_folder() save_path = os.path.join(folder, fname) resume_file.save(save_path) resume_path = os.path.relpath(save_path, current_app.root_path) cand = Candidate( full_name=full_name, email=email, phone=phone, total_experience_years=request.form.get("experience_years") or None, current_company=request.form.get("current_company"), current_ctc=request.form.get("current_ctc") or None, expected_ctc=request.form.get("expected_ctc") or None, location=request.form.get("location"), resume_path=resume_path, source="public", ) db.session.add(cand) db.session.flush() app = Application( requisition_id=req.id, candidate_id=cand.id, status="applied", ) db.session.add(app) db.session.commit() flash("Your application has been submitted. Thank you!", "success") return redirect(request.url) return render_template("recruitment/public_apply.html", req=req) @recruitment_bp.route("/applications/new-from-cv", methods=["GET", "POST"]) @login_required def new_application_from_cv(): emp = _get_current_employee() if not _is_hr_or_admin(emp): flash("Only HR/Admin can upload CVs.", "danger") return redirect(url_for("employee.dashboard")) approved_reqs = Requisition.query.filter_by(status="approved").order_by(Requisition.created_on.desc()).all() if request.method == "POST": req_id = request.form.get("requisition_id", type=int) req = Requisition.query.get_or_404(req_id) full_name = request.form.get("full_name") if not full_name: flash("Full name is required.", "danger") return redirect(request.url) resume_file = request.files.get("resume") resume_path = None if resume_file and resume_file.filename: fname = secure_filename(resume_file.filename) folder = _resume_folder() save_path = os.path.join(folder, fname) resume_file.save(save_path) resume_path = os.path.relpath(save_path, current_app.root_path) cand = Candidate( full_name=full_name, email=request.form.get("email"), phone=request.form.get("phone"), source="hr_upload", resume_path=resume_path, ) db.session.add(cand) db.session.flush() app = Application( requisition_id=req.id, candidate_id=cand.id, status="in_process", ) db.session.add(app) db.session.commit() flash("Candidate linked to requisition.", "success") return redirect(url_for("recruitment.view_application", app_id=app.id)) return render_template("recruitment/new_application_from_cv.html", requisitions=approved_reqs) @recruitment_bp.route("/applications/") @login_required def view_application(app_id): app = Application.query.get_or_404(app_id) emp = _get_current_employee() if not emp: flash("Employee profile not linked.", "danger") return redirect(url_for("employee.dashboard")) # List of interviews for this application interviews = ( Interview.query .filter_by(application_id=app.id) .order_by(Interview.scheduled_at.desc()) .all() ) # --------- NEW: feedback summary for HR -------------- feedbacks = ( InterviewFeedback.query .join(Interview, InterviewFeedback.interview_id == Interview.id) .filter(Interview.application_id == app.id) .order_by(InterviewFeedback.id.desc()) .all() ) # Aggregate overall rating avg_rating = None if feedbacks: ratings = [fb.overall_rating for fb in feedbacks if fb.overall_rating is not None] if ratings: avg_rating = round(sum(ratings) / len(ratings), 1) # Map interview_id -> list of feedbacks (normally 1, but future-proof) fb_by_interview = {} for fb in feedbacks: fb_by_interview.setdefault(fb.interview_id, []).append(fb) # Check if ALL interviews are completed AND have feedback total_interviews = len(interviews) completed_with_feedback = 0 for iv in interviews: if (iv.status or "").lower() == "completed" and iv.id in fb_by_interview: completed_with_feedback += 1 all_interviews_done = total_interviews > 0 and completed_with_feedback == total_interviews # Only HR/Admin can take final decision can_decide = ( _is_hr_or_admin(emp) and all_interviews_done and (app.status or "").lower() in ("applied", "in_process") ) return render_template( "recruitment/application_view.html", app=app, interviews=interviews, feedbacks=feedbacks, fb_by_interview=fb_by_interview, avg_rating=avg_rating, all_interviews_done=all_interviews_done, can_decide=can_decide, ) @recruitment_bp.route("/applications//schedule-interview", methods=["GET", "POST"]) @login_required def schedule_interview(app_id): emp = _get_current_employee() if not emp: flash("Employee profile not linked.", "danger") return redirect(url_for("employee.dashboard")) app = Application.query.get_or_404(app_id) stages = RequisitionStage.query.filter_by(requisition_id=app.requisition_id).order_by(RequisitionStage.order).all() interviewers = Employee.query.filter(Employee.status.in_(["active", "notice period", "notice_period"])).order_by(Employee.full_name).all() if request.method == "POST": stage_id = request.form.get("stage_id", type=int) interviewer_id = request.form.get("interviewer_id", type=int) scheduled_at_str = request.form.get("scheduled_at") mode = request.form.get("mode") location = request.form.get("location") if not stage_id or not interviewer_id: flash("Stage and interviewer are required.", "danger") return redirect(request.url) scheduled_at = None if scheduled_at_str: try: scheduled_at = datetime.fromisoformat(scheduled_at_str) except Exception: pass interview = Interview( application_id=app.id, stage_id=stage_id, interviewer_id=interviewer_id, scheduled_at=scheduled_at, mode=mode, location=location, status="scheduled", ) db.session.add(interview) db.session.commit() flash("Interview scheduled.", "success") return redirect(url_for("recruitment.view_application", app_id=app.id)) return render_template("recruitment/schedule_interview.html", app=app, stages=stages, interviewers=interviewers) @recruitment_bp.route("/interviews//feedback", methods=["GET", "POST"]) @login_required def interview_feedback(int_id): emp = _get_current_employee() if not emp: flash("Employee profile not linked.", "danger") return redirect(url_for("employee.dashboard")) interview = Interview.query.get_or_404(int_id) if interview.interviewer_id != emp.id and not _is_hr_or_admin(emp): flash("You are not allowed to submit feedback for this interview.", "danger") return redirect(url_for("employee.dashboard")) # --- Load requisition + template structure for reference --- req = interview.application.requisition if interview.application else None tpl = None structure = {"kras": [], "skills": []} if req and req.designation_id: tpl = RecruitmentTemplate.query.filter_by(designation_id=req.designation_id).first() if tpl: # KRAs + KPIs kra_rows = TemplateKRA.query.filter_by(template_id=tpl.id).order_by(TemplateKRA.id).all() for kra in kra_rows: kpi_rows = TemplateKPI.query.filter_by(kra_id=kra.id).order_by(TemplateKPI.id).all() structure["kras"].append( { "id": kra.id, "title": kra.name, "description": kra.description, "weightage": kra.weightage, "kpis": [ { "id": kpi.id, "title": kpi.name, "target": kpi.target, "weightage": kpi.weightage, } for kpi in kpi_rows ], } ) # Skills + Questions skill_rows = TemplateSkill.query.filter_by(template_id=tpl.id).order_by(TemplateSkill.id).all() for sk in skill_rows: q_rows = ( TemplateSkillQuestion.query .filter_by(skill_id=sk.id) .order_by(TemplateSkillQuestion.id) .all() ) structure["skills"].append( { "id": sk.id, "name": sk.name, "basic_requirement": sk.basic_requirement, "questions": [ {"id": q.id, "text": q.question_text} for q in q_rows ], } ) # --- Handle POST (save feedback) --- if request.method == "POST": overall_rating = request.form.get("overall_rating", type=int) recommendation = request.form.get("recommendation") comments = request.form.get("comments") fb = InterviewFeedback( interview_id=interview.id, interviewer_id=emp.id, overall_rating=overall_rating, recommendation=recommendation, comments=comments, ) db.session.add(fb) db.session.flush() # 1) Save PREDEFINED template skill questions (tplq_* fields) for sk in structure.get("skills", []): skill_name = sk.get("name") for q in sk.get("questions", []): q_id = q.get("id") field_rating = f"tplq_{q_id}_rating" field_comment = f"tplq_{q_id}_comment" rating_raw = request.form.get(field_rating) comment = request.form.get(field_comment) rating = int(rating_raw) if rating_raw and rating_raw.isdigit() else None # If neither rating nor comment given, skip if rating is None and not (comment and comment.strip()): continue db.session.add( InterviewFeedbackItem( feedback_id=fb.id, question_text=q.get("text"), rating=rating, comment=comment, skill_name=skill_name, ) ) # 2) Save ADDITIONAL / CUSTOM questions (existing logic) index = 1 while True: q_text = request.form.get(f"item_question_{index}") if not q_text: break rating_raw = request.form.get(f"item_rating_{index}") comment = request.form.get(f"item_comment_{index}") skill_name = request.form.get(f"item_skill_{index}") rating = int(rating_raw) if rating_raw and rating_raw.isdigit() else None db.session.add( InterviewFeedbackItem( feedback_id=fb.id, question_text=q_text, rating=rating, comment=comment, skill_name=skill_name, ) ) index += 1 interview.status = "completed" db.session.commit() flash("Feedback submitted.", "success") return redirect(url_for("recruitment.view_application", app_id=interview.application_id)) # GET return render_template( "recruitment/interview_feedback.html", interview=interview, req=req, tpl=tpl, structure=structure, ) @recruitment_bp.route("/applications") @login_required def applications_list(): """ Applications list: - HR/Admin: see all applications - Others: see applications for requisitions raised by them OR applications where they are an interviewer in any round. """ emp = _get_current_employee() if not emp: flash("Employee profile not linked.", "danger") return redirect(url_for("employee.dashboard")) base_q = ( Application.query .join(Requisition, Application.requisition_id == Requisition.id) .outerjoin(Interview, Interview.application_id == Application.id) ) if _is_hr_or_admin(emp): apps = ( base_q .order_by(Application.id.desc()) # <-- changed .distinct() .all() ) else: apps = ( base_q .filter( or_( Requisition.raised_by_id == emp.id, Interview.interviewer_id == emp.id, ) ) .order_by(Application.id.desc()) # <-- changed .distinct() .all() ) return render_template("recruitment/applications_list.html", applications=apps) @recruitment_bp.route("/interviewer/my-interviews") @login_required def my_interviews(): """ Show all interviews where current employee is the interviewer. """ emp = _get_current_employee() if not emp: flash("Employee profile not linked.", "danger") return redirect(url_for("employee.dashboard")) interviews = ( Interview.query .filter_by(interviewer_id=emp.id) .order_by(Interview.scheduled_at.desc()) .all() ) return render_template("recruitment/my_interviews.html", interviews=interviews) @recruitment_bp.route("/requisitions//stages", methods=["GET", "POST"]) @login_required def manage_requisition_stages(req_id): """ Manage interview stages for a requisition. HR/Admin only. """ emp = _get_current_employee() if not _is_hr_or_admin(emp): flash("Only HR/Admin can manage interview stages.", "danger") return redirect(url_for("employee.dashboard")) req = Requisition.query.get_or_404(req_id) if request.method == "POST": action = request.form.get("action") stage_id = request.form.get("stage_id", type=int) if action == "save": name = (request.form.get("name") or "").strip() order = request.form.get("order", type=int) is_final = bool(request.form.get("is_final")) if not name: flash("Stage name is required.", "danger") return redirect(url_for("recruitment.manage_requisition_stages", req_id=req.id)) # If no explicit order provided, place at end if order is None: last = ( RequisitionStage.query .filter_by(requisition_id=req.id) .order_by(RequisitionStage.order.desc()) .first() ) order = (last.order or 0) + 1 if last else 1 if stage_id: # Update existing stage = RequisitionStage.query.filter_by( id=stage_id, requisition_id=req.id, ).first_or_404() stage.name = name stage.order = order stage.is_final = is_final flash("Stage updated.", "success") else: # Create new stage = RequisitionStage( requisition_id=req.id, name=name, order=order, is_final=is_final, ) db.session.add(stage) flash("Stage added.", "success") db.session.commit() return redirect(url_for("recruitment.manage_requisition_stages", req_id=req.id)) elif action == "delete" and stage_id: stage = RequisitionStage.query.filter_by( id=stage_id, requisition_id=req.id, ).first() if stage: db.session.delete(stage) db.session.commit() flash("Stage deleted.", "success") return redirect(url_for("recruitment.manage_requisition_stages", req_id=req.id)) # GET stages = ( RequisitionStage.query .filter_by(requisition_id=req.id) .order_by(RequisitionStage.order) .all() ) edit_stage_id = request.args.get("stage_id", type=int) edit_stage = None if edit_stage_id: edit_stage = RequisitionStage.query.filter_by( id=edit_stage_id, requisition_id=req.id, ).first() return render_template( "recruitment/requisition_stages.html", req=req, stages=stages, edit_stage=edit_stage, ) @recruitment_bp.route("/applications//decide", methods=["POST"]) @login_required def decide_application(app_id): emp = _get_current_employee() if not emp: flash("Employee profile not linked.", "danger") return redirect(url_for("employee.dashboard")) if not _is_hr_or_admin(emp): flash("Only HR / Admin can finalize candidates.", "danger") return redirect(url_for("recruitment.view_application", app_id=app_id)) app = Application.query.get_or_404(app_id) req = app.requisition decision = request.form.get("decision") # "hire" or "reject" remarks = (request.form.get("decision_remarks") or "").strip() if decision not in ("hire", "reject"): flash("Invalid decision.", "danger") return redirect(url_for("recruitment.view_application", app_id=app.id)) # ---- Update application status ---- if decision == "hire": app.status = "hired" else: app.status = "rejected" # (Optional) store remarks somewhere if you have a field – otherwise ignore # Example if you add a column later: # app.decision_remarks = remarks # app.decision_by_id = emp.id # app.decision_on = datetime.utcnow() # ---- Auto-close requisition if no other active candidates ---- # Active = applied / in_process from sqlalchemy import or_ open_apps_count = ( Application.query .filter( Application.requisition_id == req.id, Application.id != app.id, Application.status.in_(["applied", "in_process"]) ) .count() ) if decision == "hire" and open_apps_count == 0: # All other candidates are already decided; close requisition req.status = "closed" db.session.commit() if decision == "hire": msg = "Candidate marked as HIRED. Requisition closed." if open_apps_count == 0 \ else "Candidate marked as HIRED." else: msg = "Candidate marked as REJECTED." flash(msg, "success") return redirect(url_for("recruitment.view_application", app_id=app.id)) @recruitment_bp.route("/requisitions//finalize", methods=["POST"]) @login_required def finalize_candidate(req_id): emp = _get_current_employee() if not _is_hr_or_admin(emp): flash("Only HR / Admin can finalize a candidate.", "danger") return redirect(url_for("recruitment.requisition_view", req_id=req_id)) req = Requisition.query.get_or_404(req_id) candidate_app_id = request.form.get("candidate_id", type=int) remarks = request.form.get("remarks") if not candidate_app_id: flash("Please select a candidate.", "danger") return redirect(url_for("recruitment.requisition_view", req_id=req.id)) app = ( Application.query .filter_by(id=candidate_app_id, requisition_id=req.id) .first() ) if not app: flash("Invalid candidate selection.", "danger") return redirect(url_for("recruitment.requisition_view", req_id=req.id)) # Mark this application as selected app.status = "selected" # Mark all other applications for this requisition as rejected ( Application.query .filter( Application.requisition_id == req.id, Application.id != app.id, ) .update({Application.status: "rejected"}) ) # Close the requisition req.status = "closed" # Safely append an HR note only if `remarks` column actually exists note = f"[HR Finalised] Candidate: {app.candidate.full_name if app.candidate else app.id}" if remarks: note += f"\nRemarks: {remarks}" if hasattr(req, "remarks"): existing = getattr(req, "remarks") or "" if existing: existing = existing + "\n\n" + note else: existing = note setattr(req, "remarks", existing) # else: no remarks column on this model – skip storing note db.session.commit() flash("Final candidate selected and requisition closed.", "success") return redirect(url_for("recruitment.requisition_view", req_id=req.id)) from datetime import datetime, date, timedelta from sqlalchemy import func, or_ # ... @recruitment_bp.route("/dashboard") @login_required def recruitment_dashboard(): emp = _get_current_employee() if not emp: flash("Employee profile not linked.", "danger") return redirect(url_for("employee.dashboard")) is_hr_admin = _is_hr_or_admin(emp) # ---------- Base query scopes (HR/Admin = full, others = own) ---------- base_req_q = Requisition.query base_app_q = ( Application.query .join(Requisition, Application.requisition_id == Requisition.id) ) base_int_q = ( Interview.query .join(Application, Interview.application_id == Application.id) .join(Requisition, Application.requisition_id == Requisition.id) ) if not is_hr_admin: visibility_filter = or_( Requisition.raised_by_id == emp.id, Requisition.manager_id == emp.id, Requisition.hr_owner_id == emp.id, ) base_req_q = base_req_q.filter(visibility_filter) base_app_q = base_app_q.filter(visibility_filter) base_int_q = base_int_q.filter(visibility_filter) today = date.today() start_today = datetime.combine(today, datetime.min.time()) end_today = datetime.combine(today, datetime.max.time()) last_30 = today - timedelta(days=30) # ---------- Metrics ---------- open_statuses = ["approved", "open", "pending_hr", "pending_manager"] open_reqs_count = ( base_req_q .filter(Requisition.status.in_(open_statuses)) .count() ) on_hold_count = base_req_q.filter(Requisition.status == "on_hold").count() closed_recent_count = ( base_req_q .filter( Requisition.status == "closed", Requisition.created_on >= last_30 ) .count() ) candidates_in_pipeline = ( base_app_q .filter(Application.status.in_(["applied", "in_process"])) .count() ) interviews_today = ( base_int_q .filter( Interview.scheduled_at >= start_today, Interview.scheduled_at <= end_today ) .count() ) metrics = { "open_reqs": open_reqs_count, "pipeline": candidates_in_pipeline, "interviews_today": interviews_today, "closed_recent": closed_recent_count, "on_hold": on_hold_count, } # ---------- Open requisitions list ---------- open_reqs = ( base_req_q .filter(Requisition.status.in_(open_statuses)) .order_by(Requisition.created_on.desc()) .all() ) # Pre-compute application counts per requisition req_ids = [r.id for r in open_reqs] or [0] apps_counts = dict( db.session.query( Application.requisition_id, func.count(Application.id) ) .filter(Application.requisition_id.in_(req_ids)) .group_by(Application.requisition_id) .all() ) # ---------- Upcoming interviews (next 7 days) ---------- in_7_days = today + timedelta(days=7) upcoming_interviews = ( base_int_q .filter( Interview.scheduled_at >= datetime.combine(today, datetime.min.time()), Interview.scheduled_at <= datetime.combine(in_7_days, datetime.max.time()), Interview.status == "scheduled", ) .order_by(Interview.scheduled_at.asc()) .limit(10) .all() ) # ---------- Recent hires (selected candidates on closed reqs) ---------- recent_hires = ( base_app_q .filter( Application.status == "selected", Requisition.status == "closed", Requisition.created_on >= last_30, ) .order_by(Requisition.created_on.desc()) .limit(10) .all() ) return render_template( "recruitment/dashboard.html", metrics=metrics, open_reqs=open_reqs, apps_counts=apps_counts, upcoming_interviews=upcoming_interviews, recent_hires=recent_hires, is_hr_admin=is_hr_admin, ) from werkzeug.utils import secure_filename import os, re from openpyxl import load_workbook # ... existing imports already present above ... @recruitment_bp.route("/templates/import-excel", methods=["GET", "POST"]) @login_required def template_import_excel(): """ Import a recruitment template (KRA / KPI / Skills / Questions) from an Excel file in the format of Requisition.xlsx (BDE sheet). """ emp = _get_current_employee() if not _is_hr_or_admin(emp): flash("You are not allowed to import recruitment templates.", "danger") return redirect(url_for("employee.dashboard")) designations = Designation.query.order_by(Designation.title).all() if request.method == "POST": designation_id = request.form.get("designation_id", type=int) file = request.files.get("excel_file") if not designation_id: flash("Please select a designation.", "danger") return redirect(request.url) if not file or not file.filename: flash("Please upload an Excel file.", "danger") return redirect(request.url) ext = os.path.splitext(file.filename)[1].lower() if ext not in (".xlsx", ".xls"): flash("Please upload a .xlsx or .xls file.", "danger") return redirect(request.url) # --- Save to a temp/import folder --- upload_folder = os.path.join(current_app.instance_path, "imports", "recruitment") os.makedirs(upload_folder, exist_ok=True) safe_name = secure_filename(file.filename) tmp_path = os.path.join(upload_folder, safe_name) file.save(tmp_path) # --- Parse workbook (first sheet, same format as Requisition.xlsx / BDE) --- try: wb = load_workbook(tmp_path, data_only=True) except Exception as e: current_app.logger.exception("Failed to read excel for template import") flash(f"Could not read Excel file: {e}", "danger") return redirect(request.url) if not wb.sheetnames: flash("Excel file has no sheets.", "danger") return redirect(request.url) ws = wb[wb.sheetnames[0]] # BDE in your sample; we just take first # Row 2 contains the basic template info in your sample: # A2: MINIMUM EDUCATIONAL QUALIFICATION # B2: MINIMUM YEARS OF EXPERIENCE & CORE SKILLS (text) # C2: JOB PURPOSE & ROLE SUMMARY # D2: Age Bracket min_education = ws["A2"].value exp_cell = ws["B2"].value or "" job_purpose = ws["C2"].value age_bracket = ws["D2"].value # Try to extract a numeric min_experience_years from the text (e.g. "1–2 years...") min_experience_years = None if isinstance(exp_cell, (int, float)): min_experience_years = exp_cell elif isinstance(exp_cell, str): m = re.search(r"(\d+)", exp_cell) if m: try: min_experience_years = int(m.group(1)) except Exception: min_experience_years = None # --- Create the RecruitmentTemplate row --- tpl = RecruitmentTemplate( designation_id=designation_id, min_education=min_education, min_experience_years=min_experience_years, age_bracket=age_bracket, salary_bracket=None, # not present in sheet accountability=None, # not present separately in sheet job_description=job_purpose, # big text from C2 ) db.session.add(tpl) db.session.flush() # tpl.id # --- Parse KRAs / KPIs / Skills / Questions from rows 2..N --- # Columns (from your header): # 5: KRA # 6: KRA Description # 7: KPI / Metric # 8: KPI Target # 9: Weightage (%) # 10: Basic Skill(s) (3–5 core attributes) # 11: Skill Evaluation Questions (for Interviewer) skills_map = {} # name -> {"name": str, "basic": str|None, "questions": [str,..]} max_row = ws.max_row for r in range(2, max_row + 1): kra_title = ws.cell(row=r, column=5).value kra_desc = ws.cell(row=r, column=6).value kpi_title = ws.cell(row=r, column=7).value kpi_target = ws.cell(row=r, column=8).value kpi_weight = ws.cell(row=r, column=9).value # ---- KRA + KPI ---- kra_obj = None if kra_title: kra_obj = TemplateKRA( template_id=tpl.id, name=str(kra_title).strip(), description=(kra_desc or "").strip() or None, weightage=kpi_weight if isinstance(kpi_weight, (int, float)) else None, ) db.session.add(kra_obj) db.session.flush() # kra_obj.id if kpi_title: db.session.add( TemplateKPI( template_id=tpl.id, kra_id=kra_obj.id, name=str(kpi_title).strip(), target=str(kpi_target).strip() if kpi_target else None, weightage=kpi_weight if isinstance(kpi_weight, (int, float)) else None, ) ) # ---- Skills + Questions ---- skill_name = ws.cell(row=r, column=10).value questions_raw = ws.cell(row=r, column=11).value if skill_name and questions_raw: key = str(skill_name).strip() sk = skills_map.setdefault( key, {"name": key, "basic": None, "questions": []} ) # Split questions on newlines (your file uses Q1...Q2... etc) for line in str(questions_raw).splitlines(): q = line.strip() if not q: continue # remove leading "Q1:", "Q2 -" etc q = re.sub(r"^Q\d+\s*[:\-]?\s*", "", q).strip() if q and q not in sk["questions"]: sk["questions"].append(q) db.session.flush() # --- Persist skills and their questions --- for sk in skills_map.values(): skill_row = TemplateSkill( template_id=tpl.id, name=sk["name"], basic_requirement=sk["basic"], ) db.session.add(skill_row) db.session.flush() # skill_row.id for q_text in sk["questions"]: db.session.add( TemplateSkillQuestion( skill_id=skill_row.id, question_text=q_text, ) ) db.session.commit() flash("Template imported from Excel. You can now review and adjust it.", "success") return redirect(url_for("recruitment.template_edit", template_id=tpl.id)) # GET return render_template( "recruitment/template_import_excel.html", designations=designations, ) def _can_manage_requisition(emp: Employee | None, req: Requisition) -> bool: if not emp: return False return ( _is_hr_or_admin(emp) or req.raised_by_id == emp.id or req.manager_id == emp.id or req.hr_owner_id == emp.id ) @recruitment_bp.route("/requisitions//hold", methods=["POST"]) @login_required def requisition_hold(req_id): emp = _get_current_employee() req = Requisition.query.get_or_404(req_id) if not _can_manage_requisition(emp, req): flash("You are not allowed to hold this requisition.", "danger") return redirect(url_for("recruitment.requisition_view", req_id=req.id)) # don't allow holding closed/rejected if (req.status or "").lower() in ("closed", "rejected"): flash("Closed/Rejected requisitions cannot be put on hold.", "warning") return redirect(url_for("recruitment.requisition_view", req_id=req.id)) # already on hold if (req.status or "").lower() == "on_hold": flash("This requisition is already on hold.", "info") return redirect(url_for("recruitment.requisition_view", req_id=req.id)) reason = (request.form.get("hold_reason") or "").strip() if not reason: flash("Please enter a reason to put requisition on hold.", "danger") return redirect(url_for("recruitment.requisition_view", req_id=req.id)) req.hold_prev_status = req.status req.status = "on_hold" req.hold_reason = reason req.hold_on = datetime.utcnow() req.hold_by_id = emp.id if emp else None db.session.commit() flash("Requisition put on hold.", "success") return redirect(url_for("recruitment.requisition_view", req_id=req.id)) @recruitment_bp.route("/requisitions//resume", methods=["POST"]) @login_required def requisition_resume(req_id): emp = _get_current_employee() req = Requisition.query.get_or_404(req_id) if not _can_manage_requisition(emp, req): flash("You are not allowed to resume this requisition.", "danger") return redirect(url_for("recruitment.requisition_view", req_id=req.id)) if (req.status or "").lower() != "on_hold": flash("This requisition is not on hold.", "info") return redirect(url_for("recruitment.requisition_view", req_id=req.id)) prev = (req.hold_prev_status or "").strip() or "approved" # safe fallback req.status = prev req.hold_prev_status = None req.hold_reason = None req.hold_on = None req.hold_by_id = None db.session.commit() flash("Requisition resumed.", "success") return redirect(url_for("recruitment.requisition_view", req_id=req.id))