blob: 28c03635d50069b3c6487d594262debb88459b8a [file] [log] [blame]
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
# Licensed to the Apache Software Foundation (ASF) under one or more
# contributor license agreements. See the NOTICE file distributed with
# this work for additional information regarding copyright ownership.
# The ASF licenses this file to You under the Apache License, Version 2.0
# (the "License"); you may not use this file except in compliance with
# the License. You may obtain a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
import sys
import random
import time
import hashlib
import os
from threading import Thread, Lock
import mailbox
import email
from urllib.request import urlopen
import re
from elastic import Elastic
import argparse
from os import listdir
from os.path import isfile, join, isdir
import glob
import multiprocessing
import tempfile
import gzip
import archiver
goodies = 0
baddies = 0
duplicates={} # detect if mid is re-used this run
block = Lock()
lists = [] # N.B. the entries in this list depend on the import type:
# globDir: [filename, list-id]
# modMbox: [list-id, mbox]
# piperMail: [filename, list-id]
# imap(s): [uids, listname, imap4]
# other: [filename, list-override]
start = time.time()
quickmode = False
private = False
appender = "apache.org"
source = "./"
maildir = False
imap = False
list_override = None
project = ""
filebased = False
fileToLID = {}
interactive = False
extension = ".mbox"
piperWeirdness = False
resendTo = None
timeout = 600
fromFilter = None
dedup = False
dedupped = 0
noMboxo = False # Don't skip MBoxo patch
# Fetch config and set up ES
es = Elastic()
# We need the index name for bulk actions
dbname = es.getdbname()
rootURL = ""
def bulk_insert(name, json, xes, dtype, wc = 'quorum'):
if args.dry:
return
sys.stderr.flush()
js_arr = []
for entry in json:
js = entry
mid = js['mid']
if dtype == 'mbox_source':
del js['mid']
js_arr.append({
'_op_type': 'index',
'_consistency': wc,
'_index': dbname,
'_type': dtype,
'_id': mid,
'doc': js,
'_source': js
})
try:
xes.bulk(js_arr,ignore=404)
# print("%s: Inserted %u entries into %s" % (name, len(js_arr),dtype))
except Exception as err:
print("%s: Warning: Could not bulk insert: %s into %s" % (name,err,dtype))
class SlurpThread(Thread):
def printid(self, message):
print("%s: %s" % (self.name, message))
def run(self):
global goodies, baddies, dedupped
self.name = Thread.getName(self)
ja = []
jas = []
self.printid("Thread started")
mla = None
ml = ""
mboxfile = ""
filename = ""
archie = archiver.Archiver(generator=args.generator, parse_html=args.html2text,
ignore_body=args.ibody[0] if args.ibody else None,
verbose=args.verbose,
skipff=args.skipff)
while len(lists) > 0:
self.printid("%u elements left to slurp" % len(lists))
block.acquire()
try:
mla = lists.pop(0)
if not mla:
self.printid("Nothing more to do here")
return
except Exception as err:
self.printid("Could not pop list: %s" % err)
return
finally:
block.release()
stime = time.time()
dFile = False
if imap:
imap4 = mla[2]
def mailgen(_list):
for uid in _list:
msgbytes = imap4.uid('fetch', uid, '(RFC822)')[1][0][1]
yield email.message_from_bytes(msgbytes)
messages = mailgen(mla[0])
elif filebased:
tmpname = mla[0]
filename = mla[0]
if filename.find(".gz") != -1:
self.printid("Decompressing %s..." % filename)
try:
with open(filename, "rb") as bf:
bmd = bf.read()
bf.close() # explicit early close
bmd = gzip.decompress(bmd)
tmpfile = tempfile.NamedTemporaryFile(mode='w+b', buffering=1, delete=False)
tmpfile.write(bmd)
tmpfile.flush()
tmpfile.close()
tmpname = tmpfile.name
dFile = True # Slated for deletion upon having been read
self.printid("%s -> %u bytes" % (tmpname, len(bmd)))
except Exception as err:
self.printid("This wasn't a gzip file: %s" % err)
self.printid("Slurping %s" % filename)
if maildir:
messages = mailbox.Maildir(tmpname, create=False)
else:
messages = mailbox.mbox(tmpname, None if noMboxo else MboxoFactory, create=False)
else:
ml = mla[0]
mboxfile = mla[1]
self.printid("Slurping %s/%s" % (ml, mboxfile))
ctx = urlopen("%s%s/%s" % (source, ml, mboxfile ))
inp = ctx.read().decode(ctx.headers.get_content_charset() or 'utf-8', errors='ignore')
tmpname = hashlib.sha224(("%f-%f-%s-%s.mbox" % (random.random(), time.time(), ml, mboxfile)).encode('utf-8') ).hexdigest()
with open(tmpname, "w") as f:
f.write(inp)
if maildir:
messages = mailbox.Maildir(tmpname, create=False)
else:
messages = mailbox.mbox(tmpname, None if noMboxo else MboxoFactory, create=False)
count = 0
bad = 0
for key in messages.iterkeys():
message=messages.get(key)
# If --filter is set, discard any messages not matching by continuing to next email
if fromFilter and 'from' in message and message['from'].find(fromFilter) == -1:
continue
if resendTo:
self.printid("Delivering message %s via MTA" % message['message-id'] if 'message-id' in message else '??')
s = SMTP('localhost')
try:
if list_override:
message.replace_header('List-ID', list_override)
message.replace_header('To', resendTo)
except:
if list_override:
message['List-ID'] = list_override
message['cc'] = None
s.send_message(message, from_addr=None, to_addrs=(resendTo))
continue
if (time.time() - stime > timeout): # break out after N seconds, it shouldn't take this long..!
self.printid("Whoa, this is taking way too long, ignoring %s for now" % tmpname)
break
# Don't pass message to archiver unless we have a list id
if not (list_override or message['list-id']):
self.printid("No list id found for %s " % message['message-id'])
bad += 1
continue
json, contents, _msgdata, _irt = archie.compute_updates(list_override, private, message)
# Not sure this can ever happen
if json and not (json['list'] and json['list_raw']):
self.printid("No list id found for %s " % json['message-id'])
bad += 1
continue
# If --dedup is active, try to filter out any messages that already exist on the list
if json and dedup and message.get('message-id', None):
res = es.search(
doc_type="mbox",
size = 1,
_source = ['mid'], # so can report the match source
body = {
'query': {
'bool': {
'must': [
{
'term': {
'message-id': message.get('message-id', None)
}
},
{
'term': {
'list_raw': json['list']
}
}
]
}
}
}
)
if res and res['hits']['total'] > 0:
self.printid("Dedupping %s - matched in %s" % (json['message-id'], res['hits']['hits'][0]['_source']['mid']))
dedupped += 1
continue
if json:
file=messages.get_file(key, True)
# If the parsed data is filtered, also need to filter the raw input
# so the source agrees with the summary info
if message.__class__.__name__ == 'MboxoFactory':
file=MboxoReader(file)
raw_msg=file.read()
file.close()
if args.dups:
try:
duplicates[json['mid']].append(json['message-id'] + " in " + filename)
except:
duplicates[json['mid']]=[json['message-id'] + " in " + filename]
try: # temporary hack to try and find an encoding issue
# needs to be replaced by proper exception handling
json_source = {
'mid': json['mid'], # needed for bulk-insert only, not needed in database
'message-id': json['message-id'],
'source': archiver.mbox_source(raw_msg)
}
except Exception as e:
self.printid("Error '%s' processing id %s msg %s " % (e, json['mid'], json['message-id']))
bad += 1
continue
count += 1
ja.append(json)
jas.append(json_source)
if args.verbose and verbose_logger:
# TODO optionally show other fields (e.g. From_ line)
verbose_logger.info("MID:%s MSGID:%s" % (json['mid'], json['message-id'].strip()))
if contents:
if not args.dry:
for key in contents:
es.index(
doc_type="attachment",
id=key,
body = {
'source': contents[key]
}
)
if len(ja) >= 40:
bulk_insert(self.name, ja, es, 'mbox')
ja = []
bulk_insert(self.name, jas, es, 'mbox_source')
jas = []
else:
self.printid("Failed to parse: Return=%s Message-Id=%s" % (message.get('Return-Path'), message.get('Message-Id')))
bad += 1
if filebased:
self.printid("Parsed %u records (failed: %u) from %s" % (count, bad, filename))
if dFile:
os.unlink(tmpname)
elif imap:
self.printid("Parsed %u records (failed: %u) from imap" % (count, bad))
else:
self.printid("Parsed %s/%s: %u records (failed: %u) from %s" % (ml, mboxfile, count, bad, tmpname))
os.unlink(tmpname)
goodies += count
baddies += bad
if len(ja) > 0:
bulk_insert(self.name, ja, es, 'mbox')
ja = []
if len(jas) > 0:
bulk_insert(self.name, jas, es, 'mbox_source')
jas = []
self.printid("Done, %u elements left to slurp" % len(lists))
parser = argparse.ArgumentParser(description='Command line options.')
parser.add_argument('--source', dest='source', type=str, nargs=1,
help='Source to scan (http(s)://, imap(s):// or file path)')
parser.add_argument('--dir', dest='dir', action='store_true',
help='Input is in Maildir format')
parser.add_argument('--interactive', dest='interactive', action='store_true',
help='Ask for help when possible')
parser.add_argument('--quick', dest='quick', action='store_true',
help='Only grab the first file you can find')
parser.add_argument('--mod-mbox', dest='modmbox', action='store_true',
help='This is mod_mbox, derive list-id and files from it')
parser.add_argument('--pipermail', dest='pipermail', action='store_true',
help='This is pipermail, derive files from it (list ID has to be set!)')
parser.add_argument('--lid', dest='listid', type=str, nargs=1,
help='Optional List-ID to override source with. Format: <list-name>@<domain>')
parser.add_argument('--project', dest='project', type=str, nargs=1,
help='Optional project to look for ($project-* will be imported as well)')
parser.add_argument('--ext', dest='ext', type=str, nargs=1,
help='Optional file extension e.g. ".gz" (or call it with an empty string to not care)')
parser.add_argument('--domain', dest='domain', type=str, nargs=1,
help='Optional domain extension for MIDs and List ID reconstruction)')
parser.add_argument('--private', dest='private', action='store_true',
help='This is a privately archived list. Filter through auth proxy.')
parser.add_argument('--dry', dest='dry', action='store_true',
help='Do not save emails to elasticsearch, only test importing')
parser.add_argument('--verbose', dest='verbose', action='store_true',
help='Show details of generated id (for use with --dry)')
parser.add_argument('--duplicates', dest='dups', action='store_true',
help='Detect duplicate mids in this run')
parser.add_argument('--html2text', dest='html2text', action='store_true',
help='If no text/plain is found, try to parse HTML using html2text')
parser.add_argument('--requirelid', dest='requirelid', action='store_true',
help='Require a List ID to be present, ignore otherwise')
parser.add_argument('--dedup', dest='dedup', action='store_true',
help='Try to dedup messages based on ID before importing')
parser.add_argument('--ignorebody', dest='ibody', type=str, nargs=1,
help='Optional email bodies to treat as empty (in conjunction with --html2text)')
parser.add_argument('--resend', dest='resend', type=str, nargs=1,
help='DANGER ZONE: Resend every read email to this recipient as a new email')
parser.add_argument('--timeout', dest='timeout', type=int, nargs=1,
help='Optional timeout in secs for importing an mbox/maildir file (default is 600 seconds)')
parser.add_argument('--filter', dest = 'fromfilter', type=str, nargs=1,
help = 'Optional sender filter: Only import emails from this address')
parser.add_argument('--nomboxo', dest = 'nomboxo', action='store_true',
help = 'Skip Mboxo processing')
parser.add_argument('--generator', dest='generator',
help='Override the generator.')
parser.add_argument('--skipff', dest = 'skipff', action='store_true',
help = 'Skip format=flowed processing (mainly for unit-testing)')
args = parser.parse_args()
if len(sys.argv) <= 2:
parser.print_help()
sys.exit(-1)
if args.source:
source = args.source[0]
if args.dir:
maildir = args.dir
if args.listid:
list_override = archiver.normalize_lid(args.listid[0])
if args.project:
project = args.project[0]
if args.domain:
appender = args.domain[0]
if args.interactive:
interactive = args.interactive
if args.quick:
quickmode = args.quick
if args.private:
private = args.private
if args.dedup:
dedup = args.dedup
if args.ext:
extension = args.ext[0]
if args.fromfilter:
fromFilter = args.fromfilter[0]
if args.nomboxo:
noMboxo = args.nomboxo
else:
# Temporary patch to fix Python email package limitation
# It must be removed when the Python package is fixed
from mboxo_patch import MboxoFactory, MboxoReader
if args.resend:
resendTo = args.resend[0]
from smtplib import SMTP
if args.timeout:
timeout = args.timeout[0]
baddies = 0
# No point continuing if the index does not exist
print("Checking that the database index %s exists ... " % dbname)
# elasticsearch logs lots of warnings on retries/connection failure
import logging
logging.getLogger("elasticsearch").setLevel(logging.ERROR)
verbose_logger = None
if args.verbose:
verbose_logger = logging.getLogger("verbose")
verbose_logger.setLevel(logging.INFO)
# The default handler is set to WARN level
verbose_logger.addHandler(logging.StreamHandler(sys.stdout))
archiver.logger = verbose_logger
if args.dry:
print("Dry-run; continuing to check input data")
else:
# Need to check the index before starting bulk operations
try:
if not es.indices.exists(index=dbname):
print("Error: the index '%s' does not exist!" % (dbname))
sys.exit(1)
print("Database exists OK")
except Exception as err:
print("Error: unable to check if the index %s exists!: %s" % (dbname, err))
sys.exit(1)
def globDir(d):
dirs = [ f for f in listdir(d) if isdir(join(d,f)) ]
mboxes = [ f for f in glob.glob(join(d,"*" + extension)) if isfile(f) ]
if not d in fileToLID and len(mboxes) > 0 and interactive:
print("Would you like to set a list-ID override for %s?:" % d)
lo = sys.stdin.readline()
if lo and len(lo) > 3:
fileToLID[d] = archiver.normalize_lid(lo.strip("\r\n"))
print("Righto, setting it to %s." % fileToLID[d])
else:
print("alright, I'll try to figure it out myself!")
for fi in sorted(mboxes):
lists.append([fi, fileToLID.get(d) if fileToLID.get(d) else list_override])
for nd in sorted(dirs):
globDir(join(d,nd))
# HTTP(S) based import?
if re.match(r"https?://", source):
data = urlopen(source).read().decode('utf-8')
print("Fetched %u bytes of main data, parsing month lists" % len(data))
if project:
# ensure there is a '-' between project and list name otherwise we match too much
# Note: It looks like mod_mbox always uses single quoted hrefs
ns = r"<a href='(%s-[-a-z0-9]+)/'" % project
if project.find("-") != -1:
ns = r"<a href='(%s)/'" % project
else: # match all possible project names
ns = r"<a href='([-a-z0-9]+)/'"
if args.modmbox:
for mlist in re.finditer(ns, data):
ml = mlist.group(1)
mldata = urlopen("%s%s/" % (source, ml)).read().decode('utf-8')
present = re.search(r"<th colspan=\"3\">Year 20[\d]{2}</th>", mldata) # Check that year 2014-2017 exists, otherwise why keep it?
if present:
qn = 0
for mbox in re.finditer(r"(\d+\.mbox)/thread", mldata):
qn += 1
mboxfile = mbox.group(1)
lists.append([ml, mboxfile])
print("Adding %s/%s to slurp list" % (ml, mboxfile))
if quickmode and qn >= 2:
break
if args.pipermail:
filebased = True
piperWeirdness = True
if not list_override:
print("You need to specify a list ID with --lid when importing from Pipermail!")
sys.exit(-1)
ns = r"href=\"(\d+-[a-zA-Z]+\.txt(\.gz)?)\""
qn = 0
for mlist in re.finditer(ns, data):
ml = mlist.group(1)
mldata = urlopen("%s%s" % (source, ml)).read()
tmpfile = tempfile.NamedTemporaryFile(mode='w+b', buffering=1, delete=False)
try:
if ml.find(".gz") != -1:
mldata = gzip.decompress(mldata)
except Exception as err:
print("This wasn't a gzip file: %s" % err )
print(len(mldata))
tmpfile.write(mldata)
tmpfile.flush()
tmpfile.close()
lists.append([tmpfile.name, list_override])
print("Adding %s/%s to slurp list as %s" % (source, ml, tmpfile.name))
qn += 1
if quickmode and qn >= 2:
break
# IMAP(S) based import?
elif re.match(r"imaps?://", source):
imap = True
import urllib, getpass, imaplib
url = urllib.parse.urlparse(source)
port = url.port or (143 if url.scheme == 'imap' else 993)
user = url.username or getpass.getuser()
password = url.password or getpass.getpass('IMAP Password: ')
folder = url.path.strip('/') or 'INBOX'
listname = list_override or "<%s/%s.%s>" % (user, folder, url.hostname)
# fetch message-id => _id pairs from elasticsearch
result = es.search(scroll = '5m',
body = {
'size': 1024,
'fields': ['message-id'],
'query': {'match': {'list': listname}}
}
)
db = {}
while len(result['hits']['hits']) > 0:
for hit in result['hits']['hits']:
db[hit['fields']['message-id'][0]] = hit['_id']
result = es.scroll(scroll='5m', scroll_id=result['_scroll_id'])
# fetch message-id => uid pairs from imap
if url.scheme == 'imaps':
imap4 = imaplib.IMAP4_SSL(url.hostname, port)
else:
imap4 = imaplib.IMAP4(url.hostname, port)
imap4.login(user, password)
imap4.select(folder, readonly=True)
results = imap4.uid('search', None, 'ALL')
uids = b','.join(results[1][0].split())
results = imap4.uid('fetch', uids, '(BODY[HEADER.FIELDS (MESSAGE-ID)])')
mail = {}
uid_re = re.compile(r'^\d+ \(UID (\d+) BODY\[')
mid_re = re.compile(r'^Message-ID:\s*(.*?)\s*$', re.I)
uid = None
for result in results[1]:
for line in result:
if isinstance(line, bytes):
match = uid_re.match(line)
if match:
uid = match.group(1)
else:
match = mid_re.match(line)
if match:
try:
mail[match.group(1).decode('utf-8')] = uid
uid = None
except ValueError:
pass
# delete items from elasticsearch that are not present in imap
queue1 = []
queue2 = []
for mid, _id in db.items():
if not mid in mail:
queue1.append({
'_op_type': 'delete',
'_index': dbname,
'_type': 'mbox',
'_id': _id
})
queue2.append({
'_op_type': 'delete',
'_index': dbname,
'_type': 'mbox_source',
'_id': _id
})
print("deleting: " + mid)
while len(queue1) > 0:
es.bulk(queue1[0:1024])
del queue1[0:1024]
while len(queue2) > 0:
es.bulk(queue2[0:1024])
del queue2[0:1024]
# add new items to elasticsearch from imap
uids = []
for mid, uid in mail.items():
if not mid in db:
uids.append(uid)
lists.append([uids, listname, imap4])
else:
# File based import??
print("Doing file based import")
filebased = True
if maildir:
lists.append([source, fileToLID.get(source) if fileToLID.get(source) else list_override])
else:
if os.path.isfile(source):
lists.append([source, fileToLID.get(source) if fileToLID.get(source) else list_override])
else:
globDir(source)
threads = []
# Don't start more threads than there are lists
cc = min(len(lists), int( multiprocessing.cpu_count() / 2) + 1)
print("Starting up to %u threads to fetch the %u %s lists" % (cc, len(lists), project))
for i in range(1,cc+1):
t = SlurpThread()
threads.append(t)
t.start()
print("Started no. %u" % i)
for t in threads:
t.join()
if args.dups:
print("Showing duplicate ids:")
for mid in duplicates:
if len(duplicates[mid]) > 1:
print("The mid %s was used by:" % mid)
for msg in duplicates[mid]:
print(msg)
print("All done! %u records inserted/updated after %u seconds. %u records were bad and ignored" % (goodies, int(time.time() - start), baddies))
if dedupped > 0:
print("%u records were not inserted due to deduplication" % dedupped)