<commit_before>from functools import wraps
import logging
from elasticsearch.exceptions import RequestError, ConnectionError, NotFoundError, ConflictError
from psycopg2._psycopg import InterfaceError as Psycopg2InterfaceError
from django.db.utils import InterfaceError as DjangoInterfaceError
from datetime import datetime, timedelta
import traceback
import math
import time

import simplejson

from dimagi.utils.decorators.memoized import memoized
from dimagi.utils.couch import LockManager
from pillowtop.checkpoints.manager import PillowCheckpoint, get_default_django_checkpoint_for_legacy_pillow_class
from pillowtop.checkpoints.util import get_machine_id, construct_checkpoint_doc_id_from_name
from pillowtop.const import CHECKPOINT_FREQUENCY
from pillowtop.couchdb import CachedCouchDB

from django import db
from pillowtop.dao.couch import CouchDocumentStore
from pillowtop.es_utils import completely_initialize_pillow_index, doc_exists
from pillowtop.feed.couch import CouchChangeFeed
from pillowtop.logger import pillow_logging
from pillowtop.pillow.interface import PillowBase
from pillowtop.utils import prepare_bulk_payloads

try:
    from corehq.util.soft_assert import soft_assert
    _assert = soft_assert(to='@'.join(['czue', 'dimagi.com']), fail_if_debug=True)
except ImportError:
    # hack for dependency resolution if corehq not available
    _assert = lambda assertion, message: None


WAIT_HEARTBEAT = 10000
CHANGES_TIMEOUT = 60000
RETRY_INTERVAL = 2  # seconds, exponentially increasing
MAX_RETRIES = 4  # exponential factor threshold for alerts


class PillowtopIndexingError(Exception):
    pass


class PillowtopNetworkError(Exception):
    pass


def ms_from_timedelta(td):
    """
    Given a timedelta object, returns a float representing milliseconds
    """
    return (td.seconds * 1000) + (td.microseconds / 1000.0)


def lock_manager(obj):
    if isinstance(obj, LockManager):
        return obj
    else:
        return LockManager(obj, None)


class BasicPillow(PillowBase):
    """
    BasicPillow is actually a CouchPillow. PillowBase defines the actual interface.
    """
    checkpoint_frequency = CHECKPOINT_FREQUENCY
    couch_filter = None  # string for filter if needed
    extra_args = {}  # filter args if needed
    document_class = None  # couchdbkit Document class
    _couch_db = None
    include_docs = True
    use_locking = False

    def __init__(self, couch_db=None, document_class=None, checkpoint=None, change_feed=None):
        if document_class:
            self.document_class = document_class

        self._couch_db = couch_db
        self._checkpoint = checkpoint
        self._change_feed = change_feed

        if self.use_locking:
            # document_class must be a CouchDocLockableMixIn
            assert hasattr(self.document_class, 'get_obj_lock_by_id')

    @property
    def pillow_id(self):
        # for legacy reasons, by default a Pillow's ID is just it's class name
        return self.__class__.__name__

    def get_couch_db(self):
        if self._couch_db is None:
            self._couch_db = self.get_default_couch_db()
        return self._couch_db

    def set_couch_db(self, couch_db):
        self._couch_db = couch_db

    def get_default_couch_db(self):
        return self.document_class.get_db() if self.document_class else None

    @property
    def couch_db(self):
        _assert(False, 'People should not be using the couch_db properties!')
        return self.get_couch_db()

    @couch_db.setter
    def couch_db(self, value):
        _assert(False, 'People should not be using the couch_db properties!')
        self._couch_db = value

    @property
    def document_store(self):
        return CouchDocumentStore(self.get_couch_db())

    @property
    def checkpoint(self):
        if self._checkpoint is None:
            self._checkpoint = self._get_default_checkpoint()
        return self._checkpoint

    def _get_default_checkpoint(self):
        return PillowCheckpoint(
            construct_checkpoint_doc_id_from_name(self.get_name()),
        )

    def get_change_feed(self):
        if self._change_feed is None:
            self._change_feed = self._get_default_change_feed()
        return self._change_feed

    def _get_default_change_feed(self):
        return CouchChangeFeed(
            couch_db=self.get_couch_db(),
            include_docs=self.include_docs,
            couch_filter=self.couch_filter,
            extra_couch_view_params=self.extra_args
        )

    @memoized
    def get_name(self):
        return self.get_legacy_name()

    @classmethod
    def get_legacy_name(cls):
        return "%s.%s.%s" % (cls._get_base_name(), cls.__name__, get_machine_id())

    @classmethod
    def _get_base_name(cls):
        return cls.__module__

    def process_change(self, change, is_retry_attempt=False):
        """
        Parent processsor for a pillow class - this should not be overridden.
        This workflow is made for the situation where 1 change yields 1 transport/transaction
        """
        with lock_manager(self.change_trigger(change)) as t:
            if t is not None:
                tr = self.change_transform(t)
                if tr is not None:
                    self.change_transport(tr)

    def fire_change_processed_event(self, change, context):
        if context.changes_seen % self.checkpoint_frequency == 0 and context.do_set_checkpoint:
            self.set_checkpoint(change)

    def change_trigger(self, changes_dict):
        """
        Step one of pillowtop process
        For a given _changes indicator, the changes dict (the id, _rev) is sent here.

        Note, a couch _changes line is: {'changes': [], 'id': 'guid',  'seq': <int>}
        a 'deleted': True might be there too

        whereas a doc_dict is _id
        Should return a doc_dict
        """
        if changes_dict.get('deleted', False):
            # override deleted behavior on consumers that deal with deletions
            return None
        id = changes_dict['id']
        if self.use_locking:
            lock = self.document_class.get_obj_lock_by_id(id)
            lock.acquire()
            return LockManager(self.get_couch_db().open_doc(id), lock)
        elif changes_dict.get('doc', None) is not None:
            return changes_dict['doc']
        elif hasattr(changes_dict, 'get_document') and changes_dict.get_document():
            return changes_dict.get_document()
        else:
            # todo: remove this in favor of always using get_document() above
            return self.get_couch_db().open_doc(id)

    def change_transform(self, doc_dict):
        """
        Step two of the pillowtop processor:
        Process/transform doc_dict if needed - by default, return the doc_dict passed.
        """
        return doc_dict

    def change_transport(self, doc_dict):
        """
        Step three of the pillowtop processor:<commit_msg>treat missing docs like deletions in pillows<commit_after>from functools import wraps
import logging
from couchdbkit.exceptions import ResourceNotFound
from elasticsearch.exceptions import RequestError, ConnectionError, NotFoundError, ConflictError
from psycopg2._psycopg import InterfaceError as Psycopg2InterfaceError
from django.db.utils import InterfaceError as DjangoInterfaceError
from datetime import datetime, timedelta
import traceback
import math
import time

import simplejson

from dimagi.utils.decorators.memoized import memoized
from dimagi.utils.couch import LockManager
from pillowtop.checkpoints.manager import PillowCheckpoint, get_default_django_checkpoint_for_legacy_pillow_class
from pillowtop.checkpoints.util import get_machine_id, construct_checkpoint_doc_id_from_name
from pillowtop.const import CHECKPOINT_FREQUENCY
from pillowtop.couchdb import CachedCouchDB

from django import db
from pillowtop.dao.couch import CouchDocumentStore
from pillowtop.es_utils import completely_initialize_pillow_index, doc_exists
from pillowtop.feed.couch import CouchChangeFeed
from pillowtop.logger import pillow_logging
from pillowtop.pillow.interface import PillowBase
from pillowtop.utils import prepare_bulk_payloads

try:
    from corehq.util.soft_assert import soft_assert
    _assert = soft_assert(to='@'.join(['czue', 'dimagi.com']), fail_if_debug=True)
except ImportError:
    # hack for dependency resolution if corehq not available
    _assert = lambda assertion, message: None


WAIT_HEARTBEAT = 10000
CHANGES_TIMEOUT = 60000
RETRY_INTERVAL = 2  # seconds, exponentially increasing
MAX_RETRIES = 4  # exponential factor threshold for alerts


class PillowtopIndexingError(Exception):
    pass


class PillowtopNetworkError(Exception):
    pass


def ms_from_timedelta(td):
    """
    Given a timedelta object, returns a float representing milliseconds
    """
    return (td.seconds * 1000) + (td.microseconds / 1000.0)


def lock_manager(obj):
    if isinstance(obj, LockManager):
        return obj
    else:
        return LockManager(obj, None)


class BasicPillow(PillowBase):
    """
    BasicPillow is actually a CouchPillow. PillowBase defines the actual interface.
    """
    checkpoint_frequency = CHECKPOINT_FREQUENCY
    couch_filter = None  # string for filter if needed
    extra_args = {}  # filter args if needed
    document_class = None  # couchdbkit Document class
    _couch_db = None
    include_docs = True
    use_locking = False

    def __init__(self, couch_db=None, document_class=None, checkpoint=None, change_feed=None):
        if document_class:
            self.document_class = document_class

        self._couch_db = couch_db
        self._checkpoint = checkpoint
        self._change_feed = change_feed

        if self.use_locking:
            # document_class must be a CouchDocLockableMixIn
            assert hasattr(self.document_class, 'get_obj_lock_by_id')

    @property
    def pillow_id(self):
        # for legacy reasons, by default a Pillow's ID is just it's class name
        return self.__class__.__name__

    def get_couch_db(self):
        if self._couch_db is None:
            self._couch_db = self.get_default_couch_db()
        return self._couch_db

    def set_couch_db(self, couch_db):
        self._couch_db = couch_db

    def get_default_couch_db(self):
        return self.document_class.get_db() if self.document_class else None

    @property
    def couch_db(self):
        _assert(False, 'People should not be using the couch_db properties!')
        return self.get_couch_db()

    @couch_db.setter
    def couch_db(self, value):
        _assert(False, 'People should not be using the couch_db properties!')
        self._couch_db = value

    @property
    def document_store(self):
        return CouchDocumentStore(self.get_couch_db())

    @property
    def checkpoint(self):
        if self._checkpoint is None:
            self._checkpoint = self._get_default_checkpoint()
        return self._checkpoint

    def _get_default_checkpoint(self):
        return PillowCheckpoint(
            construct_checkpoint_doc_id_from_name(self.get_name()),
        )

    def get_change_feed(self):
        if self._change_feed is None:
            self._change_feed = self._get_default_change_feed()
        return self._change_feed

    def _get_default_change_feed(self):
        return CouchChangeFeed(
            couch_db=self.get_couch_db(),
            include_docs=self.include_docs,
            couch_filter=self.couch_filter,
            extra_couch_view_params=self.extra_args
        )

    @memoized
    def get_name(self):
        return self.get_legacy_name()

    @classmethod
    def get_legacy_name(cls):
        return "%s.%s.%s" % (cls._get_base_name(), cls.__name__, get_machine_id())

    @classmethod
    def _get_base_name(cls):
        return cls.__module__

    def process_change(self, change, is_retry_attempt=False):
        """
        Parent processsor for a pillow class - this should not be overridden.
        This workflow is made for the situation where 1 change yields 1 transport/transaction
        """
        with lock_manager(self.change_trigger(change)) as t:
            if t is not None:
                tr = self.change_transform(t)
                if tr is not None:
                    self.change_transport(tr)

    def fire_change_processed_event(self, change, context):
        if context.changes_seen % self.checkpoint_frequency == 0 and context.do_set_checkpoint:
            self.set_checkpoint(change)

    def change_trigger(self, changes_dict):
        """
        Step one of pillowtop process
        For a given _changes indicator, the changes dict (the id, _rev) is sent here.

        Note, a couch _changes line is: {'changes': [], 'id': 'guid',  'seq': <int>}
        a 'deleted': True might be there too

        whereas a doc_dict is _id
        Should return a doc_dict
        """
        if changes_dict.get('deleted', False):
            # override deleted behavior on consumers that deal with deletions
            return None
        id = changes_dict['id']
        if self.use_locking:
            lock = self.document_class.get_obj_lock_by_id(id)
            lock.acquire()
            return LockManager(self.get_couch_db().open_doc(id), lock)
        elif changes_dict.get('doc', None) is not None:
            return changes_dict['doc']
        elif hasattr(changes_dict, 'get_document') and changes_dict.get_document():
            return changes_dict.get_document()
        else:
            # todo: remove this in favor of always using get_document() above
            try:
                return self.get_couch_db().open_doc(id)
            except ResourceNotFound:
                # doc was likely hard-deleted. treat like a deletion
                return None

    def change_transform(self, doc_dict):
        """
        Step two of the pillowtop processor:
        Process/transform doc_dict if needed - by default, return the doc_dict passed.
        """
        return doc_dict

    def change_transport(self, doc_dict):
        """
        Step three of the pillowtop processor:
{'fraction_non_alphanumeric': 0.06974927196533845, 'fraction_numerical': 0.003693444136657433, 'mean_word_length': 3.015972618368511, 'pattern_counts': {'":': 0, '<': 5, '<?xml version=': 0, '>': 5, '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': '7138665', 'n_tokens_mistral': 4166, 'n_tokens_neox': 4003, 'n_words': 1108}