mirror of https://github.com/zulip/zulip.git
544 lines
20 KiB
Python
544 lines
20 KiB
Python
from functools import partial
|
|
from typing import Any, Dict, Optional
|
|
|
|
from django.http import HttpRequest, HttpResponse
|
|
|
|
from zerver.decorator import api_key_only_webhook_view
|
|
from zerver.lib.request import REQ, has_request_variables
|
|
from zerver.lib.response import json_success
|
|
from zerver.lib.webhooks.common import UnexpectedWebhookEventType, check_send_webhook_message
|
|
from zerver.models import UserProfile
|
|
|
|
EPIC_NAME_TEMPLATE = "**{name}**"
|
|
STORY_NAME_TEMPLATE = "[{name}]({app_url})"
|
|
COMMENT_ADDED_TEMPLATE = "New comment added to the {entity} {name_template}:\n``` quote\n{text}\n```"
|
|
NEW_DESC_ADDED_TEMPLATE = "New description added to the {entity} {name_template}:\n``` quote\n{new}\n```"
|
|
DESC_CHANGED_TEMPLATE = ("Description for the {entity} {name_template} was changed from:\n"
|
|
"``` quote\n{old}\n```\nto\n``` quote\n{new}\n```")
|
|
DESC_REMOVED_TEMPLATE = "Description for the {entity} {name_template} was removed."
|
|
STATE_CHANGED_TEMPLATE = "State of the {entity} {name_template} was changed from **{old}** to **{new}**."
|
|
NAME_CHANGED_TEMPLATE = ("The name of the {entity} {name_template} was changed from:\n"
|
|
"``` quote\n{old}\n```\nto\n``` quote\n{new}\n```")
|
|
ARCHIVED_TEMPLATE = "The {entity} {name_template} was {action}."
|
|
STORY_TASK_TEMPLATE = "Task **{task_description}** was {action} the story {name_template}."
|
|
STORY_TASK_COMPLETED_TEMPLATE = "Task **{task_description}** ({name_template}) was completed. :tada:"
|
|
STORY_ADDED_REMOVED_EPIC_TEMPLATE = ("The story {story_name_template} was {action} the"
|
|
" epic {epic_name_template}.")
|
|
STORY_EPIC_CHANGED_TEMPLATE = ("The story {story_name_template} was moved from {old_epic_name_template}"
|
|
" to {new_epic_name_template}.")
|
|
STORY_ESTIMATE_TEMPLATE = "The estimate for the story {story_name_template} was set to {estimate}."
|
|
FILE_ATTACHMENT_TEMPLATE = "A {type} attachment `{file_name}` was added to the story {name_template}."
|
|
STORY_LABEL_TEMPLATE = "The label **{label_name}** was added to the story {name_template}."
|
|
STORY_UPDATE_PROJECT_TEMPLATE = ("The story {name_template} was moved from"
|
|
" the **{old}** project to **{new}**.")
|
|
STORY_UPDATE_TYPE_TEMPLATE = ("The type of the story {name_template} was changed"
|
|
" from **{old_type}** to **{new_type}**.")
|
|
DELETE_TEMPLATE = "The {entity_type} **{name}** was deleted."
|
|
STORY_UPDATE_OWNER_TEMPLATE = "New owner added to the story {name_template}."
|
|
STORY_GITHUB_PR_TEMPLATE = ("New GitHub PR [#{name}]({url}) opened for story"
|
|
" {name_template} ({old} -> {new}).")
|
|
STORY_GITHUB_BRANCH_TEMPLATE = ("New GitHub branch [{name}]({url})"
|
|
" associated with story {name_template} ({old} -> {new}).")
|
|
|
|
|
|
def get_action_with_primary_id(payload: Dict[str, Any]) -> Dict[str, Any]:
|
|
for action in payload["actions"]:
|
|
if payload["primary_id"] == action["id"]:
|
|
action_with_primary_id = action
|
|
|
|
return action_with_primary_id
|
|
|
|
def get_event(payload: Dict[str, Any]) -> Optional[str]:
|
|
action = get_action_with_primary_id(payload)
|
|
event = "{}_{}".format(action["entity_type"], action["action"])
|
|
|
|
if event in IGNORED_EVENTS:
|
|
return None
|
|
|
|
changes = action.get("changes")
|
|
if changes is not None:
|
|
if changes.get("description") is not None:
|
|
event = "{}_{}".format(event, "description")
|
|
elif changes.get("state") is not None:
|
|
event = "{}_{}".format(event, "state")
|
|
elif changes.get("workflow_state_id") is not None:
|
|
event = "{}_{}".format(event, "state")
|
|
elif changes.get("name") is not None:
|
|
event = "{}_{}".format(event, "name")
|
|
elif changes.get("archived") is not None:
|
|
event = "{}_{}".format(event, "archived")
|
|
elif changes.get("complete") is not None:
|
|
event = "{}_{}".format(event, "complete")
|
|
elif changes.get("epic_id") is not None:
|
|
event = "{}_{}".format(event, "epic")
|
|
elif changes.get("estimate") is not None:
|
|
event = "{}_{}".format(event, "estimate")
|
|
elif changes.get("file_ids") is not None:
|
|
event = "{}_{}".format(event, "attachment")
|
|
elif changes.get("label_ids") is not None:
|
|
event = "{}_{}".format(event, "label")
|
|
elif changes.get("project_id") is not None:
|
|
event = "{}_{}".format(event, "project")
|
|
elif changes.get("story_type") is not None:
|
|
event = "{}_{}".format(event, "type")
|
|
elif changes.get("owner_ids") is not None:
|
|
event = "{}_{}".format(event, "owner")
|
|
|
|
return event
|
|
|
|
def get_topic_function_based_on_type(payload: Dict[str, Any]) -> Any:
|
|
entity_type = get_action_with_primary_id(payload)["entity_type"]
|
|
return EVENT_TOPIC_FUNCTION_MAPPER.get(entity_type)
|
|
|
|
def get_delete_body(payload: Dict[str, Any]) -> str:
|
|
action = get_action_with_primary_id(payload)
|
|
return DELETE_TEMPLATE.format(**action)
|
|
|
|
def get_story_create_body(payload: Dict[str, Any]) -> str:
|
|
action = get_action_with_primary_id(payload)
|
|
|
|
if action.get("epic_id") is None:
|
|
message = "New story [{name}]({app_url}) of type **{story_type}** was created."
|
|
kwargs = action
|
|
else:
|
|
message = "New story [{name}]({app_url}) was created and added to the epic **{epic_name}**."
|
|
kwargs = {
|
|
"name": action["name"],
|
|
"app_url": action["app_url"],
|
|
}
|
|
epic_id = action["epic_id"]
|
|
refs = payload["references"]
|
|
for ref in refs:
|
|
if ref["id"] == epic_id:
|
|
kwargs["epic_name"] = ref["name"]
|
|
|
|
return message.format(**kwargs)
|
|
|
|
def get_epic_create_body(payload: Dict[str, Any]) -> str:
|
|
action = get_action_with_primary_id(payload)
|
|
message = "New epic **{name}**({state}) was created."
|
|
return message.format(**action)
|
|
|
|
def get_comment_added_body(payload: Dict[str, Any], entity: str) -> str:
|
|
actions = payload["actions"]
|
|
kwargs = {"entity": entity}
|
|
for action in actions:
|
|
if action["id"] == payload["primary_id"]:
|
|
kwargs["text"] = action["text"]
|
|
elif action["entity_type"] == entity:
|
|
name_template = get_name_template(entity).format(
|
|
name=action["name"],
|
|
app_url=action.get("app_url"),
|
|
)
|
|
kwargs["name_template"] = name_template
|
|
|
|
return COMMENT_ADDED_TEMPLATE.format(**kwargs)
|
|
|
|
def get_update_description_body(payload: Dict[str, Any], entity: str) -> str:
|
|
action = get_action_with_primary_id(payload)
|
|
desc = action["changes"]["description"]
|
|
|
|
kwargs = {
|
|
"entity": entity,
|
|
"new": desc["new"],
|
|
"old": desc["old"],
|
|
"name_template": get_name_template(entity).format(
|
|
name=action["name"],
|
|
app_url=action.get("app_url"),
|
|
),
|
|
}
|
|
|
|
if kwargs["new"] and kwargs["old"]:
|
|
body = DESC_CHANGED_TEMPLATE.format(**kwargs)
|
|
elif kwargs["new"]:
|
|
body = NEW_DESC_ADDED_TEMPLATE.format(**kwargs)
|
|
else:
|
|
body = DESC_REMOVED_TEMPLATE.format(**kwargs)
|
|
|
|
return body
|
|
|
|
def get_epic_update_state_body(payload: Dict[str, Any]) -> str:
|
|
action = get_action_with_primary_id(payload)
|
|
state = action["changes"]["state"]
|
|
kwargs = {
|
|
"entity": "epic",
|
|
"new": state["new"],
|
|
"old": state["old"],
|
|
"name_template": EPIC_NAME_TEMPLATE.format(name=action["name"]),
|
|
}
|
|
|
|
return STATE_CHANGED_TEMPLATE.format(**kwargs)
|
|
|
|
def get_story_update_state_body(payload: Dict[str, Any]) -> str:
|
|
action = get_action_with_primary_id(payload)
|
|
workflow_state_id = action["changes"]["workflow_state_id"]
|
|
references = payload["references"]
|
|
|
|
state = {}
|
|
for ref in references:
|
|
if ref["id"] == workflow_state_id["new"]:
|
|
state["new"] = ref["name"]
|
|
if ref["id"] == workflow_state_id["old"]:
|
|
state["old"] = ref["name"]
|
|
|
|
kwargs = {
|
|
"entity": "story",
|
|
"new": state["new"],
|
|
"old": state["old"],
|
|
"name_template": STORY_NAME_TEMPLATE.format(
|
|
name=action["name"],
|
|
app_url=action.get("app_url"),
|
|
),
|
|
}
|
|
|
|
return STATE_CHANGED_TEMPLATE.format(**kwargs)
|
|
|
|
def get_update_name_body(payload: Dict[str, Any], entity: str) -> str:
|
|
action = get_action_with_primary_id(payload)
|
|
name = action["changes"]["name"]
|
|
kwargs = {
|
|
"entity": entity,
|
|
"new": name["new"],
|
|
"old": name["old"],
|
|
"name_template": get_name_template(entity).format(
|
|
name=action["name"],
|
|
app_url=action.get("app_url"),
|
|
),
|
|
}
|
|
|
|
return NAME_CHANGED_TEMPLATE.format(**kwargs)
|
|
|
|
def get_update_archived_body(payload: Dict[str, Any], entity: str) -> str:
|
|
primary_action = get_action_with_primary_id(payload)
|
|
archived = primary_action["changes"]["archived"]
|
|
if archived["new"]:
|
|
action = "archived"
|
|
else:
|
|
action = "unarchived"
|
|
|
|
kwargs = {
|
|
"entity": entity,
|
|
"name_template": get_name_template(entity).format(
|
|
name=primary_action["name"],
|
|
app_url=primary_action.get("app_url"),
|
|
),
|
|
"action": action,
|
|
}
|
|
|
|
return ARCHIVED_TEMPLATE.format(**kwargs)
|
|
|
|
def get_story_task_body(payload: Dict[str, Any], action: str) -> str:
|
|
primary_action = get_action_with_primary_id(payload)
|
|
|
|
kwargs = {
|
|
"task_description": primary_action["description"],
|
|
"action": action,
|
|
}
|
|
|
|
for a in payload["actions"]:
|
|
if a["entity_type"] == "story":
|
|
kwargs["name_template"] = STORY_NAME_TEMPLATE.format(
|
|
name=a["name"],
|
|
app_url=a["app_url"],
|
|
)
|
|
|
|
return STORY_TASK_TEMPLATE.format(**kwargs)
|
|
|
|
def get_story_task_completed_body(payload: Dict[str, Any]) -> Optional[str]:
|
|
action = get_action_with_primary_id(payload)
|
|
|
|
kwargs = {
|
|
"task_description": action["description"],
|
|
}
|
|
|
|
story_id = action["story_id"]
|
|
for ref in payload["references"]:
|
|
if ref["id"] == story_id:
|
|
kwargs["name_template"] = STORY_NAME_TEMPLATE.format(
|
|
name=ref["name"],
|
|
app_url=ref["app_url"],
|
|
)
|
|
|
|
if action["changes"]["complete"]["new"]:
|
|
return STORY_TASK_COMPLETED_TEMPLATE.format(**kwargs)
|
|
else:
|
|
return None
|
|
|
|
def get_story_update_epic_body(payload: Dict[str, Any]) -> str:
|
|
action = get_action_with_primary_id(payload)
|
|
|
|
kwargs = {
|
|
"story_name_template": STORY_NAME_TEMPLATE.format(
|
|
name=action["name"],
|
|
app_url=action["app_url"],
|
|
),
|
|
}
|
|
|
|
new_id = action["changes"]["epic_id"].get("new")
|
|
old_id = action["changes"]["epic_id"].get("old")
|
|
|
|
for ref in payload["references"]:
|
|
if ref["id"] == new_id:
|
|
kwargs["new_epic_name_template"] = EPIC_NAME_TEMPLATE.format(
|
|
name=ref["name"])
|
|
|
|
if ref["id"] == old_id:
|
|
kwargs["old_epic_name_template"] = EPIC_NAME_TEMPLATE.format(
|
|
name=ref["name"])
|
|
|
|
if new_id and old_id:
|
|
return STORY_EPIC_CHANGED_TEMPLATE.format(**kwargs)
|
|
elif new_id:
|
|
kwargs["epic_name_template"] = kwargs["new_epic_name_template"]
|
|
kwargs["action"] = "added to"
|
|
else:
|
|
kwargs["epic_name_template"] = kwargs["old_epic_name_template"]
|
|
kwargs["action"] = "removed from"
|
|
|
|
return STORY_ADDED_REMOVED_EPIC_TEMPLATE.format(**kwargs)
|
|
|
|
def get_story_update_estimate_body(payload: Dict[str, Any]) -> str:
|
|
action = get_action_with_primary_id(payload)
|
|
|
|
kwargs = {
|
|
"story_name_template": STORY_NAME_TEMPLATE.format(
|
|
name=action["name"],
|
|
app_url=action["app_url"],
|
|
),
|
|
}
|
|
|
|
new = action["changes"]["estimate"].get("new")
|
|
if new:
|
|
kwargs["estimate"] = f"{new} points"
|
|
else:
|
|
kwargs["estimate"] = "*Unestimated*"
|
|
|
|
return STORY_ESTIMATE_TEMPLATE.format(**kwargs)
|
|
|
|
def get_reference_by_id(payload: Dict[str, Any], ref_id: int) -> Dict[str, Any]:
|
|
ref: Dict[str, Any] = {}
|
|
for reference in payload['references']:
|
|
if reference['id'] == ref_id:
|
|
ref = reference
|
|
|
|
return ref
|
|
|
|
def get_story_create_github_entity_body(payload: Dict[str, Any],
|
|
entity: str) -> str:
|
|
action = get_action_with_primary_id(payload)
|
|
|
|
story: Dict[str, Any] = {}
|
|
for a in payload['actions']:
|
|
if (a['entity_type'] == 'story' and
|
|
a['changes'].get('workflow_state_id') is not None):
|
|
story = a
|
|
|
|
new_state_id = story['changes']['workflow_state_id']['new']
|
|
old_state_id = story['changes']['workflow_state_id']['old']
|
|
new_state = get_reference_by_id(payload, new_state_id)['name']
|
|
old_state = get_reference_by_id(payload, old_state_id)['name']
|
|
|
|
kwargs = {
|
|
'name_template': STORY_NAME_TEMPLATE.format(**story),
|
|
'name': action.get('number') if entity == 'pull-request' else action.get('name'),
|
|
'url': action['url'],
|
|
'new': new_state,
|
|
'old': old_state,
|
|
}
|
|
|
|
template = STORY_GITHUB_PR_TEMPLATE if entity == 'pull-request' else STORY_GITHUB_BRANCH_TEMPLATE
|
|
return template.format(**kwargs)
|
|
|
|
def get_story_update_attachment_body(payload: Dict[str, Any]) -> Optional[str]:
|
|
action = get_action_with_primary_id(payload)
|
|
|
|
kwargs = {
|
|
"name_template": STORY_NAME_TEMPLATE.format(
|
|
name=action["name"],
|
|
app_url=action["app_url"],
|
|
),
|
|
}
|
|
file_ids_added = action["changes"]["file_ids"].get("adds")
|
|
|
|
# If this is a payload for when an attachment is removed, ignore it
|
|
if not file_ids_added:
|
|
return None
|
|
|
|
file_id = file_ids_added[0]
|
|
for ref in payload["references"]:
|
|
if ref["id"] == file_id:
|
|
kwargs.update({
|
|
"type": ref["entity_type"],
|
|
"file_name": ref["name"],
|
|
})
|
|
|
|
return FILE_ATTACHMENT_TEMPLATE.format(**kwargs)
|
|
|
|
def get_story_label_body(payload: Dict[str, Any]) -> Optional[str]:
|
|
action = get_action_with_primary_id(payload)
|
|
|
|
kwargs = {
|
|
"name_template": STORY_NAME_TEMPLATE.format(
|
|
name=action["name"],
|
|
app_url=action["app_url"],
|
|
),
|
|
}
|
|
label_ids_added = action["changes"]["label_ids"].get("adds")
|
|
|
|
# If this is a payload for when a label is removed, ignore it
|
|
if not label_ids_added:
|
|
return None
|
|
|
|
label_id = label_ids_added[0]
|
|
|
|
label_name = ''
|
|
for action in payload["actions"]:
|
|
if action['id'] == label_id:
|
|
label_name = action.get('name', '')
|
|
|
|
if not label_name:
|
|
for reference in payload["references"]:
|
|
if reference["id"] == label_id:
|
|
label_name = reference.get('name', '')
|
|
|
|
kwargs.update({"label_name": label_name})
|
|
|
|
return STORY_LABEL_TEMPLATE.format(**kwargs)
|
|
|
|
def get_story_update_project_body(payload: Dict[str, Any]) -> str:
|
|
action = get_action_with_primary_id(payload)
|
|
kwargs = {
|
|
"name_template": STORY_NAME_TEMPLATE.format(
|
|
name=action["name"],
|
|
app_url=action["app_url"],
|
|
),
|
|
}
|
|
|
|
new_project_id = action["changes"]["project_id"]["new"]
|
|
old_project_id = action["changes"]["project_id"]["old"]
|
|
for ref in payload["references"]:
|
|
if ref["id"] == new_project_id:
|
|
kwargs.update({"new": ref["name"]})
|
|
if ref["id"] == old_project_id:
|
|
kwargs.update({"old": ref["name"]})
|
|
|
|
return STORY_UPDATE_PROJECT_TEMPLATE.format(**kwargs)
|
|
|
|
def get_story_update_type_body(payload: Dict[str, Any]) -> str:
|
|
action = get_action_with_primary_id(payload)
|
|
kwargs = {
|
|
"name_template": STORY_NAME_TEMPLATE.format(
|
|
name=action["name"],
|
|
app_url=action["app_url"],
|
|
),
|
|
"new_type": action["changes"]["story_type"]["new"],
|
|
"old_type": action["changes"]["story_type"]["old"],
|
|
}
|
|
|
|
return STORY_UPDATE_TYPE_TEMPLATE.format(**kwargs)
|
|
|
|
def get_story_update_owner_body(payload: Dict[str, Any]) -> str:
|
|
action = get_action_with_primary_id(payload)
|
|
kwargs = {
|
|
"name_template": STORY_NAME_TEMPLATE.format(
|
|
name=action["name"],
|
|
app_url=action["app_url"],
|
|
),
|
|
}
|
|
|
|
return STORY_UPDATE_OWNER_TEMPLATE.format(**kwargs)
|
|
|
|
def get_entity_name(payload: Dict[str, Any], entity: Optional[str]=None) -> Optional[str]:
|
|
action = get_action_with_primary_id(payload)
|
|
name = action.get("name")
|
|
|
|
if name is None or action['entity_type'] == 'branch':
|
|
for action in payload["actions"]:
|
|
if action["entity_type"] == entity:
|
|
name = action["name"]
|
|
|
|
if name is None:
|
|
for ref in payload["references"]:
|
|
if ref["entity_type"] == entity:
|
|
name = ref["name"]
|
|
|
|
return name
|
|
|
|
def get_name_template(entity: str) -> str:
|
|
if entity == "story":
|
|
return STORY_NAME_TEMPLATE
|
|
return EPIC_NAME_TEMPLATE
|
|
|
|
EVENT_BODY_FUNCTION_MAPPER = {
|
|
"story_update_archived": partial(get_update_archived_body, entity='story'),
|
|
"epic_update_archived": partial(get_update_archived_body, entity='epic'),
|
|
"story_create": get_story_create_body,
|
|
"pull-request_create": partial(get_story_create_github_entity_body, entity='pull-request'),
|
|
"branch_create": partial(get_story_create_github_entity_body, entity='branch'),
|
|
"story_delete": get_delete_body,
|
|
"epic_delete": get_delete_body,
|
|
"story-task_create": partial(get_story_task_body, action="added to"),
|
|
"story-task_delete": partial(get_story_task_body, action="removed from"),
|
|
"story-task_update_complete": get_story_task_completed_body,
|
|
"story_update_epic": get_story_update_epic_body,
|
|
"story_update_estimate": get_story_update_estimate_body,
|
|
"story_update_attachment": get_story_update_attachment_body,
|
|
"story_update_label": get_story_label_body,
|
|
"story_update_owner": get_story_update_owner_body,
|
|
"story_update_project": get_story_update_project_body,
|
|
"story_update_type": get_story_update_type_body,
|
|
"epic_create": get_epic_create_body,
|
|
"epic-comment_create": partial(get_comment_added_body, entity='epic'),
|
|
"story-comment_create": partial(get_comment_added_body, entity='story'),
|
|
"epic_update_description": partial(get_update_description_body, entity='epic'),
|
|
"story_update_description": partial(get_update_description_body, entity='story'),
|
|
"epic_update_state": get_epic_update_state_body,
|
|
"story_update_state": get_story_update_state_body,
|
|
"epic_update_name": partial(get_update_name_body, entity='epic'),
|
|
"story_update_name": partial(get_update_name_body, entity='story'),
|
|
}
|
|
|
|
EVENT_TOPIC_FUNCTION_MAPPER = {
|
|
"story": partial(get_entity_name, entity='story'),
|
|
"pull-request": partial(get_entity_name, entity='story'),
|
|
"branch": partial(get_entity_name, entity='story'),
|
|
"story-comment": partial(get_entity_name, entity='story'),
|
|
"story-task": partial(get_entity_name, entity='story'),
|
|
"epic": partial(get_entity_name, entity='epic'),
|
|
"epic-comment": partial(get_entity_name, entity='epic'),
|
|
}
|
|
|
|
IGNORED_EVENTS = {
|
|
'story-comment_update',
|
|
}
|
|
|
|
@api_key_only_webhook_view('ClubHouse')
|
|
@has_request_variables
|
|
def api_clubhouse_webhook(
|
|
request: HttpRequest, user_profile: UserProfile,
|
|
payload: Optional[Dict[str, Any]]=REQ(argument_type='body'),
|
|
) -> HttpResponse:
|
|
|
|
# Clubhouse has a tendency to send empty POST requests to
|
|
# third-party endpoints. It is unclear as to which event type
|
|
# such requests correspond to. So, it is best to ignore such
|
|
# requests for now.
|
|
if payload is None:
|
|
return json_success()
|
|
|
|
event = get_event(payload)
|
|
if event is None:
|
|
return json_success()
|
|
|
|
body_func: Any = EVENT_BODY_FUNCTION_MAPPER.get(event)
|
|
topic_func = get_topic_function_based_on_type(payload)
|
|
if body_func is None or topic_func is None:
|
|
raise UnexpectedWebhookEventType('Clubhouse', event)
|
|
topic = topic_func(payload)
|
|
body = body_func(payload)
|
|
|
|
if topic and body:
|
|
check_send_webhook_message(request, user_profile, topic, body)
|
|
|
|
return json_success()
|