#!/usr/bin/env python
# Copyright (C) 2009-2012 Chris Ball <cjb@laptop.org>
# Gianluca Montecchi <gian@grys.it>
# W. Trevor King <wking@tremily.us>
#
# This file is part of Bugs Everywhere.
#
# Bugs Everywhere is free software: you can redistribute it and/or modify it
# under the terms of the GNU General Public License as published by the Free
# Software Foundation, either version 2 of the License, or (at your option) any
# later version.
#
# Bugs Everywhere is distributed in the hope that it will be useful, but
# WITHOUT ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or
# FITNESS FOR A PARTICULAR PURPOSE. See the GNU General Public License for
# more details.
#
# You should have received a copy of the GNU General Public License along with
# Bugs Everywhere. If not, see <http://www.gnu.org/licenses/>.
"""
Convert an mbox into xml suitable for input into be.
$ be-mbox-to-xml file.mbox | be import-xml -c <ID> -
mbox is a flat-file format, consisting of a series of messages.
Messages begin with a a From_ line, followed by RFC 822 email,
followed by a blank line.
"""
import base64
import email.utils
from libbe.util.encoding import get_output_encoding
from libbe.util.utility import time_to_str
import mailbox # the mailbox people really want an on-disk copy
import optparse
from time import asctime, gmtime, mktime
import types
from xml.sax.saxutils import escape
BREAK = u'--' # signature separator
DEFAULT_ENCODING = get_output_encoding()
KNOWN_IDS = []
def normalize_email_address(address):
"""
Standardize whitespace, etc.
"""
addr = email.utils.formataddr(email.utils.parseaddr(address))
if len(addr) == 0:
return None
return addr
def normalize_RFC_2822_date(date):
"""
Some email clients write non-RFC 2822-compliant date tags like:
Fri, 18 Sep 2009 08:49:02 -0400 (EDT)
with the non-standard (EDT) timezone name. This funtion attempts
to deal with such inconsistencies.
"""
time_tuple = email.utils.parsedate(date)
assert time_tuple != None, \
'unparsable date: "%s"' % date
return time_to_str(mktime(time_tuple))
def strip_footer(body):
body_lines = body.splitlines()
for i,line in enumerate(body_lines):
if line.startswith(BREAK):
break
i += 1 # increment past the current valid line.
return u'\n'.join(body_lines[:i]).strip()
def comment_message_to_xml(message, fields=None):
if fields == None:
fields = {}
new_fields = {}
new_fields[u'alt-id'] = message[u'message-id']
new_fields[u'in-reply-to'] = message[u'in-reply-to']
new_fields[u'author'] = normalize_email_address(message[u'from'])
new_fields[u'date'] = message[u'date']
if new_fields[u'date'] != None:
new_fields[u'date'] = normalize_RFC_2822_date(new_fields[u'date'])
new_fields[u'content-type'] = message.get_content_type()
for k,v in new_fields.items():
if v != None and type(v) != types.UnicodeType:
fields[k] = unicode(v, encoding=DEFAULT_ENCODING)
elif v == None and k in fields:
new_fields[k] = fields[k]
for k,v in fields.items():
if k not in new_fields:
new_fields.k = fields[k]
fields = new_fields
if fields[u'in-reply-to'] == None:
if message[u'references'] != None:
refs = message[u'references'].split()
for ref in refs: # search for a known reference id.
if ref in KNOWN_IDS:
fields[u'in-reply-to'] = ref
break
if fields[u'in-reply-to'] == None and len(refs) > 0:
fields[u'in-reply-to'] = refs[0] # default to the first
else: # check for mutliple in-reply-to references.
refs = fields[u'in-reply-to'].split()
found_ref = False
for ref in refs: # search for a known reference id.
if ref in KNOWN_IDS:
fields[u'in-reply-to'] = ref
found_ref = True
break
if found_ref == False and len(refs) > 0:
fields[u'in-reply-to'] = refs[0] # default to the first
if fields[u'alt-id'] != None:
KNOWN_IDS.append(fields[u'alt-id'])
if message.is_multipart():
ret = []
alt_id = fields[u'alt-id']
from_str = fields[u'author']
date = fields[u'date']
for m in message.walk():
if m == message:
continue
fields[u'author'] = from_str
fields[u'date'] = date
if len(ret) > 0: # we've added one part already
fields.pop(u'alt-id') # don't pass alt-id to other parts
fields[u'in-reply-to'] = alt_id # others respond to first
ret.append(comment_message_to_xml(m, fields))
return u'\n'.join(ret)
charset = message.get_content_charset(DEFAULT_ENCODING).lower()
#assert charset == DEFAULT_ENCODING.lower(), \
# u"Unknown charset: %s" % charset
if message[u'content-transfer-encoding'] == None:
encoding = DEFAULT_ENCODING
else:
encoding = message[u'content-transfer-encoding'].lower()
body = message.get_payload(decode=True) # attempt to decode
assert body != None, "Unable to decode?"
if fields[u'content-type'].startswith(u"text/"):
body = strip_footer(unicode(body, encoding=charset))
else:
body = base64.encode(body)
fields[u'body'] = body
lines = [u"<comment>"]
for tag,body in fields.items():
if body != None:
ebody = escape(body)
lines.append(u" <%s>%s</%s>" % (tag, ebody, tag))
lines.append(u"</comment>")
return u'\n'.join(lines)
def main(argv):
parser = optparse.OptionParser(usage='%prog [options] mailbox')
formats = ['mbox', 'Maildir', 'MH', 'Babyl', 'MMDF']
parser.add_option('-f', '--format', type='choice', dest='format',
help="Select the mailbox format from %s. See the mailbox module's documention for descriptions of these formats." \
% ', '.join(formats),
default='mbox', choices=formats)
options,args = parser.parse_args(argv)
mailbox_file = args[1]
reader = getattr(mailbox, options.format)
mb = reader(mailbox_file, factory=None)
print u'<?xml version="1.0" encoding="%s" ?>' % DEFAULT_ENCODING
print u"<be-xml>"
for message in mb:
print comment_message_to_xml(message)
print u"</be-xml>"
if __name__ == "__main__":
import sys
import codecs
sys.stdout = codecs.getwriter(DEFAULT_ENCODING)(sys.stdout)
main(sys.argv)
