diff --git a/backend/inbox/tasks.py b/backend/inbox/tasks.py index ae48a0b..04b4480 100644 --- a/backend/inbox/tasks.py +++ b/backend/inbox/tasks.py @@ -1,10 +1,13 @@ -"""Inbox Taskiq tasks — CV → job-post matching.""" +"""Inbox Taskiq tasks — CV → job-post matching and ATS scoring.""" from __future__ import annotations import logging from datetime import datetime,timezone +from fastapi import HTTPException +from sqlalchemy import select + from agent.execute_agent import run_agent from db_setup import session_scope from employment_agent.execute_agent import run_employment_agent @@ -19,6 +22,58 @@ logger=logging.getLogger("inbox.tasks") _DONE=frozenset({"matched","skipped","no_text","failed","dlq"}) +async def score_message_against_job(record_id:str,job_id:str) -> dict: + """ATS-score one inbox CV against one job post — the no-upload path. + + The decoded attachment already on disk is the CV; the job post in the + database is the JD. Idempotent: a (message, job) pair with a completed + score is never paid for twice; re-runs are a no-op. + """ + # Lazy imports: inbox.plugins imports job.candidate.views, so a top-level + # import here would be circular. + from job.candidate.models import Candidates + from job.candidate.views import CandidateScoring + + mid=Candidates._as_uuid(record_id) + jid=Candidates._as_uuid(job_id) + if mid is None or jid is None: + raise PermanentTaskError("record_id and job_id must be uuids") + + async with session_scope() as session: + existing=await session.execute( + select(Candidates).where( + Candidates.inbox_message_id==mid, + Candidates.job_id==jid, + Candidates.status=="completed", + ) + ) + if existing.scalars().first() is not None: + return {"status":"already_scored"} + job=await JobPosts.get_job_post_by_id(session,job_id) + if job is None or job.is_deleted: + raise PermanentTaskError("job post missing or deleted") + service=CandidateScoring(session=session) + try: + # Attribute the rows to the job's owner — there is no request user + # in a background task. + results=await service.score_inbox(job_id,[record_id],{"id":str(job.created_by)}) + except HTTPException as exc: + # 400/404 from score_inbox are permanent (no attachment, bad ids); + # retrying cannot fix them. + raise PermanentTaskError(str(exc.detail)) from exc + return {"status":"scored","results":len(results)} + + +@broker.task( + task_name="inbox.score_message", + retry_on_error=True, + max_retries=MAX_RETRIES, + delay=RETRY_DELAY, +) +async def score_inbox_message(record_id:str,job_id:str) -> dict: + return await score_message_against_job(record_id,job_id) + + @broker.task( task_name="inbox.match_message", retry_on_error=True, @@ -81,6 +136,24 @@ async def match_inbox_message(record_id:str,force:bool=False) -> dict: status=status, error=result.get("error") or "", ) + # Auto-score: the match just paired this CV with jobs, so run the ATS on the + # spot — assigned job first, else the agent's top suggestion. Scoring failures + # must not fail the match; the match result is already committed above. + suggested=[str(j) for j in (result.get("suggested_job_post_ids") or []) if j] + score_job_id=None + async with session_scope() as session: + fresh=await Inbox_Messages.get_inbox_message_by_id(session,record_id) + if fresh is not None and fresh.assigned_job_post_id: + score_job_id=str(fresh.assigned_job_post_id) + if score_job_id is None and suggested: + score_job_id=suggested[0] + if score_job_id: + try: + outcome=await score_message_against_job(record_id,score_job_id) + logger.info("ats auto-score %s vs %s: %s",record_id,score_job_id,outcome.get("status")) + except Exception as exc: + logger.warning("ats auto-score failed for %s vs %s: %s",record_id,score_job_id,exc) + return { "status":status, "suggested_job_post_ids":result.get("suggested_job_post_ids") or [], diff --git a/backend/inbox/views.py b/backend/inbox/views.py index 0b87e4d..873d88d 100644 --- a/backend/inbox/views.py +++ b/backend/inbox/views.py @@ -179,6 +179,20 @@ class Email: updated=await Inbox_Messages.set_assigned_job_post(self.session,record_id,job_post_id) if not updated: raise HTTPException(status_code=404,detail="Message not found") + if job_post_id is not None: + # Assignment pairs this CV with a JD we already have — queue the ATS + # score in the background so the recruiter is not held on an OpenAI + # call. Idempotent server-side; broker-down just logs (the profile's + # Score-with-ATS button remains the manual fallback). + from inbox.tasks import score_inbox_message + try: + await score_inbox_message.kicker().with_labels( + created_at=datetime.now(timezone.utc).isoformat(), + correlation_id=str(record_id), + queue="inbox", + ).kiq(str(record_id),str(job_post_id)) + except Exception as exc: + logger.warning("could not queue ats score for %s: %s",record_id,exc) return await self.get_inbox_message_by_id(record_id) async def mark_read(self,record_id):