<commit_before>from __future__ import absolute_import
from __future__ import division


import logging
import logging.config
import time

import redis
import slumber
import yaml

from apscheduler.scheduler import Scheduler

from conveyor.processor import Processor

# @@@ Switch all Urls to SSL


logger = logging.getLogger(__name__)


class Conveyor(object):

    def __init__(self, config_file=None, *args, **kwargs):
        super(Conveyor, self).__init__(*args, **kwargs)

        if config_file is None:
            config_file = "config.yml"

        with open(config_file) as f:
            self.config = yaml.safe_load(f.read())

        logging.config.dictConfig(self.config["logging"])

        self.redis = redis.StrictRedis(**self.config.get("redis", {}).get("connection", {}))

    def run(self):
        self.scheduler = Scheduler()
        self.scheduler.add_interval_job(self.process, **self.config["conveyor"]["schedule"])
        self.scheduler.start()

        try:
            while True:
                time.sleep(999)
        except KeyboardInterrupt:
            logger.info("Shutting down Conveyor...")
            self.scheduler.shutdown(wait=False)

    def process(self):
        warehouse = slumber.API(
                        self.config["conveyor"]["warehouse"]["url"],
                        auth=(
                            self.config["conveyor"]["warehouse"]["auth"]["username"],
                            self.config["conveyor"]["warehouse"]["auth"]["password"],
                        )
                    )

        processor_class = self.get_processor_class()
        processor = processor_class(
                        index=self.config["conveyor"]["index"],
                        warehouse=warehouse,
                        store=self.redis,
                        store_prefix=self.config.get("redis", {}).get("prefix", None)
                    )

        processor.process()

    def get_processor_class(self):
        if self.previous_time is None:
            # This is the first time we've ran so we need to do a bulk import
            raise Exception("Bulk import must be run prior to running Conveyor.")

        return Processor

    @property
    def previous_time(self):
        return None<commit_msg>Implement the ability to see if Bulk sync had been run<commit_after>from __future__ import absolute_import
from __future__ import division


import logging
import logging.config
import time

import redis
import slumber
import yaml

from apscheduler.scheduler import Scheduler

from conveyor.processor import Processor, get_key

# @@@ Switch all Urls to SSL


logger = logging.getLogger(__name__)


class Conveyor(object):

    def __init__(self, config_file=None, *args, **kwargs):
        super(Conveyor, self).__init__(*args, **kwargs)

        if config_file is None:
            config_file = "config.yml"

        with open(config_file) as f:
            self.config = yaml.safe_load(f.read())

        logging.config.dictConfig(self.config["logging"])

        self.redis = redis.StrictRedis(**self.config.get("redis", {}).get("connection", {}))

    def run(self):
        self.scheduler = Scheduler()
        self.scheduler.add_interval_job(self.process, **self.config["conveyor"]["schedule"])
        self.scheduler.start()

        try:
            while True:
                time.sleep(999)
        except KeyboardInterrupt:
            logger.info("Shutting down Conveyor...")
            self.scheduler.shutdown(wait=False)

    def process(self):
        warehouse = slumber.API(
                        self.config["conveyor"]["warehouse"]["url"],
                        auth=(
                            self.config["conveyor"]["warehouse"]["auth"]["username"],
                            self.config["conveyor"]["warehouse"]["auth"]["password"],
                        )
                    )

        processor_class = self.get_processor_class()
        processor = processor_class(
                        index=self.config["conveyor"]["index"],
                        warehouse=warehouse,
                        store=self.redis,
                        store_prefix=self.config.get("redis", {}).get("prefix", None)
                    )

        processor.process()

    def get_processor_class(self):
        if self.previous_time is None:
            # This is the first time we've ran so we need to do a bulk import
            raise Exception("Bulk import must be run prior to running Conveyor.")

        return Processor

    @property
    def previous_time(self):
        return self.redis.get(get_key(self.config.get("redis", {}).get("prefix", None), "pypi:since"))
{'fraction_non_alphanumeric': 0.11308631398586422, 'fraction_numerical': 0.001285071749839366, 'mean_word_length': 2.1133333333333333, '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': '757281', 'n_tokens_mistral': 1286, 'n_tokens_neox': 1259, 'n_words': 283}