HR-ATS-Portal/backend/job/candidate/models.py

import uuid
from datetime import datetime, timedelta, timezone
from typing import TYPE_CHECKING, List, Optional

from fastapi import HTTPException
from sqlalchemy import JSON, DateTime, Index, UniqueConstraint, and_, 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

from linkedin_utils import NO_SLUG, primary_slug_from_text, slug_from_url

if TYPE_CHECKING:
    from inbox.models import Inbox
    from users.models import Users
    from job.job_post.models import JobPosts


def _now() -> datetime:
    return datetime.now(timezone.utc)


# Every datetime below is aware (see _now, and the API parses ISO input carrying
# an offset), so each column is declared timestamptz. SQLModel maps a bare
# `datetime` to TIMESTAMP WITHOUT TIME ZONE, and asyncpg refuses to bind an aware
# value to one — "can't subtract offset-naive and offset-aware datetimes" — which
# turns every insert here into a 500. Same pairing as job/job_post/models.py.
class Manual_UPLOAD_CANDIDATE(SQLModel, table=True):
    __tablename__ = "manual_upload_candidate"
    id: uuid.UUID = Field(default_factory=uuid.uuid4, primary_key=True)
    candidate_email: str = Field(default="")
    candidate_name: str = Field(default="")
    candidate_phone: str = Field(default="")
    job_post_id: uuid.UUID | None = Field(default=None, foreign_key="job_posts.id")
    full_text: str = Field(default="")
    # Lowercase /in/<slug> from full_text ("" = scanned, none found; NULL =
    # not yet scanned — see linkedin_utils). Same contract as
    # inbox_messages.linkedin_slug; Find Talent matches on it.
    linkedin_slug: str | None = Field(default=None, index=True)
    # Canonical profile URL for the LinkedIn button. Written at CV ingest;
    # fetch reads this, it does not re-parse full_text.
    linkedin_url: str | None = Field(default=None)
    current_company: str = Field(default="")
    # Candidate's role at that company (e.g. "Senior Merchandiser"). Distinct
    # 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="")
    # Employment-agent extractions, written at CV-bank ingest (see 029). These
    # are what make the bank searchable — full_text alone cannot be filtered on.
    # `experience` above is free text from the Add Candidate form; this one is
    # the numeric years the bank filters and sorts by, so they stay separate.
    skills: list[str] = Field(default_factory=list, sa_type=JSONB, sa_column_kwargs={"server_default": "[]"})
    years_experience: int | None = Field(default=None)
    city: str | None = Field(default=None)
    professional_summary: str | None = Field(default=None)
    reapplied: list[str] = Field(default_factory=list, sa_type=JSONB, sa_column_kwargs={"server_default": "[]"})
    education: str = Field(default="", sa_column_kwargs={"server_default": ""})
    # Why the CV is held (speculative / referral) and when retention expires.
    bank_reason: str = Field(default="", sa_column_kwargs={"server_default": ""})
    bank_expires_at: datetime | None = Field(default=None, sa_type=DateTime(timezone=True))
    # Candidate_application_Status value (PENDING, SCREENING, …). Empty reads as Shortlist.
    status: str = Field(default="")
    # Free text, not a users FK: a referrer is often someone outside the system
    referral_by: str = Field(default="", sa_column_kwargs={"server_default": ""})
    file_name: str = Field(default="", sa_column_kwargs={"server_default": ""})
    file_path: str = Field(default="", sa_column_kwargs={"server_default": ""})
    # Recruiter-set, same shape as inbox.favorite/inbox.rating — this row's own
    # copy, not derived from Inbox (manual_upload_candidate has no inbox link).
    favorite: Optional[bool] = Field(default=False)
    rating: Optional[float] = Field(default=0.0)
    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, job_post_id=None, job_post_ids=None, limit=None, offset=0, search=None):
        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_name,
                    cls.candidate_phone,
                    JobPosts.title,
                    cls.status,
                    cls.current_company,
                    cls.current_position,
                    cls.experience,
                    cls.platform,
                    cls.apply_via,
                    cls.linkedin_url,
                    Users.linkedin_url.label("user_linkedin_url"),
                    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
                )
                # 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_ids is not None:
                ids=list(job_post_ids)
                if not ids:
                    return []
                qry=qry.where(cls.job_post_id.in_(ids))
            elif job_post_id:
                qry=qry.where(cls.job_post_id==job_post_id)
            if search and str(search).strip():
                like=f"%{str(search).strip()}%"
                qry=qry.where(or_(
                    Users.name.ilike(like),
                    Users.email.ilike(like),
                    cls.candidate_name.ilike(like),
                    cls.candidate_email.ilike(like),
                ))
            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": float(row["overall_score"]) if row["overall_score"] is not None else None,
                        "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"] or "").strip() or (row["candidate_name"] or "").strip() or None,
                    "candidate_phone":row["candidate_phone"],
                    "title":row["title"] or None,
                    "application_status":row["status"] or None,
                    "current_company":row["current_company"] or None,
                    "current_position":row["current_position"] or None,
                    "experience":row["experience"] or None,
                    "platform":row["platform"] or None,
                    "apply_via":row["apply_via"] or None,
                    "linkedin_url":row["linkedin_url"] or row["user_linkedin_url"] or None,
                    "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))

    @classmethod
    async def count_by_status(cls, session: AsyncSession, job_post_id=None, search=None):
        try:
            from users.models import Users
            from job.job_post.models import JobPosts
            qry=(
                select(cls.status,func.count())
                .select_from(cls)
                .join(Users,cls.user_id==Users.id)
                .join(JobPosts,cls.job_post_id==JobPosts.id)
                .group_by(cls.status)
            )
            if job_post_id:
                qry=qry.where(cls.job_post_id==job_post_id)
            if search and str(search).strip():
                like=f"%{str(search).strip()}%"
                qry=qry.where(or_(
                    Users.name.ilike(like),
                    Users.email.ilike(like),
                    cls.candidate_name.ilike(like),
                    cls.candidate_email.ilike(like),
                ))
            result=await session.execute(qry)
            counts={}
            for status,n in result.all():
                key=(status or "").strip() or "UNKNOWN"
                counts[key]=counts.get(key,0)+int(n or 0)
            return counts
        except Exception as e:
            raise HTTPException(status_code=500,detail=str(e))

    @classmethod
    async def counts_by_application_status(
        cls, session: AsyncSession, from_date=None, to_date=None, department=None, recruiter_id=None,
    ):
        """Current-stage counts for the same manual-upload population the board uses.

        Inner-joined to a job post, same as count_by_status. Optional department /
        recruiter / created_at window sit on top; with none of those this matches
        the board's manual_upload column.
        """
        from users.models import Users
        from job.job_post.models import JobPosts
        qry=(
            select(cls.status,func.count())
            .select_from(cls)
            .join(Users,cls.user_id==Users.id)
            .join(JobPosts,cls.job_post_id==JobPosts.id)
        )
        if from_date is not None:
            qry=qry.where(cls.created_at>=from_date)
        if to_date is not None:
            qry=qry.where(cls.created_at<to_date)
        if department:
            qry=qry.where(JobPosts.department==department)
        try:
            rid=uuid.UUID(str(recruiter_id)) if recruiter_id not in (None,"") else None
        except (TypeError,ValueError):
            rid=None
        if rid is not None:
            qry=qry.where(JobPosts.has_recruiter(rid))
        qry=qry.group_by(cls.status)
        result=await session.execute(qry)
        counts={}
        for status,n in result.all():
            key=(status or "").strip() or "UNKNOWN"
            counts[key]=counts.get(key,0)+int(n or 0)
        return counts

    @classmethod
    async def counts_by_job_post(
        cls, session: AsyncSession, from_date=None, to_date=None, department=None, recruiter_id=None,
    ):
        """Per-job counts for the same manual-upload population as
        counts_by_application_status, so inbox + manual per-job sums stay equal
        to the funnel total under identical filters.
        """
        from users.models import Users
        from job.job_post.models import JobPosts
        qry=(
            select(cls.job_post_id,func.count())
            .select_from(cls)
            .join(Users,cls.user_id==Users.id)
            .join(JobPosts,cls.job_post_id==JobPosts.id)
        )
        if from_date is not None:
            qry=qry.where(cls.created_at>=from_date)
        if to_date is not None:
            qry=qry.where(cls.created_at<to_date)
        if department:
            qry=qry.where(JobPosts.department==department)
        rid=cls._as_uuid(recruiter_id)
        if rid is not None:
            qry=qry.where(JobPosts.has_recruiter(rid))
        qry=qry.group_by(cls.job_post_id)
        result=await session.execute(qry)
        return {str(job_id):int(n or 0) for job_id,n in result.all()}

    @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 create_manual_upload_candidate(cls, session: AsyncSession, fields: dict):
        import os

        from users.models import Users
        from users.plugins import hash_password

        email=(fields.get("candidate_email") or "").strip().lower()
        name=(fields.get("candidate_name") or "").strip() or email
        default_pw=os.getenv("DEFAULT_CANDIDATE_PASSWORD","Utopia!@#")
        full_text=fields.get("full_text") or ""
        linkedin_url=(fields.get("linkedin_url") or "").strip() or None
        if linkedin_url:
            linkedin_slug=slug_from_url(linkedin_url) or NO_SLUG
        elif full_text:
            linkedin_slug=NO_SLUG
        else:
            linkedin_slug=None

        user=await Users.get_user_by_email(session,email)
        if not user:
            user=await Users.insert_user(session,{
                "name":name,
                "email":email,
                "role_id":8,
                "password":hash_password(default_pw),
                "is_active":True,
                "is_approved":True,
                "is_deleted":False,
                "linkedin_url":linkedin_url,
            })
        elif linkedin_url:
            await Users.set_linkedin_url_if_empty(session,user_id=user.id,url=linkedin_url)

        row=cls(
            candidate_email=email,
            candidate_name=name,
            candidate_phone=(fields.get("candidate_phone") or "").strip(),
            job_post_id=cls._as_uuid(fields.get("job_post_id")),
            full_text=full_text,
            linkedin_slug=linkedin_slug,
            linkedin_url=linkedin_url,
            current_company=(fields.get("current_company") or "").strip(),
            current_position=(fields.get("current_position") or "").strip(),
            apply_via=(fields.get("apply_via") or "manual_upload").strip() or "manual_upload",
            user_id=user.id,
            platform=(fields.get("platform") or "").strip(),
            created_by=cls._as_uuid(fields.get("created_by")),
            experience=(fields.get("experience") or "").strip(),
            status=(fields.get("status") or "").strip() or "PENDING",
            referral_by=(fields.get("referral_by") or "").strip(),
            file_name=(fields.get("file_name") or "").strip(),
            file_path=(fields.get("file_path") or "").strip(),
            city=(fields.get("city") or "").strip() or None,
        )
        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 one row — used to roll back when S3 upload fails after insert."""
        row=await cls.get_by_id(session,record_id)
        if not row:
            return False
        session.delete(row)
        await session.commit()
        return True

    @classmethod
    async def set_file_path(cls, session: AsyncSession, record_id, file_path, file_name=None):
        row=await cls.get_by_id(session,record_id)
        if not row:
            return None
        url=(file_path or "").strip()
        row.file_path=url
        if file_name is not None:
            row.file_name=(file_name or "").strip()
        session.add(row)
        if row.apply_via == "cv_bank":
            file_row = await CvBankFiles.get(session, row.id)
            if file_row:
                file_row.file_path = url or None
                session.add(file_row)
        await session.commit()
        await session.refresh(row)
        return row

    @classmethod
    async def get_by_user_id(cls, session: AsyncSession, user_id):
        uid = cls._as_uuid(user_id)
        if uid is None:
            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 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 set_rating_favorite(cls, session: AsyncSession, record_id, favorite=None, rating=None):
        row=await cls.get_by_id(session,record_id)
        if not row:
            return None
        if favorite is not None:
            row.favorite=favorite
        if rating is not None:
            row.rating=rating
        row.updated_at=_now()
        session.add(row)
        await session.commit()
        await session.refresh(row)
        return row

    @classmethod
    async def set_full_text(cls, session: AsyncSession, record_id, text, *, only_if_empty=True):
        """Keep extracted CV text on the bank/manual row (Inbox uses resume_text)."""
        row = await cls.get_by_id(session, record_id)
        if not row:
            return None
        value = (text or "").strip()
        if not value:
            return row
        if only_if_empty and (row.full_text or "").strip():
            return row
        row.full_text = value
        row.updated_at = _now()
        session.add(row)
        await session.commit()
        await session.refresh(row)
        return row

    @classmethod
    async def set_professional_summary(cls, session: AsyncSession, record_id, summary):
        row = await cls.get_by_id(session, record_id)
        if not row:
            return None
        row.professional_summary = (summary or "").strip() or None
        row.updated_at = _now()
        session.add(row)
        await session.commit()
        await session.refresh(row)
        return row

    @classmethod
    async def get_by_ids(cls, session: AsyncSession, ids):
        keys = []
        for raw in ids or []:
            uid = cls._as_uuid(raw)
            if uid is not None:
                keys.append(uid)
        if not keys:
            return []
        result = await session.execute(select(cls).where(cls.id.in_(keys)))
        return list(result.scalars().all())

    @classmethod
    async def get_by_email_and_job(cls, session: AsyncSession, email: str, job_post_id):
        """Idempotency for form / re-import promotes against the same role."""
        cleaned = (email or "").strip().lower()
        jid = cls._as_uuid(job_post_id)
        if not cleaned or jid is None:
            return None
        result = await session.execute(
            select(cls)
            .where(cls.candidate_email == cleaned, cls.job_post_id == jid)
            .order_by(cls.created_at.desc())
        )
        return result.scalars().first()

    @classmethod
    async def list_by_emails(cls, session: AsyncSession, emails):
        """Every manual/form/cv-bank row for these addresses, newest first."""
        from job.job_post.models import JobPosts

        lowers = sorted({(e or "").strip().lower() for e in (emails or []) if (e or "").strip()})
        if not lowers:
            return []
        result = await session.execute(
            select(cls, JobPosts.title)
            .outerjoin(JobPosts, cls.job_post_id == JobPosts.id)
            .where(func.lower(cls.candidate_email).in_(lowers))
            .order_by(cls.created_at.desc())
        )
        rows = []
        for rec, title in result.all():
            status = (rec.status or "").strip() or None
            rows.append({
                "source": "manual",
                "email": (rec.candidate_email or "").strip().lower() or None,
                "inbox_id": None,
                "message_id": None,
                "manual_upload_candidate_id": str(rec.id),
                "form_data_id": None,
                "candidate_id": None,
                "user_id": str(rec.user_id) if rec.user_id else None,
                "job_post_id": str(rec.job_post_id) if rec.job_post_id else None,
                "job_title": title or None,
                "status": status or "PENDING",
                "applied_at": rec.created_at.isoformat() if rec.created_at else None,
            })
        return rows

    @classmethod
    async def job_post_ids_by_emails(cls, session: AsyncSession, emails):
        """(email, job_post_id) pairs that actually have a job. Unlinked rows skipped."""
        lowers=sorted({(e or "").strip().lower() for e in (emails or []) if (e or "").strip()})
        if not lowers:
            return []
        result=await session.execute(
            select(cls.candidate_email,cls.job_post_id)
            .where(func.lower(cls.candidate_email).in_(lowers))
            .where(cls.job_post_id.is_not(None))
        )
        rows=[]
        for email,job_id in result.all():
            if job_id is None:
                continue
            rows.append(((email or "").strip().lower(),str(job_id)))
        return rows

    @classmethod
    async def set_reapplied_by_emails(cls, session: AsyncSession, mapping):
        if not mapping:
            return 0
        updated=0
        for email, ids in mapping.items():
            key=(email or "").strip().lower()
            if not key:
                continue
            result=await session.execute(
                update(cls).where(func.lower(cls.candidate_email)==key).values(reapplied=list(ids or []))
            )
            updated+=result.rowcount or 0
        await session.commit()
        return updated

    @classmethod
    def _talent_pool_filters(cls, Users, *, search=None, job_post_ids=None):
        filters = [cls.user_id.is_not(None), cls.job_post_id.is_not(None)]
        if job_post_ids is not None:
            filters.append(cls.job_post_id.in_(list(job_post_ids)))
        if search:
            like = f"%{search.strip()}%"
            filters.append(
                or_(
                    cls.candidate_name.ilike(like),
                    cls.candidate_email.ilike(like),
                    Users.name.ilike(like),
                    Users.email.ilike(like),
                )
            )
        return filters

    @classmethod
    async def list_for_talent_pool(cls, session: AsyncSession, *, limit=100, offset=0, search=None, job_post_ids=None):
        """Newest applications with a user + assigned job for Candidates / Talent Pool."""
        from users.models import Users

        if job_post_ids is not None and not list(job_post_ids):
            return []
        statement = (
            select(cls)
            .join(Users, cls.user_id == Users.id)
            .where(*cls._talent_pool_filters(Users, search=search, job_post_ids=job_post_ids))
            .order_by(cls.created_at.desc())
            .limit(limit)
            .offset(offset)
        )
        result = await session.execute(statement)
        return list(result.scalars().all())

    @classmethod
    async def count_for_talent_pool(cls, session: AsyncSession, *, search=None, job_post_ids=None):
        from users.models import Users

        if job_post_ids is not None and not list(job_post_ids):
            return 0
        statement = (
            select(func.count())
            .select_from(cls)
            .join(Users, cls.user_id == Users.id)
            .where(*cls._talent_pool_filters(Users, search=search, job_post_ids=job_post_ids))
        )
        result = await session.execute(statement)
        return result.scalar_one()

    @classmethod
    async def sources_by_user_ids(cls, session: AsyncSession, user_ids) -> dict[str, str]:
        """Newest platform/apply_via label per user — Candidates Form badges."""
        parsed = []
        for raw in (user_ids or []):
            uid = cls._as_uuid(raw)
            if uid is not None:
                parsed.append(uid)
        if not parsed:
            return {}
        result = await session.execute(
            select(cls.user_id, cls.platform, cls.apply_via, cls.created_at)
            .where(cls.user_id.in_(parsed))
            .order_by(cls.created_at.desc())
        )
        out: dict[str, str] = {}
        for user_id, platform, apply_via, _created in result.all():
            key = str(user_id)
            if key in out:
                continue
            label = (platform or "").strip() or (apply_via or "").strip()
            if label:
                out[key] = label
        return out

    @classmethod
    async def file_paths_by_user_ids(cls, session: AsyncSession, user_ids) -> dict[str, str]:
        """Newest stored CV path per user — search Open resume when there is no inbox row."""
        parsed = []
        for raw in (user_ids or []):
            uid = cls._as_uuid(raw)
            if uid is not None:
                parsed.append(uid)
        if not parsed:
            return {}
        result = await session.execute(
            select(cls.user_id, cls.file_path)
            .where(cls.user_id.in_(parsed))
            .order_by(cls.created_at.desc())
        )
        out: dict[str, str] = {}
        for user_id, file_path in result.all():
            key = str(user_id)
            if key in out:
                continue
            first = (file_path or "").strip()
            if first:
                out[key] = first
        return out

    # ---- CV bank -----------------------------------------------------------
    # apply_via="cv_bank" marks origin: the CV Import "No job" tab. Unassigned
    # rows (job_post_id IS NULL) are the bank; Job Matching assigns a job_post_id
    # (and a user account) so get_all's Users/JobPosts inner joins pick them up
    # as normal applications. apply_via stays "cv_bank" so Matching can still
    # list them. No inbox entry.

    @classmethod
    async def insert_bank_cv(cls, session: AsyncSession, *, candidate_email,
                             candidate_name, full_text, file_name,
                             created_by, pdf_bytes,
                             content_type="application/pdf",
                             linkedin_url=None, profile=None,
                             bank_reason="speculative", retention_months=None):
        """Bank a CV: metadata row + its bytes (cv_bank_files) in one commit.
        file_path is filled after S3 upload under Temp/{id}/ (see cv_bank_upload).

        linkedin_url is the employment-agent extraction (None when the CV has
        none — never a constructed slug). When the CV carries an email, the
        candidate ACCOUNT is created/reused so the person shows up on the
        Candidates screen; unlike an application there is still no inbox entry,
        no scoring, and no setup email. A CV with no detectable email banks
        fine and simply stays account-less.

        `profile` is the rest of the employment-agent extraction (company,
        title, education, phone, skills, years) — the only structured data a
        banked CV gets, since nothing scores it until it is matched to a job.
        `retention_months` stamps bank_expires_at so the CV is held for a
        stated period rather than indefinitely."""
        import os

        from role.models import EnumRoles, Roles
        from users.models import Users
        from users.plugins import hash_password

        email = (candidate_email or "").strip().lower()
        user = None
        if email:
            user = await Users.get_user_by_email(session, email)
            if not user:
                role = await Roles.get_role_by_name(session, EnumRoles.CANDIDATE.value)
                user = await Users.insert_user(session, {
                    "name": (candidate_name or "").strip() or email,
                    "email": email,
                    "role_id": role.id if role else 8,
                    "password": hash_password(os.getenv("DEFAULT_CANDIDATE_PASSWORD", "Utopia!@#")),
                    "is_active": True,
                    "is_approved": True,
                    "is_deleted": False,
                })
            elif user.is_deleted or not user.is_active:
                # get_user_by_email returns soft-deleted accounts too; a fresh
                # bank upload means the person is relevant again.
                user.is_deleted = False
                user.is_active = True
                session.add(user)

        extracted = profile or {}
        url = (linkedin_url or extracted.get("linkedin_url") or "").strip() or None
        if url:
            linkedin_slug = slug_from_url(url) or NO_SLUG
        else:
            linkedin_slug = primary_slug_from_text(full_text or "")
        if user and url:
            await Users.set_linkedin_url_if_empty(session, user_id=user.id, url=url)

        expires_at = None
        if retention_months:
            expires_at = _now() + timedelta(days=30 * int(retention_months))

        row = cls(
            candidate_email=email,
            candidate_name=(candidate_name or "").strip() or (email or ""),
            candidate_phone=(extracted.get("candidate_phone") or "").strip(),
            job_post_id=None,
            full_text=full_text or "",
            linkedin_slug=linkedin_slug,
            linkedin_url=url,
            current_company=(extracted.get("current_company") or "").strip(),
            current_position=(extracted.get("current_position") or "").strip(),
            education=(extracted.get("education") or "").strip(),
            skills=extracted.get("skills") or [],
            years_experience=extracted.get("years_experience"),
            experience="" if extracted.get("years_experience") is None else str(extracted.get("years_experience")),
            bank_reason=(bank_reason or "").strip(),
            bank_expires_at=expires_at,
            apply_via="cv_bank",
            user_id=user.id if user else None,
            created_by=cls._as_uuid(created_by),
            status="BANKED",
            file_name=(file_name or "").strip(),
            file_path="",
            city=(extracted.get("city") or "").strip() or None,
        )
        session.add(row)
        await session.flush()
        session.add(CvBankFiles(
            manual_upload_candidate_id=row.id,
            content_type=content_type,
            file_name=(file_name or "").strip(),
            file_path=None,
            data=pdf_bytes,
        ))
        if user and row.city:
            await Users.set_city_if_empty(session, user_id=user.id, city=row.city)
        await session.commit()
        await session.refresh(row)
        return row

    @classmethod
    async def list_bank(cls, session: AsyncSession, limit=100, offset=0):
        """Speculative CVs (`apply_via=cv_bank`), including ones later linked to a job.

        Run ATS assigns job_post_id so the row joins Pipeline like any other
        application. The bank still lists them so the ATS score is visible on
        this screen. Matching remains the assign queue for unassigned rows.
        """
        bank = (cls.apply_via == "cv_bank",)
        total = (
            await session.execute(
                select(func.count()).select_from(cls).where(*bank)
            )
        ).scalar() or 0
        result = await session.execute(
            select(cls)
            .where(*bank)
            .order_by(cls.created_at.desc(), cls.id.desc())
            .limit(limit)
            .offset(offset)
        )
        return list(result.scalars().all()), total

    @classmethod
    async def list_bank_needing_profile(cls, session: AsyncSession, limit=50):
        """Bank rows stored before the extraction existed (see 029).

        Skills is the marker: a CV that genuinely lists none still gets its
        title or years filled, so an empty skills array plus a blank title
        means the agent never ran, not that the CV was sparse.
        """
        result = await session.execute(
            select(cls)
            .where(
                cls.apply_via == "cv_bank",
                cls.full_text != "",
                func.coalesce(func.jsonb_array_length(cls.skills), 0) == 0,
                or_(cls.current_position == "", cls.current_position.is_(None)),
            )
            .order_by(cls.created_at.desc())
            .limit(limit)
        )
        return list(result.scalars().all())

    @classmethod
    async def set_bank_profile(cls, session: AsyncSession, record_id, profile):
        """Write an employment-agent extraction onto an existing bank row.

        Only fills blanks — a recruiter may have corrected the company or title
        by hand, and a backfill must not overwrite that.
        """
        row = await session.get(cls, cls._as_uuid(record_id))
        if row is None:
            return None
        extracted_name = (profile.get("candidate_name") or "").strip()
        current_name = (row.candidate_name or "").strip()
        email = (row.candidate_email or "").strip()
        if extracted_name and (not current_name or current_name.lower() == email.lower()):
            row.candidate_name = extracted_name
        if not (row.current_company or "").strip():
            row.current_company = (profile.get("current_company") or "").strip()
        if not (row.current_position or "").strip():
            row.current_position = (profile.get("current_position") or "").strip()
        if not (row.education or "").strip():
            row.education = (profile.get("education") or "").strip()
        if not (row.candidate_phone or "").strip():
            row.candidate_phone = (profile.get("candidate_phone") or "").strip()
        if not row.skills:
            row.skills = profile.get("skills") or []
        if row.years_experience is None:
            row.years_experience = profile.get("years_experience")
        if not (row.experience or "").strip() and row.years_experience is not None:
            row.experience = str(row.years_experience)
        if not (row.linkedin_url or "").strip() and profile.get("linkedin_url"):
            row.linkedin_url = profile["linkedin_url"]
            row.linkedin_slug = slug_from_url(profile["linkedin_url"]) or NO_SLUG
        if not (row.city or "").strip() and profile.get("city"):
            row.city = (profile.get("city") or "").strip() or None
        row.updated_at = _now()
        session.add(row)
        await session.commit()
        await session.refresh(row)
        return row

    @classmethod
    async def list_matching(cls, session: AsyncSession, *, assigned=None,
                            search=None, limit=100, offset=0):
        """No-job-tab origin (`apply_via=cv_bank`). `assigned` is tri-valued:
        None = all, False = still in the bank, True = job_post_id set."""
        filters = [cls.apply_via == "cv_bank"]
        if assigned is True:
            filters.append(cls.job_post_id.is_not(None))
        elif assigned is False:
            filters.append(cls.job_post_id.is_(None))
        if search and str(search).strip():
            like = f"%{str(search).strip()}%"
            filters.append(or_(
                cls.candidate_name.ilike(like),
                cls.candidate_email.ilike(like),
                cls.file_name.ilike(like),
            ))
        total = (
            await session.execute(
                select(func.count()).select_from(cls).where(*filters)
            )
        ).scalar() or 0
        result = await session.execute(
            select(cls)
            .where(*filters)
            .order_by(cls.created_at.desc(), cls.id.desc())
            .limit(limit)
            .offset(offset)
        )
        return list(result.scalars().all()), total

    @classmethod
    async def _ensure_bank_user(cls, session: AsyncSession, row):
        """Candidate USER so get_all / Candidates can see the row after assign.
        insert_user commits; caller must reload `row` afterwards."""
        if row.user_id:
            return None
        import os

        from role.models import EnumRoles, Roles
        from users.models import Users
        from users.plugins import hash_password

        email = (row.candidate_email or "").strip().lower()
        if not email:
            email = f"cvbank-{row.id.hex}@no-email.local"
        user = await Users.get_user_by_email(session, email)
        if not user:
            role = await Roles.get_role_by_name(session, EnumRoles.CANDIDATE.value)
            name = (row.candidate_name or "").strip() or (row.file_name or "").strip() or email
            user = await Users.insert_user(session, {
                "name": name,
                "email": email,
                "role_id": role.id if role else 8,
                "password": hash_password(os.getenv("DEFAULT_CANDIDATE_PASSWORD", "Utopia!@#")),
                "is_active": True,
                "is_approved": True,
                "is_deleted": False,
            })
        elif user.is_deleted or not user.is_active:
            user.is_deleted = False
            user.is_active = True
            session.add(user)
        return user

    @classmethod
    async def assign_job_post(cls, session: AsyncSession, record_id, job_post_id):
        """Set job_post_id on a CV-bank row. None unassigns (back to the bank).
        Origin apply_via stays cv_bank. Creates/reuses a candidate user so the
        row joins like any other manual_upload_candidate application."""
        row = await cls.get_by_id(session, record_id)
        if not row or row.apply_via != "cv_bank":
            return None
        jid = cls._as_uuid(job_post_id) if job_post_id not in (None, "") else None
        user = await cls._ensure_bank_user(session, row)
        row = await cls.get_by_id(session, record_id)
        if not row:
            return None
        if user is not None:
            row.user_id = user.id
            if not (row.candidate_email or "").strip() and user.email:
                row.candidate_email = user.email
            if not (row.candidate_name or "").strip() and user.name:
                row.candidate_name = user.name
        row.job_post_id = jid
        row.status = "PENDING" if jid else "BANKED"
        row.updated_at = _now()
        session.add(row)
        await session.commit()
        await session.refresh(row)
        return row

    @classmethod
    async def list_bank_for_ranking(cls, session: AsyncSession, limit=5000):
        """Every CV still in the bank, for tier-1 ranking against a new job.

        Expired CVs are excluded: ranking one would put a candidate in front of
        a recruiter after the retention window said to stop holding them.
        """
        result = await session.execute(
            select(cls)
            .where(
                cls.apply_via == "cv_bank",
                cls.job_post_id.is_(None),
                or_(cls.bank_expires_at.is_(None), cls.bank_expires_at > _now()),
            )
            .order_by(cls.created_at.desc())
            .limit(limit)
        )
        return list(result.scalars().all())

    @classmethod
    async def list_bank_expired(cls, session: AsyncSession, limit=500):
        """Bank CVs past their retention window, for the nightly sweep."""
        result = await session.execute(
            select(cls)
            .where(
                cls.apply_via == "cv_bank",
                cls.job_post_id.is_(None),
                cls.bank_expires_at.is_not(None),
                cls.bank_expires_at <= _now(),
            )
            .order_by(cls.bank_expires_at.asc())
            .limit(limit)
        )
        return list(result.scalars().all())

    @classmethod
    async def delete_bank_cv(cls, session: AsyncSession, record_id):
        """Hard delete unassigned bank rows only — assigned rows are applications.
        The cv_bank_files row goes with it via ON DELETE CASCADE."""
        row = await cls.get_by_id(session, record_id)
        if not row or row.apply_via != "cv_bank" or row.job_post_id is not None:
            return None
        file_row = await CvBankFiles.get(session, row.id)
        if file_row:
            await session.delete(file_row)
        await session.delete(row)
        await session.commit()
        return row


class CvBankMatches(SQLModel, table=True):
    """Tier-1 rank of one banked CV against one job post (see 030).

    Persisted rather than computed on read because the whole point is to tell a
    recruiter the bank already holds candidates the moment a job opens — a
    notification cannot wait for someone to open the screen.
    """

    __tablename__ = "cv_bank_matches"
    __table_args__ = (
        UniqueConstraint(
            "manual_upload_candidate_id", "job_post_id",
            name="uq_cv_bank_matches_pair",
        ),
    )

    id: uuid.UUID = Field(default_factory=uuid.uuid4, primary_key=True)
    manual_upload_candidate_id: uuid.UUID = Field(foreign_key="manual_upload_candidate.id")
    job_post_id: uuid.UUID = Field(foreign_key="job_posts.id", index=True)
    rank_score: int = Field(default=0)
    computed_at: datetime = Field(default_factory=_now, sa_type=DateTime(timezone=True))

    @classmethod
    async def replace_for_job(cls, session: AsyncSession, job_post_id, scores):
        """Swap in a fresh ranking for one job.

        Delete-then-insert rather than upsert: a re-rank after the job's
        requirements were edited must not leave behind scores for CVs that have
        since been assigned or deleted.
        """
        jid = job_post_id if isinstance(job_post_id, uuid.UUID) else uuid.UUID(str(job_post_id))
        existing = await session.execute(select(cls).where(cls.job_post_id == jid))
        for row in existing.scalars().all():
            await session.delete(row)
        await session.flush()
        for record_id, score in scores:
            session.add(cls(
                manual_upload_candidate_id=record_id,
                job_post_id=jid,
                rank_score=int(score or 0),
            ))
        await session.commit()
        return len(scores)

    @classmethod
    async def scores_for_job(cls, session: AsyncSession, job_post_id) -> dict:
        """rank_score keyed by manual_upload_candidate_id, as strings."""
        jid = job_post_id if isinstance(job_post_id, uuid.UUID) else uuid.UUID(str(job_post_id))
        result = await session.execute(
            select(cls.manual_upload_candidate_id, cls.rank_score).where(cls.job_post_id == jid)
        )
        return {str(record_id): int(score) for record_id, score in result.all()}


class CvBankFiles(SQLModel, table=True):
    """PDF bytes of a CV-bank entry — in the database so production redeploys
    (ephemeral container filesystems) can never lose a stored CV. Created in
    prod by migrations/manual/010_cv_bank_files.sql. file_path is the same
    permanent S3 URL written to manual_upload_candidate.file_path."""

    __tablename__ = "cv_bank_files"

    manual_upload_candidate_id: uuid.UUID = Field(
        primary_key=True, foreign_key="manual_upload_candidate.id",
    )
    content_type: str = Field(default="application/pdf")
    file_name: str | None = Field(default=None)
    file_path: str | None = Field(default=None)
    data: bytes
    created_at: datetime = Field(default_factory=_now, sa_type=DateTime(timezone=True))

    @classmethod
    async def get(cls, session: AsyncSession, manual_upload_candidate_id):
        result = await session.execute(
            select(cls).where(cls.manual_upload_candidate_id == manual_upload_candidate_id)
        )
        return result.scalars().first()


class Candidates(SQLModel, table=True):


    __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" origin metadata, not an FK
    filename: str
    file_path: str | None = Field(default=None)  # permanent S3 URL (same as manual_upload_candidate / inbox)

    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)
    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)
    professional_summary: str | None = Field(default=None)
    # Public LinkedIn URL extracted from the scored CV. Fetch reads this column.
    linkedin_url: str | None = Field(default=None)
    # Application that owns the extracted CV text. Inbox email → inbox_messages.resume_text;
    # Add Candidate / CV bank → manual_upload_candidate.full_text. Migration 036.
    inbox_message_id: uuid.UUID | None = Field(default=None, foreign_key="inbox_messages.id", index=True)
    manual_upload_candidate_id: uuid.UUID | None = Field(
        default=None, foreign_key="manual_upload_candidate.id", index=True
    )

    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,
        limit: int | None = None,
        offset: int = 0,
    ):
        """Most-recent-first list, paged. Score is only a tiebreak within an instant.

        job_id=None returns the whole pool across jobs (same ordering) for the
        frontend's unscoped Candidates/Talent Pool views. Returns (rows, total) so
        the caller can page without a second count query of its own.
        """
        statement = select(cls)
        if job_id is not None:
            uid = cls._as_uuid(job_id)
            if uid is None:
                return [], 0
            statement = statement.where(cls.job_id == uid)
        total = (
            await session.execute(select(func.count()).select_from(statement.subquery()))
        ).scalar_one()
        statement = statement.order_by(
            cls.created_at.desc(),
            cls.status.asc(),  # "completed" < "failed"
            cls.match_score.desc().nulls_last(),
            cls.id.desc(),
        )
        if offset:
            statement = statement.offset(offset)
        if limit is not None:
            statement = statement.limit(limit)
        result = await session.execute(statement)
        return list(result.scalars().all()), total

    @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 list_by_emails(cls, session: AsyncSession, emails):
        """Scored `candidates` rows for these addresses (ATS, not pipeline stage)."""
        from job.job_post.models import JobPosts

        lowers = sorted({(e or "").strip().lower() for e in (emails or []) if (e or "").strip()})
        if not lowers:
            return []
        result = await session.execute(
            select(cls, JobPosts.title)
            .outerjoin(JobPosts, cls.job_id == JobPosts.id)
            .where(func.lower(cls.candidate_email).in_(lowers))
            .order_by(cls.created_at.desc())
        )
        rows = []
        for rec, title in result.all():
            email = (rec.candidate_email or "").strip().lower() or None
            rows.append({
                "source": "ats",
                "email": email,
                "inbox_id": None,
                "message_id": str(rec.inbox_message_id) if getattr(rec, "inbox_message_id", None) else None,
                "manual_upload_candidate_id": str(rec.manual_upload_candidate_id) if getattr(rec, "manual_upload_candidate_id", None) else None,
                "form_data_id": None,
                "candidate_id": str(rec.id),
                "job_post_id": str(rec.job_id) if rec.job_id else None,
                "job_title": title or rec.job_title or None,
                "status": rec.status or None,
                "applied_at": rec.created_at.isoformat() if rec.created_at else None,
            })
        return rows

    @classmethod
    async def latest_completed_by_emails(cls, session: AsyncSession, emails):
        """Newest completed ATS score per email, with the job it ran against.

        CV Bank speculative rows have no candidates FK; email is the join the
        scorer already writes (score_bank → upsert_candidate).
        """
        from job.job_post.models import JobPosts

        lowers = sorted({(e or "").strip().lower() for e in (emails or []) if (e or "").strip()})
        if not lowers:
            return {}
        result = await session.execute(
            select(cls, JobPosts.title)
            .outerjoin(JobPosts, cls.job_id == JobPosts.id)
            .where(func.lower(cls.candidate_email).in_(lowers))
            .where(cls.status == "completed")
            .where(cls.match_score.is_not(None))
            .order_by(cls.updated_at.desc(), cls.created_at.desc())
        )
        out = {}
        for rec, title in result.all():
            email = (rec.candidate_email or "").strip().lower()
            if not email or email in out:
                continue
            out[email] = {
                "match_score": int(rec.match_score) if rec.match_score is not None else None,
                "job_post_id": str(rec.job_id) if rec.job_id else None,
                "job_title": title or rec.job_title or None,
            }
        return out

    @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

    @classmethod
    async def sync_s3_file_path(cls, session: AsyncSession, email, job_id, file_path):
        """Stamp the Manual/Email S3 URL onto every Candidates row for this email+job.

        Same link as manual_upload_candidate.file_path — scoring may create the
        Candidates row after Add Candidate, so both create and score call this.
        """
        url=(file_path or "").strip()
        normalized=(email or "").strip().lower()
        jid=cls._as_uuid(job_id)
        if not url or not normalized or jid is None:
            return 0
        result=await session.execute(
            select(cls).where(func.lower(cls.candidate_email)==normalized,cls.job_id==jid)
        )
        rows=list(result.scalars().all())
        if not rows:
            return 0
        for row in rows:
            row.file_path=url
            row.updated_at=_now()
            session.add(row)
        await session.commit()
        return len(rows)


class Interviews(SQLModel, table=True):
    __tablename__ = "interviews"

    id: uuid.UUID = Field(default_factory=uuid.uuid4, primary_key=True)
    interview_date: datetime = Field(default_factory=_now, sa_type=DateTime(timezone=True))
    interview_time: datetime = Field(default_factory=_now, sa_type=DateTime(timezone=True))
    interview_type: str = Field(default="")
    interview_status: str = Field(default="")
    inbox_id: int | None = Field(default=None, foreign_key="inbox.id")
    # Optional denorm so Recruiter Hub can join interviews → job_posts.current_recruiter_id
    # without walking inbox. Filled on create from the application's assigned job;
    # migration 011 added the columns. user_id is the candidate, not the recruiter.
    user_id: uuid.UUID | None = Field(default=None, foreign_key="users.id")
    job_post_id: uuid.UUID | None = Field(default=None, foreign_key="job_posts.id")
    graph_event_id: str | None = Field(default=None)
    web_link: str | None = Field(default=None)
    inbox: Optional["Inbox"] = Relationship(
        back_populates="interviews",
        sa_relationship_kwargs={"lazy": "selectin"},
    )

    @staticmethod
    def _as_uuid(record_id) -> uuid.UUID | None:
        try:
            return uuid.UUID(str(record_id))
        except ValueError:
            return None

    @classmethod
    def _with_inbox_message(cls):
        from inbox.models import Inbox

        return selectinload(cls.inbox).options(
            selectinload(Inbox.messages),
            selectinload(Inbox.user),
        )

    @classmethod
    async def get_interview_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).options(cls._with_inbox_message()).where(cls.id == uid)
        )
        return result.scalars().first()

    @classmethod
    async def get_interviews_by_ids(cls, session: AsyncSession, ids):
        keys=[]
        for raw in ids or []:
            uid=cls._as_uuid(raw)
            if uid is not None:
                keys.append(uid)
        if not keys:
            return []
        result=await session.execute(select(cls).where(cls.id.in_(keys)))
        return list(result.scalars().all())

    @classmethod
    async def get_interviews_by_inbox(cls, session: AsyncSession, inbox_id: int):
        result = await session.execute(
            select(cls)
            .options(cls._with_inbox_message())
            .where(cls.inbox_id == inbox_id)
            .order_by(cls.interview_date.desc())
        )
        return result.scalars().all()

    @classmethod
    def scoped_to_recruiter(cls, statement, recruiter_id):
        """Restrict an Interviews select to the recruiter who owns the job.

        COALESCE(interviews.job_post_id, inbox_messages.assigned_job_post_id)
        → job_posts.current_recruiter_id. Interviews with no job drop out.
        """
        from inbox.models import Inbox, Inbox_Messages
        from job.job_post.models import JobPosts

        rid = cls._as_uuid(recruiter_id)
        if rid is None:
            return statement
        job_id = func.coalesce(cls.job_post_id, Inbox_Messages.assigned_job_post_id)
        return (
            statement
            .outerjoin(Inbox, cls.inbox_id == Inbox.id)
            .outerjoin(Inbox_Messages, Inbox.message_id == Inbox_Messages.id)
            .outerjoin(JobPosts, JobPosts.id == job_id)
            .where(JobPosts.has_recruiter(rid))
        )

    @classmethod
    async def get_interviews_in_range(
        cls,
        session: AsyncSession,
        *,
        from_date=None,
        to_date=None,
        status: str | None = None,
        recruiter_id=None,
        top: int | None = None,
        skip: int = 0,
    ):
        statement = select(cls)
        if from_date is not None:
            statement = statement.where(cls.interview_date >= from_date)
        if to_date is not None:
            statement = statement.where(cls.interview_date < to_date)
        if status:
            statement = statement.where(cls.interview_status == status)
        if recruiter_id:
            statement = cls.scoped_to_recruiter(statement, recruiter_id)
        count_statement = select(func.count()).select_from(statement.subquery())
        total = (await session.execute(count_statement)).scalar_one()
        statement = (
            statement.options(cls._with_inbox_message()).order_by(cls.interview_date.asc())
        )
        if skip:
            statement = statement.offset(skip)
        if top is not None:
            statement = statement.limit(top)
        result = await session.execute(statement)
        return list(result.scalars().all()), total

    @classmethod
    async def count_between(cls, session: AsyncSession, start, end, recruiter_id=None):
        statement = select(func.count()).select_from(cls).where(
            cls.interview_date >= start,
            cls.interview_date < end,
        )
        if recruiter_id:
            statement = cls.scoped_to_recruiter(statement, recruiter_id)
        result = await session.execute(statement)
        return int(result.scalar_one() or 0)

    @classmethod
    async def count_upcoming(cls, session: AsyncSession, as_of, recruiter_id=None):
        statement = select(func.count()).select_from(cls).where(
            cls.interview_status.ilike("scheduled"),
            cls.interview_date >= as_of,
        )
        if recruiter_id:
            statement = cls.scoped_to_recruiter(statement, recruiter_id)
        result = await session.execute(statement)
        return int(result.scalar_one() or 0)

    @classmethod
    async def next_scheduled_at(cls, session: AsyncSession, as_of, recruiter_id=None):
        statement = select(
            func.min(func.coalesce(cls.interview_time, cls.interview_date))
        ).select_from(cls).where(
            cls.interview_status.ilike("scheduled"),
            cls.interview_date >= as_of,
        )
        if recruiter_id:
            statement = cls.scoped_to_recruiter(statement, recruiter_id)
        result = await session.execute(statement)
        return result.scalar_one()

    @classmethod
    async def job_titles_by_inbox(cls, session: AsyncSession, inbox_ids) -> dict[int, str]:
        """Resolve {inbox_id: job_title} for a page of interview rows.

        Two constraints keep this off get_interviews_in_range:
        1. Interviews.inbox is selectin, but Inbox.messages defaults to lazy="select"
           and raises MissingGreenlet under AsyncSession. Switching that relation to
           selectin would extra-query the candidate list, pipeline, activity and inbox.
        2. Widening the range statement with a join would INNER-join the count
           subquery and drop interviews whose application has no assigned requisition,
           changing `total` on Interviews and Calendar.

        INNER joins are correct here — an unassigned inbox simply produces no dict
        entry and .get() yields None. No response rows are lost because the rows
        still come from the untouched range statement.
        """
        from inbox.models import Inbox, Inbox_Messages
        from job.job_post.models import JobPosts

        ids = {int(i) for i in (inbox_ids or []) if i is not None}
        if not ids:
            return {}
        result = await session.execute(
            select(Inbox.id, JobPosts.title)
            .join(Inbox_Messages, Inbox.message_id == Inbox_Messages.id)
            .join(JobPosts, Inbox_Messages.assigned_job_post_id == JobPosts.id)
            .where(Inbox.id.in_(ids))
        )
        return {int(inbox_id): title for inbox_id, title in result.all()}

    @classmethod
    async def insert_interview(cls, session: AsyncSession, fields: dict):
        row = cls(**fields)
        session.add(row)
        await session.commit()
        return await cls.get_interview_by_id(session, row.id)

    @classmethod
    async def update_interview(cls, session: AsyncSession, record_id, fields: dict):
        row = await cls.get_interview_by_id(session, record_id)
        if not row:
            return None
        for key, value in fields.items():
            setattr(row, key, value)
        session.add(row)
        await session.commit()
        await session.refresh(row)
        return row

    @classmethod
    async def set_calendar_event(cls, session: AsyncSession, record_id, event_id, web_link):
        row = await cls.get_interview_by_id(session, record_id)
        if not row:
            return None
        row.graph_event_id = event_id
        row.web_link = web_link
        session.add(row)
        await session.commit()
        await session.refresh(row)
        return row


class Notes(SQLModel, table=True):
    __tablename__ = "notes"

    id: uuid.UUID = Field(default_factory=uuid.uuid4, primary_key=True)
    note: str = Field(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))
    user_id: uuid.UUID | None = Field(default=None, foreign_key="users.id")
    created_by: uuid.UUID | None = Field(default=None, foreign_key="users.id")
    user: Optional["Users"] = Relationship(
        back_populates="notes",
        sa_relationship_kwargs={"lazy": "selectin", "foreign_keys": "[Notes.user_id]"},
    )
    author: Optional["Users"] = Relationship(
        back_populates="authored_notes",
        sa_relationship_kwargs={"lazy": "selectin", "foreign_keys": "[Notes.created_by]"},
    )

    @staticmethod
    def _as_uuid(record_id) -> uuid.UUID | None:
        try:
            return uuid.UUID(str(record_id))
        except ValueError:
            return None

    @classmethod
    async def get_note_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_notes_by_user(cls, session: AsyncSession, user_id):
        uid = cls._as_uuid(user_id)
        if uid is None:
            return []
        result = await session.execute(
            select(cls).where(cls.user_id == uid).order_by(cls.created_at.desc())
        )
        return result.scalars().all()

    @classmethod
    async def insert_note(cls, session: AsyncSession, fields: dict):
        row = cls(**fields)
        session.add(row)
        await session.commit()
        return await cls.get_note_by_id(session, row.id)

    @classmethod
    async def update_note(cls, session: AsyncSession, record_id, fields: dict):
        row = await cls.get_note_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 Activity(SQLModel, table=True):
    __tablename__ = "activity"

    id: uuid.UUID = Field(default_factory=uuid.uuid4, primary_key=True)
    activity_type: str = Field(default="")
    activity_date: datetime = Field(default_factory=_now, sa_type=DateTime(timezone=True))
    activity_time: datetime = Field(default_factory=_now, sa_type=DateTime(timezone=True))
    activity_status: str = Field(default="")
    description: str | None = Field(default=None)
    inbox_id: int | None = Field(default=None, foreign_key="inbox.id")
    inbox: Optional["Inbox"] = Relationship(
        back_populates="activity",
        sa_relationship_kwargs={"lazy": "selectin"},
    )

    @staticmethod
    def _as_uuid(record_id) -> uuid.UUID | None:
        try:
            return uuid.UUID(str(record_id))
        except ValueError:
            return None

    @classmethod
    async def get_activity_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_activity_by_inbox(cls, session: AsyncSession, inbox_id: int):
        result = await session.execute(
            select(cls).where(cls.inbox_id == inbox_id).order_by(cls.activity_date.desc())
        )
        return result.scalars().all()

    @classmethod
    async def get_activity_feed(cls, session: AsyncSession, *, top: int | None = None, skip: int = 0):
        statement = select(cls)
        count_statement = select(func.count()).select_from(cls)
        total = (await session.execute(count_statement)).scalar_one()
        statement = statement.order_by(cls.activity_date.desc(), cls.activity_time.desc())
        if skip:
            statement = statement.offset(skip)
        if top is not None:
            statement = statement.limit(top)
        result = await session.execute(statement)
        return list(result.scalars().all()), total

    @classmethod
    async def insert_activity(cls, session: AsyncSession, fields: dict):
        row = cls(**fields)
        session.add(row)
        await session.commit()
        return await cls.get_activity_by_id(session, row.id)

    @classmethod
    async def update_activity(cls, session: AsyncSession, record_id, fields: dict):
        row = await cls.get_activity_by_id(session, record_id)
        if not row:
            return None
        for key, value in fields.items():
            setattr(row, key, value)
        session.add(row)
        await session.commit()
        await session.refresh(row)
        return row


class Feedback(SQLModel, table=True):
    __tablename__ = "feedback"

    id: uuid.UUID = Field(default_factory=uuid.uuid4, primary_key=True)
    review: str = Field(default="")
    financial_status: str = Field(default="")
    score: float = Field(default=0.0)
    note: str | None = Field(default=None)
    created_at: datetime = Field(default_factory=_now, sa_type=DateTime(timezone=True))
    updated_at: datetime = Field(default_factory=_now, sa_type=DateTime(timezone=True))
    reviewed_by: uuid.UUID | None = Field(default=None, foreign_key="users.id")
    inbox_id: int | None = Field(default=None, foreign_key="inbox.id")
    user: Optional["Users"] = Relationship(
        back_populates="feedback",
        sa_relationship_kwargs={"lazy": "selectin"},
    )
    inbox: Optional["Inbox"] = Relationship(
        back_populates="feedback",
        sa_relationship_kwargs={"lazy": "selectin"},
    )

    @staticmethod
    def _as_uuid(record_id) -> uuid.UUID | None:
        try:
            return uuid.UUID(str(record_id))
        except ValueError:
            return None

    @classmethod
    async def get_feedback_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_feedback_by_inbox(cls, session: AsyncSession, inbox_id: int):
        result = await session.execute(
            select(cls).where(cls.inbox_id == inbox_id).order_by(cls.created_at.desc())
        )
        return result.scalars().all()

    @classmethod
    async def insert_feedback(cls, session: AsyncSession, fields: dict):
        row = cls(**fields)
        session.add(row)
        await session.commit()
        return await cls.get_feedback_by_id(session, row.id)

    @classmethod
    async def update_feedback(cls, session: AsyncSession, record_id, fields: dict):
        row = await cls.get_feedback_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 ApplicationStageTransitions(SQLModel, table=True):
    """Temporal history of application stage changes.

    Inbox moves write inbox_messages.application_status; manual-upload moves
    write manual_upload_candidate.status. Exactly one of inbox_id /
    manual_upload_candidate_id is set. NULL valid_to means the stage is current.
    """

    __tablename__ = "application_stage_transitions"

    id: uuid.UUID = Field(default_factory=uuid.uuid4, primary_key=True)
    inbox_id: int | None = Field(default=None, index=True, foreign_key="inbox.id")
    manual_upload_candidate_id: uuid.UUID | None = Field(
        default=None, index=True, foreign_key="manual_upload_candidate.id"
    )
    from_stage: str | None = Field(default=None)
    to_stage: str
    valid_from: datetime = Field(default_factory=_now, sa_type=DateTime(timezone=True))
    valid_to: datetime | None = Field(default=None, sa_type=DateTime(timezone=True))
    changed_by: uuid.UUID | None = Field(default=None, foreign_key="users.id")
    actor_kind: str = Field(default="user")
    change_reason: str | None = Field(default=None)
    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_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 fetch_by_inbox(cls, session: AsyncSession, inbox_id: int):
        result = await session.execute(
            select(cls).where(cls.inbox_id == int(inbox_id)).order_by(cls.valid_from.desc())
        )
        return list(result.scalars().all())

    @classmethod
    async def fetch_by_manual(cls, session: AsyncSession, record_id):
        uid = cls._as_uuid(record_id)
        if uid is None:
            return []
        result = await session.execute(
            select(cls).where(cls.manual_upload_candidate_id == uid).order_by(cls.valid_from.desc())
        )
        return list(result.scalars().all())

    @classmethod
    async def get_open_transition(cls, session: AsyncSession, inbox_id=None, manual_upload_candidate_id=None):
        statement = select(cls).where(cls.valid_to.is_(None)).order_by(cls.valid_from.desc())
        if inbox_id is not None:
            statement = statement.where(cls.inbox_id == int(inbox_id))
        elif manual_upload_candidate_id is not None:
            uid = cls._as_uuid(manual_upload_candidate_id)
            if uid is None:
                return None
            statement = statement.where(cls.manual_upload_candidate_id == uid)
        else:
            return None
        result = await session.execute(statement)
        return result.scalars().first()

    @classmethod
    async def insert_transition(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 close_open(cls, session: AsyncSession, inbox_id=None, *, manual_upload_candidate_id=None, at: datetime | None = None, commit: bool = False):
        row = await cls.get_open_transition(
            session, inbox_id=inbox_id, manual_upload_candidate_id=manual_upload_candidate_id
        )
        if not row:
            return None
        row.valid_to = at or _now()
        session.add(row)
        if commit:
            await session.commit()
            await session.refresh(row)
        return row

    @classmethod
    async def count_by_inbox(cls, session: AsyncSession, inbox_id: int):
        statement = select(func.count()).select_from(cls).where(cls.inbox_id == int(inbox_id))
        result = await session.execute(statement)
        return result.scalar_one()

    @classmethod
    async def avg_time_to_hire(
        cls, session: AsyncSession, from_date=None, to_date=None, department=None, recruiter_id=None,
    ):
        """Mean days from first stage (from_stage IS NULL) to HIRED, optional filters."""
        from inbox.enums import Candidate_application_Status
        from inbox.models import Inbox, Inbox_Messages
        from job.job_post.models import JobPosts

        entry = cls.__table__.alias("entry")
        hire = cls.__table__.alias("hire")
        days = func.extract("epoch", hire.c.valid_from - entry.c.valid_from) / 86400.0
        statement = (
            select(func.avg(days))
            .select_from(
                hire.join(
                    entry,
                    and_(
                        hire.c.inbox_id == entry.c.inbox_id,
                        entry.c.from_stage.is_(None),
                    ),
                )
            )
            .where(hire.c.to_stage == Candidate_application_Status.HIRED.value)
        )
        if from_date is not None:
            statement = statement.where(hire.c.valid_from >= from_date)
        if to_date is not None:
            statement = statement.where(hire.c.valid_from < to_date)
        if department or recruiter_id:
            statement = (
                statement
                .outerjoin(Inbox, hire.c.inbox_id == Inbox.id)
                .outerjoin(Inbox_Messages, Inbox.message_id == Inbox_Messages.id)
                .outerjoin(JobPosts, Inbox_Messages.assigned_job_post_id == JobPosts.id)
            )
            if department:
                statement = statement.where(JobPosts.department == department)
            rid = cls._as_uuid(recruiter_id)
            if rid is not None:
                statement = statement.where(
                    or_(Inbox_Messages.recruiter_id == rid, JobPosts.has_recruiter(rid))
                )
        result = await session.execute(statement)
        value = result.scalar_one()
        return float(value) if value is not None else None

    @classmethod
    def scoped_to_job(cls, statement, department=None, recruiter_id=None):
        if not department and not recruiter_id:
            return statement
        from inbox.models import Inbox, Inbox_Messages
        from job.job_post.models import JobPosts

        statement = (
            statement
            .outerjoin(Inbox, cls.inbox_id == Inbox.id)
            .outerjoin(Inbox_Messages, Inbox.message_id == Inbox_Messages.id)
            .outerjoin(JobPosts, Inbox_Messages.assigned_job_post_id == JobPosts.id)
        )
        if department:
            statement = statement.where(JobPosts.department == department)
        rid = cls._as_uuid(recruiter_id)
        if rid is not None:
            statement = statement.where(
                or_(Inbox_Messages.recruiter_id == rid, JobPosts.has_recruiter(rid))
            )
        return statement

    @classmethod
    async def count_hires(
        cls, session: AsyncSession, from_date=None, to_date=None, department=None, recruiter_id=None,
    ):
        from inbox.enums import Candidate_application_Status

        statement = select(func.count()).select_from(cls).where(
            cls.to_stage == Candidate_application_Status.HIRED.value
        )
        if from_date is not None:
            statement = statement.where(cls.valid_from >= from_date)
        if to_date is not None:
            statement = statement.where(cls.valid_from < to_date)
        statement = cls.scoped_to_job(statement, department, recruiter_id)
        result = await session.execute(statement)
        return int(result.scalar_one() or 0)

    @classmethod
    async def counts_hires_by_month(
        cls, session: AsyncSession, start, department=None, recruiter_id=None,
    ):
        from inbox.enums import Candidate_application_Status

        month_bucket = func.date_trunc("month", cls.valid_from)
        statement = (
            select(month_bucket.label("month"), func.count().label("count"))
            .select_from(cls)
            .where(
                cls.to_stage == Candidate_application_Status.HIRED.value,
                cls.valid_from >= start,
            )
            .group_by(month_bucket)
            .order_by(month_bucket)
        )
        statement = cls.scoped_to_job(statement, department, recruiter_id)
        result = await session.execute(statement)
        return [(month, int(count or 0)) for month, count in result.all()]


class CandidateHistory(SQLModel, table=True):
    """Append-only audit log for one candidate (users.id), scoped to an application.

    user_id is the anchor: the profile modal is keyed by users.id, so every read is
    one indexed scan. inbox_id / manual_upload_candidate_id record WHICH application
    the event happened on and are BOTH nullable -- unlike application_stage_transitions,
    several events (notes, rating changes, an import before the inbox link exists) have
    neither. That is why there is no XOR check constraint here.
    """

    __tablename__ = "candidate_history"
    __table_args__ = (
        Index("ix_candidate_history_user_created", "user_id", "created_at"),
    )

    id: uuid.UUID = Field(default_factory=uuid.uuid4, primary_key=True)
    user_id: uuid.UUID = Field(foreign_key="users.id", nullable=False)
    inbox_id: int | None = Field(default=None, index=True, foreign_key="inbox.id")
    manual_upload_candidate_id: uuid.UUID | None = Field(
        default=None, index=True, foreign_key="manual_upload_candidate.id"
    )
    event_type: str = Field(index=True)
    entity_type: str | None = Field(default=None)
    entity_id: str | None = Field(default=None)
    from_value: str | None = Field(default=None)
    to_value: str | None = Field(default=None)
    description: str | None = Field(default=None)
    actor_id: uuid.UUID | None = Field(default=None, foreign_key="users.id")
    actor_kind: str = Field(default="user")
    meta: dict | None = Field(default=None, sa_type=JSON)
    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 fetch_by_user(cls, session: AsyncSession, user_id, limit=200, offset=0):
        uid = cls._as_uuid(user_id)
        if uid is None:
            return [], 0
        total_stmt = select(func.count()).select_from(cls).where(cls.user_id == uid)
        total = (await session.execute(total_stmt)).scalar_one()
        result = await session.execute(
            select(cls)
            .where(cls.user_id == uid)
            .order_by(cls.created_at.desc(), cls.id.desc())
            .offset(int(offset))
            .limit(int(limit))
        )
        return list(result.scalars().all()), int(total)

    @classmethod
    async def insert_event(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


import users.models as _users_models  # noqa: E402, F401