<commit_before>#!/usr/bin/env python
import json
import logging
import sys
import time
import dateutil.parser as parser

from redis import StrictRedis, ConnectionError
from kafka import SimpleProducer, KafkaClient
from kafka.common import (NotLeaderForPartitionError, KafkaUnavailableError)
from birdy.twitter import StreamClient, TwitterApiError

from settings import (KAFKA_HOSTS, KAFKA_TOPIC, APIKEYS, TRAPTOR_ID,
                      TRAPTOR_TYPE, REDIS_HOST)

logging.basicConfig(level=logging.INFO)

# logging = logging.getLogger(__name__)
# handler = logging.StreamHandler()
# formatter = logging.Formatter(
#         '%(asctime)s %(name)-4s %(levelname)-4s %(message)s')
# handler.setFormatter(formatter)
# logging.addHandler(handler)
# logging.setLevel(logging.INFO)


def get_redis_twitter_ids(traptor_type=TRAPTOR_TYPE, traptor_id=TRAPTOR_ID,
                          redis_host=REDIS_HOST):
    """ Return a list of twitter ids from the service server.  This function
        expects that the redis keys are set up like follows:

        traptor-<traptor_type>:<crawler_num>

        For example,

        traptor-follow:0
        traptor-follow:1

        In the case of the 'follow' twitter streaming, each traptor may only
        follow 5000 twitter ids, as per the Twitter API.
    """
    # This line is lazy, nothing touches Redis until a command is issued
    r = StrictRedis(host=redis_host, port=6379, db=0)

    twids = []
    redis_key = 'traptor-{0}:{1}'.format(traptor_type, traptor_id)
    try:
        for idx, key in enumerate(r.sscan_iter(redis_key)):
            if idx < 5000:
                twids.append(key)
                logging.debug('{0}: {1}'.format(idx, key))
    except ConnectionError as e:
        logging.critical(e)
        sys.exit(3)  # Special error code to track known failures
    return twids


def create_kafka_producer(kafka_hosts=KAFKA_HOSTS, kafka_topic=KAFKA_TOPIC):
    """ Create a kafka producer.
        If it cannot find one it will exit with error code 3.
    """
    try:
        client = KafkaClient(hosts=kafka_hosts)
        producer = SimpleProducer(client)
    except KafkaUnavailableError as e:
        logging.critical(e)
        sys.exit(3)
    try:
        client.ensure_topic_exists(kafka_topic)
    except:
        raise

    return producer


def create_birdy_stream(rules,
                        traptor_type=TRAPTOR_TYPE,
                        traptor_id=TRAPTOR_ID,
                        ):
    """ Set up a birdy twitter stream.
        If there is a TwitterApiError it will exit with status code 3.
        This was done to prevent services like supervisor from automatically
        restart the process causing the twitter API to get locked out.
    """
    # Check traptor_type
    if traptor_type == 'follow':
        # Set up a birdy twitter streaming client
        client = StreamClient(
                              APIKEYS['CONSUMER_KEY'],
                              APIKEYS['CONSUMER_SECRET'],
                              APIKEYS['ACCESS_TOKEN'],
                              APIKEYS['ACCESS_TOKEN_SECRET']
                          )
        # Try to set up a twitter stream using twitter id list
        try:
            resource = client.stream.statuses.filter.post(follow=rules)
            return resource
        except TwitterApiError as e:
            logging.critical(e)
            sys.exit(3)
    else:
        logging.critical('That traptor type has not been implemented yet')
        sys.exit(3)


def run():
    # Grab a list of twitter ids from the get_redis_twitter_ids function
    twids_str = ','.join(get_redis_twitter_ids())

    # Set up Kafka producer
    producer = create_kafka_producer()

    # Set up a birdy streaming client
    birdyclient = create_birdy_stream(twids_str)

    # Iterate through the twitter results
    for data in birdyclient.stream():
        logging.info(json.dumps(data.get('text')))
        logging.debug(json.dumps(data))
        # Convert date to ISO format
        _time = data.get('created_at')
        if _time:
            data['created_at'] = parser.parse(_time).isoformat()

        try:
            producer.send_messages(KAFKA_TOPIC, json.dumps(data))
        except:
            logging.error('Could not write to kafka topic {0}'.format(
                         KAFKA_TOPIC))
            sys.exit(3)


def main():
    run()

if __name__ == '__main__':
    main()<commit_msg>round and round we go with the logger<commit_after>#!/usr/bin/env python
import json
import logging
import sys
import time
import dateutil.parser as parser

from redis import StrictRedis, ConnectionError
from kafka import SimpleProducer, KafkaClient
from kafka.common import (NotLeaderForPartitionError, KafkaUnavailableError)
from birdy.twitter import StreamClient, TwitterApiError

from settings import (KAFKA_HOSTS, KAFKA_TOPIC, APIKEYS, TRAPTOR_ID,
                      TRAPTOR_TYPE, REDIS_HOST)

# logging.basicConfig(level=logging.INFO)

logger = logging.getLogger(__name__)
handler = logging.StreamHandler()
formatter = logging.Formatter(
        '%(asctime)s %(name)-4s %(levelname)-4s %(message)s')
handler.setFormatter(formatter)
logger.addHandler(handler)
logger.setLevel(logging.DEBUG)


def get_redis_twitter_ids(traptor_type=TRAPTOR_TYPE, traptor_id=TRAPTOR_ID,
                          redis_host=REDIS_HOST):
    """ Return a list of twitter ids from the service server.  This function
        expects that the redis keys are set up like follows:

        traptor-<traptor_type>:<crawler_num>

        For example,

        traptor-follow:0
        traptor-follow:1

        In the case of the 'follow' twitter streaming, each traptor may only
        follow 5000 twitter ids, as per the Twitter API.
    """
    # This line is lazy, nothing touches Redis until a command is issued
    r = StrictRedis(host=redis_host, port=6379, db=0)

    twids = []
    redis_key = 'traptor-{0}:{1}'.format(traptor_type, traptor_id)
    try:
        for idx, key in enumerate(r.sscan_iter(redis_key)):
            if idx < 5000:
                twids.append(key)
                logger.debug('{0}: {1}'.format(idx, key))
    except ConnectionError as e:
        logger.critical(e)
        sys.exit(3)  # Special error code to track known failures
    return twids


def create_kafka_producer(kafka_hosts=KAFKA_HOSTS, kafka_topic=KAFKA_TOPIC):
    """ Create a kafka producer.
        If it cannot find one it will exit with error code 3.
    """
    try:
        client = KafkaClient(hosts=kafka_hosts)
        producer = SimpleProducer(client)
    except KafkaUnavailableError as e:
        logger.critical(e)
        sys.exit(3)
    try:
        client.ensure_topic_exists(kafka_topic)
    except:
        raise

    return producer


def create_birdy_stream(rules,
                        traptor_type=TRAPTOR_TYPE,
                        traptor_id=TRAPTOR_ID,
                        ):
    """ Set up a birdy twitter stream.
        If there is a TwitterApiError it will exit with status code 3.
        This was done to prevent services like supervisor from automatically
        restart the process causing the twitter API to get locked out.
    """
    # Check traptor_type
    if traptor_type == 'follow':
        # Set up a birdy twitter streaming client
        client = StreamClient(
                              APIKEYS['CONSUMER_KEY'],
                              APIKEYS['CONSUMER_SECRET'],
                              APIKEYS['ACCESS_TOKEN'],
                              APIKEYS['ACCESS_TOKEN_SECRET']
                          )
        # Try to set up a twitter stream using twitter id list
        try:
            resource = client.stream.statuses.filter.post(follow=rules)
            return resource
        except TwitterApiError as e:
            logger.critical(e)
            sys.exit(3)
    else:
        logger.critical('That traptor type has not been implemented yet')
        sys.exit(3)


def run():
    # Grab a list of twitter ids from the get_redis_twitter_ids function
    twids_str = ','.join(get_redis_twitter_ids())

    # Set up Kafka producer
    producer = create_kafka_producer()

    # Set up a birdy streaming client
    birdyclient = create_birdy_stream(twids_str)

    # Iterate through the twitter results
    for data in birdyclient.stream():
        logging.info(json.dumps(data.get('text')))
        logging.debug(json.dumps(data))
        # Convert date to ISO format
        _time = data.get('created_at')
        if _time:
            data['created_at'] = parser.parse(_time).isoformat()

        try:
            producer.send_messages(KAFKA_TOPIC, json.dumps(data))
        except NotLeaderForPartitionError as e:
            logger.error(e)
            sys.exit(3)


def main():
    run()

if __name__ == '__main__':
    main()
{'fraction_non_alphanumeric': 0.0820414167703972, 'fraction_numerical': 0.006450152766776056, 'mean_word_length': 2.378440366972477, 'pattern_counts': {'":': 0, '<': 9, '<?xml version=': 0, '>': 7, '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': '1025540', 'n_tokens_mistral': 2722, 'n_tokens_neox': 2594, 'n_words': 716}