"""
SMSES
--------
"""
from . import filters
from .es_query import HQESQuery
[docs]class SMSES(HQESQuery):
index = 'sms'
@property
def builtin_filters(self):
return [
incoming_messages,
outgoing_messages,
to_commcare_user,
to_commcare_case,
to_web_user,
to_couch_user,
to_commcare_user_or_case,
received,
direction,
processed,
processed_or_incoming_messages,
] + super(SMSES, self).builtin_filters
[docs] def user_aggregation(self):
return self.terms_aggregation('couch_recipient', 'user')
[docs]def incoming_messages():
return direction("i")
[docs]def outgoing_messages():
return direction("o")
[docs]def direction(direction_):
return filters.term("direction", direction_)
[docs]def processed_or_incoming_messages():
return filters.NOT(filters.AND(outgoing_messages(), processed(False)))
[docs]def processed(processed=True):
return filters.term('processed', processed)
[docs]def to_commcare_user():
return filters.term("couch_recipient_doc_type", "commcareuser")
[docs]def to_commcare_case():
return filters.term("couch_recipient_doc_type", "commcarecase")
[docs]def to_web_user():
return filters.term("couch_recipient_doc_type", "webuser")
[docs]def to_couch_user():
return filters.term("couch_recipient_doc_type", "couchuser")
[docs]def to_commcare_user_or_case():
return filters.OR(to_commcare_user(), to_commcare_case())
[docs]def received(gt=None, gte=None, lt=None, lte=None):
return filters.date_range('date', gt, gte, lt, lte)