<commit_before># -*- coding: utf-8 -*-
from __future__ import unicode_literals

import json
import logging
from collections import Counter
from urlparse import urlparse

import click
import redis

from celery.task.control import inspect
from flask import current_app

from udata.app import cache
from udata.commands import cli, exit_with_error
from udata.tasks import celery, router

log = logging.getLogger(__name__)


@cli.group('worker')
def grp():
    '''Worker related operations'''
    pass


TASKS_LIST_CACHE_KEY = 'worker-status-tasks'
# we're using an aggressive cache in order not to hit Celery every 5 min
TASKS_LIST_CACHE_DURATION = 60 * 60 * 24  # in seconds


@grp.command()
def start():
    '''Start a worker'''
    worker = celery.Worker()
    worker.start()
    return worker.exitcode


def status_print_task(count, biggest_task_name, munin=False):
    if not munin:
        print('* %s : %s' % (count[0].ljust(biggest_task_name), count[1]))
    else:
        print('%s.value %s' % (format_field_for_munin(count[0]), count[1]))


def status_print_config(queue):
    if not queue:
        exit_with_error('--munin-config called without a --queue parameter')
    tasks = cache.get(TASKS_LIST_CACHE_KEY) or []
    if not tasks:
        registered = inspect().registered_tasks()
        if registered:
            for w, tasks_list in registered.iteritems():
                tasks += [t for t in tasks_list if t not in tasks]
            cache.set(TASKS_LIST_CACHE_KEY, tasks,
                      timeout=TASKS_LIST_CACHE_DURATION)
    print('graph_title Waiting tasks for queue %s' % queue)
    print('graph_vlabel Nb of tasks')
    print('graph_category celery')
    for task in tasks:
        print('%s.label %s' % (format_field_for_munin(task), task))


def status_print_queue(queue, munin=False):
    r = get_redis_connection()
    if not munin:
        print('-' * 40)
    queue_length = r.llen(queue)
    if not munin:
        print('Queue "%s": %s task(s)' % (queue, queue_length))
    counter = Counter()
    biggest_task_name = 0
    for task in r.lrange(queue, 0, -1):
        task = json.loads(task)<commit_msg>Ensure munin config only lists tasks for the given queue<commit_after># -*- coding: utf-8 -*-
from __future__ import unicode_literals

import json
import logging
from collections import Counter
from urlparse import urlparse

import click
import redis

from flask import current_app

from udata.commands import cli, exit_with_error
from udata.tasks import celery, router

log = logging.getLogger(__name__)


@cli.group('worker')
def grp():
    '''Worker related operations'''
    pass


@grp.command()
def start():
    '''Start a worker'''
    worker = celery.Worker()
    worker.start()
    return worker.exitcode


def status_print_task(count, biggest_task_name, munin=False):
    if not munin:
        print('* %s : %s' % (count[0].ljust(biggest_task_name), count[1]))
    else:
        print('%s.value %s' % (format_field_for_munin(count[0]), count[1]))


def status_print_config(queue):
    if not queue:
        exit_with_error('--munin-config called without a --queue parameter')
    tasks = [n for n, q in get_tasks().items() if q == queue]
    print('graph_title Waiting tasks for queue %s' % queue)
    print('graph_vlabel Nb of tasks')
    print('graph_category celery')
    for task in tasks:
        print('%s.label %s' % (format_field_for_munin(task), task))


def status_print_queue(queue, munin=False):
    r = get_redis_connection()
    if not munin:
        print('-' * 40)
    queue_length = r.llen(queue)
    if not munin:
        print('Queue "%s": %s task(s)' % (queue, queue_length))
    counter = Counter()
    biggest_task_name = 0
    for task in r.lrange(queue, 0, -1):
        task = json.loads(task)
{'fraction_non_alphanumeric': 0.11580345285524568, 'fraction_numerical': 0.007171314741035857, 'mean_word_length': 3.399532710280374, 'pattern_counts': {'":': 2, '<': 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': '1970790', 'n_tokens_mistral': 1257, 'n_tokens_neox': 1204, 'n_words': 355}