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

from sqlalchemy.ext.asyncio import AsyncSession
import asyncio,base64,dataclasses,hashlib,io,logging,os,uuid
from datetime import date,datetime,timezone
from pathlib import Path
from dotenv import load_dotenv
from fastapi import HTTPException
from pypdf import PdfReader
from app.core.errors import ATSError,ErrorCode
from app.models.scoring import CompletedCandidate
from app.services.pdf import ExtractedResume,extract_resume,sanitize_filename
from app.services.scoring import score_batch
from inbox.models import Inbox_Messages,Inbox,Inbox_Message_Triage,InboxRescanRun,AtsResults
from job.candidate.models import Candidates
from job.candidate.plugins import (
    FILE_NOT_FOUND,
    build_job_description,
    candidate_completed_fields,
    candidate_failed_fields,
    contained_download_path,
    documents_from_message,
    extract_pdf_link_uris,
    extract_pdf_text,
    get_scorer,
    get_scoring_settings,
    normalize_spaced_text,
)
from g_sheet.models import FormData
from job.candidate.serializers import is_kept_application,serialize_application_history,serialize_application_history_item,serialize_candidate,serialize_candidate_profile,serialize_form_candidate_list,serialize_manual_candidate_list,serialize_manual_candidate_profile,serialize_manual_upload_candidate,serialize_matching_candidate,serialize_manager_candidate
from job.job_post.models import JobPosts
from job.job_post.serializers import serialize_job_post
from job.candidate.models import Notes,Manual_UPLOAD_CANDIDATE
from job.history.enums import HistoryEvent
from job.history.views import HistoryRecorder
from job.notes.serializers import serialize_note
from job.candidate.plugins import extract_candidate_email
from users.models import Users
from users.permissions import is_hiring_manager,sees_all_candidates,scopes_to_own_requisitions
from employment_agent.plugins import parse_phone

load_dotenv()
logger=logging.getLogger("job.candidate.views")
CV_QUEUE_NAME=os.getenv("TASKIQ_CV_QUEUE_NAME","cv_upload")
MANUAL_UPLOAD_TO_ADDRESS=os.getenv(
    "MANUAL_UPLOAD_TO_ADDRESS","manual-cv-upload@hr-ats.local"
)
MANAGER_SCOPE_DETAIL="You can only access candidates allocated to jobs opened from your requisitions"
CREATOR_SCOPE_DETAIL="You can only access candidates allocated to jobs you created"
_HISTORY_TABLES=("users","candidates","manual_upload_candidate","form_data")


def _norm_email(value):
    raw=(value or "").strip().lower()
    if not raw:
        return ""
    if "<" in raw and ">" in raw:
        inner=raw.rsplit("<",1)[-1]
        raw=inner.split(">",1)[0].strip()
    return raw


def _payload_email(payload):
    if not isinstance(payload,dict):
        return ""
    return _norm_email(
        payload.get("email")
        or payload.get("candidate_email")
        or payload.get("fromEmail")
    )


_ID_KEYS=(
    "id","inbox_id","message_id","upstream_id",
    "form_data_id","manual_upload_candidate_id","candidate_id",
)
_PAYLOAD_TIME_KEYS=(
    "received","when","message_sent_time","message_received_time",
    "entry_date","applied_at","created_at",
)


def _row_ids(obj):
    """Stable identifiers for one application row — not email, not job_post_id.

    List payloads use `source` for the Outlook To address, so matching by source
    string cannot work. Shared id values are what make 'this row' the same
    application the recruiter just clicked.
    """
    ids=set()
    if not isinstance(obj,dict):
        return ids
    for key in _ID_KEYS:
        value=obj.get(key)
        if value is None or value=="":
            continue
        ids.add(str(value))
    return ids


def _is_current_application(item,payload):
    """True when `item` is the same row the list/detail payload is showing."""
    if not isinstance(item,dict) or not isinstance(payload,dict):
        return False
    return bool(_row_ids(item)&_row_ids(payload))


def _as_utc(value):
    if value is None or value=="":
        return None
    if isinstance(value,datetime):
        dt=value
    elif isinstance(value,date):
        dt=datetime(value.year,value.month,value.day,tzinfo=timezone.utc)
    else:
        raw=str(value).strip()
        if not raw:
            return None
        if raw.endswith("Z"):
            raw=raw[:-1]+"+00:00"
        elif "T" not in raw[:20] and " " in raw:
            raw=raw.replace(" ","T",1)
        try:
            dt=datetime.fromisoformat(raw)
        except ValueError:
            return None
    if dt.tzinfo is None:
        dt=dt.replace(tzinfo=timezone.utc)
    return dt.astimezone(timezone.utc)


def _payload_applied_at(payload):
    if not isinstance(payload,dict):
        return None
    for key in _PAYLOAD_TIME_KEYS:
        dt=_as_utc(payload.get(key))
        if dt is not None:
            return dt
    return None


def _is_earlier_application(item,payload):
    """True when `item` happened before the open application.

    A later mail from the same person is not a previous attempt. Missing
    timestamps cannot be ordered, so those rows stay visible.
    """
    if not isinstance(item,dict) or not isinstance(payload,dict):
        return False
    current=_payload_applied_at(payload)
    other=_as_utc(item.get("applied_at"))
    if current is None:
        return True
    if other is None:
        return False
    return other<current


async def assigned_job_ids_for_user(session,user_id):
    """Job posts this candidate is allocated to (inbox assignment + manual upload)."""
    ids=set()
    if not user_id:
        return ids
    rows=await Inbox.get_candidate_profile(session=session,user_id=user_id,limit=1000,offset=0)
    records=rows if isinstance(rows,list) else ([rows] if rows else [])
    for rec in records:
        msg=getattr(rec,"messages",None)
        jid=getattr(msg,"assigned_job_post_id",None) if msg is not None else None
        if jid:
            ids.add(jid)
    manual=await Manual_UPLOAD_CANDIDATE.get_by_user_id(session,user_id)
    if manual and manual.job_post_id:
        ids.add(manual.job_post_id)
    return ids


async def job_id_for_application(session,inbox_id=None,manual_id=None):
    if inbox_id is not None:
        link=await Inbox.get_inbox_with_message(session,inbox_id)
        if link is None:
            return None,None
        msg=link.messages
        return (msg.assigned_job_post_id if msg is not None else None),link.user_id
    if manual_id is not None:
        manual=await Manual_UPLOAD_CANDIDATE.get_by_id(session,manual_id)
        if manual is None:
            return None,None
        return manual.job_post_id,manual.user_id
    return None,None


async def owned_job_ids_for_candidate_scope(session,current_user,created_by=False):
    """Job ids this user may see, or None when the list is unscoped.

    requisitions.configure (or hiring-manager portal) → jobs on their requisitions
    / assigned hiring_manager_id. Recruiter assignment on the job does not hide
    those candidates. candidates.manage or admin → None (all applications).
    Otherwise → current_recruiter_ids / current_recruiter_id when set, else created_by. Never role_id.
    created_by=True skips recruiter assignment and matches job_posts.created_by
    to the session user (ignored when the user is requisition-scoped).
    """
    if scopes_to_own_requisitions(current_user):
        return await JobPosts.ids_for_manager(session,current_user.get("id"))
    if sees_all_candidates(current_user):
        return None
    return await JobPosts.ids_for_creator(session,current_user.get("id"),created_by=created_by)


def _requested_job_post_ids(assigned_job_post_id):
    """Comma-separated or list of job post ids → UUID list. None = no filter."""
    if assigned_job_post_id is None:
        return None
    if isinstance(assigned_job_post_id,(list,tuple,set)):
        parts=list(assigned_job_post_id)
    else:
        text=str(assigned_job_post_id).strip()
        if not text:
            return None
        parts=[p.strip() for p in text.split(",") if p.strip()]
    ids=[]
    seen=set()
    for part in parts:
        uid=JobPosts._as_uuid(part)
        if uid is not None and uid not in seen:
            seen.add(uid)
            ids.append(uid)
    return ids


async def job_post_ids_for_candidate_list(session,current_user,assigned_job_post_id=None,created_by=False):
    """None = unscoped list. [] = nothing visible. Else UUID list for the query."""
    owned=await owned_job_ids_for_candidate_scope(session,current_user,created_by=created_by)
    requested=_requested_job_post_ids(assigned_job_post_id)
    if owned is None:
        return requested
    if requested is not None:
        owned_set=set(owned)
        return [jid for jid in requested if jid in owned_set]
    return list(owned)


def _scope_detail(current_user):
    if scopes_to_own_requisitions(current_user):
        return MANAGER_SCOPE_DETAIL
    return CREATOR_SCOPE_DETAIL


async def assert_manager_candidate_access(
    session,current_user,*,user_id=None,job_post_id=None,inbox_id=None,manual_id=None,created_by=False,
):
    """Row access: requisition-owned jobs, unscoped (manage/admin), or jobs this user created."""
    if sees_all_candidates(current_user) and not scopes_to_own_requisitions(current_user):
        return
    owned=await owned_job_ids_for_candidate_scope(session,current_user,created_by=created_by)
    owned=set(owned or [])
    detail=_scope_detail(current_user)
    if not owned:
        raise HTTPException(status_code=403,detail=detail)
    job_id=JobPosts._as_uuid(job_post_id) if job_post_id is not None else None
    uid=user_id
    if job_id is None and (inbox_id is not None or manual_id is not None):
        job_id,uid=await job_id_for_application(session,inbox_id=inbox_id,manual_id=manual_id)
    if job_id is None and uid is not None:
        candidate_jobs=await assigned_job_ids_for_user(session,uid)
        if candidate_jobs & owned:
            return
        raise HTTPException(status_code=403,detail=detail)
    if job_id is None or job_id not in owned:
        raise HTTPException(status_code=403,detail=detail)


async def parse_linkedin_url_from_cv(resume_text) -> str | None:
    """Employment-agent `linkedin_url` from CV text. None if absent, sentinel, invented, or the call fails."""
    text=(resume_text or "").strip()
    if not text:
        return None
    try:
        from employment_agent.execute_agent import run_employment_agent
        from employment_agent.plugins import parse_linkedin
        fields=await run_employment_agent(resume_text=text)
        url_fields=parse_linkedin({"linkedin_url":fields.get("linkedin_url") or ""},text)
        return url_fields.get("linkedin_url")
    except Exception:
        logger.exception("employment agent linkedin_url parse failed")
        return None


async def extract_bank_profile_from_cv(resume_text) -> dict:
    """Full employment-agent profile for a banked CV.

    parse_linkedin_url_from_cv runs this same agent and keeps only the URL,
    which left the bank with nothing to search on. Banking is the one ingest
    path with no job attached, so this extraction is the ONLY structured data
    the CV will ever have until someone scores it against a real job.

    Never raises: a failed extraction must still bank the file. Sentinels
    normalize to "" / None so "not stated" stays distinguishable from a value.
    """
    blank={
        "linkedin_url":None,"current_company":"","current_position":"",
        "education":"","candidate_phone":"","city":None,"skills":[],
        "years_experience":None,"candidate_name":"",
    }
    text=(resume_text or "").strip()
    if not text:
        return blank
    try:
        from employment_agent.decorators import _clean_years
        from employment_agent.execute_agent import run_employment_agent
        from employment_agent.plugins import parse_linkedin
        from employment_agent.prompt import CURRENT_TITLE,EDUCATION,NO_COMPANY
        fields=await run_employment_agent(resume_text=text)
    except Exception:
        logger.exception("employment agent bank profile extraction failed")
        return blank

    def unless_sentinel(key,sentinel):
        value=(fields.get(key) or "").strip()
        return "" if not value or value.lower()==sentinel.lower() else value

    try:
        url=parse_linkedin({"linkedin_url":fields.get("linkedin_url") or ""},text).get("linkedin_url")
    except Exception:
        url=None
    years=_clean_years(fields.get("years_experience"),text)
    return {
        "linkedin_url":url,
        "current_company":unless_sentinel("current_employment",NO_COMPANY),
        "current_position":unless_sentinel("current_title",CURRENT_TITLE),
        "education":unless_sentinel("education",EDUCATION),
        "candidate_phone":(fields.get("phone") or "").strip(),
        "city":(fields.get("city") or "").strip() or None,
        "skills":fields.get("skills") if isinstance(fields.get("skills"),list) else [],
        "years_experience":years,
        "candidate_name":(fields.get("candidate_name") or "").strip(),
    }

def _bank_row_matches(row,*,search=None,skills=None,min_years=None,band=None) -> bool:
    """Client-side facets for the merged bank list.

    The two populations live in different tables, so these cannot be one WHERE
    clause; they run over the merged page instead.
    """
    if search and str(search).strip():
        needle=str(search).strip().lower()
        haystack=" ".join(str(v or "") for v in (
            row.get("name"),row.get("email"),row.get("current_company"),
            row.get("current_position"),row.get("last_job_title"),
            " ".join(row.get("skills") or []),
        )).lower()
        if needle not in haystack:
            return False
    if skills:
        owned={s.lower() for s in (row.get("skills") or [])}
        # Every requested skill must be present: filters narrow, they do not widen.
        for wanted in skills:
            key=str(wanted).strip().lower()
            if key and not any(key in owned_skill for owned_skill in owned):
                return False
    if min_years is not None:
        years=row.get("years_experience")
        if years is None or years<int(min_years):
            return False
    if band and str(band).strip():
        wanted=str(band).strip()
        if wanted.lower()=="unscored":
            if row.get("ai_score") is not None:
                return False
        elif (row.get("recommendation") or "")!=wanted:
            return False
    return True


def _sort_bank_rows(rows,*,ranked=False) -> None:
    """Newest first, best score first, and best job-rank first when ranking.

    Three stable passes rather than one composite key: created_at is an ISO
    string and cannot be negated into a descending tuple slot.
    """
    rows.sort(key=lambda r:str(r.get("created_at") or ""),reverse=True)
    rows.sort(key=lambda r:r.get("ai_score") if isinstance(r.get("ai_score"),(int,float)) else -1,reverse=True)
    if ranked:
        rows.sort(key=lambda r:r.get("rank_score") if isinstance(r.get("rank_score"),(int,float)) else -1,reverse=True)


class FileRead:
    def __init__(self,session:AsyncSession,filename=None,file=None):
        self.session=session
        self.filename=filename
        self.file=file

    async def read_file(self,file=None,filename=None):
        try:
            reader = PdfReader(io.BytesIO(self.file))
            if reader.is_encrypted:
                raise HTTPException(400, "PDF is password protected")
            # extract_pdf_text, not page.extract_text() directly: some CVs come
            # out of pypdf one character per line, which reads fine to the LLM
            # but defeats every substring check downstream. See its docstring.
            text = normalize_spaced_text(extract_pdf_text(reader))
            # Icon-only LinkedIn buttons never appear in extract_text(); the
            # URL is on the annotation. Append so the employment agent can
            # return linkedin_url as its own parsed key.
            uris = extract_pdf_link_uris(reader)
            if uris:
                extra = "\n".join(uris)
                text = f"{text}\n\n{extra}".strip() if text else extra
            return {
                "filename": self.filename,
                "num_pages": len(reader.pages),
                "text": text,
            }
        except HTTPException:
            raise
        except Exception as e:
            raise HTTPException(400, str(e))
    async def injest_manual_upload(self):
        try:
            parsed=await self.read_file()
            text=(parsed.get("text") or "").strip()
            if not text:
                raise HTTPException(status_code=400,detail="No usable text could be extracted from the PDF")
            return parsed
        except HTTPException:
            raise
        except Exception as e:
            raise HTTPException(status_code=400,detail=str(e))

    async def save_manual_upload(self):
        """Deprecated — Manual CVs go to S3 via create_candidate (no local disk)."""
        raise HTTPException(
            status_code=410,
            detail="Local CV storage was removed; use create_candidate (S3 Manual/{id}/{user_id}/)",
        )

    @staticmethod
    def discard_upload(file_path):
        """Best-effort removal of a leftover local CV (legacy rows only)."""
        if not file_path:
            return
        if str(file_path).lower().startswith("http://") or str(file_path).lower().startswith("https://"):
            return
        try:
            Path(file_path).unlink(missing_ok=True)
        except OSError as e:
            logger.warning("could not remove orphaned upload %s: %s",file_path,e)

    async def ingest_upload(self,candidate_email=None,candidate_name=None,current_user=None):
        """Persist a recruiter-uploaded CV with full email-ingestion parity (S3)."""
        from inbox.file_decoder import extract_pdf_attachments
        from inbox.cv_tasks import match_uploaded_cv
        from inbox.plugins import attach_email_pdfs_to_s3
        from inbox.views import Email
        from s3.plugins import S3ServiceError,assert_pdf

        parsed=await self.read_file()
        text=parsed.get("text") or ""
        parsed_linkedin=await parse_linkedin_url_from_cv(text)
        detected,emails_found=extract_candidate_email(text)
        supplied=(candidate_email or "").strip().lower() or None
        email=supplied or detected
        email_source="recruiter" if supplied else ("cv" if detected else None)

        if not email:
            raise HTTPException(
                status_code=422,
                detail={
                    "error_code":"CANDIDATE_EMAIL_REQUIRED",
                    "filename":parsed.get("filename"),
                    "num_pages":parsed.get("num_pages"),
                    "emails_found":emails_found,
                    "text":text,
                },
            )

        filename=self.filename or "resume.pdf"
        try:
            assert_pdf(filename,"application/pdf")
        except S3ServiceError as e:
            raise HTTPException(status_code=e.status_code,detail=e.message) from e

        # In-memory only — no decoded_attachments write.
        import base64 as _b64
        pdfs=extract_pdf_attachments([{
            "name":filename,
            "contentBytes":_b64.b64encode(self.file).decode("ascii"),
        }])
        if not pdfs:
            raise HTTPException(status_code=400,detail="Only PDF resumes are allowed")

        now=datetime.now(timezone.utc).isoformat()
        email_data={
            "id":f"manual-cv:{uuid.uuid4()}",
            "subject":f"Manual CV upload — {filename}",
            "body":{"content":"","contentType":"text"},
            "hasAttachments":True,
            "attachments":[{"name":filename}],
            "from":{"emailAddress":{"address":email,"name":(candidate_name or "").strip()}},
            "toRecipients":[{"emailAddress":{"address":MANUAL_UPLOAD_TO_ADDRESS}}],
            "ccRecipients":[],
            "bccRecipients":[],
            "replyTo":[],
            "isRead":False,
            "sentDateTime":now,
            "receivedDateTime":now,
        }
        row,new_user_email=await Inbox_Messages.insert_email(
            self.session,email_data,file_path=None,
        )
        from inbox.views import Reapplied
        await Reapplied(session=self.session).sync_for_email(email)
        try:
            row=await attach_email_pdfs_to_s3(self.session,row,pdfs,created_new=True)
        except Exception as e:
            raise HTTPException(status_code=502,detail=f"S3 upload failed: {e}") from e

        if parsed_linkedin:
            user_id=await Inbox_Messages.get_linked_user_id(self.session,row.id)
            if user_id and await Users.set_linkedin_url_if_empty(
                self.session,user_id=user_id,url=parsed_linkedin,
            ):
                await self.session.commit()

        created_at=datetime.now(timezone.utc).isoformat()
        # Enqueue failure must not fail the upload: the CV row is already
        # persisted, and with the broker down (optional locally) the kiq call
        # raises a connection error. Suggestions just arrive later, or never.
        try:
            task=await match_uploaded_cv.kicker().with_labels(
                created_at=created_at,
                correlation_id=str(row.id),
                queue=CV_QUEUE_NAME,
            ).kiq(str(row.id),force=False)
        except Exception as e:
            logger.warning("cv match enqueue skipped for %s: %s",row.id,e)
            task=None

        account_setup=None
        if new_user_email:
            try:
                account_setup=await Email(session=self.session).send_account_setup(
                    [new_user_email]
                )
            except Exception as e:
                logger.warning("account setup mail failed for %s: %s",new_user_email,e)
                account_setup=[{"email":new_user_email,"sent":False}]

        user=await Users.get_user_by_email(self.session,email)
        if user:
            await HistoryRecorder(self.session).record(
                HistoryEvent.CANDIDATE_IMPORTED.value,
                current_user=current_user,user_id=user.id,
                entity_type="inbox_message",entity_id=row.id,
                to_value=email,description=f"CV uploaded: {filename}",commit=True,
            )
            await HistoryRecorder(self.session).record(
                HistoryEvent.DOCUMENT_UPLOADED.value,
                current_user=current_user,user_id=user.id,
                entity_type="document",entity_id=row.id,
                to_value=filename,commit=True,
            )

        return {
            "queued":True,
            "inbox_message_id":str(row.id),
            "task_id":task.task_id if task else None,
            "filename":parsed.get("filename"),
            "num_pages":parsed.get("num_pages"),
            "candidate_email":email,
            "email_source":email_source,
            "account_setup":account_setup,
            "text":text,
        }

    async def match_inbox_cv(self,inbox_message_id,current_user=None):
        from inbox.plugins import load_file_bytes
        from inbox.tasks import match_inbox_message

        row=await Inbox_Messages.get_inbox_message_by_id(self.session,inbox_message_id)
        if not row:
            raise HTTPException(status_code=404,detail="Message not found")
        if not row.attachment or not row.file_path:
            raise HTTPException(status_code=400,detail="your file isnt in the system")

        found_name=None
        for path_str in (p.strip() for p in row.file_path.split(",") if p.strip()):
            # S3 URL or local — presence of bytes (or a https URL we already stored) counts.
            if path_str.lower().startswith("http://") or path_str.lower().startswith("https://"):
                found_name=Path(path_str.replace("\\","/")).name or "resume.pdf"
                break
            raw=load_file_bytes(path_str)
            if raw is not None:
                found_name=Path(path_str.replace("\\","/")).name or "resume.pdf"
                break
        if found_name is None:
            raise HTTPException(status_code=400,detail="your file isnt in the system")

        created_at=datetime.now(timezone.utc).isoformat()
        task=await match_inbox_message.kicker().with_labels(
            created_at=created_at,
            correlation_id=str(row.id),
            queue="inbox",
        ).kiq(str(row.id),force=True)

        file_name=(row.file_name or "").split(",")[0].strip() or found_name
        await HistoryRecorder(self.session).record(
            HistoryEvent.CANDIDATE_IMPORTED.value,
            current_user=current_user,message_id=inbox_message_id,
            entity_type="inbox_message",entity_id=row.id,
            to_value=file_name,description=f"CV uploaded: {file_name}",commit=True,
        )
        return {
            "queued":True,
            "inbox_message_id":str(row.id),
            "file_name":file_name,
            "task_id":task.task_id,
        }
    # async def get_intention(self,input):
    #     try:
    #         get_subject=Inbox_Messages.candidate_x_inbox(self.session,self.candidate_id)
    #         get_file=


class CandidateScoring:
    """ATS scoring of CVs against one job post, persisted to the candidates table.

    Complements the agent's inbox-match flow: the agent suggests WHICH job a CV is
    for; this service scores HOW WELL a CV fits a chosen job (0-100 leaderboard).

    Deviation from the bulk-ats HTTP API (which rejects a whole batch with 415/413
    on a bad file): here per-file problems become persisted rows with
    status="failed" so one broken attachment never sinks the rest of the batch.
    Request-level errors (unknown job, too many files) still raise.
    """

    def __init__(self,session:AsyncSession):
        self.session=session

    async def _professional_summary_for_email(self,email,stored=None):
        """Stored summary wins; Users.professional_summary is the identity fallback."""
        text=(stored or "").strip() or None
        if text:
            return text
        cleaned=(email or "").strip().lower()
        if not cleaned:
            return None
        user=await Users.get_user_by_email(self.session,cleaned)
        if user is None:
            return None
        return (user.professional_summary or "").strip() or None

    async def _gate_source(self,source,jd):
        """Suitability gradient: summary vs job post, before PDF load / ATS.

        No summary, or the gate disabled: leave the source alone so the first
        score can write professional_summary. A mismatch sets precheck and
        drops bytes so later extract / score_batch never run.
        """
        from summary_gate.execute_agent import allow_ats
        from summary_gate.plugins import SUMMARY_NOT_SUITABLE

        if source.get("precheck") is not None:
            return
        summary=(source.get("professional_summary") or "").strip()
        if not summary:
            return
        if not await allow_ats(summary,jd):
            source["precheck"]=(SUMMARY_NOT_SUITABLE,"Professional summary does not fit this job.")
            source["data"]=None

    async def score_uploads(self,job_id,files,current_user):
        """files: list of (filename, bytes) pairs from the route handler."""
        settings=get_scoring_settings()
        if len(files)>settings.max_resumes_per_request:
            raise HTTPException(
                status_code=413,
                detail=f"At most {settings.max_resumes_per_request} resumes per request",
            )
        sources=[]
        for filename,data in files:
            source={
                "filename":filename or "resume.pdf",
                "data":data,
                "file_path":None,
                "precheck":None,
            }
            if not (filename or "").lower().endswith(".pdf"):
                source["precheck"]=(ErrorCode.UNSUPPORTED_FILE_TYPE,"Only PDF resumes are supported.")
            elif len(data)>settings.max_pdf_size_bytes:
                source["precheck"]=(ErrorCode.PAYLOAD_TOO_LARGE,"The file exceeds the size limit.")
            sources.append(source)
        return await self._score_and_persist(job_id,sources,"upload",current_user)

    async def score_inbox(self,job_id,message_ids,current_user,rescan_run_id=None):
        """Score PDF attachments of inbox messages (S3 URLs or legacy local paths)."""
        from inbox.plugins import load_file_bytes

        job=await JobPosts.get_job_post_by_id(self.session,job_id)
        if job is None or job.is_deleted:
            raise HTTPException(status_code=404,detail="Job post not found")
        jd=build_job_description(job)
        sources=[]
        for mid in message_ids:
            row=await Inbox_Messages.get_inbox_message_by_id(self.session,mid)
            if row is None:
                raise HTTPException(status_code=404,detail=f"Inbox message {mid} not found")
            if not row.file_path:
                continue
            names=[n.strip() for n in (row.file_name or "").split(",") if n.strip()]
            stored=(row.resume_text or "").strip()
            for idx,path_str in enumerate(p.strip() for p in row.file_path.split(",") if p.strip()):
                name=names[idx] if idx<len(names) else Path(path_str.replace("\\","/")).name or "resume.pdf"
                source={
                    "filename":name,
                    "data":None,
                    "file_path":path_str,
                    "inbox_message_id":row.id,
                    "candidate_email":(row.message_from or "").strip().lower() or None,
                    "professional_summary":None,
                    "precheck":None,
                }
                if stored:
                    source["resume_text"]=stored
                    source["stored_resume_text"]=stored
                source["professional_summary"]=await self._professional_summary_for_email(
                    source["candidate_email"],row.professional_summary,
                )
                await self._gate_source(source,jd)
                if source["precheck"] is not None:
                    sources.append(source)
                    continue
                if stored:
                    sources.append(source)
                    continue
                lower=name.lower()
                if lower.endswith(".doc") or lower.endswith(".docx"):
                    source["precheck"]=(ErrorCode.UNSUPPORTED_FILE_TYPE,"DOC/DOCX extraction is not supported yet.")
                elif not lower.endswith(".pdf") and ".pdf" not in path_str.lower():
                    source["precheck"]=(ErrorCode.UNSUPPORTED_FILE_TYPE,"Only PDF resumes are supported.")
                else:
                    try:
                        data=await asyncio.to_thread(load_file_bytes,path_str)
                    except Exception:
                        data=None
                    if data is None:
                        source["precheck"]=(FILE_NOT_FOUND,"The CV could not be loaded from S3 (check GetObject / public read).")
                    else:
                        source["data"]=data
                sources.append(source)
        if not sources:
            raise HTTPException(status_code=400,detail="No attachments found for the given message(s)")
        return await self._score_and_persist(job_id,sources,"inbox",current_user,rescan_run_id=rescan_run_id)

    async def score_bank(self,job_id,record_ids,current_user):
        """ATS-score CVs already sitting in the bank — tier 2, the paid step.

        The bank's automatic ranking is keyword overlap and says nothing about
        whether anyone is actually qualified. This is the real score, and it is
        deliberately explicit: a recruiter picks the handful worth paying for
        rather than the whole bank being scored against every new job.

        Bytes come from cv_bank_files (the CV is already stored, so there is
        nothing to re-upload), falling back to S3 for rows banked before the
        bytes were kept in the database.
        """
        from inbox.plugins import load_file_bytes
        from job.candidate.models import CvBankFiles

        settings=get_scoring_settings()
        ids=[str(r) for r in (record_ids or [])]
        if not ids:
            raise HTTPException(status_code=400,detail="Select at least one CV to score")
        if len(ids)>settings.max_resumes_per_request:
            raise HTTPException(
                status_code=413,
                detail=f"At most {settings.max_resumes_per_request} CVs per request",
            )
        job=await JobPosts.get_job_post_by_id(self.session,job_id)
        if job is None or job.is_deleted:
            raise HTTPException(status_code=404,detail="Job post not found")
        jd=build_job_description(job)
        sources=[]
        for record_id in ids:
            row=await Manual_UPLOAD_CANDIDATE.get_by_id(self.session,record_id)
            if row is None or row.apply_via!="cv_bank":
                raise HTTPException(status_code=404,detail=f"CV {record_id} not found in the bank")
            name=(row.file_name or "").strip() or "resume.pdf"
            source={
                "filename":name,
                "data":None,
                "file_path":(row.file_path or "").strip() or None,
                "candidate_email":(row.candidate_email or "").strip().lower() or None,
                "manual_upload_candidate_id":row.id,
                "professional_summary":None,
                "precheck":None,
            }
            stored=(row.full_text or "").strip()
            if stored:
                source["resume_text"]=stored
                source["stored_resume_text"]=stored
            source["professional_summary"]=await self._professional_summary_for_email(
                source["candidate_email"],row.professional_summary,
            )
            await self._gate_source(source,jd)
            if source["precheck"] is not None:
                sources.append(source)
                continue
            if stored:
                sources.append(source)
                continue
            file_row=await CvBankFiles.get(self.session,row.id)
            data=file_row.data if file_row and file_row.data else None
            if data is None and source["file_path"]:
                try:
                    data=await asyncio.to_thread(load_file_bytes,source["file_path"])
                except Exception:
                    data=None
            if data is None:
                source["precheck"]=(FILE_NOT_FOUND,"The stored CV could not be loaded.")
            elif len(data)>settings.max_pdf_size_bytes:
                source["precheck"]=(ErrorCode.PAYLOAD_TOO_LARGE,"The file exceeds the size limit.")
            else:
                source["data"]=data
            sources.append(source)
        return await self._score_and_persist(job_id,sources,"bank",current_user)

    async def fetch_candidates(self,job_id=None,limit=10,offset=0):
        # job_id omitted -> the whole pool across jobs (frontend Candidates/TalentPool).
        if job_id is not None:
            job=await JobPosts.get_job_post_by_id(self.session,job_id)
            if job is None or job.is_deleted:
                raise HTTPException(status_code=404,detail="Job post not found")
        rows,total=await Candidates.get_candidates_by_job(
            self.session,job_id,limit=limit,offset=offset,
        )
        data=[serialize_candidate(row) for row in rows]
        data=await CandidateView(session=self.session).attach_application_history(data)
        return data,total

    async def fetch_candidate_by_id(self,candidate_id):
        row=await Candidates.get_candidate_by_id(self.session,candidate_id)
        if row is None:
            raise HTTPException(status_code=404,detail="Candidate not found")
        data=serialize_candidate(row)
        return await CandidateView(session=self.session).attach_application_history(data)

    async def rerun_ats(self,current_user,user_id=None,inbox_message_id=None,manual_upload_candidate_id=None):
        """Re-score one candidate without re-reading the PDF.

        Prefer candidates.professional_summary as the scoring text. Fall back to
        inbox_messages.resume_text / manual_upload_candidate.full_text. Only parse
        the file when nothing is stored, then write that extract onto the inbox
        or manual row and stamp the FK on candidates.
        """
        from inbox.plugins import extract_resume_text

        message=None
        manual=None
        email=None
        if inbox_message_id:
            message=await Inbox_Messages.get_inbox_message_by_id(self.session,inbox_message_id)
            if message is None:
                raise HTTPException(status_code=404,detail="Inbox message not found")
        elif user_id:
            links=await Inbox.get_candidate_profile(session=self.session,user_id=user_id,limit=100,offset=0)
            records=links if isinstance(links,list) else ([links] if links else [])
            picked=None
            for link in records:
                msg=getattr(link,"messages",None)
                if msg is not None and msg.assigned_job_post_id:
                    picked=link
                    break
            if picked is None and records:
                picked=records[0]
            if picked is None:
                manual=await Manual_UPLOAD_CANDIDATE.get_by_user_id(self.session,user_id)
                if manual is None:
                    raise HTTPException(status_code=404,detail="Candidate not found")
            else:
                message=getattr(picked,"messages",None)
                user=getattr(picked,"user",None)
                email=(getattr(user,"email",None) or "").strip().lower() or None
        elif manual_upload_candidate_id:
            manual=await Manual_UPLOAD_CANDIDATE.get_by_id(self.session,manual_upload_candidate_id)
            if manual is None:
                raise HTTPException(status_code=404,detail="Candidate not found")
        else:
            raise HTTPException(status_code=400,detail="user_id, inbox_message_id or manual_upload_candidate_id is required")

        if message is None and manual is None:
            raise HTTPException(status_code=404,detail="Candidate not found")

        job_id=None
        filename="resume.pdf"
        file_path=None
        stored_extract=None
        inbox_summary=None
        inbox_mid=None
        manual_id=None
        existing_sha=None

        if message is not None:
            inbox_mid=message.id
            email=email or (message.message_from or "").strip().lower() or None
            suggested=list(message.suggested_job_post_ids or [])
            job_id=message.assigned_job_post_id or (suggested[0] if suggested else None)
            names=[n.strip() for n in (message.file_name or "").split(",") if n.strip()]
            filename=names[0] if names else "resume.pdf"
            paths=[p.strip() for p in (message.file_path or "").split(",") if p.strip()]
            file_path=paths[0] if paths else None
            stored_extract=(message.resume_text or "").strip() or None
            inbox_summary=(message.professional_summary or "").strip() or None
        else:
            manual_id=manual.id
            email=email or (manual.candidate_email or "").strip().lower() or None
            job_id=manual.job_post_id
            filename=(manual.file_name or "").strip() or "resume.pdf"
            file_path=(manual.file_path or "").strip() or None
            stored_extract=(manual.full_text or "").strip() or None
            inbox_summary=(manual.professional_summary or "").strip() or None

        if job_id is None:
            raise HTTPException(status_code=422,detail="Assign a job before running ATS")

        scored=await Candidates.get_completed_by_email_job(self.session,email,job_id) if email else None
        table_summary=(scored.professional_summary or "").strip() if scored else None
        if scored and scored.content_sha256:
            existing_sha=scored.content_sha256
        identity_summary=await self._professional_summary_for_email(email,inbox_summary)
        summary=(table_summary or identity_summary or "").strip() or None
        stored_extract=(stored_extract or "").strip() or None

        score_text=summary or stored_extract
        if not score_text:
            paths=[p.strip() for p in (file_path or "").split(",") if p.strip()] if file_path else []
            if not paths:
                raise HTTPException(status_code=400,detail="No professional summary or stored CV text to score")
            text,extract_err=await extract_resume_text(paths)
            score_text=(text or "").strip() or None
            if not score_text:
                raise HTTPException(status_code=400,detail=extract_err or "No professional summary or stored CV text to score")
            stored_extract=score_text

        if inbox_mid is not None and stored_extract:
            await Inbox_Messages.set_resume_text(self.session,inbox_mid,stored_extract)
        if manual_id is not None and stored_extract:
            await Manual_UPLOAD_CANDIDATE.set_full_text(self.session,manual_id,stored_extract)

        source={
            "filename":filename,
            "data":None,
            "file_path":file_path,
            "resume_text":score_text,
            "stored_resume_text":stored_extract,
            "sha256":existing_sha or (
                hashlib.sha256(stored_extract.encode("utf-8")).hexdigest() if stored_extract else None
            ),
            "inbox_message_id":inbox_mid,
            "manual_upload_candidate_id":manual_id,
            "candidate_email":email,
            "professional_summary":summary,
            "precheck":None,
        }
        kind="inbox" if inbox_mid is not None else "upload"
        return await self._score_and_persist(job_id, [source], kind, current_user, skip_gate=True)

    async def _score_and_persist(self,job_id,sources,source_kind,current_user,rescan_run_id=None,skip_gate=False):
        job=await JobPosts.get_job_post_by_id(self.session,job_id)
        if job is None or job.is_deleted:
            raise HTTPException(status_code=404,detail="Job post not found")
        settings=get_scoring_settings()
        jd=build_job_description(job)
        if len(jd)>settings.max_jd_chars:
            raise HTTPException(status_code=422,detail="The job post is too large to score against")
        from summary_gate.plugins import SUMMARY_NOT_SUITABLE
        for source in sources:
            if not source.get("professional_summary"):
                source["professional_summary"]=await self._professional_summary_for_email(
                    source.get("candidate_email"),
                )
            if not skip_gate:
                await self._gate_source(source,jd)
        fields_by_slot=await self._score_sources(sources,jd,settings)
        # Prefer the Manual S3 URL for this email+job when scoring from a raw upload
        # (Add Candidate scores right after create — same link as manual_upload_candidate).
        for slot,source in enumerate(sources):
            fields=fields_by_slot.get(slot) or {}
            email=(fields.get("candidate_email") or source.get("candidate_email") or "").strip().lower()
            if email and not source.get("manual_upload_candidate_id") and not fields.get("manual_upload_candidate_id"):
                manual=await Manual_UPLOAD_CANDIDATE.get_by_email_and_job(self.session,email,job.id)
                if manual:
                    source["manual_upload_candidate_id"]=manual.id
                    fields["manual_upload_candidate_id"]=manual.id
                    if not (fields.get("file_path") or source.get("file_path") or "").strip() and (manual.file_path or "").strip():
                        fields["file_path"]=manual.file_path.strip()
                        source["file_path"]=manual.file_path.strip()
        await self._persist_extracts(sources)
        common={
            "job_id":job.id,
            "source":source_kind,
            "created_by":uuid.UUID(str(current_user["id"])),
            "model":settings.openai_model,
        }
        rows=[]
        kept_sources=[]
        for slot in range(len(sources)):
            fields={**fields_by_slot[slot],**common}
            if fields.get("error_code")==SUMMARY_NOT_SUITABLE and rescan_run_id:
                continue
            email=(fields.get("candidate_email") or "").strip().lower()
            if email and not fields.get("linkedin_url"):
                user=await Users.get_user_by_email(self.session,email)
                if user and (user.linkedin_url or "").strip():
                    fields["linkedin_url"]=user.linkedin_url
            row=await Candidates.upsert_candidate(self.session,fields)
            if row.linkedin_url and row.candidate_email:
                if await Users.set_linkedin_url_if_empty(
                    self.session,email=row.candidate_email,url=row.linkedin_url,
                ):
                    await self.session.commit()
            rows.append(row)
            kept_sources.append(sources[slot])
        await self._sync_ats_results(source_kind,job,rows,kept_sources,current_user,rescan_run_id=rescan_run_id)
        rows.sort(key=lambda r:(0,-(r.match_score or 0)) if r.status=="completed" else (1,0))
        return [serialize_candidate(row) for row in rows]

    async def _score_sources(self,sources,jd,settings):
        # Slot-indexed: results merge back by position, never by filename —
        # inbox attachments can share a basename. Stored resume_text skips PDF parse.
        fields_by_slot={}
        extracted=[]
        for slot,source in enumerate(sources):
            source["safe_name"]=sanitize_filename(source.get("filename") or "resume.pdf")
            if source.get("precheck") is not None:
                code,message=source["precheck"]
                fields_by_slot[slot]=candidate_failed_fields(source,code,message)
                continue
            stored=(source.get("resume_text") or "").strip()
            if stored:
                text=normalize_spaced_text(stored)
                if len(text)>settings.max_resume_chars:
                    text=text[:settings.max_resume_chars]
                if not source.get("sha256"):
                    source["sha256"]=hashlib.sha256(text.encode("utf-8")).hexdigest()
                if not source.get("candidate_email"):
                    detected,_=extract_candidate_email(text)
                    source["candidate_email"]=detected
                resume=ExtractedResume(
                    filename=source["safe_name"],
                    candidate_id=str(source.get("inbox_message_id") or source.get("manual_upload_candidate_id") or uuid.uuid4()),
                    text=text,
                    page_count=1,
                    truncated=False,
                )
                extracted.append((slot,resume))
                continue
            data=source.get("data")
            source["sha256"]=hashlib.sha256(data).hexdigest() if data is not None else None
            try:
                resume=await asyncio.to_thread(
                    extract_resume,data,source["safe_name"],settings.max_resume_chars
                )
                resume=dataclasses.replace(resume,text=normalize_spaced_text(resume.text))
                if not source.get("candidate_email"):
                    detected,_=extract_candidate_email(resume.text)
                    source["candidate_email"]=detected
                if not source.get("stored_resume_text"):
                    source["stored_resume_text"]=resume.text
            except ATSError as exc:
                fields_by_slot[slot]=candidate_failed_fields(source,exc.error_code,exc.public_message)
                continue
            extracted.append((slot,resume))
        scored=await score_batch(
            [resume for _,resume in extracted],
            job_description=jd,
            scorer=get_scorer(),
            concurrency=settings.scoring_concurrency,
        )
        for (slot,_resume),result in zip(extracted,scored,strict=True):
            source=sources[slot]
            if isinstance(result,CompletedCandidate):
                fields_by_slot[slot]=candidate_completed_fields(source,result)
            else:
                fields_by_slot[slot]=candidate_failed_fields(source,result.error_code,result.error_message)
        return fields_by_slot

    async def _persist_extracts(self,sources):
        """Write PDF extract onto inbox_messages / manual_upload_candidate, never onto candidates."""
        for source in sources:
            extract=(source.get("stored_resume_text") or "").strip()
            if not extract:
                continue
            mid=source.get("inbox_message_id")
            if mid is not None:
                await Inbox_Messages.set_resume_text(self.session,mid,extract)
            uid=source.get("manual_upload_candidate_id")
            if uid is not None:
                await Manual_UPLOAD_CANDIDATE.set_full_text(self.session,uid,extract)

    async def _sync_ats_results(self,source_kind,job,rows,sources,current_user=None,rescan_run_id=None):
        run_id=self._as_rescan_run_id(rescan_run_id)
        if source_kind=="inbox":
            best={}
            for source,row in zip(sources,rows):
                mid=source.get("inbox_message_id")
                if row.status=="completed" and mid:
                    cur=best.get(mid)
                    if cur is None or (row.match_score or 0)>(cur.match_score or 0):
                        best[mid]=row
            for message_id,row in best.items():
                try:
                    await self._sync_inbox_ats(message_id,job,row,current_user=current_user,rescan_run_id=run_id)
                except Exception:
                    await self.session.rollback()
                    logger.exception("inbox ATS denorm failed for message %s",message_id)
            return
        for source,row in zip(sources,rows):
            if row.status!="completed":
                continue
            try:
                await self._sync_upload_ats(job,row,source,current_user=current_user,rescan_run_id=run_id)
            except Exception:
                await self.session.rollback()
                logger.exception("upload ATS history failed for candidate %s",row.id)

    @staticmethod
    def _as_rescan_run_id(value):
        if value in (None,""):
            return None
        try:
            return uuid.UUID(str(value))
        except (TypeError,ValueError):
            return None

    @staticmethod
    def _summary_text(row):
        return (getattr(row,"professional_summary",None) or "").strip() or None

    async def _stamp_identity_summary(self,row,summary):
        identity=await AtsResults.resolve_identity(self.session,row.candidate_email,row.id)
        if identity.get("user_id"):
            await Users.set_professional_summary(self.session,identity["user_id"],summary)
        return identity

    async def _append_rescan_summary(self,rescan_run_id,kind,record_id,job,summary):
        if not rescan_run_id:
            return
        await InboxRescanRun.append_summary(self.session,rescan_run_id,{
            "kind":kind,
            "record_id":str(record_id),
            "job_post_id":str(job.id),
            "professional_summary":summary,
        })

    async def _sync_inbox_ats(self,message_id,job,row,current_user=None,rescan_run_id=None):
        """Land a completed score on inbox_messages / inbox / ats_results.

        message_id is the scoring call's known inbox_messages PK, not a column
        read off the Candidates row.
        """
        msg=await Inbox_Messages.get_inbox_message_by_id(self.session,message_id)
        if msg is None:
            return
        band=CandidateView._recommendation(row.match_score) or ""
        summary=self._summary_text(row)
        denorm=True
        assigned=msg.assigned_job_post_id
        link=await Inbox.get_inbox_by_message_id(self.session,message_id)
        if assigned and str(assigned)!=str(job.id) and link is not None:
            existing=await AtsResults.get_for_inbox_job(self.session,link.id,assigned)
            denorm=existing is None
        if denorm:
            await Inbox_Messages.set_ats_score(self.session,message_id,row.match_score,band)
        await Inbox_Messages.set_professional_summary(self.session,message_id,summary)
        identity=await self._stamp_identity_summary(row,summary)
        if link is None:
            return
        old=await AtsResults.get_current_for_inbox(self.session,link.id)
        old_score=old.overall_score if old else None
        await AtsResults.insert_result(self.session,{
            "inbox_id":link.id,
            **identity,
            "job_post_id":job.id,
            "overall_score":float(row.match_score),
            "band":band,
            "model_name":row.model,
            "professional_summary":summary,
            "rescan_run_id":rescan_run_id,
            "is_current":True,
        })
        await self._append_rescan_summary(rescan_run_id,"email",message_id,job,summary)
        await HistoryRecorder(self.session).record(
            HistoryEvent.ATS_SCORED.value,
            current_user=current_user,user_id=link.user_id,inbox_id=link.id,
            entity_type="ats_result",entity_id=row.id,
            from_value=old_score,to_value=row.match_score,
            description=f"{band} against {job.title or 'job'}",commit=True,
        )

    async def _sync_upload_ats(self,job,row,source=None,current_user=None,rescan_run_id=None):
        """History row for an upload-sourced score — inbox_id stays NULL."""
        band=CandidateView._recommendation(row.match_score) or ""
        summary=self._summary_text(row)
        identity=await self._stamp_identity_summary(row,summary)
        mid=(source or {}).get("manual_upload_candidate_id")
        if not mid and row.candidate_email:
            manual=await Manual_UPLOAD_CANDIDATE.get_by_email_and_job(self.session,row.candidate_email,job.id)
            if manual:
                mid=manual.id
        if mid:
            await Manual_UPLOAD_CANDIDATE.set_professional_summary(self.session,mid,summary)
        old=None
        if identity.get("user_id"):
            old=await AtsResults.get_current_for_user(self.session,identity["user_id"],job.id)
        elif identity.get("candidate_id"):
            old=await AtsResults.get_current_for_candidate(self.session,identity["candidate_id"])
        old_score=old.overall_score if old else None
        await AtsResults.insert_result(self.session,{
            "inbox_id":None,
            **identity,
            "job_post_id":job.id,
            "overall_score":float(row.match_score),
            "band":band,
            "model_name":row.model,
            "professional_summary":summary,
            "rescan_run_id":rescan_run_id,
            "is_current":True,
        })
        await self._append_rescan_summary(rescan_run_id,"upload",mid or row.id,job,summary)
        await HistoryRecorder(self.session).record(
            HistoryEvent.ATS_SCORED.value,
            current_user=current_user,user_id=identity.get("user_id"),
            entity_type="ats_result",entity_id=row.id,
            from_value=old_score,to_value=row.match_score,
            description=f"{band} against {job.title or 'job'}",commit=True,
        )

class CandidateView:
    def __init__(self,session:AsyncSession):
        self.session=session

    @staticmethod
    def _recommendation(score):
        if score is None:
            return None
        # Same bands the frontend uses (Candidates.jsx / seed.js).
        return "Strong Match" if score>=82 else "Potential Match" if score>=65 else "Weak Match"

    @staticmethod
    def _score_from_message(record):
        """Inbox denorm on the already-loaded messages row — not a Candidates join."""
        msg=getattr(record,"messages",None)
        if msg is None or msg.ats_score is None:
            return None,None
        band=(msg.ats_band or "").strip() or None
        return msg.ats_score,band or CandidateView._recommendation(msg.ats_score)

    async def create_candidate(self,candidate_email=None,candidate_name=None,candidate_phone=None,job_post_id=None,current_company=None,current_position=None,platform=None,experience=None,status=None,referral_by=None,file_name=None,file_path=None,full_text=None,current_user=None,file_bytes=None,content_type=None):
        """Create manual_upload_candidate, then S3 upload under Manual/{id}/{user_id}/.

        Atomicity: if S3 fails after the row insert, the row is deleted (rolled back).
        PDF gate runs before any DB write when file_bytes is supplied.
        """
        from s3.plugins import S3,S3ServiceError,S3Source,assert_pdf

        row=None
        try:
            email=(candidate_email or "").strip().lower()
            if not email:
                raise HTTPException(status_code=422,detail="candidate_email is required")
            if not current_user:
                raise HTTPException(status_code=400,detail="created_by is required")

            original_name=(file_name or "").strip() or "resume.pdf"
            if file_bytes is not None:
                try:
                    original_name=assert_pdf(original_name,content_type)
                except S3ServiceError as e:
                    raise HTTPException(status_code=e.status_code,detail=e.message) from e
                if not file_bytes:
                    raise HTTPException(status_code=422,detail="file is empty")

            parsed_linkedin=await parse_linkedin_url_from_cv(full_text)
            phone_fields=parse_phone(
                {"phone":(candidate_phone or "").strip()},
                full_text or "",
            )
            phone=phone_fields.get("phone") or ""

            data={
                "candidate_email":email,
                "candidate_name":(candidate_name or "").strip(),
                "candidate_phone":phone,
                "job_post_id":job_post_id,
                "current_company":(current_company or "").strip(),
                "current_position":(current_position or "").strip(),
                "platform":(platform or "").strip(),
                "apply_via":"manual_upload",
                "experience":(experience or "").strip(),
                "status":(status or "").strip(),
                "referral_by":(referral_by or "").strip(),
                "file_name":original_name,
                # path filled after S3 succeeds; never leave a local orphan path here
                "file_path":(file_path or "").strip() if file_bytes is None else "",
                "full_text":full_text or "",
                "linkedin_url":parsed_linkedin,
                "created_by":current_user,
            }
            row=await Manual_UPLOAD_CANDIDATE.create_manual_upload_candidate(session=self.session,fields=data)
            from inbox.views import Reapplied
            await Reapplied(session=self.session).sync_for_email(email)

            if file_bytes is not None:
                try:
                    uploaded=S3().upload_for_record(
                        file_bytes,
                        original_name,
                        source=S3Source.MANUAL,
                        record_id=row.id,
                        owner_id=row.user_id,
                        content_type=content_type,
                    )
                except S3ServiceError as e:
                    await Manual_UPLOAD_CANDIDATE.delete_by_id(self.session,row.id)
                    row=None
                    raise HTTPException(status_code=e.status_code,detail=e.message) from e
                except Exception:
                    await Manual_UPLOAD_CANDIDATE.delete_by_id(self.session,row.id)
                    row=None
                    raise
                row=await Manual_UPLOAD_CANDIDATE.set_file_path(
                    self.session,row.id,uploaded["url"],file_name=uploaded.get("filename") or original_name,
                )
                # Same permanent URL on candidates rows for this email+job (if scored already).
                try:
                    await Candidates.sync_s3_file_path(
                        self.session,
                        email=row.candidate_email,
                        job_id=row.job_post_id,
                        file_path=row.file_path,
                    )
                except Exception:
                    logger.exception("candidates.file_path sync failed for manual %s",row.id)

            await HistoryRecorder(self.session).record(
                HistoryEvent.CANDIDATE_CREATED.value,
                actor_id=current_user,user_id=row.user_id,
                manual_upload_candidate_id=row.id,
                entity_type="manual_upload_candidate",entity_id=row.id,
                to_value=row.candidate_email,
                description=(row.platform or "").strip() or "manual_upload",commit=True,
            )
            if (row.file_name or "").strip() and (row.file_path or "").strip():
                await HistoryRecorder(self.session).record(
                    HistoryEvent.DOCUMENT_UPLOADED.value,
                    actor_id=current_user,user_id=row.user_id,
                    manual_upload_candidate_id=row.id,
                    entity_type="document",entity_id=row.id,
                    to_value=row.file_name,commit=True,
                )
            return serialize_manual_upload_candidate(row)
        except HTTPException:
            raise
        except Exception as e:
            if row is not None:
                try:
                    await Manual_UPLOAD_CANDIDATE.delete_by_id(self.session,row.id)
                except Exception:
                    logger.exception("manual candidate rollback failed for %s",getattr(row,"id",None))
            raise HTTPException(status_code=500,detail=str(e))

    async def list_manager_candidates(self,current_user,limit=50,offset=0):
        """Candidates allocated to jobs this manager owns (requisition → job post)."""
        job_ids=await JobPosts.ids_for_manager(self.session,current_user.get("id"))
        if not job_ids:
            return [],0
        inbox_rows=await Inbox.get_all(self.session,job_post_ids=job_ids)
        manual_rows=await Manual_UPLOAD_CANDIDATE.get_all(self.session,job_post_ids=job_ids)
        merged=[]
        seen=set()
        for row in inbox_rows:
            uid=row.get("user_id")
            payload=serialize_manager_candidate(row,source="inbox")
            if uid and uid not in seen:
                seen.add(uid)
                merged.append(payload)
            elif not uid:
                merged.append(payload)
        for row in manual_rows:
            uid=row.get("user_id")
            if uid and uid in seen:
                continue
            payload=serialize_manager_candidate(row,source="manual")
            if uid:
                seen.add(uid)
            merged.append(payload)
        merged.sort(key=lambda r: r.get("created_at") or "",reverse=True)
        total=len(merged)
        start=max(0,int(offset or 0))
        cap=max(1,int(limit or 50))
        page=merged[start:start+cap]
        return await self.attach_application_history(page),total

    async def get_candidate(self,user_id=None,limit=10,offset=0,search=None,current_user=None,assigned_job_post_id=None,created_by=False,assignment=None):
        try:
            if not user_id and is_hiring_manager(current_user):
                raise HTTPException(status_code=403,detail=MANAGER_SCOPE_DETAIL)
            if user_id:
                await assert_manager_candidate_access(
                    self.session,current_user,user_id=user_id,created_by=created_by,
                )
                return await self._get_candidate_detail(
                    user_id,limit=limit,offset=offset,search=search,
                    current_user=current_user,created_by=created_by,
                )
            list_job_ids=await job_post_ids_for_candidate_list(
                self.session,current_user,assigned_job_post_id=assigned_job_post_id,created_by=created_by,
            )
            if list_job_ids is not None and not list_job_ids:
                return []
            return await self._list_candidates(
                limit=limit,offset=offset,search=search,job_post_ids=list_job_ids,assignment=assignment,
            )
        except HTTPException:
            raise
        except Exception as e:
            raise HTTPException(status_code=500,detail=str(e))

    async def _get_candidate_detail(self,user_id,limit=10,offset=0,search=None,current_user=None,created_by=False):
        rows=await Inbox.get_candidate_profile(
            session=self.session,user_id=user_id,limit=limit,offset=offset,search=search,
        )
        records=rows if isinstance(rows,list) else ([rows] if rows else [])
        owned=await owned_job_ids_for_candidate_scope(self.session,current_user,created_by=created_by)
        if records and owned is not None:
            owned_set=set(owned)
            kept=[]
            for rec in records:
                msg=getattr(rec,"messages",None)
                jid=getattr(msg,"assigned_job_post_id",None) if msg is not None else None
                if jid and jid in owned_set:
                    kept.append(rec)
            if kept:
                return await self.attach_application_history(await self.attach_profile_detail(kept))
            records=[]
        if records:
            return await self.attach_application_history(await self.attach_profile_detail(rows))
        return await self._get_manual_detail(user_id,owned,current_user)

    async def _get_manual_detail(self,user_id,owned,current_user):
        from inbox.plugins import get_ats_score_for_manual_user

        manual=await Manual_UPLOAD_CANDIDATE.get_by_user_id(self.session,user_id)
        if not manual:
            return []
        if owned is not None:
            owned_set=set(owned)
            if not manual.job_post_id or manual.job_post_id not in owned_set:
                raise HTTPException(status_code=403,detail=_scope_detail(current_user))
        user=await Users.get_user_by_id(self.session,user_id)
        job_post=None
        if manual.job_post_id:
            job_post=await JobPosts.get_job_post_by_id(self.session,str(manual.job_post_id))
        payload=serialize_manual_candidate_profile(manual,user,job_post)
        score=await get_ats_score_for_manual_user(self.session,user_id,manual.job_post_id)
        if score:
            payload["ai_score"]=score["overall_score"]
            payload["recommendation"]=self._recommendation(score["overall_score"])
            payload["scored_at"]=score["computed_at"]
            payload["candidate_id"]=score.get("candidate_id")
            if score.get("user_id") and not payload.get("user_id"):
                payload["user_id"]=score["user_id"]
        if score and score.get("job_post_id"):
            payload["scored_job_post_id"]=score["job_post_id"]
        return await self.attach_application_history(payload)

    async def _list_candidates(self,limit=10,offset=0,search=None,job_post_ids=None,assignment=None):
        rows=await Inbox.get_candidate_profile(
            session=self.session,limit=limit,offset=offset,search=search,job_post_ids=job_post_ids,assignment=assignment,
        )
        inbox_payloads=await self.attach_job_posts(rows)
        if not isinstance(inbox_payloads,list):
            inbox_payloads=[inbox_payloads] if inbox_payloads else []
        seen={p.get("user_id") for p in inbox_payloads if p.get("user_id")}
        seen_emails={(p.get("email") or "").strip().lower() for p in inbox_payloads if (p.get("email") or "").strip()}
        manual_payloads=await self._list_manual_payloads(
            limit=limit,search=search,job_post_ids=job_post_ids,seen=seen,seen_emails=seen_emails,assignment=assignment,
        )
        form_payloads=await self._list_form_payloads(
            limit=limit,search=search,job_post_ids=job_post_ids,seen_emails=seen_emails,assignment=assignment,
        )
        return await self.attach_application_history(inbox_payloads+manual_payloads+form_payloads)

    async def _list_manual_payloads(self,limit,search,job_post_ids,seen,seen_emails,assignment=None):
        if assignment == "unassigned":
            return []
        rows=await Manual_UPLOAD_CANDIDATE.list_for_talent_pool(
            self.session,limit=limit,offset=0,search=search,job_post_ids=job_post_ids,
        )
        payloads=[]
        for row in rows:
            uid=str(row.user_id) if row.user_id else None
            if uid and uid in seen:
                continue
            user=await Users.get_user_by_id(self.session,row.user_id) if row.user_id else None
            job_post=None
            if row.job_post_id:
                job_post=await JobPosts.get_job_post_by_id(self.session,str(row.job_post_id))
            payload=serialize_manual_candidate_list(serialize_manual_candidate_profile(row,user,job_post))
            payloads.append(payload)
            if uid:
                seen.add(uid)
            email=(payload.get("email") or "").strip().lower()
            if email:
                seen_emails.add(email)
        await self._attach_user_ats(payloads)
        return payloads

    async def _list_form_payloads(self,limit,search,job_post_ids,seen_emails,assignment=None):
        rows=await FormData.list_for_talent_pool(
            self.session,limit=limit,offset=0,search=search,job_post_ids=job_post_ids,assignment=assignment,
        )
        payloads=[]
        for rec in rows:
            payload=serialize_form_candidate_list(rec)
            email=(payload.get("email") or "").strip().lower()
            if email and email in seen_emails:
                continue
            payloads.append(payload)
            if email:
                seen_emails.add(email)
        payloads=await self._hydrate_list_jobs(payloads)
        await self._attach_form_ats(payloads)
        return payloads

    async def _attach_user_ats(self,payloads):
        from inbox.plugins import get_ats_scores_for_users
        owners=[p.get("user_id") for p in payloads if p.get("user_id")]
        ats=await get_ats_scores_for_users(self.session,owners)
        for payload in payloads:
            row=ats.get(str(payload.get("user_id") or ""))
            if row and row.get("overall_score") is not None:
                payload["ai_score"]=row["overall_score"]
                payload["recommendation"]=row.get("band") or self._recommendation(row["overall_score"])

    async def count_candidates(self,user_id=None,search=None,current_user=None,assigned_job_post_id=None,created_by=False,assignment=None):
        try:
            job_post_ids=None
            if not user_id:
                job_post_ids=await job_post_ids_for_candidate_list(
                    self.session,current_user,assigned_job_post_id=assigned_job_post_id,created_by=created_by,
                )
                if job_post_ids is not None and not job_post_ids:
                    return 0
            inbox_n=await Inbox.count_candidate_profiles(
                session=self.session,user_id=user_id,search=search,job_post_ids=job_post_ids,assignment=assignment,
            )
            if user_id:
                return inbox_n
            manual_n=0 if assignment == "unassigned" else await Manual_UPLOAD_CANDIDATE.count_for_talent_pool(
                self.session,search=search,job_post_ids=job_post_ids,
            )
            form_n=await FormData.count_for_talent_pool(
                self.session,search=search,job_post_ids=job_post_ids,assignment=assignment,
            )
            return inbox_n+manual_n+form_n
        except HTTPException:
            raise
        except Exception as e:
            raise HTTPException(status_code=500,detail=str(e))

    async def update_candidate(self,user_id,payload,current_user=None,created_by=False):
        try:
            if not user_id:
                raise HTTPException(status_code=400,detail="user_id is required")
            await assert_manager_candidate_access(self.session,current_user,user_id=user_id,created_by=created_by)
            fields={k:v for k,v in (payload or {}).items() if k in ("favorite","rating") and v is not None}
            if not fields:
                raise HTTPException(status_code=400,detail="favorite or rating is required")
            links=await Inbox.get_candidate_profile(session=self.session,user_id=user_id,limit=100,offset=0)
            records=links if isinstance(links,list) else ([links] if links else [])
            if not records:
                raise HTTPException(status_code=404,detail="Candidate not found")
            first=records[0]
            old_favorite=getattr(first,"favorite",None)
            old_rating=getattr(first,"rating",None)
            for link in records:
                await Inbox.update_inbox(self.session,link.id,fields)
            if "favorite" in fields and fields["favorite"]!=old_favorite:
                await HistoryRecorder(self.session).record(
                    HistoryEvent.FAVORITE_CHANGED.value,
                    current_user=current_user,user_id=user_id,inbox_id=first.id,
                    entity_type="candidate",entity_id=user_id,
                    from_value=old_favorite,to_value=fields["favorite"],commit=True,
                )
            if "rating" in fields and fields["rating"]!=old_rating:
                await HistoryRecorder(self.session).record(
                    HistoryEvent.RATING_CHANGED.value,
                    current_user=current_user,user_id=user_id,inbox_id=first.id,
                    entity_type="candidate",entity_id=user_id,
                    from_value=old_rating,to_value=fields["rating"],commit=True,
                )
            refreshed=await Inbox.get_candidate_profile(session=self.session,user_id=user_id,limit=100,offset=0)
            return await self.attach_application_history(await self.attach_profile_detail(refreshed))
        except HTTPException:
            raise
        except Exception as e:
            raise HTTPException(status_code=500,detail=str(e))

    @staticmethod
    def _attach_job_post(data,payload,*,as_assigned=False):
        """Merge one serialized job post onto a candidate payload.

        Pure, so the per-id path and the batched one below cannot drift. A missing
        post attaches nothing, matching the old early return.
        """
        if not isinstance(data,dict) or not payload:
            return payload
        if as_assigned:
            data["assigned_job_post"]=payload
            if payload.get("created_by_name"):
                data["recruiter"]=payload.get("created_by_name")
                data["recruiter_id"]=payload.get("created_by")
            if payload.get("title"):
                data["job_title"]=payload.get("title")
        else:
            data.setdefault("job_posts",[]).append(payload)
            if data.get("recruiter") is None and payload.get("created_by_name"):
                data["recruiter"]=payload.get("created_by_name")
                data["recruiter_id"]=payload.get("created_by")
            if data.get("job_title") is None and payload.get("title"):
                data["job_title"]=payload.get("title")
        return payload

    async def _hydrate_list_jobs(self,payloads):
        """Attach assigned + suggested job posts onto already-serialized list rows."""
        wanted=[]
        for payload in payloads:
            if payload.get("assigned_job_post_id"):
                wanted.append(payload["assigned_job_post_id"])
            wanted.extend(payload.get("suggested_job_post_ids") or [])
        posts=await self._job_posts_by_id(wanted)
        for payload in payloads:
            assigned_id=payload.get("assigned_job_post_id")
            if assigned_id:
                post=posts.get(str(assigned_id))
                self._attach_job_post(payload,dict(post) if post else None,as_assigned=True)
            for job_id in payload.get("suggested_job_post_ids") or []:
                if assigned_id and str(job_id)==str(assigned_id):
                    continue
                post=posts.get(str(job_id))
                self._attach_job_post(payload,dict(post) if post else None)
        return payloads

    async def _attach_form_ats(self,payloads):
        grouped=await AtsResults.get_current_for_forms(
            self.session,[p.get("form_data_id") for p in payloads],
        )
        for payload in payloads:
            fid=payload.get("form_data_id")
            try:
                key=uuid.UUID(str(fid)) if fid else None
            except (TypeError,ValueError):
                key=None
            rows=grouped.get(key) or [] if key is not None else []
            assigned=payload.get("assigned_job_post_id")
            chosen=None
            if assigned:
                chosen=next((r for r in rows if str(r.job_post_id)==str(assigned)),None)
            if chosen is None and rows:
                chosen=rows[0]
            if chosen is not None and chosen.overall_score is not None:
                payload["ai_score"]=chosen.overall_score
                payload["recommendation"]=chosen.band or self._recommendation(chosen.overall_score)

    async def _job_posts_by_id(self,ids):
        """Serialized job posts keyed by id — one query for a whole page of rows.

        active_only=False mirrors get_job_post_by_id, which filters neither flag:
        an application assigned to a closed post must keep its title.
        """
        wanted=[]
        seen=set()
        for raw in ids or []:
            key=str(raw) if raw else None
            if not key or key in seen:
                continue
            seen.add(key)
            wanted.append(key)
        if not wanted:
            return {}
        rows=await JobPosts.get_by_ids(self.session,wanted,active_only=False)
        return {str(r.id):serialize_job_post(r) for r in rows}

    async def get_job_post_by_id(self,record_id,data=None,*,as_assigned=False):
        """Load full job_posts row and optionally append it onto a candidate payload."""
        try:
            job_post_data=await JobPosts.get_job_post_by_id(session=self.session,record_id=record_id)
            if not job_post_data:
                return None
            return self._attach_job_post(data,serialize_job_post(job_post_data),as_assigned=as_assigned)
        except Exception as e:
            raise HTTPException(status_code=500,detail=str(e))

    async def attach_job_posts(self,data):
        """Normalize list/single, serialize each record, attach full job_posts rows.

        Three batched queries per page, not two per row — a 200-row pipeline board
        issued 600+ sequential job_posts round trips before.
        """
        from inbox.plugins import get_ats_scores_for_users

        single=not isinstance(data,list)
        records=[data] if single else list(data or [])

        payloads=[]
        wanted=[]
        owners=[]
        for record in records:
            payload=serialize_candidate_profile(record)
            payload["job_posts"]=[]
            payload["assigned_job_post"]=None
            score,band=self._score_from_message(record)
            if score is not None:
                payload["ai_score"]=score
                payload["recommendation"]=band
            payloads.append(payload)
            if payload.get("user_id"):
                owners.append(payload["user_id"])
            if payload.get("assigned_job_post_id"):
                wanted.append(payload["assigned_job_post_id"])
            wanted.extend(payload.get("suggested_job_post_ids") or [])

        # ats_results is the source the inbox denorm above is copied FROM, so a
        # live score wins over it. A candidate with neither keeps ai_score None —
        # the list rows must not invent a number the scoring engine never produced.
        ats=await get_ats_scores_for_users(self.session,owners)
        for payload in payloads:
            row=ats.get(str(payload.get("user_id") or ""))
            if row and row.get("overall_score") is not None:
                payload["ai_score"]=row["overall_score"]
                payload["recommendation"]=row.get("band") or self._recommendation(row["overall_score"])

        posts=await self._job_posts_by_id(wanted)
        # Copy per attach: two candidates on the same post held independent dicts
        # back when every row re-serialized its own.
        for payload in payloads:
            assigned_id=payload.get("assigned_job_post_id")
            if assigned_id:
                post=posts.get(str(assigned_id))
                self._attach_job_post(payload,dict(post) if post else None,as_assigned=True)
            for job_id in payload.get("suggested_job_post_ids") or []:
                post=posts.get(str(job_id))
                self._attach_job_post(payload,dict(post) if post else None)
        return payloads[0] if single else payloads

    async def attach_profile_detail(self,data):
        """Detail mode: flatten child collections across every Inbox row for the candidate."""
        single=not isinstance(data,list)
        records=[data] if single else list(data or [])
        if not records:
            return {} if single else []

        interviews=[]
        activity=[]
        feedback=[]
        documents=[]
        job_posts=[]
        assigned_job_post=None
        base=None
        user_id=None
        favorite=None
        rating=None
        for record in records:
            payload=serialize_candidate_profile(record,detail=True)
            if base is None:
                base=payload
                user_id=payload.get("user_id")
                favorite=payload.get("favorite")
                rating=payload.get("rating")
            interviews.extend(payload.get("interviews") or [])
            activity.extend(payload.get("activity") or [])
            feedback.extend(payload.get("feedback") or [])
            documents.extend(payload.get("documents") or [])
            if payload.get("assigned_job_post_id") and assigned_job_post is None:
                await self.get_job_post_by_id(
                    record_id=payload.get("assigned_job_post_id"),
                    data=payload,
                    as_assigned=True,
                )
                assigned_job_post=payload.get("assigned_job_post")
                if base.get("recruiter") is None and payload.get("recruiter"):
                    base["recruiter"]=payload.get("recruiter")
                    base["recruiter_id"]=payload.get("recruiter_id")
                if base.get("job_title") is None and payload.get("job_title"):
                    base["job_title"]=payload.get("job_title")
            for job_id in payload.get("suggested_job_post_ids") or []:
                await self.get_job_post_by_id(record_id=job_id,data=payload)
            for jp in payload.get("job_posts") or []:
                if not any(x.get("id")==jp.get("id") for x in job_posts):
                    job_posts.append(jp)
            if base.get("recruiter") is None and payload.get("recruiter"):
                base["recruiter"]=payload.get("recruiter")
                base["recruiter_id"]=payload.get("recruiter_id")
            if base.get("job_title") is None and payload.get("job_title"):
                base["job_title"]=payload.get("job_title")

        notes=[]
        uid=Notes._as_uuid(user_id) if user_id else None
        if uid is not None:
            notes=[serialize_note(r) for r in await Notes.get_notes_by_user(self.session,uid)]

        # ATS score from inbox denorm / ats_results via Inbox.ats_id — never from
        # a Candidates join on message id. Keywords live on the scored Candidates
        # row: candidate_id when set, else email+job for the matched-user path.
        ats_ids=[r.ats_id for r in records if getattr(r,"ats_id",None)]
        ats_rows=await AtsResults.get_by_ids(self.session,ats_ids) if ats_ids else []
        assigned_uid=AtsResults._as_uuid(base.get("assigned_job_post_id")) if base.get("assigned_job_post_id") else None
        chosen=None
        if assigned_uid is not None:
            chosen=next((a for a in ats_rows if a.job_post_id==assigned_uid),None)
        if chosen is None and ats_rows:
            chosen=max(ats_rows,key=lambda a:a.computed_at)
        if chosen is not None:
            base["ai_score"]=chosen.overall_score
            base["recommendation"]=chosen.band or self._recommendation(chosen.overall_score)
            base["scored_job_post_id"]=str(chosen.job_post_id) if chosen.job_post_id else None
            base["scored_at"]=chosen.computed_at.isoformat() if chosen.computed_at else None
            base["candidate_id"]=str(chosen.candidate_id) if chosen.candidate_id else None
            if chosen.user_id and not base.get("user_id"):
                base["user_id"]=str(chosen.user_id)
            if chosen.professional_summary and not base.get("professional_summary"):
                base["professional_summary"]=chosen.professional_summary
            scored=None
            if chosen.candidate_id:
                scored=await Candidates.get_candidate_by_id(self.session,str(chosen.candidate_id))
            elif chosen.user_id:
                owner=await Users.get_user_by_id(self.session,str(chosen.user_id))
                if owner is not None:
                    scored=await Candidates.get_completed_by_email_job(self.session,owner.email,chosen.job_post_id)
            if scored is not None:
                base["matched_keywords"]=list(scored.matched_keywords or [])
                base["missing_keywords"]=list(scored.missing_keywords or [])
                base["summary_critique"]=scored.summary_critique
                if scored.professional_summary and not base.get("professional_summary"):
                    base["professional_summary"]=scored.professional_summary
        else:
            for record in records:
                score,band=self._score_from_message(record)
                if score is not None:
                    base["ai_score"]=score
                    base["recommendation"]=band
                    break

        activity.sort(key=lambda r:(r.get("activity_date") or ""),reverse=True)
        base["favorite"]=favorite
        base["rating"]=rating
        base["interviews"]=interviews
        base["activity"]=activity
        base["feedback"]=feedback
        base["documents"]=documents
        base["notes"]=notes
        base["job_posts"]=job_posts or base.get("job_posts") or []
        base["assigned_job_post"]=assigned_job_post
        if assigned_job_post:
            base["assigned_job_post_id"]=assigned_job_post.get("id")
            if assigned_job_post.get("created_by_name"):
                base["recruiter"]=assigned_job_post.get("created_by_name")
                base["recruiter_id"]=assigned_job_post.get("created_by")
            if assigned_job_post.get("title"):
                base["job_title"]=assigned_job_post.get("title")
        return base

    async def download_document(self,inbox_id=None,manual_upload_candidate_id=None,index=0):
        has_inbox=inbox_id is not None and str(inbox_id).strip()!=""
        has_manual=manual_upload_candidate_id is not None and str(manual_upload_candidate_id).strip()!=""
        if has_inbox==has_manual:
            raise HTTPException(status_code=404,detail="Not found")
        try:
            index=int(index or 0)
        except (TypeError,ValueError):
            raise HTTPException(status_code=404,detail="Not found")
        if index<0:
            raise HTTPException(status_code=404,detail="Not found")

        docs=[]
        if has_inbox:
            link=await Inbox.get_inbox_with_message(self.session,inbox_id)
            if not link or not link.messages:
                raise HTTPException(status_code=404,detail="Not found")
            docs=documents_from_message(link.messages.file_name,link.messages.file_path)
        else:
            row=await Manual_UPLOAD_CANDIDATE.get_by_id(self.session,manual_upload_candidate_id)
            if not row:
                raise HTTPException(status_code=404,detail="Not found")
            docs=documents_from_message(row.file_name,row.file_path)

        if index>=len(docs):
            raise HTTPException(status_code=404,detail="Not found")
        entry=docs[index]
        path=contained_download_path(entry.get("path"))
        if path is None:
            raise HTTPException(status_code=404,detail="Not found")
        name=(entry.get("name") or path.name).strip() or path.name
        return path,name

    # The bank holds two populations that live in different tables, so paging
    # cannot happen in SQL. Both are read up to this cap, merged, filtered, and
    # paged in Python. A bank larger than this needs a materialized view, not a
    # bigger number.
    BANK_SCAN_CAP=2000

    async def list_bank(self,*,source=None,search=None,skills=None,min_years=None,
                        band=None,job_post_id=None,limit=50,offset=0):
        """The unified CV Bank: speculative uploads plus scored rejections.

        job_post_id does not filter — it is only used by /cv-bank/suggestions
        to attach rank_score. The CV Bank screen itself no longer ranks.
        """
        from inbox.models import Inbox
        from job.candidate.serializers import serialize_bank_candidate,serialize_bank_silver_medalist

        rows=[]
        if source in (None,"","speculative"):
            bank_rows,_=await Manual_UPLOAD_CANDIDATE.list_bank(
                self.session,limit=self.BANK_SCAN_CAP,offset=0,
            )
            rows.extend(serialize_bank_candidate(r) for r in bank_rows)
        if source in (None,"","silver_medalist"):
            medalists=await Inbox.list_silver_medalists(
                self.session,min_score=self._silver_floor(),limit=self.BANK_SCAN_CAP,
            )
            rows.extend(serialize_bank_silver_medalist(r) for r in medalists)

        rows=await self._attach_bank_ats(rows)
        rows=await self._hydrate_bank_job_titles(rows)
        if job_post_id:
            rows=await self._attach_rank_scores(rows,job_post_id)

        rows=[r for r in rows if _bank_row_matches(r,search=search,skills=skills,
                                                   min_years=min_years,band=band)]
        _sort_bank_rows(rows,ranked=bool(job_post_id))
        total=len(rows)
        return rows[offset:offset+limit],total

    @staticmethod
    def _silver_floor():
        import os

        return int(os.getenv("CV_BANK_SILVER_FLOOR","60"))

    async def _attach_bank_ats(self,rows):
        """Join the latest completed ATS score onto speculative bank rows.

        serialize_bank_candidate leaves ai_score None; scores live on
        candidates (and ats_results) keyed by email, written by score_bank.
        Silver medalists already carry the inbox denorm score.
        """
        emails=[]
        for row in rows:
            if row.get("bank_source")!="speculative":
                continue
            email=(row.get("email") or "").strip().lower()
            if email:
                emails.append(email)
        if not emails:
            return rows
        latest=await Candidates.latest_completed_by_emails(self.session,emails)
        for row in rows:
            if row.get("bank_source")!="speculative":
                continue
            hit=latest.get((row.get("email") or "").strip().lower())
            if not hit:
                continue
            row["ai_score"]=hit.get("match_score")
            row["recommendation"]=self._recommendation(hit.get("match_score"))
            row["scored_job_post_id"]=hit.get("job_post_id")
            row["scored_job_title"]=hit.get("job_title")
        return rows

    async def _hydrate_bank_job_titles(self,rows):
        """Resolve assigned / scored / suggested job ids to titles in one fetch."""
        ids=[]
        seen=set()
        def add(jid):
            key=str(jid) if jid not in (None,"") else ""
            if key and key not in seen:
                seen.add(key)
                ids.append(key)
        for row in rows:
            add(row.get("assigned_job_post_id"))
            add(row.get("scored_job_post_id"))
            for jid in row.get("suggested_job_post_ids") or []:
                add(jid)
        titles={}
        if ids:
            posts=await JobPosts.titles_by_ids(self.session,ids,active_only=False)
            titles={str(p.id):p.title for p in posts}
        for row in rows:
            assigned=str(row.get("assigned_job_post_id") or "")
            scored=str(row.get("scored_job_post_id") or "")
            if not row.get("assigned_job_title"):
                row["assigned_job_title"]=titles.get(assigned)
            if not row.get("scored_job_title"):
                row["scored_job_title"]=titles.get(scored)
            suggested=[]
            for jid in row.get("suggested_job_post_ids") or []:
                title=titles.get(str(jid))
                if title:
                    suggested.append({"id":str(jid),"title":title})
            row["suggested_jobs"]=suggested
        return rows

    async def _attach_rank_scores(self,rows,job_post_id):
        """Fill rank_score from the stored tier-1 ranking for one job.

        Speculative rows are ranked by the background task. Silver medalists
        are ranked here, in-process: they are read live and never had a row in
        cv_bank_matches to begin with.
        """
        from job.candidate.models import CvBankMatches
        from matching.ranking import rank_profile

        job=await JobPosts.get_job_post_by_id(self.session,str(job_post_id))
        if not job:
            return rows
        job_fields={
            "title":job.title,
            "requirements":job.requirements,
            "optional_skills":job.optional_skills,
        }
        stored=await CvBankMatches.scores_for_job(self.session,job.id)
        for row in rows:
            if row["bank_source"]=="speculative":
                row["rank_score"]=stored.get(row["record_id"])
                if row["rank_score"] is None:
                    # Banked after the job opened, so the task never saw it.
                    row["rank_score"]=rank_profile(job_fields,{
                        "current_title":row.get("current_position"),
                        "headline":row.get("current_company"),
                        "skills":row.get("skills") or [],
                        "summary":None,
                    })
            else:
                row["rank_score"]=rank_profile(job_fields,{
                    "current_title":row.get("current_position"),
                    "headline":row.get("current_company"),
                    "skills":row.get("skills") or [],
                    "summary":None,
                })
        return rows

    async def list_matching(self,assigned=None,search=None,limit=10,offset=0):
        rows,total=await Manual_UPLOAD_CANDIDATE.list_matching(
            self.session,assigned=assigned,search=search,limit=limit,offset=offset,
        )
        job_ids=[str(r.job_post_id) for r in rows if r.job_post_id]
        posts=await JobPosts.get_by_ids(self.session,job_ids,active_only=False) if job_ids else []
        by_id={str(p.id):p for p in posts}
        data=[
            serialize_matching_candidate(r,by_id.get(str(r.job_post_id)) if r.job_post_id else None)
            for r in rows
        ]
        return data,total

    async def get_matching(self,record_id):
        row=await Manual_UPLOAD_CANDIDATE.get_by_id(self.session,record_id)
        if not row or row.apply_via!="cv_bank":
            raise HTTPException(status_code=404,detail="CV not found")
        job_post=None
        if row.job_post_id:
            job_post=await JobPosts.get_job_post_by_id(self.session,str(row.job_post_id))
        return serialize_matching_candidate(row,job_post)

    async def assign_matching(self,record_id,job_post_id,current_user=None):
        row=await Manual_UPLOAD_CANDIDATE.get_by_id(self.session,record_id)
        if not row or row.apply_via!="cv_bank":
            raise HTTPException(status_code=404,detail="CV not found")
        job_post=None
        if job_post_id not in (None,""):
            job_post=await JobPosts.get_job_post_by_id(self.session,str(job_post_id))
            if not job_post or job_post.is_deleted:
                raise HTTPException(status_code=404,detail="Job post not found")
        was_unassigned=row.job_post_id is None
        row=await Manual_UPLOAD_CANDIDATE.assign_job_post(self.session,record_id,job_post_id)
        if not row:
            raise HTTPException(status_code=404,detail="CV not found")
        if job_post is None and row.job_post_id:
            job_post=await JobPosts.get_job_post_by_id(self.session,str(row.job_post_id))
        if was_unassigned and row.job_post_id and row.user_id:
            title=(job_post.title if job_post else "") or str(row.job_post_id)
            await HistoryRecorder(self.session).record(
                HistoryEvent.CANDIDATE_CREATED.value,
                actor_id=current_user,user_id=row.user_id,
                manual_upload_candidate_id=row.id,
                entity_type="manual_upload_candidate",entity_id=row.id,
                to_value=title,
                description="cv_bank",commit=True,
            )
        return serialize_matching_candidate(row,job_post)

    async def application_history_by_emails(self,emails):
        """Prior applications keyed by lowercase email across users / inbox /
        manual_upload_candidate / form_data / candidates."""
        lowers=[]
        seen=set()
        for raw in emails or []:
            email=_norm_email(raw)
            if email and email not in seen:
                seen.add(email)
                lowers.append(email)
        if not lowers:
            return {}
        users=await Users.get_users_by_emails(self.session,lowers)
        user_by_email={_norm_email(u.email):u for u in users}
        inbox_rows=await Inbox.list_applications_by_emails(self.session,lowers)
        filtered_rows=await Inbox_Message_Triage.list_filtered_by_emails(self.session,lowers)
        manual_rows=await Manual_UPLOAD_CANDIDATE.list_by_emails(self.session,lowers)
        form_rows=await FormData.list_by_emails(self.session,lowers)
        ats_rows=await Candidates.list_by_emails(self.session,lowers)
        packed={email:{"present_in":[],"user":user_by_email.get(email),"applications":[]} for email in lowers}
        for email,user in user_by_email.items():
            if email in packed:
                packed[email]["present_in"].append("users")
                packed[email]["user"]=user
        ingested_upstream={
            str(row.get("upstream_id")) for row in inbox_rows if row.get("upstream_id")
        }
        for row in inbox_rows:
            email=_norm_email(row.get("email"))
            if email in packed:
                packed[email]["applications"].append(row)
        for row in filtered_rows:
            email=_norm_email(row.get("email"))
            if email not in packed:
                continue
            upstream=str(row.get("upstream_id") or row.get("message_id") or "")
            if upstream and upstream in ingested_upstream:
                continue
            packed[email]["applications"].append(row)
        for row in manual_rows:
            email=_norm_email(row.get("email"))
            if email not in packed:
                continue
            packed[email]["applications"].append(row)
            if "manual_upload_candidate" not in packed[email]["present_in"]:
                packed[email]["present_in"].append("manual_upload_candidate")
        for row in form_rows:
            email=_norm_email(row.get("email"))
            if email not in packed:
                continue
            packed[email]["applications"].append(row)
            if "form_data" not in packed[email]["present_in"]:
                packed[email]["present_in"].append("form_data")
        for row in ats_rows:
            email=_norm_email(row.get("email"))
            if email not in packed:
                continue
            # ATS scores are job-assignment history, not applications.
            if "candidates" not in packed[email]["present_in"]:
                packed[email]["present_in"].append("candidates")
        for pack in packed.values():
            pack["applications"].sort(key=lambda r: r.get("applied_at") or "",reverse=True)
            present=pack["present_in"]
            pack["present_in"]=[name for name in _HISTORY_TABLES if name in present]
        return packed

    async def get_application_history(self,email):
        cleaned=_norm_email(email)
        if not cleaned:
            raise HTTPException(status_code=422,detail="email is required")
        packed=await self.application_history_by_emails([cleaned])
        pack=packed.get(cleaned) or {"present_in":[],"user":None,"applications":[]}
        return serialize_application_history(
            cleaned,user=pack.get("user"),present_in=pack.get("present_in"),
            applications=pack.get("applications"),
        )

    async def attach_application_history(self,payloads):
        """Stamp is_reapplicant + previous_applications onto list/detail dicts.

        ``previous_applications`` is every application for that email, including
        the open row. ``is_reapplicant`` means a *different* kept attempt —
        another email, form, or upload, even when neither has a job assigned.
        """
        single=not isinstance(payloads,list)
        records=[payloads] if single else list(payloads or [])
        emails=[_payload_email(p) for p in records]
        history=await self.application_history_by_emails(emails)
        for payload in records:
            if not isinstance(payload,dict):
                continue
            email=_payload_email(payload)
            pack=history.get(email) or {"present_in":[],"user":None,"applications":[]}
            items=[]
            reapplied=False
            for row in pack.get("applications") or []:
                item=serialize_application_history_item(row)
                items.append(item)
                if is_kept_application(item) and not _is_current_application(row,payload):
                    reapplied=True
            items.sort(key=lambda r: r.get("applied_at") or "",reverse=True)
            payload["present_in"]=list(pack.get("present_in") or [])
            payload["is_reapplicant"]=reapplied
            payload["previous_applications"]=items
        return records[0] if single else records