<commit_before>from celery import task
from celery.utils.log import get_task_logger
from celery.exceptions import SoftTimeLimitExceeded
from django.db.models import Max

from go_http.send import HttpApiSender
import control.settings as settings
from subscription.models import Subscription, Message

logger = get_task_logger(__name__)


@task()
def process_message_queue(schedule):
    # Get all active and incomplete subscribers for schedule
    # subscribers = 
    with Subscription.objects.filter(
        schedule=schedule, active=True, completed=False, process_status=0).all() as subscribers:
        if subscribers.exists():
            # Make a reusable session to Vumi
            sender = HttpApiSender(
                account_key=settings.VUMI_GO_ACCOUNT_KEY,
                conversation_key=settings.VUMI_GO_CONVERSATION_KEY,
                conversation_token=settings.VUMI_GO_ACCOUNT_TOKEN
            )
            # Fire off message processor for each
            for subscriber in subscribers:
                subscriber.process_status = 1
                subscriber.save()
                processes_message.delay(subscriber, sender)
        return subscribers.count()

@task()
def processes_message(subscriber, sender):
    try:
        # Get next message
        message = Message.objects.get(
            message_set=subscriber.message_set, lang=subscriber.lang,
            sequence_number=subscriber.next_sequence_number)
        # Send message
        response = sender.send_text(subscriber.to_addr, message.content)
        # Post process moving to next message, next set or finished
        # Get set max
        set_max = Message.objects.all().aggregate(Max('sequence_number'))["sequence_number__max"]
        # Compare user position to max
        if subscriber.next_sequence_number == set_max:
            # Mark current as completed
            subscriber.completed = True
            subscriber.active = False
            subscriber.save()
            # If next set defined create new subscription
            message_set = subscriber.message_set
            if message_set.next_set:
                # clone existing minus PK as recommended in
                # https://docs.djangoproject.com/en/1.6/topics/db/queries/#copying-model-instances
                subscriber.pk = None
                subscriber.process_status = 0
                subscription = subscriber
                subscription.message_set = message_set.next_set
                subscription.save()
        else:
            # More in this set so interate by one
            subscriber.next_sequence_number = subscriber.next_sequence_number + 1
            subscriber.process_status = 0
            subscriber.save()
        return response
    except SoftTimeLimitExceeded:
        logger.error('Soft time limit exceed processing message to Vumi HTTP API via Celery', exc_info=True)

<commit_msg>Better subsend error handling<commit_after>from celery import task
from celery.utils.log import get_task_logger
from celery.exceptions import SoftTimeLimitExceeded
from django.db.models import Max
from django.core.exceptions import ObjectDoesNotExist

from go_http.send import HttpApiSender, LoggingSender
import control.settings as settings
from subscription.models import Subscription, Message

logger = get_task_logger(__name__)


@task()
def process_message_queue(schedule, sender=None):
    # Get all active and incomplete subscribers for schedule
    subscribers = Subscription.objects.filter(
        schedule=schedule, active=True, completed=False, process_status=0).all()

    # Make a reusable session to Vumi
    if sender is None:
        sender = HttpApiSender(
            account_key=settings.VUMI_GO_ACCOUNT_KEY,
            conversation_key=settings.VUMI_GO_CONVERSATION_KEY,
            conversation_token=settings.VUMI_GO_ACCOUNT_TOKEN
        )
        # sender = LoggingSender('go_http.test')
            # Fire off message processor for each
    for subscriber in subscribers:
        subscriber.process_status = 1
        subscriber.save()
        processes_message.delay(subscriber, sender)
    return subscribers.count()

@task()
def processes_message(subscriber, sender):
    try:
        # Get next message
        try:
            message = Message.objects.get(
                message_set=subscriber.message_set, lang=subscriber.lang,
                sequence_number=subscriber.next_sequence_number)
            # Send message
            response = sender.send_text(subscriber.to_addr, message.content)
            # Post process moving to next message, next set or finished
            # Get set max
            set_max = Message.objects.all().aggregate(Max('sequence_number'))["sequence_number__max"]
            # Compare user position to max
            if subscriber.next_sequence_number == set_max:
                # Mark current as completed
                subscriber.completed = True
                subscriber.active = False
                subscriber.save()
                # If next set defined create new subscription
                message_set = subscriber.message_set
                if message_set.next_set:
                    # clone existing minus PK as recommended in
                    # https://docs.djangoproject.com/en/1.6/topics/db/queries/#copying-model-instances
                    subscriber.pk = None
                    subscriber.process_status = 0
                    subscription = subscriber
                    subscription.message_set = message_set.next_set
                    subscription.save()
            else:
                # More in this set so interate by one
                subscriber.next_sequence_number = subscriber.next_sequence_number + 1
                subscriber.process_status = 0
                subscriber.save()
            return response
        except ObjectDoesNotExist:
            subscriber.process_status = -1
            subscriber.save()
            logger.error('Missing subscription message', exc_info=True)
    except SoftTimeLimitExceeded:
        logger.error('Soft time limit exceed processing message to Vumi HTTP API via Celery', exc_info=True)


{'fraction_non_alphanumeric': 0.06303818034118602, 'fraction_numerical': 0.002437043054427295, 'mean_word_length': 2.3365853658536584, 'pattern_counts': {'":': 0, '<': 3, '<?xml version=': 0, '>': 3, 'https://': 2, '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': '1727975', 'n_tokens_mistral': 1598, 'n_tokens_neox': 1508, 'n_words': 402}