forked from futrime/thuasta-email-submission-system
-
Notifications
You must be signed in to change notification settings - Fork 0
/
Copy pathemail_message_utils.py
84 lines (57 loc) · 2 KB
/
email_message_utils.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
import email.headerregistry
import email.message
import re
from typing import Optional
def calc_submission_id_from_message_id(message_id: str) -> str:
"""Calculate submission ID from a message.
Args:
message: Message to calculate submission ID from.
Returns:
Submission ID.
"""
return message_id
def calc_message_id_from_submission_id(submission_id: str) -> str:
"""Calculate message ID from a submission ID.
Args:
submission_id: Submission ID to calculate message ID from.
Returns:
Message ID.
"""
return submission_id
def extract_body_content(message: email.message.EmailMessage) -> Optional[str]:
"""Extract body content from a message.
Args:
message: Message to extract body content from.
Returns:
Body content.
"""
submission_body = message.get_body(preferencelist=("html", "plain"))
submission_body_content = (
str(submission_body.get_content()) if submission_body is not None else None
)
return submission_body_content
def extract_first_from_address(message: email.message.EmailMessage) -> Optional[str]:
"""Extract the first from address from a message.
Args:
message: Message to extract from address from.
Returns:
From address.
"""
address_header = message["From"]
if not isinstance(address_header, email.headerregistry.AddressHeader):
return None
addresses = address_header.addresses
return addresses[0].addr_spec if len(addresses) > 0 else None
def extract_submission_id_from_subject(
message: email.message.EmailMessage,
) -> Optional[str]:
"""Extract submission ID from a subject.
Args:
subject: Subject to extract submission ID from.
Returns:
Submission ID.
"""
subject = str(message["Subject"])
submission_matcher = re.compile(r"#(.+)#")
submission_match = submission_matcher.search(subject)
return submission_match.group(1) if submission_match is not None else None