126 lines
5.0 KiB
Python
126 lines
5.0 KiB
Python
import uuid
|
|
from datetime import datetime, timezone
|
|
|
|
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 | None = None):
|
|
"""Leaderboard order: completed by score desc, failures last, ties stable.
|
|
|
|
job_id=None returns the whole pool across jobs (same ordering) for the
|
|
frontend's unscoped Candidates/Talent Pool views.
|
|
"""
|
|
statement = select(cls)
|
|
if job_id is not None:
|
|
uid = cls._as_uuid(job_id)
|
|
if uid is None:
|
|
return []
|
|
statement = statement.where(cls.job_id == uid)
|
|
statement = statement.order_by(
|
|
cls.status.asc(), # "completed" < "failed"
|
|
cls.match_score.desc().nulls_last(),
|
|
cls.created_at.asc(),
|
|
)
|
|
result = await session.execute(statement)
|
|
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
|