<commit_before>from __future__ import print_function
from __future__ import absolute_import
import weakref
from abc import ABCMeta, abstractmethod
from datetime import datetime, timedelta

import six

from corehq.util.pagination import PaginationEventHandler, TooManyRetries


class BulkProcessingFailed(Exception):
    pass


DOCS_SKIPPED_WARNING = """
        WARNING {} documents were not processed due to concurrent modification
        during migration. Run the migration again until you do not see this
        message.
        """

MIN_PROGRESS_INTERVAL = timedelta(minutes=5)


class BaseDocProcessor(six.with_metaclass(ABCMeta)):
    """Base class for processors that get passed"""

    def __enter__(self):
        pass

    def __exit__(self, exc_type, exc_val, exc_tb):
        pass

    def process_doc(self, doc):
        """Process a single document

        :param doc: The document dict to be processed.
        :returns: True if doc was processed successfully else False. If this returns False
        the document migration will be retried later.
        """
        raise NotImplementedError

    def process_bulk_docs(self, docs):
        """Process a batch of documents. The default implementation passes
        each doc in turn to ``process_doc``.

        :param docs: A list of document dicts to be processed.
        :returns: True if doc was processed successfully else False.
        If this returns False the processing will be halted.
        """
        return all(self.process_doc(doc) for doc in docs)

    def handle_skip(self, doc):
        """Called when a document is going to be skipped i.e. it has been
        retried > max_retries.

        :returns: True to indicate that the skip has been handled
                  or False to stop execution
        """
        return False

    def processing_complete(self, skipped):
        pass

    def should_process(self, doc):
        """
        :param doc: the document to filter
        :return: True if this doc should be migrated
        """
        return True


class ProcessorProgressLogger(object):
    def progress_starting(self, total, previously_visited):
        print("Processing {} documents{}: ...".format(
            total,
            " (~{} already processed)".format(previously_visited) if previously_visited else ""
        ))

    def document_skipped(self, doc_dict):
        print("Skip: {doc_type} {_id}".format(**doc_dict))

    def progress(self, processed, visited, total, time_elapsed, time_remaining):
        print("Processed {}/{} of {} documents in {} ({} remaining)"
              .format(processed, visited, total, time_elapsed, time_remaining))

    def progress_complete(self, processed, visited, total, previously_visited, filtered):
        print("Processed {}/{} of {} documents ({} previously processed, {} filtered out).".format(
            processed,
            visited,
            total,
            previously_visited,
            filtered
        ))


class DocumentProvider(six.with_metaclass(ABCMeta)):
    @abstractmethod
    def get_document_iterator(self, chunk_size, event_handler=None):
        """
        :param chunk_size: Maximum number of records to read from the database at one time
        :param event_handler: instance of ``PaginateViewLogHandler`` to be notified of view events.
        :return: an instance of ``ResumableFunctionIterator``
        """
        raise NotImplementedError

    @abstractmethod
    def get_total_document_count(self):
        """
        :return: the total count of documents expected
        """
        raise NotImplementedError


class DocumentProcessorController(object):
    """Process Docs

    :param document_provider: A ``DocumentProvider`` object
    :param doc_processor: A ``BaseDocProcessor`` object used to process documents.
    :param reset: Reset existing processor state (if any), causing all
    documents to be reconsidered for processing, if this is true.
    :param max_retry: Number of times to retry processing a document before giving up.
    :param chunk_size: Maximum number of records to read from couch at
    one time. It may be necessary to use a smaller chunk size if the
    records being processed are very large and the default chunk size of
    100 would exceed available memory.
    :param event_handler: A ``PaginateViewLogHandler`` object to be notified of pagination events.
    :param progress_logger: A ``ProcessorProgressLogger`` object to notify of progress events.
    """
    def __init__(self, document_provider, doc_processor, reset=False, max_retry=2,
                 chunk_size=100, event_handler=None, progress_logger=None):
        self.doc_processor = doc_processor
        self.reset = reset
        self.max_retry = max_retry
        self.chunk_size = chunk_size
        self.progress_logger = progress_logger or ProcessorProgressLogger()

        self.document_provider = document_provider

        self.document_iterator = self.document_provider.get_document_iterator(chunk_size, event_handler)

        self.visited = 0
        self.previously_visited = 0
        self.total = 0

        self.processed = 0
        self.skipped = 0

        self.start = None

    def has_started(self):
        return bool(self.document_iterator.get_iterator_detail('progress'))

    @property
    def session_visited(self):
        return self.visited - self.previously_visited

    @property
    def session_total(self):
        return self.total - self.previously_visited

    @property
    def attempted(self):
        return self.processed + self.skipped

    @property
    def timing(self):
        """Returns a tuple of (elapsed, remaining)"""
        elapsed = datetime.now() - self.start
        if self.session_visited > self.session_total:
            remaining = "?"
        else:
            session_remaining = self.session_total - self.session_visited
            remaining = elapsed / self.session_visited * session_remaining
        return elapsed, remaining
<commit_msg>one floating point division, one timedelta division<commit_after>from __future__ import print_function
from __future__ import absolute_import
from __future__ import division
import weakref
from abc import ABCMeta, abstractmethod
from datetime import datetime, timedelta

import six

from corehq.util.pagination import PaginationEventHandler, TooManyRetries


class BulkProcessingFailed(Exception):
    pass


DOCS_SKIPPED_WARNING = """
        WARNING {} documents were not processed due to concurrent modification
        during migration. Run the migration again until you do not see this
        message.
        """

MIN_PROGRESS_INTERVAL = timedelta(minutes=5)


class BaseDocProcessor(six.with_metaclass(ABCMeta)):
    """Base class for processors that get passed"""

    def __enter__(self):
        pass

    def __exit__(self, exc_type, exc_val, exc_tb):
        pass

    def process_doc(self, doc):
        """Process a single document

        :param doc: The document dict to be processed.
        :returns: True if doc was processed successfully else False. If this returns False
        the document migration will be retried later.
        """
        raise NotImplementedError

    def process_bulk_docs(self, docs):
        """Process a batch of documents. The default implementation passes
        each doc in turn to ``process_doc``.

        :param docs: A list of document dicts to be processed.
        :returns: True if doc was processed successfully else False.
        If this returns False the processing will be halted.
        """
        return all(self.process_doc(doc) for doc in docs)

    def handle_skip(self, doc):
        """Called when a document is going to be skipped i.e. it has been
        retried > max_retries.

        :returns: True to indicate that the skip has been handled
                  or False to stop execution
        """
        return False

    def processing_complete(self, skipped):
        pass

    def should_process(self, doc):
        """
        :param doc: the document to filter
        :return: True if this doc should be migrated
        """
        return True


class ProcessorProgressLogger(object):
    def progress_starting(self, total, previously_visited):
        print("Processing {} documents{}: ...".format(
            total,
            " (~{} already processed)".format(previously_visited) if previously_visited else ""
        ))

    def document_skipped(self, doc_dict):
        print("Skip: {doc_type} {_id}".format(**doc_dict))

    def progress(self, processed, visited, total, time_elapsed, time_remaining):
        print("Processed {}/{} of {} documents in {} ({} remaining)"
              .format(processed, visited, total, time_elapsed, time_remaining))

    def progress_complete(self, processed, visited, total, previously_visited, filtered):
        print("Processed {}/{} of {} documents ({} previously processed, {} filtered out).".format(
            processed,
            visited,
            total,
            previously_visited,
            filtered
        ))


class DocumentProvider(six.with_metaclass(ABCMeta)):
    @abstractmethod
    def get_document_iterator(self, chunk_size, event_handler=None):
        """
        :param chunk_size: Maximum number of records to read from the database at one time
        :param event_handler: instance of ``PaginateViewLogHandler`` to be notified of view events.
        :return: an instance of ``ResumableFunctionIterator``
        """
        raise NotImplementedError

    @abstractmethod
    def get_total_document_count(self):
        """
        :return: the total count of documents expected
        """
        raise NotImplementedError


class DocumentProcessorController(object):
    """Process Docs

    :param document_provider: A ``DocumentProvider`` object
    :param doc_processor: A ``BaseDocProcessor`` object used to process documents.
    :param reset: Reset existing processor state (if any), causing all
    documents to be reconsidered for processing, if this is true.
    :param max_retry: Number of times to retry processing a document before giving up.
    :param chunk_size: Maximum number of records to read from couch at
    one time. It may be necessary to use a smaller chunk size if the
    records being processed are very large and the default chunk size of
    100 would exceed available memory.
    :param event_handler: A ``PaginateViewLogHandler`` object to be notified of pagination events.
    :param progress_logger: A ``ProcessorProgressLogger`` object to notify of progress events.
    """
    def __init__(self, document_provider, doc_processor, reset=False, max_retry=2,
                 chunk_size=100, event_handler=None, progress_logger=None):
        self.doc_processor = doc_processor
        self.reset = reset
        self.max_retry = max_retry
        self.chunk_size = chunk_size
        self.progress_logger = progress_logger or ProcessorProgressLogger()

        self.document_provider = document_provider

        self.document_iterator = self.document_provider.get_document_iterator(chunk_size, event_handler)

        self.visited = 0
        self.previously_visited = 0
        self.total = 0

        self.processed = 0
        self.skipped = 0

        self.start = None

    def has_started(self):
        return bool(self.document_iterator.get_iterator_detail('progress'))

    @property
    def session_visited(self):
        return self.visited - self.previously_visited

    @property
    def session_total(self):
        return self.total - self.previously_visited

    @property
    def attempted(self):
        return self.processed + self.skipped

    @property
    def timing(self):
        """Returns a tuple of (elapsed, remaining)"""
        elapsed = datetime.now() - self.start
        if self.session_visited > self.session_total:
            remaining = "?"
        else:
            session_remaining = self.session_total - self.session_visited
            remaining = elapsed // self.session_visited * session_remaining
        return elapsed, remaining

{'fraction_non_alphanumeric': 0.07602677464672342, 'fraction_numerical': 0.002148582761755227, 'mean_word_length': 2.839467005076142, 'pattern_counts': {'":': 0, '<': 3, '<?xml version=': 0, '>': 7, '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': '5092949', 'n_tokens_mistral': 3189, 'n_tokens_neox': 3071, 'n_words': 1068}