khushalps
cc3e160339
git-subtree-dir: utils_v2 git-subtree-split: 758ed9a0ab460f322a8691337bad42c7722672df
236 lines
9.1 KiB
Python
236 lines
9.1 KiB
Python
"""
|
|
|
|
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 various encodings:
|
|
import base64
|
|
import quopri
|
|
|
|
|
|
# *****************************************************************************************************************
|
|
# ***** ****
|
|
# *** MACROS / ONE-TIME INIT ***
|
|
# ***** ****
|
|
# *****************************************************************************************************************
|
|
|
|
|
|
# --- Nothing Yet
|
|
|
|
|
|
# *****************************************************************************************************************
|
|
# ***** ****
|
|
# *** VARIABLES ***
|
|
# ***** ****
|
|
# *****************************************************************************************************************
|
|
|
|
|
|
# --- Nothing Yet
|
|
|
|
|
|
# *****************************************************************************************************************
|
|
# ***** ****
|
|
# *** FUNCTIONS ***
|
|
# ***** ****
|
|
# *****************************************************************************************************************
|
|
|
|
|
|
def from_quoted_printable(text: str) -> str:
|
|
|
|
decoded_text = ""
|
|
decoded_bytes = quopri.decodestring(text)
|
|
for encoding in ["utf-8", "utf-16", "utf-32", "latin1"]:
|
|
try: text = decoded_bytes.decode(encoding)
|
|
except: text = ""
|
|
if text.find("From") >= 0:
|
|
decoded_text = text
|
|
break
|
|
return decoded_text
|
|
|
|
|
|
# ---------------------------------------------------------------------------------------------------------------------
|
|
|
|
|
|
def find_in_raw_mail(
|
|
raw_mail: str,
|
|
text: str
|
|
) -> int:
|
|
|
|
# We first treat it as un-encoded text:
|
|
offset = raw_mail.find(text)
|
|
if offset >= 0: return offset
|
|
|
|
# Then we try Base64 encoding:
|
|
offset = raw_mail.find(base64.b64encode(text.encode("utf-8")).decode("utf-8"))
|
|
if offset >= 0: return offset
|
|
|
|
# then we try Quoted-Printable encoding:
|
|
offset = from_quoted_printable(raw_mail).find(text)
|
|
print("MAIL:")
|
|
print(raw_mail)
|
|
print("\n\n\n---\n\n\n")
|
|
print("TEXT:")
|
|
print(quopri.encodestring(text.encode("utf-8")).decode("utf-8"))
|
|
if offset >= 0: return offset
|
|
|
|
# Done here, even if nothing worked:
|
|
return offset
|
|
|
|
|
|
# ---------------------------------------------------------------------------------------------------------------------
|
|
|
|
|
|
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)
|
|
|
|
# print(parsed_mail.mail_json)
|
|
# return
|
|
|
|
# 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
|
|
]
|
|
|
|
# print("PRINTING PARTS")
|
|
# print("LIBRARY PARTS:", type(parsed_mail))
|
|
|
|
# # 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 = [
|
|
# # {
|
|
# # "partNo": 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": find_in_raw_mail(raw_mail = parsed_mail.message_as_string, text = h),
|
|
# "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["partNo"] = i
|
|
|
|
# Get the unformatted text from everything in the mail:
|
|
unformatted_text = []
|
|
for p in parsed_mail.text_html:
|
|
html_parser = BeautifulSoup(p, "html.parser")
|
|
unformatted_text.append(html_parser.get_text())
|
|
|
|
# Put everything together:
|
|
return {
|
|
"ts": date_time.to_timezone(parsed_mail.date, timezone = date_time.TIMEZONE_UTC),
|
|
"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__":
|
|
|
|
from utils_v2.system import files
|
|
|
|
mail_string_raw = files.read_file(r"/home/developer/Downloads/raw_mail.txt")
|
|
parse_results = parse(mail_string_raw)
|
|
|
|
print(json.to_string(parse_results, default = str))
|