""" AUTHOR: Khushal P Soonderji DATE: Tuesday, 26th Nov., 2024 OBJECTIVE: To parse raw mail bodies and give a structure that is suitable for storing in No-SQL databases like MongoDB. The raw mail's text is expected to be compliant with standard defined in RFC 5322, RFC 2045, and maybe a few more. REFERENCES: 1. GitHub: https://github.com/SpamScope/mail-parser 2. RFC 5322: https://datatracker.ietf.org/doc/html/rfc5322 3. RFC 2045: https://datatracker.ietf.org/doc/html/rfc2045 DOWNLOADS: N/A """ # ***************************************************************************************************************** # ***** **** # *** IMPORT *** # ***** **** # ***************************************************************************************************************** # To make sibling directories accessible for imports: import sys sys.path.append(".") sys.path.append("..") # System-level activities: import io # My utils: from utils_v2.string import json from utils_v2.string import regex from utils_v2.date_time import date_time # To work with mails: import mailparser # To parse the HTML content in the mail: from bs4 import BeautifulSoup # To work with datatypes: from typing import Any, Dict # To work with base-64 encoding: import base64 # ***************************************************************************************************************** # ***** **** # *** MACROS / ONE-TIME INIT *** # ***** **** # ***************************************************************************************************************** # --- Nothing Yet # ***************************************************************************************************************** # ***** **** # *** VARIABLES *** # ***** **** # ***************************************************************************************************************** # --- Nothing Yet # ***************************************************************************************************************** # ***** **** # *** FUNCTIONS *** # ***** **** # ***************************************************************************************************************** def parse(raw_mail: str | bytes) -> Dict[str, Any]: """ To parse the raw mail text to a usable JSON that can even be stored on a No-SQL database like MongoDB. DOCUMENTATION: 1. GitHub: https://github.com/SpamScope/mail-parser 2. RFC 5322: https://datatracker.ietf.org/doc/html/rfc5322 3. RFC 2045: https://datatracker.ietf.org/doc/html/rfc2045 :param raw_mail: The raw mail body that adheres to RFC 5322 and RFC 2045 (among others). :return: The parsed JSON format (dict) of the mail. """ # Parse the raw format: if isinstance(raw_mail, str): parsed_mail = mailparser.parse_from_string(raw_mail) else: parsed_mail = mailparser.parse_from_bytes(raw_mail) # Format the attachments: message_attachments = [ { "filename": attachment["filename"], "type": attachment["mail_content_type"], "cid": regex.find_first(text = attachment["content-id"], pattern = r"(?<=<).*(?=>)"), "rawCid": attachment["content-id"], "contentDisposition": (cd := attachment["content-disposition"]), "isInline": True if cd.lower().find("inline") >= 0 else False, "charset": attachment["charset"], "contentTransferEncoding": attachment["content_transfer_encoding"], "payload": attachment["payload"] } for attachment in parsed_mail.attachments ] # Figure out which entity (text and HTML) came in which sequence. # The library doesn't give us any sequence info so we do some custom string processing here to figure out the order # in which to render the contents of the page. parts = [ { "no": None, "offset": max( parsed_mail.message_as_string.find(t), parsed_mail.message_as_string.find(base64.b64encode(t.encode()).decode()) ), "type": "text/plain", "data": t } for t in parsed_mail.text_plain ] parts = parts + [ { "partNo": None, "offset": max( parsed_mail.message_as_string.find(h), parsed_mail.message_as_string.find(base64.b64encode(h.encode()).decode()) ), "type": "text/html", "data": h } for h in parsed_mail.text_html ] parts = sorted(parts, key = lambda x: x["offset"]) for i, p in enumerate(parts): p["no"] = i # Get the unformatted text from everything in the mail: unformatted_text = [] for p in parts: if p["type"] == "text/html": html_parser = BeautifulSoup(p["data"], "html.parser") unformatted_text.append(html_parser.get_text()) else: unformatted_text.append(p["data"]) # Put everything together: return { "ts": date_time.to_timezone(parsed_mail.date, timezone = date_time.TIMEZONE_UTC), "readTs": date_time.get_current_utc_date_time(as_string = False), "headers": parsed_mail.headers, "from": [{"name": _[0] or _[1], "email": _[1]} for _ in parsed_mail.headers["From"]], "to": [{"name": _[0] or _[1], "email": _[1]} for _ in parsed_mail.headers["To"]], "cc": [{"name": _[0] or _[1], "email": _[1]} for _ in parsed_mail.headers.get("Cc", [])], "bcc": [{"name": _[0] or _[1], "email": _[1]} for _ in parsed_mail.headers.get("Bcc", [])], "subject": parsed_mail.headers["Subject"], "text": parsed_mail.text_plain, "html": parsed_mail.text_html, "parts": parts, "unformattedText": "\n".join(unformatted_text), "attachments": message_attachments, "isInbox": None } # ***************************************************************************************************************** # ***** **** # *** MAIN PROGRAM *** # ***** **** # ***************************************************************************************************************** if __name__ == "__main__": pass