931 lines
43 KiB
Python
931 lines
43 KiB
Python
from sqlalchemy.ext.asyncio import AsyncSession
|
|
import asyncio,base64,dataclasses,hashlib,io,logging,os,uuid
|
|
from datetime import datetime,timezone
|
|
from pathlib import Path
|
|
from dotenv import load_dotenv
|
|
from fastapi import HTTPException
|
|
from pypdf import PdfReader
|
|
from sqlalchemy import select
|
|
from sqlalchemy.orm import selectinload
|
|
from sqlmodel import true
|
|
from app.core.errors import ATSError,ErrorCode
|
|
from app.models.scoring import CompletedCandidate
|
|
from app.services.pdf import extract_resume,sanitize_filename
|
|
from app.services.scoring import score_batch
|
|
from inbox.models import Inbox_Messages,Inbox,AtsResults
|
|
from job.candidate.models import Candidates
|
|
from job.candidate.plugins import (
|
|
FILE_NOT_FOUND,
|
|
build_job_description,
|
|
candidate_completed_fields,
|
|
candidate_failed_fields,
|
|
contained_download_path,
|
|
documents_from_message,
|
|
get_scorer,
|
|
get_scoring_settings,
|
|
normalize_spaced_text,
|
|
)
|
|
from job.candidate.serializers import serialize_candidate,serialize_candidate_profile,serialize_manual_candidate_profile,serialize_manual_upload_candidate
|
|
from job.job_post.models import JobPosts
|
|
from job.job_post.serializers import serialize_job_post
|
|
from job.candidate.models import Notes,Manual_UPLOAD_CANDIDATE
|
|
from job.history.enums import HistoryEvent
|
|
from job.history.views import HistoryRecorder
|
|
from job.notes.serializers import serialize_note
|
|
from job.candidate.plugins import extract_candidate_email
|
|
from users.models import Users
|
|
|
|
load_dotenv()
|
|
logger=logging.getLogger("job.candidate.views")
|
|
CV_QUEUE_NAME=os.getenv("TASKIQ_CV_QUEUE_NAME","cv_upload")
|
|
MANUAL_UPLOAD_TO_ADDRESS=os.getenv(
|
|
"MANUAL_UPLOAD_TO_ADDRESS","manual-cv-upload@hr-ats.local"
|
|
)
|
|
|
|
class FileRead:
|
|
def __init__(self,session:AsyncSession,filename=None,file=None):
|
|
self.session=session
|
|
self.filename=filename
|
|
self.file=file
|
|
|
|
async def read_file(self,file=None,filename=None):
|
|
try:
|
|
reader = PdfReader(io.BytesIO(self.file))
|
|
if reader.is_encrypted:
|
|
raise HTTPException(400, "PDF is password protected")
|
|
pages = [(page.extract_text() or "") for page in reader.pages]
|
|
return {
|
|
"filename": self.filename,
|
|
"num_pages": len(reader.pages),
|
|
"text": normalize_spaced_text("\n".join(pages)),
|
|
}
|
|
except HTTPException:
|
|
raise
|
|
except Exception as e:
|
|
raise HTTPException(400, str(e))
|
|
async def injest_manual_upload(self):
|
|
try:
|
|
parsed=await self.read_file()
|
|
text=(parsed.get("text") or "").strip()
|
|
if not text:
|
|
raise HTTPException(status_code=400,detail="No usable text could be extracted from the PDF")
|
|
return parsed
|
|
except HTTPException:
|
|
raise
|
|
except Exception as e:
|
|
raise HTTPException(status_code=400,detail=str(e))
|
|
|
|
async def save_manual_upload(self):
|
|
"""Write the uploaded CV under inbox/decoded_attachments.
|
|
|
|
Returns ``{"file_name", "file_path"}``: the recruiter-facing original
|
|
name, and the absolute path actually written.
|
|
|
|
Those two differ deliberately. decode_attachment writes ``Path(name).name``
|
|
with plain ``write_bytes`` — no collision handling — so two candidates
|
|
uploading "resume.pdf" would silently clobber each other and the first
|
|
row's file_path would then serve the second candidate's CV. Prefixing the
|
|
stored basename with a uuid makes every upload its own file, while
|
|
file_name keeps what the recruiter recognises. resolve_attachment_path
|
|
handles the result either way: the stored absolute path wins, and its
|
|
basename-under-attachments fallback still finds the prefixed name.
|
|
"""
|
|
from inbox.file_decoder import AttachmentDecodeError,decode_attachment
|
|
|
|
# Separators normalized before taking the basename: a Windows client can
|
|
# send "C:\Users\x\cv.pdf", whose Path(...).name on Linux is the whole
|
|
# string. Same reasoning as inbox.plugins.resolve_attachment_path.
|
|
original=Path((self.filename or "resume.pdf").replace("\\","/")).name or "resume.pdf"
|
|
stored=f"{uuid.uuid4().hex}-{original}"
|
|
try:
|
|
paths=await decode_attachment([{
|
|
"name":stored,
|
|
"contentBytes":base64.b64encode(self.file).decode("ascii"),
|
|
}])
|
|
except AttachmentDecodeError as e:
|
|
raise HTTPException(status_code=400,detail=str(e))
|
|
if not paths:
|
|
# decode_attachment skips rather than raises on an unsupported
|
|
# extension, so an empty list is the only signal that nothing landed.
|
|
raise HTTPException(status_code=400,detail="attachment could not be saved")
|
|
return {"file_name":original,"file_path":paths[0]}
|
|
|
|
@staticmethod
|
|
def discard_upload(file_path):
|
|
"""Best-effort removal of a saved CV whose row never got created.
|
|
|
|
Called on the failure path so a rejected request (a missing email, a DB
|
|
error) does not leave an orphan PDF behind. Failure to delete is logged
|
|
and swallowed — it must never mask the error that got us here.
|
|
"""
|
|
if not file_path:
|
|
return
|
|
try:
|
|
Path(file_path).unlink(missing_ok=True)
|
|
except OSError as e:
|
|
logger.warning("could not remove orphaned upload %s: %s",file_path,e)
|
|
|
|
async def ingest_upload(self,candidate_email=None,candidate_name=None,current_user=None):
|
|
"""Persist a recruiter-uploaded CV with full email-ingestion parity."""
|
|
from inbox.file_decoder import AttachmentDecodeError,decode_attachment
|
|
from inbox.cv_tasks import match_uploaded_cv
|
|
from inbox.views import Email
|
|
|
|
parsed=await self.read_file()
|
|
text=parsed.get("text") or ""
|
|
detected,emails_found=extract_candidate_email(text)
|
|
supplied=(candidate_email or "").strip().lower() or None
|
|
email=supplied or detected
|
|
email_source="recruiter" if supplied else ("cv" if detected else None)
|
|
|
|
if not email:
|
|
raise HTTPException(
|
|
status_code=422,
|
|
detail={
|
|
"error_code":"CANDIDATE_EMAIL_REQUIRED",
|
|
"filename":parsed.get("filename"),
|
|
"num_pages":parsed.get("num_pages"),
|
|
"emails_found":emails_found,
|
|
"text":text,
|
|
},
|
|
)
|
|
|
|
filename=self.filename or "resume.pdf"
|
|
try:
|
|
paths=await decode_attachment([{
|
|
"name":filename,
|
|
"contentBytes":base64.b64encode(self.file).decode("ascii"),
|
|
}])
|
|
except AttachmentDecodeError as e:
|
|
raise HTTPException(status_code=400,detail=str(e))
|
|
if not paths:
|
|
raise HTTPException(status_code=400,detail="attachment could not be saved")
|
|
|
|
now=datetime.now(timezone.utc).isoformat()
|
|
email_data={
|
|
"id":f"manual-cv:{uuid.uuid4()}",
|
|
"subject":f"Manual CV upload — {filename}",
|
|
"body":{"content":"","contentType":"text"},
|
|
"hasAttachments":True,
|
|
"attachments":[{"name":filename}],
|
|
"from":{"emailAddress":{"address":email,"name":(candidate_name or "").strip()}},
|
|
"toRecipients":[{"emailAddress":{"address":MANUAL_UPLOAD_TO_ADDRESS}}],
|
|
"ccRecipients":[],
|
|
"bccRecipients":[],
|
|
"replyTo":[],
|
|
"isRead":False,
|
|
"sentDateTime":now,
|
|
"receivedDateTime":now,
|
|
}
|
|
row,new_user_email=await Inbox_Messages.insert_email(
|
|
self.session,email_data,file_path=paths,
|
|
)
|
|
|
|
created_at=datetime.now(timezone.utc).isoformat()
|
|
task=await match_uploaded_cv.kicker().with_labels(
|
|
created_at=created_at,
|
|
correlation_id=str(row.id),
|
|
queue=CV_QUEUE_NAME,
|
|
).kiq(str(row.id),force=False)
|
|
|
|
account_setup=None
|
|
if new_user_email:
|
|
try:
|
|
account_setup=await Email(session=self.session).send_account_setup(
|
|
[new_user_email]
|
|
)
|
|
except Exception as e:
|
|
logger.warning("account setup mail failed for %s: %s",new_user_email,e)
|
|
account_setup=[{"email":new_user_email,"sent":False}]
|
|
|
|
# Inbox link may not exist yet (match task creates it later); resolve
|
|
# by email because insert_email creates the Users row synchronously.
|
|
user=await Users.get_user_by_email(self.session,email)
|
|
if user:
|
|
await HistoryRecorder(self.session).record(
|
|
HistoryEvent.CANDIDATE_IMPORTED.value,
|
|
current_user=current_user,user_id=user.id,
|
|
entity_type="inbox_message",entity_id=row.id,
|
|
to_value=email,description=f"CV uploaded: {filename}",commit=True,
|
|
)
|
|
await HistoryRecorder(self.session).record(
|
|
HistoryEvent.DOCUMENT_UPLOADED.value,
|
|
current_user=current_user,user_id=user.id,
|
|
entity_type="document",entity_id=row.id,
|
|
to_value=filename,commit=True,
|
|
)
|
|
|
|
return {
|
|
"queued":True,
|
|
"inbox_message_id":str(row.id),
|
|
"task_id":task.task_id,
|
|
"filename":parsed.get("filename"),
|
|
"num_pages":parsed.get("num_pages"),
|
|
"candidate_email":email,
|
|
"email_source":email_source,
|
|
"account_setup":account_setup,
|
|
"text":text,
|
|
}
|
|
|
|
async def match_inbox_cv(self,inbox_message_id,current_user=None):
|
|
from inbox.plugins import resolve_attachment_path
|
|
from inbox.tasks import match_inbox_message
|
|
|
|
row=await Inbox_Messages.get_inbox_message_by_id(self.session,inbox_message_id)
|
|
if not row:
|
|
raise HTTPException(status_code=404,detail="Message not found")
|
|
if not row.attachment or not row.file_path:
|
|
raise HTTPException(status_code=400,detail="your file isnt in the system")
|
|
|
|
found=None
|
|
for path_str in (p.strip() for p in row.file_path.split(",") if p.strip()):
|
|
path=resolve_attachment_path(path_str)
|
|
if path.is_file():
|
|
found=path
|
|
break
|
|
if found is None:
|
|
raise HTTPException(status_code=400,detail="your file isnt in the system")
|
|
|
|
created_at=datetime.now(timezone.utc).isoformat()
|
|
task=await match_inbox_message.kicker().with_labels(
|
|
created_at=created_at,
|
|
correlation_id=str(row.id),
|
|
queue="inbox",
|
|
).kiq(str(row.id),force=True)
|
|
|
|
file_name=(row.file_name or "").split(",")[0].strip() or found.name
|
|
await HistoryRecorder(self.session).record(
|
|
HistoryEvent.CANDIDATE_IMPORTED.value,
|
|
current_user=current_user,message_id=inbox_message_id,
|
|
entity_type="inbox_message",entity_id=row.id,
|
|
to_value=file_name,description=f"CV uploaded: {file_name}",commit=True,
|
|
)
|
|
return {
|
|
"queued":True,
|
|
"inbox_message_id":str(row.id),
|
|
"file_name":file_name,
|
|
"task_id":task.task_id,
|
|
}
|
|
# async def get_intention(self,input):
|
|
# try:
|
|
# get_subject=Inbox_Messages.candidate_x_inbox(self.session,self.candidate_id)
|
|
# get_file=
|
|
|
|
|
|
class CandidateScoring:
|
|
"""ATS scoring of CVs against one job post, persisted to the candidates table.
|
|
|
|
Complements the agent's inbox-match flow: the agent suggests WHICH job a CV is
|
|
for; this service scores HOW WELL a CV fits a chosen job (0-100 leaderboard).
|
|
|
|
Deviation from the bulk-ats HTTP API (which rejects a whole batch with 415/413
|
|
on a bad file): here per-file problems become persisted rows with
|
|
status="failed" so one broken attachment never sinks the rest of the batch.
|
|
Request-level errors (unknown job, too many files) still raise.
|
|
"""
|
|
|
|
def __init__(self,session:AsyncSession):
|
|
self.session=session
|
|
|
|
async def score_uploads(self,job_id,files,current_user):
|
|
"""files: list of (filename, bytes) pairs from the route handler."""
|
|
settings=get_scoring_settings()
|
|
if len(files)>settings.max_resumes_per_request:
|
|
raise HTTPException(
|
|
status_code=413,
|
|
detail=f"At most {settings.max_resumes_per_request} resumes per request",
|
|
)
|
|
sources=[]
|
|
for filename,data in files:
|
|
source={
|
|
"filename":filename or "resume.pdf",
|
|
"data":data,
|
|
"file_path":None,
|
|
"precheck":None,
|
|
}
|
|
if not (filename or "").lower().endswith(".pdf"):
|
|
source["precheck"]=(ErrorCode.UNSUPPORTED_FILE_TYPE,"Only PDF resumes are supported.")
|
|
elif len(data)>settings.max_pdf_size_bytes:
|
|
source["precheck"]=(ErrorCode.PAYLOAD_TOO_LARGE,"The file exceeds the size limit.")
|
|
sources.append(source)
|
|
return await self._score_and_persist(job_id,sources,"upload",current_user)
|
|
|
|
async def score_inbox(self,job_id,message_ids,current_user):
|
|
"""Score the decoded attachments of inbox messages (PK uuids, not Graph ids)."""
|
|
# Local import: inbox.plugins imports this module (FileRead), so a top-level
|
|
# import would be circular — same pattern as match_inbox_cv above.
|
|
from inbox.plugins import resolve_attachment_path
|
|
|
|
sources=[]
|
|
for mid in message_ids:
|
|
row=await Inbox_Messages.get_inbox_message_by_id(self.session,mid)
|
|
if row is None:
|
|
raise HTTPException(status_code=404,detail=f"Inbox message {mid} not found")
|
|
if not row.file_path:
|
|
continue
|
|
for path_str in (p.strip() for p in row.file_path.split(",") if p.strip()):
|
|
path=resolve_attachment_path(path_str)
|
|
source={
|
|
"filename":path.name,
|
|
"data":None,
|
|
"file_path":str(path),
|
|
"inbox_message_id":row.id, # call-scoped; not persisted on Candidates
|
|
"candidate_email":(row.message_from or "").strip().lower() or None,
|
|
"precheck":None,
|
|
}
|
|
suffix=path.suffix.lower()
|
|
if suffix in (".doc",".docx"):
|
|
source["precheck"]=(ErrorCode.UNSUPPORTED_FILE_TYPE,"DOC/DOCX extraction is not supported yet.")
|
|
elif suffix!=".pdf":
|
|
source["precheck"]=(ErrorCode.UNSUPPORTED_FILE_TYPE,"Only PDF resumes are supported.")
|
|
elif not path.is_file():
|
|
source["precheck"]=(FILE_NOT_FOUND,"The decoded attachment is missing on disk.")
|
|
else:
|
|
try:
|
|
source["data"]=await asyncio.to_thread(path.read_bytes)
|
|
except OSError:
|
|
source["precheck"]=(FILE_NOT_FOUND,"The decoded attachment could not be read.")
|
|
sources.append(source)
|
|
if not sources:
|
|
raise HTTPException(status_code=400,detail="No attachments found for the given message(s)")
|
|
return await self._score_and_persist(job_id,sources,"inbox",current_user)
|
|
|
|
async def fetch_candidates(self,job_id=None):
|
|
# job_id omitted -> the whole pool across jobs (frontend Candidates/TalentPool).
|
|
if job_id is not None:
|
|
job=await JobPosts.get_job_post_by_id(self.session,job_id)
|
|
if job is None or job.is_deleted:
|
|
raise HTTPException(status_code=404,detail="Job post not found")
|
|
rows=await Candidates.get_candidates_by_job(self.session,job_id)
|
|
return [serialize_candidate(row) for row in rows]
|
|
|
|
async def fetch_candidate_by_id(self,candidate_id):
|
|
row=await Candidates.get_candidate_by_id(self.session,candidate_id)
|
|
if row is None:
|
|
raise HTTPException(status_code=404,detail="Candidate not found")
|
|
return serialize_candidate(row)
|
|
|
|
async def _score_and_persist(self,job_id,sources,source_kind,current_user):
|
|
job=await JobPosts.get_job_post_by_id(self.session,job_id)
|
|
if job is None or job.is_deleted:
|
|
raise HTTPException(status_code=404,detail="Job post not found")
|
|
settings=get_scoring_settings()
|
|
jd=build_job_description(job)
|
|
if len(jd)>settings.max_jd_chars:
|
|
raise HTTPException(status_code=422,detail="The job post is too large to score against")
|
|
fields_by_slot=await self._score_sources(sources,jd,settings)
|
|
common={
|
|
"job_id":job.id,
|
|
"source":source_kind,
|
|
"created_by":uuid.UUID(str(current_user["id"])),
|
|
"model":settings.openai_model,
|
|
}
|
|
rows=[]
|
|
for slot in range(len(sources)):
|
|
rows.append(await Candidates.upsert_candidate(self.session,{**fields_by_slot[slot],**common}))
|
|
await self._sync_ats_results(source_kind,job,rows,sources,current_user)
|
|
rows.sort(key=lambda r:(0,-(r.match_score or 0)) if r.status=="completed" else (1,0))
|
|
return [serialize_candidate(row) for row in rows]
|
|
|
|
async def _score_sources(self,sources,jd,settings):
|
|
# Slot-indexed: results merge back by position, never by filename —
|
|
# inbox attachments can share a basename.
|
|
fields_by_slot={}
|
|
extracted=[]
|
|
for slot,source in enumerate(sources):
|
|
source["safe_name"]=sanitize_filename(source["filename"])
|
|
data=source["data"]
|
|
source["sha256"]=hashlib.sha256(data).hexdigest() if data is not None else None
|
|
if source["precheck"] is not None:
|
|
code,message=source["precheck"]
|
|
fields_by_slot[slot]=candidate_failed_fields(source,code,message)
|
|
continue
|
|
try:
|
|
resume=await asyncio.to_thread(
|
|
extract_resume,data,source["safe_name"],settings.max_resume_chars
|
|
)
|
|
resume=dataclasses.replace(resume,text=normalize_spaced_text(resume.text))
|
|
if not source.get("candidate_email"):
|
|
detected,_=extract_candidate_email(resume.text)
|
|
source["candidate_email"]=detected
|
|
except ATSError as exc:
|
|
fields_by_slot[slot]=candidate_failed_fields(source,exc.error_code,exc.public_message)
|
|
continue
|
|
extracted.append((slot,resume))
|
|
scored=await score_batch(
|
|
[resume for _,resume in extracted],
|
|
job_description=jd,
|
|
scorer=get_scorer(),
|
|
concurrency=settings.scoring_concurrency,
|
|
)
|
|
for (slot,_),result in zip(extracted,scored,strict=True):
|
|
source=sources[slot]
|
|
if isinstance(result,CompletedCandidate):
|
|
fields_by_slot[slot]=candidate_completed_fields(source,result)
|
|
else:
|
|
fields_by_slot[slot]=candidate_failed_fields(source,result.error_code,result.error_message)
|
|
return fields_by_slot
|
|
|
|
async def _sync_ats_results(self,source_kind,job,rows,sources,current_user=None):
|
|
if source_kind=="inbox":
|
|
best={}
|
|
for source,row in zip(sources,rows):
|
|
mid=source.get("inbox_message_id")
|
|
if row.status=="completed" and mid:
|
|
cur=best.get(mid)
|
|
if cur is None or (row.match_score or 0)>(cur.match_score or 0):
|
|
best[mid]=row
|
|
for message_id,row in best.items():
|
|
try:
|
|
await self._sync_inbox_ats(message_id,job,row,current_user=current_user)
|
|
except Exception:
|
|
await self.session.rollback()
|
|
logger.exception("inbox ATS denorm failed for message %s",message_id)
|
|
return
|
|
for row in rows:
|
|
if row.status!="completed":
|
|
continue
|
|
try:
|
|
await self._sync_upload_ats(job,row,current_user=current_user)
|
|
except Exception:
|
|
await self.session.rollback()
|
|
logger.exception("upload ATS history failed for candidate %s",row.id)
|
|
|
|
async def _sync_inbox_ats(self,message_id,job,row,current_user=None):
|
|
"""Land a completed score on inbox_messages / inbox / ats_results.
|
|
|
|
message_id is the scoring call's known inbox_messages PK, not a column
|
|
read off the Candidates row.
|
|
"""
|
|
msg=await Inbox_Messages.get_inbox_message_by_id(self.session,message_id)
|
|
if msg is None:
|
|
return
|
|
band=CandidateView._recommendation(row.match_score) or ""
|
|
denorm=True
|
|
assigned=msg.assigned_job_post_id
|
|
link=await Inbox.get_inbox_by_message_id(self.session,message_id)
|
|
if assigned and str(assigned)!=str(job.id) and link is not None:
|
|
existing=await AtsResults.get_for_inbox_job(self.session,link.id,assigned)
|
|
denorm=existing is None
|
|
if denorm:
|
|
await Inbox_Messages.set_ats_score(self.session,message_id,row.match_score,band)
|
|
if link is None:
|
|
return
|
|
old=await AtsResults.get_current_for_inbox(self.session,link.id)
|
|
old_score=old.overall_score if old else None
|
|
identity=await AtsResults.resolve_identity(self.session,row.candidate_email,row.id)
|
|
await AtsResults.insert_result(self.session,{
|
|
"inbox_id":link.id,
|
|
**identity,
|
|
"job_post_id":job.id,
|
|
"overall_score":float(row.match_score),
|
|
"band":band,
|
|
"model_name":row.model,
|
|
"is_current":True,
|
|
})
|
|
await HistoryRecorder(self.session).record(
|
|
HistoryEvent.ATS_SCORED.value,
|
|
current_user=current_user,user_id=link.user_id,inbox_id=link.id,
|
|
entity_type="ats_result",entity_id=row.id,
|
|
from_value=old_score,to_value=row.match_score,
|
|
description=f"{band} against {job.title or 'job'}",commit=True,
|
|
)
|
|
|
|
async def _sync_upload_ats(self,job,row,current_user=None):
|
|
"""History row for an upload-sourced score — inbox_id stays NULL."""
|
|
band=CandidateView._recommendation(row.match_score) or ""
|
|
identity=await AtsResults.resolve_identity(self.session,row.candidate_email,row.id)
|
|
old=None
|
|
if identity.get("user_id"):
|
|
old=await AtsResults.get_current_for_user(self.session,identity["user_id"],job.id)
|
|
elif identity.get("candidate_id"):
|
|
old=await AtsResults.get_current_for_candidate(self.session,identity["candidate_id"])
|
|
old_score=old.overall_score if old else None
|
|
await AtsResults.insert_result(self.session,{
|
|
"inbox_id":None,
|
|
**identity,
|
|
"job_post_id":job.id,
|
|
"overall_score":float(row.match_score),
|
|
"band":band,
|
|
"model_name":row.model,
|
|
"is_current":True,
|
|
})
|
|
await HistoryRecorder(self.session).record(
|
|
HistoryEvent.ATS_SCORED.value,
|
|
current_user=current_user,user_id=identity.get("user_id"),
|
|
entity_type="ats_result",entity_id=row.id,
|
|
from_value=old_score,to_value=row.match_score,
|
|
description=f"{band} against {job.title or 'job'}",commit=True,
|
|
)
|
|
|
|
class CandidateView:
|
|
def __init__(self,session:AsyncSession):
|
|
self.session=session
|
|
|
|
@staticmethod
|
|
def _recommendation(score):
|
|
if score is None:
|
|
return None
|
|
# Same bands the frontend uses (Candidates.jsx / seed.js).
|
|
return "Strong Match" if score>=82 else "Potential Match" if score>=65 else "Weak Match"
|
|
|
|
@staticmethod
|
|
def _score_from_message(record):
|
|
"""Inbox denorm on the already-loaded messages row — not a Candidates join."""
|
|
msg=getattr(record,"messages",None)
|
|
if msg is None or msg.ats_score is None:
|
|
return None,None
|
|
band=(msg.ats_band or "").strip() or None
|
|
return msg.ats_score,band or CandidateView._recommendation(msg.ats_score)
|
|
|
|
async def create_candidate(self,candidate_email=None,candidate_name=None,candidate_phone=None,job_post_id=None,current_company=None,current_position=None,platform=None,experience=None,status=None,referral_by=None,file_name=None,file_path=None,full_text=None,current_user=None):
|
|
try:
|
|
email=(candidate_email or "").strip().lower()
|
|
if not email:
|
|
raise HTTPException(status_code=422,detail="candidate_email is required")
|
|
if not current_user:
|
|
raise HTTPException(status_code=400,detail="created_by is required")
|
|
data={
|
|
"candidate_email":email,
|
|
"candidate_name":(candidate_name or "").strip(),
|
|
"candidate_phone":(candidate_phone or "").strip(),
|
|
"job_post_id":job_post_id,
|
|
"current_company":(current_company or "").strip(),
|
|
"current_position":(current_position or "").strip(),
|
|
"platform":(platform or "").strip(),
|
|
"experience":(experience or "").strip(),
|
|
"status":(status or "").strip(),
|
|
"referral_by":(referral_by or "").strip(),
|
|
"file_name":(file_name or "").strip(),
|
|
"file_path":(file_path or "").strip(),
|
|
"full_text":full_text or "",
|
|
"created_by":current_user,
|
|
|
|
}
|
|
row=await Manual_UPLOAD_CANDIDATE.create_manual_upload_candidate(session=self.session,fields=data)
|
|
await HistoryRecorder(self.session).record(
|
|
HistoryEvent.CANDIDATE_CREATED.value,
|
|
actor_id=current_user,user_id=row.user_id,
|
|
manual_upload_candidate_id=row.id,
|
|
entity_type="manual_upload_candidate",entity_id=row.id,
|
|
to_value=row.candidate_email,
|
|
description=(row.platform or "").strip() or "manual_upload",commit=True,
|
|
)
|
|
if (row.file_name or "").strip():
|
|
await HistoryRecorder(self.session).record(
|
|
HistoryEvent.DOCUMENT_UPLOADED.value,
|
|
actor_id=current_user,user_id=row.user_id,
|
|
manual_upload_candidate_id=row.id,
|
|
entity_type="document",entity_id=row.id,
|
|
to_value=row.file_name,commit=True,
|
|
)
|
|
return serialize_manual_upload_candidate(row)
|
|
except HTTPException:
|
|
raise
|
|
except Exception as e:
|
|
raise HTTPException(status_code=500,detail=str(e))
|
|
|
|
async def get_candidate(self,user_id=None,limit=10,offset=0,search=None):
|
|
try:
|
|
detail=bool(user_id)
|
|
# Detail mode must see every application for the candidate, not one page.
|
|
fetch_limit=1000 if detail else limit
|
|
rows=await Inbox.get_candidate_profile(session=self.session,user_id=user_id,limit=fetch_limit,offset=offset,search=search)
|
|
if detail:
|
|
records=rows if isinstance(rows,list) else ([rows] if rows else [])
|
|
if records:
|
|
return await self.attach_profile_detail(rows)
|
|
# Manual uploads create users + manual_upload_candidate but no inbox
|
|
# row — resolve the profile from that table instead of returning [].
|
|
manual=await Manual_UPLOAD_CANDIDATE.get_by_user_id(self.session,user_id)
|
|
if not manual:
|
|
return []
|
|
user=await Users.get_user_by_id(self.session,user_id)
|
|
job_post=None
|
|
if manual.job_post_id:
|
|
job_post=await JobPosts.get_job_post_by_id(self.session,str(manual.job_post_id))
|
|
payload=serialize_manual_candidate_profile(manual,user,job_post)
|
|
from inbox.plugins import get_ats_score_for_manual_user
|
|
score=await get_ats_score_for_manual_user(self.session,user_id,manual.job_post_id)
|
|
if score:
|
|
payload["ai_score"]=score["overall_score"]
|
|
payload["recommendation"]=self._recommendation(score["overall_score"])
|
|
payload["scored_at"]=score["computed_at"]
|
|
payload["candidate_id"]=score.get("candidate_id")
|
|
if score.get("user_id") and not payload.get("user_id"):
|
|
payload["user_id"]=score["user_id"]
|
|
if score.get("job_post_id"):
|
|
payload["scored_job_post_id"]=score["job_post_id"]
|
|
return payload
|
|
return await self.attach_job_posts(rows)
|
|
except HTTPException:
|
|
raise
|
|
except Exception as e:
|
|
raise HTTPException(status_code=500,detail=str(e))
|
|
|
|
async def count_candidates(self,user_id=None,search=None):
|
|
try:
|
|
return await Inbox.count_candidate_profiles(session=self.session,user_id=user_id,search=search)
|
|
except HTTPException:
|
|
raise
|
|
except Exception as e:
|
|
raise HTTPException(status_code=500,detail=str(e))
|
|
|
|
async def update_candidate(self,user_id,payload,current_user=None):
|
|
try:
|
|
if not user_id:
|
|
raise HTTPException(status_code=400,detail="user_id is required")
|
|
fields={k:v for k,v in (payload or {}).items() if k in ("favorite","rating") and v is not None}
|
|
if not fields:
|
|
raise HTTPException(status_code=400,detail="favorite or rating is required")
|
|
links=await Inbox.get_candidate_profile(session=self.session,user_id=user_id,limit=100,offset=0)
|
|
records=links if isinstance(links,list) else ([links] if links else [])
|
|
if not records:
|
|
raise HTTPException(status_code=404,detail="Candidate not found")
|
|
first=records[0]
|
|
old_favorite=getattr(first,"favorite",None)
|
|
old_rating=getattr(first,"rating",None)
|
|
for link in records:
|
|
await Inbox.update_inbox(self.session,link.id,fields)
|
|
if "favorite" in fields and fields["favorite"]!=old_favorite:
|
|
await HistoryRecorder(self.session).record(
|
|
HistoryEvent.FAVORITE_CHANGED.value,
|
|
current_user=current_user,user_id=user_id,inbox_id=first.id,
|
|
entity_type="candidate",entity_id=user_id,
|
|
from_value=old_favorite,to_value=fields["favorite"],commit=True,
|
|
)
|
|
if "rating" in fields and fields["rating"]!=old_rating:
|
|
await HistoryRecorder(self.session).record(
|
|
HistoryEvent.RATING_CHANGED.value,
|
|
current_user=current_user,user_id=user_id,inbox_id=first.id,
|
|
entity_type="candidate",entity_id=user_id,
|
|
from_value=old_rating,to_value=fields["rating"],commit=True,
|
|
)
|
|
refreshed=await Inbox.get_candidate_profile(session=self.session,user_id=user_id,limit=100,offset=0)
|
|
return await self.attach_profile_detail(refreshed)
|
|
except HTTPException:
|
|
raise
|
|
except Exception as e:
|
|
raise HTTPException(status_code=500,detail=str(e))
|
|
|
|
@staticmethod
|
|
def _attach_job_post(data,payload,*,as_assigned=False):
|
|
"""Merge one serialized job post onto a candidate payload.
|
|
|
|
Pure, so the per-id path and the batched one below cannot drift. A missing
|
|
post attaches nothing, matching the old early return.
|
|
"""
|
|
if not isinstance(data,dict) or not payload:
|
|
return payload
|
|
if as_assigned:
|
|
data["assigned_job_post"]=payload
|
|
if payload.get("created_by_name"):
|
|
data["recruiter"]=payload.get("created_by_name")
|
|
data["recruiter_id"]=payload.get("created_by")
|
|
if payload.get("title"):
|
|
data["job_title"]=payload.get("title")
|
|
else:
|
|
data.setdefault("job_posts",[]).append(payload)
|
|
if data.get("recruiter") is None and payload.get("created_by_name"):
|
|
data["recruiter"]=payload.get("created_by_name")
|
|
data["recruiter_id"]=payload.get("created_by")
|
|
if data.get("job_title") is None and payload.get("title"):
|
|
data["job_title"]=payload.get("title")
|
|
return payload
|
|
|
|
async def _job_posts_by_id(self,ids):
|
|
"""Serialized job posts keyed by id — one query for a whole page of rows.
|
|
|
|
active_only=False mirrors get_job_post_by_id, which filters neither flag:
|
|
an application assigned to a closed post must keep its title.
|
|
"""
|
|
wanted=[]
|
|
seen=set()
|
|
for raw in ids or []:
|
|
key=str(raw) if raw else None
|
|
if not key or key in seen:
|
|
continue
|
|
seen.add(key)
|
|
wanted.append(key)
|
|
if not wanted:
|
|
return {}
|
|
rows=await JobPosts.get_by_ids(self.session,wanted,active_only=False)
|
|
return {str(r.id):serialize_job_post(r) for r in rows}
|
|
|
|
async def get_job_post_by_id(self,record_id,data=None,*,as_assigned=False):
|
|
"""Load full job_posts row and optionally append it onto a candidate payload."""
|
|
try:
|
|
job_post_data=await JobPosts.get_job_post_by_id(session=self.session,record_id=record_id)
|
|
if not job_post_data:
|
|
return None
|
|
return self._attach_job_post(data,serialize_job_post(job_post_data),as_assigned=as_assigned)
|
|
except Exception as e:
|
|
raise HTTPException(status_code=500,detail=str(e))
|
|
|
|
async def attach_job_posts(self,data):
|
|
"""Normalize list/single, serialize each record, attach full job_posts rows.
|
|
|
|
Three batched queries per page, not two per row — a 200-row pipeline board
|
|
issued 600+ sequential job_posts round trips before.
|
|
"""
|
|
from inbox.plugins import get_ats_scores_for_users
|
|
|
|
single=not isinstance(data,list)
|
|
records=[data] if single else list(data or [])
|
|
|
|
payloads=[]
|
|
wanted=[]
|
|
owners=[]
|
|
for record in records:
|
|
payload=serialize_candidate_profile(record)
|
|
payload["job_posts"]=[]
|
|
payload["assigned_job_post"]=None
|
|
score,band=self._score_from_message(record)
|
|
if score is not None:
|
|
payload["ai_score"]=score
|
|
payload["recommendation"]=band
|
|
payloads.append(payload)
|
|
if payload.get("user_id"):
|
|
owners.append(payload["user_id"])
|
|
if payload.get("assigned_job_post_id"):
|
|
wanted.append(payload["assigned_job_post_id"])
|
|
wanted.extend(payload.get("suggested_job_post_ids") or [])
|
|
|
|
# ats_results is the source the inbox denorm above is copied FROM, so a
|
|
# live score wins over it. A candidate with neither keeps ai_score None —
|
|
# the list rows must not invent a number the scoring engine never produced.
|
|
ats=await get_ats_scores_for_users(self.session,owners)
|
|
for payload in payloads:
|
|
row=ats.get(str(payload.get("user_id") or ""))
|
|
if row and row.get("overall_score") is not None:
|
|
payload["ai_score"]=row["overall_score"]
|
|
payload["recommendation"]=row.get("band") or self._recommendation(row["overall_score"])
|
|
|
|
posts=await self._job_posts_by_id(wanted)
|
|
# Copy per attach: two candidates on the same post held independent dicts
|
|
# back when every row re-serialized its own.
|
|
for payload in payloads:
|
|
assigned_id=payload.get("assigned_job_post_id")
|
|
if assigned_id:
|
|
post=posts.get(str(assigned_id))
|
|
self._attach_job_post(payload,dict(post) if post else None,as_assigned=True)
|
|
for job_id in payload.get("suggested_job_post_ids") or []:
|
|
post=posts.get(str(job_id))
|
|
self._attach_job_post(payload,dict(post) if post else None)
|
|
return payloads[0] if single else payloads
|
|
|
|
async def attach_profile_detail(self,data):
|
|
"""Detail mode: flatten child collections across every Inbox row for the candidate."""
|
|
single=not isinstance(data,list)
|
|
records=[data] if single else list(data or [])
|
|
if not records:
|
|
return {} if single else []
|
|
|
|
interviews=[]
|
|
activity=[]
|
|
feedback=[]
|
|
documents=[]
|
|
job_posts=[]
|
|
assigned_job_post=None
|
|
base=None
|
|
user_id=None
|
|
favorite=None
|
|
rating=None
|
|
for record in records:
|
|
payload=serialize_candidate_profile(record,detail=True)
|
|
if base is None:
|
|
base=payload
|
|
user_id=payload.get("user_id")
|
|
favorite=payload.get("favorite")
|
|
rating=payload.get("rating")
|
|
interviews.extend(payload.get("interviews") or [])
|
|
activity.extend(payload.get("activity") or [])
|
|
feedback.extend(payload.get("feedback") or [])
|
|
documents.extend(payload.get("documents") or [])
|
|
if payload.get("assigned_job_post_id") and assigned_job_post is None:
|
|
await self.get_job_post_by_id(
|
|
record_id=payload.get("assigned_job_post_id"),
|
|
data=payload,
|
|
as_assigned=True,
|
|
)
|
|
assigned_job_post=payload.get("assigned_job_post")
|
|
if base.get("recruiter") is None and payload.get("recruiter"):
|
|
base["recruiter"]=payload.get("recruiter")
|
|
base["recruiter_id"]=payload.get("recruiter_id")
|
|
if base.get("job_title") is None and payload.get("job_title"):
|
|
base["job_title"]=payload.get("job_title")
|
|
for job_id in payload.get("suggested_job_post_ids") or []:
|
|
await self.get_job_post_by_id(record_id=job_id,data=payload)
|
|
for jp in payload.get("job_posts") or []:
|
|
if not any(x.get("id")==jp.get("id") for x in job_posts):
|
|
job_posts.append(jp)
|
|
if base.get("recruiter") is None and payload.get("recruiter"):
|
|
base["recruiter"]=payload.get("recruiter")
|
|
base["recruiter_id"]=payload.get("recruiter_id")
|
|
if base.get("job_title") is None and payload.get("job_title"):
|
|
base["job_title"]=payload.get("job_title")
|
|
|
|
notes=[]
|
|
uid=Notes._as_uuid(user_id) if user_id else None
|
|
if uid is not None:
|
|
result=await self.session.execute(
|
|
select(Notes)
|
|
.options(selectinload(Notes.author))
|
|
.where(Notes.user_id==uid)
|
|
.order_by(Notes.created_at.desc())
|
|
)
|
|
notes=[serialize_note(r) for r in result.scalars().all()]
|
|
|
|
# ATS score from inbox denorm / ats_results via Inbox.ats_id — never from
|
|
# a Candidates join on message id. Keywords live on the scored Candidates
|
|
# row: candidate_id when set, else email+job for the matched-user path.
|
|
ats_ids=[r.ats_id for r in records if getattr(r,"ats_id",None)]
|
|
ats_rows=[]
|
|
if ats_ids:
|
|
result=await self.session.execute(select(AtsResults).where(AtsResults.id.in_(ats_ids)))
|
|
ats_rows=list(result.scalars().all())
|
|
assigned_uid=AtsResults._as_uuid(base.get("assigned_job_post_id")) if base.get("assigned_job_post_id") else None
|
|
chosen=None
|
|
if assigned_uid is not None:
|
|
chosen=next((a for a in ats_rows if a.job_post_id==assigned_uid),None)
|
|
if chosen is None and ats_rows:
|
|
chosen=max(ats_rows,key=lambda a:a.computed_at)
|
|
if chosen is not None:
|
|
base["ai_score"]=chosen.overall_score
|
|
base["recommendation"]=chosen.band or self._recommendation(chosen.overall_score)
|
|
base["scored_job_post_id"]=str(chosen.job_post_id) if chosen.job_post_id else None
|
|
base["scored_at"]=chosen.computed_at.isoformat() if chosen.computed_at else None
|
|
base["candidate_id"]=str(chosen.candidate_id) if chosen.candidate_id else None
|
|
if chosen.user_id and not base.get("user_id"):
|
|
base["user_id"]=str(chosen.user_id)
|
|
scored=None
|
|
if chosen.candidate_id:
|
|
scored=await Candidates.get_candidate_by_id(self.session,str(chosen.candidate_id))
|
|
elif chosen.user_id:
|
|
owner=await Users.get_user_by_id(self.session,str(chosen.user_id))
|
|
if owner is not None:
|
|
scored=await Candidates.get_completed_by_email_job(self.session,owner.email,chosen.job_post_id)
|
|
if scored is not None:
|
|
base["matched_keywords"]=list(scored.matched_keywords or [])
|
|
base["missing_keywords"]=list(scored.missing_keywords or [])
|
|
base["summary_critique"]=scored.summary_critique
|
|
else:
|
|
for record in records:
|
|
score,band=self._score_from_message(record)
|
|
if score is not None:
|
|
base["ai_score"]=score
|
|
base["recommendation"]=band
|
|
break
|
|
|
|
activity.sort(key=lambda r:(r.get("activity_date") or ""),reverse=True)
|
|
base["favorite"]=favorite
|
|
base["rating"]=rating
|
|
base["interviews"]=interviews
|
|
base["activity"]=activity
|
|
base["feedback"]=feedback
|
|
base["documents"]=documents
|
|
base["notes"]=notes
|
|
base["job_posts"]=job_posts or base.get("job_posts") or []
|
|
base["assigned_job_post"]=assigned_job_post
|
|
if assigned_job_post:
|
|
base["assigned_job_post_id"]=assigned_job_post.get("id")
|
|
if assigned_job_post.get("created_by_name"):
|
|
base["recruiter"]=assigned_job_post.get("created_by_name")
|
|
base["recruiter_id"]=assigned_job_post.get("created_by")
|
|
if assigned_job_post.get("title"):
|
|
base["job_title"]=assigned_job_post.get("title")
|
|
return base
|
|
|
|
async def download_document(self,inbox_id=None,manual_upload_candidate_id=None,index=0):
|
|
has_inbox=inbox_id is not None and str(inbox_id).strip()!=""
|
|
has_manual=manual_upload_candidate_id is not None and str(manual_upload_candidate_id).strip()!=""
|
|
if has_inbox==has_manual:
|
|
raise HTTPException(status_code=404,detail="Not found")
|
|
try:
|
|
index=int(index or 0)
|
|
except (TypeError,ValueError):
|
|
raise HTTPException(status_code=404,detail="Not found")
|
|
if index<0:
|
|
raise HTTPException(status_code=404,detail="Not found")
|
|
|
|
docs=[]
|
|
if has_inbox:
|
|
link=await Inbox.get_inbox_with_message(self.session,inbox_id)
|
|
if not link or not link.messages:
|
|
raise HTTPException(status_code=404,detail="Not found")
|
|
docs=documents_from_message(link.messages.file_name,link.messages.file_path)
|
|
else:
|
|
row=await Manual_UPLOAD_CANDIDATE.get_by_id(self.session,manual_upload_candidate_id)
|
|
if not row:
|
|
raise HTTPException(status_code=404,detail="Not found")
|
|
docs=documents_from_message(row.file_name,row.file_path)
|
|
|
|
if index>=len(docs):
|
|
raise HTTPException(status_code=404,detail="Not found")
|
|
entry=docs[index]
|
|
path=contained_download_path(entry.get("path"))
|
|
if path is None:
|
|
raise HTTPException(status_code=404,detail="Not found")
|
|
name=(entry.get("name") or path.name).strip() or path.name
|
|
return path,name
|