#!/usr/bin/env python3 """ Convert emails extracted from an OLM file (as XML) to standard .eml format. Usage: python olm_xml_to_eml.py /path/to/emails/root /path/to/output_dir """ import argparse import re import sys import xml.etree.ElementTree as ET from email import encoders from email.mime.base import MIMEBase from email.mime.image import MIMEImage from email.mime.multipart import MIMEMultipart 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" ATTACH_EXT = "OPFAttachmentContentExtension" ATTACH_SIZE = "OPFAttachmentContentFileSize" ATTACH_CID = "OPFAttachmentContentID" ATTACH_TYPE = "OPFAttachmentContentType" ATTACH_NAME = "OPFAttachmentName" ATTACH_URL = "OPFAttachmentURL" def parse_addresses(parent_elem): addresses = [] 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, "") addresses.append((name, addr)) return addresses def format_address_list(addresses): parts = [] for name, addr in addresses: if name and addr: parts.append(formataddr((name, addr))) elif addr: parts.append(addr) return ", ".join(parts) if parts else "" def parse_attachment_list(parent_elem): attachments = [] if parent_elem is None: return attachments for child in parent_elem: 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"), } attachments.append(att) return attachments def find_email_xmls(root_dir): xml_files = [] root_path = Path(root_dir) for path in root_path.rglob("message_*.xml"): xml_files.append(path) return sorted(xml_files) def get_email_element(xml_path): tree = ET.parse(str(xml_path)) root = tree.getroot() return root[0] def text_of(elem): if elem is None: 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")) 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")) 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")) html_body = text_of(email.find("OPFMessageCopyHTMLBody")) body = text_of(email.find("OPFMessageCopyBody")) attachments = parse_attachment_list(email.find("OPFMessageCopyAttachmentList")) mime_msg = MIMEMultipart("mixed") mime_msg["Subject"] = subject from_str = format_address_list(from_addrs) if from_str: mime_msg["From"] = from_str elif sender_addr: mime_msg["From"] = sender_addr else: mime_msg["From"] = "unknown@unknown.com" if from_addrs and display_to: mime_msg["To"] = display_to else: to_str = format_address_list(to_addrs) if to_str: mime_msg["To"] = to_str else: mime_msg["To"] = "undisclosed-recipients:;" cc_str = format_address_list(cc_addrs) if 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 date_str = sent_time or received_time or mod_time or server_time if date_str: mime_msg["Date"] = date_str else: mime_msg["Date"] = formatdate(localtime=True) if message_id: mime_msg["Message-ID"] = message_id if in_reply_to: mime_msg["In-Reply-To"] = in_reply_to if references: mime_msg["References"] = references if thread_topic: mime_msg["Thread-Topic"] = thread_topic if conversation_id: mime_msg["Thread-Index"] = conversation_id 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 "" if has_html: 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")) mime_msg.attach(body_part) elif plain_text: mime_msg.attach(MIMEText(plain_text, "plain", "utf-8")) else: mime_msg.attach(MIMEText("", "plain", "utf-8")) else: mime_msg.attach(MIMEText("", "plain", "utf-8")) for att in attachments: att_url = att["url"] if not att_url: continue att_path = Path(emails_root) / att_url if not att_path.exists(): 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) else: main_type, sub_type = "application", "octet-stream" 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"] if cid: part.add_header("Content-ID", f"<{cid}>") part.add_header("Content-Disposition", "inline", filename=filename) else: part.add_header("Content-Disposition", "attachment", filename=filename) mime_msg.attach(part) except Exception as e: print(f" Warning: failed to attach {att_url}: {e}", file=sys.stderr) return mime_msg.as_bytes() def strip_html(html_text): text = re.sub(r"]*>.*?", "", html_text, flags=re.DOTALL) text = re.sub(r"]*>.*?", "", text, flags=re.DOTALL) text = re.sub(r"<[^>]+>", " ", text) text = re.sub(r"\s+", " ", text) return text.strip() def build_output_path(xml_path, emails_root, output_dir, preserve_structure): if preserve_structure: rel = xml_path.parent.relative_to(Path(emails_root).resolve()) out = Path(output_dir) / rel / f"{xml_path.stem}.eml" else: out = Path(output_dir) / f"{xml_path.stem}.eml" out.parent.mkdir(parents=True, exist_ok=True) return out 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" ) args = parser.parse_args() input_dir = Path(args.input_dir).resolve() output_dir = Path(args.output_dir).resolve() if not input_dir.is_dir(): print(f"Error: input directory not found: {input_dir}", file=sys.stderr) sys.exit(1) output_dir.mkdir(parents=True, exist_ok=True) xml_files = find_email_xmls(input_dir) if not xml_files: print(f"No message_*.xml files found in {input_dir}", file=sys.stderr) sys.exit(1) print(f"Found {len(xml_files)} email XML files") success = 0 skipped = 0 errors = 0 for xml_path in xml_files: out_path = build_output_path( xml_path, input_dir, output_dir, args.preserve_structure ) if out_path.exists() and not args.overwrite: skipped += 1 continue try: eml_bytes = convert_email_xml_to_eml(xml_path, input_dir) out_path.write_bytes(eml_bytes) success += 1 except Exception as e: print(f"Error converting {xml_path}: {e}", file=sys.stderr) errors += 1 print(f"\nDone. {success} converted, {skipped} skipped, {errors} errors.") if __name__ == "__main__": main()