<commit_before>import typing
from google.cloud import datastore  # pylint:disable=import-error
from google.cloud import storage  # pylint:disable=import-error
from scrapy import exporters
from scrapy import signals
from justice.justice.items import ActItem


class DataStoreExporter(exporters.BaseItemExporter):

    def __init__(self, ds: datastore.Client, st: storage.Client, **kwargs) -> None:
        super().__init__(**kwargs)
        self.__datastore = ds
        self.__bucket = st.get_bucket('gitlawca.appspot.com')
        self.export_empty_fields = True

    def __act_in_datastore(self, item: ActItem) -> typing.Optional[datastore.Entity]:
        query = self.__datastore.query(kind='Act')
        query.add_filter('code', '=', item['code'])
        acts = list(query.fetch())
        if acts:
            return acts[0]

    def __store_raw_in_storage(self, item: ActItem) -> str:
        path = 'acts/raw/{}'.format(item['code'])
        blob = self.__bucket.blob(path)
        blob.upload_from_string(item['body'])
        return path

    def __store_act_in_datastore(self, item: ActItem) -> datastore.Entity:
        key = self.__datastore.key('Act')
        task = datastore.Entity(key, exclude_from_indexes=('body',))

        task.update({
            'title': item['title'],
            'code': item['code'],
        })
        return task

    def export_item(self, item: ActItem) -> ActItem:
        act_entity = self.__act_in_datastore(item)

        if act_entity:
            if not act_entity['raw_blob']:
                act_entity['raw_blob'] = self.__store_raw_in_storage(item)
                self.__datastore.put(act_entity)
        else:
            act_entity = self.__store_act_in_datastore(item)
            act_entity['raw_blob'] = self.__store_raw_in_storage(item)
            self.__datastore.put(act_entity)

        return item


class JusticePipeline(object):

    def __init__(self):
        self.exporter = None  # type: DataStoreExporter

    @classmethod
    def from_crawler(cls, crawler):
        pipeline = cls()
        crawler.signals.connect(pipeline.spider_opened, signals.spider_opened)
        crawler.signals.connect(pipeline.spider_closed, signals.spider_closed)
        return pipeline

    def spider_opened(self, spider):  # pylint:disable=unused-argument
        dstore = datastore.Client('gitlawca')
        stor = storage.Client('gitlawca')

        self.exporter = DataStoreExporter(dstore, stor)
        self.exporter.start_exporting()

    def spider_closed(self, _):
        self.exporter.finish_exporting()

    def process_item(self, item, _):
        self.exporter.export_item(item)
        return item<commit_msg>fix spider_closed<commit_after>import typing
from google.cloud import datastore  # pylint:disable=import-error
from google.cloud import storage  # pylint:disable=import-error
from scrapy import exporters
from scrapy import signals
from justice.justice.items import ActItem


class DataStoreExporter(exporters.BaseItemExporter):

    def __init__(self, ds: datastore.Client, st: storage.Client, **kwargs) -> None:
        super().__init__(**kwargs)
        self.__datastore = ds
        self.__bucket = st.get_bucket('gitlawca.appspot.com')
        self.export_empty_fields = True

    def __act_in_datastore(self, item: ActItem) -> typing.Optional[datastore.Entity]:
        query = self.__datastore.query(kind='Act')
        query.add_filter('code', '=', item['code'])
        acts = list(query.fetch())
        if acts:
            return acts[0]

    def __store_raw_in_storage(self, item: ActItem) -> str:
        path = 'acts/raw/{}'.format(item['code'])
        blob = self.__bucket.blob(path)
        blob.upload_from_string(item['body'])
        return path

    def __store_act_in_datastore(self, item: ActItem) -> datastore.Entity:
        key = self.__datastore.key('Act')
        task = datastore.Entity(key, exclude_from_indexes=('body',))

        task.update({
            'title': item['title'],
            'code': item['code'],
        })
        return task

    def export_item(self, item: ActItem) -> ActItem:
        act_entity = self.__act_in_datastore(item)

        if act_entity:
            if not act_entity['raw_blob']:
                act_entity['raw_blob'] = self.__store_raw_in_storage(item)
                self.__datastore.put(act_entity)
        else:
            act_entity = self.__store_act_in_datastore(item)
            act_entity['raw_blob'] = self.__store_raw_in_storage(item)
            self.__datastore.put(act_entity)

        return item


class JusticePipeline(object):

    def __init__(self):
        self.exporter = None  # type: typing.Optional[DataStoreExporter]

    @classmethod
    def from_crawler(cls, crawler):
        pipeline = cls()
        crawler.signals.connect(pipeline.spider_opened, signals.spider_opened)
        crawler.signals.connect(pipeline.spider_closed, signals.spider_closed)
        return pipeline

    def spider_opened(self, spider):  # pylint:disable=unused-argument
        dstore = datastore.Client('gitlawca')
        stor = storage.Client('gitlawca')

        self.exporter = DataStoreExporter(dstore, stor)
        self.exporter.start_exporting()

    def spider_closed(self, spider):  # pylint:disable=unused-argument
        self.exporter.finish_exporting()

    def process_item(self, item, _):
        self.exporter.export_item(item)
        return item
{'fraction_non_alphanumeric': 0.11145910095799558, 'fraction_numerical': 0.00036845983787767134, 'mean_word_length': 2.9142033165104544, 'pattern_counts': {'":': 0, '<': 3, '<?xml version=': 0, '>': 13, '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': '1334390', 'n_tokens_mistral': 1704, 'n_tokens_neox': 1659, 'n_words': 300}