<commit_before>import logging
import re
from itertools import chain
from urllib.parse import urlparse

import celery
import celery.exceptions
import celery.states
import mwapi.errors
import revscoring.errors

from ores.score_request import ScoreRequest
from .. import errors
from .scoring_system import ScoringSystem
from ..task_tracker import RedisTaskTracker, NullTaskTracker

logger = logging.getLogger(__name__)

_applications = []

DEFAULT_CELERY_QUEUE = "celery"
SENT = "SENT"
REQUESTED = "REQUESTED"


class CeleryQueue(ScoringSystem):

    def __init__(self, *args, application, queue_maxsize=None,
                 task_tracker=None, **kwargs):
        super().__init__(*args, **kwargs)
        global _applications
        self.application = application
        self.queue_maxsize = int(queue_maxsize) if queue_maxsize is not None \
                             else None

        self.redis = redis_from_url(self.application.conf.BROKER_URL)

        self.task_tracker = task_tracker or NullTaskTracker()

        if self.queue_maxsize is not None and self.redis is None:
            logger.warning("No redis connection.  Can't check queue size")

        self._initialize_tasks()

        _applications.append(application)

    def _initialize_tasks(self):
        expected_errors = (revscoring.errors.RevisionNotFound,
                           revscoring.errors.PageNotFound,
                           revscoring.errors.UserNotFound,
                           revscoring.errors.DependencyError,
                           mwapi.errors.RequestError,
                           mwapi.errors.TimeoutError,
                           errors.TimeoutError)

        @self.application.task(throws=expected_errors,
                               queue=DEFAULT_CELERY_QUEUE)
        def _process_score_map(request, model_names, rev_id, root_cache):
            if not isinstance(request, ScoreRequest):
                request = ScoreRequest.from_json(request)

            logger.info("Generating a score map for {0}"
                        .format(request.format(rev_id, model_names)))

            score_map = ScoringSystem._process_score_map(
                self, request, rev_id, model_names,
                root_cache=root_cache)
            logger.info("Completed generating score map for {0}"
                        .format(request.format(rev_id, model_names)))
            return score_map

        self._process_score_map = _process_score_map

    def _process_missing_scores(self, request, missing_model_set_revs,
                                root_caches, inprogress_results=None):
        logger.debug("Processing missing scores {0}:{1}."
                     .format(request.context_name, missing_model_set_revs))
        context = self[request.context_name]

        inprogress_results = inprogress_results or {}

        # Generate score results
        results = {}
        for missing_models, rev_ids in missing_model_set_revs.items():
            for rev_id in rev_ids:
                injection_cache = request.injection_caches.get(rev_id)
                if rev_id not in root_caches:
                    for model_name in missing_models:
                        task_id = context.format_id_string(
                            model_name, rev_id, request,
                            injection_cache=injection_cache)
                        self.application.backend.mark_as_failure(
                            task_id, RuntimeError("Never started"))
                    continue
                root_cache = {str(k): v for k, v in root_caches[rev_id].items()}
                result = self._process_score_map.delay(
                    request, missing_models, rev_id, root_cache)
                self._lock_process(missing_models, rev_id, request,
                                   injection_cache, result.id)

                for model_name in missing_models:
                    if rev_id in results:
                        results[rev_id][model_name] = result
                    else:
                        results[rev_id] = {model_name: result}

        # Read results
        rev_scores = {}<commit_msg>Send ScoreRequest as json object over the celery wire<commit_after>import logging
import re
from itertools import chain
from urllib.parse import urlparse

import celery
import celery.exceptions
import celery.states
import mwapi.errors
import revscoring.errors

from ores.score_request import ScoreRequest
from .. import errors
from .scoring_system import ScoringSystem
from ..task_tracker import RedisTaskTracker, NullTaskTracker

logger = logging.getLogger(__name__)

_applications = []

DEFAULT_CELERY_QUEUE = "celery"
SENT = "SENT"
REQUESTED = "REQUESTED"


class CeleryQueue(ScoringSystem):

    def __init__(self, *args, application, queue_maxsize=None,
                 task_tracker=None, **kwargs):
        super().__init__(*args, **kwargs)
        global _applications
        self.application = application
        self.queue_maxsize = int(queue_maxsize) if queue_maxsize is not None \
                             else None

        self.redis = redis_from_url(self.application.conf.BROKER_URL)

        self.task_tracker = task_tracker or NullTaskTracker()

        if self.queue_maxsize is not None and self.redis is None:
            logger.warning("No redis connection.  Can't check queue size")

        self._initialize_tasks()

        _applications.append(application)

    def _initialize_tasks(self):
        expected_errors = (revscoring.errors.RevisionNotFound,
                           revscoring.errors.PageNotFound,
                           revscoring.errors.UserNotFound,
                           revscoring.errors.DependencyError,
                           mwapi.errors.RequestError,
                           mwapi.errors.TimeoutError,
                           errors.TimeoutError)

        @self.application.task(throws=expected_errors,
                               queue=DEFAULT_CELERY_QUEUE)
        def _process_score_map(request, model_names, rev_id, root_cache):
            if not isinstance(request, ScoreRequest):
                request = ScoreRequest.from_json(request)

            logger.info("Generating a score map for {0}"
                        .format(request.format(rev_id, model_names)))

            score_map = ScoringSystem._process_score_map(
                self, request, rev_id, model_names,
                root_cache=root_cache)
            logger.info("Completed generating score map for {0}"
                        .format(request.format(rev_id, model_names)))
            return score_map

        self._process_score_map = _process_score_map

    def _process_missing_scores(self, request, missing_model_set_revs,
                                root_caches, inprogress_results=None):
        logger.debug("Processing missing scores {0}:{1}."
                     .format(request.context_name, missing_model_set_revs))
        context = self[request.context_name]

        inprogress_results = inprogress_results or {}

        # Generate score results
        results = {}
        for missing_models, rev_ids in missing_model_set_revs.items():
            for rev_id in rev_ids:
                injection_cache = request.injection_caches.get(rev_id)
                if rev_id not in root_caches:
                    for model_name in missing_models:
                        task_id = context.format_id_string(
                            model_name, rev_id, request,
                            injection_cache=injection_cache)
                        self.application.backend.mark_as_failure(
                            task_id, RuntimeError("Never started"))
                    continue
                root_cache = {str(k): v for k, v in root_caches[rev_id].items()}
                result = self._process_score_map.delay(
                    request.to_json(), missing_models, rev_id, root_cache)
                self._lock_process(missing_models, rev_id, request,
                                   injection_cache, result.id)

                for model_name in missing_models:
                    if rev_id in results:
                        results[rev_id][model_name] = result
                    else:
                        results[rev_id] = {model_name: result}

        # Read results
        rev_scores = {}
{'fraction_non_alphanumeric': 0.06934131736526947, 'fraction_numerical': 0.0009580838323353293, 'mean_word_length': 1.9342937456078706, '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': 2, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '6629246', 'n_tokens_mistral': 2215, 'n_tokens_neox': 2119, 'n_words': 474}