141 lines
5.9 KiB
Python
141 lines
5.9 KiB
Python
import logging
|
|
import httpx,os
|
|
from fastapi import HTTPException
|
|
from inbox.models import Inbox_Messages
|
|
from inbox.file_decoder import decode_attachment
|
|
from inbox.serializers import serialize_message
|
|
from inbox.plugins import (
|
|
EMAIL_API_TOKEN,
|
|
fetch_message_read_status,
|
|
load_message_files,
|
|
)
|
|
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]=[]
|
|
|
|
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):
|
|
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"))
|
|
insert_func=await Inbox_Messages.insert_email(session=self.session,email_data=data,file_path=re_create_file)
|
|
if (
|
|
insert_func.attachment
|
|
and insert_func.file_path
|
|
and insert_func.match_status is None
|
|
):
|
|
self.pending_match_ids.append(str(insert_func.id))
|
|
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_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
|
|
return item
|
|
|
|
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 count_inbox_messages(self,search=None):
|
|
return await Inbox_Messages.count_inbox_messages(self.session,search)
|
|
|
|
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)
|