Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
lock_messages.append(message.backend_id)
if not lock_messages:
for message in messages:
message.user_lock(user)
elif action == "unlock":
for message in messages:
message.user_unlock()
else: # pragma: no cover
return HttpResponseBadRequest("Invalid action: %s", action)
return JsonResponse({"messages": lock_messages}, encoder=JSONEncoder)
class Action(OrgPermsMixin, SmartTemplateView):
"""
AJAX endpoint for bulk message actions. Takes a list of message ids.
"""
@classmethod
def derive_url_pattern(cls, path, action):
return r"^message/action/(?P\w+)/$"
def post(self, request, *args, **kwargs):
org = request.org
user = request.user
action = kwargs["action"]
message_ids = request.json["messages"]
messages = org.incoming_messages.filter(org=org, backend_id__in=message_ids)
class Fetch(OrgObjPermsMixin, SmartReadView):
"""
JSON endpoint for fetching a single case
"""
permission = "cases.case_read"
def render_to_response(self, context, **response_kwargs):
case_json = self.object.as_json()
# augment usual case JSON
case_json["watching"] = self.object.is_watched_by(self.request.user)
return JsonResponse(case_json, encoder=JSONEncoder)
class Search(OrgPermsMixin, CaseSearchMixin, SmartTemplateView):
"""
JSON endpoint for searching for cases
"""
permission = "cases.case_list"
def get_context_data(self, **kwargs):
context = super(CaseCRUDL.Search, self).get_context_data(**kwargs)
org = self.request.org
user = self.request.user
page = int(self.request.GET.get("page", 1))
search = self.derive_search()
cases = Case.search(org, user, search)
paginator = LazyPaginator(cases, 50)
"cases": {
"average_closed_this_month": humanize_seconds(
average_closed_this_month.get(partner, 0)
),
"opened_this_month": cases_opened_this_month.get(partner, 0),
"closed_this_month": cases_closed_this_month.get(partner, 0),
"total": cases_total.get(partner, 0),
},
}
)
return obj
return JsonResponse({"results": [as_json(p) for p in partners]})
class BaseInboxView(OrgPermsMixin, SmartTemplateView):
"""
Mixin to add site metadata to the context in JSON format which can then used
"""
title = None
folder = None
folder_icon = None
template_name = None
permission = "orgs.org_inbox"
def get_context_data(self, **kwargs):
context = super(BaseInboxView, self).get_context_data(**kwargs)
org = self.request.org
user = self.request.user
partner = user.get_partner(org)
context["has_more"] = paginator.num_pages > page
return context
def render_to_response(self, context, **response_kwargs):
results = []
for m in context["object_list"]:
msg = m.as_json()
msg["lock"] = m.get_lock(self.request.user)
results.append(msg)
return JsonResponse({"results": results, "has_more": context["has_more"]}, encoder=JSONEncoder)
class Lock(OrgPermsMixin, SmartTemplateView):
"""
AJAX endpoint for updating messages with a date and user id.
Takes a list of message ids.
"""
@classmethod
def derive_url_pattern(cls, path, action):
return r"^message/lock/(?P\w+)/$"
def post(self, request, *args, **kwargs):
org = request.org
user = request.user
action = kwargs["action"]
message_ids = request.json["messages"]
from __future__ import unicode_literals
from smartmin.views import SmartTemplateView
class PartialTemplate(SmartTemplateView):
"""
Simple view for fetching partial templates for Angular
"""
def pre_process(self, request, *args, **kwargs):
self.template = kwargs['template']
return
def get_template_names(self):
return "partials/%s.haml" % self.template
elif action == "unflag":
Message.bulk_unflag(org, user, messages)
elif action == "label":
Message.bulk_label(org, user, messages, label)
elif action == "unlabel":
Message.bulk_unlabel(org, user, messages, label)
elif action == "archive":
Message.bulk_archive(org, user, messages)
elif action == "restore":
Message.bulk_restore(org, user, messages)
else: # pragma: no cover
return HttpResponseBadRequest("Invalid action: %s", action)
return HttpResponse(status=204)
class Label(OrgPermsMixin, SmartTemplateView):
"""
AJAX endpoint for labelling a message.
"""
@classmethod
def derive_url_pattern(cls, path, action):
return r"^message/label/(?P\d+)/$"
def post(self, request, *args, **kwargs):
org = request.org
user = request.user
user_labels = Label.get_all(self.org, user)
message_id = int(kwargs["id"])
message = org.incoming_messages.filter(org=org, backend_id=message_id).first()
return {
"folder": folder,
"label": label_id,
"text": text,
"contact": contact_id,
"after": after,
"before": before,
}
class MessageCRUDL(SmartCRUDL):
actions = ("search", "lock", "action", "label", "bulk_reply", "forward", "history")
model = Message
class Search(OrgPermsMixin, MessageSearchMixin, SmartTemplateView):
"""
JSON endpoint for fetching incoming messages
"""
page_size = 50
def get_messages(self, search, last_refresh=None):
org = self.request.org
user = self.request.user
queryset = Message.search(org, user, search, modified_after=last_refresh, all=False)
return queryset.prefetch_related("contact", "labels", "case__assignee", "case__user_assignee")
def get_context_data(self, **kwargs):
context = super(MessageCRUDL.Search, self).get_context_data(**kwargs)
page = int(self.request.GET.get("page", 1))
user_labels = Label.get_all(self.org, user)
message_id = int(kwargs["id"])
message = org.incoming_messages.filter(org=org, backend_id=message_id).first()
label_ids = request.json["labels"]
specified_labels = list(user_labels.filter(pk__in=label_ids))
# user can't remove labels that they can't see
unseen_labels = [l for l in message.labels.all() if l not in user_labels]
message.update_labels(user, specified_labels + unseen_labels)
return HttpResponse(status=204)
class BulkReply(OrgPermsMixin, SmartTemplateView):
"""
JSON endpoint for bulk messages replies
"""
@classmethod
def derive_url_pattern(cls, path, action):
return r"^message/bulk_reply/$"
def post(self, request, *args, **kwargs):
text = request.json["text"]
message_ids = request.json["messages"]
messages = Message.objects.filter(org=request.org, backend_id__in=message_ids).select_related("contact")
# organize messages by contact
messages_by_contact = defaultdict(list)
for msg in messages:
"""
Collects and prepares reply search parameters into JSON serializable dict
"""
params = self.request.GET
partner = params.get("partner")
after = parse_iso8601(params.get("after"))
before = parse_iso8601(params.get("before"))
return {"partner": partner, "after": after, "before": before}
class OutgoingCRUDL(SmartCRUDL):
actions = ("search", "search_replies")
model = Outgoing
class Search(OrgPermsMixin, SmartTemplateView):
"""
JSON endpoint for fetching outgoing messages
"""
def derive_search(self):
folder = OutgoingFolder[self.request.GET["folder"]]
text = self.request.GET.get("text", None)
contact = self.request.GET.get("contact", None)
return {"folder": folder, "text": text, "contact": contact}
def get_context_data(self, **kwargs):
context = super(OutgoingCRUDL.Search, self).get_context_data(**kwargs)
org = self.request.org
user = self.request.user
# organize messages by contact
messages_by_contact = defaultdict(list)
for msg in messages:
messages_by_contact[msg.contact].append(msg)
# the actual message that will be replied to is the oldest selected message for each contact
reply_tos = []
for contact, contact_messages in messages_by_contact.items():
contact_messages = sorted(contact_messages, key=lambda m: m.created_on, reverse=True)
reply_tos.append(contact_messages[0])
outgoing = Outgoing.create_bulk_replies(request.org, request.user, text, reply_tos)
return JsonResponse({"messages": len(outgoing)})
class Forward(OrgPermsMixin, SmartTemplateView):
"""
JSON endpoint for forwarding a message to a URN
"""
@classmethod
def derive_url_pattern(cls, path, action):
return r"^message/forward/(?P\d+)/$"
def post(self, request, *args, **kwargs):
text = request.json["text"]
message = Message.objects.get(org=request.org, backend_id=int(kwargs["id"]))
urns = request.json["urns"]
outgoing = Outgoing.create_forwards(request.org, request.user, text, urns, message)
return JsonResponse({"messages": len(outgoing)})