194 lines
7.2 KiB
Python
194 lines
7.2 KiB
Python
"""ATS-score a form_data CV against every linked job post.
|
|
|
|
A form row can match many jobs (suggested_job_post_ids). Recruiter assignment
|
|
is assigned_job_post_id. If assigned is set, ATS scores only that job; else
|
|
every suggested id. Resume text comes from extracted_data. Each
|
|
(form_data_id, job_post_id) pair lands as its own ats_results row.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import logging
|
|
import uuid
|
|
from datetime import datetime, timezone
|
|
|
|
from app.models.scoring import CompletedCandidate
|
|
from app.services.pdf import ExtractedResume
|
|
from app.services.scoring import score_batch
|
|
from db_setup import session_scope
|
|
from g_sheet.models import FormData
|
|
from inbox.models import AtsResults, InboxRescanRun
|
|
from job.candidate.plugins import build_job_description, get_scorer, get_scoring_settings
|
|
from job.job_post.models import JobPosts
|
|
|
|
logger = logging.getLogger("g_sheet.scoring")
|
|
|
|
|
|
def resume_text_from_extracted(payload) -> str | None:
|
|
"""Usable CV text from form_data.extracted_data, or None."""
|
|
if not isinstance(payload, dict):
|
|
return None
|
|
if payload.get("status") != "completed":
|
|
return None
|
|
text = (payload.get("text") or "").strip()
|
|
return text or None
|
|
|
|
|
|
def _band(score) -> str:
|
|
if score is None:
|
|
return ""
|
|
return "Strong Match" if score >= 82 else "Potential Match" if score >= 65 else "Weak Match"
|
|
|
|
|
|
def serialize_form_ats(row) -> dict:
|
|
"""One current ats_results row for a Sheet Forms applicant."""
|
|
return {
|
|
"job_post_id": str(row.job_post_id) if row.job_post_id else None,
|
|
"overall_score": row.overall_score,
|
|
"band": row.band or None,
|
|
"professional_summary": row.professional_summary or None,
|
|
"computed_at": row.computed_at.isoformat() if row.computed_at else None,
|
|
}
|
|
|
|
|
|
async def enqueue_form_score(form_data_id, job_post_id) -> None:
|
|
"""Queue ATS for one form row against one job. Broker-down only logs."""
|
|
if not form_data_id or not job_post_id:
|
|
return
|
|
await enqueue_form_scores(form_data_id, [job_post_id])
|
|
|
|
|
|
async def enqueue_form_scores(form_data_id, job_post_ids) -> None:
|
|
"""Queue ATS for one form row against each job. Broker-down only logs."""
|
|
if not form_data_id:
|
|
return
|
|
from inbox.tasks import score_form_data
|
|
|
|
seen: set[str] = set()
|
|
for raw in job_post_ids or []:
|
|
job_id = str(raw or "").strip()
|
|
if not job_id or job_id in seen:
|
|
continue
|
|
seen.add(job_id)
|
|
try:
|
|
await score_form_data.kicker().with_labels(
|
|
created_at=datetime.now(timezone.utc).isoformat(),
|
|
correlation_id=str(form_data_id),
|
|
queue="inbox",
|
|
).kiq(str(form_data_id), job_id)
|
|
except Exception as exc:
|
|
logger.warning(
|
|
"could not queue form ats score for %s vs %s: %s",
|
|
form_data_id, job_id, exc,
|
|
)
|
|
|
|
|
|
async def enqueue_form_row_scores(form_row) -> None:
|
|
"""Queue ATS: assigned job only, else every suggested job."""
|
|
if form_row is None:
|
|
return
|
|
assigned=getattr(form_row,"assigned_job_post_id",None) or getattr(form_row,"job_post_id",None)
|
|
if assigned:
|
|
await enqueue_form_score(form_row.id,assigned)
|
|
return
|
|
job_ids=FormData.score_job_ids(form_row)
|
|
if not job_ids:
|
|
return
|
|
await enqueue_form_scores(form_row.id,job_ids)
|
|
|
|
|
|
async def score_form_against_job(form_data_id: str, job_id: str, rescan_run_id=None) -> dict:
|
|
"""Score one Sheet Forms CV against one job. Idempotent per (form, job)."""
|
|
try:
|
|
uuid.UUID(str(form_data_id))
|
|
uuid.UUID(str(job_id))
|
|
except (TypeError, ValueError):
|
|
return {"status": "skipped", "reason": "invalid_ids"}
|
|
|
|
async with session_scope() as session:
|
|
existing = await AtsResults.get_for_form_job(session, form_data_id, job_id)
|
|
if existing is not None:
|
|
return {"status": "already_scored"}
|
|
form_row, job = await FormData.get_with_job(session, form_data_id, job_id)
|
|
if form_row is None or job is None:
|
|
return {"status": "skipped", "reason": "no_join"}
|
|
settings = get_scoring_settings()
|
|
jd = build_job_description(job)
|
|
if len(jd) > settings.max_jd_chars:
|
|
return {"status": "skipped", "reason": "jd_too_large"}
|
|
stored_summary = (form_row.professional_summary or "").strip() or None
|
|
text = resume_text_from_extracted(form_row.extracted_data)
|
|
filename = (form_row.extracted_data or {}).get("filename") or "resume.pdf"
|
|
page_count = int((form_row.extracted_data or {}).get("page_count") or 1)
|
|
truncated = bool((form_row.extracted_data or {}).get("truncated"))
|
|
form_pk = form_row.id
|
|
job_pk = job.id
|
|
|
|
from summary_gate.execute_agent import allow_ats
|
|
if not await allow_ats(stored_summary, jd):
|
|
return {"status": "skipped", "reason": "not_suitable"}
|
|
if not text:
|
|
return {"status": "skipped", "reason": "no_extract"}
|
|
|
|
resume = ExtractedResume(
|
|
filename=str(filename),
|
|
candidate_id=str(form_pk),
|
|
text=text,
|
|
page_count=page_count,
|
|
truncated=truncated,
|
|
)
|
|
scored = await score_batch(
|
|
[resume],
|
|
job_description=jd,
|
|
scorer=get_scorer(),
|
|
concurrency=1,
|
|
)
|
|
result = scored[0] if scored else None
|
|
if not isinstance(result, CompletedCandidate):
|
|
error = getattr(result, "error_code", None) if result is not None else "MODEL_UNAVAILABLE"
|
|
logger.warning("form ats failed form_data=%s job=%s code=%s", form_data_id, job_id, error)
|
|
return {"status": "failed", "error_code": error}
|
|
|
|
summary = (result.professional_summary or "").strip() or None
|
|
run_id = None
|
|
if rescan_run_id not in (None, ""):
|
|
try:
|
|
run_id = uuid.UUID(str(rescan_run_id))
|
|
except (TypeError, ValueError):
|
|
run_id = None
|
|
|
|
async with session_scope() as session:
|
|
existing = await AtsResults.get_for_form_job(session, form_pk, job_pk)
|
|
if existing is not None:
|
|
return {"status": "already_scored"}
|
|
job = await JobPosts.get_job_post_by_id(session, job_pk)
|
|
if job is None or job.is_deleted:
|
|
return {"status": "skipped", "reason": "job_gone"}
|
|
await FormData.set_professional_summary(session, form_pk, summary)
|
|
await AtsResults.insert_result(session, {
|
|
"inbox_id": None,
|
|
"user_id": None,
|
|
"candidate_id": None,
|
|
"form_data_id": form_pk,
|
|
"job_post_id": job_pk,
|
|
"overall_score": float(result.match_score),
|
|
"band": _band(result.match_score),
|
|
"model_name": settings.openai_model,
|
|
"professional_summary": summary,
|
|
"rescan_run_id": run_id,
|
|
"is_current": True,
|
|
})
|
|
if run_id:
|
|
await InboxRescanRun.append_summary(session, run_id, {
|
|
"kind": "form",
|
|
"record_id": str(form_pk),
|
|
"job_post_id": str(job_pk),
|
|
"professional_summary": summary,
|
|
})
|
|
return {
|
|
"status": "scored",
|
|
"overall_score": result.match_score,
|
|
"band": _band(result.match_score),
|
|
"job_post_id": str(job_pk),
|
|
}
|