mirror of
https://github.com/zulip/zulip.git
synced 2025-11-23 16:01:24 +00:00
ac14a8bcf5
This kind of payload that's loaded from json in the body of the request is not only used for webhooks, but also in the push bouncer, and may get used elsewhere too - so a general name is better.
128 lines
4.1 KiB
Python
128 lines
4.1 KiB
Python
# Webhooks for external integrations.
|
|
from typing import Callable, Dict, Optional
|
|
|
|
from django.http import HttpRequest, HttpResponse
|
|
from returns.curry import partial
|
|
|
|
from zerver.decorator import webhook_view
|
|
from zerver.lib.exceptions import UnsupportedWebhookEventTypeError
|
|
from zerver.lib.response import json_success
|
|
from zerver.lib.typed_endpoint import JsonBodyPayload, typed_endpoint
|
|
from zerver.lib.validator import WildValue, check_int, check_none_or, check_string, check_url
|
|
from zerver.lib.webhooks.common import (
|
|
check_send_webhook_message,
|
|
get_http_headers_from_filename,
|
|
validate_extract_webhook_http_header,
|
|
)
|
|
from zerver.models import UserProfile
|
|
|
|
TICKET_STARTED_TEMPLATE = """
|
|
{customer_name} submitted new ticket [#{number}: {title}]({app_url}):
|
|
|
|
``` quote
|
|
{summary}
|
|
```
|
|
""".strip()
|
|
|
|
TICKET_ASSIGNED_TEMPLATE = "[#{number}: {title}]({app_url}) ({state}) assigned to {assignee_info}."
|
|
|
|
AGENT_REPLIED_TEMPLATE = """
|
|
{actor} {action} [ticket #{number}]({app_ticket_url}):
|
|
|
|
``` quote
|
|
{plain_text_body}
|
|
```
|
|
""".strip()
|
|
|
|
|
|
def ticket_started_body(payload: WildValue) -> str:
|
|
return TICKET_STARTED_TEMPLATE.format(
|
|
customer_name=payload["customer_name"].tame(check_string),
|
|
number=payload["number"].tame(check_int),
|
|
title=payload["title"].tame(check_string),
|
|
app_url=payload["app_url"].tame(check_url),
|
|
summary=payload["summary"].tame(check_string),
|
|
)
|
|
|
|
|
|
def ticket_assigned_body(payload: WildValue) -> Optional[str]:
|
|
state = payload["state"].tame(check_string)
|
|
kwargs = {
|
|
"state": "open" if state == "opened" else state,
|
|
"number": payload["number"].tame(check_int),
|
|
"title": payload["title"].tame(check_string),
|
|
"app_url": payload["app_url"].tame(check_url),
|
|
}
|
|
|
|
assignee = payload["assignee"].tame(check_none_or(check_string))
|
|
assigned_group = payload["assigned_group"].tame(check_none_or(check_string))
|
|
|
|
if assignee or assigned_group:
|
|
if assignee and assigned_group:
|
|
kwargs["assignee_info"] = f"{assignee} from {assigned_group}"
|
|
elif assignee:
|
|
kwargs["assignee_info"] = f"{assignee}"
|
|
elif assigned_group:
|
|
kwargs["assignee_info"] = f"{assigned_group}"
|
|
|
|
return TICKET_ASSIGNED_TEMPLATE.format(**kwargs)
|
|
else:
|
|
return None
|
|
|
|
|
|
def replied_body(payload: WildValue, actor: str, action: str) -> str:
|
|
actor_url = "http://api.groovehq.com/v1/{}/".format(actor + "s")
|
|
actor = payload["links"]["author"]["href"].tame(check_url).split(actor_url)[1]
|
|
number = (
|
|
payload["links"]["ticket"]["href"]
|
|
.tame(check_url)
|
|
.split("http://api.groovehq.com/v1/tickets/")[1]
|
|
)
|
|
|
|
body = AGENT_REPLIED_TEMPLATE.format(
|
|
actor=actor,
|
|
action=action,
|
|
number=number,
|
|
app_ticket_url=payload["app_ticket_url"].tame(check_url),
|
|
plain_text_body=payload["plain_text_body"].tame(check_string),
|
|
)
|
|
|
|
return body
|
|
|
|
|
|
EVENTS_FUNCTION_MAPPER: Dict[str, Callable[[WildValue], Optional[str]]] = {
|
|
"ticket_started": ticket_started_body,
|
|
"ticket_assigned": ticket_assigned_body,
|
|
"agent_replied": partial(replied_body, actor="agent", action="replied to"),
|
|
"customer_replied": partial(replied_body, actor="customer", action="replied to"),
|
|
"note_added": partial(replied_body, actor="agent", action="left a note on"),
|
|
}
|
|
|
|
ALL_EVENT_TYPES = list(EVENTS_FUNCTION_MAPPER.keys())
|
|
|
|
|
|
@webhook_view("Groove", all_event_types=ALL_EVENT_TYPES)
|
|
@typed_endpoint
|
|
def api_groove_webhook(
|
|
request: HttpRequest,
|
|
user_profile: UserProfile,
|
|
*,
|
|
payload: JsonBodyPayload[WildValue],
|
|
) -> HttpResponse:
|
|
event = validate_extract_webhook_http_header(request, "X-Groove-Event", "Groove")
|
|
assert event is not None
|
|
handler = EVENTS_FUNCTION_MAPPER.get(event)
|
|
if handler is None:
|
|
raise UnsupportedWebhookEventTypeError(event)
|
|
|
|
body = handler(payload)
|
|
topic = "notifications"
|
|
|
|
if body is not None:
|
|
check_send_webhook_message(request, user_profile, topic, body, event)
|
|
|
|
return json_success(request)
|
|
|
|
|
|
fixture_to_headers = get_http_headers_from_filename("HTTP_X_GROOVE_EVENT")
|