diff --git a/olm_xml_to_eml.py b/olm_xml_to_eml.py index dcd57ee..ab9ee44 100644 --- a/olm_xml_to_eml.py +++ b/olm_xml_to_eml.py @@ -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']*>.*?', '', html_text, flags=re.DOTALL) - text = re.sub(r']*>.*?', '', text, flags=re.DOTALL) - text = re.sub(r'<[^>]+>', ' ', text) - text = re.sub(r'\s+', ' ', 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() @@ -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()