333 lines
13 KiB
Python
333 lines
13 KiB
Python
import logging
|
|
import uuid
|
|
from datetime import timezone
|
|
|
|
import httpx
|
|
from fastapi import HTTPException
|
|
from sqlalchemy import select
|
|
from sqlalchemy.ext.asyncio import AsyncSession
|
|
from sqlalchemy.orm import selectinload
|
|
|
|
from assessments.models import Assessments, _now
|
|
from assessments.serializers import serialize_assessment
|
|
from inbox.models import Inbox
|
|
from inbox.plugins import send_mail
|
|
from job.candidate.models import Manual_UPLOAD_CANDIDATE
|
|
from job.job_post.models import JobPosts
|
|
from notifications.models import Notifications
|
|
from users.models import Users
|
|
|
|
logger = logging.getLogger("assessments")
|
|
|
|
VALID_TYPES = (
|
|
"Coding Challenge",
|
|
"Take-home Project",
|
|
"Cognitive Test",
|
|
"Personality Assessment",
|
|
"SQL Test",
|
|
"Case Study",
|
|
)
|
|
VALID_STATUS = ("pending", "in_progress", "completed", "expired")
|
|
|
|
|
|
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
|
|
|
|
|
|
class Assessment:
|
|
def __init__(self, session: AsyncSession):
|
|
self.session = session
|
|
|
|
async def _emit(self, current_user, kind, title, body, *, inbox_id=None, job_post_id=None, link_path=None):
|
|
uid = _as_uuid(current_user.get("id") if current_user else None)
|
|
if uid is None:
|
|
return
|
|
try:
|
|
await Notifications.insert_notification(self.session, {
|
|
"user_id": uid,
|
|
"kind": kind,
|
|
"title": title,
|
|
"body": body,
|
|
"link_path": link_path,
|
|
"inbox_id": inbox_id,
|
|
"job_post_id": job_post_id,
|
|
})
|
|
except Exception as exc:
|
|
logger.warning("notification insert skipped: %s", exc)
|
|
|
|
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()}
|
|
return inbox_by_id, manual_by_id, jobs_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 = 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_assessment(row, candidate_name=name, job_title=title))
|
|
return out
|
|
|
|
async def _recipient(self, row):
|
|
if row.inbox_id is not None:
|
|
link = await Inbox.get_inbox_with_message(self.session, row.inbox_id)
|
|
if link is None:
|
|
return None, None, None
|
|
user = link.user
|
|
if user is None and link.user_id:
|
|
user = await Users.get_user_by_id(self.session, str(link.user_id))
|
|
email = user.email if user else None
|
|
name = user.name if user else None
|
|
return email, name, link.id
|
|
if row.manual_upload_candidate_id:
|
|
manual = await Manual_UPLOAD_CANDIDATE.get_by_id(
|
|
self.session, row.manual_upload_candidate_id
|
|
)
|
|
if manual is None:
|
|
return None, None, None
|
|
return (manual.candidate_email or None), (manual.candidate_name or None), None
|
|
return None, None, None
|
|
|
|
async def get_assessments(
|
|
self,
|
|
assessment_id=None,
|
|
inbox_id=None,
|
|
manual_upload_candidate_id=None,
|
|
job_post_id=None,
|
|
assessment_status=None,
|
|
top=None,
|
|
skip=0,
|
|
):
|
|
if assessment_status and assessment_status not in VALID_STATUS:
|
|
raise HTTPException(
|
|
status_code=422, detail=f"assessment_status must be one of {', '.join(VALID_STATUS)}"
|
|
)
|
|
rows, total = await Assessments.fetch_assessments(
|
|
self.session,
|
|
assessment_id=assessment_id,
|
|
inbox_id=inbox_id,
|
|
manual_upload_candidate_id=manual_upload_candidate_id,
|
|
job_post_id=job_post_id,
|
|
assessment_status=assessment_status,
|
|
top=top,
|
|
skip=skip or 0,
|
|
)
|
|
return await self._serialize_rows(rows), total
|
|
|
|
async def get_counts(self):
|
|
counts = await Assessments.count_by_status(self.session)
|
|
return {status: int(counts.get(status, 0)) for status in VALID_STATUS}
|
|
|
|
async def create_assessment(self, payload, current_user):
|
|
assessment_type = (payload.get("assessment_type") or "").strip()
|
|
if assessment_type not in VALID_TYPES:
|
|
raise HTTPException(
|
|
status_code=422, detail=f"assessment_type must be one of {', '.join(VALID_TYPES)}"
|
|
)
|
|
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_by_id(self.session, inbox_id)
|
|
if link is None:
|
|
raise HTTPException(status_code=404, detail="Inbox record not found")
|
|
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")
|
|
|
|
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")
|
|
|
|
fields = {
|
|
"inbox_id": inbox_id,
|
|
"manual_upload_candidate_id": manual_id,
|
|
"job_post_id": job_post_id,
|
|
"assessment_type": assessment_type,
|
|
"assessment_status": "pending",
|
|
"duration_minutes": payload.get("duration_minutes"),
|
|
"assigned_at": _now(),
|
|
"created_by": _user_id(current_user),
|
|
}
|
|
if payload.get("due_at") is not None:
|
|
fields["due_at"] = _aware(payload["due_at"])
|
|
row = await Assessments.insert_assessment(self.session, fields)
|
|
data = (await self._serialize_rows([row]))[0]
|
|
await self._emit(
|
|
current_user,
|
|
"assessment",
|
|
"Assessment assigned",
|
|
f"{assessment_type} assigned to {data.get('candidate_name') or 'a candidate'}",
|
|
inbox_id=inbox_id,
|
|
job_post_id=job_post_id,
|
|
link_path="/assessments",
|
|
)
|
|
return data
|
|
|
|
async def update_assessment(self, assessment_id, payload, current_user):
|
|
_user_id(current_user)
|
|
row = await Assessments.get_assessment_by_id(self.session, assessment_id)
|
|
if not row:
|
|
raise HTTPException(status_code=404, detail="Assessment not found")
|
|
|
|
fields = {}
|
|
if "assessment_status" in payload:
|
|
status = payload["assessment_status"]
|
|
if status not in VALID_STATUS:
|
|
raise HTTPException(
|
|
status_code=422, detail=f"assessment_status must be one of {', '.join(VALID_STATUS)}"
|
|
)
|
|
fields["assessment_status"] = status
|
|
if status == "completed" and row.completed_at is None:
|
|
fields["completed_at"] = _now()
|
|
if "score" in payload:
|
|
score = payload["score"]
|
|
if score is not None and (not isinstance(score, int) or score < 0 or score > 100):
|
|
raise HTTPException(status_code=422, detail="score must be an integer 0-100")
|
|
fields["score"] = score
|
|
if "section_scores" in payload:
|
|
sections = payload["section_scores"]
|
|
if sections is not None and not isinstance(sections, list):
|
|
raise HTTPException(status_code=422, detail="section_scores must be a list")
|
|
fields["section_scores"] = sections
|
|
if "due_at" in payload:
|
|
fields["due_at"] = _aware(payload["due_at"])
|
|
if not fields:
|
|
raise HTTPException(status_code=400, detail="No fields to update")
|
|
|
|
updated = await Assessments.update_assessment(self.session, assessment_id, fields)
|
|
if not updated:
|
|
raise HTTPException(status_code=404, detail="Assessment not found")
|
|
return (await self._serialize_rows([updated]))[0]
|
|
|
|
async def delete_assessment(self, assessment_id, current_user):
|
|
_user_id(current_user)
|
|
row = await Assessments.soft_delete_assessment(self.session, assessment_id)
|
|
if not row:
|
|
raise HTTPException(status_code=404, detail="Assessment not found")
|
|
return {"id": str(row.id), "deleted": True}
|
|
|
|
async def remind_assessment(self, assessment_id, current_user):
|
|
_user_id(current_user)
|
|
row = await Assessments.get_assessment_by_id(self.session, assessment_id)
|
|
if not row:
|
|
raise HTTPException(status_code=404, detail="Assessment not found")
|
|
email, name, inbox_id = await self._recipient(row)
|
|
if not email:
|
|
raise HTTPException(status_code=422, detail="Candidate has no email address")
|
|
subject = f"Reminder: {row.assessment_type}"
|
|
body = (
|
|
f"<p>This is a reminder that your {row.assessment_type} assessment is pending"
|
|
f"{' for ' + name if name else ''}.</p>"
|
|
)
|
|
try:
|
|
await send_mail(email, subject, body, content_type="html")
|
|
except (httpx.HTTPError, RuntimeError) as e:
|
|
raise HTTPException(status_code=502, detail="Failed to send reminder email") from e
|
|
|
|
updated = await Assessments.update_assessment(
|
|
self.session, assessment_id, {"reminded_at": _now()}
|
|
)
|
|
data = (await self._serialize_rows([updated]))[0]
|
|
await self._emit(
|
|
current_user,
|
|
"assessment",
|
|
"Assessment reminder sent",
|
|
f"Reminder sent for {row.assessment_type}",
|
|
inbox_id=inbox_id,
|
|
job_post_id=row.job_post_id,
|
|
link_path="/assessments",
|
|
)
|
|
return data
|