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


try:
except ImportError, exc:
    keyword_indexer = str(exc)

logger = logging.getLogger(__name__)


class WrongNumberException(Exception):
    pass


def iter_one_or_ex(gen):
    # for when you want exactly one result out of an iterable.
    # raise WrongNumberException otherwise.
    val = None
    any = False
    for tv in gen:
        if not any:
            any = True
            val = tv
        else:
            raise WrongNumberException("only wanted one, but got another")
    if not any:
        raise WrongNumberException("didn't get any")
    return val


class from_kvlayer(Configured):
    '''Read documents from a :mod:`kvlayer` supported database.

    .. currentmodule:: streamcorpus_pipeline._kvlayer

    :mod:`kvlayer` must be included in the top-level configuration.
    Typically this reads stream items previously written by
    :class:`to_kvlayer`.

    The input string is a sequence of stream IDs; see :meth:`__call__`
    for details.  If no values are specified, scans the entire
    :data:`~streamcorpus_pipeline._kvlayer_table_names.STREAM_ITEMS_TABLE`
    table.

    .. automethod:: __call__

    '''
    config_name = 'from_kvlayer'

    @staticmethod
    def check_config(config, name):
        yakonfig.check_toplevel_config(kvlayer, name)

    def __init__(self, *args, **kwargs):
        super(from_kvlayer, self).__init__(*args, **kwargs)
        self.client = kvlayer.client()
        self.client.setup_namespace(STREAM_ITEM_TABLE_DEFS,
                                    STREAM_ITEM_VALUE_DEFS)

    def _loadkey(self, k):
        key, data = iter_one_or_ex(self.client.get(STREAM_ITEMS_TABLE, k))
        errors, data = streamcorpus.decrypt_and_uncompress(data)
        if errors:
            if not isinstance(errors, Exception):
                errors = Exception(repr(errors))
            raise errors
        yield streamcorpus.deserialize(data)

    def _loadrange(self, keya, keyb):
        for key, data in self.client.scan(STREAM_ITEMS_TABLE, (keya, keyb)):
            errors, data = streamcorpus.decrypt_and_uncompress(data)
            if errors:
                logger.error('could not decrypet_and_uncompress %s: %s',
                             key, errors)
                continue
            yield streamcorpus.deserialize(data)

    def __call__(self, i_str):
        '''Actually scan the table, yielding stream items.

        `i_str` consists of a sequence of 20-byte encoded stream IDs.
        These are the 16 bytes of the document ID, plus the 4-byte
        big-endian stream time.  :func:`serialize_si` can produce
        these.  `i_str` can then consist of a single encoded stream
        ID; an encoded stream ID, a literal ``<``, and a second
        encoded stream ID to specify a range of documents; or a list
        of either of the preceding separated by literal ``;``.

        .. todo:: make this support keyword index strings

        '''
        if not i_str:
            for si in self._loadrange(None, None):
                yield si
            return
        for si in parse_keys_and_ranges(i_str, self._loadkey,
                                        self._loadrange):
            yield si


_STREAM_ID_RE = re.compile(r'[0-9]+-[0-9a-fA-F]{32}')


def parse_keys_and_ranges(i_str, keyfunc, rangefunc):
    '''Parse the :class:`from_kvlayer` input string.

    This accepts two formats.  In the textual format, it accepts any
    number of stream IDs in timestamp-docid format, separated by ``,``
    or ``;``, and processes those as individual stream IDs.  In the
    binary format, it accepts 20-byte key blobs (16 bytes md5 hash, 4
    bytes timestamp) split by ``;`` or ``<``; e.g., ``a<f;x`` loads
    scans keys `a` through `f` and loads singly key `x`.

    `keyfunc` and `rangefunc` are run as generators and their yields
    are yielded from this function.

    '''
    while i_str:
        m = _STREAM_ID_RE.match(i_str)
        if m:
            # old style text stream_id

            for retval in keyfunc(stream_id_to_kvlayer_key(m.group())):