kpinquan/stmp_server

1import datetime2import json3import logging4import email5 6from email.message import Message7from email.mime.multipart import MIMEMultipart8from email.mime.text import MIMEText9 10 11from models import EmailModel12 13_logger = logging.getLogger(__name__)14_logger.setLevel(logging.INFO)15 16 17def get_email_model(msg: Message):18    subparts = [19        get_email_model(subpart)20        for subpart in msg.get_payload()21    ] if msg.is_multipart() else []22    body = "" if msg.is_multipart() else msg._payload23    return EmailModel(24        headers={k: v for k, v in msg.items()},25        body=body,26        content_type=msg.get_content_type(),27        size=len(body) + sum(subpart.size for subpart in subparts),28        subparts=subparts,29    )30 31 32def parse_email(raw: str) -> EmailModel:33    try:34        msg = email.message_from_string(raw)35        return get_email_model(msg)36    except Exception as e:37        _logger.error(f"Could not parse email: {e}")38        return EmailModel(39            headers={},40            body="could not parse email",41            content_type="text/plain",42            size=len("could not parse email"),43            subparts=[],44        )45 46 47def generate_email_model(item: dict) -> EmailModel:48    email_json = json.loads(item["raw"])49    message = MIMEMultipart()50    if email_json.get("version") == "v2":51        message['From'] = f'{email_json["from_name"]} <{item["address"]}>' if email_json.get("from_name") else item["address"]52        message['To'] = f'{email_json["to_name"]} <{email_json["to_mail"]}>' if email_json.get("to_name") else email_json["to_mail"]53        message.attach(MIMEText(54            email_json["content"],55            "html" if email_json.get("is_html") else "plain"56        ))57    else:58        message['From'] = f'{email_json["from"]["name"]} <{email_json["from"]["email"]}>'59        message['To'] = ", ".join(60            [f"{to['name']} <{to['email']}>" for to in email_json["personalizations"][0]["to"]])61        message.attach(MIMEText(62            email_json["content"][0]["value"],63            "html" if "html" in email_json["content"][0]["type"] else "plain"64        ))65    message['Subject'] = email_json["subject"]66    message["Date"] = datetime.datetime.strptime(67        item["created_at"], "%Y-%m-%d %H:%M:%S"68    ).strftime("%a, %d %b %Y %H:%M:%S +0000")69    return parse_email(message.as_string())70