HR-ATS-Portal/backend/inbox/plugins.py

"""Inbox helpers — attachment loading, resume text extraction, read-status sync."""

from __future__ import annotations

import asyncio
import base64
import logging
import os
import uuid
from pathlib import Path
from urllib.parse import quote

import httpx
from dotenv import load_dotenv
from sqlalchemy.ext.asyncio import AsyncSession
from sqlmodel import select

from inbox.models import AtsResults, Inbox, Inbox_Messages
from job.candidate.models import Candidates, Manual_UPLOAD_CANDIDATE
from job.candidate.views import FileRead

load_dotenv()

logger=logging.getLogger("inbox.plugins")

EMAIL_URL=os.getenv("EMAIL_URL")
EMAIL_API_TOKEN=os.getenv("EMAIL_API_TOKEN")
BACKEND_URL=os.getenv("BACKEND_URL","http://localhost:8000")
TEAMS_MAIL_API_URL=os.getenv("TEAMS_MAIL_API_URL")
TEAMS_API_TOKEN=os.getenv("TEAMS_API_TOKEN")
MAIL_ACCEPTED_STATUS=202

_ATTACHMENTS_DIR=Path(__file__).resolve().parent/"decoded_attachments"


async def request_email_confirmation(email):
    """POST /users/confirm-email/resend on this same service -> status code.

    Goes through the endpoint rather than importing Confirmation so the token row,
    resend cooldown and mail send stay on one code path.
    """
    async with httpx.AsyncClient(timeout=20.0) as client:
        response=await client.post(
            f"{BACKEND_URL.rstrip('/')}/users/confirm-email/resend",
            json={"email":email},
        )
    return response.status_code


async def fetch_read_status_delta(folder, since=None, limit=100, max_pages=10, token=None):
    """GET /sync/read-status -> the raw round dict.

    Upstream Email API caps `limit` at 100; keep the default at that ceiling.
    """
    if not EMAIL_URL:
        raise RuntimeError("EMAIL_URL must be set")
    auth_token=token or EMAIL_API_TOKEN
    if not auth_token:
        raise RuntimeError("EMAIL_API_TOKEN must be set")
    params={"folder":folder,"limit":min(int(limit or 100),100),"max_pages":max_pages}
    if since:
        params["since"]=since
    async with httpx.AsyncClient(timeout=15.0) as client:
        response=await client.get(
            f"{EMAIL_URL.rstrip('/')}/sync/read-status",
            params=params,
            headers={"Authorization":f"Bearer {auth_token}"},
        )
    if response.status_code>=400:
        raise httpx.HTTPStatusError(
            response.text,
            request=response.request,
            response=response,
        )
    return response.json()


async def fetch_message_read_status(message_id, token=None):
    """GET /sync/read-status/message/{id} -> record dict, or None on 404."""
    if not EMAIL_URL:
        raise RuntimeError("EMAIL_URL must be set")
    auth_token=token or EMAIL_API_TOKEN
    if not auth_token:
        raise RuntimeError("EMAIL_API_TOKEN must be set")
    encoded_id=quote(str(message_id),safe="")
    async with httpx.AsyncClient(timeout=15.0) as client:
        response=await client.get(
            f"{EMAIL_URL.rstrip('/')}/sync/read-status/message/{encoded_id}",
            headers={"Authorization":f"Bearer {auth_token}"},
        )
    if response.status_code==404:
        return None
    if response.status_code>=400:
        raise httpx.HTTPStatusError(
            response.text,
            request=response.request,
            response=response,
        )
    return response.json()


def resolve_attachment_path(path_str:str) -> Path:
    """Legacy local-path resolver — kept for any old rows still on disk.

    New Email/Manual rows store HTTPS S3 URLs in file_path; callers should use
    ``load_file_bytes`` / ``extract_resume_text`` which handle URLs first.
    """
    raw=path_str.strip()
    path=Path(raw)
    if path.is_file():
        return path
    basename=Path(raw.replace("\\","/")).name
    fallback=_ATTACHMENTS_DIR/basename
    if fallback.is_file():
        return fallback
    return path


def load_file_bytes(path_or_url: str) -> bytes | None:
    """Load CV bytes from an S3 URL (preferred) or a leftover local path."""
    raw=(path_or_url or "").strip()
    if not raw:
        return None
    if raw.lower().startswith("http://") or raw.lower().startswith("https://"):
        from s3.plugins import S3,S3ServiceError
        try:
            return S3().download_bytes(raw)
        except S3ServiceError:
            logger.exception("s3 download failed for %s",raw[:120])
            return None
    path=resolve_attachment_path(raw)
    if not path.is_file():
        return None
    try:
        return path.read_bytes()
    except OSError:
        return None


def load_message_files(message:Inbox_Messages) -> list[dict]:
    if not message.file_path:
        return []
    names=[n.strip() for n in (message.file_name or "").split(",") if n.strip()]
    files=[]
    for idx,path_str in enumerate(p.strip() for p in message.file_path.split(",") if p.strip()):
        name=names[idx] if idx<len(names) else Path(path_str.replace("\\","/")).name
        entry={"file_name":name or "resume.pdf","url":None,"content_base64":None,"size":0}
        if path_str.lower().startswith("http://") or path_str.lower().startswith("https://"):
            entry["url"]=path_str
            raw=load_file_bytes(path_str)
            if raw is not None:
                entry["content_base64"]=base64.b64encode(raw).decode("ascii")
                entry["size"]=len(raw)
            files.append(entry)
            continue
        path=resolve_attachment_path(path_str)
        if not path.is_file():
            continue
        try:
            raw=path.read_bytes()
        except OSError:
            continue
        entry["file_name"]=path.name
        entry["content_base64"]=base64.b64encode(raw).decode("ascii")
        entry["size"]=len(raw)
        files.append(entry)
    return files


async def attach_email_pdfs_to_s3(session,row,pdfs,*,created_new:bool):
    """Upload PDFs under Email/{row.id}/{user_id}/ and set file_path to permanent URLs.

    Atomicity: if upload fails and ``created_new`` is True, delete the inbox_messages
    row (and inbox links). Re-sync of an existing row does not delete on failure.
    Returns the refreshed row.
    """
    from s3.plugins import S3,S3Source

    if not pdfs:
        return row
    owner_id=await Inbox_Messages.get_linked_user_id(session,row.id)
    if owner_id is None:
        owner_id="unlinked"
    s3=S3()
    urls=[]
    names=[]
    uploaded_keys=[]
    try:
        for pdf in pdfs:
            result=s3.upload_for_record(
                pdf["body"],
                pdf.get("name") or "resume.pdf",
                source=S3Source.EMAIL,
                record_id=row.id,
                owner_id=owner_id,
                content_type="application/pdf",
            )
            urls.append(result["url"])
            names.append(result.get("filename") or pdf.get("name") or "resume.pdf")
            uploaded_keys.append(result["key"])
        return await Inbox_Messages.set_file_paths(session,row.id,urls,names)
    except Exception:
        for key in uploaded_keys:
            try:
                s3.delete_object(key)
            except Exception:
                logger.exception("s3 cleanup failed key=%s",key)
        if created_new:
            await Inbox_Messages.delete_by_id(session,row.id)
        raise


async def extract_resume_text(file_paths:list[str]) -> tuple[str,str]:
    """Extract text from S3 URLs or leftover local PDF paths."""
    refs=[p.strip() for p in (file_paths or []) if p and p.strip()]
    if not refs:
        return "","no PDF attachment to extract"

    texts=[]
    errors=[]
    for ref in refs:
        name=Path(ref.replace("\\","/")).name or "resume.pdf"
        is_url=ref.lower().startswith("http://") or ref.lower().startswith("https://")
        if not is_url and not name.lower().endswith(".pdf"):
            continue
        if is_url and ".pdf" not in ref.lower() and not name.lower().endswith(".pdf"):
            # still try — key may omit extension rarely
            pass
        try:
            raw=await asyncio.to_thread(load_file_bytes,ref)
            if raw is None:
                errors.append(f"{name}: could not load file (S3 Access Denied or missing)")
                continue
            result=await FileRead(session=None,filename=name if name.lower().endswith(".pdf") else f"{name}.pdf",file=raw).read_file()
            text=(result.get("text") or "").strip()
            if text:
                texts.append(text)
            else:
                errors.append(f"{name}: no text extracted")
        except Exception as exc:
            errors.append(f"{name}: {exc}")

    if not texts:
        return "","; ".join(errors) if errors else "no text extracted from PDF"
    return "\n\n---\n\n".join(texts),""


def _ats_score_payload(row):
    if row is None:
        return None
    return {
        "overall_score":row.overall_score,
        "band":row.band or None,
        "job_post_id":str(row.job_post_id) if row.job_post_id else None,
        "computed_at":row.computed_at.isoformat() if row.computed_at else None,
        "candidate_id":str(row.candidate_id) if row.candidate_id else None,
        "user_id":str(row.user_id) if row.user_id else None,
    }


async def get_ats_score_for_user(session:AsyncSession,user_id,job_post_id=None):
    """Current ats_results overall score for a candidate -> dict, or None.

    Prefer ats_results.user_id (CV email matched that user). Fall back to the
    inbox.user_id join for scores whose email did not match any user, where
    candidate_id is set and user_id is NULL.

    Pass job_post_id to pin one application when a candidate has several;
    without it the newest current score across their applications wins.
    """
    try:
        uid=uuid.UUID(str(user_id))
    except (TypeError,ValueError):
        return None

    direct=(
        select(AtsResults)
        .where(AtsResults.user_id==uid,AtsResults.is_current==True)  # noqa: E712
        .order_by(AtsResults.computed_at.desc())
    )
    if job_post_id:
        try:
            direct=direct.where(AtsResults.job_post_id==uuid.UUID(str(job_post_id)))
        except (TypeError,ValueError):
            return None
    row=(await session.execute(direct)).scalars().first()
    if row is not None:
        return _ats_score_payload(row)

    qry=(
        select(AtsResults)
        .join(Inbox,AtsResults.inbox_id==Inbox.id)
        .join(Inbox_Messages,Inbox.message_id==Inbox_Messages.id)
        .where(
            Inbox.user_id==uid,
            Inbox_Messages.assigned_job_post_id.is_not(None),
            AtsResults.job_post_id==Inbox_Messages.assigned_job_post_id,
            AtsResults.is_current==True,  # noqa: E712
        )
        .order_by(AtsResults.computed_at.desc())
    )
    if job_post_id:
        try:
            qry=qry.where(Inbox_Messages.assigned_job_post_id==uuid.UUID(str(job_post_id)))
        except (TypeError,ValueError):
            return None

    return _ats_score_payload((await session.execute(qry)).scalars().first())


async def get_ats_scores_for_users(session:AsyncSession,user_ids):
    """{str(user_id): score payload} for a whole page of candidates.

    Same source and same three resolution paths as get_ats_score_for_user /
    get_ats_score_for_manual_user, but three queries for the page instead of two
    per row — a 100-card talent pool called the single-row helpers 200 times.

    Paths are applied in precedence order and a user found by an earlier one is
    never overwritten: direct ats_results.user_id, then the inbox join for scores
    whose email matched no user, then the manual_upload email join. Within a path
    the newest current score wins, which is what the unpinned single-row helpers
    return when a candidate has several applications.
    """
    uids=[]
    seen=set()
    for raw in user_ids or []:
        try:
            uid=uuid.UUID(str(raw))
        except (TypeError,ValueError):
            continue
        if uid not in seen:
            seen.add(uid)
            uids.append(uid)
    if not uids:
        return {}

    scores={}

    def collect(pairs):
        # computed_at DESC on every query, so the first row seen for a user is
        # the newest, and a later path can never displace an earlier one.
        for ats,owner in pairs:
            key=str(owner) if owner else None
            if key and key not in scores:
                scores[key]=_ats_score_payload(ats)

    direct=(
        select(AtsResults)
        .where(AtsResults.user_id.in_(uids),AtsResults.is_current==True)  # noqa: E712
        .order_by(AtsResults.computed_at.desc())
    )
    collect((r,r.user_id) for r in (await session.execute(direct)).scalars().all())

    remaining=[u for u in uids if str(u) not in scores]
    if remaining:
        via_inbox=(
            select(AtsResults,Inbox.user_id)
            .join(Inbox,AtsResults.inbox_id==Inbox.id)
            .join(Inbox_Messages,Inbox.message_id==Inbox_Messages.id)
            .where(
                Inbox.user_id.in_(remaining),
                Inbox_Messages.assigned_job_post_id.is_not(None),
                AtsResults.job_post_id==Inbox_Messages.assigned_job_post_id,
                AtsResults.is_current==True,  # noqa: E712
            )
            .order_by(AtsResults.computed_at.desc())
        )
        collect((await session.execute(via_inbox)).all())

    remaining=[u for u in uids if str(u) not in scores]
    if remaining:
        via_manual=(
            select(AtsResults,Manual_UPLOAD_CANDIDATE.user_id)
            .join(Candidates,AtsResults.candidate_id==Candidates.id)
            .join(
                Manual_UPLOAD_CANDIDATE,
                (Candidates.job_id==Manual_UPLOAD_CANDIDATE.job_post_id)
                &(Candidates.candidate_email==Manual_UPLOAD_CANDIDATE.candidate_email),
            )
            .where(
                Manual_UPLOAD_CANDIDATE.user_id.in_(remaining),
                Manual_UPLOAD_CANDIDATE.apply_via=="manual_upload",
                Candidates.status=="completed",
                AtsResults.job_post_id==Manual_UPLOAD_CANDIDATE.job_post_id,
                AtsResults.is_current==True,  # noqa: E712
            )
            .order_by(AtsResults.computed_at.desc())
        )
        collect((await session.execute(via_manual)).all())

    return scores


async def get_ats_score_for_manual_user(session:AsyncSession,user_id,job_post_id=None):
    """Current ats_results overall score for an Add Candidate user -> dict, or None.

    Prefer ats_results.user_id (Add Candidate always creates a users row, so a
    later score against that email lands on user_id). Fall back to the
    email+candidate_id join for rows written before user_id existed.
    apply_via=manual_upload is the Add Candidate gate; /import never writes
    that table.

    Pass job_post_id to pin one application when a candidate has several;
    without it the newest current score across their applications wins.
    """
    try:
        uid=uuid.UUID(str(user_id))
    except (TypeError,ValueError):
        return None

    direct=(
        select(AtsResults)
        .where(AtsResults.user_id==uid,AtsResults.is_current==True)  # noqa: E712
        .order_by(AtsResults.computed_at.desc())
    )
    if job_post_id:
        try:
            jid=uuid.UUID(str(job_post_id))
        except (TypeError,ValueError):
            return None
        direct=direct.where(AtsResults.job_post_id==jid)
    row=(await session.execute(direct)).scalars().first()
    if row is not None:
        return _ats_score_payload(row)

    qry=(
        select(AtsResults)
        .join(Candidates,AtsResults.candidate_id==Candidates.id)
        .join(
            Manual_UPLOAD_CANDIDATE,
            (Candidates.job_id==Manual_UPLOAD_CANDIDATE.job_post_id)
            &(Candidates.candidate_email==Manual_UPLOAD_CANDIDATE.candidate_email),
        )
        .where(
            Manual_UPLOAD_CANDIDATE.user_id==uid,
            Manual_UPLOAD_CANDIDATE.apply_via=="manual_upload",
            Candidates.status=="completed",
            AtsResults.job_post_id==Manual_UPLOAD_CANDIDATE.job_post_id,
            AtsResults.is_current==True,  # noqa: E712
        )
        .order_by(AtsResults.computed_at.desc())
    )
    if job_post_id:
        try:
            jid=uuid.UUID(str(job_post_id))
        except (TypeError,ValueError):
            return None
        qry=qry.where(
            Manual_UPLOAD_CANDIDATE.job_post_id==jid,
            AtsResults.job_post_id==jid,
        )

    return _ats_score_payload((await session.execute(qry)).scalars().first())


async def send_mail(to_email: str, subject: str, body: str, content_type: str = "html") -> None:
    """POST multipart to TEAMS_MAIL_API_URL. Treats 202 as accepted.

    Same shape as notifications.plugins.send_confirmation_mail — duplicated
    rather than imported so each domain owns its own mail copy and env reads.
    """
    if not TEAMS_MAIL_API_URL or not TEAMS_API_TOKEN:
        raise RuntimeError("TEAMS_MAIL_API_URL and TEAMS_API_TOKEN must be set")
    fields=[
        ("subject",(None,subject)),
        ("body",(None,body)),
        ("content_type",(None,content_type or "html")),
        ("save_to_sent_items",(None,"false")),
        ("to",(None,to_email)),
    ]
    async with httpx.AsyncClient(timeout=15.0) as client:
        response=await client.post(
            TEAMS_MAIL_API_URL,
            files=fields,
            headers={"Authorization":f"Bearer {TEAMS_API_TOKEN}"},
        )
    if response.status_code!=MAIL_ACCEPTED_STATUS:
        raise httpx.HTTPStatusError(
            response.text,
            request=response.request,
            response=response,
        )