<commit_before>import traceback
import sys
import os
import taskcluster

from mozillapulse.config import PulseConfiguration
from mozillapulse.consumers import GenericConsumer


class AuthenticationError(Exception):
    pass


def debug_print(msg):
    print(msg)
    # running in heroku causes stdio to be buffered therefore we flush
    sys.stdout.flush()


class Options(object):

    options = {}

    def __init__(self):
        self._parse_env()
        self._parse_args()

    def _parse_env(self):
        try:
            self.options['user'] = os.environ['PULSE_USER']
            self.options['passwd'] = os.environ['PULSE_PASSWD']
        except KeyError:
            raise AuthenticationError
            # DEBUG statement: please remove before release
            debug_print("Auth error")
            sys.exit(1)

    def _parse_args(self):
        # TODO: parse args and return them as options
        pass


class TcPulseConsumer(GenericConsumer):
    def __init__(self, exchanges, **kwargs):
        super(TcPulseConsumer, self).__init__(
            PulseConfiguration(**kwargs), exchanges, **kwargs)


class TaskEventApp(object):

    # pendingTasks is a dict where all pending tasks and their task definitions
    # are kept. Tasks are added or removed based on msgs from task-pending,
    # task-running, and task-exception {'taskid': task_definition}
    pendingTasks = {}

    # ampq/pulse listener
    listener = None

    # TODO: move these to args and env options
    # TODO: add task-exception exchange
    # State transitions
    # pending --> running
    #         \-> exception
    exchanges = ['exchange/taskcluster-queue/v1/task-pending',
                 'exchange/taskcluster-queue/v1/task-running']

    # TODO: make perm coalescer service pulse creds
    consumer_args = {
        'applabel': '<EMAIL>|pulse-test2',
        'topic': ['#', '#'],
        'durable': True,
        'user': 'public',
        'password': '<PASSWORD>'
    }

    # Setup task cluster queue object
    tc_queue = taskcluster.Queue()

    options = None

    def __init__(self, options):
        self.options = options
        self.consumer_args['user'] = self.options['user']
        self.consumer_args['password'] = self.options['<PASSWORD>']

    def run(self):
        # TODO: bind with better topic to limit queue
        # DEBUG statement: please remove before release
        debug_print(self.consumer_args)
        self.listener = TcPulseConsumer(self.exchanges,
                                        callback=self._route_callback_handler,
                                        **self.consumer_args)
        while True:
            try:
                self.listener.listen()
            except KeyboardInterrupt:
                # TODO: delete_queue doesn't work. fix me
                # self.listener.delete_queue()
                sys.exit(1)
            except:
                traceback.print_exc()

    def _route_callback_handler(self, body, message):
        """
        Route call body and msg to proper callback handler
        """
        taskState = body['status']['state']
        taskId = body['status']['taskId']
        # DEBUG statement: please remove before release
        debug_print("taskId: %s (%s)" % (taskId, taskState))
        # DEBUG statement: please remove before release
        # print json.dumps(body, sort_keys=True, indent=4)
        if taskState == 'pending':
            self._add_task_callback(body, message, taskId)
        elif taskState == 'running' or taskState == 'exception':
            self._remove_task_callback(body, message, taskId)
        else:
            # TODO: Handle unknown states; for now just ack the msg
            pass
        # DEBUG statement: please remove before release
        debug_print("PendingTasks: %s" % (len(self.pendingTasks)))
        message.ack()

    def _add_task_callback(self, body, message, taskId):
        # Insert taskId into pendingTask with None as place holder
        # TODO: handle if task already exists
        taskDef = self._retrieve_taskdef(taskId)
        self.pendingTasks[taskId] = taskDef

    def _remove_task_callback(self, body, message, taskId):
        # TODO: make idempotent; handle KeyError
        try:
            del self.pendingTasks[taskId]
        except:
            pass

    def _retrieve_taskdef(self, taskId):
        # TODO: retry api call
        # DEBUG: api call disabled
        taskDef = "TaskDef goes here"
        # taskDef = self.tc_queue.task(taskId)
        # TODO: validate response
        # DEBUG statement: please remove before release
        return taskDef

    def _spawn_taskdef_worker(self, taskId):
        """
        Spawn an async worker thread to fetch taskdef and fill in pendingTasks
        value. Once worker has taskdef, Coalescer object is applied to possible
        add taskId to a current 'defined commonality list' already indexed or
        to build a new list
        """
        pass
<commit_msg>Remove taskcluster py lib (not compat w/py3.5)<commit_after>import traceback
import sys
import os

from mozillapulse.config import PulseConfiguration
from mozillapulse.consumers import GenericConsumer


class AuthenticationError(Exception):
    pass


def debug_print(msg):
    print(msg)
    # running in heroku causes stdio to be buffered therefore we flush
    sys.stdout.flush()


class Options(object):

    options = {}

    def __init__(self):
        self._parse_env()
        self._parse_args()

    def _parse_env(self):
        try:
            self.options['user'] = os.environ['PULSE_USER']
            self.options['passwd'] = <PASSWORD>['<PASSWORD>']
        except KeyError:
            raise AuthenticationError
            # DEBUG statement: please remove before release
            debug_print("Auth error")
            sys.exit(1)

    def _parse_args(self):
        # TODO: parse args and return them as options
        pass


class TcPulseConsumer(GenericConsumer):
    def __init__(self, exchanges, **kwargs):
        super(TcPulseConsumer, self).__init__(
            PulseConfiguration(**kwargs), exchanges, **kwargs)


class TaskEventApp(object):

    # pendingTasks is a dict where all pending tasks and their task definitions
    # are kept. Tasks are added or removed based on msgs from task-pending,
    # task-running, and task-exception {'taskid': task_definition}
    pendingTasks = {}

    # ampq/pulse listener
    listener = None

    # TODO: move these to args and env options
    # TODO: add task-exception exchange
    # State transitions
    # pending --> running
    #         \-> exception
    exchanges = ['exchange/taskcluster-queue/v1/task-pending',
                 'exchange/taskcluster-queue/v1/task-running']

    # TODO: make perm coalescer service pulse creds
    consumer_args = {
        'applabel': '<EMAIL>|pulse-test2',
        'topic': ['#', '#'],
        'durable': True,
        'user': 'public',
        'password': '<PASSWORD>'
    }


    options = None

    def __init__(self, options):
        self.options = options
        self.consumer_args['user'] = self.options['user']
        self.consumer_args['password'] = self.options['passwd']

    def run(self):
        # TODO: bind with better topic to limit queue
        # DEBUG statement: please remove before release
        debug_print(self.consumer_args)
        self.listener = TcPulseConsumer(self.exchanges,
                                        callback=self._route_callback_handler,
                                        **self.consumer_args)
        while True:
            try:
                self.listener.listen()
            except KeyboardInterrupt:
                # TODO: delete_queue doesn't work. fix me
                # self.listener.delete_queue()
                sys.exit(1)
            except:
                traceback.print_exc()

    def _route_callback_handler(self, body, message):
        """
        Route call body and msg to proper callback handler
        """
        taskState = body['status']['state']
        taskId = body['status']['taskId']
        # DEBUG statement: please remove before release
        debug_print("taskId: %s (%s)" % (taskId, taskState))
        # DEBUG statement: please remove before release
        # print json.dumps(body, sort_keys=True, indent=4)
        if taskState == 'pending':
            self._add_task_callback(body, message, taskId)
        elif taskState == 'running' or taskState == 'exception':
            self._remove_task_callback(body, message, taskId)
        else:
            # TODO: Handle unknown states; for now just ack the msg
            pass
        # DEBUG statement: please remove before release
        debug_print("PendingTasks: %s" % (len(self.pendingTasks)))
        message.ack()

    def _add_task_callback(self, body, message, taskId):
        # Insert taskId into pendingTask with None as place holder
        # TODO: handle if task already exists
        taskDef = self._retrieve_taskdef(taskId)
        self.pendingTasks[taskId] = taskDef

    def _remove_task_callback(self, body, message, taskId):
        # TODO: make idempotent; handle KeyError
        try:
            del self.pendingTasks[taskId]
        except:
            pass

    def _retrieve_taskdef(self, taskId):
        # TODO: retry api call
        # DEBUG: api call disabled
        taskDef = "TaskDef goes here"
        # TODO: validate response
        # DEBUG statement: please remove before release
        return taskDef

    def _spawn_taskdef_worker(self, taskId):
        """
        Spawn an async worker thread to fetch taskdef and fill in pendingTasks
        value. Once worker has taskdef, Coalescer object is applied to possible
        add taskId to a current 'defined commonality list' already indexed or
        to build a new list
        """
        pass

{'fraction_non_alphanumeric': 0.08961593172119488, 'fraction_numerical': 0.001422475106685633, 'mean_word_length': 2.2788141239173885, 'pattern_counts': {'":': 0, '<': 10, '<?xml version=': 0, '>': 14, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 4, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '4591060', 'n_tokens_mistral': 2658, 'n_tokens_neox': 2569, 'n_words': 787}