From a68d4bc2f6ebcd2209e7dd50ca3c59d6f0c46ffd Mon Sep 17 00:00:00 2001 From: Talha Ahmed Date: Mon, 10 Aug 2026 21:44:04 +0500 Subject: [PATCH] Integrate bulk-ats scoring into backend: candidates table + scoring routes Embeds the tested bulk-ats engine (pip install -e ., app.services.*) rather than duplicating it. New Candidates table (auto-migrated) persists one row per CV per job, deduped on (job_id, content_sha256) so re-scores update in place. CandidateScoring service extracts (pypdf off-loop), despaces, scores with bounded concurrency + cache priming via llm_setup's shared client, and isolates per-file failures as failed rows. Routes: /candidate/score (uploads), /candidate/score_inbox (decoded attachments), /candidate/fetch, /candidate/fetch_by_id. Complements the agent inbox-match flow: it routes, this scores. Co-Authored-By: Claude Fable 5 --- backend/.env.example | 10 ++ backend/job/app.py | 95 +++++++++--- backend/job/candidate/models.py | 125 +++++++++++++++- backend/job/candidate/plugins.py | 81 ++++++++++- backend/job/candidate/serializers.py | 25 ++++ backend/job/candidate/views.py | 210 ++++++++++++++++++++++++++- backend/requirements.txt | 6 + 7 files changed, 524 insertions(+), 28 deletions(-) diff --git a/backend/.env.example b/backend/.env.example index 32f5440..438ab71 100644 --- a/backend/.env.example +++ b/backend/.env.example @@ -44,6 +44,16 @@ OPENAI_BASE_URL= OPENAI_ORGANIZATION= OPENAI_PROJECT= +# ATS scoring (bulk-ats engine embedded via `pip install -e ..`). +# OPENAI_API_KEY / OPENAI_MODEL / OPENAI_MAX_OUTPUT_TOKENS above are shared. +OPENAI_EFFORT=low +OPENAI_ENABLE_PROMPT_CACHE=true +SCORING_CONCURRENCY=5 +MAX_RESUMES_PER_REQUEST=50 +MAX_PDF_SIZE_MB=10 +MAX_JD_CHARS=30000 +MAX_RESUME_CHARS=60000 + REDIS_URL=redis://localhost:6379/0 TASKIQ_QUEUE_NAME=inbox TASKIQ_MAX_RETRIES=3 diff --git a/backend/job/app.py b/backend/job/app.py index 682bb75..2eefa7f 100644 --- a/backend/job/app.py +++ b/backend/job/app.py @@ -2,13 +2,14 @@ from fastapi import APIRouter,Depends,Query from fastapi.responses import JSONResponse from fastapi import HTTPException from db_setup import get_session -from job.candidate.views import FileRead +from job.candidate.views import CandidateScoring,FileRead from sqlalchemy.ext.asyncio import AsyncSession from users.permissions import PermissionTag, require_permission from job.job_post.views import JobPost,JobPostCreate import logging from job.job_post.plugins import PlatformAlias -from fastapi import UploadFile, File +from fastapi import UploadFile, File, Form +from pydantic import BaseModel from dotenv import load_dotenv from datetime import datetime, time, timezone load_dotenv() @@ -102,21 +103,75 @@ async def buffer_channels( except Exception as e: raise HTTPException(status_code=500,detail=str(e)) -# @router.get("/candidate/fetch") -# async def fetch_candidate( -# user_id:str=Query(None), -# current_user: dict = Depends(require_permission(PermissionTag.CANDIDATES_VIEW)), -# session: AsyncSession = Depends(get_session), -# ): -# try: -# service=CandidateView(session=session) -# if user_id: -# data=await service.get_candidate(user_id=user_id) -# else: -# data=await service.get_candidate() -# return JSONResponse(content={"data":data,"status_code":200}) -# except HTTPException: -# raise -# except Exception as e: -# raise HTTPException(status_code=500,detail=str(e)) - \ No newline at end of file +class InboxScoreRequest(BaseModel): + job_id: str + message_ids: list[str] # inbox_messages PK uuids, not Graph message ids + + +@router.post("/candidate/score") +async def score_candidates( + job_id: str = Form(...), + files: list[UploadFile] = File(...), + current_user: dict = Depends(require_permission(PermissionTag.CANDIDATES_CREATE)), + session: AsyncSession = Depends(get_session), +): + """Score uploaded CV PDFs against a job post; persists and returns the leaderboard.""" + try: + pairs=[(f.filename,await f.read()) for f in files] + service=CandidateScoring(session=session) + data=await service.score_uploads(job_id,pairs,current_user) + return JSONResponse(content={"data":data,"total":len(data),"status_code":200}) + except HTTPException: + raise + except Exception as e: + raise HTTPException(status_code=500,detail=str(e)) + + +@router.post("/candidate/score_inbox") +async def score_inbox_candidates( + payload: InboxScoreRequest, + current_user: dict = Depends(require_permission(PermissionTag.CANDIDATES_CREATE)), + session: AsyncSession = Depends(get_session), +): + """Score the decoded attachments of inbox messages against a job post.""" + try: + service=CandidateScoring(session=session) + data=await service.score_inbox(payload.job_id,payload.message_ids,current_user) + return JSONResponse(content={"data":data,"total":len(data),"status_code":200}) + except HTTPException: + raise + except Exception as e: + raise HTTPException(status_code=500,detail=str(e)) + + +@router.get("/candidate/fetch") +async def fetch_candidates( + job_id: str = Query(...), + current_user: dict = Depends(require_permission(PermissionTag.CANDIDATES_VIEW)), + session: AsyncSession = Depends(get_session), +): + """Persisted leaderboard for a job: completed by score desc, failures last.""" + try: + service=CandidateScoring(session=session) + data=await service.fetch_candidates(job_id) + return JSONResponse(content={"data":data,"total":len(data),"status_code":200}) + except HTTPException: + raise + except Exception as e: + raise HTTPException(status_code=500,detail=str(e)) + + +@router.get("/candidate/fetch_by_id") +async def fetch_candidate_by_id( + candidate_id: str = Query(...), + current_user: dict = Depends(require_permission(PermissionTag.CANDIDATES_VIEW)), + session: AsyncSession = Depends(get_session), +): + try: + service=CandidateScoring(session=session) + data=await service.fetch_candidate_by_id(candidate_id) + return JSONResponse(content={"data":data,"total":1,"status_code":200}) + except HTTPException: + raise + except Exception as e: + raise HTTPException(status_code=500,detail=str(e)) diff --git a/backend/job/candidate/models.py b/backend/job/candidate/models.py index 482916b..3f5bb4f 100644 --- a/backend/job/candidate/models.py +++ b/backend/job/candidate/models.py @@ -1,6 +1,121 @@ -# from sqlmodel import SQLModel, Field -# from uuid import UUID, uuid4 -# from datetime import datetime -# from enum import Enum +import uuid +from datetime import datetime, timezone -# class CV_extraction(SQLModel,table=True): +from sqlalchemy import JSON, DateTime, UniqueConstraint +from sqlalchemy.exc import IntegrityError +from sqlalchemy.ext.asyncio import AsyncSession +from sqlmodel import Field, SQLModel, select + + +def _now() -> datetime: + return datetime.now(timezone.utc) + + +class Candidates(SQLModel, table=True): + """One scored (or failed-to-score) CV against one job post. + + The dedupe key is (job_id, content_sha256), not the filename: inbox attachments + are stored by basename so different candidates can collide on "resume.pdf", while + identical bytes can arrive via both upload and email. Re-scoring the same bytes + against the same job updates the existing row (fresh model output, updated_at + bumped) instead of duplicating it. content_sha256 is NULL when the file bytes + were never readable (missing on disk); NULLs never conflict in the unique index. + """ + + __tablename__ = "candidates" + __table_args__ = (UniqueConstraint("job_id", "content_sha256"),) + + id: uuid.UUID = Field(default_factory=uuid.uuid4, primary_key=True) + job_id: uuid.UUID = Field(foreign_key="job_posts.id", index=True) + source: str = Field(default="upload") # "upload" | "inbox" + inbox_message_id: uuid.UUID | None = Field(default=None, foreign_key="inbox_messages.id") + filename: str + file_path: str | None = Field(default=None) # decoded-attachment path (inbox only) + content_sha256: str | None = Field(default=None, index=True) + + candidate_name: str | None = Field(default=None) + job_title: str | None = Field(default=None) + current_company: str | None = Field(default=None) + years_experience: int | None = Field(default=None) + match_score: int | None = Field(default=None) # None on failed rows + matched_keywords: list[str] = Field(default_factory=list, sa_type=JSON) + missing_keywords: list[str] = Field(default_factory=list, sa_type=JSON) + summary_critique: str | None = Field(default=None) + + status: str # "completed" | "failed" + error_code: str | None = Field(default=None) + error_message: str | None = Field(default=None) + model: str | None = Field(default=None) # which OPENAI_MODEL produced the score + + created_by: uuid.UUID = Field(foreign_key="users.id") + created_at: datetime = Field(default_factory=_now, sa_type=DateTime(timezone=True)) + updated_at: datetime = Field(default_factory=_now, sa_type=DateTime(timezone=True)) + + @staticmethod + def _as_uuid(record_id) -> uuid.UUID | None: + try: + return uuid.UUID(str(record_id)) + except ValueError: + return None + + @classmethod + async def get_candidate_by_id(cls, session: AsyncSession, record_id: str): + uid = cls._as_uuid(record_id) + if uid is None: + return None + result = await session.execute(select(cls).where(cls.id == uid)) + return result.scalars().first() + + @classmethod + async def get_candidates_by_job(cls, session: AsyncSession, job_id: str): + """Leaderboard order: completed by score desc, failures last, ties stable.""" + uid = cls._as_uuid(job_id) + if uid is None: + return [] + result = await session.execute( + select(cls) + .where(cls.job_id == uid) + .order_by( + cls.status.asc(), # "completed" < "failed" + cls.match_score.desc().nulls_last(), + cls.created_at.asc(), + ) + ) + return result.scalars().all() + + @classmethod + async def upsert_candidate(cls, session: AsyncSession, fields: dict): + existing = None + sha = fields.get("content_sha256") + if sha: + result = await session.execute( + select(cls).where(cls.job_id == fields["job_id"], cls.content_sha256 == sha) + ) + existing = result.scalars().first() + + if existing is None: + row = cls(**fields) + session.add(row) + try: + await session.commit() + except IntegrityError: + # A concurrent request inserted the same (job_id, sha) first; take over + # that row and update it instead. + await session.rollback() + result = await session.execute( + select(cls).where(cls.job_id == fields["job_id"], cls.content_sha256 == sha) + ) + existing = result.scalars().first() + if existing is None: + raise + else: + await session.refresh(row) + return row + + for key, value in fields.items(): + setattr(existing, key, value) + existing.updated_at = _now() + session.add(existing) + await session.commit() + await session.refresh(existing) + return existing diff --git a/backend/job/candidate/plugins.py b/backend/job/candidate/plugins.py index f545b15..29f55d8 100644 --- a/backend/job/candidate/plugins.py +++ b/backend/job/candidate/plugins.py @@ -1,4 +1,4 @@ -"""CV text cleanup helpers for the PDF extractor. +"""CV text cleanup, scoring-JD builder, and the ATS scorer singleton. Pure module: no FastAPI imports and no HTTPException. @@ -6,14 +6,93 @@ Designer-made resumes position every glyph individually, so pypdf hands back "S K I L L S" instead of "SKILLS". In that layout a single space is glyph padding and a run of two or more spaces is the real word gap, which is what the `despace_line` decorator keys off to rebuild readable lines. + +The scoring pieces reuse the bulk-ats engine, installed editable from the repo +root (`pip install -e ..` -> `import app.*`), and share llm_setup's process-wide +AsyncOpenAI client rather than opening a second connection pool. """ from __future__ import annotations import re +from app.core.config import Settings, get_settings +from app.services.llm import OpenAIScorer +from dotenv import load_dotenv + from job.candidate.decorators import despace_line, normalize_unicode +load_dotenv() + +# Backend-local per-candidate error code for cases the bulk-ats engine never sees +# (its uploads always have bytes; inbox attachments can vanish from disk). +FILE_NOT_FOUND = "FILE_NOT_FOUND" + +_scorer: OpenAIScorer | None = None + + +def get_scoring_settings() -> Settings: + """Validated scoring knobs (model family, token floor, size limits). + + Reads real env vars, which load_dotenv() above has populated from the nearest + .env (backend/.env, else the repo root). Shared names (OPENAI_MODEL, + OPENAI_MAX_OUTPUT_TOKENS) therefore match what llm_setup uses. + """ + return get_settings() + + +def get_scorer() -> OpenAIScorer: + """Process-wide scorer over llm_setup's shared AsyncOpenAI client. + + Lazy so that a missing/invalid OPENAI configuration surfaces on the first + scoring request, not at import; llm_setup.init_llm() in the app lifespan has + normally created and verified the client before this ever runs. + """ + global _scorer + if _scorer is None: + from llm_setup import get_client + + settings = get_scoring_settings() + _scorer = OpenAIScorer( + get_client(), + model=settings.openai_model, + max_output_tokens=settings.openai_max_output_tokens, + effort=settings.openai_effort, + enable_cache=settings.openai_enable_prompt_cache, + ) + return _scorer + + +def build_job_description(job) -> str: + """Deterministic scoring JD from a JobPosts row. + + Byte-stable per job: derived only from stored column values, in fixed order, + because OpenAI prompt caching works on exact prefix match — one volatile byte + (an id, a timestamp) would stop the whole batch from reusing the cache. + Deliberately excludes post_text (hashtags, salary, LinkedIn formatting) and + salary; list order is taken as stored. + """ + lines = [f"Job Title: {job.title}"] + if job.employment_type: + lines.append(f"Employment Type: {job.employment_type}") + if job.location: + lines.append(f"Location: {job.location}") + if job.experience_min is not None and job.experience_max is not None: + lines.append(f"Experience Required: {job.experience_min}-{job.experience_max} years") + elif job.experience_min is not None: + lines.append(f"Experience Required: {job.experience_min}+ years") + elif job.experience_max is not None: + lines.append(f"Experience Required: up to {job.experience_max} years") + if job.description: + lines += ["", "Description:", job.description.strip()] + if job.requirements: + lines += ["", "Mandatory Requirements:"] + lines += [f"- {item}" for item in job.requirements] + if job.optional_skills: + lines += ["", "Preferred (nice to have):"] + lines += [f"- {item}" for item in job.optional_skills] + return "\n".join(lines) + @normalize_unicode @despace_line diff --git a/backend/job/candidate/serializers.py b/backend/job/candidate/serializers.py index e69de29..5f45d4b 100644 --- a/backend/job/candidate/serializers.py +++ b/backend/job/candidate/serializers.py @@ -0,0 +1,25 @@ +def serialize_candidate(row) -> dict: + return { + "id": str(row.id), + "job_id": str(row.job_id), + "inbox_message_id": str(row.inbox_message_id) if row.inbox_message_id else None, + "source": row.source, + "filename": row.filename, + "file_path": row.file_path, + "content_sha256": row.content_sha256, + "candidate_name": row.candidate_name, + "job_title": row.job_title, + "current_company": row.current_company, + "years_experience": row.years_experience, + "match_score": row.match_score, + "matched_keywords": list(row.matched_keywords or []), + "missing_keywords": list(row.missing_keywords or []), + "summary_critique": row.summary_critique, + "status": row.status, + "error_code": row.error_code, + "error_message": row.error_message, + "model": row.model, + "created_by": str(row.created_by), + "created_at": row.created_at.isoformat() if row.created_at else None, + "updated_at": row.updated_at.isoformat() if row.updated_at else None, + } diff --git a/backend/job/candidate/views.py b/backend/job/candidate/views.py index 6b10aa3..c97cf82 100644 --- a/backend/job/candidate/views.py +++ b/backend/job/candidate/views.py @@ -1,12 +1,25 @@ from sqlalchemy.ext.asyncio import AsyncSession -import os,logging,io +import asyncio,dataclasses,hashlib,os,logging,io,uuid from datetime import datetime,timezone from fastapi import HTTPException from pypdf import PdfReader from sqlalchemy import select from sqlmodel import true +from app.core.errors import ATSError,ErrorCode +from app.models.scoring import CompletedCandidate +from app.services.pdf import extract_resume,sanitize_filename +from app.services.scoring import score_batch from inbox.models import Inbox_Messages -from job.candidate.plugins import normalize_spaced_text +from job.candidate.models import Candidates +from job.candidate.plugins import ( + FILE_NOT_FOUND, + build_job_description, + get_scorer, + get_scoring_settings, + normalize_spaced_text, +) +from job.candidate.serializers import serialize_candidate +from job.job_post.models import JobPosts class FileRead: def __init__(self,session:AsyncSession,filename=None,file=None): @@ -68,6 +81,199 @@ class FileRead: # get_subject=Inbox_Messages.candidate_x_inbox(self.session,self.candidate_id) # get_file= + +class CandidateScoring: + """ATS scoring of CVs against one job post, persisted to the candidates table. + + Complements the agent's inbox-match flow: the agent suggests WHICH job a CV is + for; this service scores HOW WELL a CV fits a chosen job (0-100 leaderboard). + + Deviation from the bulk-ats HTTP API (which rejects a whole batch with 415/413 + on a bad file): here per-file problems become persisted rows with + status="failed" so one broken attachment never sinks the rest of the batch. + Request-level errors (unknown job, too many files) still raise. + """ + + def __init__(self,session:AsyncSession): + self.session=session + + async def score_uploads(self,job_id,files,current_user): + """files: list of (filename, bytes) pairs from the route handler.""" + settings=get_scoring_settings() + if len(files)>settings.max_resumes_per_request: + raise HTTPException( + status_code=413, + detail=f"At most {settings.max_resumes_per_request} resumes per request", + ) + sources=[] + for filename,data in files: + source={ + "filename":filename or "resume.pdf", + "data":data, + "file_path":None, + "inbox_message_id":None, + "precheck":None, + } + if not (filename or "").lower().endswith(".pdf"): + source["precheck"]=(ErrorCode.UNSUPPORTED_FILE_TYPE,"Only PDF resumes are supported.") + elif len(data)>settings.max_pdf_size_bytes: + source["precheck"]=(ErrorCode.PAYLOAD_TOO_LARGE,"The file exceeds the size limit.") + sources.append(source) + return await self._score_and_persist(job_id,sources,"upload",current_user) + + async def score_inbox(self,job_id,message_ids,current_user): + """Score the decoded attachments of inbox messages (PK uuids, not Graph ids).""" + # Local import: inbox.plugins imports this module (FileRead), so a top-level + # import would be circular — same pattern as match_inbox_cv above. + from inbox.plugins import resolve_attachment_path + + sources=[] + for mid in message_ids: + row=await Inbox_Messages.get_inbox_message_by_id(self.session,mid) + if row is None: + raise HTTPException(status_code=404,detail=f"Inbox message {mid} not found") + if not row.file_path: + continue + for path_str in (p.strip() for p in row.file_path.split(",") if p.strip()): + path=resolve_attachment_path(path_str) + source={ + "filename":path.name, + "data":None, + "file_path":str(path), + "inbox_message_id":row.id, + "precheck":None, + } + suffix=path.suffix.lower() + if suffix in (".doc",".docx"): + source["precheck"]=(ErrorCode.UNSUPPORTED_FILE_TYPE,"DOC/DOCX extraction is not supported yet.") + elif suffix!=".pdf": + source["precheck"]=(ErrorCode.UNSUPPORTED_FILE_TYPE,"Only PDF resumes are supported.") + elif not path.is_file(): + source["precheck"]=(FILE_NOT_FOUND,"The decoded attachment is missing on disk.") + else: + try: + source["data"]=await asyncio.to_thread(path.read_bytes) + except OSError: + source["precheck"]=(FILE_NOT_FOUND,"The decoded attachment could not be read.") + sources.append(source) + if not sources: + raise HTTPException(status_code=400,detail="No attachments found for the given message(s)") + return await self._score_and_persist(job_id,sources,"inbox",current_user) + + async def fetch_candidates(self,job_id): + job=await JobPosts.get_job_post_by_id(self.session,job_id) + if job is None or job.is_deleted: + raise HTTPException(status_code=404,detail="Job post not found") + rows=await Candidates.get_candidates_by_job(self.session,job_id) + return [serialize_candidate(row) for row in rows] + + async def fetch_candidate_by_id(self,candidate_id): + row=await Candidates.get_candidate_by_id(self.session,candidate_id) + if row is None: + raise HTTPException(status_code=404,detail="Candidate not found") + return serialize_candidate(row) + + async def _score_and_persist(self,job_id,sources,source_kind,current_user): + job=await JobPosts.get_job_post_by_id(self.session,job_id) + if job is None or job.is_deleted: + raise HTTPException(status_code=404,detail="Job post not found") + settings=get_scoring_settings() + jd=build_job_description(job) + if len(jd)>settings.max_jd_chars: + raise HTTPException(status_code=422,detail="The job post is too large to score against") + + # Slot-indexed like app/api/routes.py: results merge back by position, never + # by filename — inbox attachments can share a basename. + results_by_slot={} + extracted=[] + for slot,source in enumerate(sources): + source["safe_name"]=sanitize_filename(source["filename"]) + data=source["data"] + source["sha256"]=hashlib.sha256(data).hexdigest() if data is not None else None + if source["precheck"] is not None: + code,message=source["precheck"] + results_by_slot[slot]=self._failed_fields(source,code,message) + continue + try: + # pypdf is CPU-bound: keep it off the event loop. Despace BEFORE + # scoring so keyword verification sees the exact text the model saw; + # ExtractedResume is frozen, hence dataclasses.replace. + resume=await asyncio.to_thread( + extract_resume,data,source["safe_name"],settings.max_resume_chars + ) + resume=dataclasses.replace(resume,text=normalize_spaced_text(resume.text)) + except ATSError as exc: + results_by_slot[slot]=self._failed_fields(source,exc.error_code,exc.public_message) + continue + extracted.append((slot,resume)) + + scored=await score_batch( + [resume for _,resume in extracted], + job_description=jd, + scorer=get_scorer(), + concurrency=settings.scoring_concurrency, + ) + for (slot,_),result in zip(extracted,scored,strict=True): + source=sources[slot] + if isinstance(result,CompletedCandidate): + results_by_slot[slot]={ + **self._base_fields(source), + "status":"completed", + "candidate_name":result.candidate_name, + "job_title":result.job_title, + "current_company":result.current_company, + "years_experience":result.years_experience, + "match_score":result.match_score, + "matched_keywords":result.matched_keywords, + "missing_keywords":result.missing_keywords, + "summary_critique":result.summary_critique, + "error_code":None, + "error_message":None, + } + else: + results_by_slot[slot]=self._failed_fields(source,result.error_code,result.error_message) + + common={ + "job_id":job.id, + "source":source_kind, + "created_by":uuid.UUID(str(current_user["id"])), + "model":settings.openai_model, + } + rows=[] + for slot in range(len(sources)): + fields={**results_by_slot[slot],**common} + rows.append(await Candidates.upsert_candidate(self.session,fields)) + + # Leaderboard order: completed by score desc, failures last, stable. + rows.sort(key=lambda r:(0,-(r.match_score or 0)) if r.status=="completed" else (1,0)) + return [serialize_candidate(row) for row in rows] + + @staticmethod + def _base_fields(source): + return { + "inbox_message_id":source["inbox_message_id"], + "filename":source["safe_name"], + "file_path":source["file_path"], + "content_sha256":source["sha256"], + } + + @classmethod + def _failed_fields(cls,source,code,message): + return { + **cls._base_fields(source), + "status":"failed", + "error_code":str(code), + "error_message":message, + "candidate_name":None, + "job_title":None, + "current_company":None, + "years_experience":None, + "match_score":None, + "matched_keywords":[], + "missing_keywords":[], + "summary_critique":None, + } + # class CandidateView: # def __init__(self,session:AsyncSession): # self.session=session diff --git a/backend/requirements.txt b/backend/requirements.txt index 63ad39f..dfb3ac5 100644 --- a/backend/requirements.txt +++ b/backend/requirements.txt @@ -38,3 +38,9 @@ redis>=5.0,<6.0 # DLQ middleware (taskiq_management/middleware.py) as # --- LLM ------------------------------------------------------------------- openai==2.53.0 # AsyncOpenAI client in llm_setup.py langgraph==1.2.10 # StateGraph agent framework in agent/agent_setup.py + +# --- ATS scoring ----------------------------------------------------------- +# The bulk-ats scoring engine (app.services.pdf / llm / scoring) is installed +# editable from the repo root — run once per environment: +# pip install -e .. +# Its dependencies are already satisfied by the pins above.