458 lines
18 KiB
Python
458 lines
18 KiB
Python
import uuid
|
||
from fastapi import HTTPException
|
||
from datetime import timezone
|
||
|
||
def _as_uuid(value):
|
||
if value in (None, ""):
|
||
return None
|
||
try:
|
||
return uuid.UUID(str(value))
|
||
except (TypeError, ValueError):
|
||
return None
|
||
|
||
|
||
def _user_id(current_user):
|
||
if not current_user or not current_user.get("id"):
|
||
raise HTTPException(status_code=401, detail="Not authenticated")
|
||
uid = _as_uuid(current_user["id"])
|
||
if uid is None:
|
||
raise HTTPException(status_code=401, detail="Invalid user id")
|
||
return uid
|
||
|
||
|
||
def _aware(value):
|
||
if value is not None and getattr(value, "tzinfo", None) is None:
|
||
return value.replace(tzinfo=timezone.utc)
|
||
return value
|
||
|
||
|
||
def _stage_value(status) -> str:
|
||
return str(getattr(status, "value", status) or "").upper()
|
||
|
||
|
||
def _recommendation(form_type, value):
|
||
definition = FORM_DEFINITIONS.get(form_type) or {}
|
||
if not definition.get("has_recommendation"):
|
||
return None
|
||
if value in (None, ""):
|
||
return None
|
||
value = str(value).strip()
|
||
if value not in RECOMMENDATIONS:
|
||
raise HTTPException(
|
||
status_code=422,
|
||
detail=f"recommendation must be one of {', '.join(RECOMMENDATIONS)}",
|
||
)
|
||
return value
|
||
|
||
|
||
def _score_sections(form_type, sections):
|
||
try:
|
||
return normalize_sections(form_type, sections)
|
||
except ValueError as exc:
|
||
raise HTTPException(status_code=422, detail=str(exc))
|
||
|
||
|
||
def _score_fields(form_type, fields):
|
||
try:
|
||
return normalize_fields(form_type, fields)
|
||
except ValueError as exc:
|
||
raise HTTPException(status_code=422, detail=str(exc))
|
||
|
||
|
||
FORM_TYPES = ("requisition", "interview_analysis", "cultural_fit")
|
||
|
||
RATING_POINTS = (25, 50, 75, 100)
|
||
# Paper ticks used to be 1–4; coerce those to the matching percentage.
|
||
_LEGACY_TICK = {1: 25, 2: 50, 3: 75, 4: 100}
|
||
RATING_LABELS = {
|
||
25: "Below Average (25%)",
|
||
50: "Average (50%)",
|
||
75: "Good (75%)",
|
||
100: "Excellent (100%)",
|
||
}
|
||
RATING_SCALE_NOTE = (
|
||
"Rating Scale: Below Average = 25% | Average = 50% | Good = 75% | Excellent = 100%. "
|
||
"Tick the box that applies for each criterion."
|
||
)
|
||
|
||
RECOMMENDATIONS = (
|
||
"selected",
|
||
"hold",
|
||
"next_round",
|
||
"not_selected",
|
||
"other_position",
|
||
"offer_placement",
|
||
)
|
||
RECOMMENDATION_LABELS = {
|
||
"selected": "Selected",
|
||
"hold": "Hold for now",
|
||
"next_round": "Shortlist for next round",
|
||
"not_selected": "Not selected",
|
||
"other_position": "Consider for other position",
|
||
"offer_placement": "Offer Placement",
|
||
}
|
||
|
||
# INTERVIEW onward; APPROVED is the legacy spelling the UI maps to Hired.
|
||
FORM_READY_STATUSES = ("INTERVIEW", "OFFER", "HIRED", "APPROVED")
|
||
|
||
EMPLOYMENT_TYPES = ("permanent", "temporary", "contract", "internee")
|
||
EMPLOYMENT_TYPE_LABELS = {
|
||
"permanent": "Permanent",
|
||
"temporary": "Temporary",
|
||
"contract": "Contract",
|
||
"internee": "Internee",
|
||
}
|
||
|
||
_EVALUATION_HEADER_FIELDS = [
|
||
{"key": "interviewer_name", "label": "Interviewer Name", "kind": "text"},
|
||
{"key": "department", "label": "Department/Division", "kind": "text"},
|
||
{"key": "position_title", "label": "Position Interviewed For", "kind": "text"},
|
||
]
|
||
|
||
_EVALUATION_FOOTER_FIELDS = [
|
||
{"key": "strengths", "label": "Key Strengths", "kind": "textarea"},
|
||
{"key": "concerns", "label": "Main Concerns or Gaps", "kind": "textarea"},
|
||
{
|
||
"key": "overall_observation",
|
||
"label": "Overall Observation of the Candidate",
|
||
"kind": "textarea",
|
||
},
|
||
]
|
||
|
||
FORM_DEFINITIONS = {
|
||
"interview_analysis": {
|
||
"title": "Interview Analysis",
|
||
"source": "Annexure E - Interview Evaluation Form",
|
||
"scale_note": RATING_SCALE_NOTE,
|
||
"sections": [
|
||
{
|
||
"key": "technical",
|
||
"title": "TECHNICAL COMPETENCY ASSESSMENT",
|
||
"average_label": "TECHNICAL SECTION AVERAGE",
|
||
"criteria": [
|
||
{"key": "core_job_knowledge", "label": "Core Job Knowledge & Domain Expertise"},
|
||
{"key": "relevant_experience", "label": "Depth of Relevant Experience"},
|
||
{"key": "problem_solving", "label": "Problem Solving"},
|
||
{"key": "analytical_reasoning", "label": "Analytical Reasoning"},
|
||
{"key": "tools_proficiency", "label": "Technical Tools & Systems Proficiency"},
|
||
{"key": "quality_of_work", "label": "Quality of Work & Attention to Detail"},
|
||
],
|
||
},
|
||
{
|
||
"key": "behavioral",
|
||
"title": "BEHAVIORAL COMPETENCY ASSESSMENT",
|
||
"average_label": "BEHAVIORAL SECTION AVERAGE",
|
||
"criteria": [
|
||
{"key": "communication", "label": "Communication & Clarity of Expression"},
|
||
{"key": "active_listening", "label": "Active Listening & Comprehension"},
|
||
{"key": "ownership", "label": "Ownership & Accountability"},
|
||
{"key": "resilience", "label": "Resilience Under Pressure"},
|
||
{"key": "learning_agility", "label": "Learning Agility & Coachability"},
|
||
],
|
||
},
|
||
],
|
||
"fields": (
|
||
_EVALUATION_HEADER_FIELDS
|
||
+ [
|
||
{"key": "summary", "label": "BRIEF SUMMARY OF THE CANDIDATE", "kind": "textarea"},
|
||
{"key": "technical_note", "label": "Technical Competency — Notes", "kind": "text"},
|
||
{"key": "behavioral_note", "label": "Behavioral Competency — Notes", "kind": "text"},
|
||
]
|
||
+ _EVALUATION_FOOTER_FIELDS
|
||
),
|
||
"has_recommendation": True,
|
||
},
|
||
# form_type/section/field keys below stay "cultural_fit"/"cultural"/"cultural_note" —
|
||
# renamed labels only. Titles and criterion labels are denormalized into every
|
||
# saved row at write time (see module docstring), so historical rows keep the
|
||
# "Cultural Fit" wording they were saved under while new rows pick up the fuller
|
||
# revision 2 "HR Evaluation" section below; the key stays stable so old rows keep
|
||
# validating and combined_summary()'s "cultural" lookup keeps matching both.
|
||
"cultural_fit": {
|
||
"title": "HR Evaluation",
|
||
"source": "Annexure E - Interview Evaluation Form",
|
||
"scale_note": RATING_SCALE_NOTE,
|
||
"sections": [
|
||
{
|
||
"key": "cultural",
|
||
"title": "HR EVALUATION",
|
||
"average_label": "HR EVALUATION SECTION",
|
||
"criteria": [
|
||
{"key": "basic_jd_requirement", "label": "Basic JD requirement"},
|
||
{"key": "company_values", "label": "Alignment with Company Culture"},
|
||
{"key": "professionalism", "label": "Professionalism & Integrity"},
|
||
{"key": "collaboration", "label": "Collaboration & Team Orientation"},
|
||
{"key": "adaptability", "label": "Adaptability"},
|
||
{"key": "agility", "label": "Agility"},
|
||
{"key": "work_ethic", "label": "Work Ethics"},
|
||
{"key": "communication_articulation", "label": "Communication & Articulation"},
|
||
{"key": "problem_solving_orientation", "label": "Problem Solving & Solution Orientation"},
|
||
{"key": "critical_thinking", "label": "Critical Thinking & Analytical Capability"},
|
||
{"key": "initiative", "label": "Initiative & Proactiveness"},
|
||
{"key": "decision_making", "label": "Decision Making"},
|
||
{"key": "leadership", "label": "Leadership"},
|
||
],
|
||
},
|
||
],
|
||
"fields": (
|
||
_EVALUATION_HEADER_FIELDS
|
||
+ [{"key": "cultural_note", "label": "HR Evaluation — Notes", "kind": "text"}]
|
||
+ _EVALUATION_FOOTER_FIELDS
|
||
),
|
||
"has_recommendation": True,
|
||
},
|
||
"requisition": {
|
||
"title": "Employee Requisition",
|
||
"source": "Annexure A - Employee Requisition Form",
|
||
"header_note": "To: Human Resource Department",
|
||
"sections": [],
|
||
"fields": [
|
||
{"key": "department", "label": "From: (Dept.)", "kind": "text"},
|
||
{"key": "job_title", "label": "Job Title", "kind": "text"},
|
||
{"key": "date_needed", "label": "Date Needed", "kind": "date"},
|
||
{
|
||
"key": "employment_type",
|
||
"label": "Permanent / Temporary / Contract / Internee",
|
||
"kind": "select",
|
||
"options": list(EMPLOYMENT_TYPES),
|
||
},
|
||
{"key": "period_from", "label": "If not permanent, specify the period — From", "kind": "date"},
|
||
{"key": "period_to", "label": "If not permanent, specify the period — To", "kind": "date"},
|
||
{
|
||
"key": "jd_available",
|
||
"label": (
|
||
"JD Available (JD is mandatory, TA team will not proceed with "
|
||
"sourcing until JD is provided)"
|
||
),
|
||
"kind": "bool",
|
||
},
|
||
{"key": "is_replacement", "label": "IF A REPLACEMENT, COMPLETE THE FOLLOWING", "kind": "bool"},
|
||
{"key": "replacement_employee", "label": "Employee to be replaced", "kind": "text"},
|
||
{"key": "replacement_grade", "label": "Grade", "kind": "text"},
|
||
{"key": "replacement_job_title", "label": "Job Title (replaced employee)", "kind": "text"},
|
||
{"key": "replacement_date_separated", "label": "Date Separated", "kind": "date"},
|
||
{
|
||
"key": "headcount_justification",
|
||
"label": "IN CASE OF NEW/ADDITIONAL HEADCOUNT PLEASE PROVIDE JUSTIFICATION",
|
||
"kind": "textarea",
|
||
},
|
||
{"key": "proposed_budget", "label": "PROPOSE BUDGET", "kind": "text"},
|
||
{"key": "recommended_grade", "label": "RECOMMENDED GRADE", "kind": "text"},
|
||
{"key": "internal_recommendation", "label": "INCASE OF INTERNAL RECOMMENDATE", "kind": "bool"},
|
||
{"key": "recommended_employee_name", "label": "EMPLOYEE NAME", "kind": "text"},
|
||
{"key": "recommended_employee_department", "label": "EMPLOYEE DEPARTMENT", "kind": "text"},
|
||
{"key": "entity", "label": "Entity", "kind": "text"},
|
||
{"key": "initiated_by", "label": "Initiated By — Name", "kind": "text"},
|
||
{"key": "initiated_date", "label": "Initiated By — Date", "kind": "date"},
|
||
{"key": "recommended_by", "label": "Recommended By — Name (Director)", "kind": "text"},
|
||
{"key": "recommended_date", "label": "Recommended By — Date", "kind": "date"},
|
||
{"key": "approved_by", "label": "Approved By — Name (Director HR)", "kind": "text"},
|
||
{"key": "approved_date", "label": "Approved By — Date", "kind": "date"},
|
||
{"key": "vp_approved_by", "label": "Approved By — Name (VP/SVP)", "kind": "text"},
|
||
{"key": "vp_approved_date", "label": "Approved By — Date (VP/SVP)", "kind": "date"},
|
||
],
|
||
"field_enums": {"employment_type": EMPLOYMENT_TYPES},
|
||
"has_recommendation": False,
|
||
},
|
||
}
|
||
|
||
|
||
def definitions_payload() -> dict:
|
||
"""The response body for GET /forms/definitions."""
|
||
return {
|
||
"form_types": list(FORM_TYPES),
|
||
"forms": FORM_DEFINITIONS,
|
||
"rating_labels": {str(k): v for k, v in RATING_LABELS.items()},
|
||
"rating_points": list(RATING_POINTS),
|
||
"recommendations": list(RECOMMENDATIONS),
|
||
"recommendation_labels": dict(RECOMMENDATION_LABELS),
|
||
"employment_types": list(EMPLOYMENT_TYPES),
|
||
"employment_type_labels": dict(EMPLOYMENT_TYPE_LABELS),
|
||
"form_ready_statuses": list(FORM_READY_STATUSES),
|
||
}
|
||
|
||
|
||
def _coerce_rating(value):
|
||
if value in (None, ""):
|
||
return None
|
||
try:
|
||
number = float(value)
|
||
except (TypeError, ValueError):
|
||
raise ValueError(f"rating must be a number, got {value!r}")
|
||
if number != int(number):
|
||
raise ValueError(f"rating must be a whole number, got {value!r}")
|
||
rating = _LEGACY_TICK.get(int(number), int(number))
|
||
if rating not in RATING_POINTS:
|
||
allowed = ", ".join(str(p) for p in RATING_POINTS)
|
||
raise ValueError(f"rating must be one of {allowed}, got {rating}")
|
||
return rating
|
||
|
||
|
||
def _mean(values, digits=2):
|
||
values = [v for v in values if v is not None]
|
||
if not values:
|
||
return None
|
||
return round(sum(values) / len(values), digits)
|
||
|
||
|
||
def to_percent(score):
|
||
"""Keep derived scores on 0–100.
|
||
|
||
New ticks are 25/50/75/100 and averages are already percentages. Legacy
|
||
1–4 ticks or means (0, 4] convert once via (score / 4) × 100.
|
||
"""
|
||
if score is None:
|
||
return None
|
||
try:
|
||
number = float(score)
|
||
except (TypeError, ValueError):
|
||
return None
|
||
if 0 < number <= 4:
|
||
return round((number / 4) * 100, 2)
|
||
return round(number, 2)
|
||
|
||
|
||
def normalize_sections(form_type: str, sections):
|
||
"""Validate submitted rated sections against the form definition and
|
||
recompute all derived numbers. Returns (normalized_sections, overall_score).
|
||
|
||
Every definition section is emitted in definition order with denormalized
|
||
labels; submitted per-criterion ratings are merged in; client-sent averages
|
||
are discarded and recomputed. Criterion ticks are 25/50/75/100. A section
|
||
average is the mean of those percentages; the overall score is the mean of
|
||
the section averages. Legacy 1–4 ticks are coerced to the matching percent
|
||
before averaging. Raises ValueError on unknown section/criterion keys or
|
||
ratings outside the scale (422 material).
|
||
"""
|
||
definition = FORM_DEFINITIONS.get(form_type)
|
||
if definition is None:
|
||
raise ValueError(f"unknown form_type {form_type!r}")
|
||
if not definition["sections"]:
|
||
return None, None
|
||
if sections is None:
|
||
sections = []
|
||
if not isinstance(sections, list):
|
||
raise ValueError("sections must be a list")
|
||
|
||
known_sections = {s["key"]: s for s in definition["sections"]}
|
||
submitted = {}
|
||
for entry in sections:
|
||
if not isinstance(entry, dict):
|
||
raise ValueError("each section must be an object")
|
||
key = entry.get("key")
|
||
if key not in known_sections:
|
||
raise ValueError(f"unknown section {key!r} for {form_type}")
|
||
criteria = entry.get("criteria") or []
|
||
if not isinstance(criteria, list):
|
||
raise ValueError("section criteria must be a list")
|
||
known_criteria = {c["key"] for c in known_sections[key]["criteria"]}
|
||
ratings = {}
|
||
for criterion in criteria:
|
||
if not isinstance(criterion, dict):
|
||
raise ValueError("each criterion must be an object")
|
||
ckey = criterion.get("key")
|
||
if ckey not in known_criteria:
|
||
raise ValueError(f"unknown criterion {ckey!r} in section {key!r}")
|
||
ratings[ckey] = _coerce_rating(criterion.get("rating"))
|
||
submitted[key] = ratings
|
||
|
||
normalized = []
|
||
section_averages = []
|
||
for section_def in definition["sections"]:
|
||
ratings = submitted.get(section_def["key"], {})
|
||
criteria = [
|
||
{
|
||
"key": c["key"],
|
||
"label": c["label"],
|
||
"rating": ratings.get(c["key"]),
|
||
}
|
||
for c in section_def["criteria"]
|
||
]
|
||
average = to_percent(_mean([c["rating"] for c in criteria]))
|
||
if average is not None:
|
||
section_averages.append(average)
|
||
normalized.append(
|
||
{
|
||
"key": section_def["key"],
|
||
"title": section_def["title"],
|
||
"criteria": criteria,
|
||
"average": average,
|
||
}
|
||
)
|
||
return normalized, _mean(section_averages)
|
||
|
||
|
||
def normalize_fields(form_type: str, fields):
|
||
"""Keep only the definition's field keys, validate enums, coerce booleans."""
|
||
definition = FORM_DEFINITIONS.get(form_type)
|
||
if definition is None:
|
||
raise ValueError(f"unknown form_type {form_type!r}")
|
||
if fields is None:
|
||
return {}
|
||
if not isinstance(fields, dict):
|
||
raise ValueError("fields must be an object")
|
||
|
||
known = {f["key"]: f for f in definition["fields"]}
|
||
enums = definition.get("field_enums", {})
|
||
normalized = {}
|
||
for key, value in fields.items():
|
||
spec = known.get(key)
|
||
if spec is None:
|
||
continue
|
||
if value in (None, ""):
|
||
normalized[key] = None
|
||
continue
|
||
if key in enums:
|
||
value = str(value).strip().lower()
|
||
if value not in enums[key]:
|
||
raise ValueError(f"{key} must be one of {', '.join(enums[key])}")
|
||
elif spec["kind"] == "bool":
|
||
if isinstance(value, str):
|
||
value = value.strip().lower() in ("true", "yes", "1", "on")
|
||
else:
|
||
value = bool(value)
|
||
else:
|
||
value = str(value).strip() or None
|
||
normalized[key] = value
|
||
return normalized
|
||
|
||
|
||
def combined_summary(rows):
|
||
"""Annexure E's OVERALL SCORE SUMMARY across the two evaluation forms.
|
||
|
||
`rows` are candidate_forms records (attribute access: form_type, created_at,
|
||
sections). The latest interview_analysis row supplies the technical and
|
||
behavioral averages, the latest cultural_fit row the "cultural" section
|
||
average — cultural_fit's own section carries the fuller HR Evaluation
|
||
criteria as of revision 2, but the key stays "cultural" so this lookup
|
||
(and the `cultural_avg` key below) don't need to change with it.
|
||
The combined overall (mean of the three section averages, 2 dp, already
|
||
ranged onto 0–100) appears only once all three exist. Returns None when
|
||
neither evaluation exists. Legacy 1–4 section averages are converted
|
||
through to_percent so mixed old/new rows stay comparable.
|
||
"""
|
||
latest = {}
|
||
for row in rows:
|
||
if row.form_type not in ("interview_analysis", "cultural_fit"):
|
||
continue
|
||
current = latest.get(row.form_type)
|
||
if current is None or (row.created_at and current.created_at and row.created_at > current.created_at):
|
||
latest[row.form_type] = row
|
||
if not latest:
|
||
return None
|
||
|
||
averages = {"technical": None, "behavioral": None, "cultural": None}
|
||
for row in latest.values():
|
||
for section in row.sections or []:
|
||
key = section.get("key")
|
||
if key in averages:
|
||
averages[key] = to_percent(section.get("average"))
|
||
|
||
complete = all(v is not None for v in averages.values())
|
||
return {
|
||
"technical_avg": averages["technical"],
|
||
"behavioral_avg": averages["behavioral"],
|
||
"cultural_avg": averages["cultural"],
|
||
"combined_overall": _mean(list(averages.values())) if complete else None,
|
||
}
|