190 lines
7.5 KiB
Python
190 lines
7.5 KiB
Python
"""
|
|
|
|
AUTHOR:
|
|
|
|
Khushal P Soonderji
|
|
|
|
DATE:
|
|
|
|
Tuesday, 10th Dec., 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
|
|
4. StackOverflow: https://stackoverflow.com/questions/17874360/python-how-to-parse-the-body-from-a-raw-email-given-that-raw-email-does-not
|
|
|
|
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 email
|
|
|
|
# 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
|
|
|
|
import time
|
|
|
|
|
|
# *****************************************************************************************************************
|
|
# ***** ****
|
|
# *** MACROS / ONE-TIME INIT ***
|
|
# ***** ****
|
|
# *****************************************************************************************************************
|
|
|
|
|
|
# --- Nothing Yet
|
|
|
|
|
|
# *****************************************************************************************************************
|
|
# ***** ****
|
|
# *** VARIABLES ***
|
|
# ***** ****
|
|
# *****************************************************************************************************************
|
|
|
|
|
|
# --- Nothing Yet
|
|
|
|
|
|
# *****************************************************************************************************************
|
|
# ***** ****
|
|
# *** FUNCTIONS ***
|
|
# ***** ****
|
|
# *****************************************************************************************************************
|
|
|
|
|
|
def parse_part(
|
|
# part: email.message.Message
|
|
part
|
|
) -> Dict[str, Any]:
|
|
|
|
part_json = {
|
|
"contentType": part.get_content_type(),
|
|
"contentMainType": part.get_content_maintype(),
|
|
"contentSubType": part.get_content_subtype(),
|
|
"contentCharset": part.get_content_charset(),
|
|
"contentDisposition": part.get_content_disposition(),
|
|
"filename": part.get_filename(),
|
|
"payload": part.get_payload(decode = True)
|
|
}
|
|
if part_json["contentCharset"] is not None:
|
|
part_json["payload"] = part_json["payload"].decode(part_json["contentCharset"])
|
|
|
|
print("PARSED PART:", json.to_string(part_json, default = str))
|
|
|
|
return part_json
|
|
|
|
|
|
# ---------------------------------------------------------------------------------------------------------------------
|
|
|
|
|
|
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 = email.message_from_string(raw_mail)
|
|
else: parsed_mail = email.message_from_bytes(raw_mail)
|
|
|
|
# Make variables:
|
|
parts = []
|
|
attachments = []
|
|
|
|
# Iterate through each part of the mail for multipart mails:
|
|
if parsed_mail.is_multipart():
|
|
for part in parsed_mail.walk():
|
|
parse_part(part)
|
|
|
|
# # Start by extracting basic details about the part:
|
|
# part_json = {
|
|
# "contentType": part.get("Content-Type"),
|
|
# "contentDisposition": part.get("Content-Disposition"),
|
|
# "contentTransferEncoding": part.get("Content-Transfer-Encoding"),
|
|
# "contentId": part.get("Content-ID"),
|
|
# # "part": part.get_payload(decode=True)
|
|
# }
|
|
#
|
|
# # Now check if this part is either plain or HTML text:
|
|
# content_type = part_json["contentType"].lower()
|
|
# if (
|
|
# content_type.find("text/html") >= 0 or
|
|
# content_type.find("text/plain") >= 0
|
|
# ): parts.append(parse_part(part))
|
|
#
|
|
# # print("MULTIPART PART:", json.to_string(part_json, default = str))
|
|
# # if (
|
|
# # isinstance(part_json["part"], bytes) and
|
|
# # part_json[""].find()
|
|
# # ):
|
|
# # time.sleep(1.0)
|
|
# # try: part_json["part"] = part_json["part"].decode()
|
|
# # except: pass
|
|
# # print("MULTIPART PART:", json.to_string(part_json))
|
|
# print("\n\n---\n\n")
|
|
# # if part_json[]
|
|
|
|
# When the mails are not multipart, just plaintext:
|
|
else: print("PLAINTEXT PART:", parsed_mail.get_payload())
|
|
|
|
|
|
# *****************************************************************************************************************
|
|
# ***** ****
|
|
# *** MAIN PROGRAM ***
|
|
# ***** ****
|
|
# *****************************************************************************************************************
|
|
|
|
|
|
if __name__ == "__main__":
|
|
|
|
from utils_v2.system import files
|
|
|
|
mail_string_raw = files.read_file(r"/home/developer/Downloads/raw_mail_test.txt")
|
|
parse_results = parse(mail_string_raw)
|
|
|
|
# print(json.to_string(parse_results, default = str))
|