gitignore
This commit is contained in:
w.pomp
@@ -0,0 +1,297 @@
|
||||
#!/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'<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()
|
||||
|
||||
|
||||
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()
|
||||
Reference in New Issue
Block a user