317 lines
15 KiB
Python
317 lines
15 KiB
Python
import logging
|
|
import uuid
|
|
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,
|
|
send_mail,
|
|
)
|
|
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)
|
|
|
|
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 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}
|