<commit_before>from casexml.apps.case.models import CommCareCase
from corehq.apps.change_feed import data_sources
from corehq.apps.change_feed.connection import get_kafka_client_or_none
from corehq.apps.change_feed.document_types import get_doc_meta_object_from_document, \
    change_meta_from_doc_meta_and_document
from corehq.apps.change_feed.exceptions import MissingMetaInformationError
from corehq.apps.change_feed.producer import ChangeProducer
from corehq.apps.change_feed.topics import get_topic
from corehq.apps.domain.models import Domain
from corehq.apps.users.models import CommCareUser
from corehq.util.couchdb_management import couch_config
from corehq.util.soft_assert import soft_assert
from pillowtop.checkpoints.manager import PillowCheckpoint, PillowCheckpointEventHandler
from pillowtop.feed.couch import CouchChangeFeed
from pillowtop.pillow.interface import ConstructedPillow
from pillowtop.processors import PillowProcessor


class KafkaProcessor(PillowProcessor):
    """
    Processor that pushes changes to Kafka
    """
    def __init__(self, kafka, data_source_type, data_source_name):
        self._kafka = kafka
        self._producer = ChangeProducer(self._kafka)
        self._data_source_type = data_source_type
        self._data_source_name = data_source_name

    def process_change(self, pillow_instance, change, do_set_checkpoint=False):
        try:
            doc_meta = get_doc_meta_object_from_document(change.document)
            change_meta = change_meta_from_doc_meta_and_document(
                doc_meta=doc_meta,
                document=change.document,
                data_source_type=self._data_source_type,
                data_source_name=self._data_source_name,
                doc_id=change.id,
            )
        except MissingMetaInformationError:
            pass
        else:
            # change.deleted is used for hard deletions, from which we don't currently
            # get any metadata from so this should have raised a MissingMetaInformationError above
            _assert = soft_assert(to='@'.join(['czue', 'dim<EMAIL>']), send_to_ops=False)
            _assert(not change.deleted, u'change {} (meta: {}) should not have been deleted but was.'.format(
                change,
                change_meta.to_json()
            ))
            self._producer.send_change(get_topic(doc_meta), change_meta)


def get_default_couch_db_change_feed_pillow(pillow_id):
    return get_change_feed_pillow_for_db(pillow_id, CommCareCase.get_db())


def get_user_groups_db_kafka_pillow(pillow_id):<commit_msg>support hard deletions as deletions when we can get information about them<commit_after>from casexml.apps.case.models import CommCareCase
from corehq.apps.change_feed import data_sources
from corehq.apps.change_feed.connection import get_kafka_client_or_none
from corehq.apps.change_feed.document_types import get_doc_meta_object_from_document, \
    change_meta_from_doc_meta_and_document
from corehq.apps.change_feed.exceptions import MissingMetaInformationError
from corehq.apps.change_feed.producer import ChangeProducer
from corehq.apps.change_feed.topics import get_topic
from corehq.apps.domain.models import Domain
from corehq.apps.users.models import CommCareUser
from corehq.util.couchdb_management import couch_config
from pillowtop.checkpoints.manager import PillowCheckpoint, PillowCheckpointEventHandler
from pillowtop.feed.couch import CouchChangeFeed
from pillowtop.pillow.interface import ConstructedPillow
from pillowtop.processors import PillowProcessor


class KafkaProcessor(PillowProcessor):
    """
    Processor that pushes changes to Kafka
    """
    def __init__(self, kafka, data_source_type, data_source_name):
        self._kafka = kafka
        self._producer = ChangeProducer(self._kafka)
        self._data_source_type = data_source_type
        self._data_source_name = data_source_name

    def process_change(self, pillow_instance, change, do_set_checkpoint=False):
        try:
            doc_meta = get_doc_meta_object_from_document(change.document)
            change_meta = change_meta_from_doc_meta_and_document(
                doc_meta=doc_meta,
                document=change.document,
                data_source_type=self._data_source_type,
                data_source_name=self._data_source_name,
                doc_id=change.id,
            )
        except MissingMetaInformationError:
            pass
        else:
            # change.deleted is used for hard deletions whereas change_meta.is_deletion is for soft deletions.
            # from the consumer's perspective both should be counted as deletions so just "or" them
            # note: it is strange and hard to reproduce that the couch changes feed is providing a "doc"
            # along with a hard deletion, but it is doing that in the wild so we might as well support it.
            change_meta.is_deletion = change_meta.is_deletion or change.deleted
            self._producer.send_change(get_topic(doc_meta), change_meta)


def get_default_couch_db_change_feed_pillow(pillow_id):
    return get_change_feed_pillow_for_db(pillow_id, CommCareCase.get_db())


def get_user_groups_db_kafka_pillow(pillow_id):
{'fraction_non_alphanumeric': 0.05806575658527206, 'fraction_numerical': 0.0, 'mean_word_length': 3.759377859103385, 'pattern_counts': {'":': 0, '<': 4, '<?xml version=': 0, '>': 4, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 2}, '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': '7128886', 'n_tokens_mistral': 1558, 'n_tokens_neox': 1509, 'n_words': 335}