339 lines
13 KiB
Python
339 lines
13 KiB
Python
import logging
|
|
import uuid
|
|
from datetime import timezone
|
|
|
|
from fastapi import HTTPException
|
|
from sqlalchemy import select
|
|
from sqlalchemy.ext.asyncio import AsyncSession
|
|
from sqlalchemy.orm import selectinload
|
|
|
|
from candidate_forms.models import CandidateForms, _now
|
|
from candidate_forms.plugins import (
|
|
FORM_READY_STATUSES,
|
|
FORM_TYPES,
|
|
RECOMMENDATIONS,
|
|
combined_summary,
|
|
normalize_fields,
|
|
normalize_sections,
|
|
)
|
|
from candidate_forms.serializers import serialize_form
|
|
from inbox.models import Inbox
|
|
from job.candidate.models import Manual_UPLOAD_CANDIDATE
|
|
from job.history.enums import HistoryEvent
|
|
from job.history.views import HistoryRecorder
|
|
from job.job_post.models import JobPosts
|
|
from users.models import Users
|
|
|
|
logger = logging.getLogger("candidate_forms")
|
|
|
|
|
|
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()
|
|
|
|
|
|
class CandidateForm:
|
|
def __init__(self, session: AsyncSession):
|
|
self.session = session
|
|
|
|
async def _validate_link(self, payload):
|
|
"""Exactly one of inbox_id / manual_upload_candidate_id; both rows must
|
|
exist. Returns (inbox_id, manual_id, job_post_id, current_stage)."""
|
|
inbox_id = payload.get("inbox_id")
|
|
manual_id = _as_uuid(payload.get("manual_upload_candidate_id"))
|
|
has_inbox = inbox_id is not None
|
|
has_manual = manual_id is not None
|
|
if has_inbox == has_manual:
|
|
raise HTTPException(
|
|
status_code=422,
|
|
detail="Exactly one of inbox_id or manual_upload_candidate_id is required",
|
|
)
|
|
if has_inbox:
|
|
try:
|
|
inbox_id = int(inbox_id)
|
|
except (TypeError, ValueError):
|
|
raise HTTPException(status_code=422, detail="Invalid inbox_id")
|
|
link = await Inbox.get_inbox_with_message(self.session, inbox_id)
|
|
if link is None:
|
|
raise HTTPException(status_code=404, detail="Inbox record not found")
|
|
stage = _stage_value(
|
|
link.messages.application_status if link.messages is not None else None
|
|
)
|
|
else:
|
|
inbox_id = None
|
|
manual = await Manual_UPLOAD_CANDIDATE.get_by_id(self.session, manual_id)
|
|
if manual is None:
|
|
raise HTTPException(status_code=404, detail="Manual upload candidate not found")
|
|
stage = _stage_value(manual.status)
|
|
|
|
job_post_id = _as_uuid(payload.get("job_post_id"))
|
|
if payload.get("job_post_id") and job_post_id is None:
|
|
raise HTTPException(status_code=422, detail="Invalid job_post_id")
|
|
if job_post_id is not None:
|
|
post = await JobPosts.get_job_post_by_id(self.session, str(job_post_id))
|
|
if not post or post.is_deleted:
|
|
raise HTTPException(status_code=404, detail="Job post not found")
|
|
return inbox_id, manual_id, job_post_id, stage
|
|
|
|
def _normalize_payload(self, form_type, payload):
|
|
"""Shared create/update normalization. Returns the writable fields dict
|
|
for the keys present in `payload`."""
|
|
fields = {}
|
|
if "sections" in payload:
|
|
try:
|
|
sections, overall = normalize_sections(form_type, payload.get("sections"))
|
|
except ValueError as exc:
|
|
raise HTTPException(status_code=422, detail=str(exc))
|
|
fields["sections"] = sections
|
|
fields["overall_score"] = overall
|
|
if "fields" in payload:
|
|
try:
|
|
fields["fields"] = normalize_fields(form_type, payload.get("fields"))
|
|
except ValueError as exc:
|
|
raise HTTPException(status_code=422, detail=str(exc))
|
|
if "recommendation" in payload:
|
|
recommendation = payload.get("recommendation") or None
|
|
if recommendation is not None and recommendation not in RECOMMENDATIONS:
|
|
raise HTTPException(
|
|
status_code=422,
|
|
detail=f"recommendation must be one of {', '.join(RECOMMENDATIONS)}",
|
|
)
|
|
fields["recommendation"] = recommendation
|
|
if "interviewer_id" in payload:
|
|
interviewer_id = _as_uuid(payload.get("interviewer_id"))
|
|
if payload.get("interviewer_id") and interviewer_id is None:
|
|
raise HTTPException(status_code=422, detail="Invalid interviewer_id")
|
|
fields["interviewer_id"] = interviewer_id
|
|
if "form_date" in payload:
|
|
fields["form_date"] = _aware(payload.get("form_date"))
|
|
return fields
|
|
|
|
async def _context_maps(self, rows):
|
|
inbox_ids = [r.inbox_id for r in rows if r.inbox_id is not None]
|
|
manual_ids = [r.manual_upload_candidate_id for r in rows if r.manual_upload_candidate_id]
|
|
job_ids = [r.job_post_id for r in rows if r.job_post_id]
|
|
|
|
inbox_by_id = {}
|
|
if inbox_ids:
|
|
result = await self.session.execute(
|
|
select(Inbox)
|
|
.options(selectinload(Inbox.messages), selectinload(Inbox.user))
|
|
.where(Inbox.id.in_(inbox_ids))
|
|
)
|
|
inbox_by_id = {row.id: row for row in result.scalars().all()}
|
|
for row in inbox_by_id.values():
|
|
msg = row.messages
|
|
if msg is not None and msg.assigned_job_post_id:
|
|
job_ids.append(msg.assigned_job_post_id)
|
|
|
|
manual_by_id = {}
|
|
if manual_ids:
|
|
result = await self.session.execute(
|
|
select(Manual_UPLOAD_CANDIDATE).where(Manual_UPLOAD_CANDIDATE.id.in_(manual_ids))
|
|
)
|
|
manual_by_id = {row.id: row for row in result.scalars().all()}
|
|
for row in manual_by_id.values():
|
|
if row.job_post_id:
|
|
job_ids.append(row.job_post_id)
|
|
|
|
jobs_by_id = {}
|
|
uids = [j for j in set(job_ids) if j]
|
|
if uids:
|
|
result = await self.session.execute(select(JobPosts).where(JobPosts.id.in_(uids)))
|
|
jobs_by_id = {row.id: row for row in result.scalars().all()}
|
|
|
|
user_ids = {r.interviewer_id for r in rows if r.interviewer_id}
|
|
user_ids |= {r.created_by for r in rows if r.created_by}
|
|
users_by_id = {}
|
|
if user_ids:
|
|
result = await self.session.execute(
|
|
select(Users.id, Users.name).where(Users.id.in_(user_ids))
|
|
)
|
|
users_by_id = {uid: name for uid, name in result.all()}
|
|
return inbox_by_id, manual_by_id, jobs_by_id, users_by_id
|
|
|
|
def _labels(self, row, inbox_by_id, manual_by_id, jobs_by_id):
|
|
candidate_name = None
|
|
job_title = None
|
|
if row.job_post_id and row.job_post_id in jobs_by_id:
|
|
job_title = jobs_by_id[row.job_post_id].title
|
|
if row.inbox_id is not None:
|
|
link = inbox_by_id.get(row.inbox_id)
|
|
if link is not None:
|
|
if link.user is not None:
|
|
candidate_name = link.user.name
|
|
msg = link.messages
|
|
if job_title is None and msg is not None and msg.assigned_job_post_id:
|
|
job = jobs_by_id.get(msg.assigned_job_post_id)
|
|
if job is not None:
|
|
job_title = job.title
|
|
if row.manual_upload_candidate_id:
|
|
manual = manual_by_id.get(row.manual_upload_candidate_id)
|
|
if manual is not None:
|
|
candidate_name = candidate_name or manual.candidate_name or None
|
|
if job_title is None and manual.job_post_id:
|
|
job = jobs_by_id.get(manual.job_post_id)
|
|
if job is not None:
|
|
job_title = job.title
|
|
return candidate_name, job_title
|
|
|
|
async def _serialize_rows(self, rows):
|
|
inbox_by_id, manual_by_id, jobs_by_id, users_by_id = await self._context_maps(rows)
|
|
out = []
|
|
for row in rows:
|
|
name, title = self._labels(row, inbox_by_id, manual_by_id, jobs_by_id)
|
|
out.append(
|
|
serialize_form(
|
|
row,
|
|
candidate_name=name,
|
|
job_title=title,
|
|
interviewer_name=users_by_id.get(row.interviewer_id),
|
|
created_by_name=users_by_id.get(row.created_by),
|
|
)
|
|
)
|
|
return out
|
|
|
|
async def get_forms(
|
|
self,
|
|
form_id=None,
|
|
inbox_id=None,
|
|
manual_upload_candidate_id=None,
|
|
job_post_id=None,
|
|
form_type=None,
|
|
top=None,
|
|
skip=0,
|
|
):
|
|
if form_type and form_type not in FORM_TYPES:
|
|
raise HTTPException(
|
|
status_code=422, detail=f"form_type must be one of {', '.join(FORM_TYPES)}"
|
|
)
|
|
rows, total = await CandidateForms.fetch_forms(
|
|
self.session,
|
|
form_id=form_id,
|
|
inbox_id=inbox_id,
|
|
manual_upload_candidate_id=manual_upload_candidate_id,
|
|
job_post_id=job_post_id,
|
|
form_type=form_type,
|
|
top=top,
|
|
skip=skip or 0,
|
|
)
|
|
|
|
summary = None
|
|
if inbox_id is not None or manual_upload_candidate_id is not None:
|
|
if form_type:
|
|
# The filtered fetch may not include both evaluation forms.
|
|
summary_rows, _ = await CandidateForms.fetch_forms(
|
|
self.session,
|
|
inbox_id=inbox_id,
|
|
manual_upload_candidate_id=manual_upload_candidate_id,
|
|
)
|
|
else:
|
|
summary_rows = rows
|
|
summary = combined_summary(summary_rows)
|
|
return await self._serialize_rows(rows), summary, total
|
|
|
|
async def create_form(self, payload, current_user):
|
|
form_type = (payload.get("form_type") or "").strip()
|
|
if form_type not in FORM_TYPES:
|
|
raise HTTPException(
|
|
status_code=422, detail=f"form_type must be one of {', '.join(FORM_TYPES)}"
|
|
)
|
|
inbox_id, manual_id, job_post_id, stage = await self._validate_link(payload)
|
|
if stage not in FORM_READY_STATUSES:
|
|
raise HTTPException(
|
|
status_code=409,
|
|
detail=(
|
|
"Forms unlock at the Interview stage — this candidate is at "
|
|
f"{stage or 'Shortlist'}"
|
|
),
|
|
)
|
|
|
|
fields = {
|
|
"inbox_id": inbox_id,
|
|
"manual_upload_candidate_id": manual_id,
|
|
"job_post_id": job_post_id,
|
|
"form_type": form_type,
|
|
"created_by": _user_id(current_user),
|
|
}
|
|
fields.update(
|
|
self._normalize_payload(
|
|
form_type,
|
|
{
|
|
key: payload.get(key)
|
|
for key in ("sections", "fields", "recommendation", "interviewer_id", "form_date")
|
|
},
|
|
)
|
|
)
|
|
if form_type != "requisition" and fields.get("interviewer_id") is None:
|
|
fields["interviewer_id"] = _user_id(current_user)
|
|
if fields.get("form_date") is None:
|
|
fields["form_date"] = _now()
|
|
|
|
row = await CandidateForms.insert_form(self.session, fields)
|
|
await HistoryRecorder(self.session).record(
|
|
HistoryEvent.FORM_CREATED,
|
|
current_user=current_user,
|
|
inbox_id=inbox_id,
|
|
manual_upload_candidate_id=manual_id,
|
|
entity_type="candidate_form",
|
|
entity_id=row.id,
|
|
to_value=form_type,
|
|
commit=True,
|
|
)
|
|
return (await self._serialize_rows([row]))[0]
|
|
|
|
async def update_form(self, form_id, payload, current_user):
|
|
_user_id(current_user)
|
|
row = await CandidateForms.get_form_by_id(self.session, form_id)
|
|
if not row:
|
|
raise HTTPException(status_code=404, detail="Form not found")
|
|
|
|
fields = self._normalize_payload(row.form_type, payload)
|
|
if not fields:
|
|
raise HTTPException(status_code=400, detail="No fields to update")
|
|
|
|
updated = await CandidateForms.update_form(self.session, form_id, fields)
|
|
if not updated:
|
|
raise HTTPException(status_code=404, detail="Form not found")
|
|
await HistoryRecorder(self.session).record(
|
|
HistoryEvent.FORM_UPDATED,
|
|
current_user=current_user,
|
|
inbox_id=updated.inbox_id,
|
|
manual_upload_candidate_id=updated.manual_upload_candidate_id,
|
|
entity_type="candidate_form",
|
|
entity_id=updated.id,
|
|
to_value=updated.form_type,
|
|
commit=True,
|
|
)
|
|
return (await self._serialize_rows([updated]))[0]
|
|
|
|
async def delete_form(self, form_id, current_user):
|
|
_user_id(current_user)
|
|
row = await CandidateForms.soft_delete_form(self.session, form_id)
|
|
if not row:
|
|
raise HTTPException(status_code=404, detail="Form not found")
|
|
return {"id": str(row.id), "deleted": True}
|