<commit_before>from .interface import PillowProcessor
from pillowtop.listener import send_to_elasticsearch


class ElasticProcessor(PillowProcessor):

    def __init__(self, elasticseach, index_meta, doc_prep_fn):
        self.elasticsearch = elasticseach
        self.index_meta = index_meta
        self.doc_transform_fn = doc_prep_fn

    def es_getter(self):
        return self.elasticsearch

    def process_change(self, pillow_instance, change, do_set_checkpoint):
        # todo: if deletion - delete
        # prepare doc for es
        doc_ready_to_save = self.doc_transform_fn(change.get_document())
        # send it across
        send_to_elasticsearch(
            index=self.index_meta.index,
            doc_type=self.index_meta.type,
            doc_id=change.id,
            es_getter=self.es_getter,
            name=pillow_instance.get_name(),
            data=doc_ready_to_save,
            update=self.elasticsearch.exists(self.index_meta.index, self.index_meta.type, change.id),
        )
<commit_msg>don't try to process missing/deleted docs<commit_after>from .interface import PillowProcessor
from pillowtop.listener import send_to_elasticsearch
from pillowtop.logger import pillow_logging


class ElasticProcessor(PillowProcessor):

    def __init__(self, elasticseach, index_meta, doc_prep_fn):
        self.elasticsearch = elasticseach
        self.index_meta = index_meta
        self.doc_transform_fn = doc_prep_fn

    def es_getter(self):
        return self.elasticsearch

    def process_change(self, pillow_instance, change, do_set_checkpoint):
        # todo: if deletion - delete
        # prepare doc for es
        doc = change.get_document()
        if doc is None:
            pillow_logging.warning("Unable to get document from change: {}".format(change))
            return

        doc_ready_to_save = self.doc_transform_fn(doc)
        # send it across
        send_to_elasticsearch(
            index=self.index_meta.index,
            doc_type=self.index_meta.type,
            doc_id=change.id,
            es_getter=self.es_getter,
            name=pillow_instance.get_name(),
            data=doc_ready_to_save,
            update=self.elasticsearch.exists(self.index_meta.index, self.index_meta.type, change.id),
        )

{'fraction_non_alphanumeric': 0.07563764291996482, 'fraction_numerical': 0.0, 'mean_word_length': 2.7790697674418605, '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': '2975478', 'n_tokens_mistral': 675, 'n_tokens_neox': 664, 'n_words': 129}