102 lines
3.5 KiB
Python
102 lines
3.5 KiB
Python
"""Inbox Taskiq tasks — CV → job-post matching."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import logging
|
|
from datetime import datetime, timezone
|
|
|
|
from db_setup import session_scope
|
|
from inbox.models import Inbox_Messages
|
|
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_STATUSES=frozenset({"matched","skipped","no_text","failed","dlq"})
|
|
|
|
|
|
@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_STATUSES:
|
|
logger.info("skip %s — already %s",record_id,row.match_status)
|
|
return {"status":row.match_status,"skipped":True}
|
|
|
|
if not row.attachment or not row.file_path:
|
|
raise PermanentTaskError("message has no attachment to match")
|
|
|
|
row.match_status="processing"
|
|
row.match_error=None
|
|
row.matched_at=datetime.now(timezone.utc)
|
|
session.add(row)
|
|
await session.commit()
|
|
await session.refresh(row)
|
|
|
|
paths=[p.strip() for p in (row.file_path or "").split(",") if p.strip()]
|
|
subject=row.message_subject or ""
|
|
|
|
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}
|
|
|
|
from agent.execute_agent import run_agent
|
|
|
|
async with session_scope() as session:
|
|
posts=await JobPosts.get_active_job_posts(session)
|
|
job_posts=[serialize_job_post(p) for p in posts]
|
|
|
|
try:
|
|
result=await run_agent(subject=subject,resume_text=text,job_posts=job_posts)
|
|
except Exception as exc:
|
|
logger.exception("agent failed for %s",record_id)
|
|
raise RuntimeError(f"agent matching failed: {exc}") from exc
|
|
|
|
status=result.get("status") or "failed"
|
|
error=result.get("error") or ""
|
|
|
|
if status=="failed":
|
|
raise RuntimeError(error or "agent returned failed status")
|
|
logger.info("agent result: %s,%s",result,result.get("experience"))
|
|
async with session_scope() as session:
|
|
await Inbox_Messages.set_match_result(
|
|
session,
|
|
record_id,
|
|
resume_text=text,
|
|
experience=result.get("experience") or "",
|
|
suggested_job_post_ids=result.get("suggested_job_post_ids") or [],
|
|
summary=result.get("summary") or "",
|
|
reasoning=result.get("reasoning") or "",
|
|
status=status,
|
|
error=error,
|
|
)
|
|
|
|
logger.info("matched inbox %s status=%s ids=%s",record_id,status,result.get("suggested_job_post_ids"))
|
|
return {
|
|
"status":status,
|
|
"suggested_job_post_ids":result.get("suggested_job_post_ids") or [],
|
|
}
|