467 lines
17 KiB
Python
467 lines
17 KiB
Python
"""Inbox helpers — attachment loading, resume text extraction, read-status sync."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import asyncio
|
|
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]:
|
|
"""Filename + public URL only. Open-resume uses the S3 link; do not pull bytes."""
|
|
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
|
|
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,
|
|
)
|