<commit_before>from __future__ import print_function, unicode_literals

import json
import logging
import pytz
import regex
import six
import time
import traceback

from datetime import datetime, timedelta
from django.conf import settings
from django.contrib.auth.models import User
from django.core.cache import cache
from django.core.files.temp import NamedTemporaryFile
from django.db import models, transaction
from django.db.models import Q, Count, Prefetch, Sum
from django.utils import timezone
from django.utils.html import escape
from django.utils.translation import ugettext, ugettext_lazy as _
from temba_expressions.evaluator import EvaluationContext, DateStyle
from temba.assets.models import register_asset_store
from temba.contacts.models import Contact, ContactGroup, ContactURN, URN, TEL_SCHEME
from temba.channels.models import Channel, ChannelEvent
from temba.orgs.models import Org, TopUp, Language, UNREAD_INBOX_MSGS
from temba.schedules.models import Schedule
from temba.utils import get_datetime_format, datetime_to_str, analytics, chunk_list, on_transaction_commit
from temba.utils.cache import get_cacheable_attr
from temba.utils.export import BaseExportTask, BaseExportAssetStore
from temba.utils.expressions import evaluate_template
from temba.utils.models import SquashableModel, TembaModel
from temba.utils.queues import DEFAULT_PRIORITY, push_task, LOW_PRIORITY, HIGH_PRIORITY
from .handler import MessageHandler

logger = logging.getLogger(__name__)
__message_handlers = None

MSG_QUEUE = 'msgs'
SEND_MSG_TASK = 'send_msg_task'

HANDLER_QUEUE = 'handler'
HANDLE_EVENT_TASK = 'handle_event_task'
MSG_EVENT = 'msg'
FIRE_EVENT = 'fire'
TIMEOUT_EVENT = 'timeout'

BATCH_SIZE = 500

INITIALIZING = 'I'
PENDING = 'P'
QUEUED = 'Q'
WIRED = 'W'
SENT = 'S'
DELIVERED = 'D'
HANDLED = 'H'
ERRORED = 'E'
FAILED = 'F'
RESENT = 'R'

INCOMING = 'I'
OUTGOING = 'O'

INBOX = 'I'
FLOW = 'F'
IVR = 'V'
USSD = 'U'

MSG_SENT_KEY = 'msgs_sent_%y_%m_%d'

# status codes used for both messages and broadcasts (single char constant, human readable, API readable)
STATUS_CONFIG = (
    # special state for flows used to hold off sending the message until the flow is ready to receive a response
    (INITIALIZING, _("Initializing"), 'initializing'),

    (PENDING, _("Pending"), 'pending'),        # initial state for all messages

    # valid only for outgoing messages
    (QUEUED, _("Queued"), 'queued'),
    (WIRED, _("Wired"), 'wired'),              # message was handed off to the provider and credits were deducted for it
    (SENT, _("Sent"), 'sent'),                 # we have confirmation that a message was sent
    (DELIVERED, _("Delivered"), 'delivered'),

    # valid only for incoming messages
    (HANDLED, _("Handled"), 'handled'),

    (ERRORED, _("Error Sending"), 'errored'),  # there was an error during delivery
    (FAILED, _("Failed Sending"), 'failed'),   # we gave up on sending this message
    (RESENT, _("Resent message"), 'resent'),   # we retried this message
)


def get_message_handlers():
    """
    Initializes all our message handlers
    """
    global __message_handlers
    if not __message_handlers:
        handlers = []
        for handler_class in settings.MESSAGE_HANDLERS:
            try:
                cls = MessageHandler.find(handler_class)
                handlers.append(cls())
            except Exception as ee:  # pragma: no cover
                traceback.print_exc(ee)

        __message_handlers = handlers

    return __message_handlers


def get_unique_recipients(urns, contacts, groups):
    """
    Builds a list of the unique contacts and URNs by merging urns, contacts and groups
    """
    unique_urns = set()
    unique_contacts = set()
    included_by_urn = set()  # contact ids of contacts included by URN

    for urn in urns:
        unique_urns.add(urn)
        included_by_urn.add(urn.contact_id)

    for group in groups:
        for contact in group.contacts.all():
            if contact.id not in included_by_urn:
                unique_contacts.add(contact)

    for contact in contacts:
        if contact.id not in included_by_urn:
            unique_contacts.add(contact)

    return unique_urns, unique_contacts


class UnreachableException(Exception):
    """
    Exception thrown when a message is being sent to a contact that we don't have a sendable URN for
    """
    pass


class BroadcastRecipient(models.Model):
    """
    Through table for broadcast recipients many-to-many
    """
    broadcast = models.ForeignKey('msgs.Broadcast')

    contact = models.ForeignKey(Contact)

    purged_status = models.CharField(null=True, max_length=1,
                                     help_text=_("Used when broadcast is purged to record contact's message's state"))

    class Meta:
        db_table = 'msgs_broadcast_recipients'


@six.python_2_unicode_compatible
class Broadcast(models.Model):
    """
    A broadcast is a message that is sent out to more than one recipient, such
    as a ContactGroup or a list of Contacts. It's nothing more than a way to tie
    messages sent from the same bundle together
    """
    STATUS_CHOICES = [(s[0], s[1]) for s in STATUS_CONFIG]

    BULK_THRESHOLD = 50  # use bulk priority for messages if number of recipients greater than this

    org = models.ForeignKey(Org, verbose_name=_("Org"),
                            help_text=_("The org this broadcast is connected to"))

    groups = models.ManyToManyField(ContactGroup, verbose_name=_("Groups"), related_name='addressed_broadcasts',
                                    help_text=_("The groups to send the message to"))

    contacts = models.ManyToManyField(Contact, verbose_name=_("Contacts"), related_name='addressed_broadcasts',
                                      help_text=_("Individual contacts included in this message"))

    urns = models.ManyToManyField(ContactURN, verbose_name=_("URNs"), related_name='addressed_broadcasts',
                                  help_text=_("Individual URNs included in this message"))

    recipients = models.ManyToManyField(Contact, through=BroadcastRecipient, verbose_name=_("Recipients"),
                                        related_name='broadcasts',
                                        help_text=_("The contacts which received this message"))

    recipient_count = models.IntegerField(verbose_name=_("Number of recipients"), null=True,
                                          help_text=_("Number of urns which received this broadcast"))

    text = models.TextField(max_length=settings.MSG_FIELD_SIZE, verbose_name=_("Text"),
                            help_text=_("The message to send out"))

    channel = models.ForeignKey(Channel, null=True, verbose_name=_("Channel"),
                                help_text=_("Channel to use for message sending"))

    status = models.CharField(max_length=1, verbose_name=_("Status"), choices=STATUS_CHOICES, default=INITIALIZING,
                              help_text=_("The current status for this broadcast"))

    schedule = models.OneToOneField(Schedule, verbose_name=_("Schedule"), null=True,
                                    help_text=_("Our recurring schedule if we have one"), related_name="broadcast")

    parent = models.ForeignKey('Broadcast', verbose_name=_("Parent"), null=True, related_name='children')

    language_dict = models.TextField(verbose_name=_("Translations"),
                                     help_text=_("The localized versions of the broadcast"), null=True)

    base_language = models.CharField(max_length=4, null=True, blank=True,
                                     help_text=_('The language used to send this to contacts without a language'))

    is_active = models.BooleanField(default=True, help_text="Whether this broadcast is active")

    created_by = models.ForeignKey(User, related_name="%(app_label)s_%(class)s_creations",
                                   help_text="The user which originally created this item")

    created_on = models.DateTimeField(default=timezone.now, blank=True, editable=False, db_index=True,
                                      help_text=_("When this broadcast was created"))

    modified_by = models.ForeignKey(User, related_name="%(app_label)s_%(class)s_modifications",
                                    help_text="The user which last modified this item")

    modified_on = models.DateTimeField(auto_now=True,
                                       help_text="When this item was last modified")

    purged = models.BooleanField(default=False,
                                 help_text="If the messages for this broadcast have been purged")

    media_dict = models.TextField(verbose_name=_("Media"),
                                  help_text=_("The localized versions of the media"), null=True)

    @classmethod
    def create(cls, org, user, text, recipients, channel=None, media_dict=None, **kwargs):
        create_args = dict(org=org, text=text, channel=channel, media_dict=media_dict, created_by=user,
                           modified_by=user)
        create_args.update(kwargs)
        broadcast = Broadcast.objects.create(**create_args)
        broadcast.update_recipients(recipients)
        return broadcast

    def update_contacts(self, contact_ids):
        """
        Optimization for broadcasts that only contain contacts. Updates our contacts according to the passed in
        queryset or array.
        """
        self.urns.clear()
        self.groups.clear()
        self.contacts.clear()

        # get our through model
        RelatedModel = self.contacts.through

        # clear called automatically by django
        for chunk in chunk_list(contact_ids, 1000):
            bulk_contacts = [RelatedModel(contact_id=id, broadcast_id=self.id) for id in chunk]
            RelatedModel.objects.bulk_create(bulk_contacts)

        self.recipient_count = len(contact_ids)
        self.save(update_fields=('recipient_count',))

        # cache on object for use in subsequent send(..) calls
        delattr(self, '_recipient_cache')

    def update_recipients(self, recipients):
        """
        Updates the recipients which may be contact groups, contacts or contact URNs. Normally you can't update a
        broadcast after it has been created - the exception is scheduled broadcasts which are never really sent (clones
        of them are sent).
        """
        urns = []
        contacts = []
        groups = []

        for recipient in recipients:
            if isinstance(recipient, ContactURN):
                urns.append(recipient)
            elif isinstance(recipient, Contact):
                contacts.append(recipient)
            elif isinstance(recipient, ContactGroup):
                groups.append(recipient)
            else:  # pragma: needs cover
                raise ValueError("Recipient item is not a Contact, ContactURN or ContactGroup")

        self.urns.clear()
        self.urns.add(*urns)
        self.contacts.clear()
        self.contacts.add(*contacts)
        self.groups.clear()
        self.groups.add(*groups)

        urns, contacts = get_unique_recipients(urns, contacts, groups)

        # update the recipient count - the number of messages we intend to send
        self.recipient_count = len(urns) + len(contacts)
        self.save(update_fields=('recipient_count',))

        # cache on object for use in subsequent send(..) calls
        setattr(self, '_recipient_cache', (urns, contacts))

        return urns, contacts

    def has_pending_fire(self):  # pragma: needs cover
        return self.schedule and self.schedule.has_pending_fire()

    def fire(self):
        recipients = list(self.urns.all()) + list(self.contacts.all()) + list(self.groups.all())
        broadcast = Broadcast.create(self.org, self.created_by, self.text, recipients,
                                     parent=self, modified_by=self.modified_by)

        broadcast.send(trigger_send=True)

        return broadcast

    @classmethod
    def get_broadcasts(cls, org, scheduled=False):
        qs = Broadcast.objects.filter(org=org).exclude(contacts__is_test=True)
        return qs.exclude(schedule=None) if scheduled else qs.filter(schedule=None)

    def get_messages(self):
        return self.msgs.exclude(status=RESENT)

    def get_messages_substitution_complete(self):  # pragma: needs cover
        return self.get_messages().filter(has_template_error=False)

    def get_messages_substitution_incomplete(self):  # pragma: needs cover
        return self.get_messages().filter(has_template_error=True)

    def get_message_count(self):
        return self.get_messages().count()

    def get_message_sending_count(self):  # pragma: needs cover
        return self.get_messages().filter(status__in=[PENDING, QUEUED]).count()

    def get_message_sent_count(self):  # pragma: needs cover
        return self.get_messages().filter(status__in=[SENT, DELIVERED, WIRED]).count()

    def get_message_delivered_count(self):  # pragma: needs cover
        return self.get_messages().filter(status=DELIVERED).count()

    def get_message_failed_count(self):  # pragma: needs cover
        return self.get_messages().filter(status__in=[FAILED, RESENT]).count()

    def get_sync_commands(self, channel):
        """
        Returns the minimal # of broadcast commands for the given Android channel to uniquely represent all the
        messages which are being sent to tel URNs. This will return an array of dicts that look like:
             dict(cmd="mt_bcast", to=[dict(phone=msg.contact.tel, id=msg.pk) for msg in msgs], msg=broadcast.text))
        """
        commands = []
        current_msg = None
        contact_id_pairs = []

        pending = self.get_messages().filter(status__in=[PENDING, QUEUED, WIRED], channel=channel,
                                             contact_urn__scheme=TEL_SCHEME).select_related('contact_urn').order_by('text', 'pk')

        for msg in pending:
            if msg.text != current_msg and contact_id_pairs:
                commands.append(dict(cmd='mt_bcast', to=contact_id_pairs, msg=current_msg))
                contact_id_pairs = []

            current_msg = msg.text
            contact_id_pairs.append(dict(phone=msg.contact_urn.path, id=msg.pk))

        if contact_id_pairs:
            commands.append(dict(cmd='mt_bcast', to=contact_id_pairs, msg=current_msg))

        return commands

    def get_preferred_languages(self, contact, base_language=None, org=None):
        """
        Gets the ordered list of language preferences for the given contact
        """
        org = org or self.org  # org object can be provided to allow caching of org languages
        preferred_languages = []

        if org.primary_language:
            preferred_languages.append(org.primary_language.iso_code)

        if base_language:
            preferred_languages.append(base_language)

        # if contact has a language and it's a valid org language, it has priority
        if contact.language and contact.language in org.get_language_codes():
            preferred_languages = [contact.language] + preferred_languages

        return preferred_languages

    def get_translations(self):
        if not self.language_dict:
            return []
        return get_cacheable_attr(self, '_translations', lambda: json.loads(self.language_dict))

    def get_media_translations(self):
        if not self.media_dict:
            return dict()
        return get_cacheable_attr(self, '_media_translations', lambda: json.loads(self.media_dict))

    def get_translated_text(self, contact, org=None):
        """
        Gets the appropriate translation for the given contact. base_language may be provided
        """
        translations = self.get_translations()
        preferred_languages = self.get_preferred_languages(contact, self.base_language, org)
        return Language.get_localized_text(translations, preferred_languages, self.text)

    def get_translated_media(self, contact, org=None):
        """
        Gets the appropriate media for the given contact. base_language may be provided
        """
        media_translations = self.get_media_translations()
        preferred_languages = self.get_preferred_languages(contact, self.base_language, org)
        return Language.get_localized_text(media_translations, preferred_languages, None)

    def send(self, trigger_send=True, message_context=None, response_to=None, status=PENDING, msg_type=INBOX,
             created_on=None, partial_recipients=None, run_map=None):
        """
        Sends this broadcast by creating outgoing messages for each recipient.
        """
        # ignore mock messages
        if response_to and not response_to.id:
            response_to = None

        # cannot ask for sending by us AND specify a created on, blow up in that case
        if trigger_send and created_on:  # pragma: no cover
            raise ValueError("Cannot trigger send and specify a created_on, breaks creating batches")

        if partial_recipients:
            # if flow is being started, it'll provide a batch of unique contacts itself
            urns, contacts = partial_recipients
        elif hasattr(self, '_recipient_cache'):
            # look to see if previous call to update_recipients left a cached value
            urns, contacts = self._recipient_cache
        else:
            # otherwise fetch everything and calculate
            urns, contacts = get_unique_recipients(self.urns.all(), self.contacts.all(), self.groups.all())

        Contact.bulk_cache_initialize(self.org, contacts)
        recipients = list(urns) + list(contacts)

        RelatedRecipient = Broadcast.recipients.through

        # we batch up our SQL calls to speed up the creation of our SMS objects
        batch = []
        recipient_batch = []

        # our priority is based on the number of recipients
        priority = Msg.PRIORITY_NORMAL
        if len(recipients) == 1:
            priority = Msg.PRIORITY_HIGH
        elif len(recipients) >= self.BULK_THRESHOLD:
            priority = Msg.PRIORITY_BULK

        # if they didn't pass in a created on, create one ourselves
        if not created_on:
            created_on = timezone.now()

        # pre-fetch channels to reduce database hits
        org = Org.objects.filter(pk=self.org.id).prefetch_related('channels').first()

        for recipient in recipients:
            contact = recipient if isinstance(recipient, Contact) else recipient.contact

            # get the appropriate translation for this contact
            text = self.get_translated_text(contact)

            media = self.get_translated_media(contact)
            if media:
                media_type, media_url = media.split(':')

                # if we have a localized media, create the url
                if media_url:
                    media = "%s:https://%s/%s" % (media_type, settings.AWS_BUCKET_DOMAIN, media_url)

            # add in our parent context if the message references @parent
            if run_map:
                run = run_map.get(recipient.pk, None)
                if run and run.flow:
                    # a bit kludgy here, but should avoid most unnecessary context creations.
                    # since this path is an optimization for flow starts, we don't need to
                    # worry about the @child context.
                    if 'parent' in text:
                        if run.parent:
                            from temba.flows.models import Flow
                            message_context = message_context.copy()
                            message_context.update(dict(parent=Flow.build_flow_context(run.parent.flow, run.parent.contact)))

            try:
                msg = Msg.create_outgoing(org,
                                          self.created_by,
                                          recipient,
                                          text,
                                          broadcast=self,
                                          channel=self.channel,
                                          response_to=response_to,
                                          message_context=message_context,
                                          status=status,
                                          msg_type=msg_type,
                                          insert_object=False,
                                          media=media,
                                          priority=priority,
                                          created_on=created_on)

            except UnreachableException:
                # there was no way to reach this contact, do not create a message
                msg = None

            # only add it to our batch if it was legit
            if msg:
                batch.append(msg)
                # keep track of this URN as a recipient
                recipient_batch.append(RelatedRecipient(contact_id=msg.contact_id, broadcast_id=self.id))

            # we commit our messages in batches
            if len(batch) >= BATCH_SIZE:
                Msg.objects.bulk_create(batch)
                RelatedRecipient.objects.bulk_create(recipient_batch)

                # send any messages
                if trigger_send:
                    self.org.trigger_send(Msg.objects.filter(broadcast=self, created_on=created_on).select_related('contact', 'contact_urn', 'channel'))

                    # increment our created on so we can load our next batch
                    created_on = created_on + timedelta(seconds=1)

                batch = []
                recipient_batch = []

        # commit any remaining objects
        if batch:
            Msg.objects.bulk_create(batch)
            RelatedRecipient.objects.bulk_create(recipient_batch)

            if trigger_send:
                self.org.trigger_send(Msg.objects.filter(broadcast=self, created_on=created_on).select_related('contact', 'contact_urn', 'channel'))

        # for large batches, status is handled externally
        # we do this as with the high concurrency of sending we can run into postgresl deadlocks
        # (this could be our fault, or could be: http://www.postgresql.org/message-id/20140731233051.GN17765@andrew-ThinkPad-X230)
        if not partial_recipients:
            self.status = QUEUED if len(recipients) > 0 else SENT
            self.save(update_fields=('status',))

    def update(self):
        """
        Check the status of our messages and update ours accordingly
        """
        # build a map from status to the count for that status
        statuses = self.get_messages().values('status').order_by('status').annotate(count=Count('status'))
        total = 0
        status_map = dict()
        for status in statuses:
            status_map[status['status']] = status['count']
            total += status['count']

        # if errored msgs are greater than the half of all msgs
        if status_map.get(ERRORED, 0) > total / 2:
            self.status = ERRORED

        # if there are more than half failed, show failed
        elif status_map.get(FAILED, 0) > total / 2:
            self.status = FAILED

        # if there are any in Q, we are Q
        elif status_map.get(QUEUED, 0) or status_map.get(PENDING, 0):
            self.status = QUEUED

        # at this point we are either sent or delivered

        # if there are any messages that are only in a sent state
        elif status_map.get(SENT, 0) or status_map.get(WIRED, 0):
            self.status = SENT

        # otherwise, all messages delivered
        elif status_map.get(DELIVERED, 0) == total:
            self.status = DELIVERED

        self.save(update_fields=['status'])

    def __str__(self):
        return "%s (%s)" % (self.org.name, self.pk)


@six.python_2_unicode_compatible
class Msg(models.Model):
    """
    Messages are the main building blocks of a RapidPro application. Channels send and receive
    these, Triggers and Flows handle them when appropriate.

    Messages are either inbound or outbound and can have varying states depending on their
    direction. Generally an outbound message will go through the following states:

      INITIALIZING > QUEUED > WIRED > SENT > DELIVERED

    If things go wrong, they can be put into an ERRORED state where they can be retried. Once
    we've given up then they can be put in the FAILED state.

    Inbound messages are much simpler. They start as PENDING and the can be picked up by Triggers
    or Flows where they would get set to the HANDLED state once they've been dealt with.
    """
    STATUS_CHOICES = [(s[0], s[1]) for s in STATUS_CONFIG]

    VISIBILITY_VISIBLE = 'V'
    VISIBILITY_ARCHIVED = 'A'
    VISIBILITY_DELETED = 'D'

    # single char flag, human readable name, API readable name
    VISIBILITY_CONFIG = ((VISIBILITY_VISIBLE, _("Visible"), 'visible'),
                         (VISIBILITY_ARCHIVED, _("Archived"), 'archived'),
                         (VISIBILITY_DELETED, _("Deleted"), 'deleted'))

    VISIBILITY_CHOICES = [(s[0], s[1]) for s in VISIBILITY_CONFIG]

    DIRECTION_CHOICES = ((INCOMING, _("Incoming")),
                         (OUTGOING, _("Outgoing")))

    MSG_TYPES = ((INBOX, _("Inbox Message")),
                 (FLOW, _("Flow Message")),
                 (IVR, _("IVR Message")),
                 (USSD, _("USSD Message")))

    MEDIA_GPS = 'geo'
    MEDIA_IMAGE = 'image'
    MEDIA_VIDEO = 'video'
    MEDIA_AUDIO = 'audio'

    MEDIA_TYPES = [MEDIA_AUDIO, MEDIA_GPS, MEDIA_IMAGE, MEDIA_VIDEO]

    PRIORITY_HIGH = 1000
    PRIORITY_NORMAL = 500
    PRIORITY_BULK = 100

    MAX_SIZE = settings.MSG_FIELD_SIZE

    org = models.ForeignKey(Org, related_name='msgs', verbose_name=_("Org"),
                            help_text=_("The org this message is connected to"))

    channel = models.ForeignKey(Channel, null=True,
                                related_name='msgs', verbose_name=_("Channel"),
                                help_text=_("The channel object that this message is associated with"))

    contact = models.ForeignKey(Contact,
                                related_name='msgs', verbose_name=_("Contact"),
                                help_text=_("The contact this message is communicating with"))

    contact_urn = models.ForeignKey(ContactURN, null=True,
                                    related_name='msgs', verbose_name=_("Contact URN"),
                                    help_text=_("The URN this message is communicating with"))

    broadcast = models.ForeignKey(Broadcast, null=True, blank=True,
                                  related_name='msgs', verbose_name=_("Broadcast"),
                                  help_text=_("If this message was sent to more than one recipient"))

    text = models.TextField(max_length=MAX_SIZE, verbose_name=_("Text"),
                            help_text=_("The actual message content that was sent"))

    priority = models.IntegerField(default=PRIORITY_NORMAL,
                                   help_text=_("The priority for this message to be sent, higher is higher priority"))

    created_on = models.DateTimeField(verbose_name=_("Created On"), db_index=True,
                                      help_text=_("When this message was created"))

    modified_on = models.DateTimeField(null=True, blank=True, verbose_name=_("Modified On"), auto_now=True,
                                       help_text=_("When this message was last modified"))

    sent_on = models.DateTimeField(null=True, blank=True, verbose_name=_("Sent On"),
                                   help_text=_("When this message was sent to the endpoint"))

    queued_on = models.DateTimeField(null=True, blank=True, verbose_name=_("Queued On"),
                                     help_text=_("When this message was queued to be sent or handled."))

    direction = models.CharField(max_length=1, choices=DIRECTION_CHOICES, verbose_name=_("Direction"),
                                 help_text=_("The direction for this message, either incoming or outgoing"))

    status = models.CharField(max_length=1, choices=STATUS_CHOICES, default='P', verbose_name=_("Status"), db_index=True,
                              help_text=_("The current status for this message"))

    response_to = models.ForeignKey('Msg', null=True, blank=True, related_name='responses',
                                    verbose_name=_("Response To"), db_index=False,
                                    help_text=_("The message that this message is in reply to"))

    labels = models.ManyToManyField('Label', related_name='msgs', verbose_name=_("Labels"),
                                    help_text=_("Any labels on this message"))

    visibility = models.CharField(max_length=1, choices=VISIBILITY_CHOICES, default=VISIBILITY_VISIBLE, db_index=True,
                                  verbose_name=_("Visibility"),
                                  help_text=_("The current visibility of this message, either visible, archived or deleted"))

    has_template_error = models.BooleanField(default=False, verbose_name=_("Has Template Error"),
                                             help_text=_("Whether data for variable substitution are missing"))

    msg_type = models.CharField(max_length=1, choices=MSG_TYPES, null=True, verbose_name=_("Message Type"),
                                help_text=_('The type of this message'))

    msg_count = models.IntegerField(default=1, verbose_name=_("Message Count"),
                                    help_text=_("The number of messages that were used to send this message, calculated on Twilio channels"))

    error_count = models.IntegerField(default=0, verbose_name=_("Error Count"),
                                      help_text=_("The number of times this message has errored"))

    next_attempt = models.DateTimeField(auto_now_add=True, verbose_name=_("Next Attempt"),
                                        help_text=_("When we should next attempt to deliver this message"))

    external_id = models.CharField(max_length=255, null=True, blank=True, verbose_name=_("External ID"),
                                   help_text=_("External id used for integrating with callbacks from other APIs"))

    topup = models.ForeignKey(TopUp, null=True, blank=True, related_name='msgs', on_delete=models.SET_NULL,
                              help_text="The topup that this message was deducted from")

    media = models.URLField(null=True, blank=True, max_length=255,
                            help_text=_("The media associated with this message if any"))

    session = models.ForeignKey('channels.ChannelSession', null=True,
                                help_text=_("The session this message was a part of if any"))

    @classmethod
    def send_messages(cls, all_msgs):
        """
        Adds the passed in messages to our sending queue, this will also update the status of the message to
        queued.
        :return:
        """
        task_msgs = []
        task_priority = None
        last_contact = None

        # we send in chunks of 1,000 to help with contention
        for msg_chunk in chunk_list(all_msgs, 1000):
            # create a temporary list of our chunk so we can iterate more than once
            msgs = [msg for msg in msg_chunk]

            # build our id list
            msg_ids = set([m.id for m in msgs])

            with transaction.atomic():
                queued_on = timezone.now()

                # update them to queued
                send_messages = Msg.objects.filter(id__in=msg_ids)\
                                           .exclude(channel__channel_type=Channel.TYPE_ANDROID)\
                                           .exclude(msg_type__in=(IVR, USSD))\
                                           .exclude(topup=None)\
                                           .exclude(contact__is_test=True)
                send_messages.update(status=QUEUED, queued_on=queued_on, modified_on=queued_on)

                # now push each onto our queue
                for msg in msgs:
                    if (msg.msg_type != IVR and msg.msg_type != USSD and msg.channel and msg.channel.channel_type != Channel.TYPE_ANDROID) and \
                            msg.topup and not msg.contact.is_test:

                        # if this is a different contact than our last, and we have msgs for that last contact, queue the task
                        if task_msgs and last_contact != msg.contact_id:
                            # if no priority was set, default to DEFAULT
                            if task_priority is None:  # pragma: needs cover
                                task_priority = DEFAULT_PRIORITY

                            push_task(task_msgs[0]['org'], MSG_QUEUE, SEND_MSG_TASK, task_msgs, priority=task_priority)
                            task_msgs = []
                            task_priority = None

                        # serialize the model to a dictionary
                        msg.queued_on = queued_on
                        task = msg.as_task_json()

                        # only be low priority if no priority has been set for this task group
                        if msg.priority == Msg.PRIORITY_BULK and task_priority is None:
                            task_priority = LOW_PRIORITY
                        elif msg.priority == Msg.PRIORITY_HIGH:
                            task_priority = HIGH_PRIORITY

                        task_msgs.append(task)
                        last_contact = msg.contact_id

        # send our last msgs
        if task_msgs:
            if task_priority is None:
                task_priority = DEFAULT_PRIORITY
            push_task(task_msgs[0]['org'], MSG_QUEUE, SEND_MSG_TASK, task_msgs, priority=task_priority)

    @classmethod
    def process_message(cls, msg):
        """
        Processes a message, running it through all our handlers
        """
        handlers = get_message_handlers()

        if msg.contact.is_blocked:
            msg.visibility = Msg.VISIBILITY_ARCHIVED
            msg.save(update_fields=['visibility', 'modified_on'])
        else:
            for handler in handlers:
                try:
                    start = None
                    if settings.DEBUG:  # pragma: no cover
                        start = time.time()

                    handled = handler.handle(msg)

                    if start:  # pragma: no cover
                        print("[%0.2f] %s for %d" % (time.time() - start, handler.name, msg.pk or 0))

                    if handled:
                        break
                except Exception as e:  # pragma: no cover
                    import traceback
                    traceback.print_exc(e)
                    logger.exception("Error in message handling: %s" % e)

        cls.mark_handled(msg)

        # if this is an inbox message, increment our unread inbox count
        if msg.msg_type == INBOX:
            msg.org.increment_unread_msg_count(UNREAD_INBOX_MSGS)

        # record our handling latency for this object
        if msg.queued_on:
            analytics.gauge('temba.handling_latency', (msg.modified_on - msg.queued_on).total_seconds())

        # this is the latency from when the message was received at the channel, which may be different than
        # above if people above us are queueing (or just because clocks are out of sync)
        analytics.gauge('temba.channel_handling_latency', (msg.modified_on - msg.created_on).total_seconds())

    @classmethod
    def get_messages(cls, org, is_archived=False, direction=None, msg_type=None):
        messages = Msg.objects.filter(org=org)

        if is_archived:  # pragma: needs cover
            messages = messages.filter(visibility=Msg.VISIBILITY_ARCHIVED)
        else:
            messages = messages.filter(visibility=Msg.VISIBILITY_VISIBLE)

        if direction:  # pragma: needs cover
            messages = messages.filter(direction=direction)

        if msg_type:  # pragma: needs cover
            messages = messages.filter(msg_type=msg_type)

        return messages.filter(contact__is_test=False)

    @classmethod
    def fail_old_messages(cls):  # pragma: needs cover
        """
        Looks for any errored or queued messages more than a week old and fails them. Messages that old would
        probably be confusing to go out.
        """
        one_week_ago = timezone.now() - timedelta(days=7)
        failed_messages = Msg.objects.filter(created_on__lte=one_week_ago, direction=OUTGOING,
                                             status__in=[QUEUED, PENDING, ERRORED])

        failed_broadcasts = list(failed_messages.order_by('broadcast').values('broadcast').distinct())

        # fail our messages
        failed_messages.update(status='F', modified_on=timezone.now())

        # and update all related broadcast statuses
        for broadcast in Broadcast.objects.filter(id__in=[b['broadcast'] for b in failed_broadcasts]):
            broadcast.update()

    @classmethod
    def get_unread_msg_count(cls, user):
        org = user.get_org()

        key = 'org_unread_msg_count_%d' % org.pk
        unread_count = cache.get(key, None)

        if unread_count is None:
            unread_count = Msg.objects.filter(org=org, visibility=Msg.VISIBILITY_VISIBLE, direction=INCOMING,
                                              msg_type=INBOX, contact__is_test=False,
                                              created_on__gt=org.msg_last_viewed, labels=None).count()
            cache.set(key, unread_count, 900)

        return unread_count

    @classmethod
    def mark_handled(cls, msg):
        """
        Marks an incoming message as HANDLED
        """
        update_fields = ['status', 'modified_on']

        # if flows or IVR haven't claimed this message, then it's going to the inbox
        if not msg.msg_type:
            msg.msg_type = INBOX
            update_fields.append('msg_type')

        msg.status = HANDLED

        # make sure we don't overwrite any async message changes by only saving specific fields
        msg.save(update_fields=update_fields)

    @classmethod
    def mark_error(cls, r, channel, msg, fatal=False):
        """
        Marks an outgoing message as FAILED or ERRORED
        :param msg: a JSON representation of the message or a Msg object
        """
        msg.error_count += 1
        if msg.error_count >= 3 or fatal:
            if isinstance(msg, Msg):
                msg.status_fail()
            else:
                Msg.objects.select_related('org').get(pk=msg.id).status_fail()

            if channel:
                analytics.gauge('temba.msg_failed_%s' % channel.channel_type.lower())
        else:
            msg.status = ERRORED
            msg.next_attempt = timezone.now() + timedelta(minutes=5 * msg.error_count)

            if isinstance(msg, Msg):
                msg.save(update_fields=('status', 'modified_on', 'next_attempt', 'error_count'))
            else:
                Msg.objects.filter(id=msg.id).update(status=msg.status, next_attempt=msg.next_attempt,
                                                     error_count=msg.error_count, modified_on=msg.modified_on)

            # clear that we tried to send this message (otherwise we'll ignore it when we retry)
            pipe = r.pipeline()
            pipe.srem(timezone.now().strftime(MSG_SENT_KEY), str(msg.id))
            pipe.srem((timezone.now() - timedelta(days=1)).strftime(MSG_SENT_KEY), str(msg.id))
            pipe.execute()

            if channel:
                analytics.gauge('temba.msg_errored_%s' % channel.channel_type.lower())

    @classmethod
    def mark_sent(cls, r, msg, status, external_id=None):
        """
        Marks an outgoing message as WIRED or SENT
        :param msg: a JSON representation of the message
        """
        msg.status = status
        msg.sent_on = timezone.now()
        if external_id:
            msg.external_id = external_id

        # use redis to mark this message sent
        pipe = r.pipeline()
        sent_key = timezone.now().strftime(MSG_SENT_KEY)
        pipe.sadd(sent_key, str(msg.id))
        pipe.expire(sent_key, 86400)
        pipe.execute()

        if external_id:
            Msg.objects.filter(id=msg.id).update(status=status, sent_on=msg.sent_on, external_id=external_id)
        else:
            Msg.objects.filter(id=msg.id).update(status=status, sent_on=msg.sent_on)

    @classmethod
    def get_media(cls, msg):
        if msg.media:
            parts = msg.media.split(':', 1)
            if len(parts) == 2:
                return parts
        return None, None

    def as_json(self):
        return dict(direction=self.direction,
                    text=self.text,
                    id=self.id,
                    media=self.media,
                    created_on=self.created_on.strftime('%x %X'),
                    model="msg")

    def simulator_json(self):
        msg_json = self.as_json()
        msg_json['text'] = escape(self.text).replace('\n', "<br/>")
        return msg_json

    @classmethod
    def get_text_parts(cls, text, max_length=160):
        """
        Breaks our message into 160 character parts
        """
        if len(text) < max_length or max_length <= 0:
            return [text]

        else:
            def next_part(text):
                if len(text) <= max_length:
                    return text, None

                else:
                    # search for a space to split on, up to 140 characters in
                    index = max_length
                    while index > max_length - 20:
                        if text[index] == ' ':
                            break
                        index -= 1

                    # couldn't find a good split, oh well, 160 it is
                    if index == max_length - 20:
                        return text[:max_length], text[max_length:]
                    else:
                        return text[:index], text[index + 1:]

            parts = []
            rest = text
            while rest:
                (part, rest) = next_part(rest)
                parts.append(part)

            return parts

    @classmethod
    def get_sync_commands(self, channel, msgs):
        """
        Returns the minimal # of broadcast commands for the given Android channel to uniquely represent all the
        messages which are being sent to tel URNs. This will return an array of dicts that look like:
             dict(cmd="mt_bcast", to=[dict(phone=msg.contact.tel, id=msg.pk) for msg in msgs], msg=broadcast.text))
        """
        commands = []
        current_msg = None
        contact_id_pairs = []

        ordered_msgs = Msg.objects.filter(id__in=[m.id for m in msgs]).order_by('created_on')

        for msg in ordered_msgs:
            if msg.text != current_msg and contact_id_pairs:
                commands.append(dict(cmd='mt_bcast', to=contact_id_pairs, msg=current_msg))
                contact_id_pairs = []

            current_msg = msg.text
            contact_id_pairs.append(dict(phone=msg.contact_urn.path, id=msg.pk))

        if contact_id_pairs:
            commands.append(dict(cmd='mt_bcast', to=contact_id_pairs, msg=current_msg))

        return commands

    def get_last_log(self):
        """
        Gets the last channel log for this message. Performs sorting in Python to ease pre-fetching.
        """
        sorted_logs = None
        if self.channel and self.channel.is_active:
            sorted_logs = sorted(self.channel_logs.all(), key=lambda l: l.created_on, reverse=True)
        return sorted_logs[0] if sorted_logs else None

    def get_media_path(self):

        if self.media:
            # TODO: remove after migration msgs.0053
            if self.media.startswith('http'):  # pragma: needs cover
                return self.media

            if ':' in self.media:
                return self.media.split(':', 1)[1]

    def get_media_type(self):

        if self.media:
            # TODO: remove after migration msgs.0053
            if self.media.startswith('http'):  # pragma: needs cover
                return 'audio'

        if self.media and ':' in self.media:
            type = self.media.split(':', 1)[0]
            if type == 'application/octet-stream':  # pragma: needs cover
                return 'audio'
            return type.split('/', 1)[0]

    def is_media_type_audio(self):
        return Msg.MEDIA_AUDIO == self.get_media_type()

    def is_media_type_video(self):
        return Msg.MEDIA_VIDEO == self.get_media_type()

    def is_media_type_image(self):
        return Msg.MEDIA_IMAGE == self.get_media_type()

    def reply(self, text, user, trigger_send=False, message_context=None, session=None, media=None, msg_type=None):
        return self.contact.send(text, user, trigger_send=trigger_send, message_context=message_context,
                                 response_to=self if self.id else None, session=session, media=media,
                                 msg_type=msg_type or self.msg_type)

    def update(self, cmd):
        """
        Updates our message according to the provided client command
        """
        from temba.api.models import WebHookEvent, SMS_DELIVERED, SMS_SENT, SMS_FAIL
        date = datetime.fromtimestamp(int(cmd['ts']) / 1000).replace(tzinfo=pytz.utc)

        keyword = cmd['cmd']
        handled = False

        if keyword == 'mt_error':
            self.status = ERRORED
            handled = True

        elif keyword == 'mt_fail':
            self.status = FAILED
            handled = True
            WebHookEvent.trigger_sms_event(SMS_FAIL, self, date)

        elif keyword == 'mt_sent':
            self.status = SENT
            self.sent_on = date
            handled = True
            WebHookEvent.trigger_sms_event(SMS_SENT, self, date)

        elif keyword == 'mt_dlvd':
            self.status = DELIVERED
            handled = True
            WebHookEvent.trigger_sms_event(SMS_DELIVERED, self, date)

        self.save()  # first save message status before updating the broadcast status

        # update our broadcast if we have one
        if self.broadcast:
            self.broadcast.update()

        return handled

    def handle(self):
        if self.direction == OUTGOING:
            raise ValueError(ugettext("Cannot process an outgoing message."))

        # process Android and test contact messages inline
        if not self.channel or self.channel.channel_type == Channel.TYPE_ANDROID or self.contact.is_test:
            Msg.process_message(self)

        # others do in celery
        else:
            on_transaction_commit(lambda: push_task(self.org, HANDLER_QUEUE, HANDLE_EVENT_TASK,
                                                    dict(type=MSG_EVENT, id=self.id, from_mage=False, new_contact=False)))

    def build_expressions_context(self, contact_context=None):
        date_format = get_datetime_format(self.org.get_dayfirst())[1]

        return {
            '__default__': self.text,
            'value': self.text,
            'contact': contact_context or self.contact.build_expressions_context(),
            'time': datetime_to_str(self.created_on, format=date_format, tz=self.org.timezone)
        }

    def resend(self):
        """
        Resends this message by creating a clone and triggering a send of that clone
        """
        now = timezone.now()
        (topup_id, amount) = self.org.decrement_credit()  # costs 1 credit to resend message

        # see if we should use a new channel
        channel = self.org.get_send_channel(contact_urn=self.contact_urn)<commit_msg>prioritize msg handling over timeotus and event fires<commit_after>from __future__ import print_function, unicode_literals

import json
import logging
import pytz
import regex
import six
import time
import traceback

from datetime import datetime, timedelta
from django.conf import settings
from django.contrib.auth.models import User
from django.core.cache import cache
from django.core.files.temp import NamedTemporaryFile
from django.db import models, transaction
from django.db.models import Q, Count, Prefetch, Sum
from django.utils import timezone
from django.utils.html import escape
from django.utils.translation import ugettext, ugettext_lazy as _
from temba_expressions.evaluator import EvaluationContext, DateStyle
from temba.assets.models import register_asset_store
from temba.contacts.models import Contact, ContactGroup, ContactURN, URN, TEL_SCHEME
from temba.channels.models import Channel, ChannelEvent
from temba.orgs.models import Org, TopUp, Language, UNREAD_INBOX_MSGS
from temba.schedules.models import Schedule
from temba.utils import get_datetime_format, datetime_to_str, analytics, chunk_list, on_transaction_commit
from temba.utils.cache import get_cacheable_attr
from temba.utils.export import BaseExportTask, BaseExportAssetStore
from temba.utils.expressions import evaluate_template
from temba.utils.models import SquashableModel, TembaModel
from temba.utils.queues import DEFAULT_PRIORITY, push_task, LOW_PRIORITY, HIGH_PRIORITY
from .handler import MessageHandler

logger = logging.getLogger(__name__)
__message_handlers = None

MSG_QUEUE = 'msgs'
SEND_MSG_TASK = 'send_msg_task'

HANDLER_QUEUE = 'handler'
HANDLE_EVENT_TASK = 'handle_event_task'
MSG_EVENT = 'msg'
FIRE_EVENT = 'fire'
TIMEOUT_EVENT = 'timeout'

BATCH_SIZE = 500

INITIALIZING = 'I'
PENDING = 'P'
QUEUED = 'Q'
WIRED = 'W'
SENT = 'S'
DELIVERED = 'D'
HANDLED = 'H'
ERRORED = 'E'
FAILED = 'F'
RESENT = 'R'

INCOMING = 'I'
OUTGOING = 'O'

INBOX = 'I'
FLOW = 'F'
IVR = 'V'
USSD = 'U'

MSG_SENT_KEY = 'msgs_sent_%y_%m_%d'

# status codes used for both messages and broadcasts (single char constant, human readable, API readable)
STATUS_CONFIG = (
    # special state for flows used to hold off sending the message until the flow is ready to receive a response
    (INITIALIZING, _("Initializing"), 'initializing'),

    (PENDING, _("Pending"), 'pending'),        # initial state for all messages

    # valid only for outgoing messages
    (QUEUED, _("Queued"), 'queued'),
    (WIRED, _("Wired"), 'wired'),              # message was handed off to the provider and credits were deducted for it
    (SENT, _("Sent"), 'sent'),                 # we have confirmation that a message was sent
    (DELIVERED, _("Delivered"), 'delivered'),

    # valid only for incoming messages
    (HANDLED, _("Handled"), 'handled'),

    (ERRORED, _("Error Sending"), 'errored'),  # there was an error during delivery
    (FAILED, _("Failed Sending"), 'failed'),   # we gave up on sending this message
    (RESENT, _("Resent message"), 'resent'),   # we retried this message
)


def get_message_handlers():
    """
    Initializes all our message handlers
    """
    global __message_handlers
    if not __message_handlers:
        handlers = []
        for handler_class in settings.MESSAGE_HANDLERS:
            try:
                cls = MessageHandler.find(handler_class)
                handlers.append(cls())
            except Exception as ee:  # pragma: no cover
                traceback.print_exc(ee)

        __message_handlers = handlers

    return __message_handlers


def get_unique_recipients(urns, contacts, groups):
    """
    Builds a list of the unique contacts and URNs by merging urns, contacts and groups
    """
    unique_urns = set()
    unique_contacts = set()
    included_by_urn = set()  # contact ids of contacts included by URN

    for urn in urns:
        unique_urns.add(urn)
        included_by_urn.add(urn.contact_id)

    for group in groups:
        for contact in group.contacts.all():
            if contact.id not in included_by_urn:
                unique_contacts.add(contact)

    for contact in contacts:
        if contact.id not in included_by_urn:
            unique_contacts.add(contact)

    return unique_urns, unique_contacts


class UnreachableException(Exception):
    """
    Exception thrown when a message is being sent to a contact that we don't have a sendable URN for
    """
    pass


class BroadcastRecipient(models.Model):
    """
    Through table for broadcast recipients many-to-many
    """
    broadcast = models.ForeignKey('msgs.Broadcast')

    contact = models.ForeignKey(Contact)

    purged_status = models.CharField(null=True, max_length=1,
                                     help_text=_("Used when broadcast is purged to record contact's message's state"))

    class Meta:
        db_table = 'msgs_broadcast_recipients'


@six.python_2_unicode_compatible
class Broadcast(models.Model):
    """
    A broadcast is a message that is sent out to more than one recipient, such
    as a ContactGroup or a list of Contacts. It's nothing more than a way to tie
    messages sent from the same bundle together
    """
    STATUS_CHOICES = [(s[0], s[1]) for s in STATUS_CONFIG]

    BULK_THRESHOLD = 50  # use bulk priority for messages if number of recipients greater than this

    org = models.ForeignKey(Org, verbose_name=_("Org"),
                            help_text=_("The org this broadcast is connected to"))

    groups = models.ManyToManyField(ContactGroup, verbose_name=_("Groups"), related_name='addressed_broadcasts',
                                    help_text=_("The groups to send the message to"))

    contacts = models.ManyToManyField(Contact, verbose_name=_("Contacts"), related_name='addressed_broadcasts',
                                      help_text=_("Individual contacts included in this message"))

    urns = models.ManyToManyField(ContactURN, verbose_name=_("URNs"), related_name='addressed_broadcasts',
                                  help_text=_("Individual URNs included in this message"))

    recipients = models.ManyToManyField(Contact, through=BroadcastRecipient, verbose_name=_("Recipients"),
                                        related_name='broadcasts',
                                        help_text=_("The contacts which received this message"))

    recipient_count = models.IntegerField(verbose_name=_("Number of recipients"), null=True,
                                          help_text=_("Number of urns which received this broadcast"))

    text = models.TextField(max_length=settings.MSG_FIELD_SIZE, verbose_name=_("Text"),
                            help_text=_("The message to send out"))

    channel = models.ForeignKey(Channel, null=True, verbose_name=_("Channel"),
                                help_text=_("Channel to use for message sending"))

    status = models.CharField(max_length=1, verbose_name=_("Status"), choices=STATUS_CHOICES, default=INITIALIZING,
                              help_text=_("The current status for this broadcast"))

    schedule = models.OneToOneField(Schedule, verbose_name=_("Schedule"), null=True,
                                    help_text=_("Our recurring schedule if we have one"), related_name="broadcast")

    parent = models.ForeignKey('Broadcast', verbose_name=_("Parent"), null=True, related_name='children')

    language_dict = models.TextField(verbose_name=_("Translations"),
                                     help_text=_("The localized versions of the broadcast"), null=True)

    base_language = models.CharField(max_length=4, null=True, blank=True,
                                     help_text=_('The language used to send this to contacts without a language'))

    is_active = models.BooleanField(default=True, help_text="Whether this broadcast is active")

    created_by = models.ForeignKey(User, related_name="%(app_label)s_%(class)s_creations",
                                   help_text="The user which originally created this item")

    created_on = models.DateTimeField(default=timezone.now, blank=True, editable=False, db_index=True,
                                      help_text=_("When this broadcast was created"))

    modified_by = models.ForeignKey(User, related_name="%(app_label)s_%(class)s_modifications",
                                    help_text="The user which last modified this item")

    modified_on = models.DateTimeField(auto_now=True,
                                       help_text="When this item was last modified")

    purged = models.BooleanField(default=False,
                                 help_text="If the messages for this broadcast have been purged")

    media_dict = models.TextField(verbose_name=_("Media"),
                                  help_text=_("The localized versions of the media"), null=True)

    @classmethod
    def create(cls, org, user, text, recipients, channel=None, media_dict=None, **kwargs):
        create_args = dict(org=org, text=text, channel=channel, media_dict=media_dict, created_by=user,
                           modified_by=user)
        create_args.update(kwargs)
        broadcast = Broadcast.objects.create(**create_args)
        broadcast.update_recipients(recipients)
        return broadcast

    def update_contacts(self, contact_ids):
        """
        Optimization for broadcasts that only contain contacts. Updates our contacts according to the passed in
        queryset or array.
        """
        self.urns.clear()
        self.groups.clear()
        self.contacts.clear()

        # get our through model
        RelatedModel = self.contacts.through

        # clear called automatically by django
        for chunk in chunk_list(contact_ids, 1000):
            bulk_contacts = [RelatedModel(contact_id=id, broadcast_id=self.id) for id in chunk]
            RelatedModel.objects.bulk_create(bulk_contacts)

        self.recipient_count = len(contact_ids)
        self.save(update_fields=('recipient_count',))

        # cache on object for use in subsequent send(..) calls
        delattr(self, '_recipient_cache')

    def update_recipients(self, recipients):
        """
        Updates the recipients which may be contact groups, contacts or contact URNs. Normally you can't update a
        broadcast after it has been created - the exception is scheduled broadcasts which are never really sent (clones
        of them are sent).
        """
        urns = []
        contacts = []
        groups = []

        for recipient in recipients:
            if isinstance(recipient, ContactURN):
                urns.append(recipient)
            elif isinstance(recipient, Contact):
                contacts.append(recipient)
            elif isinstance(recipient, ContactGroup):
                groups.append(recipient)
            else:  # pragma: needs cover
                raise ValueError("Recipient item is not a Contact, ContactURN or ContactGroup")

        self.urns.clear()
        self.urns.add(*urns)
        self.contacts.clear()
        self.contacts.add(*contacts)
        self.groups.clear()
        self.groups.add(*groups)

        urns, contacts = get_unique_recipients(urns, contacts, groups)

        # update the recipient count - the number of messages we intend to send
        self.recipient_count = len(urns) + len(contacts)
        self.save(update_fields=('recipient_count',))

        # cache on object for use in subsequent send(..) calls
        setattr(self, '_recipient_cache', (urns, contacts))

        return urns, contacts

    def has_pending_fire(self):  # pragma: needs cover
        return self.schedule and self.schedule.has_pending_fire()

    def fire(self):
        recipients = list(self.urns.all()) + list(self.contacts.all()) + list(self.groups.all())
        broadcast = Broadcast.create(self.org, self.created_by, self.text, recipients,
                                     parent=self, modified_by=self.modified_by)

        broadcast.send(trigger_send=True)

        return broadcast

    @classmethod
    def get_broadcasts(cls, org, scheduled=False):
        qs = Broadcast.objects.filter(org=org).exclude(contacts__is_test=True)
        return qs.exclude(schedule=None) if scheduled else qs.filter(schedule=None)

    def get_messages(self):
        return self.msgs.exclude(status=RESENT)

    def get_messages_substitution_complete(self):  # pragma: needs cover
        return self.get_messages().filter(has_template_error=False)

    def get_messages_substitution_incomplete(self):  # pragma: needs cover
        return self.get_messages().filter(has_template_error=True)

    def get_message_count(self):
        return self.get_messages().count()

    def get_message_sending_count(self):  # pragma: needs cover
        return self.get_messages().filter(status__in=[PENDING, QUEUED]).count()

    def get_message_sent_count(self):  # pragma: needs cover
        return self.get_messages().filter(status__in=[SENT, DELIVERED, WIRED]).count()

    def get_message_delivered_count(self):  # pragma: needs cover
        return self.get_messages().filter(status=DELIVERED).count()

    def get_message_failed_count(self):  # pragma: needs cover
        return self.get_messages().filter(status__in=[FAILED, RESENT]).count()

    def get_sync_commands(self, channel):
        """
        Returns the minimal # of broadcast commands for the given Android channel to uniquely represent all the
        messages which are being sent to tel URNs. This will return an array of dicts that look like:
             dict(cmd="mt_bcast", to=[dict(phone=msg.contact.tel, id=msg.pk) for msg in msgs], msg=broadcast.text))
        """
        commands = []
        current_msg = None
        contact_id_pairs = []

        pending = self.get_messages().filter(status__in=[PENDING, QUEUED, WIRED], channel=channel,
                                             contact_urn__scheme=TEL_SCHEME).select_related('contact_urn').order_by('text', 'pk')

        for msg in pending:
            if msg.text != current_msg and contact_id_pairs:
                commands.append(dict(cmd='mt_bcast', to=contact_id_pairs, msg=current_msg))
                contact_id_pairs = []

            current_msg = msg.text
            contact_id_pairs.append(dict(phone=msg.contact_urn.path, id=msg.pk))

        if contact_id_pairs:
            commands.append(dict(cmd='mt_bcast', to=contact_id_pairs, msg=current_msg))

        return commands

    def get_preferred_languages(self, contact, base_language=None, org=None):
        """
        Gets the ordered list of language preferences for the given contact
        """
        org = org or self.org  # org object can be provided to allow caching of org languages
        preferred_languages = []

        if org.primary_language:
            preferred_languages.append(org.primary_language.iso_code)

        if base_language:
            preferred_languages.append(base_language)

        # if contact has a language and it's a valid org language, it has priority
        if contact.language and contact.language in org.get_language_codes():
            preferred_languages = [contact.language] + preferred_languages

        return preferred_languages

    def get_translations(self):
        if not self.language_dict:
            return []
        return get_cacheable_attr(self, '_translations', lambda: json.loads(self.language_dict))

    def get_media_translations(self):
        if not self.media_dict:
            return dict()
        return get_cacheable_attr(self, '_media_translations', lambda: json.loads(self.media_dict))

    def get_translated_text(self, contact, org=None):
        """
        Gets the appropriate translation for the given contact. base_language may be provided
        """
        translations = self.get_translations()
        preferred_languages = self.get_preferred_languages(contact, self.base_language, org)
        return Language.get_localized_text(translations, preferred_languages, self.text)

    def get_translated_media(self, contact, org=None):
        """
        Gets the appropriate media for the given contact. base_language may be provided
        """
        media_translations = self.get_media_translations()
        preferred_languages = self.get_preferred_languages(contact, self.base_language, org)
        return Language.get_localized_text(media_translations, preferred_languages, None)

    def send(self, trigger_send=True, message_context=None, response_to=None, status=PENDING, msg_type=INBOX,
             created_on=None, partial_recipients=None, run_map=None):
        """
        Sends this broadcast by creating outgoing messages for each recipient.
        """
        # ignore mock messages
        if response_to and not response_to.id:
            response_to = None

        # cannot ask for sending by us AND specify a created on, blow up in that case
        if trigger_send and created_on:  # pragma: no cover
            raise ValueError("Cannot trigger send and specify a created_on, breaks creating batches")

        if partial_recipients:
            # if flow is being started, it'll provide a batch of unique contacts itself
            urns, contacts = partial_recipients
        elif hasattr(self, '_recipient_cache'):
            # look to see if previous call to update_recipients left a cached value
            urns, contacts = self._recipient_cache
        else:
            # otherwise fetch everything and calculate
            urns, contacts = get_unique_recipients(self.urns.all(), self.contacts.all(), self.groups.all())

        Contact.bulk_cache_initialize(self.org, contacts)
        recipients = list(urns) + list(contacts)

        RelatedRecipient = Broadcast.recipients.through

        # we batch up our SQL calls to speed up the creation of our SMS objects
        batch = []
        recipient_batch = []

        # our priority is based on the number of recipients
        priority = Msg.PRIORITY_NORMAL
        if len(recipients) == 1:
            priority = Msg.PRIORITY_HIGH
        elif len(recipients) >= self.BULK_THRESHOLD:
            priority = Msg.PRIORITY_BULK

        # if they didn't pass in a created on, create one ourselves
        if not created_on:
            created_on = timezone.now()

        # pre-fetch channels to reduce database hits
        org = Org.objects.filter(pk=self.org.id).prefetch_related('channels').first()

        for recipient in recipients:
            contact = recipient if isinstance(recipient, Contact) else recipient.contact

            # get the appropriate translation for this contact
            text = self.get_translated_text(contact)

            media = self.get_translated_media(contact)
            if media:
                media_type, media_url = media.split(':')

                # if we have a localized media, create the url
                if media_url:
                    media = "%s:https://%s/%s" % (media_type, settings.AWS_BUCKET_DOMAIN, media_url)

            # add in our parent context if the message references @parent
            if run_map:
                run = run_map.get(recipient.pk, None)
                if run and run.flow:
                    # a bit kludgy here, but should avoid most unnecessary context creations.
                    # since this path is an optimization for flow starts, we don't need to
                    # worry about the @child context.
                    if 'parent' in text:
                        if run.parent:
                            from temba.flows.models import Flow
                            message_context = message_context.copy()
                            message_context.update(dict(parent=Flow.build_flow_context(run.parent.flow, run.parent.contact)))

            try:
                msg = Msg.create_outgoing(org,
                                          self.created_by,
                                          recipient,
                                          text,
                                          broadcast=self,
                                          channel=self.channel,
                                          response_to=response_to,
                                          message_context=message_context,
                                          status=status,
                                          msg_type=msg_type,
                                          insert_object=False,
                                          media=media,
                                          priority=priority,
                                          created_on=created_on)

            except UnreachableException:
                # there was no way to reach this contact, do not create a message
                msg = None

            # only add it to our batch if it was legit
            if msg:
                batch.append(msg)
                # keep track of this URN as a recipient
                recipient_batch.append(RelatedRecipient(contact_id=msg.contact_id, broadcast_id=self.id))

            # we commit our messages in batches
            if len(batch) >= BATCH_SIZE:
                Msg.objects.bulk_create(batch)
                RelatedRecipient.objects.bulk_create(recipient_batch)

                # send any messages
                if trigger_send:
                    self.org.trigger_send(Msg.objects.filter(broadcast=self, created_on=created_on).select_related('contact', 'contact_urn', 'channel'))

                    # increment our created on so we can load our next batch
                    created_on = created_on + timedelta(seconds=1)

                batch = []
                recipient_batch = []

        # commit any remaining objects
        if batch:
            Msg.objects.bulk_create(batch)
            RelatedRecipient.objects.bulk_create(recipient_batch)

            if trigger_send:
                self.org.trigger_send(Msg.objects.filter(broadcast=self, created_on=created_on).select_related('contact', 'contact_urn', 'channel'))

        # for large batches, status is handled externally
        # we do this as with the high concurrency of sending we can run into postgresl deadlocks
        # (this could be our fault, or could be: http://www.postgresql.org/message-id/20140731233051.GN17765@andrew-ThinkPad-X230)
        if not partial_recipients:
            self.status = QUEUED if len(recipients) > 0 else SENT
            self.save(update_fields=('status',))

    def update(self):
        """
        Check the status of our messages and update ours accordingly
        """
        # build a map from status to the count for that status
        statuses = self.get_messages().values('status').order_by('status').annotate(count=Count('status'))
        total = 0
        status_map = dict()
        for status in statuses:
            status_map[status['status']] = status['count']
            total += status['count']

        # if errored msgs are greater than the half of all msgs
        if status_map.get(ERRORED, 0) > total / 2:
            self.status = ERRORED

        # if there are more than half failed, show failed
        elif status_map.get(FAILED, 0) > total / 2:
            self.status = FAILED

        # if there are any in Q, we are Q
        elif status_map.get(QUEUED, 0) or status_map.get(PENDING, 0):
            self.status = QUEUED

        # at this point we are either sent or delivered

        # if there are any messages that are only in a sent state
        elif status_map.get(SENT, 0) or status_map.get(WIRED, 0):
            self.status = SENT

        # otherwise, all messages delivered
        elif status_map.get(DELIVERED, 0) == total:
            self.status = DELIVERED

        self.save(update_fields=['status'])

    def __str__(self):
        return "%s (%s)" % (self.org.name, self.pk)


@six.python_2_unicode_compatible
class Msg(models.Model):
    """
    Messages are the main building blocks of a RapidPro application. Channels send and receive
    these, Triggers and Flows handle them when appropriate.

    Messages are either inbound or outbound and can have varying states depending on their
    direction. Generally an outbound message will go through the following states:

      INITIALIZING > QUEUED > WIRED > SENT > DELIVERED

    If things go wrong, they can be put into an ERRORED state where they can be retried. Once
    we've given up then they can be put in the FAILED state.

    Inbound messages are much simpler. They start as PENDING and the can be picked up by Triggers
    or Flows where they would get set to the HANDLED state once they've been dealt with.
    """
    STATUS_CHOICES = [(s[0], s[1]) for s in STATUS_CONFIG]

    VISIBILITY_VISIBLE = 'V'
    VISIBILITY_ARCHIVED = 'A'
    VISIBILITY_DELETED = 'D'

    # single char flag, human readable name, API readable name
    VISIBILITY_CONFIG = ((VISIBILITY_VISIBLE, _("Visible"), 'visible'),
                         (VISIBILITY_ARCHIVED, _("Archived"), 'archived'),
                         (VISIBILITY_DELETED, _("Deleted"), 'deleted'))

    VISIBILITY_CHOICES = [(s[0], s[1]) for s in VISIBILITY_CONFIG]

    DIRECTION_CHOICES = ((INCOMING, _("Incoming")),
                         (OUTGOING, _("Outgoing")))

    MSG_TYPES = ((INBOX, _("Inbox Message")),
                 (FLOW, _("Flow Message")),
                 (IVR, _("IVR Message")),
                 (USSD, _("USSD Message")))

    MEDIA_GPS = 'geo'
    MEDIA_IMAGE = 'image'
    MEDIA_VIDEO = 'video'
    MEDIA_AUDIO = 'audio'

    MEDIA_TYPES = [MEDIA_AUDIO, MEDIA_GPS, MEDIA_IMAGE, MEDIA_VIDEO]

    PRIORITY_HIGH = 1000
    PRIORITY_NORMAL = 500
    PRIORITY_BULK = 100

    MAX_SIZE = settings.MSG_FIELD_SIZE

    org = models.ForeignKey(Org, related_name='msgs', verbose_name=_("Org"),
                            help_text=_("The org this message is connected to"))

    channel = models.ForeignKey(Channel, null=True,
                                related_name='msgs', verbose_name=_("Channel"),
                                help_text=_("The channel object that this message is associated with"))

    contact = models.ForeignKey(Contact,
                                related_name='msgs', verbose_name=_("Contact"),
                                help_text=_("The contact this message is communicating with"))

    contact_urn = models.ForeignKey(ContactURN, null=True,
                                    related_name='msgs', verbose_name=_("Contact URN"),
                                    help_text=_("The URN this message is communicating with"))

    broadcast = models.ForeignKey(Broadcast, null=True, blank=True,
                                  related_name='msgs', verbose_name=_("Broadcast"),
                                  help_text=_("If this message was sent to more than one recipient"))

    text = models.TextField(max_length=MAX_SIZE, verbose_name=_("Text"),
                            help_text=_("The actual message content that was sent"))

    priority = models.IntegerField(default=PRIORITY_NORMAL,
                                   help_text=_("The priority for this message to be sent, higher is higher priority"))

    created_on = models.DateTimeField(verbose_name=_("Created On"), db_index=True,
                                      help_text=_("When this message was created"))

    modified_on = models.DateTimeField(null=True, blank=True, verbose_name=_("Modified On"), auto_now=True,
                                       help_text=_("When this message was last modified"))

    sent_on = models.DateTimeField(null=True, blank=True, verbose_name=_("Sent On"),
                                   help_text=_("When this message was sent to the endpoint"))

    queued_on = models.DateTimeField(null=True, blank=True, verbose_name=_("Queued On"),
                                     help_text=_("When this message was queued to be sent or handled."))

    direction = models.CharField(max_length=1, choices=DIRECTION_CHOICES, verbose_name=_("Direction"),
                                 help_text=_("The direction for this message, either incoming or outgoing"))

    status = models.CharField(max_length=1, choices=STATUS_CHOICES, default='P', verbose_name=_("Status"), db_index=True,
                              help_text=_("The current status for this message"))

    response_to = models.ForeignKey('Msg', null=True, blank=True, related_name='responses',
                                    verbose_name=_("Response To"), db_index=False,
                                    help_text=_("The message that this message is in reply to"))

    labels = models.ManyToManyField('Label', related_name='msgs', verbose_name=_("Labels"),
                                    help_text=_("Any labels on this message"))

    visibility = models.CharField(max_length=1, choices=VISIBILITY_CHOICES, default=VISIBILITY_VISIBLE, db_index=True,
                                  verbose_name=_("Visibility"),
                                  help_text=_("The current visibility of this message, either visible, archived or deleted"))

    has_template_error = models.BooleanField(default=False, verbose_name=_("Has Template Error"),
                                             help_text=_("Whether data for variable substitution are missing"))

    msg_type = models.CharField(max_length=1, choices=MSG_TYPES, null=True, verbose_name=_("Message Type"),
                                help_text=_('The type of this message'))

    msg_count = models.IntegerField(default=1, verbose_name=_("Message Count"),
                                    help_text=_("The number of messages that were used to send this message, calculated on Twilio channels"))

    error_count = models.IntegerField(default=0, verbose_name=_("Error Count"),
                                      help_text=_("The number of times this message has errored"))

    next_attempt = models.DateTimeField(auto_now_add=True, verbose_name=_("Next Attempt"),
                                        help_text=_("When we should next attempt to deliver this message"))

    external_id = models.CharField(max_length=255, null=True, blank=True, verbose_name=_("External ID"),
                                   help_text=_("External id used for integrating with callbacks from other APIs"))

    topup = models.ForeignKey(TopUp, null=True, blank=True, related_name='msgs', on_delete=models.SET_NULL,
                              help_text="The topup that this message was deducted from")

    media = models.URLField(null=True, blank=True, max_length=255,
                            help_text=_("The media associated with this message if any"))

    session = models.ForeignKey('channels.ChannelSession', null=True,
                                help_text=_("The session this message was a part of if any"))

    @classmethod
    def send_messages(cls, all_msgs):
        """
        Adds the passed in messages to our sending queue, this will also update the status of the message to
        queued.
        :return:
        """
        task_msgs = []
        task_priority = None
        last_contact = None

        # we send in chunks of 1,000 to help with contention
        for msg_chunk in chunk_list(all_msgs, 1000):
            # create a temporary list of our chunk so we can iterate more than once
            msgs = [msg for msg in msg_chunk]

            # build our id list
            msg_ids = set([m.id for m in msgs])

            with transaction.atomic():
                queued_on = timezone.now()

                # update them to queued
                send_messages = Msg.objects.filter(id__in=msg_ids)\
                                           .exclude(channel__channel_type=Channel.TYPE_ANDROID)\
                                           .exclude(msg_type__in=(IVR, USSD))\
                                           .exclude(topup=None)\
                                           .exclude(contact__is_test=True)
                send_messages.update(status=QUEUED, queued_on=queued_on, modified_on=queued_on)

                # now push each onto our queue
                for msg in msgs:
                    if (msg.msg_type != IVR and msg.msg_type != USSD and msg.channel and msg.channel.channel_type != Channel.TYPE_ANDROID) and \
                            msg.topup and not msg.contact.is_test:

                        # if this is a different contact than our last, and we have msgs for that last contact, queue the task
                        if task_msgs and last_contact != msg.contact_id:
                            # if no priority was set, default to DEFAULT
                            if task_priority is None:  # pragma: needs cover
                                task_priority = DEFAULT_PRIORITY

                            push_task(task_msgs[0]['org'], MSG_QUEUE, SEND_MSG_TASK, task_msgs, priority=task_priority)
                            task_msgs = []
                            task_priority = None

                        # serialize the model to a dictionary
                        msg.queued_on = queued_on
                        task = msg.as_task_json()

                        # only be low priority if no priority has been set for this task group
                        if msg.priority == Msg.PRIORITY_BULK and task_priority is None:
                            task_priority = LOW_PRIORITY
                        elif msg.priority == Msg.PRIORITY_HIGH:
                            task_priority = HIGH_PRIORITY

                        task_msgs.append(task)
                        last_contact = msg.contact_id

        # send our last msgs
        if task_msgs:
            if task_priority is None:
                task_priority = DEFAULT_PRIORITY
            push_task(task_msgs[0]['org'], MSG_QUEUE, SEND_MSG_TASK, task_msgs, priority=task_priority)

    @classmethod
    def process_message(cls, msg):
        """
        Processes a message, running it through all our handlers
        """
        handlers = get_message_handlers()

        if msg.contact.is_blocked:
            msg.visibility = Msg.VISIBILITY_ARCHIVED
            msg.save(update_fields=['visibility', 'modified_on'])
        else:
            for handler in handlers:
                try:
                    start = None
                    if settings.DEBUG:  # pragma: no cover
                        start = time.time()

                    handled = handler.handle(msg)

                    if start:  # pragma: no cover
                        print("[%0.2f] %s for %d" % (time.time() - start, handler.name, msg.pk or 0))

                    if handled:
                        break
                except Exception as e:  # pragma: no cover
                    import traceback
                    traceback.print_exc(e)
                    logger.exception("Error in message handling: %s" % e)

        cls.mark_handled(msg)

        # if this is an inbox message, increment our unread inbox count
        if msg.msg_type == INBOX:
            msg.org.increment_unread_msg_count(UNREAD_INBOX_MSGS)

        # record our handling latency for this object
        if msg.queued_on:
            analytics.gauge('temba.handling_latency', (msg.modified_on - msg.queued_on).total_seconds())

        # this is the latency from when the message was received at the channel, which may be different than
        # above if people above us are queueing (or just because clocks are out of sync)
        analytics.gauge('temba.channel_handling_latency', (msg.modified_on - msg.created_on).total_seconds())

    @classmethod
    def get_messages(cls, org, is_archived=False, direction=None, msg_type=None):
        messages = Msg.objects.filter(org=org)

        if is_archived:  # pragma: needs cover
            messages = messages.filter(visibility=Msg.VISIBILITY_ARCHIVED)
        else:
            messages = messages.filter(visibility=Msg.VISIBILITY_VISIBLE)

        if direction:  # pragma: needs cover
            messages = messages.filter(direction=direction)

        if msg_type:  # pragma: needs cover
            messages = messages.filter(msg_type=msg_type)

        return messages.filter(contact__is_test=False)

    @classmethod
    def fail_old_messages(cls):  # pragma: needs cover
        """
        Looks for any errored or queued messages more than a week old and fails them. Messages that old would
        probably be confusing to go out.
        """
        one_week_ago = timezone.now() - timedelta(days=7)
        failed_messages = Msg.objects.filter(created_on__lte=one_week_ago, direction=OUTGOING,
                                             status__in=[QUEUED, PENDING, ERRORED])

        failed_broadcasts = list(failed_messages.order_by('broadcast').values('broadcast').distinct())

        # fail our messages
        failed_messages.update(status='F', modified_on=timezone.now())

        # and update all related broadcast statuses
        for broadcast in Broadcast.objects.filter(id__in=[b['broadcast'] for b in failed_broadcasts]):
            broadcast.update()

    @classmethod
    def get_unread_msg_count(cls, user):
        org = user.get_org()

        key = 'org_unread_msg_count_%d' % org.pk
        unread_count = cache.get(key, None)

        if unread_count is None:
            unread_count = Msg.objects.filter(org=org, visibility=Msg.VISIBILITY_VISIBLE, direction=INCOMING,
                                              msg_type=INBOX, contact__is_test=False,
                                              created_on__gt=org.msg_last_viewed, labels=None).count()
            cache.set(key, unread_count, 900)

        return unread_count

    @classmethod
    def mark_handled(cls, msg):
        """
        Marks an incoming message as HANDLED
        """
        update_fields = ['status', 'modified_on']

        # if flows or IVR haven't claimed this message, then it's going to the inbox
        if not msg.msg_type:
            msg.msg_type = INBOX
            update_fields.append('msg_type')

        msg.status = HANDLED

        # make sure we don't overwrite any async message changes by only saving specific fields
        msg.save(update_fields=update_fields)

    @classmethod
    def mark_error(cls, r, channel, msg, fatal=False):
        """
        Marks an outgoing message as FAILED or ERRORED
        :param msg: a JSON representation of the message or a Msg object
        """
        msg.error_count += 1
        if msg.error_count >= 3 or fatal:
            if isinstance(msg, Msg):
                msg.status_fail()
            else:
                Msg.objects.select_related('org').get(pk=msg.id).status_fail()

            if channel:
                analytics.gauge('temba.msg_failed_%s' % channel.channel_type.lower())
        else:
            msg.status = ERRORED
            msg.next_attempt = timezone.now() + timedelta(minutes=5 * msg.error_count)

            if isinstance(msg, Msg):
                msg.save(update_fields=('status', 'modified_on', 'next_attempt', 'error_count'))
            else:
                Msg.objects.filter(id=msg.id).update(status=msg.status, next_attempt=msg.next_attempt,
                                                     error_count=msg.error_count, modified_on=msg.modified_on)

            # clear that we tried to send this message (otherwise we'll ignore it when we retry)
            pipe = r.pipeline()
            pipe.srem(timezone.now().strftime(MSG_SENT_KEY), str(msg.id))
            pipe.srem((timezone.now() - timedelta(days=1)).strftime(MSG_SENT_KEY), str(msg.id))
            pipe.execute()

            if channel:
                analytics.gauge('temba.msg_errored_%s' % channel.channel_type.lower())

    @classmethod
    def mark_sent(cls, r, msg, status, external_id=None):
        """
        Marks an outgoing message as WIRED or SENT
        :param msg: a JSON representation of the message
        """
        msg.status = status
        msg.sent_on = timezone.now()
        if external_id:
            msg.external_id = external_id

        # use redis to mark this message sent
        pipe = r.pipeline()
        sent_key = timezone.now().strftime(MSG_SENT_KEY)
        pipe.sadd(sent_key, str(msg.id))
        pipe.expire(sent_key, 86400)
        pipe.execute()

        if external_id:
            Msg.objects.filter(id=msg.id).update(status=status, sent_on=msg.sent_on, external_id=external_id)
        else:
            Msg.objects.filter(id=msg.id).update(status=status, sent_on=msg.sent_on)

    @classmethod
    def get_media(cls, msg):
        if msg.media:
            parts = msg.media.split(':', 1)
            if len(parts) == 2:
                return parts
        return None, None

    def as_json(self):
        return dict(direction=self.direction,
                    text=self.text,
                    id=self.id,
                    media=self.media,
                    created_on=self.created_on.strftime('%x %X'),
                    model="msg")

    def simulator_json(self):
        msg_json = self.as_json()
        msg_json['text'] = escape(self.text).replace('\n', "<br/>")
        return msg_json

    @classmethod
    def get_text_parts(cls, text, max_length=160):
        """
        Breaks our message into 160 character parts
        """
        if len(text) < max_length or max_length <= 0:
            return [text]

        else:
            def next_part(text):
                if len(text) <= max_length:
                    return text, None

                else:
                    # search for a space to split on, up to 140 characters in
                    index = max_length
                    while index > max_length - 20:
                        if text[index] == ' ':
                            break
                        index -= 1

                    # couldn't find a good split, oh well, 160 it is
                    if index == max_length - 20:
                        return text[:max_length], text[max_length:]
                    else:
                        return text[:index], text[index + 1:]

            parts = []
            rest = text
            while rest:
                (part, rest) = next_part(rest)
                parts.append(part)

            return parts

    @classmethod
    def get_sync_commands(self, channel, msgs):
        """
        Returns the minimal # of broadcast commands for the given Android channel to uniquely represent all the
        messages which are being sent to tel URNs. This will return an array of dicts that look like:
             dict(cmd="mt_bcast", to=[dict(phone=msg.contact.tel, id=msg.pk) for msg in msgs], msg=broadcast.text))
        """
        commands = []
        current_msg = None
        contact_id_pairs = []

        ordered_msgs = Msg.objects.filter(id__in=[m.id for m in msgs]).order_by('created_on')

        for msg in ordered_msgs:
            if msg.text != current_msg and contact_id_pairs:
                commands.append(dict(cmd='mt_bcast', to=contact_id_pairs, msg=current_msg))
                contact_id_pairs = []

            current_msg = msg.text
            contact_id_pairs.append(dict(phone=msg.contact_urn.path, id=msg.pk))

        if contact_id_pairs:
            commands.append(dict(cmd='mt_bcast', to=contact_id_pairs, msg=current_msg))

        return commands

    def get_last_log(self):
        """
        Gets the last channel log for this message. Performs sorting in Python to ease pre-fetching.
        """
        sorted_logs = None
        if self.channel and self.channel.is_active:
            sorted_logs = sorted(self.channel_logs.all(), key=lambda l: l.created_on, reverse=True)
        return sorted_logs[0] if sorted_logs else None

    def get_media_path(self):

        if self.media:
            # TODO: remove after migration msgs.0053
            if self.media.startswith('http'):  # pragma: needs cover
                return self.media

            if ':' in self.media:
                return self.media.split(':', 1)[1]

    def get_media_type(self):

        if self.media:
            # TODO: remove after migration msgs.0053
            if self.media.startswith('http'):  # pragma: needs cover
                return 'audio'

        if self.media and ':' in self.media:
            type = self.media.split(':', 1)[0]
            if type == 'application/octet-stream':  # pragma: needs cover
                return 'audio'
            return type.split('/', 1)[0]

    def is_media_type_audio(self):
        return Msg.MEDIA_AUDIO == self.get_media_type()

    def is_media_type_video(self):
        return Msg.MEDIA_VIDEO == self.get_media_type()

    def is_media_type_image(self):
        return Msg.MEDIA_IMAGE == self.get_media_type()

    def reply(self, text, user, trigger_send=False, message_context=None, session=None, media=None, msg_type=None):
        return self.contact.send(text, user, trigger_send=trigger_send, message_context=message_context,
                                 response_to=self if self.id else None, session=session, media=media,
                                 msg_type=msg_type or self.msg_type)

    def update(self, cmd):
        """
        Updates our message according to the provided client command
        """
        from temba.api.models import WebHookEvent, SMS_DELIVERED, SMS_SENT, SMS_FAIL
        date = datetime.fromtimestamp(int(cmd['ts']) / 1000).replace(tzinfo=pytz.utc)

        keyword = cmd['cmd']
        handled = False

        if keyword == 'mt_error':
            self.status = ERRORED
            handled = True

        elif keyword == 'mt_fail':
            self.status = FAILED
            handled = True
            WebHookEvent.trigger_sms_event(SMS_FAIL, self, date)

        elif keyword == 'mt_sent':
            self.status = SENT
            self.sent_on = date
            handled = True
            WebHookEvent.trigger_sms_event(SMS_SENT, self, date)

        elif keyword == 'mt_dlvd':
            self.status = DELIVERED
            handled = True
            WebHookEvent.trigger_sms_event(SMS_DELIVERED, self, date)

        self.save()  # first save message status before updating the broadcast status

        # update our broadcast if we have one
        if self.broadcast:
            self.broadcast.update()

        return handled

    def handle(self):
        if self.direction == OUTGOING:
            raise ValueError(ugettext("Cannot process an outgoing message."))

        # process Android and test contact messages inline
        if not self.channel or self.channel.channel_type == Channel.TYPE_ANDROID or self.contact.is_test:
            Msg.process_message(self)

        # others do in celery
        else:
            on_transaction_commit(lambda: push_task(self.org, HANDLER_QUEUE, HANDLE_EVENT_TASK,
                                                    dict(type=MSG_EVENT, id=self.id, from_mage=False, new_contact=False),
                                                    priority=HIGH_PRIORITY))

    def build_expressions_context(self, contact_context=None):
        date_format = get_datetime_format(self.org.get_dayfirst())[1]

        return {
            '__default__': self.text,
            'value': self.text,
            'contact': contact_context or self.contact.build_expressions_context(),
            'time': datetime_to_str(self.created_on, format=date_format, tz=self.org.timezone)
        }

    def resend(self):
        """
        Resends this message by creating a clone and triggering a send of that clone
        """
        now = timezone.now()
        (topup_id, amount) = self.org.decrement_credit()  # costs 1 credit to resend message

        # see if we should use a new channel
        channel = self.org.get_send_channel(contact_urn=self.contact_urn)
{'fraction_non_alphanumeric': 0.08621954030213506, 'fraction_numerical': 0.0030381125987386593, 'mean_word_length': 2.173898586866168, 'pattern_counts': {'":': 0, '<': 11, '<?xml version=': 0, '>': 27, 'https://': 2, 'lorem ipsum': 0, 'www.': 2, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 176, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '6847082', 'n_tokens_mistral': 26641, 'n_tokens_neox': 25234, 'n_words': 7063}