wim/olm2eml
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

ruff format

Browse Source
This commit is contained in:
w.pomp
2026-07-02 16:26:03 +02:00
parent cfc00fe355
commit d0af8f9654
1 changed files with 100 additions and 91 deletions
Download Patch File Download Diff File Expand all files Collapse all files
olm_xml_to_eml.py
+100 -91
View File
@@ -18,17 +18,17 @@ from email.mime.text import MIMEText
from email.utils import formataddr, formatdate
from pathlib import Path
EMAILS_ROOT_ATTR = '{http://www.w3.org/XML/1998/namespace}space'
ADDR_ADDRESS = 'OPFContactEmailAddressAddress'
ADDR_NAME = 'OPFContactEmailAddressName'
ADDR_TYPE = 'OPFContactEmailAddressType'
EMAILS_ROOT_ATTR = "{http://www.w3.org/XML/1998/namespace}space"
ADDR_ADDRESS = "OPFContactEmailAddressAddress"
ADDR_NAME = "OPFContactEmailAddressName"
ADDR_TYPE = "OPFContactEmailAddressType"
ATTACH_EXT = 'OPFAttachmentContentExtension'
ATTACH_SIZE = 'OPFAttachmentContentFileSize'
ATTACH_CID = 'OPFAttachmentContentID'
ATTACH_TYPE = 'OPFAttachmentContentType'
ATTACH_NAME = 'OPFAttachmentName'
ATTACH_URL = 'OPFAttachmentURL'
ATTACH_EXT = "OPFAttachmentContentExtension"
ATTACH_SIZE = "OPFAttachmentContentFileSize"
ATTACH_CID = "OPFAttachmentContentID"
ATTACH_TYPE = "OPFAttachmentContentType"
ATTACH_NAME = "OPFAttachmentName"
ATTACH_URL = "OPFAttachmentURL"
def parse_addresses(parent_elem):
@@ -36,9 +36,9 @@ def parse_addresses(parent_elem):
if parent_elem is None:
return addresses
for child in parent_elem:
if child.tag == 'emailAddress':
addr = child.get(ADDR_ADDRESS, '')
name = child.get(ADDR_NAME, '')
if child.tag == "emailAddress":
addr = child.get(ADDR_ADDRESS, "")
name = child.get(ADDR_NAME, "")
addresses.append((name, addr))
return addresses
@@ -50,7 +50,7 @@ def format_address_list(addresses):
parts.append(formataddr((name, addr)))
elif addr:
parts.append(addr)
return ', '.join(parts) if parts else ''
return ", ".join(parts) if parts else ""
def parse_attachment_list(parent_elem):
@@ -58,14 +58,14 @@ def parse_attachment_list(parent_elem):
if parent_elem is None:
return attachments
for child in parent_elem:
if child.tag == 'messageAttachment':
if child.tag == "messageAttachment":
att = {
'name': child.get(ATTACH_NAME, 'attachment'),
'content_type': child.get(ATTACH_TYPE, 'application/octet-stream'),
'extension': child.get(ATTACH_EXT, ''),
'content_id': child.get(ATTACH_CID, ''),
'url': child.get(ATTACH_URL, ''),
'size': child.get(ATTACH_SIZE, '0'),
"name": child.get(ATTACH_NAME, "attachment"),
"content_type": child.get(ATTACH_TYPE, "application/octet-stream"),
"extension": child.get(ATTACH_EXT, ""),
"content_id": child.get(ATTACH_CID, ""),
"url": child.get(ATTACH_URL, ""),
"size": child.get(ATTACH_SIZE, "0"),
}
attachments.append(att)
return attachments
@@ -74,7 +74,7 @@ def parse_attachment_list(parent_elem):
def find_email_xmls(root_dir):
xml_files = []
root_path = Path(root_dir)
for path in root_path.rglob('message_*.xml'):
for path in root_path.rglob("message_*.xml"):
xml_files.append(path)
return sorted(xml_files)
@@ -87,111 +87,111 @@ def get_email_element(xml_path):
def text_of(elem):
if elem is None:
return ''
return (elem.text or '').strip()
return ""
return (elem.text or "").strip()
def convert_email_xml_to_eml(xml_path, emails_root):
email = get_email_element(xml_path)
subject = text_of(email.find('OPFMessageCopySubject'))
subject = text_of(email.find("OPFMessageCopySubject"))
from_addrs = parse_addresses(email.find('OPFMessageCopyFromAddresses'))
to_addrs = parse_addresses(email.find('OPFMessageCopyToAddresses'))
cc_addrs = parse_addresses(email.find('OPFMessageCopyCCAddresses'))
reply_to_addrs = parse_addresses(email.find('OPFMessageCopyReplyToAddresses'))
from_addrs = parse_addresses(email.find("OPFMessageCopyFromAddresses"))
to_addrs = parse_addresses(email.find("OPFMessageCopyToAddresses"))
cc_addrs = parse_addresses(email.find("OPFMessageCopyCCAddresses"))
reply_to_addrs = parse_addresses(email.find("OPFMessageCopyReplyToAddresses"))
sender_addr = text_of(email.find('OPFMessageCopySenderAddress'))
display_to = text_of(email.find('OPFMessageCopyDisplayTo'))
sender_addr = text_of(email.find("OPFMessageCopySenderAddress"))
display_to = text_of(email.find("OPFMessageCopyDisplayTo"))
sent_time = text_of(email.find('OPFMessageCopySentTime'))
received_time = text_of(email.find('OPFMessageCopyReceivedTime'))
mod_time = text_of(email.find('OPFMessageCopyModDate'))
server_time = text_of(email.find('ExchangeServerLastModifiedTime'))
sent_time = text_of(email.find("OPFMessageCopySentTime"))
received_time = text_of(email.find("OPFMessageCopyReceivedTime"))
mod_time = text_of(email.find("OPFMessageCopyModDate"))
server_time = text_of(email.find("ExchangeServerLastModifiedTime"))
message_id = text_of(email.find('OPFMessageCopyMessageID'))
in_reply_to = text_of(email.find('OPFMessageCopyInReplyTo'))
references = text_of(email.find('OPFMessageCopyReferences'))
thread_topic = text_of(email.find('OPFMessageCopyThreadTopic'))
conversation_id = text_of(email.find('OPFMessageCopyExchangeConversationId'))
message_id = text_of(email.find("OPFMessageCopyMessageID"))
in_reply_to = text_of(email.find("OPFMessageCopyInReplyTo"))
references = text_of(email.find("OPFMessageCopyReferences"))
thread_topic = text_of(email.find("OPFMessageCopyThreadTopic"))
conversation_id = text_of(email.find("OPFMessageCopyExchangeConversationId"))
html_body = text_of(email.find('OPFMessageCopyHTMLBody'))
body = text_of(email.find('OPFMessageCopyBody'))
html_body = text_of(email.find("OPFMessageCopyHTMLBody"))
body = text_of(email.find("OPFMessageCopyBody"))
attachments = parse_attachment_list(email.find('OPFMessageCopyAttachmentList'))
attachments = parse_attachment_list(email.find("OPFMessageCopyAttachmentList"))
mime_msg = MIMEMultipart('mixed')
mime_msg = MIMEMultipart("mixed")
mime_msg['Subject'] = subject
mime_msg["Subject"] = subject
from_str = format_address_list(from_addrs)
if from_str:
mime_msg['From'] = from_str
mime_msg["From"] = from_str
elif sender_addr:
mime_msg['From'] = sender_addr
mime_msg["From"] = sender_addr
else:
mime_msg['From'] = 'unknown@unknown.com'
mime_msg["From"] = "unknown@unknown.com"
if from_addrs and display_to:
mime_msg['To'] = display_to
mime_msg["To"] = display_to
else:
to_str = format_address_list(to_addrs)
if to_str:
mime_msg['To'] = to_str
mime_msg["To"] = to_str
else:
mime_msg['To'] = 'undisclosed-recipients:;'
mime_msg["To"] = "undisclosed-recipients:;"
cc_str = format_address_list(cc_addrs)
if cc_str:
mime_msg['Cc'] = cc_str
mime_msg["Cc"] = cc_str
reply_to_str = format_address_list(reply_to_addrs)
if reply_to_str:
mime_msg['Reply-To'] = reply_to_str
mime_msg["Reply-To"] = reply_to_str
date_str = sent_time or received_time or mod_time or server_time
if date_str:
mime_msg['Date'] = date_str
mime_msg["Date"] = date_str
else:
mime_msg['Date'] = formatdate(localtime=True)
mime_msg["Date"] = formatdate(localtime=True)
if message_id:
mime_msg['Message-ID'] = message_id
mime_msg["Message-ID"] = message_id
if in_reply_to:
mime_msg['In-Reply-To'] = in_reply_to
mime_msg["In-Reply-To"] = in_reply_to
if references:
mime_msg['References'] = references
mime_msg["References"] = references
if thread_topic:
mime_msg['Thread-Topic'] = thread_topic
mime_msg["Thread-Topic"] = thread_topic
if conversation_id:
mime_msg['Thread-Index'] = conversation_id
mime_msg["Thread-Index"] = conversation_id
mime_msg['MIME-Version'] = '1.0'
mime_msg["MIME-Version"] = "1.0"
has_html = bool(html_body)
has_text = bool(body)
if has_html or has_text:
plain_text = strip_html(body) if body else ''
plain_text = strip_html(body) if body else ""
if has_html:
body_part = MIMEMultipart('alternative')
body_part = MIMEMultipart("alternative")
if plain_text:
body_part.attach(MIMEText(plain_text, 'plain', 'utf-8'))
body_part.attach(MIMEText(html_body, 'html', 'utf-8'))
body_part.attach(MIMEText(plain_text, "plain", "utf-8"))
body_part.attach(MIMEText(html_body, "html", "utf-8"))
mime_msg.attach(body_part)
elif plain_text:
mime_msg.attach(MIMEText(plain_text, 'plain', 'utf-8'))
mime_msg.attach(MIMEText(plain_text, "plain", "utf-8"))
else:
mime_msg.attach(MIMEText('', 'plain', 'utf-8'))
mime_msg.attach(MIMEText("", "plain", "utf-8"))
else:
mime_msg.attach(MIMEText('', 'plain', 'utf-8'))
mime_msg.attach(MIMEText("", "plain", "utf-8"))
for att in attachments:
att_url = att['url']
att_url = att["url"]
if not att_url:
continue
att_path = Path(emails_root) / att_url
@@ -199,26 +199,26 @@ def convert_email_xml_to_eml(xml_path, emails_root):
continue
try:
data = att_path.read_bytes()
content_type = att['content_type'] or 'application/octet-stream'
if '/' in content_type:
main_type, sub_type = content_type.split('/', 1)
content_type = att["content_type"] or "application/octet-stream"
if "/" in content_type:
main_type, sub_type = content_type.split("/", 1)
else:
main_type, sub_type = 'application', 'octet-stream'
main_type, sub_type = "application", "octet-stream"
if main_type == 'image':
if main_type == "image":
part = MIMEImage(data, _subtype=sub_type)
else:
part = MIMEBase(main_type, sub_type)
part.set_payload(data)
encoders.encode_base64(part)
filename = att['name']
cid = att['content_id']
filename = att["name"]
cid = att["content_id"]
if cid:
part.add_header('Content-ID', f'<{cid}>')
part.add_header('Content-Disposition', 'inline', filename=filename)
part.add_header("Content-ID", f"<{cid}>")
part.add_header("Content-Disposition", "inline", filename=filename)
else:
part.add_header('Content-Disposition', 'attachment', filename=filename)
part.add_header("Content-Disposition", "attachment", filename=filename)
mime_msg.attach(part)
except Exception as e:
@@ -228,10 +228,10 @@ def convert_email_xml_to_eml(xml_path, emails_root):
def strip_html(html_text):
text = re.sub(r'<style[^>]*>.*?</style>', '', html_text, flags=re.DOTALL)
text = re.sub(r'<script[^>]*>.*?</script>', '', text, flags=re.DOTALL)
text = re.sub(r'<[^>]+>', ' ', text)
text = re.sub(r'\s+', ' ', text)
text = re.sub(r"<style[^>]*>.*?</style>", "", html_text, flags=re.DOTALL)
text = re.sub(r"<script[^>]*>.*?</script>", "", text, flags=re.DOTALL)
text = re.sub(r"<[^>]+>", " ", text)
text = re.sub(r"\s+", " ", text)
return text.strip()
@@ -246,12 +246,19 @@ def build_output_path(xml_path, emails_root, output_dir, preserve_structure):
def main():
parser = argparse.ArgumentParser(description='Convert OLM-extracted XML emails to EML format')
parser.add_argument('input_dir', help='Root directory of the extracted OLM emails')
parser.add_argument('output_dir', help='Output directory for EML files')
parser.add_argument('--preserve-structure', action='store_true',
help='Preserve folder structure in output (mirrors account/folder hierarchy)')
parser.add_argument('--overwrite', action='store_true', help='Overwrite existing EML files')
parser = argparse.ArgumentParser(
description="Convert OLM-extracted XML emails to EML format"
)
parser.add_argument("input_dir", help="Root directory of the extracted OLM emails")
parser.add_argument("output_dir", help="Output directory for EML files")
parser.add_argument(
"--preserve-structure",
action="store_true",
help="Preserve folder structure in output (mirrors account/folder hierarchy)",
)
parser.add_argument(
"--overwrite", action="store_true", help="Overwrite existing EML files"
)
args = parser.parse_args()
@@ -276,7 +283,9 @@ def main():
errors = 0
for xml_path in xml_files:
out_path = build_output_path(xml_path, input_dir, output_dir, args.preserve_structure)
out_path = build_output_path(
xml_path, input_dir, output_dir, args.preserve_structure
)
if out_path.exists() and not args.overwrite:
skipped += 1
@@ -293,5 +302,5 @@ def main():
print(f"\nDone. {success} converted, {skipped} skipped, {errors} errors.")
if __name__ == '__main__':
if __name__ == "__main__":
main()