HR-ATS-Portal/backend/inbox/tasks.py

"""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) -> 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")

    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:
                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,
    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"
    if status=="failed":
        raise RuntimeError(result.get("error") or "agent returned failed status")

    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"]

    async with session_scope() as session:
        await Inbox_Messages.set_match_result(
            session,
            record_id,
            resume_text=text,
            experience=result.get("experience") or "",
            candidate_phone_number=phone,
            current_employment=current_employment,
            current_title=current_title,
            candidate_education=education,
            linkedin_url=linkedin_url,
            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: 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 [],
        "current_employment":current_employment,
        "current_title":current_title,
        "education":education,
        "linkedin_url":linkedin_url,
    }