1370 lines
58 KiB
Python
1370 lines
58 KiB
Python
import logging
|
|
import os
|
|
from shlex import join
|
|
import uuid
|
|
from datetime import datetime, timezone
|
|
from typing import Any, List, Optional
|
|
|
|
from dotenv import load_dotenv
|
|
from fastapi import HTTPException
|
|
from inbox.enums import Candidate_application_Status
|
|
from role.models import EnumRoles, Roles
|
|
from sqlalchemy import Column, DateTime, case, func, or_, update
|
|
from sqlalchemy.dialects.postgresql import JSONB
|
|
from sqlalchemy.exc import IntegrityError
|
|
from sqlalchemy.ext.asyncio import AsyncSession
|
|
from sqlalchemy.orm import selectinload
|
|
from sqlmodel import Field, Relationship, SQLModel, select, true
|
|
|
|
from job.candidate.models import Activity, Feedback, Interviews
|
|
from linkedin_utils import slug_from_url, NO_SLUG
|
|
from users.models import Users
|
|
from users.plugins import hash_password
|
|
|
|
load_dotenv()
|
|
logger = logging.getLogger("inbox.models")
|
|
|
|
# Placeholder only. The account lands inactive and the candidate is mailed a
|
|
# confirmation link; the real password comes from the reset flow afterwards.
|
|
DEFAULT_CANDIDATE_PASSWORD = os.getenv("DEFAULT_CANDIDATE_PASSWORD", "Utopia!@#")
|
|
CANDIDATE_ROLE_ID = 8 # seeded candidate role (id 4 is hiring_manager, the signup default)
|
|
SKIP_SENDER_PREFIXES = ("noreply", "no-reply", "donotreply", "do-not-reply",
|
|
"mailer-daemon", "postmaster", "bounce")
|
|
|
|
|
|
def _now() -> datetime:
|
|
return datetime.now(timezone.utc)
|
|
|
|
|
|
class Inbox(SQLModel, table=True):
|
|
__tablename__ = "inbox"
|
|
|
|
id: int | None = Field(default=None, primary_key=True)
|
|
user_id: uuid.UUID | None = Field(default=None, foreign_key="users.id")
|
|
|
|
alert_id: uuid.UUID | None = Field(default=None, foreign_key="inbox_alerts.id")
|
|
alerts: Optional["Inbox_Alerts"] = Relationship(back_populates="inbox")
|
|
|
|
message_id: uuid.UUID | None = Field(default=None, foreign_key="inbox_messages.id")
|
|
messages: Optional["Inbox_Messages"] = Relationship(back_populates="inbox")
|
|
|
|
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)
|
|
|
|
ats_id: uuid.UUID | None = Field(default=None, foreign_key="ats_results.id")
|
|
|
|
interviews: List["Interviews"] = Relationship(
|
|
back_populates="inbox",
|
|
sa_relationship_kwargs={"lazy": "selectin"},
|
|
)
|
|
activity: List["Activity"] = Relationship(
|
|
back_populates="inbox",
|
|
sa_relationship_kwargs={"lazy": "selectin"},
|
|
)
|
|
feedback: List["Feedback"] = Relationship(
|
|
back_populates="inbox",
|
|
sa_relationship_kwargs={"lazy": "selectin"},
|
|
)
|
|
user: Optional["Users"] = Relationship(
|
|
back_populates="inbox",
|
|
sa_relationship_kwargs={"lazy": "joined"},
|
|
)
|
|
|
|
@classmethod
|
|
async def get_all(cls,session:AsyncSession,job_post_id=None,limit=None,offset=0):
|
|
try:
|
|
from job.job_post.models import JobPosts
|
|
qry=(
|
|
select(
|
|
cls.id.label("inbox_id"),
|
|
cls.user_id,
|
|
Users.name,
|
|
Users.email,
|
|
Users.linkedin_url,
|
|
Inbox_Messages.candidate_phone_number.label("phone"),
|
|
Inbox_Messages.assigned_job_post_id,
|
|
Inbox_Messages.application_status,
|
|
Inbox_Messages.current_employment,
|
|
Inbox_Messages.current_title,
|
|
Inbox_Messages.experience,
|
|
cls.created_at,
|
|
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)
|
|
# Newest-first is the list contract; score is only a tiebreak
|
|
# within the same instant. id keeps paging stable.
|
|
.order_by(
|
|
cls.created_at.desc(),
|
|
AtsResults.overall_score.desc().nulls_last(),
|
|
cls.id.desc(),
|
|
)
|
|
)
|
|
if job_post_id:
|
|
qry=qry.where(Inbox_Messages.assigned_job_post_id==job_post_id)
|
|
if limit is not None:
|
|
qry=qry.limit(limit).offset(offset)
|
|
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,
|
|
}
|
|
status=row["application_status"]
|
|
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"],
|
|
"linkedin_url":row["linkedin_url"] or None,
|
|
"application_status":status.value if status else None,
|
|
"phone":row["phone"],
|
|
"assigned_job_post_id":str(row["assigned_job_post_id"]) if row["assigned_job_post_id"] else None,
|
|
"title":row["title"] or None,
|
|
"current_employment":row["current_employment"] or None,
|
|
"current_title":row["current_title"] or None,
|
|
"experience":row["experience"] or None,
|
|
"created_at":row["created_at"].isoformat() if row["created_at"] else None,
|
|
"ats_result":ats,
|
|
})
|
|
return rows
|
|
except Exception as e:
|
|
raise HTTPException(status_code=500,detail=str(e))
|
|
|
|
@classmethod
|
|
async def linkedin_urls_by_message_ids(cls, session: AsyncSession, message_ids) -> dict:
|
|
"""users.linkedin_url keyed by inbox_messages.id for one list page."""
|
|
ids = [mid for mid in (message_ids or []) if mid is not None]
|
|
if not ids:
|
|
return {}
|
|
result = await session.execute(
|
|
select(cls.message_id, Users.linkedin_url)
|
|
.join(Users, Users.id == cls.user_id)
|
|
.where(cls.message_id.in_(ids))
|
|
.where(Users.linkedin_url.is_not(None))
|
|
.where(Users.linkedin_url != "")
|
|
)
|
|
out = {}
|
|
for mid, url in result.all():
|
|
if mid not in out and url:
|
|
out[mid] = url
|
|
return out
|
|
|
|
@classmethod
|
|
async def count_by_status(cls,session:AsyncSession,job_post_id=None):
|
|
try:
|
|
from job.job_post.models import JobPosts
|
|
qry=(
|
|
select(Inbox_Messages.application_status,func.count())
|
|
.select_from(cls)
|
|
.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)
|
|
.where(Inbox_Messages.assigned_job_post_id.is_not(None))
|
|
.where(Roles.role_name==EnumRoles.CANDIDATE.value)
|
|
.group_by(Inbox_Messages.application_status)
|
|
)
|
|
if job_post_id:
|
|
qry=qry.where(Inbox_Messages.assigned_job_post_id==job_post_id)
|
|
result=await session.execute(qry)
|
|
counts={}
|
|
for status,n in result.all():
|
|
key=status.value if hasattr(status,"value") else (str(status) if status else None)
|
|
if not key:
|
|
continue
|
|
counts[key]=int(n or 0)
|
|
return counts
|
|
except Exception as e:
|
|
raise HTTPException(status_code=500,detail=str(e))
|
|
|
|
@classmethod
|
|
def _candidate_search_filter(cls, search: str):
|
|
pattern = f"%{search}%"
|
|
return or_(Users.name.ilike(pattern), Users.email.ilike(pattern))
|
|
|
|
@classmethod
|
|
async def get_candidate_profile(cls,session:AsyncSession,user_id:uuid.UUID|None=None,limit:int=10,offset:int=0,search:str|None=None):
|
|
try:
|
|
options=[selectinload(cls.messages)]
|
|
if user_id:
|
|
options.extend([
|
|
selectinload(cls.interviews),
|
|
selectinload(cls.activity),
|
|
selectinload(cls.feedback),
|
|
])
|
|
qry = (
|
|
select(cls)
|
|
.options(*options)
|
|
.join(Users, cls.user_id == Users.id)
|
|
.join(Roles, Users.role_id == Roles.id)
|
|
.where(Roles.role_name == EnumRoles.CANDIDATE.value)
|
|
)
|
|
if user_id:
|
|
qry = qry.where(cls.user_id == user_id)
|
|
if search:
|
|
qry = qry.where(cls._candidate_search_filter(search))
|
|
# Most-recent-first is the list contract; id breaks ties so a page
|
|
# boundary can't drop or repeat a row when created_at collides.
|
|
qry = qry.order_by(cls.created_at.desc(), cls.id.desc())
|
|
qry = qry.limit(limit).offset(offset)
|
|
result = await session.execute(qry)
|
|
rows = result.scalars().all()
|
|
if user_id and len(rows) == 1:
|
|
return rows[0]
|
|
return rows
|
|
except Exception as e:
|
|
raise HTTPException(status_code=500,detail=str(e))
|
|
|
|
@classmethod
|
|
async def count_candidate_profiles(cls,session:AsyncSession,user_id:uuid.UUID|None=None,search:str|None=None):
|
|
"""Result-set size for the same predicate get_candidate_profile pages over."""
|
|
try:
|
|
qry = (
|
|
select(func.count())
|
|
.select_from(cls)
|
|
.join(Users, cls.user_id == Users.id)
|
|
.join(Roles, Users.role_id == Roles.id)
|
|
.where(Roles.role_name == EnumRoles.CANDIDATE.value)
|
|
)
|
|
if user_id:
|
|
qry = qry.where(cls.user_id == user_id)
|
|
if search:
|
|
qry = qry.where(cls._candidate_search_filter(search))
|
|
result = await session.execute(qry)
|
|
return result.scalar_one()
|
|
except Exception as e:
|
|
raise HTTPException(status_code=500,detail=str(e))
|
|
|
|
@classmethod
|
|
async def get_inbox_by_id(cls,session:AsyncSession,record_id:int|str|None):
|
|
if record_id is None:
|
|
return None
|
|
try:
|
|
iid=int(record_id)
|
|
except (TypeError,ValueError):
|
|
return None
|
|
result=await session.execute(select(cls).where(cls.id==iid))
|
|
return result.scalars().first()
|
|
|
|
@classmethod
|
|
async def get_inbox_with_message(cls,session:AsyncSession,record_id:int|str|None):
|
|
"""Inbox row with `messages` selectin-loaded for stage / application writers."""
|
|
if record_id is None:
|
|
return None
|
|
try:
|
|
iid=int(record_id)
|
|
except (TypeError,ValueError):
|
|
return None
|
|
result=await session.execute(
|
|
select(cls).options(selectinload(cls.messages)).where(cls.id==iid)
|
|
)
|
|
return result.scalars().first()
|
|
|
|
@classmethod
|
|
async def get_inbox_by_message_id(cls,session:AsyncSession,message_id):
|
|
try:
|
|
mid=uuid.UUID(str(message_id))
|
|
except ValueError:
|
|
return None
|
|
result=await session.execute(
|
|
select(cls).where(cls.message_id==mid).order_by(cls.created_at.desc())
|
|
)
|
|
return result.scalars().first()
|
|
|
|
@classmethod
|
|
async def get_inbox_by_user_id(cls,session:AsyncSession,user_id):
|
|
try:
|
|
uid=uuid.UUID(str(user_id))
|
|
except ValueError:
|
|
return None
|
|
result=await session.execute(
|
|
select(cls).where(cls.user_id==uid).order_by(cls.created_at.desc())
|
|
)
|
|
return result.scalars().first()
|
|
|
|
@classmethod
|
|
async def newest_cv_by_user_ids(cls,session:AsyncSession,user_ids):
|
|
"""Newest inbox.id + first file_path per user — search Open resume."""
|
|
ids=[]
|
|
for raw in (user_ids or []):
|
|
try:
|
|
ids.append(uuid.UUID(str(raw)))
|
|
except (TypeError,ValueError):
|
|
continue
|
|
if not ids:
|
|
return {}
|
|
result=await session.execute(
|
|
select(cls.user_id,cls.id,Inbox_Messages.file_path)
|
|
.join(Inbox_Messages,cls.message_id==Inbox_Messages.id)
|
|
.where(cls.user_id.in_(ids))
|
|
.order_by(cls.created_at.desc())
|
|
)
|
|
out={}
|
|
for user_id,inbox_id,file_path in result.all():
|
|
key=str(user_id)
|
|
if key in out:
|
|
continue
|
|
first=(file_path or "").split(",")[0].strip() or None
|
|
out[key]={"inbox_id":inbox_id,"file_path":first}
|
|
return out
|
|
|
|
@classmethod
|
|
async def update_inbox(cls,session:AsyncSession,record_id,fields:dict):
|
|
row=await cls.get_inbox_by_id(session,record_id)
|
|
if not row:
|
|
return None
|
|
for key,value in fields.items():
|
|
setattr(row,key,value)
|
|
row.updated_at=_now()
|
|
session.add(row)
|
|
await session.commit()
|
|
await session.refresh(row)
|
|
return row
|
|
|
|
|
|
class Inbox_Alerts(SQLModel, table=True):
|
|
__tablename__ = "inbox_alerts"
|
|
|
|
id: uuid.UUID = Field(default_factory=uuid.uuid4, primary_key=True)
|
|
alert_sender_name: str
|
|
alert_sender_email: str
|
|
is_read: bool = Field(default=False)
|
|
recieve_time: datetime = Field(default_factory=_now, sa_type=DateTime(timezone=True))
|
|
|
|
inbox: list[Inbox] = Relationship(back_populates="alerts")
|
|
|
|
|
|
class Inbox_Messages(SQLModel, table=True):
|
|
__tablename__ = "inbox_messages"
|
|
|
|
id: uuid.UUID = Field(default_factory=uuid.uuid4, primary_key=True)
|
|
message_id: str | None = Field(default=None, index=True, unique=True)
|
|
full_email_response: dict[str, Any] | None = Field(
|
|
default=None, sa_column=Column(JSONB)
|
|
)
|
|
application_status: Candidate_application_Status = Field(default=Candidate_application_Status.CLOSED)
|
|
message_subject: str
|
|
message_body: str
|
|
message_sent_time: str
|
|
message_received_time: str
|
|
message_from: str
|
|
message_to: str
|
|
message_cc: str | None = Field(default=None)
|
|
message_bcc: str | None = Field(default=None)
|
|
message_read: bool = Field(default=False)
|
|
# Stamped whenever a human flips read state from the app (single row, bulk, or
|
|
# whole view). apply_read_status skips these rows: nothing pushes local state
|
|
# back to Outlook, so without the stamp the every-minute sync_read_status sweep
|
|
# would silently re-read a mail the recruiter deliberately marked unread.
|
|
read_overridden_at: datetime | None = Field(default=None, sa_type=DateTime(timezone=True))
|
|
attachment: bool = Field(default=False)
|
|
message_reply: str | None = Field(default=None)
|
|
file_name: str | None = Field(default=None)
|
|
file_path: str | None = Field(default=None)
|
|
resume_text: str | None = Field(default=None)
|
|
# Lowercase /in/<slug> extracted from resume_text ("" = scanned, none
|
|
# found; NULL = not yet scanned — see linkedin_utils). Lets Find Talent
|
|
# flag sourced profiles that already applied.
|
|
linkedin_slug: str | None = Field(default=None, index=True)
|
|
experience: str | None = Field(default=None)
|
|
suggested_job_post_ids: list[str] | None = Field(default=None, sa_column=Column(JSONB))
|
|
assigned_job_post_id: uuid.UUID | None = Field(default=None, foreign_key="job_posts.id", index=True)
|
|
|
|
match_summary: str | None = Field(default=None)
|
|
match_reasoning: str | None = Field(default=None)
|
|
match_status: str | None = Field(default=None)
|
|
match_error: str | None = Field(default=None)
|
|
matched_at: datetime | None = Field(default=None, sa_type=DateTime(timezone=True))
|
|
candidate_phone_number: str | None = Field(default="xxx-xxx-xxxx")
|
|
candidate_education: str | None = Field(default=None)
|
|
current_employment: str | None = Field(default=None)
|
|
current_title: str | None = Field(default=None)
|
|
# Denormalised dashboard / list-screen fields. server_default is load-bearing
|
|
# for every NOT NULL column — these arrive as ALTERs on a populated table.
|
|
ats_score: float | None = Field(default=None)
|
|
ats_band: str = Field(default="", sa_column_kwargs={"server_default": ""})
|
|
recruiter_id: uuid.UUID | None = Field(default=None, foreign_key="users.id")
|
|
is_duplicate: bool = Field(default=False, sa_column_kwargs={"server_default": "false"})
|
|
source_channel_id: int | None = Field(default=None, foreign_key="source_channels.id")
|
|
processing_state: str = Field(default="unread", sa_column_kwargs={"server_default": "unread"})
|
|
# Ingestion time — list screens order by this (newest first). server_default
|
|
# backfills existing rows on the ALTER so NOT NULL is safe on a populated table.
|
|
created_at: datetime = Field(
|
|
default_factory=_now,
|
|
sa_type=DateTime(timezone=True),
|
|
sa_column_kwargs={"server_default": "now()"},
|
|
)
|
|
inbox: list[Inbox] = Relationship(back_populates="messages")
|
|
|
|
@staticmethod
|
|
def _body_text(email_data: dict) -> str:
|
|
body = email_data.get("body")
|
|
if isinstance(body, dict):
|
|
return body.get("content") or ""
|
|
if isinstance(body, str):
|
|
return body
|
|
return email_data.get("bodyPreview") or ""
|
|
|
|
|
|
@classmethod
|
|
async def get_all_applications(cls,session:AsyncSession,message_id:uuid.UUID|int|None=None):
|
|
try:
|
|
qry=select(cls.message_id,cls.full_email_response,cls.message_subject,cls.message_from,cls.message_to,cls.message_sent_time,cls.message_read,cls.attachment)
|
|
if message_id:
|
|
qry=qry.where(cls.message_id==message_id)
|
|
result=await session.execute(qry)
|
|
return result.scalars().all()
|
|
|
|
except Exception as e:
|
|
raise HTTPException(status_code=500,detail=str(e))
|
|
@classmethod
|
|
async def set_match_result(
|
|
cls,
|
|
session: AsyncSession,
|
|
record_id,
|
|
*,
|
|
resume_text=None,
|
|
experience=None,
|
|
candidate_education=None,
|
|
candidate_phone_number=None,
|
|
current_employment=None,
|
|
current_title=None,
|
|
linkedin_url=None,
|
|
suggested_job_post_ids=None,
|
|
summary="",
|
|
reasoning="",
|
|
status="",
|
|
error="",
|
|
):
|
|
"""Persist agent output onto one inbox row; returns the row or None."""
|
|
row = await cls.get_inbox_message_by_id(session, record_id)
|
|
if not row:
|
|
return None
|
|
if resume_text is not None:
|
|
row.resume_text = resume_text
|
|
url = (linkedin_url or "").strip() or None
|
|
if url:
|
|
row.linkedin_slug = slug_from_url(url) or NO_SLUG
|
|
user_id = await cls.get_linked_user_id(session, row.id)
|
|
if user_id:
|
|
await Users.set_linkedin_url_if_empty(
|
|
session, user_id=user_id, url=url,
|
|
)
|
|
elif resume_text is not None:
|
|
# Agent ran and found no profile — mark scanned so talent backfill
|
|
# does not regex-scan this CV again.
|
|
row.linkedin_slug = NO_SLUG
|
|
if candidate_phone_number is not None:
|
|
row.candidate_phone_number = candidate_phone_number
|
|
if candidate_education is not None:
|
|
row.candidate_education = candidate_education
|
|
if current_employment is not None:
|
|
row.current_employment = current_employment
|
|
if current_title is not None:
|
|
row.current_title = current_title
|
|
row.suggested_job_post_ids = suggested_job_post_ids
|
|
row.match_summary = summary or None
|
|
row.match_reasoning = reasoning or None
|
|
row.match_status = status or None
|
|
row.match_error = error or None
|
|
row.experience = experience or None
|
|
row.matched_at = datetime.now(timezone.utc)
|
|
session.add(row)
|
|
await session.commit()
|
|
await session.refresh(row)
|
|
return row
|
|
|
|
@classmethod
|
|
def _fields_from_email(cls, email_data: dict, file_path: list[str] | None = None) -> dict:
|
|
return {
|
|
"message_subject": email_data.get("subject") or "",
|
|
"message_body": cls._body_text(email_data),
|
|
"message_sent_time": email_data.get("sentDateTime") or "",
|
|
"message_read": bool(email_data.get("isRead")),
|
|
"message_received_time": email_data.get("receivedDateTime") or "",
|
|
"message_from": email_data.get("from", {})
|
|
.get("emailAddress", {})
|
|
.get("address", ""),
|
|
"message_to": ",".join(
|
|
[r["emailAddress"]["address"] for r in email_data.get("toRecipients", [])]
|
|
),
|
|
"message_cc": ",".join(
|
|
[r["emailAddress"]["address"] for r in email_data.get("ccRecipients", [])]
|
|
),
|
|
"message_bcc": ",".join(
|
|
[r["emailAddress"]["address"] for r in email_data.get("bccRecipients", [])]
|
|
),
|
|
"attachment": bool(email_data.get("hasAttachments")),
|
|
"message_reply": ",".join(
|
|
[r["emailAddress"]["address"] for r in email_data.get("replyTo", [])]
|
|
),
|
|
"message_id": email_data.get("id"),
|
|
"file_name": ",".join(
|
|
[r.get("name") for r in email_data.get("attachments", [])]
|
|
),
|
|
"file_path": ",".join(file_path) if file_path else None,
|
|
"full_email_response": email_data,
|
|
}
|
|
|
|
@classmethod
|
|
def _sender_address(cls, email_data: dict) -> str:
|
|
return (
|
|
email_data.get("from", {})
|
|
.get("emailAddress", {})
|
|
.get("address", "")
|
|
or ""
|
|
).strip().lower()
|
|
|
|
@classmethod
|
|
def _sender_display_name(cls, email_data: dict, address: str) -> str:
|
|
name = (
|
|
email_data.get("from", {})
|
|
.get("emailAddress", {})
|
|
.get("name")
|
|
or ""
|
|
).strip()
|
|
if name:
|
|
return name
|
|
return address.split("@", 1)[0] if address else "candidate"
|
|
|
|
@classmethod
|
|
def _is_linkable_sender(cls, address: str) -> bool:
|
|
if not address or "@" not in address:
|
|
return False
|
|
local = address.split("@", 1)[0]
|
|
return not local.startswith(SKIP_SENDER_PREFIXES)
|
|
|
|
@classmethod
|
|
async def _link_sender(cls,session:AsyncSession,email_data:dict,email):
|
|
address=cls._sender_address(email_data)
|
|
if not cls._is_linkable_sender(address):
|
|
return None
|
|
try:
|
|
# id-only: avoid Users.job_posts selectin / role lazy loads under asyncio
|
|
user_id=(await session.execute(
|
|
select(Users.id).where(func.lower(Users.email)==address)
|
|
)).scalar_one_or_none()
|
|
|
|
if user_id is None:
|
|
user=Users(
|
|
name=cls._sender_display_name(email_data,address),
|
|
email=address,
|
|
role_id=CANDIDATE_ROLE_ID,
|
|
password=hash_password(DEFAULT_CANDIDATE_PASSWORD),
|
|
)
|
|
session.add(user)
|
|
# autoflush=False: flush so users.id exists before inbox FK insert
|
|
# (Relationship helps ordering, but flush keeps this path explicit).
|
|
await session.flush()
|
|
session.add(Inbox(user_id=user.id,message_id=email.id))
|
|
await session.commit()
|
|
return address
|
|
|
|
link=(await session.execute(
|
|
select(Inbox.id).where(Inbox.message_id==email.id,Inbox.user_id==user_id)
|
|
)).scalar_one_or_none()
|
|
if link is None:
|
|
session.add(Inbox(user_id=user_id,message_id=email.id))
|
|
await session.commit()
|
|
return None
|
|
except IntegrityError:
|
|
await session.rollback()
|
|
return None
|
|
except Exception as e:
|
|
await session.rollback()
|
|
logger.warning("sender link failed for %s: %s",address,e)
|
|
return None
|
|
|
|
@classmethod
|
|
async def insert_email(
|
|
cls,
|
|
session: AsyncSession,
|
|
email_data: dict,
|
|
file_path: list[str] | None = None,
|
|
):
|
|
"""Returns (row, new_user_email). new_user_email is set only when this call
|
|
created the sender's Users row."""
|
|
fields = cls._fields_from_email(email_data, file_path)
|
|
external_id = fields.get("message_id")
|
|
link_user=None
|
|
if external_id:
|
|
existing = (
|
|
await session.execute(
|
|
select(cls).where(cls.message_id == external_id)
|
|
)
|
|
).scalars().first()
|
|
if existing:
|
|
for key, value in fields.items():
|
|
# Keep prior S3 URLs until attach_email_pdfs_to_s3 replaces them.
|
|
if key in ("file_path","file_name") and not value:
|
|
continue
|
|
setattr(existing, key, value)
|
|
session.add(existing)
|
|
await session.commit()
|
|
await session.refresh(existing)
|
|
|
|
if fields.get("attachment") or existing.attachment:
|
|
link_user=await cls._link_sender(session, email_data, existing)
|
|
# _link_sender may rollback (IntegrityError); that expires this row
|
|
await session.refresh(existing)
|
|
return existing, link_user
|
|
|
|
email = cls(**fields)
|
|
session.add(email)
|
|
await session.commit()
|
|
await session.refresh(email)
|
|
|
|
if fields.get("attachment"):
|
|
link_user=await cls._link_sender(session, email_data, email)
|
|
await session.refresh(email)
|
|
return email, link_user
|
|
|
|
@classmethod
|
|
async def get_linked_user_id(cls,session:AsyncSession,message_id):
|
|
try:
|
|
mid=uuid.UUID(str(message_id))
|
|
except (ValueError,TypeError):
|
|
return None
|
|
return (
|
|
await session.execute(select(Inbox.user_id).where(Inbox.message_id==mid))
|
|
).scalar_one_or_none()
|
|
|
|
@classmethod
|
|
async def set_file_paths(cls,session:AsyncSession,record_id,file_paths,file_names=None):
|
|
row=await cls.get_inbox_message_by_id(session,record_id)
|
|
if not row:
|
|
return None
|
|
paths=file_paths if isinstance(file_paths,list) else ([file_paths] if file_paths else [])
|
|
cleaned=[str(p).strip() for p in paths if p and str(p).strip()]
|
|
row.file_path=",".join(cleaned) if cleaned else None
|
|
row.attachment=bool(cleaned)
|
|
if file_names is not None:
|
|
names=file_names if isinstance(file_names,list) else [file_names]
|
|
row.file_name=",".join(str(n).strip() for n in names if n and str(n).strip()) or row.file_name
|
|
session.add(row)
|
|
await session.commit()
|
|
await session.refresh(row)
|
|
return row
|
|
|
|
@classmethod
|
|
async def delete_by_id(cls,session:AsyncSession,record_id):
|
|
"""Hard-delete message + inbox links — roll back when S3 upload fails after insert."""
|
|
row=await cls.get_inbox_message_by_id(session,record_id)
|
|
if not row:
|
|
return False
|
|
links=(await session.execute(select(Inbox).where(Inbox.message_id==row.id))).scalars().all()
|
|
for link in links:
|
|
session.delete(link)
|
|
session.delete(row)
|
|
await session.commit()
|
|
return True
|
|
|
|
@classmethod
|
|
def _search_filter(cls, search: str):
|
|
pattern = f"%{search}%"
|
|
return or_(
|
|
cls.message_subject.ilike(pattern),
|
|
cls.message_from.ilike(pattern),
|
|
cls.message_body.ilike(pattern),
|
|
)
|
|
|
|
@classmethod
|
|
def _apply_filters(
|
|
cls, statement, search: str | None=None, isread: bool=True,
|
|
application_status: Candidate_application_Status=Candidate_application_Status.CLOSED,
|
|
assigned: bool | None=None,
|
|
is_duplicate: bool | None=None,
|
|
):
|
|
"""The one WHERE chain shared by the list, the count and the bulk read UPDATE.
|
|
|
|
Works on a Select or an Update — both expose .where() — which is the whole
|
|
point: "mark all read in this view" must narrow on exactly the predicates the
|
|
list narrowed on. A scope filter that drifts from the list filter silently
|
|
touches rows the user never saw, and there is no undo for that.
|
|
"""
|
|
if search:
|
|
statement = statement.where(cls._search_filter(search))
|
|
if application_status == Candidate_application_Status.PROCESS or application_status==Candidate_application_Status.REJECTED:
|
|
statement = statement.where(cls.application_status==application_status)
|
|
if assigned is True:
|
|
statement = statement.where(cls.assigned_job_post_id.is_not(None))
|
|
elif assigned is False:
|
|
statement = statement.where(cls.assigned_job_post_id.is_(None))
|
|
if isread==False:
|
|
statement = statement.where(cls.message_read==False)
|
|
if is_duplicate is True:
|
|
statement = statement.where(cls.is_duplicate==True) # noqa: E712
|
|
elif is_duplicate is False:
|
|
statement = statement.where(cls.is_duplicate==False) # noqa: E712
|
|
return statement
|
|
|
|
@classmethod
|
|
async def get_inbox_messages(
|
|
cls, session: AsyncSession, top: int | None, skip: int, search: str | None, isread: bool=True, application_status: Candidate_application_Status=Candidate_application_Status.CLOSED, assigned: bool | None=None, is_duplicate: bool | None=None
|
|
):
|
|
# Page size is the caller's `top` (Inbox sends 10/25/50/100); `skip` is
|
|
# (page-1)*top so page 1 of 25 -> 0..24, page 2 -> 25..49. Newest first
|
|
# via created_at.
|
|
statement = cls._apply_filters(
|
|
select(cls).order_by(cls.created_at.desc()),
|
|
search, isread, application_status, assigned, is_duplicate,
|
|
)
|
|
if skip:
|
|
statement = statement.offset(skip)
|
|
|
|
if top is not None:
|
|
statement = statement.limit(top)
|
|
|
|
result = await session.execute(statement)
|
|
return result.scalars().all()
|
|
|
|
@classmethod
|
|
async def get_inbox_message_by_id(cls, session: AsyncSession, record_id: str):
|
|
try:
|
|
uid = uuid.UUID(str(record_id))
|
|
except ValueError:
|
|
return None
|
|
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."""
|
|
row = await cls.get_inbox_message_by_id(session, record_id)
|
|
if not row:
|
|
return None
|
|
if job_post_id is None:
|
|
row.assigned_job_post_id = None
|
|
else:
|
|
try:
|
|
row.assigned_job_post_id = uuid.UUID(str(job_post_id))
|
|
except ValueError:
|
|
return None
|
|
session.add(row)
|
|
await session.commit()
|
|
await session.refresh(row)
|
|
return row
|
|
|
|
@classmethod
|
|
async def counts_by_job_post_ids(cls, session: AsyncSession, job_post_ids) -> dict[str, int]:
|
|
"""Resolve {job_post_id: applicant_count} for a page of rows in a single query.
|
|
|
|
Counts Inbox_Messages rows, not Inbox rows: one message fans out to several
|
|
Inbox rows (one per recipient), so counting Inbox would over-count.
|
|
"""
|
|
uids = {u for u in (job_post_ids or []) if u}
|
|
if not uids:
|
|
return {}
|
|
result = await session.execute(
|
|
select(cls.assigned_job_post_id, func.count().label("applicants"))
|
|
.where(cls.assigned_job_post_id.in_(uids))
|
|
.group_by(cls.assigned_job_post_id)
|
|
)
|
|
return {str(job_id): int(n) for job_id, n in result.all()}
|
|
|
|
@classmethod
|
|
async def count_inbox_messages(cls, session: AsyncSession, search: str | None, isread: bool=True, application_status: Candidate_application_Status=Candidate_application_Status.CLOSED, assigned: bool | None=None, is_duplicate: bool | None=None):
|
|
statement = cls._apply_filters(
|
|
select(func.count()).select_from(cls),
|
|
search, isread, application_status, assigned, is_duplicate,
|
|
)
|
|
result = await session.execute(statement)
|
|
return result.scalar_one()
|
|
|
|
@classmethod
|
|
async def apply_read_status(cls, session: AsyncSession, changes) -> int:
|
|
"""[{id, isRead, ...}] -> bulk UPDATE message_read. Returns rows touched.
|
|
|
|
read is a ONE-WAY LATCH: only false -> true is applied, never the reverse.
|
|
mark_message_read writes the local column only — nothing pushes the state
|
|
back to Outlook — so upstream keeps reporting isRead=false and the
|
|
every-minute sync_read_status sweep would otherwise revert a mail the user
|
|
just opened. Cost of the latch: un-reading a mail in Outlook no longer
|
|
propagates here.
|
|
|
|
Rows with read_overridden_at set are excluded outright. The latch alone is not
|
|
enough once the UI can mark UNREAD: a mail that is read in Outlook keeps being
|
|
reported isRead=true, so the next sweep would undo the recruiter's click within
|
|
the minute. A human decision on this row wins permanently; the only rows
|
|
excluded are ones somebody already decided about.
|
|
"""
|
|
if not changes:
|
|
return 0
|
|
read_ids=[c.get("id") for c in changes if c.get("id") and c.get("isRead")]
|
|
if not read_ids:
|
|
return 0
|
|
result=await session.execute(
|
|
update(cls)
|
|
.where(cls.message_id.in_(read_ids),cls.read_overridden_at.is_(None))
|
|
.values(message_read=True)
|
|
)
|
|
await session.commit()
|
|
return result.rowcount or 0
|
|
|
|
@classmethod
|
|
async def get_by_upstream_id(cls, session: AsyncSession, message_id):
|
|
"""Lookup by the UPSTREAM Graph id, not the local PK.
|
|
|
|
get_inbox_message_by_id above takes the uuid primary key; the triage ledger is
|
|
keyed on the upstream id, so overturning a verdict needs this direction.
|
|
"""
|
|
result=await session.execute(select(cls).where(cls.message_id == str(message_id)))
|
|
return result.scalars().first()
|
|
|
|
@classmethod
|
|
async def existing_message_ids(cls, session: AsyncSession, message_ids) -> set:
|
|
"""The subset of upstream ids already persisted — the free half of the gate.
|
|
|
|
A message already in this table was judged an application once, so the intake
|
|
classifier must never be paid for a second time; insert_email's upsert still
|
|
refreshes the row. One query per fetch round, columns only.
|
|
"""
|
|
ids=[str(m) for m in message_ids or [] if m]
|
|
if not ids:
|
|
return set()
|
|
result=await session.execute(
|
|
select(cls.message_id).where(cls.message_id.in_(ids))
|
|
)
|
|
return {row for (row,) in result.all() if row}
|
|
|
|
@classmethod
|
|
async def mark_message_read(cls, session: AsyncSession, record_id, read: bool=True):
|
|
row=await cls.get_inbox_message_by_id(session,record_id)
|
|
if not row:
|
|
return None
|
|
row.message_read=bool(read)
|
|
row.read_overridden_at=_now()
|
|
session.add(row)
|
|
await session.commit()
|
|
await session.refresh(row)
|
|
return row
|
|
|
|
@classmethod
|
|
async def set_read_bulk(cls, session: AsyncSession, record_ids, read: bool) -> int:
|
|
"""Flip read state for an explicit id list in ONE statement. Returns rows matched.
|
|
|
|
Unparseable ids are dropped rather than raising: a stale row id in a selection
|
|
must not sink the other 49 the recruiter ticked. The caller compares `updated`
|
|
against `requested` to notice.
|
|
|
|
No `message_read != read` predicate here — the caller wants to know how many of
|
|
its ids actually EXIST, which is what rowcount reports without it.
|
|
"""
|
|
uids=[]
|
|
for raw in record_ids or []:
|
|
try:
|
|
uids.append(uuid.UUID(str(raw)))
|
|
except (AttributeError, TypeError, ValueError):
|
|
continue
|
|
if not uids:
|
|
return 0
|
|
result=await session.execute(
|
|
update(cls).where(cls.id.in_(uids)).values(message_read=bool(read),read_overridden_at=_now())
|
|
)
|
|
await session.commit()
|
|
return result.rowcount or 0
|
|
|
|
@classmethod
|
|
async def set_read_scope(
|
|
cls, session: AsyncSession, read: bool, search: str | None=None, isread: bool=True,
|
|
application_status: Candidate_application_Status=Candidate_application_Status.CLOSED,
|
|
assigned: bool | None=None,
|
|
is_duplicate: bool | None=None,
|
|
) -> int:
|
|
"""Mark every row matching a list filter. Returns rows actually CHANGED.
|
|
|
|
The extra `message_read != read` predicate is what makes the count honest: the
|
|
recruiter is told "12 marked read", not "1,240 rows touched" on a mailbox that
|
|
was already read. It also keeps read_overridden_at off rows nobody decided
|
|
anything about, so the Outlook sweep keeps its reach over untouched mail.
|
|
"""
|
|
statement=cls._apply_filters(update(cls),search,isread,application_status,assigned,is_duplicate)
|
|
statement=statement.where(cls.message_read!=bool(read))
|
|
result=await session.execute(
|
|
statement.values(message_read=bool(read),read_overridden_at=_now())
|
|
)
|
|
await session.commit()
|
|
return result.rowcount or 0
|
|
|
|
@classmethod
|
|
async def count_processing(cls, session: AsyncSession):
|
|
statement = select(
|
|
func.count().label("all_count"),
|
|
func.coalesce(func.sum(case((cls.message_read == False, 1), else_=0)), 0).label("unread"), # noqa: E712
|
|
func.coalesce(func.sum(case((cls.processing_state == "imported", 1), else_=0)), 0).label("imported"),
|
|
func.coalesce(func.sum(case((cls.application_status == Candidate_application_Status.PROCESS, 1), else_=0)), 0).label("processed"),
|
|
func.coalesce(func.sum(case((cls.application_status == Candidate_application_Status.REJECTED, 1), else_=0)), 0).label("rejected"),
|
|
func.coalesce(func.sum(case((cls.is_duplicate == True, 1), else_=0)), 0).label("duplicates"), # noqa: E712
|
|
func.coalesce(func.sum(case((cls.assigned_job_post_id.is_not(None), 1), else_=0)), 0).label("assigned"),
|
|
func.coalesce(func.sum(case((cls.assigned_job_post_id.is_(None), 1), else_=0)), 0).label("unassigned"),
|
|
)
|
|
row = (await session.execute(statement)).one()
|
|
return {
|
|
"all": int(row.all_count or 0),
|
|
"unread": int(row.unread or 0),
|
|
"imported": int(row.imported or 0),
|
|
"processed": int(row.processed or 0),
|
|
"rejected": int(row.rejected or 0),
|
|
"duplicates": int(row.duplicates or 0),
|
|
"assigned": int(row.assigned or 0),
|
|
"unassigned": int(row.unassigned or 0),
|
|
}
|
|
|
|
@classmethod
|
|
async def set_processing_state(cls, session: AsyncSession, record_id, processing_state: str):
|
|
row = await cls.get_inbox_message_by_id(session, record_id)
|
|
if not row:
|
|
return None
|
|
row.processing_state = processing_state
|
|
session.add(row)
|
|
await session.commit()
|
|
await session.refresh(row)
|
|
return row
|
|
|
|
@classmethod
|
|
async def set_duplicate(cls, session: AsyncSession, record_id, is_duplicate: bool):
|
|
row = await cls.get_inbox_message_by_id(session, record_id)
|
|
if not row:
|
|
return None
|
|
row.is_duplicate = bool(is_duplicate)
|
|
session.add(row)
|
|
await session.commit()
|
|
await session.refresh(row)
|
|
return row
|
|
|
|
|
|
class Inbox_Message_Triage(SQLModel, table=True):
|
|
"""One intake verdict per upstream message id — the gate before inbox_messages.
|
|
|
|
Rows land here for BOTH outcomes. Rejections are the point: inbox_messages stays
|
|
application-only, and a repeated /email/fetch never re-pays for the same
|
|
classification. Acceptances are recorded too, so a round that classified and then
|
|
failed to insert does not pay twice either.
|
|
|
|
Deliberately no message_body column: the body is what this feature keeps out of the
|
|
database, and the override route re-reads the mail from upstream by message_id.
|
|
message_subject is kept (capped in inbox_classifier.decorators.triage_fields)
|
|
because a review screen without it is unusable — it is stored, never logged.
|
|
|
|
server_default is load-bearing on every NOT NULL column: alembic_setup runs with
|
|
compare_server_default=True, so a model default without a matching server default
|
|
autogenerates a drift revision on every boot.
|
|
"""
|
|
|
|
__tablename__ = "inbox_message_triage"
|
|
|
|
id: uuid.UUID = Field(default_factory=uuid.uuid4, primary_key=True)
|
|
# Upstream Graph id — the same key insert_email upserts on.
|
|
message_id: str = Field(index=True, unique=True)
|
|
is_application: bool = Field(default=False, sa_column_kwargs={"server_default": "false"})
|
|
reason_code: str = Field(default="", sa_column_kwargs={"server_default": ""})
|
|
confidence: float | None = Field(default=None)
|
|
evidence: str = Field(default="", sa_column_kwargs={"server_default": ""})
|
|
# Triage_Status: classified | low_confidence | error. Plain text, not a PG enum —
|
|
# alembic autogenerate cannot see new enum labels, and the enum in
|
|
# inbox_classifier/enums.py already gates what code writes here.
|
|
status: str = Field(default="classified", sa_column_kwargs={"server_default": "classified"})
|
|
error: str | None = Field(default=None)
|
|
model_name: str = Field(default="", sa_column_kwargs={"server_default": ""})
|
|
message_subject: str = Field(default="", sa_column_kwargs={"server_default": ""})
|
|
message_from: str = Field(default="", sa_column_kwargs={"server_default": ""})
|
|
message_received_time: str = Field(default="", sa_column_kwargs={"server_default": ""})
|
|
file_name: str = Field(default="", sa_column_kwargs={"server_default": ""})
|
|
attachment: bool = Field(default=False, sa_column_kwargs={"server_default": "false"})
|
|
ingested: bool = Field(default=False, sa_column_kwargs={"server_default": "false"})
|
|
overridden_by_id: uuid.UUID | None = Field(default=None, foreign_key="users.id")
|
|
overridden_at: datetime | None = Field(default=None, sa_type=DateTime(timezone=True))
|
|
# _now(), never datetime.now(): a naive local value bound to a timestamptz column
|
|
# is read back as UTC and silently backdates the row (see AtsResults below).
|
|
classified_at: datetime = Field(default_factory=_now, sa_type=DateTime(timezone=True))
|
|
created_at: datetime = Field(default_factory=_now, sa_type=DateTime(timezone=True))
|
|
|
|
@staticmethod
|
|
def _as_uuid(record_id) -> uuid.UUID | None:
|
|
if record_id in (None, ""):
|
|
return None
|
|
try:
|
|
return uuid.UUID(str(record_id))
|
|
except ValueError:
|
|
return None
|
|
|
|
@classmethod
|
|
async def get_by_message_id(cls, session: AsyncSession, message_id):
|
|
result=await session.execute(select(cls).where(cls.message_id == str(message_id)))
|
|
return result.scalars().first()
|
|
|
|
@classmethod
|
|
async def get_triage_by_id(cls, session: AsyncSession, record_id):
|
|
rid=cls._as_uuid(record_id)
|
|
if rid is None:
|
|
return None
|
|
result=await session.execute(select(cls).where(cls.id == rid))
|
|
return result.scalars().first()
|
|
|
|
@classmethod
|
|
async def verdicts_for_message_ids(cls, session: AsyncSession, message_ids) -> dict:
|
|
"""{upstream message_id: is_application} for a whole fetch round, one query."""
|
|
ids=[str(m) for m in message_ids or [] if m]
|
|
if not ids:
|
|
return {}
|
|
result=await session.execute(
|
|
select(cls.message_id, cls.is_application).where(cls.message_id.in_(ids))
|
|
)
|
|
return {message_id: bool(is_application) for message_id, is_application in result.all()}
|
|
|
|
@classmethod
|
|
async def record_verdict(cls, session: AsyncSession, fields: dict):
|
|
"""Upsert one verdict on message_id.
|
|
|
|
Two fetch rounds can race the unique index, so IntegrityError rolls back and
|
|
re-reads rather than failing the round — same shape as _link_sender above.
|
|
"""
|
|
message_id=str(fields.get("message_id") or "")
|
|
if not message_id:
|
|
return None
|
|
existing=await cls.get_by_message_id(session, message_id)
|
|
if existing:
|
|
for key, value in fields.items():
|
|
setattr(existing, key, value)
|
|
existing.classified_at=_now()
|
|
session.add(existing)
|
|
await session.commit()
|
|
await session.refresh(existing)
|
|
return existing
|
|
row=cls(**fields)
|
|
session.add(row)
|
|
try:
|
|
await session.commit()
|
|
except IntegrityError:
|
|
await session.rollback()
|
|
return await cls.get_by_message_id(session, message_id)
|
|
await session.refresh(row)
|
|
return row
|
|
|
|
@classmethod
|
|
def _triage_filter(cls, statement, is_application, status, search):
|
|
if is_application is not None:
|
|
statement=statement.where(cls.is_application == bool(is_application))
|
|
if status:
|
|
statement=statement.where(cls.status == str(status))
|
|
if search:
|
|
pattern=f"%{search}%"
|
|
statement=statement.where(
|
|
or_(cls.message_subject.ilike(pattern), cls.message_from.ilike(pattern))
|
|
)
|
|
return statement
|
|
|
|
@classmethod
|
|
async def list_triage(cls, session: AsyncSession, top, skip, is_application=None,
|
|
status=None, search=None):
|
|
statement=cls._triage_filter(select(cls), is_application, status, search)
|
|
statement=statement.order_by(cls.classified_at.desc()).offset(skip).limit(top)
|
|
result=await session.execute(statement)
|
|
return list(result.scalars().all())
|
|
|
|
@classmethod
|
|
async def count_triage(cls, session: AsyncSession, is_application=None, status=None,
|
|
search=None) -> int:
|
|
statement=cls._triage_filter(select(func.count(cls.id)), is_application, status, search)
|
|
result=await session.execute(statement)
|
|
return int(result.scalar() or 0)
|
|
|
|
@classmethod
|
|
async def set_override(cls, session: AsyncSession, record_id, is_application, user_id=None):
|
|
row=await cls.get_triage_by_id(session, record_id)
|
|
if not row:
|
|
return None
|
|
row.is_application=bool(is_application)
|
|
row.overridden_by_id=cls._as_uuid(user_id)
|
|
row.overridden_at=_now()
|
|
session.add(row)
|
|
await session.commit()
|
|
await session.refresh(row)
|
|
return row
|
|
|
|
@classmethod
|
|
async def mark_ingested(cls, session: AsyncSession, message_id, ingested: bool = True):
|
|
row=await cls.get_by_message_id(session, message_id)
|
|
if not row:
|
|
return None
|
|
row.ingested=bool(ingested)
|
|
session.add(row)
|
|
await session.commit()
|
|
await session.refresh(row)
|
|
return row
|
|
|
|
|
|
class SourceChannels(SQLModel, table=True):
|
|
__tablename__ = "source_channels"
|
|
|
|
id: int | None = Field(default=None, primary_key=True)
|
|
key: str = Field(max_length=40, unique=True, index=True)
|
|
label: str
|
|
is_active: bool = Field(default=True)
|
|
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_by_id(cls, session: AsyncSession, record_id: int):
|
|
result = await session.execute(select(cls).where(cls.id == record_id))
|
|
return result.scalars().first()
|
|
|
|
@classmethod
|
|
async def get_by_key(cls, session: AsyncSession, key: str):
|
|
result = await session.execute(select(cls).where(cls.key == key))
|
|
return result.scalars().first()
|
|
|
|
@classmethod
|
|
async def list_active(cls, session: AsyncSession):
|
|
result = await session.execute(
|
|
select(cls).where(cls.is_active == True).order_by(cls.id.asc()) # noqa: E712
|
|
)
|
|
return list(result.scalars().all())
|
|
|
|
|
|
class AtsResults(SQLModel, table=True):
|
|
__tablename__ = "ats_results"
|
|
|
|
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. 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="")
|
|
is_current: bool = Field(default=True)
|
|
superseded_by_id: uuid.UUID | None = Field(default=None, foreign_key="ats_results.id")
|
|
model_name: str | None = Field(default=None)
|
|
# default_factory was datetime.now: a naive LOCAL value bound to a timestamptz
|
|
# column, which asyncpg reads as UTC. That silently backdated every row by the
|
|
# host's offset (+5 h here) instead of raising, unlike the naive-column case.
|
|
computed_at: datetime = Field(default_factory=_now, sa_type=DateTime(timezone=True))
|
|
created_at: datetime = Field(default_factory=_now, sa_type=DateTime(timezone=True))
|
|
|
|
@staticmethod
|
|
def _as_uuid(record_id) -> uuid.UUID | None:
|
|
if record_id in (None, ""):
|
|
return None
|
|
try:
|
|
return uuid.UUID(str(record_id))
|
|
except ValueError:
|
|
return None
|
|
|
|
@classmethod
|
|
async def get_current_for_inbox(cls, session: AsyncSession, inbox_id: int):
|
|
result = await session.execute(
|
|
select(cls)
|
|
.where(cls.inbox_id == int(inbox_id), cls.is_current == True) # noqa: E712
|
|
.order_by(cls.computed_at.desc())
|
|
)
|
|
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 —
|
|
stable across re-scores because upsert_candidate keeps the same id for
|
|
the same (job, content_sha256)."""
|
|
result = await session.execute(
|
|
select(cls)
|
|
.where(cls.candidate_id == candidate_id, cls.is_current == True) # noqa: E712
|
|
.order_by(cls.computed_at.desc())
|
|
)
|
|
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
|
|
|
|
|
|
class MailboxSyncRun(SQLModel, table=True):
|
|
"""One Outlook mailbox sync job — survives tab close because work runs in Taskiq.
|
|
|
|
The Sync button enqueues a run and returns immediately. The UI polls this row for
|
|
status / per-message entries / triage totals. Closing the browser does not cancel
|
|
the worker.
|
|
"""
|
|
|
|
__tablename__ = "mailbox_sync_runs"
|
|
|
|
id: uuid.UUID = Field(default_factory=uuid.uuid4, primary_key=True)
|
|
status: str = Field(default="queued", index=True) # queued|running|completed|failed
|
|
task_id: str | None = Field(default=None)
|
|
created_by: uuid.UUID | None = Field(default=None, foreign_key="users.id")
|
|
top: int = Field(default=100)
|
|
skip: int = Field(default=0)
|
|
test_on: bool = Field(default=True)
|
|
triage: dict | None = Field(default=None, sa_column=Column(JSONB))
|
|
entries: list | None = Field(default=None, sa_column=Column(JSONB))
|
|
error: str | None = Field(default=None)
|
|
created_at: datetime = Field(default_factory=_now, sa_type=DateTime(timezone=True))
|
|
started_at: datetime | None = Field(default=None, sa_type=DateTime(timezone=True))
|
|
finished_at: datetime | None = Field(default=None, sa_type=DateTime(timezone=True))
|
|
|
|
@staticmethod
|
|
def _as_uuid(record_id) -> uuid.UUID | None:
|
|
if record_id in (None, ""):
|
|
return None
|
|
try:
|
|
return uuid.UUID(str(record_id))
|
|
except ValueError:
|
|
return None
|
|
|
|
@classmethod
|
|
async def get_by_id(cls, session: AsyncSession, record_id):
|
|
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_active(cls, session: AsyncSession):
|
|
result = await session.execute(
|
|
select(cls)
|
|
.where(cls.status.in_(("queued", "running")))
|
|
.order_by(cls.created_at.desc())
|
|
)
|
|
return result.scalars().first()
|
|
|
|
@classmethod
|
|
async def insert_run(cls, session: AsyncSession, fields: dict, *, commit: bool = True):
|
|
row = cls(**fields)
|
|
session.add(row)
|
|
if commit:
|
|
await session.commit()
|
|
await session.refresh(row)
|
|
return row
|
|
|
|
@classmethod
|
|
async def update_run(cls, session: AsyncSession, record_id, fields: dict, *, commit: bool = True):
|
|
row = await cls.get_by_id(session, record_id)
|
|
if not row:
|
|
return None
|
|
for key, value in fields.items():
|
|
setattr(row, key, value)
|
|
session.add(row)
|
|
if commit:
|
|
await session.commit()
|
|
await session.refresh(row)
|
|
return row
|