import logging import httpx,os from fastapi import HTTPException from inbox.enums import Candidate_application_Status from inbox.models import Inbox_Messages from inbox.file_decoder import decode_attachment from inbox.serializers import serialize_application, serialize_message from inbox.plugins import ( EMAIL_API_TOKEN, fetch_message_read_status, load_message_files, request_email_confirmation, ) from dotenv import load_dotenv load_dotenv() from sqlalchemy.ext.asyncio import AsyncSession from datetime import datetime,timezone logger=logging.getLogger("inbox.match") 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]=[] # 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 get_email_by_id(self,message_id,test_on=True): async with httpx.AsyncClient() as client: try: response=await client.get(f"{self.get_url}/emails/{message_id}", headers={"Authorization":f"Bearer {self.token}"} ) if response.status_code==200: data=response.json() 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 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 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 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): 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) elif isread==False: messages=await Inbox_Messages.get_inbox_messages(self.session,top,skip,search,isread,assigned=assigned) else: messages=await Inbox_Messages.get_inbox_messages(self.session,top,skip,search,assigned=assigned) 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 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): 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) elif isread==False: return await Inbox_Messages.count_inbox_messages(self.session,search,isread=False,assigned=assigned) else: return await Inbox_Messages.count_inbox_messages(self.session,search,assigned=assigned) 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): message=await Inbox_Messages.mark_message_read(self.session,record_id) if not message: raise HTTPException(status_code=404,detail="Message not found") return serialize_message(message) 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)