<commit_before>from celery.task import task
import sys
from couchdbkit.exceptions import ResourceNotFound
from django.conf import settings
from dimagi.utils.couch import release_lock
from dimagi.utils.couch.cache import cache_core
from dimagi.utils.logging import notify_error
from pillow_retry.models import PillowError
from pillowtop.utils import import_pillow_string, get_pillow_by_name
from celery.utils.log import get_task_logger

logger = get_task_logger(__name__)


@task(queue='pillow_retry_queue', ignore_result=True)
def process_pillow_retry(error_doc_id):
    # Redis error logged in get_redis_client
    try:
        client = cache_core.get_redis_client()
    except cache_core.RedisClientError:
        return

    # Prevent more than one task from processing this error, just in case
    # it got enqueued twice.
    lock = client.lock(
        "pillow-retry-processing-%s" % error_doc_id,
        timeout=settings.PILLOW_RETRY_PROCESSING_LOCK_TIMEOUT*60
    )

    if lock.acquire(blocking=False):
        try:
            error_doc = PillowError.objects.get(id=error_doc_id)
        except PillowError.DoesNotExist:
            return

        pillow_class = error_doc.pillow
        try:
            pillow = import_pillow_string(pillow_class)
        except ValueError:
            # all fluff pillows have module path of 'fluff' so can't be imported directly
            _, pillow_class_name = pillow_class.rsplit('.', 1)
            pillow = get_pillow_by_name(pillow_class_name)

        if not pillow:
            notify_error("Could not find pillowtop class '%s' while attempting a retry." % pillow_class)
            try:
                error_doc.total_attempts = PillowError.multi_attempts_cutoff() + 1
            finally:
                release_lock(lock, True)
                return

        change = error_doc.change_dict
        if pillow.include_docs:
            try:
                change['doc'] = pillow.couch_db.open_doc(change['id'])
            except ResourceNotFound:
                change['deleted'] = True


        try:
            from corehq.apps.userreports.pillow import ConfigurableIndicatorPillow
            if isinstance(pillow, ConfigurableIndicatorPillow):
                raise Exception('this is temporarily not supported!')
            else:
                pillow.process_change(change, is_retry_attempt=True)
        except Exception:
            ex_type, ex_value, ex_tb = sys.exc_info()
            error_doc.add_attempt(ex_value, ex_tb)
            error_doc.save()
        else:
            error_doc.delete()
        finally:
            release_lock(lock, True)<commit_msg>fix tests<commit_after>from celery.task import task
import sys
from couchdbkit.exceptions import ResourceNotFound
from django.conf import settings
from dimagi.utils.couch import release_lock
from dimagi.utils.couch.cache import cache_core
from dimagi.utils.logging import notify_error
from pillow_retry.models import PillowError
from pillowtop.utils import import_pillow_string, get_pillow_by_name
from celery.utils.log import get_task_logger

logger = get_task_logger(__name__)


@task(queue='pillow_retry_queue', ignore_result=True)
def process_pillow_retry(error_doc_id):
    # Redis error logged in get_redis_client
    try:
        client = cache_core.get_redis_client()
    except cache_core.RedisClientError:
        return

    # Prevent more than one task from processing this error, just in case
    # it got enqueued twice.
    lock = client.lock(
        "pillow-retry-processing-%s" % error_doc_id,
        timeout=settings.PILLOW_RETRY_PROCESSING_LOCK_TIMEOUT*60
    )

    if lock.acquire(blocking=False):
        try:
            error_doc = PillowError.objects.get(id=error_doc_id)
        except PillowError.DoesNotExist:
            return

        pillow_class = error_doc.pillow
        try:
            pillow = import_pillow_string(pillow_class)
        except ValueError:
            # all fluff pillows have module path of 'fluff' so can't be imported directly
            _, pillow_class_name = pillow_class.rsplit('.', 1)
            pillow = get_pillow_by_name(pillow_class_name)

        if not pillow:
            notify_error("Could not find pillowtop class '%s' while attempting a retry." % pillow_class)
            try:
                error_doc.total_attempts = PillowError.multi_attempts_cutoff() + 1
            finally:
                release_lock(lock, True)
                return

        change = error_doc.change_dict
        if pillow.include_docs:
            try:
                change['doc'] = pillow.couch_db.open_doc(change['id'])
            except ResourceNotFound:
                change['deleted'] = True


        try:
            try:
                from corehq.apps.userreports.pillow import ConfigurableIndicatorPillow
                if isinstance(pillow, ConfigurableIndicatorPillow):
                    raise Exception('this is temporarily not supported!')
            except ImportError:
                pass
            pillow.process_change(change, is_retry_attempt=True)
        except Exception:
            ex_type, ex_value, ex_tb = sys.exc_info()
            error_doc.add_attempt(ex_value, ex_tb)
            error_doc.save()
        else:
            error_doc.delete()
        finally:
            release_lock(lock, True)
{'fraction_non_alphanumeric': 0.06837766953915324, 'fraction_numerical': 0.0014986886474334957, 'mean_word_length': 2.4691358024691357, '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': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '7281347', 'n_tokens_mistral': 1567, 'n_tokens_neox': 1537, 'n_words': 351}