 int((end-start).total_seconds()))
    
    for bkt in buckets:
        yield (bkt, '/view/%s/%s/%s/%s/%s' % (data[0][0], data[0][1], data[0][2], hashlib.md5(document.text).hexdigest()[:12], document._id))

def lsh_bucket(key, values):
    """LSH reduce function."""
    yield '%s\n' % str({key: values})

class LshPipeline(base_handler.PipelineBase):
    """A pipeline to run LSH.

    Args:
      blobkey: blobkey to process as string. Should be a zip archive with
        text files inside.
    """

    def run(self, filename, blobkey, ds_key):
        params = "filename %s \tblobkey %s\tds_key %s" % (filename, blobkey, ds_key)
        logging.info(params)

        dataset = ndb.Key(urlsafe=ds_key).get()
        rows = dataset.rows
        hashes = rows * dataset.bands
        if len(dataset.random_seeds)!= hashes:
            dataset.random_seeds = [random.getrandbits(max_bits) for _ in xrange(hashes)]
            logging.warning('Recalculated %d random seeds', hashes)
            dataset.put()
    
        dataset.buckets = []
        dataset.put()
        output = yield mapreduce_pipeline.MapreducePipeline(
            "locality_sensitive_hashing",
            "blobs.lsh_map",
            "blobs.lsh_bucket",
           'mapreduce.input_readers.BlobstoreZipLineInputReader', 
            "mapreduce.output_writers.BlobstoreOutputWriter",
            mapper_params={
                "blob_keys": blobkey,
            },
            reducer_params={
                "mime_type": "text/plain",
            },
            shards=16)
        yield StoreLshResults('OpenLSH', blobkey, ds_key, output)
    def finalized(self):
        pass

class StoreLshResults(base_handler.PipelineBase):
    """A pipeline to store the result of the MapReduce job in the database.

    Args:
      report: Name of the run
      encoded_key: the DB key corresponding to the metadata of this job
      output: the blobstore location where the output of the job is stored
    """
    def run(self, report, blobkey, ds_key, output):
        logging.info("blobkey is %s, output is %s", blobkey, str(output))
        dataset = ndb.Key(urlsafe=ds_key).get()
        dataset.output_link = output[0]
        dataset.result_link = ''
        dataset.put()
        return

    def finalized(self):
        logging.info('StoreLshResults finalized')

class ViewHandler(webapp2.RequestHandler):
    def get(self, zip_key, file_no, offset, md5, _id):
        def cleanup(text):
            return text.replace('\\n','').replace('\\"', '"').replace('http://inventures.euhttp://inventures.eu/', 'http://inventures.eu/')
        blob_reader = blobstore.BlobReader(zip_key)
        zip_reader = zipfile.ZipFile(blob_reader)
        infolist = zip_reader.infolist()
        zipinfo = infolist[int(file_no)]
        with zip_reader.open(zipinfo) as f:
            f.read(int(offset))
            text = f.readline()
            found_pattern = text_file_pattern.search(text)
            html = found_pattern.group(2)#.replace('/sites/','http://inventures.eu/sites/' )
            self.response.out.write(cleanup(html))
            return
        message = 'ID %s not found' % _id
        self.response.out.write('<html><body><p>%s</p></body></html>' % message)
        return

class EvalPipeline(base_handler.PipelineBase):
    def run(self, resource):
        pairs = 0
        with blobstore.BlobReader(resource) as blob_reader:
            for line in blob_reader.readlines():
                # {'33700': ['/view/Mla4PRwYe1ZpNJN4hluceA==/0/1060348/59518bb889e6/mpmoIZY6S4Si89wdEyX9IA', etc ]}
                kdocs = json.loads(line.replace("'", '"'))
                k = kdocs.keys()[0]
                docs = kdocs[k]
                pairs += len(docs) * (len(docs) - 1) / 2
        logging.info('Total number of pairs to compute: %d', pairs)
        output = yield mapreduce_pipeline.MapreducePipeline(
            "results-eval",
            "blobs.eval_map",
            "blobs.eval_reduce",
           'mapreduce.input_readers.BlobstoreLineInputReader', 
            "mapreduce.output_writers.BlobstoreOutputWriter",
            mapper_params={
                "blob_keys": resource,
            },
            reducer_params={
                "mime_type": "text/plain",
            },
            shards=16)
        yield StoreEvalResults(resource, output)
    def finalized(self):
        pass

def eval_map2(data):
    (offset, line) = data
#     logging.info('eval %s',line)
    kv = json.loads(line.replace("'", '"'))
    k = kv.keys()[0]
    vs = kv[k]
    hv = {}
    for v in vs:
        h = v.split('/')[-2]
        if h not in hv:
            hv[h] = [v]
        else:
            hv[h] += [v]
    for h1 in hv:
        for h2 in hv:
            if h1 <= h2: continue
            yield {k: [hv[h1], hv[h2]]}, ""

def eval_reduce2(khv, values):
    def retrieve_doc(v):
        (zip_key, file_no, offset, h, id1) = tuple(v[6:].split('/'))
        dataset = DatasetPB.query(DatasetPB.blob_key == blobstore.BlobKey(zip_key)).get()
        blob_reader = blobstore.BlobReader(zip_key)
        zip_reader = zipfile.ZipFile(blob_reader)
        infolist = zip_reader.infolist()
        zipinfo = infolist[int(file_no)]
        with zip_reader.open(zipinfo) as f:
            f.read(int(offset))
            text = f.readline()
            found_pattern = text_file_pattern.search(text)
            doc = Document(found_pattern.group(1), found_pattern.group(2), dataset)
        shingles = set(doc.shingles())
        minhashes = doc.calc_minhashes()
        return shingles, minhashes, len(doc.text)
    khv = khv.replace("{u'", "{'").replace("[u'", "['").replace(" u'", " '").replace("'", '"')
    try:
        khv2 = json.loads(khv)
    except:
        logging.warning('json.loads failure for %s', khv)
        return
    k = khv2.keys()[0]
    hv = khv2[k]
    v1 = hv[0][0]
    v2 = hv[1][0]
    (shingles1, minhashes1, len1) = retrieve_doc(v1)
    (shingles2, minhashes2, len2) = retrieve_doc(v2)
    jac_txt = float(len(shingles1 & shingles2)) / float(len(shingles1 | shingles2)) 
    jac_min = reduce(lambda x, y: x+y, map(lambda a,b: a == b, minhashes1,minhashes2)) / float(len(minhashes1))
    emitting = {'set1': [str(addr.split('/')[-1]) for addr in hv[0