khushalps/api_utils_converse_v2
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

(20241209) Many changes to the SMS section.

Browse Source
This commit is contained in:
khushal
2024-12-09 19:38:47 +05:30
parent 725d68f195
commit d10b156c0b
14 changed files with 1406 additions and 250 deletions
Download Patch File Download Diff File Expand all files Collapse all files
utils_v2/mail/mail_parser.py
+87 -36
View File
@@ -55,8 +55,9 @@ from bs4 import BeautifulSoup
# To work with datatypes:
from typing import Any, Dict
# To work with base-64 encoding:
# To work with various encodings:
import base64
import quopri
# *****************************************************************************************************************
@@ -86,6 +87,51 @@ import base64
# *****************************************************************************************************************
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]:
"""
@@ -102,6 +148,9 @@ def parse(raw_mail: str | bytes) -> Dict[str, Any]:
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 = [
{
@@ -117,46 +166,43 @@ def parse(raw_mail: str | bytes) -> Dict[str, Any]:
} 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 = [
{
"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": 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["partNo"] = i
# 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 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"])
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),
"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"]],
@@ -165,7 +211,7 @@ def parse(raw_mail: str | bytes) -> Dict[str, Any]:
"subject": parsed_mail.headers["Subject"],
"text": parsed_mail.text_plain,
"html": parsed_mail.text_html,
"parts": parts,
# "parts": parts,
"unformattedText": "\n".join(unformatted_text),
"attachments": message_attachments,
"isInbox": None
@@ -181,4 +227,9 @@ def parse(raw_mail: str | bytes) -> Dict[str, Any]:
if __name__ == "__main__":
pass
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))