class Configured(object):
    '''Any object containing a configuration.

    The configuration is passed to the constructor.

    .. attribute:: config

        The configuration dictionary for this object.

    '''
    def __init__(self, config, *args, **kwargs):
        super(Configured, self).__init__(*args, **kwargs)
        self.config = config BY_TIME = 'by_time' WITH_SOURCE = 'with_source' KEYWORDS = 'keywords' HASH_TF_SID = 'hash_tf_sid' HASH_FREQUENCY = 'hash_frequency' HASH_KEYWORD = 'hash_keyword' def stream_id_to_kvlayer_key(stream_id):
    '''Convert a text stream ID to a kvlayer key.

    The return tuple can be used directly as a key in the
    :data:`STREAM_ITEMS_TABLE` table.

    :param str stream_id: stream ID to convert
    :return: :mod:`kvlayer` key tuple
    :raise exceptions.KeyError: if `stream_id` is malformed

    '''
    # Reminder: stream_id is 1234567890-123456789abcdef...0
    # where the first part is the (decimal) epoch_ticks and the second
    # part is the (hex) doc_id
    parts = stream_id.split('-')
    if len(parts) != 2:
        raise KeyError('invalid stream_id ' + stream_id)
    epoch_ticks_s = parts[0]
    doc_id_s = parts[1]
    if not epoch_ticks_s.isdigit():
        raise KeyError('invalid stream_id ' + stream_id)
    if doc_id_s.lstrip(string.hexdigits) != '':
        raise KeyError('invalid stream_id ' + stream_id)
    return (base64.b16decode(doc_id_s.upper()), int(epoch_ticks_s)) def kvlayer_key_to_stream_id(k):
    '''Convert a kvlayer key to a text stream ID.

    `k` should be of the same form produced by
    :func:`stream_id_to_kvlayer_key`.

    :param k: :mod:`kvlayer` key tuple
    :return: converted stream ID
    :returntype str:

    '''
    abs_url_hash, epoch_ticks = k
    return '{0}-{1}'.format(epoch_ticks,
                            base64.b16encode(abs_url_hash).lower()) def key_for_stream_item(si):
    '''Get a kvlayer key from a stream item.

    The return tuple can be used directly as a key in the
    :data:`STREAM_ITEMS_TABLE` table.  Note that this recalculates the
    stream ID, and if the internal data on the stream item is inconsistent
    then this could return a different result from
    :func:`stream_id_to_kvlayer_key`.

    :param si: stream item to get key for
    :return: :mod:`kvlayer` key tuple

    '''
    # get binary 16 byte digest
    urlhash = hashlib.md5(si.abs_url).digest()
    return (urlhash, int(si.stream_time.epoch_ticks)) STREAM_ITEM_TABLE_DEFS = {
    # StreamItem primary key is:
    # (binary 16 byte md5 of abs_url, timestamp)
    STREAM_ITEMS_TABLE: (str, int),
    STREAM_ITEMS_SOURCE_INDEX: (str, int),
    STREAM_ITEMS_TIME_INDEX: (int, str),

    # keyword index tables:
    HASH_TF_INDEX_TABLE: (int, str, int),
    HASH_FREQUENCY_TABLE: (int,),
    HASH_KEYWORD_INDEX_TABLE: (int, str),

    # TODO: maybe reintroduce index-by-time
} STREAM_ITEM_VALUE_DEFS = {
    # Unspecified tables default to str value
    HASH_TF_INDEX_TABLE: int,
    HASH_FREQUENCY_TABLE: COUNTER,
    HASH_KEYWORD_INDEX_TABLE: COUNTER,
} INDEX_TABLE_NAMES = {
    WITH_SOURCE: STREAM_ITEMS_SOURCE_INDEX,
    BY_TIME: STREAM_ITEMS_TIME_INDEX,
    HASH_FREQUENCY: HASH_FREQUENCY_TABLE,
    HASH_KEYWORD: HASH_KEYWORD_INDEX_TABLE,
    HASH_TF_SID: HASH_TF_INDEX_TABLE,
} STREAM_ITEMS_TABLE = 'si' STREAM_ITEMS_SOURCE_INDEX = 'sisi' STREAM_ITEMS_TIME_INDEX = 'siti'
import base64
import logging
import re
import struct
import kvlayer
import streamcorpus
import yakonfig
from streamcorpus_pipeline.stages import Configured
from streamcorpus_pipeline._kvlayer_table_names import \
    BY_TIME, WITH_SOURCE, KEYWORDS, HASH_TF_SID, HASH_FREQUENCY, \
    HASH_KEYWORD, \
    stream_id_to_kvlayer_key, kvlayer_key_to_stream_id, key_for_stream_item, \
    STREAM_ITEM_TABLE_DEFS, STREAM_ITEM_VALUE_DEFS, INDEX_TABLE_NAMES, \
    STREAM_ITEMS_TABLE, STREAM_ITEMS_SOURCE_INDEX, STREAM_ITEMS_TIME_INDEX
    from ._kvlayer_keyword_search import keyword_indexer
    :param client: kvlayer client object
    :type client: :class:`kvlayer.AbstractStorage`
    :param str stream_id: stream Id to retrieve
    :return: corresponding :class:`streamcorpus.StreamItem`
    :raise exceptions.KeyError: if `stream_id` is malformed or does
      not correspond to anything in the database

    '''
    if client is None:
        client = kvlayer.client()
        client.setup_namespace(STREAM_ITEM_TABLE_DEFS,
                               STREAM_ITEM_VALUE_DEFS)
    key = stream_id_to_kvlayer_key(stream_id)
    for k, v in client.get(STREAM_ITEMS_TABLE, key):
        if v is not None:
            errors, bytestr = streamcorpus.decrypt_and_uncompress(v)
            return streamcorpus.deserialize(bytestr)
    raise KeyError(stream_id)


def make_doc_id_range(doc_id):
    '''Construct a tuple(begin, end) of one-tuple kvlayer keys from a
    hexdigest doc_id.

    '''
    assert len(doc_id) == 32, 'expecting 32 hex string, not: %r' % doc_id
    bin_docid = base64.b16decode(doc_id.upper())
    doc_id_range = ((bin_docid,), (bin_docid,))
    return doc_id_range


def get_kvlayer_stream_item_by_doc_id(client, doc_id):
    '''Retrieve :class:`streamcorpus.StreamItem`s from :mod:`kvlayer`.

    Namely, it returns an iterator over all documents with the given
    docid. The docid should be an md5 hash of the document's abs_url.

    :param client: kvlayer client object
    :type client: :class:`kvlayer.AbstractStorage`
    :param str doc_id: doc id of documents to retrieve
    :return: generator of :class:`streamcorpus.StreamItem`
    '''
    if client is None:
        client = kvlayer.client()
        client.setup_namespace(STREAM_ITEM_TABLE_DEFS,
                               STREAM_ITEM_VALUE_DEFS)
    doc_id_range = make_doc_id_range(doc_id)
    for k, v in client.scan(STREAM_ITEMS_TABLE, doc_id_range):
        if v is not None:
            errors, bytestr = streamcorpus.decrypt_and_uncompress(v)
            yield streamcorpus.deserialize(bytestr)


def get_kvlayer_stream_ids_by_doc_id(client, doc_id):
    '''Retrieve stream ids from :mod:`kvlayer`.

    Namely, it returns an iterator over all stream ids with the given
    docid. The docid should be an md5 hash of the document's abs_url.

    :param client: kvlayer client object
    :type client: :class:`kvlayer.AbstractStorage`
    :param str doc_id: doc id of documents to retrieve
    :return: generator of str
    '''
    if client is None:
        client = kvlayer.client()
        client.setup_namespace(STREAM_ITEM_TABLE_DEFS,
                               STREAM_ITEM_VALUE_DEFS)
    doc_id_range = make_doc_id_range(doc_id)
    for k in client.scan_keys(STREAM_ITEMS_TABLE, doc_id_range):
        yield kvlayer_key_to_stream_id(k)


def delete_kvlayer_stream_item(client, stream_id):
    if client is None:
        client = kvlayer.client()
        client.setup_namespace(STREAM_ITEM_TABLE_DEFS,
                               STREAM_ITEM_VALUE_DEFS)
    key = stream_id_to_kvlayer_key(stream_id)
    client.delete(STREAM_ITEMS_TABLE, key)


class to_kvlayer(Configured):
    '''Writer that puts stream items in :mod:`kvlayer`.

    :mod:`kvlayer` must be included in the top-level configuration.
    Compressed stream items are written to the
    :data:`~streamcorpus_pipeline._kvlayer_table_names.STREAM_ITEMS_TABLE`
    table.

    This writer has one configuration parameter: `indexes` adds
    additional indexes on the data.  Supported index types include
    :data:`~streamcorpus_pipeline._kvlayer_table_names.WITH_SOURCE`,
    :data:`~streamcorpus_pipeline._kvlayer_table_names.BY_TIME`,
    :data:`~streamcorpus_pipeline._kvlayer_table_names.HASH_TF_SID`,
    :data:`~streamcorpus_pipeline._kvlayer_table_names.HASH_FREQUENCIES`,
    and
    :data:`~streamcorpus_pipeline._kvlayer_table_names.HASH_KEYWORD`.
    The special index type
    :data:`~streamcorpus_pipeline._kvlayer_table_names.KEYWORDS`
    expands to all of the keyword-index tables.

    If keyword indexing is enabled, it indexes the tokens generated
    by some or all of the taggers that have been run, less a set of
    stop words.  The configuration parameter `keyword_tagger_ids` may
    be set to a list of tagger IDs; if it is set, only the tokens
    from those tagger IDs will be indexed.  If it is set to ``null``
    or left unset, all taggers' tokens will be indexed as distinct
    words.  Any tokens whose size is greater than `keyword_size_limit`
    (default 128) will not be indexed.

    '''
    config_name = 'to_kvlayer'
    default_config = {'indexes': [],
                      'keyword_size_limit': 128}

    @staticmethod
    def check_config(config, name):
        yakonfig.check_toplevel_config(kvlayer, name)
        for ndx in config['indexes']:

            if ndx != KEYWORDS and ndx not in INDEX_TABLE_NAMES: