"""Webhooks for external integrations.""" import re from typing import Any import orjson from defusedxml.ElementTree import fromstring as xml_fromstring from django.http import HttpRequest, HttpResponse from django.utils.translation import gettext as _ from zerver.decorator import webhook_view from zerver.lib.exceptions import JsonableError, UnsupportedWebhookEventTypeError from zerver.lib.response import json_success from zerver.lib.typed_endpoint import typed_endpoint_without_parameters from zerver.lib.utils import assert_is_not_none from zerver.lib.webhooks.common import check_send_webhook_message from zerver.models import UserProfile def api_pivotal_webhook_v3(request: HttpRequest, user_profile: UserProfile) -> tuple[str, str, str]: payload = xml_fromstring(request.body) def get_text(attrs: list[str]) -> str: start = payload for attr in attrs: child = start.find(attr) if child is None: return "" start = child assert start.text is not None return start.text event_type = assert_is_not_none(payload.find("event_type")).text description = assert_is_not_none(payload.find("description")).text assert description is not None project_id = assert_is_not_none(payload.find("project_id")).text story_id = get_text(["stories", "story", "id"]) # Ugh, the URL in the XML data is not a clickable URL that works for the user # so we try to build one that the user can actually click on url = f"https://www.pivotaltracker.com/s/projects/{project_id}/stories/{story_id}" # Pivotal doesn't tell us the name of the story, but it's usually in the # description in quotes as the first quoted string name_re = re.compile(r'[^"]+"([^"]+)".*') match = name_re.match(description) if match and len(match.groups()): name = match.group(1) else: name = "Story changed" # Failed for an unknown reason, show something more_info = f" [(view)]({url})." if event_type == "story_update": topic_name = name content = description + more_info elif event_type == "note_create": topic_name = "Comment added" content = description + more_info elif event_type == "story_create": issue_desc = get_text(["stories", "story", "description"]) issue_type = get_text(["stories", "story", "story_type"]) issue_status = get_text(["stories", "story", "current_state"]) estimate = get_text(["stories", "story", "estimate"]) if estimate != "": estimate = f" worth {estimate} story points" topic_name = name content = f"{description} ({issue_status} {issue_type}{estimate}):\n\n~~~ quote\n{issue_desc}\n~~~\n\n{more_info}" return topic_name, content, f"{event_type}_v3" UNSUPPORTED_EVENT_TYPES = [ "task_create_activity", "comment_delete_activity", "task_delete_activity", "task_update_activity", "story_move_from_project_activity", "story_delete_activity", "story_move_into_project_activity", "epic_update_activity", "label_create_activity", ] ALL_EVENT_TYPES = [ "story_update_v3", "note_create_v3", "story_create_v3", "story_move_activity_v5", "story_create_activity_v5", "story_update_activity_v5", "comment_create_activity_v5", ] def api_pivotal_webhook_v5(request: HttpRequest, user_profile: UserProfile) -> tuple[str, str, str]: payload = orjson.loads(request.body) event_type = payload["kind"] project_name = payload["project"]["name"] project_id = payload["project"]["id"] primary_resources = payload["primary_resources"][0] story_url = primary_resources["url"] story_type = primary_resources.get("story_type") story_id = primary_resources["id"] story_name = primary_resources["name"] performed_by = payload.get("performed_by", {}).get("name", "") story_info = f"[{project_name}](https://www.pivotaltracker.com/s/projects/{project_id}): [{story_name}]({story_url})" changes = payload.get("changes", []) content = "" topic_name = f"#{story_id}: {story_name}" def extract_comment(change: dict[str, Any]) -> str | None: if change.get("kind") == "comment": return change.get("new_values", {}).get("text", None) return None if event_type == "story_update_activity": # Find the changed valued and build a message content += f"{performed_by} updated {story_info}:\n" for change in changes: old_values = change.get("original_values", {}) new_values = change["new_values"] if "current_state" in old_values and "current_state" in new_values: content += "* state changed from **{}** to **{}**\n".format( old_values["current_state"], new_values["current_state"] ) if "estimate" in old_values and "estimate" in new_values: old_estimate = old_values.get("estimate", None) if old_estimate is None: estimate = "is now" else: estimate = f"changed from {old_estimate} to" new_estimate = new_values["estimate"] if new_values["estimate"] is not None else "0" content += f"* estimate {estimate} **{new_estimate} points**\n" if "story_type" in old_values and "story_type" in new_values: content += "* type changed from **{}** to **{}**\n".format( old_values["story_type"], new_values["story_type"] ) comment = extract_comment(change) if comment is not None: content += f"* Comment added:\n~~~quote\n{comment}\n~~~\n" elif event_type == "comment_create_activity": for change in changes: comment = extract_comment(change) if comment is not None: content += ( f"{performed_by} added a comment to {story_info}:\n~~~quote\n{comment}\n~~~" ) elif event_type == "story_create_activity": content += f"{performed_by} created {story_type}: {story_info}\n" for change in changes: new_values = change.get("new_values", {}) if "current_state" in new_values: content += "* State is **{}**\n".format(new_values["current_state"]) if "description" in new_values: content += "* Description is\n\n> {}".format(new_values["description"]) elif event_type == "story_move_activity": content = f"{performed_by} moved {story_info}" for change in changes: old_values = change.get("original_values", {}) new_values = change["new_values"] if "current_state" in old_values and "current_state" in new_values: content += " from **{}** to **{}**.".format( old_values["current_state"], new_values["current_state"] ) elif event_type in UNSUPPORTED_EVENT_TYPES: # Known but unsupported Pivotal event types pass else: raise UnsupportedWebhookEventTypeError(event_type) return topic_name, content, f"{event_type}_v5" @webhook_view("Pivotal", all_event_types=ALL_EVENT_TYPES) @typed_endpoint_without_parameters def api_pivotal_webhook(request: HttpRequest, user_profile: UserProfile) -> HttpResponse: topic_name = content = None try: topic_name, content, event_type = api_pivotal_webhook_v3(request, user_profile) except Exception: # Attempt to parse v5 JSON payload topic_name, content, event_type = api_pivotal_webhook_v5(request, user_profile) if not content: raise JsonableError(_("Unable to handle Pivotal payload")) check_send_webhook_message(request, user_profile, topic_name, content, event_type) return json_success(request)