Spaces:
Runtime error
Runtime error
File size: 6,927 Bytes
fe6855c | 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 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 | import os
import email
from email import policy
from email.parser import BytesParser
import re
from typing import Dict, List, Optional
import dateutil
from fastapi import HTTPException
def read_emails_from_directory(directory: str) -> List[str]:
"""Reads email files from a given directory"""
emails = []
for file in os.listdir(directory):
if file.endswith(".eml") or file.endswith(".msg") or file.endswith(".txt"):
emails.append(os.path.join(directory, file))
return emails
def parse_email(file_path: str) -> dict:
"""Parses email file and extracts metadata, body, and attachments."""
with open(file_path, "rb") as f:
msg = BytesParser(policy=policy.default).parse(f)
body = ""
attachments = []
for part in msg.walk():
if part.get_content_type() == "text/plain":
body += part.get_payload(decode=True).decode("utf-8", errors="ignore")
elif part.get_filename():
attachments.append(part.get_filename())
return {
"sender": msg["From"],
"subject": msg["Subject"],
"date": msg["Date"],
"body": body,
"attachments": attachments
}
def parse_email_bytes(file_content: bytes, filename: str) -> Optional[Dict]:
"""Parses email bytes, extracts attachments, and handles email chains."""
try:
msg = email.message_from_bytes(file_content)
sender = msg["from"]
subject = msg["subject"]
body = ""
attachments = []
email_chain_text = ""
if msg.is_multipart():
for part in msg.walk():
content_type = part.get_content_type()
content_disposition = str(part.get("Content-Disposition"))
if content_type == "text/plain" and "attachment" not in content_disposition:
body += part.get_payload(decode=True).decode()
elif content_type == "text/html" and "attachment" not in content_disposition:
html = part.get_payload(decode=True).decode()
body += ''.join(c if ord(c) < 128 else ' ' for c in html.replace("<br>", "\n").replace("<p>", "\n").replace("</p>","\n").replace("<div>","\n").replace("</div>","\n").replace("<span>"," ").replace("</span>", " "))
elif "attachment" in content_disposition:
attachment_data = part.get_payload(decode=True)
attachments.append({
"filename": part.get_filename(),
"content": attachment_data,
})
email_chain_text = process_email_chain(msg) #only process if it is multipart.
else:
body = msg.get_payload(decode=True).decode()
# Handle email chains (add logic based on your needs)
# email_chain_text = process_email_chain(msg) #see full code in previous response.
return {
"sender": sender if sender else "Unknown Sender", #added default value.
"subject": subject if subject else "No Subject", #added default value.
"body": body,
"attachments": attachments,
"email_chain_text": email_chain_text
}
except Exception as e:
print(f"Error parsing email: {e}")
return None
def process_email_chain(email_message):
"""
Detects and processes email chains, extracting text from each email.
"""
try:
if isinstance(email_message, str):
msg = email.message_from_string(email_message)
else:
msg = email_message
chain = []
full_text = ""
# Check for nested email headers (From:, Date:) or quoted text
if has_nested_emails(msg):
# Parse the email and extract the chain
chain = extract_email_chain(msg)
for email_part in chain:
full_text += extract_text_from_email(email_part) + "\n"
else:
# No nested emails, just extract the text from the current email
full_text = extract_text_from_email(msg)
return full_text
except Exception as e:
print(f"Error processing email chain: {e}")
return extract_text_from_email(email_message) #default to just the email.
def has_nested_emails(msg):
"""
Detects if an email contains nested emails based on headers or quoted text.
"""
body = get_email_body_text(msg)
# Check for multiple 'From:' and 'Date:' headers in the body
if body:
if len(re.findall(r"^From:.*", body, re.MULTILINE)) > 1 or \
len(re.findall(r"^Date:.*", body, re.MULTILINE)) > 1 or \
len(re.findall(r"^>.*", body, re.MULTILINE)) > 5: #arbitrary number of quoted lines.
return True
return False
def extract_email_chain(msg):
"""
Extracts the individual emails from a nested email chain.
"""
chain = []
#this is a very basic attempt at parsing the email chain. It is not perfect, and will need to be improved based on specific email formatting.
body = get_email_body_text(msg)
if not body:
return [msg] #if no body, then return the message.
#basic email chain splitting.
emails = re.split(r"(^From:.*?\n^Date:.*?(\n\n|\r\n\r\n))", body, flags=re.MULTILINE | re.DOTALL)
if len(emails) > 1:
for i in range(1, len(emails), 2):
email_part = emails[i] + emails[i+1]
try:
chain.append(email.message_from_string(email_part))
except Exception as e:
print(f"Error parsing email part: {e}")
pass #if error, skip the email part.
if not chain:
chain = [msg] #if no chain, then return the original email.
return chain
def extract_text_from_email(email_message):
try:
text = ""
for part in email_message.walk():
if part.get_content_type() == "text/plain":
text += part.get_payload(decode=True).decode()
elif part.get_content_type() == "text/html":
html = part.get_payload(decode=True).decode()
text += ''.join(c if ord(c) < 128 else ' ' for c in html.replace("<br>", "\n").replace("<p>", "\n").replace("</p>","\n").replace("<div>","\n").replace("</div>","\n").replace("<span>"," ").replace("</span>", " "))
return text
except Exception as e:
print(f"Error parsing email: {e}")
return ""
def get_email_body_text(msg):
"""
Gets the email body text.
"""
body = ""
if msg.is_multipart():
for part in msg.walk():
if part.get_content_type() == "text/plain":
body += part.get_payload(decode=True).decode()
elif part.get_content_type() == "text/html":
body += part.get_payload(decode=True).decode()
else:
body = msg.get_payload(decode=True).decode()
return body |