<commit_before>import time
from abc import ABCMeta, abstractmethod
from datetime import datetime
from elasticsearch import TransportError

import six
from corehq.apps.change_feed.document_types import is_deletion
from corehq.util.doc_processor.interface import BaseDocProcessor, BulkDocProcessor
from pillowtop.es_utils import set_index_reindex_settings, \
    set_index_normal_settings, initialize_mapping_if_necessary
from pillowtop.feed.interface import Change
from pillowtop.logger import pillow_logging
from pillowtop.utils import prepare_bulk_payloads, build_bulk_payload, ErrorCollector

MAX_TRIES = 3
RETRY_TIME_DELAY_FACTOR = 15


class Reindexer(six.with_metaclass(ABCMeta)):
    def consume_options(self, options):
        """Called from the management command with the command line
        options.

        :param options: command line options dict
        :return: dict of unprocessed options
        """
        return options

    def clean(self):
        """
            Cleans the index.

            This can be called prior to reindex to ensure starting from a clean slate.
            Should be overridden on a case-by-case basis by subclasses.
            """
        pass

    @abstractmethod
    def reindex(self):
        """Perform the reindex"""
        raise NotImplementedError


class PillowReindexer(Reindexer):
    def __init__(self, pillow):
        self.pillow = pillow


class PillowChangeProviderReindexer(PillowReindexer):
    start_from = None

    def __init__(self, pillow, change_provider):
        super(PillowChangeProviderReindexer, self).__init__(pillow)
        self.change_provider = change_provider

    def consume_options(self, options):
        self.start_from = options.pop("start_from", None)
        return options

    def reindex(self):
        for change in self.change_provider.iter_all_changes(start_from=self.start_from):
            try:
                self.pillow.process_change(change)
            except Exception:
                pillow_logging.exception("Unable to process change: %s", change.id)


def _clean_index(es, index_info):
    if es.indices.exists(index_info.index):
        es.indices.delete(index=index_info.index)


def _prepare_index_for_reindex(es, index_info):
    if not es.indices.exists(index_info.index):
        es.indices.create(index=index_info.index, body=index_info.meta)
    initialize_mapping_if_necessary(es, index_info)
    set_index_reindex_settings(es, index_info.index)


def _prepare_index_for_usage(es, index_info):
    set_index_normal_settings(es, index_info.index)
    es.indices.refresh(index_info.index)


def _set_checkpoint(pillow):
    checkpoint_value = pillow.get_change_feed().get_checkpoint_value()
    pillow_logging.info('setting checkpoint to {}'.format(checkpoint_value))
    pillow.checkpoint.update_to(checkpoint_value)


class ElasticPillowReindexer(PillowChangeProviderReindexer):
    in_place = False
<commit_msg>simple progress reporting for reindexers<commit_after>import time
from abc import ABCMeta, abstractmethod
from datetime import datetime
from elasticsearch import TransportError

import six
from corehq.apps.change_feed.document_types import is_deletion
from corehq.util.doc_processor.interface import BaseDocProcessor, BulkDocProcessor
from pillowtop.es_utils import set_index_reindex_settings, \
    set_index_normal_settings, initialize_mapping_if_necessary
from pillowtop.feed.interface import Change
from pillowtop.logger import pillow_logging
from pillowtop.utils import prepare_bulk_payloads, build_bulk_payload, ErrorCollector

MAX_TRIES = 3
RETRY_TIME_DELAY_FACTOR = 15


class Reindexer(six.with_metaclass(ABCMeta)):
    def consume_options(self, options):
        """Called from the management command with the command line
        options.

        :param options: command line options dict
        :return: dict of unprocessed options
        """
        return options

    def clean(self):
        """
            Cleans the index.

            This can be called prior to reindex to ensure starting from a clean slate.
            Should be overridden on a case-by-case basis by subclasses.
            """
        pass

    @abstractmethod
    def reindex(self):
        """Perform the reindex"""
        raise NotImplementedError


class PillowReindexer(Reindexer):
    def __init__(self, pillow):
        self.pillow = pillow


class PillowChangeProviderReindexer(PillowReindexer):
    start_from = None

    def __init__(self, pillow, change_provider):
        super(PillowChangeProviderReindexer, self).__init__(pillow)
        self.change_provider = change_provider

    def consume_options(self, options):
        self.start_from = options.pop("start_from", None)
        return options

    def reindex(self):
        for i, change in enumerate(self.change_provider.iter_all_changes(start_from=self.start_from)):
            try:
                self.pillow.process_change(change)
            except Exception:
                pillow_logging.exception("Unable to process change: %s", change.id)

            if i % 1000:
                pillow_logging.info("Processed %s docs", i)


def _clean_index(es, index_info):
    if es.indices.exists(index_info.index):
        es.indices.delete(index=index_info.index)


def _prepare_index_for_reindex(es, index_info):
    if not es.indices.exists(index_info.index):
        es.indices.create(index=index_info.index, body=index_info.meta)
    initialize_mapping_if_necessary(es, index_info)
    set_index_reindex_settings(es, index_info.index)


def _prepare_index_for_usage(es, index_info):
    set_index_normal_settings(es, index_info.index)
    es.indices.refresh(index_info.index)


def _set_checkpoint(pillow):
    checkpoint_value = pillow.get_change_feed().get_checkpoint_value()
    pillow_logging.info('setting checkpoint to {}'.format(checkpoint_value))
    pillow.checkpoint.update_to(checkpoint_value)


class ElasticPillowReindexer(PillowChangeProviderReindexer):
    in_place = False

{'fraction_non_alphanumeric': 0.07337001834250459, 'fraction_numerical': 0.0016675004168751042, 'mean_word_length': 3.708006279434851, 'pattern_counts': {'":': 0, '<': 3, '<?xml version=': 0, '>': 3, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '5995833', 'n_tokens_mistral': 1784, 'n_tokens_neox': 1739, 'n_words': 402}