import asyncio import logging import uuid import httpx,os from fastapi import HTTPException from inbox.enums import Candidate_application_Status from inbox.models import Inbox_Messages,Inbox_Message_Triage,MailboxSyncRun from inbox.file_decoder import decode_attachment from inbox.serializers import serialize_application, serialize_message, serialize_triage, serialize_mailbox_sync_run from inbox.plugins import ( EMAIL_API_TOKEN, fetch_message_read_status, load_message_files, request_email_confirmation, send_mail, ) from inbox_classifier.decorators import is_manual_upload,triage_fields from inbox_classifier.execute_agent import classify_email from inbox_classifier.plugins import ( TRIAGE_CONCURRENCY, TRIAGE_ENABLED, TRIAGE_STATUSES, sender_domain, should_ingest, triage_model_name, ) from dotenv import load_dotenv load_dotenv() from sqlalchemy.ext.asyncio import AsyncSession from datetime import datetime,timezone logger=logging.getLogger("inbox.match") triage_logger=logging.getLogger("inbox.triage") # One statement, one round trip — but an unbounded id list is still a client-supplied # IN () of arbitrary size, so the batch is capped and the route answers 413. MAX_BULK_READ_IDS=100 class Email: def __init__(self,session:AsyncSession,token=None): self.session=session self.get_url=os.getenv("EMAIL_URL") self.token=token or EMAIL_API_TOKEN self.pending_match_ids:list[str]=[] self.pending_confirmation_emails:list[str]=[] # Upstream ids the intake gate judged not to be job applications. They get a # verdict row and no inbox_messages row. self.skipped_message_ids:list[str]=[] self.triage_errors:list[str]=[] # async def get_all_applications(self,app_id=None): # try: # if app_id: # application_lst=await Inbox_Messages.get_all_applications(self.session,message_id=app_id) # else: # application_lst=await Inbox_Messages.get_all_applications(self.session) # return application_lst # except Exception as e: # raise HTTPException(status_code=500,detail=str(e)) async def service_email(self,top,skip): async with httpx.AsyncClient() as client: try: response=await client.get(f"{self.get_url}/emails", params={"skip":skip,"top":top}, headers={"Authorization":f"Bearer {self.token}"} ) if response.status_code==200: return response.json() else: raise HTTPException(status_code=response.status_code,detail=response.text) except Exception as e: raise HTTPException(status_code=500,detail=str(e)) async def fetch_message(self,message_id): """GET /emails/{id} on the upstream Email API -> the Graph payload.""" async with httpx.AsyncClient() as client: response=await client.get(f"{self.get_url}/emails/{message_id}", headers={"Authorization":f"Bearer {self.token}"} ) if response.status_code!=200: raise HTTPException(status_code=response.status_code,detail=response.text) return response.json() async def triage_round(self,message_ids): """Fetch and classify a whole /email/fetch page, bounded by a semaphore. Returns {message_id: decision}. The caller replays the page in upstream order, so pending_match_ids and pending_confirmation_emails keep the exact sequence they have today. Only the upstream GET and the OpenAI call run concurrently, and nothing inside the gather touches self.session — Depends(get_session) yields ONE AsyncSession, which cannot be shared across tasks. All DB work stays in the serial replay. Two pre-filters run first and cost no tokens: a message already in inbox_messages was judged an application once, and a message already in inbox_message_triage has a stored verdict to replay. That is what makes a repeated fetch free. """ ids=[str(m) for m in message_ids or [] if m] decisions={} if not ids: return decisions known=await Inbox_Messages.existing_message_ids(self.session,ids) recorded=await Inbox_Message_Triage.verdicts_for_message_ids(self.session,ids) pending=[] for message_id in ids: if message_id in known: decisions[message_id]={"ingest":True,"status":"known","fresh":False} elif message_id in recorded: decisions[message_id]={"ingest":recorded[message_id],"status":"recorded","fresh":False} else: pending.append(message_id) if not pending: triage_logger.info("triage round: page=%s known=%s classified=0",len(ids),len(decisions)) return decisions semaphore=asyncio.Semaphore(TRIAGE_CONCURRENCY) async def run(message_id): async with semaphore: data=await self.fetch_message(message_id) if not TRIAGE_ENABLED or is_manual_upload(data): return message_id,{"data":data,"ingest":True,"status":"disabled","fresh":False} verdict,error=await classify_email(data) ingest,status,reason=should_ingest(verdict,error) return message_id,{"data":data,"verdict":verdict,"error":error,"ingest":ingest, "status":status,"reason":reason,"fresh":True} results=await asyncio.gather(*(run(m) for m in pending),return_exceptions=True) accepted=rejected=errors=0 for result in results: if isinstance(result,BaseException): # First failure wins, preserving today's all-or-nothing behaviour for a # failing upstream message. Never catch BaseException itself: a # CancelledError must keep propagating. raise result message_id,decision=result decisions[message_id]=decision if decision.get("status")=="error": errors+=1 if decision.get("ingest"): accepted+=1 else: rejected+=1 triage_logger.info( "triage round: page=%s known=%s classified=%s accepted=%s rejected=%s errors=%s", len(ids),len(known)+len(recorded),len(pending),accepted,rejected,errors, ) return decisions async def record_triage(self,data,decision,ingested): """Persist one verdict. Never raises into the ingestion path. A failed audit write must not cost us a candidate: the worst case is that the next fetch re-classifies this message. """ try: fields=triage_fields( data, decision.get("verdict"), decision.get("status") or "classified", decision.get("reason") or "", error=decision.get("error") or "", model_name=triage_model_name(), ingested=ingested, ) await Inbox_Message_Triage.record_verdict(self.session,fields) # Allowlisted keys only: sender DOMAIN not address, attachment COUNT not # names, no subject or body text, no evidence text. verdict=decision.get("verdict") triage_logger.info( "triage %s: application=%s reason=%s confidence=%s domain=%s attachments=%s", fields["message_id"], fields["is_application"], fields["reason_code"], getattr(verdict,"confidence",None), sender_domain(fields["message_from"]), len(data.get("attachments") or []), ) except Exception as e: triage_logger.warning("triage record failed: %s",type(e).__name__) async def get_email_by_id(self,message_id,test_on=True,decision=None): """Persist one upstream message, gated by the application classifier. `decision` is the pre-computed verdict from triage_round; without one this classifies inline, so a single-message call still works. Ordering is deliberate. The verdict comes BEFORE decode_attachment: a rejected mail must not write a file into decoded_attachments (nothing on this path ever deletes one, and _write uses the basename only, so a vendor "resume.pdf" would clobber a candidate's stored CV), and must not reach _link_sender, which would create a candidate Users row and queue a confirmation mail for a stranger. The gate lives here, not in Inbox_Messages.insert_email, so FileRead.ingest_upload bypasses it for free — that path fabricates an EMPTY body and would be a guaranteed false negative under a subject+body classifier. """ try: if decision is None: decision=(await self.triage_round([message_id])).get(str(message_id)) or {} data=decision.get("data") or await self.fetch_message(message_id) if not decision.get("ingest"): if decision.get("fresh"): await self.record_triage(data,decision,ingested=False) self.skipped_message_ids.append(str(message_id)) if decision.get("status")=="error": self.triage_errors.append(str(message_id)) return {"message_id":str(message_id),"skipped":"not_application", "reason":decision.get("reason") or "","status":decision.get("status") or ""} re_create_file=await decode_attachment(data.get("attachments")) row,new_user_email=await Inbox_Messages.insert_email(session=self.session,email_data=data,file_path=re_create_file) if decision.get("fresh"): await self.record_triage(data,decision,ingested=True) if row.attachment and row.file_path and row.match_status is None: self.pending_match_ids.append(str(row.id)) if test_on: return data if new_user_email: self.pending_confirmation_emails.append(new_user_email) return data except HTTPException: # Was missing: the bare `except Exception` below caught the upstream-status # HTTPException and re-raised every one of them as a 500. raise except Exception as e: raise HTTPException(status_code=500,detail=str(e)) async def get_inbox_messages(self,top,skip,search=None): messages=await Inbox_Messages.get_inbox_messages(self.session,top,skip,search) items=[] for m in messages: item=serialize_message(m) files=load_message_files(m) if files: item["files"]=files items.append(item) return items async def get_inbox_message_by_id(self,record_id): message=await Inbox_Messages.get_inbox_message_by_id(self.session,record_id) if not message: raise HTTPException(status_code=404,detail="Message not found") item=serialize_message(message) files=load_message_files(message) if files: item["files"]=files from job.candidate.views import CandidateView cv=CandidateView(session=self.session) suggested=[] for job_id in item.get("suggested_job_post_ids") or []: jp=await cv.get_job_post_by_id(record_id=job_id) if jp: if jp.get("is_deleted") or not jp.get("is_active"): suggested.append({**jp,"unavailable":True}) else: suggested.append(jp) else: suggested.append({"id":str(job_id),"unavailable":True}) item["suggested_job_posts"]=suggested assigned_id=item.get("assigned_job_post_id") if assigned_id: item["assigned_job_post"]=await cv.get_job_post_by_id(record_id=assigned_id) else: item["assigned_job_post"]=None return item async def get_all_applications(self,top,skip,search=None,isread:bool=True,application_status:Candidate_application_Status=Candidate_application_Status.CLOSED,assigned=None,is_duplicate=None): if application_status in (Candidate_application_Status.PROCESS, Candidate_application_Status.REJECTED, Candidate_application_Status.SCREENING, Candidate_application_Status.ASSESSMENT, Candidate_application_Status.INTERVIEW, Candidate_application_Status.OFFER, Candidate_application_Status.HIRED): messages=await Inbox_Messages.get_inbox_messages(self.session,top,skip,search,application_status=application_status,assigned=assigned,is_duplicate=is_duplicate) elif isread==False: messages=await Inbox_Messages.get_inbox_messages(self.session,top,skip,search,isread,assigned=assigned,is_duplicate=is_duplicate) else: messages=await Inbox_Messages.get_inbox_messages(self.session,top,skip,search,assigned=assigned,is_duplicate=is_duplicate) return [serialize_application(m) for m in messages] async def get_application_by_id(self,record_id): message=await Inbox_Messages.get_inbox_message_by_id(self.session,record_id) if not message: raise HTTPException(status_code=404,detail="Application not found") return serialize_application(message) async def queue_rematch(self,record_id): message=await Inbox_Messages.get_inbox_message_by_id(self.session,record_id) if not message: raise HTTPException(status_code=404,detail="Message not found") if not message.attachment or not message.file_path: raise HTTPException(status_code=400,detail="Message has no attachment to match") return str(message.id) async def enqueue_matching(self,inbox_ids,force=False): from inbox.tasks import match_inbox_message task_ids=[] for record_id in inbox_ids or []: created_at=datetime.now(timezone.utc).isoformat() task=await match_inbox_message.kicker().with_labels( created_at=created_at, correlation_id=str(record_id), queue="inbox", ).kiq(str(record_id),force=force) task_ids.append(task.task_id) return task_ids async def start_mailbox_sync(self,current_user=None,top=100,skip=0,test_on=True): """Enqueue Outlook pull on the mailbox_sync queue; return the run row. If a queued/running sync already exists, return it instead of stacking another. """ active=await MailboxSyncRun.get_active(self.session) if active: return serialize_mailbox_sync_run(active) created_by=None if isinstance(current_user,dict) and current_user.get("id"): created_by=MailboxSyncRun._as_uuid(current_user.get("id")) row=await MailboxSyncRun.insert_run(self.session,{ "status":"queued", "created_by":created_by, "top":int(top or 100), "skip":int(skip or 0), "test_on":bool(test_on) if test_on is not None else True, }) from inbox.mailbox_sync_tasks import sync_mailbox task=await sync_mailbox.kicker().with_labels( created_at=datetime.now(timezone.utc).isoformat(), correlation_id=str(row.id), queue="mailbox_sync", ).kiq(str(row.id)) row=await MailboxSyncRun.update_run(self.session,row.id,{"task_id":task.task_id}) return serialize_mailbox_sync_run(row) async def get_mailbox_sync(self,run_id=None): if run_id: row=await MailboxSyncRun.get_by_id(self.session,run_id) if not row: raise HTTPException(status_code=404,detail="Sync run not found") return serialize_mailbox_sync_run(row) row=await MailboxSyncRun.get_active(self.session) if row: return serialize_mailbox_sync_run(row) # Latest finished run so the UI can still show the last result after refresh. from sqlmodel import select result=await self.session.execute( select(MailboxSyncRun).order_by(MailboxSyncRun.created_at.desc()).limit(1) ) row=result.scalars().first() if not row: raise HTTPException(status_code=404,detail="No sync runs yet") return serialize_mailbox_sync_run(row) async def run_mailbox_sync_page(self,top=100,skip=0,test_on=True): """Pull one Outlook page, triage, ingest, enqueue matching. Returns summary. Shared by the legacy synchronous /email/fetch and the background Taskiq worker. """ data=await self.service_email(top,skip) value=data.get("value") or [] decisions=await self.triage_round([item.get("id") for item in value]) entries=[] for item in value: message_id=item.get("id") decision=decisions.get(str(message_id)) or {} try: result=await self.get_email_by_id(message_id,test_on,decision=decision) if isinstance(result,dict) and result.get("skipped"): entries.append({ "message_id":str(message_id), "status":"skipped", "reason":result.get("reason") or "", "triage_status":result.get("status") or "", }) else: # Prefer the decision status (known/recorded/…) when present. status=decision.get("status") or "ingested" if status in ("known","recorded","disabled"): entry_status="known" if status=="known" else "ingested" else: entry_status="ingested" entries.append({ "message_id":str(message_id), "status":entry_status, "reason":decision.get("reason") or "", "triage_status":status, }) except Exception as e: entries.append({ "message_id":str(message_id), "status":"error", "reason":str(e), "triage_status":"error", }) self.triage_errors.append(str(message_id)) if self.pending_match_ids: await self.enqueue_matching(list(self.pending_match_ids),force=False) account_setup=[] if not test_on and self.pending_confirmation_emails: account_setup=await self.send_account_setup(list(self.pending_confirmation_emails)) skipped=len(self.skipped_message_ids) ingested=sum(1 for e in entries if e.get("status") in ("ingested","known")) triage={ "ingested":ingested, "skipped":skipped, "errors":len(self.triage_errors), "total":len(entries), } return {"entries":entries,"triage":triage,"account_setup":account_setup} async def send_account_setup(self,emails): results=[] for email in emails or []: try: status=await request_email_confirmation(email) results.append({"email":email,"sent":status==200}) except Exception as e: logger.warning("confirmation request failed for %s: %s",email,e) results.append({"email":email,"sent":False}) return results async def count_inbox_messages(self,search=None,isread:bool=True,application_status:Candidate_application_Status=Candidate_application_Status.CLOSED,assigned=None,is_duplicate=None): if application_status == Candidate_application_Status.PROCESS or application_status==Candidate_application_Status.REJECTED: return await Inbox_Messages.count_inbox_messages(self.session,search,application_status=application_status,assigned=assigned,is_duplicate=is_duplicate) elif isread==False: return await Inbox_Messages.count_inbox_messages(self.session,search,isread=False,assigned=assigned,is_duplicate=is_duplicate) else: return await Inbox_Messages.count_inbox_messages(self.session,search,assigned=assigned,is_duplicate=is_duplicate) async def assign_job_post(self,record_id,job_post_id): message=await Inbox_Messages.get_inbox_message_by_id(self.session,record_id) if not message: raise HTTPException(status_code=404,detail="Message not found") if job_post_id is not None: from job.job_post.models import JobPosts post=await JobPosts.get_job_post_by_id(self.session,job_post_id) if not post or post.is_deleted or not post.is_active: raise HTTPException(status_code=422,detail="Job post is missing, deleted, or inactive") updated=await Inbox_Messages.set_assigned_job_post(self.session,record_id,job_post_id) if not updated: raise HTTPException(status_code=404,detail="Message not found") if job_post_id is not None: # Assignment pairs this CV with a JD we already have — queue the ATS # score in the background so the recruiter is not held on an OpenAI # call. Idempotent server-side; broker-down just logs (the profile's # Score-with-ATS button remains the manual fallback). from inbox.tasks import score_inbox_message try: await score_inbox_message.kicker().with_labels( created_at=datetime.now(timezone.utc).isoformat(), correlation_id=str(record_id), queue="inbox", ).kiq(str(record_id),str(job_post_id)) except Exception as exc: logger.warning("could not queue ats score for %s: %s",record_id,exc) return await self.get_inbox_message_by_id(record_id) async def mark_read(self,record_id,read=True): message=await Inbox_Messages.mark_message_read(self.session,record_id,read) if not message: raise HTTPException(status_code=404,detail="Message not found") return serialize_message(message) async def set_read_bulk(self,record_ids,read): """Flip read state for a hand-picked selection. Returns counts, never rows: a 500-id selection would otherwise serialize 500 full messages back at a client that only needs to know it worked. `updated` < `requested` means some ids no longer exist — a stale selection against a list that moved. That is reported, not raised, because the rows that DID exist were already committed. """ ids=[str(r).strip() for r in (record_ids or []) if str(r or "").strip()] if not ids: raise HTTPException(status_code=422,detail="record_ids must contain at least one id") if len(ids)>MAX_BULK_READ_IDS: raise HTTPException(status_code=413, detail=f"At most {MAX_BULK_READ_IDS} ids per request") updated=await Inbox_Messages.set_read_bulk(self.session,ids,read) logger.info("bulk read: requested=%s updated=%s read=%s",len(ids),updated,bool(read)) return {"requested":len(ids),"updated":updated,"read":bool(read)} async def set_read_all(self,read,search=None,isread:bool=True, application_status:Candidate_application_Status=Candidate_application_Status.CLOSED, assigned=None,is_duplicate=None): """Mark every row the SAME filter set would have listed. The filter arguments are the caller's current view, not a free-form query: the button says "mark all read in this view" and the WHERE chain is literally the list's own (Inbox_Messages._apply_filters), so the two cannot drift. """ updated=await Inbox_Messages.set_read_scope( self.session,read,search=search,isread=isread, application_status=application_status,assigned=assigned,is_duplicate=is_duplicate, ) logger.info("scope read: updated=%s read=%s isread=%s assigned=%s status=%s dup=%s", updated,bool(read),isread,assigned,getattr(application_status,"value",application_status),is_duplicate) return {"updated":updated,"read":bool(read)} async def refresh_read_status(self,record_id): message=await Inbox_Messages.get_inbox_message_by_id(self.session,record_id) if not message: raise HTTPException(status_code=404,detail="Message not found") if not message.message_id: raise HTTPException(status_code=400,detail="Message has no upstream id") try: status=await fetch_message_read_status(message.message_id,token=self.token) except httpx.HTTPStatusError as e: raise HTTPException(status_code=e.response.status_code,detail=e.response.text) except Exception as e: raise HTTPException(status_code=500,detail=str(e)) if status is None: raise HTTPException(status_code=404,detail="Message not found upstream") await Inbox_Messages.apply_read_status(self.session,[status]) refreshed=await Inbox_Messages.get_inbox_message_by_id(self.session,record_id) return serialize_message(refreshed) async def get_counts(self): return await Inbox_Messages.count_processing(self.session) async def set_processing_state(self,record_id,processing_state): allowed=("unread","imported","processed","rejected") if processing_state not in allowed: raise HTTPException(status_code=422,detail=f"processing_state must be one of {', '.join(allowed)}") message=await Inbox_Messages.set_processing_state(self.session,record_id,processing_state) if not message: raise HTTPException(status_code=404,detail="Message not found") return serialize_application(message) async def set_duplicate(self,record_id,is_duplicate): if not isinstance(is_duplicate,bool): raise HTTPException(status_code=422,detail="is_duplicate must be a boolean") message=await Inbox_Messages.set_duplicate(self.session,record_id,is_duplicate) if not message: raise HTTPException(status_code=404,detail="Message not found") return serialize_application(message) async def get_triage_messages(self,top,skip,search=None,is_application=None,status=None): """The intake gate's verdict log — mostly the mail that never became a row. A hard gate's only real risk is the silent false negative, so the rejections have to be reviewable. """ if status is not None and status not in TRIAGE_STATUSES: raise HTTPException(status_code=422,detail=f"status must be one of {', '.join(TRIAGE_STATUSES)}") rows=await Inbox_Message_Triage.list_triage(self.session,top,skip,is_application,status,search) return [serialize_triage(row) for row in rows] async def count_triage(self,search=None,is_application=None,status=None): return await Inbox_Message_Triage.count_triage(self.session,is_application,status,search) async def override_triage(self,record_id,is_application,current_user=None): """Overturn a verdict a recruiter disagrees with. false -> true re-fetches the mail from upstream and runs the normal ingestion path, which is why the body was never stored. true -> false does NOT delete the inbox_messages row: inbox, ats_results, assessments, notifications and application_stage_transitions all reference it, so a purge would take candidate accounts and scores with it. It moves the row to processing_state 'rejected' instead, an already-allowlisted value. """ if not isinstance(is_application,bool): raise HTTPException(status_code=422,detail="is_application must be a boolean") row=await Inbox_Message_Triage.get_triage_by_id(self.session,record_id) if not row: raise HTTPException(status_code=404,detail="Triage record not found") user_id=(current_user or {}).get("id") if is_application and not row.ingested: data=await self.fetch_message(row.message_id) re_create_file=await decode_attachment(data.get("attachments")) message,new_user_email=await Inbox_Messages.insert_email(session=self.session,email_data=data,file_path=re_create_file) await Inbox_Message_Triage.mark_ingested(self.session,row.message_id,True) if message.attachment and message.file_path and message.match_status is None: await self.enqueue_matching([str(message.id)],force=False) if new_user_email: await self.send_account_setup([new_user_email]) elif not is_application and row.ingested: message=await Inbox_Messages.get_by_upstream_id(self.session,row.message_id) if message: await Inbox_Messages.set_processing_state(self.session,message.id,"rejected") updated=await Inbox_Message_Triage.set_override(self.session,record_id,is_application,user_id) return serialize_triage(updated) async def send_email(self,payload,current_user): to_email=(payload.get("to") or "").strip() subject=(payload.get("subject") or "").strip() body=payload.get("body") or "" content_type=(payload.get("content_type") or "html").strip() or "html" if not to_email: raise HTTPException(status_code=422,detail="to is required") if not subject: raise HTTPException(status_code=422,detail="subject is required") if not body: raise HTTPException(status_code=422,detail="body is required") try: await send_mail(to_email,subject,body,content_type=content_type) except (httpx.HTTPError,RuntimeError) as e: raise HTTPException(status_code=502,detail="Failed to send email") from e inbox_id=payload.get("inbox_id") job_post_id=None try: from notifications.models import Notifications uid=None raw=current_user.get("id") if current_user else None if raw: uid=uuid.UUID(str(raw)) if uid: await Notifications.insert_notification(self.session,{ "user_id":uid, "kind":"message", "title":"Email sent", "body":f"Sent “{subject}” to {to_email}", "link_path":"/inbox", "inbox_id":int(inbox_id) if inbox_id is not None else None, "job_post_id":job_post_id, }) except Exception as exc: logger.warning("notification insert skipped: %s",exc) return {"accepted":True,"to":to_email,"subject":subject} async def reply_email(self,payload,current_user): record_id=payload.get("record_id") body=payload.get("body") or "" if not record_id: raise HTTPException(status_code=422,detail="record_id is required") if not str(body).strip(): raise HTTPException(status_code=422,detail="body is required") message=await Inbox_Messages.get_inbox_message_by_id(self.session,record_id) if not message: raise HTTPException(status_code=404,detail="Message not found") to_email=(message.message_from or "").strip() if not to_email: raise HTTPException(status_code=422,detail="Message has no sender address") original=(message.message_subject or "").strip() subject=original if original.lower().startswith("re:") else f"Re: {original}" if original else "Re:" try: await send_mail(to_email,subject,body,content_type="html") except (httpx.HTTPError,RuntimeError) as e: raise HTTPException(status_code=502,detail="Failed to send email") from e try: from notifications.models import Notifications uid=None raw=current_user.get("id") if current_user else None if raw: uid=uuid.UUID(str(raw)) if uid: await Notifications.insert_notification(self.session,{ "user_id":uid, "kind":"message", "title":"Email sent", "body":f"Replied to {to_email}", "link_path":"/inbox", }) except Exception as exc: logger.warning("notification insert skipped: %s",exc) return {"accepted":True,"to":to_email,"subject":subject}