utopia-ai
/
HR-ATS-Portal
5
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

listy of pipeline done

pull/15/head
ahmed.mujtaba 2026-08-13 17:25:50 +05:00
parent 4f9b03bd4f
commit 0899342c34
14 changed files with 732 additions and 354 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

162
backend/inbox/models.py
View File

@ -1,5 +1,6 @@
import logging
import os
from shlex import join
import uuid
from datetime import datetime, timezone
from typing import Any, List, Optional
@ -46,23 +47,14 @@ class Inbox(SQLModel, table=True):
message_id: uuid.UUID | None = Field(default=None, foreign_key="inbox_messages.id")
messages: Optional["Inbox_Messages"] = Relationship(back_populates="inbox")
# tz-AWARE, matching every other timestamp the analytics layer filters on.
# A naive column here made asyncpg reject the aware UTC bounds that
# analytics/views.py builds, so /analytics/hiring-trend and /analytics/kpis
# both 500'd before the query ever reached Postgres.
created_at: datetime = Field(default_factory=_now, sa_type=DateTime(timezone=True))
updated_at: datetime = Field(default_factory=_now, sa_type=DateTime(timezone=True))
favorite: Optional[bool] = Field(default=False)
rating: Optional[float] = Field(default=0.0)
# The CURRENT ats_results row for this application. Written together with the
# supersede chain in CandidateScoring._sync_inbox_ats: every completed inbox
# score inserts an ats_results row and repoints this at it, so the score is
# one direct id join away instead of a filter on ats_results.is_current.
ats_id: uuid.UUID | None = Field(default=None, foreign_key="ats_results.id")
# selectin on one-to-many: joined would repeat the inbox row per child
interviews: List["Interviews"] = Relationship(
back_populates="inbox",
sa_relationship_kwargs={"lazy": "selectin"},
@ -80,6 +72,64 @@ class Inbox(SQLModel, table=True):
sa_relationship_kwargs={"lazy": "joined"},
)
@classmethod
async def get_all(cls,session:AsyncSession):
try:
from job.job_post.models import JobPosts
qry=(
select(
cls.id.label("inbox_id"),
cls.user_id,
Users.name,
Users.email,
Inbox_Messages.candidate_phone_number.label("phone"),
Inbox_Messages.assigned_job_post_id,
JobPosts.title,
AtsResults.id.label("ats_result_id"),
AtsResults.overall_score,
AtsResults.band,
AtsResults.job_post_id.label("ats_job_post_id"),
AtsResults.computed_at,
AtsResults.candidate_id,
AtsResults.user_id.label("ats_user_id"),
)
.join(Users,cls.user_id==Users.id)
.join(Roles,Users.role_id==Roles.id)
.join(Inbox_Messages,cls.message_id==Inbox_Messages.id)
.join(JobPosts,Inbox_Messages.assigned_job_post_id==JobPosts.id)
.outerjoin(AtsResults,cls.ats_id==AtsResults.id)
.where(Inbox_Messages.assigned_job_post_id.is_not(None))
.where(Roles.role_name==EnumRoles.CANDIDATE.value)
.order_by(cls.created_at.desc())
)
result=await session.execute(qry)
rows=[]
for row in result.mappings().all():
ats=None
if row["ats_result_id"] is not None:
ats={
"id":str(row["ats_result_id"]),
"overall_score":row["overall_score"],
"band":row["band"] or None,
"job_post_id":str(row["ats_job_post_id"]) if row["ats_job_post_id"] else None,
"computed_at":row["computed_at"].isoformat() if row["computed_at"] else None,
"candidate_id":str(row["candidate_id"]) if row["candidate_id"] else None,
"user_id":str(row["ats_user_id"]) if row["ats_user_id"] else None,
}
rows.append({
"inbox_id":row["inbox_id"],
"user_id":str(row["user_id"]) if row["user_id"] else None,
"name":row["name"],
"email":row["email"],
"phone":row["phone"],
"assigned_job_post_id":str(row["assigned_job_post_id"]) if row["assigned_job_post_id"] else None,
"title":row["title"],
"ats_result":ats,
})
return rows
except Exception as e:
raise HTTPException(status_code=500,detail=str(e))
@classmethod
def _candidate_search_filter(cls, search: str):
pattern = f"%{search}%"
@ -507,6 +557,18 @@ class Inbox_Messages(SQLModel, table=True):
result = await session.execute(select(cls).where(cls.id == uid))
return result.scalars().first()
@classmethod
async def set_ats_score(cls, session: AsyncSession, record_id, score, band):
row = await cls.get_inbox_message_by_id(session, record_id)
if not row:
return None
row.ats_score = float(score)
row.ats_band = band or ""
session.add(row)
await session.commit()
await session.refresh(row)
return row
@classmethod
async def set_assigned_job_post(cls, session: AsyncSession, record_id, job_post_id):
"""Set or clear assigned_job_post_id; returns the row or None if missing."""
@ -608,9 +670,11 @@ class AtsResults(SQLModel, table=True):
id: uuid.UUID = Field(default_factory=uuid.uuid4, primary_key=True)
# Inbox applications link here; NULL for upload-sourced scores, which have no
# inbox row. candidate_id links every score to the candidates row it scored.
# inbox row. Identity is XOR: matching users.email -> user_id (candidate_id
# NULL); otherwise candidate_id (user_id NULL). Blank email is the latter.
inbox_id: int | None = Field(default=None, index=True, foreign_key="inbox.id")
candidate_id: uuid.UUID | None = Field(default=None, index=True, foreign_key="candidates.id")
user_id: uuid.UUID | None = Field(default=None, index=True, foreign_key="users.id")
job_post_id: uuid.UUID | None = Field(default=None, foreign_key="job_posts.id")
overall_score: float = Field(default=0.0)
band: str = Field(default="")
@ -641,6 +705,41 @@ class AtsResults(SQLModel, table=True):
)
return result.scalars().first()
@classmethod
async def get_for_inbox_job(cls, session: AsyncSession, inbox_id, job_post_id):
"""Any score for this application against this job — current or superseded.
Inbox-side idempotency is (inbox_id, job_post_id) only. candidate_id may
be NULL when the CV email matched a user; do not key off it here.
"""
jid = cls._as_uuid(job_post_id)
if jid is None:
return None
result = await session.execute(
select(cls)
.where(cls.inbox_id == int(inbox_id), cls.job_post_id == jid)
.order_by(cls.computed_at.desc())
)
return result.scalars().first()
@classmethod
async def resolve_identity(cls, session: AsyncSession, email, candidate_id):
"""XOR identity for a score row from the scored candidate's email.
Matching users.email (case-insensitive) -> user_id, candidate_id NULL.
Missing/blank email or no user -> candidate_id, user_id NULL.
"""
normalized = (email or "").strip().lower()
if normalized:
user_id = (
await session.execute(
select(Users.id).where(func.lower(Users.email) == normalized)
)
).scalar_one_or_none()
if user_id is not None:
return {"candidate_id": None, "user_id": user_id}
return {"candidate_id": candidate_id, "user_id": None}
@classmethod
async def get_current_for_candidate(cls, session: AsyncSession, candidate_id):
"""Current row for an upload-sourced score, chained per candidates row —
@ -653,9 +752,52 @@ class AtsResults(SQLModel, table=True):
)
return result.scalars().first()
@classmethod
async def get_current_for_user(cls, session: AsyncSession, user_id, job_post_id=None):
"""Current upload-sourced score for a matched user, scoped per job."""
uid = cls._as_uuid(user_id)
if uid is None:
return None
qry = select(cls).where(cls.user_id == uid, cls.is_current == True) # noqa: E712
jid = cls._as_uuid(job_post_id) if job_post_id is not None else None
if jid is not None:
qry = qry.where(cls.job_post_id == jid)
result = await session.execute(qry.order_by(cls.computed_at.desc()))
return result.scalars().first()
@classmethod
async def insert_result(cls, session: AsyncSession, fields: dict):
"""Insert a score row and supersede the previous current one.
Inbox scores chain on inbox_id and repoint inbox.ats_id (candidate_id
may be NULL). Upload scores chain on candidate_id, or on (user_id,
job_post_id) when identity resolved to a user. Flush the INSERT first:
with no relationship() edge the unit of work emits the UPDATEs first,
and the FKs reject a pointer to a row not yet inserted."""
inbox_id = fields.get("inbox_id")
candidate_id = fields.get("candidate_id")
user_id = fields.get("user_id")
job_post_id = fields.get("job_post_id")
if inbox_id is not None:
prev = await cls.get_current_for_inbox(session, inbox_id)
elif candidate_id is not None:
prev = await cls.get_current_for_candidate(session, candidate_id)
elif user_id is not None:
prev = await cls.get_current_for_user(session, user_id, job_post_id)
else:
prev = None
row = cls(**fields)
session.add(row)
await session.flush()
if prev is not None:
prev.is_current = False
prev.superseded_by_id = row.id
session.add(prev)
if inbox_id is not None:
link = await Inbox.get_inbox_by_id(session, inbox_id)
if link is not None:
link.ats_id = row.id
link.updated_at = _now()
session.add(link)
await session.commit()
return row

131
backend/inbox/plugins.py
View File

@ -5,13 +5,17 @@ from __future__ import annotations
import base64
import os
import re
import uuid
from pathlib import Path
from urllib.parse import quote
import httpx
from dotenv import load_dotenv
from sqlalchemy.ext.asyncio import AsyncSession
from sqlmodel import select
from inbox.models import Inbox_Messages
from inbox.models import AtsResults, Inbox, Inbox_Messages
from job.candidate.models import Candidates, Manual_UPLOAD_CANDIDATE
from job.candidate.views import FileRead
load_dotenv()
@ -161,3 +165,128 @@ async def extract_resume_text(file_paths:list[str]) -> tuple[str,str]:
if not texts:
return "","; ".join(errors) if errors else "no text extracted from PDF"
return "\n\n---\n\n".join(texts),""
def _ats_score_payload(row):
if row is None:
return None
return {
"overall_score":row.overall_score,
"band":row.band or None,
"job_post_id":str(row.job_post_id) if row.job_post_id else None,
"computed_at":row.computed_at.isoformat() if row.computed_at else None,
"candidate_id":str(row.candidate_id) if row.candidate_id else None,
"user_id":str(row.user_id) if row.user_id else None,
}
async def get_ats_score_for_user(session:AsyncSession,user_id,job_post_id=None):
"""Current ats_results overall score for a candidate -> dict, or None.
Prefer ats_results.user_id (CV email matched that user). Fall back to the
inbox.user_id join for scores whose email did not match any user, where
candidate_id is set and user_id is NULL.
Pass job_post_id to pin one application when a candidate has several;
without it the newest current score across their applications wins.
"""
try:
uid=uuid.UUID(str(user_id))
except (TypeError,ValueError):
return None
direct=(
select(AtsResults)
.where(AtsResults.user_id==uid,AtsResults.is_current==True) # noqa: E712
.order_by(AtsResults.computed_at.desc())
)
if job_post_id:
try:
direct=direct.where(AtsResults.job_post_id==uuid.UUID(str(job_post_id)))
except (TypeError,ValueError):
return None
row=(await session.execute(direct)).scalars().first()
if row is not None:
return _ats_score_payload(row)
qry=(
select(AtsResults)
.join(Inbox,AtsResults.inbox_id==Inbox.id)
.join(Inbox_Messages,Inbox.message_id==Inbox_Messages.id)
.where(
Inbox.user_id==uid,
Inbox_Messages.assigned_job_post_id.is_not(None),
AtsResults.job_post_id==Inbox_Messages.assigned_job_post_id,
AtsResults.is_current==True, # noqa: E712
)
.order_by(AtsResults.computed_at.desc())
)
if job_post_id:
try:
qry=qry.where(Inbox_Messages.assigned_job_post_id==uuid.UUID(str(job_post_id)))
except (TypeError,ValueError):
return None
return _ats_score_payload((await session.execute(qry)).scalars().first())
async def get_ats_score_for_manual_user(session:AsyncSession,user_id,job_post_id=None):
"""Current ats_results overall score for an Add Candidate user -> dict, or None.
Prefer ats_results.user_id (Add Candidate always creates a users row, so a
later score against that email lands on user_id). Fall back to the
email+candidate_id join for rows written before user_id existed.
apply_via=manual_upload is the Add Candidate gate; /import never writes
that table.
Pass job_post_id to pin one application when a candidate has several;
without it the newest current score across their applications wins.
"""
try:
uid=uuid.UUID(str(user_id))
except (TypeError,ValueError):
return None
direct=(
select(AtsResults)
.where(AtsResults.user_id==uid,AtsResults.is_current==True) # noqa: E712
.order_by(AtsResults.computed_at.desc())
)
if job_post_id:
try:
jid=uuid.UUID(str(job_post_id))
except (TypeError,ValueError):
return None
direct=direct.where(AtsResults.job_post_id==jid)
row=(await session.execute(direct)).scalars().first()
if row is not None:
return _ats_score_payload(row)
qry=(
select(AtsResults)
.join(Candidates,AtsResults.candidate_id==Candidates.id)
.join(
Manual_UPLOAD_CANDIDATE,
(Candidates.job_id==Manual_UPLOAD_CANDIDATE.job_post_id)
&(Candidates.candidate_email==Manual_UPLOAD_CANDIDATE.candidate_email),
)
.where(
Manual_UPLOAD_CANDIDATE.user_id==uid,
Manual_UPLOAD_CANDIDATE.apply_via=="manual_upload",
Candidates.status=="completed",
AtsResults.job_post_id==Manual_UPLOAD_CANDIDATE.job_post_id,
AtsResults.is_current==True, # noqa: E712
)
.order_by(AtsResults.computed_at.desc())
)
if job_post_id:
try:
jid=uuid.UUID(str(job_post_id))
except (TypeError,ValueError):
return None
qry=qry.where(
Manual_UPLOAD_CANDIDATE.job_post_id==jid,
AtsResults.job_post_id==jid,
)
return _ats_score_payload((await session.execute(qry)).scalars().first())

32
backend/inbox/tasks.py
View File

@ -3,15 +3,15 @@
from __future__ import annotations
import logging
import uuid
from datetime import datetime,timezone
from fastapi import HTTPException
from sqlalchemy import select
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
from inbox.models import Inbox_Messages,Inbox,AtsResults
from inbox.plugins import extract_phone,extract_resume_text
from job.job_post.models import JobPosts
from job.job_post.serializers import serialize_job_post
@ -26,29 +26,27 @@ 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 a completed
score is never paid for twice; re-runs are a no-op.
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.models import Candidates
from job.candidate.views import CandidateScoring
mid=Candidates._as_uuid(record_id)
jid=Candidates._as_uuid(job_id)
if mid is None or jid is None:
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:
existing=await session.execute(
select(Candidates).where(
Candidates.inbox_message_id==mid,
Candidates.job_id==jid,
Candidates.status=="completed",
)
)
if existing.scalars().first() is not None:
return {"status":"already_scored"}
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")

22
backend/job/app.py
View File

@ -21,7 +21,8 @@ from dotenv import load_dotenv
from datetime import datetime, time, timezone
from pydantic import BaseModel
from uuid import UUID
from typing import Optional
import uuid
load_dotenv()
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)
@ -674,6 +675,25 @@ async def change_candidate_stage(
except Exception as e:
raise HTTPException(status_code=500,detail=str(e))
@router.get("/pipeline/candidates/fetch")
async def fetch_pipeline_candidates(
user_id:Optional[uuid.UUID]=Query(None),
job_post_id:Optional[uuid.UUID]=Query(None),
current_user: dict = Depends(require_permission(PermissionTag.PIPELINE_VIEW)),
session: AsyncSession = Depends(get_session),
):
try:
service=Pipeline(session=session)
if user_id and job_post_id:
data=await service.get_pipeline_candidates(user_id=user_id,job_post_id=job_post_id)
else:
data=await service.get_all()
return JSONResponse(content={"data":data,"status_code":200})
except HTTPException:
raise
except Exception as e:
raise HTTPException(status_code=500,detail=str(e))
@router.get("/pipeline/transitions/fetch")
async def fetch_pipeline_transitions(

117
backend/job/candidate/models.py
View File

@ -2,6 +2,7 @@ import uuid
from datetime import datetime, timezone
from typing import TYPE_CHECKING, List, Optional
from fastapi import HTTPException
from sqlalchemy import JSON, DateTime, func, UniqueConstraint
from sqlalchemy.exc import IntegrityError
from sqlalchemy.ext.asyncio import AsyncSession
@ -35,33 +36,86 @@ class Manual_UPLOAD_CANDIDATE(SQLModel, table=True):
# from job_posts.title — that is the role they applied to, not their own.
# Same ALTER-on-a-populated-table reasoning as referral_by below.
current_position: str = Field(default="", sa_column_kwargs={"server_default": ""})
# Add Candidate only (POST /candidate/create/candidate). /import never writes this table.
apply_via:Optional[str]=Field(nullable=True)
user_id: uuid.UUID | None = Field(default=None, foreign_key="users.id")
platform: str = Field(default="")
created_by: uuid.UUID | None = Field(default=None, foreign_key="users.id")
experience: str = Field(default="")
status: str = Field(default="")
# Free text, not a users FK: a referrer is often someone outside the system
# (a client, a former colleague), and recruiters type whatever the candidate
# told them. "" rather than NULL keeps it consistent with the columns above.
#
# server_default is load-bearing and NOT decoration, unlike the columns above
# — they arrived with the CREATE TABLE, this one arrives as an ALTER. The
# startup autogenerate would emit `ADD COLUMN referral_by VARCHAR NOT NULL`,
# which Postgres rejects outright on a table that already holds rows. The
# DEFAULT backfills them. Pass the bare "" — SQLAlchemy quotes a plain string
# into DEFAULT '', whereas "''" would render DEFAULT '''''' instead.
referral_by: str = Field(default="", sa_column_kwargs={"server_default": ""})
# The CV as uploaded: file_name is the recruiter-facing original, file_path
# the absolute location under inbox/decoded_attachments. They differ on
# purpose — the stored basename is uniquified so two candidates uploading
# "resume.pdf" cannot overwrite one another (see FileRead.save_manual_upload).
# Same ALTER-on-a-populated-table reasoning as referral_by above, so both
# carry a server default.
file_name: str = Field(default="", sa_column_kwargs={"server_default": ""})
file_path: str = Field(default="", sa_column_kwargs={"server_default": ""})
created_at: datetime = Field(default_factory=_now, sa_type=DateTime(timezone=True))
updated_at: datetime = Field(default_factory=_now, sa_type=DateTime(timezone=True))
@classmethod
async def get_all(cls, session: AsyncSession):
try:
from inbox.models import AtsResults
from users.models import Users
from job.job_post.models import JobPosts
qry=(
select(
cls.id,
cls.candidate_email,
cls.user_id,
Users.email,
cls.job_post_id,
Users.name,
cls.candidate_phone,
JobPosts.title,
cls.created_at,
cls.updated_at,
AtsResults.id.label("ats_result_id"),
AtsResults.overall_score,
AtsResults.band,
AtsResults.job_post_id.label("ats_job_post_id"),
AtsResults.computed_at,
AtsResults.candidate_id,
AtsResults.user_id.label("ats_user_id"),
)
.join(Users,cls.user_id==Users.id)
.join(JobPosts,cls.job_post_id==JobPosts.id)
.outerjoin(
AtsResults,
(Users.id==AtsResults.user_id)
&(AtsResults.job_post_id==cls.job_post_id)
&(AtsResults.is_current==True), # noqa: E712
)
.order_by(cls.created_at.desc())
)
result=await session.execute(qry)
rows=[]
for row in result.mappings().all():
ats=None
if row["ats_result_id"] is not None:
ats={
"id":str(row["ats_result_id"]),
"overall_score":row["overall_score"],
"band":row["band"] or None,
"job_post_id":str(row["ats_job_post_id"]) if row["ats_job_post_id"] else None,
"computed_at":row["computed_at"].isoformat() if row["computed_at"] else None,
"candidate_id":str(row["candidate_id"]) if row["candidate_id"] else None,
"user_id":str(row["ats_user_id"]) if row["ats_user_id"] else None,
}
rows.append({
"id":str(row["id"]) if row["id"] else None,
"candidate_email":row["candidate_email"],
"user_id":str(row["user_id"]) if row["user_id"] else None,
"email":row["email"],
"job_post_id":str(row["job_post_id"]) if row["job_post_id"] else None,
"name":row["name"],
"candidate_phone":row["candidate_phone"],
"title":row["title"],
"created_at":row["created_at"].isoformat() if row["created_at"] else None,
"updated_at":row["updated_at"].isoformat() if row["updated_at"] else None,
"ats_result":ats,
})
return rows
except Exception as e:
raise HTTPException(status_code=500,detail=str(e))
@staticmethod
def _as_uuid(record_id) -> uuid.UUID | None:
if record_id in (None, ""):
@ -103,6 +157,7 @@ class Manual_UPLOAD_CANDIDATE(SQLModel, table=True):
full_text=fields.get("full_text") or "",
current_company=(fields.get("current_company") or "").strip(),
current_position=(fields.get("current_position") or "").strip(),
apply_via="manual_upload",
user_id=user.id,
platform=(fields.get("platform") or "").strip(),
created_by=cls._as_uuid(fields.get("created_by")),
@ -129,27 +184,18 @@ class Manual_UPLOAD_CANDIDATE(SQLModel, table=True):
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")
source: str = Field(default="upload") # "upload" | "inbox" origin metadata, not an FK
filename: str
file_path: str | None = Field(default=None) # decoded-attachment path (inbox only)
content_sha256: str | None = Field(default=None, index=True)
candidate_email: str | None = Field(default=None)
candidate_name: str | None = Field(default=None)
job_title: str | None = Field(default=None)
current_company: str | None = Field(default=None)
@ -204,6 +250,25 @@ class Candidates(SQLModel, table=True):
result = await session.execute(statement)
return result.scalars().all()
@classmethod
async def get_completed_by_email_job(cls, session: AsyncSession, email, job_id):
"""Newest completed score for this email against this job — keywords when
ats_results.candidate_id is NULL (matched-user identity)."""
normalized = (email or "").strip().lower()
jid = cls._as_uuid(job_id)
if not normalized or jid is None:
return None
result = await session.execute(
select(cls)
.where(
func.lower(cls.candidate_email) == normalized,
cls.job_id == jid,
cls.status == "completed",
)
.order_by(cls.updated_at.desc())
)
return result.scalars().first()
@classmethod
async def upsert_candidate(cls, session: AsyncSession, fields: dict):
existing = None

43
backend/job/candidate/plugins.py
View File

@ -94,6 +94,49 @@ def build_job_description(job) -> str:
return "\n".join(lines)
def candidate_base_fields(source):
return {
"filename": source["safe_name"],
"file_path": source["file_path"],
"content_sha256": source["sha256"],
"candidate_email": source.get("candidate_email"),
}
def candidate_failed_fields(source, code, message):
return {
**candidate_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,
}
def candidate_completed_fields(source, result):
return {
**candidate_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,
}
@normalize_unicode
@despace_line
def normalize_spaced_text(text) -> str:

5
backend/job/candidate/serializers.py
View File

@ -11,11 +11,11 @@ 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_email": row.candidate_email,
"candidate_name": row.candidate_name,
"job_title": row.job_title,
"current_company": row.current_company,
@ -44,6 +44,7 @@ def serialize_manual_upload_candidate(row) -> Dict[str,Any]:
"full_text":row.full_text,
"current_company":row.current_company,
"current_position":row.current_position,
"apply_via":row.apply_via,
"user_id":str(row.user_id) if row.user_id else None,
"platform":row.platform,
"created_by":str(row.created_by) if row.created_by else None,
@ -72,6 +73,7 @@ def serialize_candidate_profile(
payload = {
"inbox_id": link.id,
"user_id": str(link.user_id) if link.user_id else None,
"candidate_id": None,
"name": user.name if user else None,
"email": user.email if user else None,
"is_active": user.is_active if user else None,
@ -138,6 +140,7 @@ def serialize_manual_candidate_profile(row, user, job_post) -> Dict[str, Any]:
return {
"inbox_id": None,
"user_id": str(user.id) if user else (str(row.user_id) if row.user_id else None),
"candidate_id": None,
"name": (user.name if user else None) or row.candidate_name or None,
"email": (user.email if user else None) or row.candidate_email or None,
"is_active": user.is_active if user else None,

341
backend/job/candidate/views.py
View File

@ -17,6 +17,8 @@ from job.candidate.models import Candidates
from job.candidate.plugins import (
FILE_NOT_FOUND,
build_job_description,
candidate_completed_fields,
candidate_failed_fields,
get_scorer,
get_scoring_settings,
normalize_spaced_text,
@ -272,7 +274,6 @@ class CandidateScoring:
"filename":filename or "resume.pdf",
"data":data,
"file_path":None,
"inbox_message_id":None,
"precheck":None,
}
if not (filename or "").lower().endswith(".pdf"):
@ -301,7 +302,8 @@ class CandidateScoring:
"filename":path.name,
"data":None,
"file_path":str(path),
"inbox_message_id":row.id,
"inbox_message_id":row.id, # call-scoped; not persisted on Candidates
"candidate_email":(row.message_from or "").strip().lower() or None,
"precheck":None,
}
suffix=path.suffix.lower()
@ -344,10 +346,24 @@ class CandidateScoring:
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")
fields_by_slot=await self._score_sources(sources,jd,settings)
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)):
rows.append(await Candidates.upsert_candidate(self.session,{**fields_by_slot[slot],**common}))
await self._sync_ats_results(source_kind,job,rows,sources)
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]
# Slot-indexed like app/api/routes.py: results merge back by position, never
# by filename — inbox attachments can share a basename.
results_by_slot={}
async def _score_sources(self,sources,jd,settings):
# Slot-indexed: results merge back by position, never by filename —
# inbox attachments can share a basename.
fields_by_slot={}
extracted=[]
for slot,source in enumerate(sources):
source["safe_name"]=sanitize_filename(source["filename"])
@ -355,21 +371,20 @@ class CandidateScoring:
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)
fields_by_slot[slot]=candidate_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))
if not source.get("candidate_email"):
detected,_=extract_candidate_email(resume.text)
source["candidate_email"]=detected
except ATSError as exc:
results_by_slot[slot]=self._failed_fields(source,exc.error_code,exc.public_message)
fields_by_slot[slot]=candidate_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,
@ -379,178 +394,80 @@ class CandidateScoring:
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,
}
fields_by_slot[slot]=candidate_completed_fields(source,result)
else:
results_by_slot[slot]=self._failed_fields(source,result.error_code,result.error_message)
fields_by_slot[slot]=candidate_failed_fields(source,result.error_code,result.error_message)
return fields_by_slot
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))
# Every completed score lands in ats_results. Inbox scores additionally
# denormalise onto inbox_messages / inbox (one sync per message: a
# multi-attachment mail keeps its best completed score); upload scores
# chain per candidates row instead — they have no inbox application.
async def _sync_ats_results(self,source_kind,job,rows,sources):
if source_kind=="inbox":
best={}
for row in rows:
if row.status=="completed" and row.inbox_message_id:
cur=best.get(row.inbox_message_id)
for source,row in zip(sources,rows):
mid=source.get("inbox_message_id")
if row.status=="completed" and mid:
cur=best.get(mid)
if cur is None or (row.match_score or 0)>(cur.match_score or 0):
best[row.inbox_message_id]=row
best[mid]=row
for message_id,row in best.items():
try:
await self._sync_inbox_ats(message_id,job,row)
except Exception:
# The candidates row is the primary outcome and is already
# committed; a denorm failure must not fail the scoring call.
await self.session.rollback()
logger.exception("inbox ATS denorm failed for message %s",message_id)
else:
for row in rows:
if row.status!="completed":
continue
try:
await self._sync_upload_ats(job,row)
except Exception:
await self.session.rollback()
logger.exception("upload ATS history failed for candidate %s",row.id)
# 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,
}
return
for row in rows:
if row.status!="completed":
continue
try:
await self._sync_upload_ats(job,row)
except Exception:
await self.session.rollback()
logger.exception("upload ATS history failed for candidate %s",row.id)
async def _sync_inbox_ats(self,message_id,job,row):
"""Land a completed score on the inbox tables (README "What is missing").
"""Land a completed score on inbox_messages / inbox / ats_results.
inbox_messages.ats_score / ats_band are what serialize_application renders
on the Applications tab; ats_results keeps the per-application history that
candidates' in-place upsert cannot. Precedence mirrors _recommendation: a
score against the assigned job always wins the denormalised columns, any
other job's score only lands while no completed assigned-job score exists.
The history row is appended regardless — it records the scoring event.
message_id is the scoring call's known inbox_messages PK, not a column
read off the Candidates row.
"""
msg=await Inbox_Messages.get_inbox_message_by_id(self.session,message_id)
if msg is None:
return
band=CandidateView._recommendation(row.match_score) or ""
denorm=True
assigned=msg.assigned_job_post_id
if assigned and str(assigned)!=str(job.id):
outranked=await self.session.execute(
select(Candidates).where(
Candidates.inbox_message_id==msg.id,
Candidates.job_id==assigned,
Candidates.status=="completed",
)
)
denorm=outranked.scalars().first() is None
if denorm:
msg.ats_score=float(row.match_score)
msg.ats_band=band
self.session.add(msg)
# ats_results hangs off the inbox JOIN row (int PK), which only exists once
# the sender is linked to a users account; without it there is no history row.
link=await Inbox.get_inbox_by_message_id(self.session,message_id)
if link is not None:
prev=await AtsResults.get_current_for_inbox(self.session,link.id)
entry=AtsResults(
inbox_id=link.id,
candidate_id=row.id,
job_post_id=job.id,
overall_score=float(row.match_score),
band=band,
model_name=row.model,
is_current=True,
)
self.session.add(entry)
# Flush the INSERT before touching prev/link: with no relationship()
# edge the unit of work emits the UPDATEs first, and both FKs
# (superseded_by_id, inbox.ats_id) reject a pointer to a row that is
# not inserted yet.
await self.session.flush()
if prev is not None:
prev.is_current=False
prev.superseded_by_id=entry.id
self.session.add(prev)
# inbox.ats_id always points at the CURRENT score row.
link.ats_id=entry.id
link.updated_at=datetime.now(timezone.utc)
self.session.add(link)
await self.session.commit()
if assigned and str(assigned)!=str(job.id) and link is not None:
existing=await AtsResults.get_for_inbox_job(self.session,link.id,assigned)
denorm=existing is None
if denorm:
await Inbox_Messages.set_ats_score(self.session,message_id,row.match_score,band)
if link is None:
return
identity=await AtsResults.resolve_identity(self.session,row.candidate_email,row.id)
await AtsResults.insert_result(self.session,{
"inbox_id":link.id,
**identity,
"job_post_id":job.id,
"overall_score":float(row.match_score),
"band":band,
"model_name":row.model,
"is_current":True,
})
async def _sync_upload_ats(self,job,row):
"""History row for an upload-sourced score — no inbox application exists,
so inbox_id stays NULL and the supersede chain runs per candidates row
(stable across re-scores: upsert keeps the id for the same job+bytes)."""
"""History row for an upload-sourced score — inbox_id stays NULL."""
band=CandidateView._recommendation(row.match_score) or ""
prev=await AtsResults.get_current_for_candidate(self.session,row.id)
entry=AtsResults(
inbox_id=None,
candidate_id=row.id,
job_post_id=job.id,
overall_score=float(row.match_score),
band=band,
model_name=row.model,
is_current=True,
)
self.session.add(entry)
# Same flush-before-pointing rule as the inbox path: the FK on
# superseded_by_id must see the new row inserted first.
await self.session.flush()
if prev is not None:
prev.is_current=False
prev.superseded_by_id=entry.id
self.session.add(prev)
await self.session.commit()
identity=await AtsResults.resolve_identity(self.session,row.candidate_email,row.id)
await AtsResults.insert_result(self.session,{
"inbox_id":None,
**identity,
"job_post_id":job.id,
"overall_score":float(row.match_score),
"band":band,
"model_name":row.model,
"is_current":True,
})
class CandidateView:
@ -564,23 +481,14 @@ class CandidateView:
# Same bands the frontend uses (Candidates.jsx / seed.js).
return "Strong Match" if score>=82 else "Potential Match" if score>=65 else "Weak Match"
async def _scores_by_message(self,message_ids):
"""Completed ATS scores (candidates table) per inbox message id, newest first.
One batched query — the profile list would otherwise pay a query per row.
"""
mids=[m for m in message_ids if m]
if not mids:
return {}
result=await self.session.execute(
select(Candidates)
.where(Candidates.inbox_message_id.in_(mids),Candidates.status=="completed")
.order_by(Candidates.updated_at.desc())
)
scores={}
for row in result.scalars().all():
scores.setdefault(row.inbox_message_id,[]).append(row)
return scores
@staticmethod
def _score_from_message(record):
"""Inbox denorm on the already-loaded messages row — not a Candidates join."""
msg=getattr(record,"messages",None)
if msg is None or msg.ats_score is None:
return None,None
band=(msg.ats_band or "").strip() or None
return msg.ats_score,band or CandidateView._recommendation(msg.ats_score)
async def create_candidate(self,candidate_email=None,candidate_name=None,candidate_phone=None,job_post_id=None,current_company=None,current_position=None,platform=None,experience=None,status=None,referral_by=None,file_name=None,file_path=None,full_text=None,current_user=None):
try:
@ -632,7 +540,19 @@ class CandidateView:
job_post=None
if manual.job_post_id:
job_post=await JobPosts.get_job_post_by_id(self.session,str(manual.job_post_id))
return serialize_manual_candidate_profile(manual,user,job_post)
payload=serialize_manual_candidate_profile(manual,user,job_post)
from inbox.plugins import get_ats_score_for_manual_user
score=await get_ats_score_for_manual_user(self.session,user_id,manual.job_post_id)
if score:
payload["ai_score"]=score["overall_score"]
payload["recommendation"]=self._recommendation(score["overall_score"])
payload["scored_at"]=score["computed_at"]
payload["candidate_id"]=score.get("candidate_id")
if score.get("user_id") and not payload.get("user_id"):
payload["user_id"]=score["user_id"]
if score.get("job_post_id"):
payload["scored_job_post_id"]=score["job_post_id"]
return payload
return await self.attach_job_posts(rows)
except HTTPException:
raise
@ -729,7 +649,6 @@ class CandidateView:
"""
single=not isinstance(data,list)
records=[data] if single else list(data or [])
scores=await self._scores_by_message([getattr(r,"message_id",None) for r in records])
payloads=[]
wanted=[]
@ -737,10 +656,10 @@ class CandidateView:
payload=serialize_candidate_profile(record)
payload["job_posts"]=[]
payload["assigned_job_post"]=None
scored=scores.get(getattr(record,"message_id",None)) or []
if scored:
payload["ai_score"]=scored[0].match_score
payload["recommendation"]=self._recommendation(scored[0].match_score)
score,band=self._score_from_message(record)
if score is not None:
payload["ai_score"]=score
payload["recommendation"]=band
payloads.append(payload)
if payload.get("assigned_job_post_id"):
wanted.append(payload["assigned_job_post_id"])
@ -821,26 +740,46 @@ class CandidateView:
)
notes=[serialize_note(r) for r in result.scalars().all()]
# ATS score: join the scoring engine's `candidates` rows onto the profile
# by inbox message. The serializer stubs ai_score/recommendation to None;
# this is where they get real values. Prefer the score against the
# assigned job post, else the most recent completed score.
scores=await self._scores_by_message([getattr(r,"message_id",None) for r in records])
scored_rows=[row for rows in scores.values() for row in rows]
if scored_rows:
assigned_uid=Candidates._as_uuid(base.get("assigned_job_post_id")) if base.get("assigned_job_post_id") else None
chosen=None
if assigned_uid is not None:
chosen=next((r for r in scored_rows if r.job_id==assigned_uid),None)
if chosen is None:
chosen=max(scored_rows,key=lambda r:r.updated_at)
base["ai_score"]=chosen.match_score
base["recommendation"]=self._recommendation(chosen.match_score)
base["matched_keywords"]=list(chosen.matched_keywords or [])
base["missing_keywords"]=list(chosen.missing_keywords or [])
base["summary_critique"]=chosen.summary_critique
base["scored_job_post_id"]=str(chosen.job_id)
base["scored_at"]=chosen.updated_at.isoformat() if chosen.updated_at else None
# ATS score from inbox denorm / ats_results via Inbox.ats_id — never from
# a Candidates join on message id. Keywords live on the scored Candidates
# row: candidate_id when set, else email+job for the matched-user path.
ats_ids=[r.ats_id for r in records if getattr(r,"ats_id",None)]
ats_rows=[]
if ats_ids:
result=await self.session.execute(select(AtsResults).where(AtsResults.id.in_(ats_ids)))
ats_rows=list(result.scalars().all())
assigned_uid=AtsResults._as_uuid(base.get("assigned_job_post_id")) if base.get("assigned_job_post_id") else None
chosen=None
if assigned_uid is not None:
chosen=next((a for a in ats_rows if a.job_post_id==assigned_uid),None)
if chosen is None and ats_rows:
chosen=max(ats_rows,key=lambda a:a.computed_at)
if chosen is not None:
base["ai_score"]=chosen.overall_score
base["recommendation"]=chosen.band or self._recommendation(chosen.overall_score)
base["scored_job_post_id"]=str(chosen.job_post_id) if chosen.job_post_id else None
base["scored_at"]=chosen.computed_at.isoformat() if chosen.computed_at else None
base["candidate_id"]=str(chosen.candidate_id) if chosen.candidate_id else None
if chosen.user_id and not base.get("user_id"):
base["user_id"]=str(chosen.user_id)
scored=None
if chosen.candidate_id:
scored=await Candidates.get_candidate_by_id(self.session,str(chosen.candidate_id))
elif chosen.user_id:
owner=await Users.get_user_by_id(self.session,str(chosen.user_id))
if owner is not None:
scored=await Candidates.get_completed_by_email_job(self.session,owner.email,chosen.job_post_id)
if scored is not None:
base["matched_keywords"]=list(scored.matched_keywords or [])
base["missing_keywords"]=list(scored.missing_keywords or [])
base["summary_critique"]=scored.summary_critique
else:
for record in records:
score,band=self._score_from_message(record)
if score is not None:
base["ai_score"]=score
base["recommendation"]=band
break
activity.sort(key=lambda r:(r.get("activity_date") or ""),reverse=True)
base["favorite"]=favorite

24
backend/job/pipeline/views.py
View File

@ -3,19 +3,39 @@ from sqlalchemy.ext.asyncio import AsyncSession
from inbox.enums import Candidate_application_Status
from inbox.models import Inbox
from job.candidate.models import ApplicationStageTransitions
from job.candidate.models import ApplicationStageTransitions, Manual_UPLOAD_CANDIDATE
from job.pipeline.serializers import serialize_stage_transition
from inbox.plugins import get_ats_score_for_manual_user, get_ats_score_for_user
class Pipeline:
def __init__(self,session:AsyncSession):
self.session=session
async def get_all(self):
try:
inbox_data=await Inbox.get_all(self.session)
manual_upload_data=await Manual_UPLOAD_CANDIDATE.get_all(self.session)
data={"inbox":inbox_data,"manual_upload":manual_upload_data}
return data
except Exception as e:
raise HTTPException(status_code=500,detail=str(e))
async def get_pipeline_candidates(self,user_id=None,job_post_id=None):
try:
manual_data=await get_ats_score_for_manual_user(self.session,user_id,job_post_id)
inbox_data=await get_ats_score_for_user(self.session,user_id,job_post_id)
data={"manual":manual_data,"inbox":inbox_data}
return data
except Exception as e:
raise HTTPException(status_code=500,detail=str(e))
async def get_transitions(self,inbox_id=None,transition_id=None):
if transition_id:
row=await ApplicationStageTransitions.get_by_id(self.session,transition_id)
if not row:
raise HTTPException(status_code=404,detail="Transition not found")
return serialize_stage_transition(row)
if inbox_id is None:
raise HTTPException(status_code=400,detail="transition_id or inbox_id is required")

200
backend/migrations/manual/002_backfill_inbox_ats.sql
View File

@ -8,87 +8,119 @@
-- INSERT skips any application that already has an ats_results row.
-- Applied automatically at startup by alembic_setup.run_manual_sql() once the
-- schema is at head; recorded in manual_migrations. Safe to re-run by hand.
--
-- The Candidates.inbox_message_id join is historical: that column is dropped
-- after revision b7d4e8f1a203. Those three statements run only when the column
-- is still present (a DB that has not yet reached that revision). Fresh installs
-- skip them — there is no pre-denorm history to copy.
-- Winning score per message, mirroring CandidateView._recommendation precedence:
-- the completed score against the ASSIGNED job wins, else the newest completed.
WITH ranked AS (
SELECT
c.inbox_message_id,
c.job_id,
c.match_score,
c.model,
c.updated_at,
ROW_NUMBER() OVER (
PARTITION BY c.inbox_message_id
ORDER BY (c.job_id = m.assigned_job_post_id) DESC NULLS LAST,
c.updated_at DESC
) AS rn
FROM app.candidates c
JOIN app.inbox_messages m ON m.id = c.inbox_message_id
WHERE c.status = 'completed'
AND c.inbox_message_id IS NOT NULL
)
UPDATE app.inbox_messages m
SET ats_score = r.match_score,
ats_band = CASE
WHEN r.match_score >= 82 THEN 'Strong Match'
WHEN r.match_score >= 65 THEN 'Potential Match'
ELSE 'Weak Match'
END
FROM ranked r
WHERE r.rn = 1
AND m.id = r.inbox_message_id
AND m.ats_score IS NULL;
DO $backfill$
BEGIN
IF EXISTS (
SELECT 1 FROM information_schema.columns
WHERE table_schema = 'app'
AND table_name = 'candidates'
AND column_name = 'inbox_message_id'
) THEN
-- Winning score per message, mirroring CandidateView._recommendation precedence:
-- the completed score against the ASSIGNED job wins, else the newest completed.
EXECUTE $sql$
WITH ranked AS (
SELECT
c.inbox_message_id,
c.job_id,
c.match_score,
c.model,
c.updated_at,
ROW_NUMBER() OVER (
PARTITION BY c.inbox_message_id
ORDER BY (c.job_id = m.assigned_job_post_id) DESC NULLS LAST,
c.updated_at DESC
) AS rn
FROM app.candidates c
JOIN app.inbox_messages m ON m.id = c.inbox_message_id
WHERE c.status = 'completed'
AND c.inbox_message_id IS NOT NULL
)
UPDATE app.inbox_messages m
SET ats_score = r.match_score,
ats_band = CASE
WHEN r.match_score >= 82 THEN 'Strong Match'
WHEN r.match_score >= 65 THEN 'Potential Match'
ELSE 'Weak Match'
END
FROM ranked r
WHERE r.rn = 1
AND m.id = r.inbox_message_id
AND m.ats_score IS NULL
$sql$;
-- One current ats_results row per already-scored application. computed_at takes
-- the candidates row's timestamp so the history reflects when the score happened.
-- inbox can hold several join rows per message; DISTINCT ON keeps the newest.
WITH ranked AS (
SELECT
c.inbox_message_id,
c.job_id,
c.match_score,
c.model,
c.updated_at,
ROW_NUMBER() OVER (
PARTITION BY c.inbox_message_id
ORDER BY (c.job_id = m.assigned_job_post_id) DESC NULLS LAST,
c.updated_at DESC
) AS rn
FROM app.candidates c
JOIN app.inbox_messages m ON m.id = c.inbox_message_id
WHERE c.status = 'completed'
AND c.inbox_message_id IS NOT NULL
),
links AS (
SELECT DISTINCT ON (message_id) message_id, id AS inbox_id
FROM app.inbox
ORDER BY message_id, created_at DESC
)
INSERT INTO app.ats_results
(id, inbox_id, job_post_id, overall_score, band, is_current,
superseded_by_id, model_name, computed_at, created_at)
SELECT
gen_random_uuid(),
l.inbox_id,
r.job_id,
r.match_score,
CASE
WHEN r.match_score >= 82 THEN 'Strong Match'
WHEN r.match_score >= 65 THEN 'Potential Match'
ELSE 'Weak Match'
END,
true,
NULL,
r.model,
r.updated_at,
NOW()
FROM ranked r
JOIN links l ON l.message_id = r.inbox_message_id
WHERE r.rn = 1
AND NOT EXISTS (
SELECT 1 FROM app.ats_results a WHERE a.inbox_id = l.inbox_id
);
-- One current ats_results row per already-scored application. computed_at takes
-- the candidates row's timestamp so the history reflects when the score happened.
-- inbox can hold several join rows per message; DISTINCT ON keeps the newest.
EXECUTE $sql$
WITH ranked AS (
SELECT
c.inbox_message_id,
c.job_id,
c.match_score,
c.model,
c.updated_at,
ROW_NUMBER() OVER (
PARTITION BY c.inbox_message_id
ORDER BY (c.job_id = m.assigned_job_post_id) DESC NULLS LAST,
c.updated_at DESC
) AS rn
FROM app.candidates c
JOIN app.inbox_messages m ON m.id = c.inbox_message_id
WHERE c.status = 'completed'
AND c.inbox_message_id IS NOT NULL
),
links AS (
SELECT DISTINCT ON (message_id) message_id, id AS inbox_id
FROM app.inbox
ORDER BY message_id, created_at DESC
)
INSERT INTO app.ats_results
(id, inbox_id, job_post_id, overall_score, band, is_current,
superseded_by_id, model_name, computed_at, created_at)
SELECT
gen_random_uuid(),
l.inbox_id,
r.job_id,
r.match_score,
CASE
WHEN r.match_score >= 82 THEN 'Strong Match'
WHEN r.match_score >= 65 THEN 'Potential Match'
ELSE 'Weak Match'
END,
true,
NULL,
r.model,
r.updated_at,
NOW()
FROM ranked r
JOIN links l ON l.message_id = r.inbox_message_id
WHERE r.rn = 1
AND NOT EXISTS (
SELECT 1 FROM app.ats_results a WHERE a.inbox_id = l.inbox_id
)
$sql$;
-- Older inbox history rows predate candidate_id; link the current ones back to
-- the candidates row that produced them.
EXECUTE $sql$
UPDATE app.ats_results a
SET candidate_id = c.id
FROM app.inbox i
JOIN app.candidates c ON c.inbox_message_id = i.message_id AND c.status = 'completed'
WHERE a.inbox_id = i.id
AND a.candidate_id IS NULL
AND c.job_id = a.job_post_id
$sql$;
END IF;
END
$backfill$;
-- inbox.ats_id -> the CURRENT ats_results row for that application, so the
-- score is one direct id join away. Newest current row wins if several exist.
@ -131,13 +163,3 @@ WHERE c.status = 'completed'
AND NOT EXISTS (
SELECT 1 FROM app.ats_results a WHERE a.candidate_id = c.id
);
-- Older inbox history rows predate candidate_id; link the current ones back to
-- the candidates row that produced them.
UPDATE app.ats_results a
SET candidate_id = c.id
FROM app.inbox i
JOIN app.candidates c ON c.inbox_message_id = i.message_id AND c.status = 'completed'
WHERE a.inbox_id = i.id
AND a.candidate_id IS NULL
AND c.job_id = a.job_post_id;

1
backend/users/models.py
View File

@ -57,7 +57,6 @@ class Users(SQLModel, table=True):
)
password: str
created_at: datetime = Field(default_factory=_now, sa_type=DateTime(timezone=True))
updated_at: datetime = Field(default_factory=_now, sa_type=DateTime(timezone=True))
is_active: bool = Field(default=False)

2
frontend/dist/index.html vendored
View File

@ -23,7 +23,7 @@
<link rel="preconnect" href="https://fonts.gstatic.com" crossorigin />
<link rel="stylesheet" href="https://fonts.googleapis.com/css2?family=Belleza&family=Inter:wght@400;500;600;700;800&display=swap" />
<link rel="icon" href="data:image/svg+xml,<svg xmlns='http://www.w3.org/2000/svg' viewBox='0 0 100 100'><rect width='100' height='100' rx='22' fill='%23004d43'/><g transform='translate(14 32) scale(0.72)'><path d='M100 3.65C97.86 20.99 91.89 43.03 79.48 55.33 76 58.77 71.84 61.46 66.96 62.14 50.4 64.46 41.84 47.5 29.07 42.7 21.85 39.98 14.5 42.02 9.66 47.95 6.54 51.78 4.49 56.35 2.97 61.13 2.41 61.64 0.97 61.66 0 61.31L0 0.13C1.05 0 2.27 0.02 3.09 0.28 14.9 15.86 26.77 30.82 40.15 45.28L60.79 24.7C67.38 18.22 74.41 12.74 82.59 8.51 88.11 5.83 93.64 3.93 100 3.65Z' fill='%23ceff71'/></g></svg>" />
<script type="module" crossorigin src="/assets/index-D1YVNnju.js"></script>
<script type="module" crossorigin src="/assets/index-CnIht1Bn.js"></script>
<link rel="stylesheet" crossorigin href="/assets/index-CSn67wit.css">
</head>
<body>

4
frontend/src/api/candidates.js
View File

@ -83,7 +83,6 @@ export function toCandidateView(row) {
errorCode: row.error_code ?? null,
errorMessage: row.error_message ?? null,
applied: row.created_at ? new Date(row.created_at) : null,
inboxMessageId: row.inbox_message_id ?? null,
}
}
@ -115,7 +114,7 @@ export function listCandidateUsers({ roleId = 8, top = 500, skip = 0 } = {}) {
*
* A user account carries identity only. Everything the ATS produces
* (score, matched skills, critique, the job it was scored against) lives in the
* `candidates` table keyed by inbox_message_id, with no user_id to join on, so
* `candidates` table keyed by job_id + content hash, with no user_id to join on, so
* those fields are null here by construction rather than by omission.
*/
export function toCandidateUserView(row) {
@ -141,7 +140,6 @@ export function toCandidateUserView(row) {
scoringStatus: null,
errorCode: null,
errorMessage: null,
inboxMessageId: null,
}
}

2
frontend/src/screens/ScoredCandidateProfile.jsx
View File

@ -64,7 +64,7 @@ export default function ScoredCandidateProfile({ candidate: c, jobTitle, onClose
const filename =
live?.documents?.[0]?.name || c.filename || null
const matchSummary = live?.match_summary ?? null
const messageId = live?.message_id ?? c.inboxMessageId ?? null
const messageId = live?.message_id ?? null
const aiScore = live?.ai_score ?? c.aiScore ?? null
const matchedSkills = live?.matched_keywords ?? c.matchedSkills ?? []
const missingSkills = live?.missing_keywords ?? c.missingSkills ?? []