292 lines
12 KiB
Python
292 lines
12 KiB
Python
"""Inbox Taskiq tasks — CV → job-post matching and ATS scoring."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import logging
|
|
import uuid
|
|
from datetime import datetime,timezone
|
|
|
|
from fastapi import HTTPException
|
|
|
|
from agent.execute_agent import run_agent
|
|
from db_setup import session_scope
|
|
from employment_agent.execute_agent import run_employment_agent
|
|
from inbox.models import Inbox_Messages,Inbox,AtsResults
|
|
from inbox.plugins import extract_resume_text
|
|
from job.job_post.models import JobPosts
|
|
from job.job_post.serializers import serialize_job_post
|
|
from taskiq_management.broker_setup import MAX_RETRIES,RETRY_DELAY,broker
|
|
from taskiq_management.middleware import PermanentTaskError
|
|
|
|
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,rescan_run_id=None) -> 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 an ats_results
|
|
row 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.views import CandidateScoring
|
|
|
|
try:
|
|
mid=uuid.UUID(str(record_id))
|
|
jid=uuid.UUID(str(job_id))
|
|
except ValueError:
|
|
raise PermanentTaskError("record_id and job_id must be uuids")
|
|
|
|
already=False
|
|
results=[]
|
|
async with session_scope() as session:
|
|
link=await Inbox.get_inbox_by_message_id(session,mid)
|
|
if link is not None:
|
|
# (inbox, job) only — candidate_id is NULL when the CV email matched
|
|
# a user, so already_scored must not depend on it.
|
|
existing=await AtsResults.get_for_inbox_job(session,link.id,jid)
|
|
if existing is not None:
|
|
already=True
|
|
if not already:
|
|
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)},rescan_run_id=rescan_run_id)
|
|
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
|
|
if already:
|
|
# Assigning a job already scored as a suggestion must still flip the
|
|
# denormed chip from max-of-suggestions to that job.
|
|
await denorm_message_ats_score(record_id)
|
|
return {"status":"already_scored"}
|
|
await denorm_message_ats_score(record_id)
|
|
return {"status":"scored","results":len(results)}
|
|
|
|
|
|
async def denorm_message_ats_score(record_id:str) -> None:
|
|
"""Stamp inbox_messages.ats_score: assigned job if set, else max of suggestions."""
|
|
async with session_scope() as session:
|
|
msg=await Inbox_Messages.get_inbox_message_by_id(session,record_id)
|
|
if msg is None:
|
|
return
|
|
link=await Inbox.get_inbox_by_message_id(session,record_id)
|
|
if link is None:
|
|
return
|
|
rows=await AtsResults.get_latest_by_job_for_inbox(session,link.id)
|
|
if not rows:
|
|
return
|
|
chosen=None
|
|
assigned=msg.assigned_job_post_id
|
|
if assigned:
|
|
chosen=next((r for r in rows if str(r.job_post_id)==str(assigned)),None)
|
|
if chosen is None:
|
|
chosen=max(rows,key=lambda r:float(r.overall_score or 0))
|
|
await Inbox_Messages.set_ats_score(session,record_id,chosen.overall_score,chosen.band)
|
|
|
|
|
|
async def score_message_against_jobs(record_id:str,job_ids) -> None:
|
|
"""Score one inbox CV against each job. Failures do not abort the rest."""
|
|
seen=set()
|
|
for raw in job_ids or []:
|
|
job_id=str(raw or "").strip()
|
|
if not job_id or job_id in seen:
|
|
continue
|
|
seen.add(job_id)
|
|
try:
|
|
outcome=await score_message_against_job(record_id,job_id)
|
|
logger.info("ats auto-score %s vs %s: %s",record_id,job_id,outcome.get("status"))
|
|
except Exception as exc:
|
|
logger.warning("ats auto-score failed for %s vs %s: %s",record_id,job_id,exc)
|
|
|
|
|
|
@broker.task(
|
|
task_name="g_sheet.score_form",
|
|
retry_on_error=True,
|
|
max_retries=MAX_RETRIES,
|
|
delay=RETRY_DELAY,
|
|
)
|
|
async def score_form_data(form_data_id:str,job_id:str) -> dict:
|
|
"""ATS-score one Sheet Forms CV (extracted_data) against one job post."""
|
|
from g_sheet.scoring import score_form_against_job
|
|
|
|
try:
|
|
uuid.UUID(str(form_data_id))
|
|
uuid.UUID(str(job_id))
|
|
except (TypeError,ValueError):
|
|
raise PermanentTaskError("form_data_id and job_id must be uuids")
|
|
result=await score_form_against_job(form_data_id,job_id)
|
|
if result.get("status")=="failed":
|
|
raise RuntimeError(result.get("error_code") or "form ats failed")
|
|
return result
|
|
|
|
|
|
@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.rescan_on_hold",
|
|
retry_on_error=True,
|
|
max_retries=MAX_RETRIES,
|
|
delay=RETRY_DELAY,
|
|
)
|
|
async def rescan_on_hold_run(run_id:str,cursor:int=0) -> dict:
|
|
"""Score On-Hold CVs against every job post, in short idempotent chunks."""
|
|
from g_sheet.scoring import score_form_against_job
|
|
from inbox.views import Email
|
|
|
|
async with session_scope() as session:
|
|
prepared=await Email(session=session).prepare_on_hold_rescan_chunk(run_id,cursor)
|
|
status=prepared.get("status")
|
|
if status in ("completed","failed","missing"):
|
|
return prepared
|
|
for item in prepared.get("batch") or []:
|
|
kind=item.get("kind")
|
|
record_id=item.get("record_id")
|
|
job_id=item.get("job_id")
|
|
try:
|
|
if kind=="form":
|
|
await score_form_against_job(record_id,job_id,run_id)
|
|
else:
|
|
await score_message_against_job(record_id,job_id,run_id)
|
|
except Exception as exc:
|
|
logger.warning("on-hold rescan failed for %s vs %s: %s",record_id,job_id,exc)
|
|
async with session_scope() as session:
|
|
return await Email(session=session).finish_on_hold_rescan_chunk(
|
|
run_id,prepared.get("next_cursor") or 0,bool(prepared.get("more")),
|
|
)
|
|
|
|
|
|
@broker.task(
|
|
task_name="inbox.match_message",
|
|
retry_on_error=True,
|
|
max_retries=MAX_RETRIES,
|
|
delay=RETRY_DELAY,
|
|
)
|
|
async def match_inbox_message(record_id:str,force:bool=False) -> dict:
|
|
if not record_id or not str(record_id).strip():
|
|
raise PermanentTaskError("record_id is required")
|
|
record_id=str(record_id).strip()
|
|
|
|
async with session_scope() as session:
|
|
row=await Inbox_Messages.get_inbox_message_by_id(session,record_id)
|
|
if not row:
|
|
raise PermanentTaskError(f"inbox message {record_id} not found")
|
|
if not force and row.match_status in _DONE:
|
|
return {"status":row.match_status,"skipped":True}
|
|
if not row.attachment or not row.file_path:
|
|
raise PermanentTaskError("message has no attachment to match")
|
|
|
|
paths=[p.strip() for p in row.file_path.split(",") if p.strip()]
|
|
subject=row.message_subject or ""
|
|
body=row.message_body or ""
|
|
row.match_status="processing"
|
|
row.match_error=None
|
|
row.matched_at=datetime.now(timezone.utc)
|
|
session.add(row)
|
|
await session.commit()
|
|
|
|
posts=await JobPosts.get_active_job_posts(session)
|
|
job_posts=[serialize_job_post(p) for p in posts]
|
|
|
|
text,extract_err=await extract_resume_text(paths)
|
|
if not text:
|
|
async with session_scope() as session:
|
|
await Inbox_Messages.set_match_result(
|
|
session,record_id,status="no_text",error=extract_err or "no text extracted",
|
|
)
|
|
return {"status":"no_text","error":extract_err}
|
|
|
|
result=await run_agent(subject=subject,resume_text=text,job_posts=job_posts)
|
|
status=result.get("status") or "failed"
|
|
# Extract the profile even when matching finds no job — On-Hold / unassigned
|
|
# CVs still need name, title, years, and phone on every screen.
|
|
fields=await run_employment_agent(
|
|
resume_text=text if not body else f"{text}\n\n{body}",
|
|
)
|
|
current_employment=fields["current_employment"]
|
|
education=fields["education"]
|
|
current_title=fields["current_title"]
|
|
linkedin_url=fields["linkedin_url"]
|
|
phone=fields["phone"]
|
|
city=fields.get("city") or None
|
|
years=fields.get("years_experience")
|
|
experience=(result.get("experience") or "").strip()
|
|
if not experience and years is not None:
|
|
experience=str(int(years)) if isinstance(years,(int,float)) and not isinstance(years,bool) else str(years)
|
|
|
|
if status=="failed":
|
|
async with session_scope() as session:
|
|
await Inbox_Messages.set_match_result(
|
|
session,
|
|
record_id,
|
|
resume_text=text,
|
|
experience=experience,
|
|
candidate_phone_number=phone if phone else "",
|
|
current_employment=current_employment,
|
|
current_title=current_title,
|
|
candidate_education=education,
|
|
linkedin_url=linkedin_url,
|
|
city=city,
|
|
suggested_job_post_ids=[],
|
|
summary=result.get("summary") or "",
|
|
reasoning=result.get("reasoning") or "",
|
|
status="failed",
|
|
error=result.get("error") or "agent returned failed status",
|
|
)
|
|
raise RuntimeError(result.get("error") or "agent returned failed status")
|
|
|
|
async with session_scope() as session:
|
|
await Inbox_Messages.set_match_result(
|
|
session,
|
|
record_id,
|
|
resume_text=text,
|
|
experience=experience,
|
|
candidate_phone_number=phone if phone else "",
|
|
current_employment=current_employment,
|
|
current_title=current_title,
|
|
candidate_education=education,
|
|
linkedin_url=linkedin_url,
|
|
city=city,
|
|
suggested_job_post_ids=result.get("suggested_job_post_ids") or [],
|
|
summary=result.get("summary") or "",
|
|
reasoning=result.get("reasoning") or "",
|
|
status=status,
|
|
error=result.get("error") or "",
|
|
)
|
|
# Auto-score every suggested job. The list chip is the max until a recruiter
|
|
# assigns one; assignment then re-runs ATS against that job_post_id.
|
|
# Scoring failures must not fail the match; the match result is committed above.
|
|
suggested=[str(j) for j in (result.get("suggested_job_post_ids") or []) if j]
|
|
score_job_ids=list(suggested)
|
|
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:
|
|
assigned=str(fresh.assigned_job_post_id)
|
|
if assigned not in score_job_ids:
|
|
score_job_ids.append(assigned)
|
|
await score_message_against_jobs(record_id,score_job_ids)
|
|
|
|
return {
|
|
"status":status,
|
|
"suggested_job_post_ids":result.get("suggested_job_post_ids") or [],
|
|
"current_employment":current_employment,
|
|
"current_title":current_title,
|
|
"education":education,
|
|
"linkedin_url":linkedin_url,
|
|
}
|