<commit_before># coding=utf-8
from __future__ import absolute_import, print_function, unicode_literals

import socket
import time

import six
from celery import states
from collections import OrderedDict
from kombu import Exchange, Queue, binding

from clearly.task_info import TaskInfo
from clearly.utils.colors import colors

HEADER_SIZE = 8
SEPARATOR = colors.RED(', ')
KWARGS_SEPARATOR = colors.RED('=')
DICT_SEPARATOR = colors.RED(': ')
EMPTY = colors.DIM(':)')


class Clearly(object):
    """
        Simple and real-time monitor for celery.
    """

    def __init__(self, app, exchange):
        """Constructs a monitor instance.
        
        Args:
            app (celery.app): the configured celery app instance
            exchange (Union[str, Exchange]): the exchange instance or name 
                the celery publishes messages to

        """
        self._app = app
        if isinstance(exchange, Exchange):
            self._exchange = exchange
        else:
            self._exchange = Exchange(exchange, type='topic')

        # initialize variables
        self.reset()

    def capture(self, routing_keys,
                show_params=False, show_success=False, show_error=True):
        """Gathers all tasks being sent to celery that matches routing keys.
        You can press CTRL+C at any time to stop capturing, without losing
         anything already captured.
        
        Args:
            routing_keys (str): a string to be split into routing keys.
                use * as exactly one part or # as zero or more.
                ex.: 'dispatch.# email.#' to filter messages starting with
                      or 'dispatch.#.123456.#' to filter that exact id and pos
                      or even '#.123456.#' to filter that exact id anywhere.
            show_params (bool): if True shows params of all tasks
            show_success (bool): if True shows successful tasks' results
            show_error (bool): if True shows failed tasks' results
    
        """
        monitor_queue = Queue(exclusive=True,
                              durable=False,
                              bindings=[
                                  binding(exchange=self._exchange,
                                          routing_key=x)
                                  for x in routing_keys.split()])

        def process_message(body, message):
            task_id = body['id']

            # RETRY tasks pass again here.
            if task_id in self._waiting_tasks:
                async_result = self._waiting_tasks[task_id].async
            else:
                async_result = self._app.AsyncResult(task_id)

            # There's a race condition in task async results, which can be in
            # the process of update in the backend yet. Sometimes I would
            # receive a PENDING with a 1 retry count or a RETRY with a 0 retry
            # count... But the current retry count seems to be reliable.
            current_retry = body['retries']
            # Wow, even a SUCCESS has appeared here.
            current_state = states.RETRY if current_retry else None

            task = TaskInfo(id=task_id, name=body['task'],
                            args=body['args'], kwargs=body['kwargs'],
                            async=async_result,
                            state=current_state,
                            retries=current_retry)

            if current_retry == 0:
                self._display_task(task, True, show_params, False)<commit_msg>fix(core) initial state is pending, to avoid printing again in fetch as an update<commit_after># coding=utf-8
from __future__ import absolute_import, print_function, unicode_literals

import socket
import time

import six
from celery import states
from collections import OrderedDict
from kombu import Exchange, Queue, binding

from clearly.task_info import TaskInfo
from clearly.utils.colors import colors

HEADER_SIZE = 8
SEPARATOR = colors.RED(', ')
KWARGS_SEPARATOR = colors.RED('=')
DICT_SEPARATOR = colors.RED(': ')
EMPTY = colors.DIM(':)')


class Clearly(object):
    """
        Simple and real-time monitor for celery.
    """

    def __init__(self, app, exchange):
        """Constructs a monitor instance.
        
        Args:
            app (celery.app): the configured celery app instance
            exchange (Union[str, Exchange]): the exchange instance or name 
                the celery publishes messages to

        """
        self._app = app
        if isinstance(exchange, Exchange):
            self._exchange = exchange
        else:
            self._exchange = Exchange(exchange, type='topic')

        # initialize variables
        self.reset()

    def capture(self, routing_keys,
                show_params=False, show_success=False, show_error=True):
        """Gathers all tasks being sent to celery that matches routing keys.
        You can press CTRL+C at any time to stop capturing, without losing
         anything already captured.
        
        Args:
            routing_keys (str): a string to be split into routing keys.
                use * as exactly one part or # as zero or more.
                ex.: 'dispatch.# email.#' to filter messages starting with
                      or 'dispatch.#.123456.#' to filter that exact id and pos
                      or even '#.123456.#' to filter that exact id anywhere.
            show_params (bool): if True shows params of all tasks
            show_success (bool): if True shows successful tasks' results
            show_error (bool): if True shows failed tasks' results
    
        """
        monitor_queue = Queue(exclusive=True,
                              durable=False,
                              bindings=[
                                  binding(exchange=self._exchange,
                                          routing_key=x)
                                  for x in routing_keys.split()])

        def process_message(body, message):
            task_id = body['id']

            # RETRY tasks pass again here.
            if task_id in self._waiting_tasks:
                async_result = self._waiting_tasks[task_id].async
            else:
                async_result = self._app.AsyncResult(task_id)

            # There's a race condition in task async results, which can be in
            # the process of update in the backend yet. Sometimes I would
            # receive a PENDING with a 1 retry count or a RETRY with a 0 retry
            # count... But the current retry count seems to be reliable.
            current_retry = body['retries']
            # Wow, even a SUCCESS has appeared here.
            current_state = states.RETRY if current_retry else states.PENDING

            task = TaskInfo(id=task_id, name=body['task'],
                            args=body['args'], kwargs=body['kwargs'],
                            async=async_result,
                            state=current_state,
                            retries=current_retry)

            if current_retry == 0:
                self._display_task(task, True, show_params, False)
{'fraction_non_alphanumeric': 0.07384441939120631, 'fraction_numerical': 0.004791431792559188, 'mean_word_length': 1.8062475286674575, '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': 10, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '6150780', 'n_tokens_mistral': 1817, 'n_tokens_neox': 1710, 'n_words': 628}