zerver/webhooks/github/view.py - main - githubmirror/zulip

zerver/webhooks/github/view.py
1 351 строка52 KB
Aman Agrawal
github: Handle ping events from GitHub Sponsors webhooks.
19 май 2026, 11:54
19 май 2026, 11:54573cd1c
import re
from collections.abc import Callable
from datetime import datetime

from django.http import HttpRequest, HttpResponse
from pydantic import Json
from typing_extensions import override

from zerver.decorator import log_unsupported_webhook_event, webhook_view
from zerver.lib.exceptions import UnsupportedWebhookEventTypeError
from zerver.lib.external_accounts import DEFAULT_EXTERNAL_ACCOUNTS
from zerver.lib.markdown.fenced_code import get_unused_fence
from zerver.lib.mention import silent_mention_syntax_for_user
from zerver.lib.partial import partial
from zerver.lib.response import json_success
from zerver.lib.typed_endpoint import JsonBodyPayload, typed_endpoint
from zerver.lib.validator import WildValue, check_bool, check_int, check_none_or, check_string
from zerver.lib.webhooks.common import (
    OptionalUserSpecifiedTopicStr,
    check_send_webhook_message,
    default_fixture_to_headers,
    get_event_header,
    get_setup_webhook_message,
    guess_zulip_user_from_external_account,
)
from zerver.lib.webhooks.git import (
    CONTENT_MESSAGE_TEMPLATE,
    TOPIC_WITH_BRANCH_TEMPLATE,
    TOPIC_WITH_PR_OR_ISSUE_INFO_TEMPLATE,
    get_commits_comment_action_message,
    get_issue_event_message,
    get_issue_labeled_or_unlabeled_event_message,
    get_issue_milestoned_or_demilestoned_event_message,
    get_pull_request_event_message,
    get_push_commits_event_message,
    get_push_tag_event_message,
    get_release_event_message,
    get_short_sha,
    is_branch_name_notifiable,
)
from zerver.models import UserProfile

fixture_to_headers = default_fixture_to_headers("HTTP_X_GITHUB_EVENT")

TOPIC_FOR_DISCUSSION = "{repo} discussion #{number}: {title}"
DISCUSSION_TEMPLATES = {
    "created": "{sender} created [discussion #{discussion_number}]({url}) in {category}:\n\n{body_fence} quote\n### {title}\n{body}\n{body_fence}",
    "generic_action": "{sender} {action} [discussion #{discussion_number}{configured_title}]({url}).",
    "deleted": "{sender} {action} discussion #{discussion_number}{configured_title}.",
    "closed": "{sender} {action} [discussion #{discussion_number}{configured_title}]({url}) as {closed_reason}.",
    "locked": "{sender} {action} [discussion #{discussion_number}{configured_title}]({url}){locked_reason}.",
    "labeled": "{sender} added the {label} label to [discussion #{discussion_number}{configured_title}]({url}).",
    "unlabeled": "{sender} removed the {label} label from [discussion #{discussion_number}{configured_title}]({url}).",
    "category_changed": "{sender} changed the category of [discussion #{discussion_number}{configured_title}]({url}) from {old_category} to {category}.",
    "transferred": "{sender} {action} discussion #{discussion_number}{configured_title} from {repository_name} to {new_repository_name} as [discussion #{new_discussion_number}]({url}).",
    "answered": "{sender} marked [comment #{comment_id}]({answer_url}) as the answer:\n\n{answer_body_fence} quote\n{answer_body}\n{answer_body_fence}",
    "unanswered": "{sender} marked [comment #{comment_id}]({answer_url}) as not the answer.",
    "edited_title": "{sender} edited the title of [discussion #{discussion_number}{configured_title}]({url}):\n\n``` quote\n### {title}\n```",
    "edited_body": "{sender} edited [discussion #{discussion_number}{configured_title}]({url}):\n\n{body_fence} quote\n{body}\n{body_fence}",
}

CHECK_RUN_CONCLUSION_EMOJI = {
    "success": ":check:",
    "failure": ":warning:",
    "cancelled": ":not_allowed:",
    "skipped": ":fast_forward:",
    "timed_out": ":times_up:",
    "action_required": ":wrench:",
    "neutral": ":minus:",
    "stale": ":sleeping:",
}

PR_REVIEW_STATE_EMOJI = {
    "approved": ":thumbs_up:",
    "changes_requested": ":repeat:",
    "commented": ":memo:",
}

PR_REVIEW_COMMENT_EMOJI = ":speech_balloon:"

PR_CLOSE_ACTION_EMOJI = {
    "merged": ":check:",
    "closed without merge": ":cross_mark:",
}

DEPLOYMENT_AND_COMMIT_STATUS_EMOJI = {
    "success": ":check:",
    "failure": ":warning:",
    "error": ":rotating_light:",
    "pending": ":time_ticking:",
}


class Helper:
    def __init__(
        self,
        request: HttpRequest,
        payload: WildValue,
        include_title: bool,
        include_repository_name: bool,
        include_emoji_indicators: bool,
        user_profile: UserProfile,
    ) -> None:
        self.request = request
        self.payload = payload
        self.include_title = include_title
        self.include_repository_name = include_repository_name
        self.include_emoji_indicators = include_emoji_indicators
        self.realm = user_profile.realm

    def log_unsupported(self, event: str) -> None:
        summary = f"The '{event}' event isn't currently supported by the GitHub webhook; ignoring"
        log_unsupported_webhook_event(request=self.request, summary=summary)


def get_opened_or_update_pull_request_body(helper: Helper) -> str:
    payload = helper.payload
    include_title = helper.include_title
    pull_request = payload["pull_request"]
    action = payload["action"].tame(check_string)
    if action == "synchronize":
        action = "updated"
    assignee = None
    if pull_request.get("assignee"):
        assignee = pull_request["assignee"]["login"].tame(check_string)
    description = None
    changes = payload.get("changes", {})
    if "body" in changes or action == "opened":
        description = pull_request["body"].tame(check_none_or(check_string))
    target_branch = None
    base_branch = None
    if action in ("opened", "merged", "reopened"):
        target_branch = pull_request["head"]["label"].tame(check_string)
        base_branch = pull_request["base"]["label"].tame(check_string)

    return get_pull_request_event_message(
        user_name=get_sender_name(helper),
        action=action,
        url=pull_request["html_url"].tame(check_string),
        target_branch=target_branch,
        base_branch=base_branch,
        message=description,
        assignee=assignee,
        number=pull_request["number"].tame(check_int),
        title=pull_request["title"].tame(check_string) if include_title else None,
    )


def get_assigned_or_unassigned_pull_request_body(helper: Helper) -> str:
    payload = helper.payload
    include_title = helper.include_title
    pull_request = payload["pull_request"]
    assignee = payload["assignee"]["login"].tame(check_string)

    return get_pull_request_event_message(
        user_name=get_sender_name(helper),
        action=payload["action"].tame(check_string),
        url=pull_request["html_url"].tame(check_string),
        number=pull_request["number"].tame(check_int),
        title=pull_request["title"].tame(check_string) if include_title else None,
        assignee_updated=assignee,
    )


def get_closed_pull_request_body(helper: Helper) -> str:
    payload = helper.payload
    include_title = helper.include_title
    pull_request = payload["pull_request"]
    action = "merged" if pull_request["merged"].tame(check_bool) else "closed without merge"
    return get_pull_request_event_message(
        user_name=get_sender_name(helper),
        action=action,
        url=pull_request["html_url"].tame(check_string),
        number=pull_request["number"].tame(check_int),
        title=pull_request["title"].tame(check_string) if include_title else None,
        emoji=PR_CLOSE_ACTION_EMOJI.get(action) if helper.include_emoji_indicators else None,
    )


def get_membership_body(helper: Helper) -> str:
    payload = helper.payload
    action = payload["action"].tame(check_string)
    member = payload["member"]
    team_name = payload["team"]["name"].tame(check_string)

    return "{sender} {action} [{username}]({html_url}) {preposition} the {team_name} team.".format(
        sender=get_sender_name(helper),
        action=action,
        username=member["login"].tame(check_string),
        html_url=member["html_url"].tame(check_string),
        preposition="from" if action == "removed" else "to",
        team_name=team_name,
    )


def get_member_body(helper: Helper) -> str:
    payload = helper.payload
    return "{} {} [{}]({}) to [{}]({}).".format(
        get_sender_name(helper),
        payload["action"].tame(check_string),
        payload["member"]["login"].tame(check_string),
        payload["member"]["html_url"].tame(check_string),
        get_repository_name(payload),
        payload["repository"]["html_url"].tame(check_string),
    )


def get_issue_body(helper: Helper) -> str:
    payload = helper.payload
    include_title = helper.include_title
    action = payload["action"].tame(check_string)
    issue = payload["issue"]
    return get_issue_event_message(
        user_name=get_sender_name(helper),
        action=action,
        url=issue["html_url"].tame(check_string),
        number=issue["number"].tame(check_int),
        message=(
            None
            if action in ("assigned", "unassigned")
            else issue["body"].tame(check_none_or(check_string))
        ),
        title=issue["title"].tame(check_string) if include_title else None,
        assignee_updated=(
            payload["assignee"]["login"].tame(check_string) if "assignee" in payload else None
        ),
    )


def get_issue_comment_body(helper: Helper) -> str:
    payload = helper.payload
    include_title = helper.include_title
    comment = payload["comment"]
    issue = payload["issue"]

    return get_pull_request_event_message(
        user_name=get_sender_name(helper),
        action=get_comment_action(payload),
        url=issue["html_url"].tame(check_string),
        number=issue["number"].tame(check_int),
        message=comment["body"].tame(check_string),
        title=issue["title"].tame(check_string) if include_title else None,
        type="PR" if is_pull_request_comment_event(payload) else "issue",
    )


def get_issue_labeled_or_unlabeled_body(helper: Helper) -> str:
    payload = helper.payload
    include_title = helper.include_title
    issue = payload["issue"]

    return get_issue_labeled_or_unlabeled_event_message(
        user_name=get_sender_name(helper),
        action="added" if payload["action"].tame(check_string) == "labeled" else "removed",
        url=issue["html_url"].tame(check_string),
        number=issue["number"].tame(check_int),
        label_name=payload["label"]["name"].tame(check_string),
        user_url=get_sender_url(payload),
        title=issue["title"].tame(check_string) if include_title else None,
    )


def get_issue_milestoned_or_demilestoned_body(helper: Helper) -> str:
    payload = helper.payload
    include_title = helper.include_title
    issue = payload["issue"]

    return get_issue_milestoned_or_demilestoned_event_message(
        user_name=get_sender_name(helper),
        action="added" if payload["action"].tame(check_string) == "milestoned" else "removed",
        url=issue["html_url"].tame(check_string),
        number=issue["number"].tame(check_int),
        milestone_name=payload["milestone"]["title"].tame(check_string),
        milestone_url=payload["milestone"]["html_url"].tame(check_string),
        user_url=get_sender_url(payload),
        title=issue["title"].tame(check_string) if include_title else None,
    )


def get_fork_body(helper: Helper) -> str:
    payload = helper.payload
    forkee = payload["forkee"]
    return "{} forked [{}]({}).".format(
        get_sender_name(helper),
        forkee["name"].tame(check_string),
        forkee["html_url"].tame(check_string),
    )


def get_deployment_body(helper: Helper) -> str:
    return f"{get_sender_name(helper)} created new deployment."


def get_change_deployment_status_body(helper: Helper) -> str:
    payload = helper.payload
    state = payload["deployment_status"]["state"].tame(check_string)
    emoji = (
        DEPLOYMENT_AND_COMMIT_STATUS_EMOJI.get(state, "") if helper.include_emoji_indicators else ""
    )
    return f"{emoji} Deployment changed status to {state}.".strip()


def get_create_or_delete_body(action: str, helper: Helper) -> str:
    payload = helper.payload
    ref_type = payload["ref_type"].tame(check_string)
    return "{} {} {} {}.".format(
        get_sender_name(helper),
        action,
        ref_type,
        payload["ref"].tame(check_string),
    ).rstrip()


def get_commit_comment_body(helper: Helper) -> str:
    payload = helper.payload
    comment = payload["comment"]
    comment_url = comment["html_url"].tame(check_string)
    commit_url = comment_url.split("#", 1)[0]
    action = f"[commented]({comment_url})"
    return get_commits_comment_action_message(
        get_sender_name(helper),
        action,
        commit_url,
        comment["commit_id"].tame(check_string),
        comment["body"].tame(check_string),
    )


def get_push_tags_body(helper: Helper) -> str:
    payload = helper.payload
    return get_push_tag_event_message(
        get_sender_name(helper),
        get_tag_name_from_ref(payload["ref"].tame(check_string)),
        action="pushed" if payload["created"].tame(check_bool) else "removed",
    )


def get_push_commits_body(helper: Helper) -> str:
    payload = helper.payload
    commits_data = []
    for commit in payload["commits"]:
        if commit["author"].get("username"):
            name = commit["author"]["username"].tame(check_string)
        else:
            name = commit["author"]["name"].tame(check_string)
        commits_data.append(
            {
                "name": name,
                "sha": commit["id"].tame(check_string),
                "url": commit["url"].tame(check_string),
                "message": commit["message"].tame(check_string),
            }
        )
    return get_push_commits_event_message(
        get_sender_name(helper),
        payload["compare"].tame(check_string),
        get_branch_name_from_ref(payload["ref"].tame(check_string)),
        commits_data,
        deleted=payload["deleted"].tame(check_bool),
        force_push=payload["forced"].tame(check_bool),
        repository_name=get_repository_full_name(payload)
        if helper.include_repository_name
        else None,
        repository_url=get_repository_url(payload) if helper.include_repository_name else None,
    )


class LazyContext(dict[str, str | int]):
    """Template rendering context for discussions."""

    def __init__(self, helper: Helper) -> None:
        super().__init__()
        self.helper = helper
        self.payload = helper.payload
        self.include_title = helper.include_title
        self.template_values: dict[str, Callable[[], str | int]] = {
            "sender": lambda: get_sender_name(self.helper),
            "author": lambda: get_user_mention(
                self.helper,
                self.payload["discussion"]["user"]["login"].tame(check_string),
            ),
            "url": lambda: self.payload["discussion"]["html_url"].tame(check_string),
            "action": lambda: self.payload["action"].tame(check_string),
            "configured_title": lambda: (
                f" {self.template_values['title']()}" if self.include_title else ""
            ),
            "category": lambda: self.payload["discussion"]["category"]["name"].tame(check_string),
            "title": lambda: self.payload["discussion"]["title"].tame(check_string),
            "body": lambda: self.payload["discussion"]["body"].tame(check_string),
            "body_fence": lambda: get_unused_fence(str(self.template_values["body"]())),
            "repository_name": lambda: self.payload["repository"]["name"].tame(check_string),
            "new_repository_name": lambda: self.payload["changes"]["new_repository"]["name"].tame(
                check_string
            ),
            "discussion_number": lambda: self.payload["discussion"]["number"].tame(check_int),
            "new_discussion_number": lambda: self.payload["changes"]["new_discussion"][
                "number"
            ].tame(check_int),
            "label": lambda: self.payload["label"]["name"].tame(check_string),
            "old_category": lambda: self.payload["changes"]["category"]["from"]["name"].tame(
                check_string
            ),
            # locked_reason includes the " as " as prefix,
            # because locked_reason could be null too, in which case,
            # we drop this entire part from the message.
            "locked_reason": lambda: (
                f" as {self.payload['discussion']['active_lock_reason'].tame(check_string)}"
                if self.payload["discussion"]["active_lock_reason"]
                else ""
            ),
            "closed_reason": lambda: self.payload["discussion"]["state_reason"].tame(check_string),
            # answer_field is used to determine which payload field to use.
            # It is either "answer" (for answered action)
            # or "old_answer" (for unanswered action)
            "answer_field": lambda: (
                "old_answer"
                if self.payload["action"].tame(check_string) == "unanswered"
                else "answer"
            ),
            "answer_url": lambda: self.payload[self.template_values["answer_field"]()][
                "html_url"
            ].tame(check_string),
            "answer_body": lambda: self.payload[self.template_values["answer_field"]()][
                "body"
            ].tame(check_string),
            "answer_body_fence": lambda: get_unused_fence(
                str(self.template_values["answer_body"]())
            ),
            "comment_id": lambda: self.payload[self.template_values["answer_field"]()]["id"].tame(
                check_int
            ),
        }

    @override
    def __getitem__(self, key: str) -> str | int:
        return self.template_values[key]()


def get_discussion_body(helper: Helper) -> str:
    payload = helper.payload
    action = get_discussion_action(payload)
    DISCUSSION_TEMPLATE = DISCUSSION_TEMPLATES[action]
    context = LazyContext(helper)
    return DISCUSSION_TEMPLATE.format_map(context)


def get_discussion_action(payload: WildValue) -> str:
    action = payload["action"].tame(check_string)
    if action in ("unlocked", "pinned", "unpinned", "reopened"):
        action = "generic_action"
    if action == "edited":
        edited_field = "body" if "body" in payload["changes"] else "title"
        action = f"edited_{edited_field}"
    return action


def get_discussion_comment_body(helper: Helper) -> str:
    payload = helper.payload
    return get_pull_request_event_message(
        user_name=get_sender_name(helper),
        action=get_comment_action(payload),
        url=payload["discussion"]["html_url"].tame(check_string),
        number=payload["discussion"]["number"].tame(check_int),
        message=payload["comment"]["body"].tame(check_string),
        title=payload["discussion"]["title"].tame(check_string) if helper.include_title else None,
        type="discussion",
    )


def get_comment_action(payload: WildValue) -> str:
    action = payload["action"].tame(check_string)
    if action == "created":
        action = "[commented]"
    else:
        action = f"{action} a [comment]"
    action += "({}) on".format(payload["comment"]["html_url"].tame(check_string))
    return action


def get_public_body(helper: Helper) -> str:
    payload = helper.payload
    return "{} made the repository [{}]({}) public.".format(
        get_sender_name(helper),
        get_repository_full_name(payload),
        payload["repository"]["html_url"].tame(check_string),
    )


def get_wiki_pages_body(helper: Helper) -> str:
    payload = helper.payload
    wiki_page_info_template = "* {action} [{title}]({url})\n"
    wiki_info = ""
    for page in payload["pages"]:
        wiki_info += wiki_page_info_template.format(
            action=page["action"].tame(check_string),
            title=page["title"].tame(check_string),
            url=page["html_url"].tame(check_string),
        )
    return f"{get_sender_name(helper)}:\n{wiki_info.rstrip()}"


def get_watch_body(helper: Helper) -> str:
    payload = helper.payload
    return "{} starred the repository [{}]({}).".format(
        get_sender_name(helper),
        get_repository_full_name(payload),
        payload["repository"]["html_url"].tame(check_string),
    )


def get_repository_advisory_body(helper: Helper) -> str:
    payload = helper.payload
    action = payload["action"].tame(check_string)
    if action == "reported":
        description = payload["repository_advisory"]["description"].tame(check_string)
        fence = get_unused_fence(description)
        return "{} reported [{}]({}) in {}: {}\n\n{}quote\n{}\n{}".format(
            get_sender_name(helper),
            payload["repository_advisory"]["ghsa_id"].tame(check_string),
            payload["repository_advisory"]["html_url"].tame(check_string),
            get_repository_full_name(payload),
            payload["repository_advisory"]["summary"].tame(check_string),
            fence,
            description,
            fence,
        )
    else:
        return "{} published [{}]({})".format(
            get_sender_name(helper),
            payload["repository_advisory"]["ghsa_id"].tame(check_string),
            payload["repository_advisory"]["html_url"].tame(check_string),
        )


def get_repository_body(helper: Helper) -> str:
    payload = helper.payload
    return "{} {} the repository [{}]({}).".format(
        get_sender_name(helper),
        payload["action"].tame(check_string),
        get_repository_full_name(payload),
        payload["repository"]["html_url"].tame(check_string),
    )


def get_add_team_body(helper: Helper) -> str:
    payload = helper.payload
    return "The repository [{}]({}) was added to team {}.".format(
        get_repository_full_name(payload),
        payload["repository"]["html_url"].tame(check_string),
        payload["team"]["name"].tame(check_string),
    )


def get_team_body(helper: Helper) -> str:
    payload = helper.payload
    changes = payload["changes"]
    if "description" in changes:
        actor = get_sender_name(helper)
        new_description = payload["team"]["description"].tame(check_string)
        fence = get_unused_fence(new_description)
        return (
            f"{actor} changed the team description to:\n\n{fence} quote\n{new_description}\n{fence}"
        )
    if "name" in changes:
        original_name = changes["name"]["from"].tame(check_string)
        new_name = payload["team"]["name"].tame(check_string)
        return f"Team `{original_name}` was renamed to `{new_name}`."
    if "privacy" in changes:
        new_visibility = payload["team"]["privacy"].tame(check_string)
        return f"Team visibility changed to `{new_visibility}`"

    missing_keys = "/".join(sorted(changes.keys()))
    helper.log_unsupported(f"team/edited (changes: {missing_keys})")

    # Do our best to give useful info to the customer--at least
    # if they know something changed, they can go to GitHub for
    # more details.  And if it's just spam, you can control that
    # from GitHub.
    return f"Team has changes to `{missing_keys}` data."


def get_release_body(helper: Helper) -> str:
    payload = helper.payload
    if payload["release"]["name"]:
        release_name = payload["release"]["name"].tame(check_string)
    else:
        release_name = payload["release"]["tag_name"].tame(check_string)
    data = {
        "user_name": get_sender_name(helper),
        "action": payload["action"].tame(check_string),
        "tagname": payload["release"]["tag_name"].tame(check_string),
        # Not every GitHub release has a "name" set; if not there, use the tag name.
        "release_name": release_name,
        "url": payload["release"]["html_url"].tame(check_string),
    }

    return get_release_event_message(**data)


def get_page_build_body(helper: Helper) -> str:
    payload = helper.payload
    build = payload["build"]
    status = build["status"].tame(check_string)
    actions = {
        "null": "has yet to be built",
        "building": "is being built",
        "errored": "has failed: {}",
        "built": "has finished building",
    }

    action = actions.get(status, f"is {status}")
    if build["error"]["message"]:
        message = build["error"]["message"].tame(check_string)
        action = action.format(
            CONTENT_MESSAGE_TEMPLATE.format(
                message=message,
                fence=get_unused_fence(message),
            ),
        )

    return "GitHub Pages build, triggered by {}, {}.".format(
        payload["build"]["pusher"]["login"].tame(check_string),
        action,
    )


def get_status_body(helper: Helper) -> str:
    payload = helper.payload
    state = payload["state"].tame(check_string)
    if payload["target_url"]:
        status = "[{state}]({target_url})".format(
            state=state,
            target_url=payload["target_url"].tame(check_string),
        )
    else:
        status = state
    return "{emoji} [{commit}]({url}) changed its status to {status}.".format(
        emoji=DEPLOYMENT_AND_COMMIT_STATUS_EMOJI.get(state, "")
        if helper.include_emoji_indicators
        else "",
        commit=get_short_sha(payload["sha"].tame(check_string)),
        url=payload["commit"]["html_url"].tame(check_string),
        status=status,
    ).strip()


def get_locked_or_unlocked_pull_request_body(helper: Helper) -> str:
    payload = helper.payload

    action = payload["action"].tame(check_string)

    message = "{sender} has locked [PR #{pr_number}]({pr_url}) as {reason} and limited conversation to collaborators."
    if action == "unlocked":
        message = "{sender} has unlocked [PR #{pr_number}]({pr_url})."
    if payload["pull_request"]["active_lock_reason"]:
        active_lock_reason = payload["pull_request"]["active_lock_reason"].tame(check_string)
    else:
        active_lock_reason = None
    return message.format(
        sender=get_sender_name(helper),
        pr_number=payload["pull_request"]["number"].tame(check_int),
        pr_url=payload["pull_request"]["html_url"].tame(check_string),
        reason=active_lock_reason,
    )


def get_pull_request_auto_merge_body(helper: Helper) -> str:
    payload = helper.payload

    action = payload["action"].tame(check_string)

    message = "{sender} has enabled auto merge for [PR #{pr_number}]({pr_url})."
    if action == "auto_merge_disabled":
        message = "{sender} has disabled auto merge for [PR #{pr_number}]({pr_url})."
    return message.format(
        sender=get_sender_name(helper),
        pr_number=payload["pull_request"]["number"].tame(check_int),
        pr_url=payload["pull_request"]["html_url"].tame(check_string),
    )


def get_pull_request_ready_for_review_body(helper: Helper) -> str:
    payload = helper.payload

    message = "{sender} has marked [PR #{pr_number}]({pr_url}) as ready for review."
    return message.format(
        sender=get_sender_name(helper),
        pr_number=payload["pull_request"]["number"].tame(check_int),
        pr_url=payload["pull_request"]["html_url"].tame(check_string),
    )


def get_pull_request_review_body(helper: Helper) -> str:
    payload = helper.payload
    include_title = helper.include_title
    title = "for #{} {}".format(
        payload["pull_request"]["number"].tame(check_int),
        payload["pull_request"]["title"].tame(check_string),
    )
    return get_pull_request_event_message(
        user_name=get_sender_name(helper),
        action="submitted",
        url=payload["review"]["html_url"].tame(check_string),
        type="PR review",
        title=title if include_title else None,
        message=payload["review"]["body"].tame(check_none_or(check_string)),
        emoji=PR_REVIEW_STATE_EMOJI.get(payload["review"]["state"].tame(check_string))
        if helper.include_emoji_indicators
        else None,
    )


def get_pull_request_review_request_removed_body(helper: Helper) -> str:
    payload = helper.payload

    sender = get_sender_name(helper)
    reviewer = payload["requested_reviewer"]["login"].tame(check_string)
    pr_number = payload["pull_request"]["number"].tame(check_int)
    title = payload["pull_request"]["title"].tame(check_string)
    pr_url = payload["pull_request"]["html_url"].tame(check_string)

    return f"{sender} unassigned {reviewer} from [PR #{pr_number} {title}]({pr_url})."


def get_pull_request_converted_to_draft_body(helper: Helper) -> str:
    payload = helper.payload
    return get_pull_request_event_message(
        user_name=get_sender_name(helper),
        action="converted",
        url=payload["pull_request"]["html_url"].tame(check_string),
        number=payload["pull_request"]["number"].tame(check_int),
        title=payload["pull_request"]["title"].tame(check_string),
        suffix="to a draft",
    )


def get_pull_request_labeled_or_unlabeled_body(helper: Helper) -> str:
    payload = helper.payload
    label_name = payload["label"]["name"].tame(check_string)
    action = payload["action"].tame(check_string)
    sender = get_sender_name(helper)
    pr_number = payload["pull_request"]["number"].tame(check_int)
    pr_url = payload["pull_request"]["html_url"].tame(check_string)
    preposition = "on" if action == "labeled" else "from"
    action_word = "added" if action == "labeled" else "removed"

    return f"{sender} {action_word} the label `{label_name}` {preposition} [PR #{pr_number}]({pr_url})."


def get_pull_request_milestoned_or_demilestoned_body(helper: Helper) -> str:
    payload = helper.payload
    action = payload["action"].tame(check_string)

    if action == "milestoned":
        suffix = (
            f"to the milestone `{payload['pull_request']['milestone']['title'].tame(check_string)}`"
        )
    else:
        suffix = f"from the milestone `{payload['milestone']['title'].tame(check_string)}`"

    return get_pull_request_event_message(
        user_name=get_sender_name(helper),
        action=action,
        url=payload["pull_request"]["html_url"].tame(check_string),
        number=payload["pull_request"]["number"].tame(check_int),
        suffix=suffix,
    )


def get_pull_request_enqueued_or_dequeued_body(helper: Helper) -> str:
    payload = helper.payload
    action = payload["action"].tame(check_string)
    return get_pull_request_event_message(
        user_name=get_sender_name(helper),
        action=action,
        url=payload["pull_request"]["html_url"].tame(check_string),
        number=payload["pull_request"]["number"].tame(check_int),
        title=payload["pull_request"]["title"].tame(check_string),
        suffix=("to the merge queue" if action == "enqueued" else "from the merge queue"),
    )


def get_pull_request_review_comment_body(helper: Helper) -> str:
    payload = helper.payload
    include_title = helper.include_title
    action = payload["action"].tame(check_string)
    message = None
    if action == "created":
        message = payload["comment"]["body"].tame(check_string)

    title = "on #{} {}".format(
        payload["pull_request"]["number"].tame(check_int),
        payload["pull_request"]["title"].tame(check_string),
    )

    return get_pull_request_event_message(
        user_name=get_sender_name(helper),
        action=action,
        url=payload["comment"]["html_url"].tame(check_string),
        message=message,
        type="PR review comment",
        title=title if include_title else None,
        emoji=PR_REVIEW_COMMENT_EMOJI if helper.include_emoji_indicators else None,
    )


def get_pull_request_review_requested_body(helper: Helper) -> str:
    payload = helper.payload
    include_title = helper.include_title

    sender = get_sender_name(helper)
    pr_number = payload["pull_request"]["number"].tame(check_int)
    pr_url = payload["pull_request"]["html_url"].tame(check_string)
    message = "{sender} requested {reviewers} for a review on [PR #{pr_number}]({pr_url})."
    message_with_title = (
        "{sender} requested {reviewers} for a review on [PR #{pr_number} {title}]({pr_url})."
    )
    body = message_with_title if include_title else message

    if "requested_reviewer" in payload:
        reviewer = payload["requested_reviewer"]
        reviewers = "[{login}]({html_url})".format(
            login=reviewer["login"].tame(check_string),
            html_url=reviewer["html_url"].tame(check_string),
        )
    else:
        team_reviewer = payload["requested_team"]
        reviewers = "[{name}]({html_url})".format(
            name=team_reviewer["name"].tame(check_string),
            html_url=team_reviewer["html_url"].tame(check_string),
        )

    return body.format(
        sender=sender,
        reviewers=reviewers,
        pr_number=pr_number,
        pr_url=pr_url,
        title=payload["pull_request"]["title"].tame(check_string) if include_title else None,
    )


def get_check_run_body(helper: Helper) -> str:
    payload = helper.payload
    template = """
{emoji} Check [{name}]({html_url}) {status} ({conclusion}). ([{short_hash}]({commit_url}))
"""

    kwargs = {
        "name": payload["check_run"]["name"].tame(check_string),
        "html_url": payload["check_run"]["html_url"].tame(check_string),
        "status": payload["check_run"]["status"].tame(check_string),
        "short_hash": get_short_sha(payload["check_run"]["head_sha"].tame(check_string)),
        "commit_url": "{}/commit/{}".format(
            payload["repository"]["html_url"].tame(check_string),
            payload["check_run"]["head_sha"].tame(check_string),
        ),
        "conclusion": payload["check_run"]["conclusion"].tame(check_string),
        "emoji": CHECK_RUN_CONCLUSION_EMOJI.get(
            payload["check_run"]["conclusion"].tame(check_string), ""
        )
        if helper.include_emoji_indicators
        else "",
    }

    return template.format(**kwargs).strip()


def get_star_body(helper: Helper) -> str:
    payload = helper.payload
    template = "[{user}]({user_url}) {action} the repository [{repo}]({url})."
    return template.format(
        user=get_sender_name(helper),
        user_url=get_sender_url(payload),
        action="starred" if payload["action"].tame(check_string) == "created" else "unstarred",
        repo=get_repository_full_name(payload),
        url=payload["repository"]["html_url"].tame(check_string),
    )


def get_ping_body(helper: Helper) -> str:
    return get_setup_webhook_message("GitHub", get_sender_name(helper))


def get_cancelled_body(helper: Helper) -> str:
    payload = helper.payload
    template = "{user_name} cancelled their {subscription} subscription."
    return template.format(
        user_name=get_sender_name(helper),
        subscription=get_subscription(payload),
    ).rstrip()


def get_created_body(helper: Helper) -> str:
    payload = helper.payload
    template = "{user_name} subscribed for {subscription}."
    return template.format(
        user_name=get_sender_name(helper),
        subscription=get_subscription(payload),
    ).rstrip()


def get_edited_body(helper: Helper) -> str:
    payload = helper.payload
    template = "{user_name} changed who can see their sponsorship from {prior_privacy_level} to {privacy_level}."
    return template.format(
        user_name=get_sender_name(helper),
        prior_privacy_level=payload["changes"]["privacy_level"]["from"].tame(check_string),
        privacy_level=payload["sponsorship"]["privacy_level"].tame(check_string),
    ).rstrip()


def get_pending_cancellation_body(helper: Helper) -> str:
    payload = helper.payload
    template = "{user_name}'s {subscription} subscription will be cancelled on {effective_date}."
    return template.format(
        user_name=get_sender_name(helper),
        subscription=get_subscription(payload),
        effective_date=get_effective_date(payload),
    ).rstrip()


def get_pending_tier_change_body(helper: Helper) -> str:
    payload = helper.payload
    template = "{user_name}'s subscription will change from {prior_subscription} to {subscription} on {effective_date}."
    return template.format(
        user_name=get_sender_name(helper),
        prior_subscription=get_prior_subscription(payload),
        subscription=get_subscription(payload),
        effective_date=get_effective_date(payload),
    ).rstrip()


def get_tier_changed_body(helper: Helper) -> str:
    payload = helper.payload
    template = "{user_name} changed their subscription from {prior_subscription} to {subscription}."
    return template.format(
        user_name=get_sender_name(helper),
        prior_subscription=get_prior_subscription(payload),
        subscription=get_subscription(payload),
    ).rstrip()


def get_issue_transferred_body(helper: Helper) -> str:
    payload = helper.payload
    template = "{sender} transferred [issue #{old_issue_number} {title}]({old_issue_url}) to [{new_repo_full_name}/#{new_issue_number}]({new_issue_url})."
    return template.format(
        sender=get_sender_name(helper),
        old_issue_number=payload["issue"]["number"].tame(check_int),
        old_issue_url=payload["issue"]["html_url"].tame(check_string),
        title=payload["issue"]["title"].tame(check_string),
        new_repo_full_name=payload["changes"]["new_repository"]["full_name"].tame(check_string),
        new_issue_number=payload["changes"]["new_issue"]["number"].tame(check_int),
        new_issue_url=payload["changes"]["new_issue"]["html_url"].tame(check_string),
    )


def get_issue_opened_via_transfer_body(helper: Helper) -> str:
    payload = helper.payload
    template = "[Issue #{new_issue_number} {title}]({new_issue_url}) was transferred from [{old_repo_full_name}#{old_issue_number}]({old_issue_url})."
    return template.format(
        new_issue_number=payload["issue"]["number"].tame(check_int),
        new_issue_url=payload["issue"]["html_url"].tame(check_string),
        title=payload["issue"]["title"].tame(check_string),
        old_repo_full_name=payload["changes"]["old_repository"]["full_name"].tame(check_string),
        old_issue_number=payload["changes"]["old_issue"]["number"].tame(check_int),
        old_issue_url=payload["changes"]["old_issue"]["html_url"].tame(check_string),
    )


def get_subscription(payload: WildValue) -> str:
    return payload["sponsorship"]["tier"]["name"].tame(check_string)


def get_effective_date(payload: WildValue) -> str:
    effective_date = payload["effective_date"].tame(check_string)[:10]
    return datetime.fromisoformat(effective_date).strftime("%B %d, %Y")


def get_prior_subscription(payload: WildValue) -> str:
    return payload["changes"]["tier"]["from"]["name"].tame(check_string)


def get_repository_name(payload: WildValue) -> str:
    return payload["repository"]["name"].tame(check_string)


def get_repository_full_name(payload: WildValue) -> str:
    return payload["repository"]["full_name"].tame(check_string)


def get_repository_url(payload: WildValue) -> str:
    return payload["repository"]["html_url"].tame(check_string)


def get_organization_name(payload: WildValue) -> str:
    return payload["organization"]["login"].tame(check_string)


def get_user_mention(helper: Helper, github_username: str) -> str:
    external_account_field_name = str(DEFAULT_EXTERNAL_ACCOUNTS["github"].name)
    zulip_user = guess_zulip_user_from_external_account(
        helper.realm,
        github_username,
        external_account_field_name,
        external_username_case_insensitive=True,
    )
    if zulip_user is not None:
        return silent_mention_syntax_for_user(zulip_user)
    return github_username


def get_sender_name(helper: Helper) -> str:
    return get_user_mention(helper, helper.payload["sender"]["login"].tame(check_string))


def get_sender_url(payload: WildValue) -> str:
    return payload["sender"]["html_url"].tame(check_string)


def get_branch_name_from_ref(ref_string: str) -> str:
    return re.sub(r"^refs/heads/", "", ref_string)


def get_tag_name_from_ref(ref_string: str) -> str:
    return re.sub(r"^refs/tags/", "", ref_string)


def is_commit_push_event(payload: WildValue) -> bool:
    return payload["ref"].tame(check_string).startswith("refs/heads/")


def is_merge_queue_push_event(payload: WildValue) -> bool:
    return payload["ref"].tame(check_string).startswith("refs/heads/gh-readonly-queue/")


def is_pull_request_comment_event(payload: WildValue) -> bool:
    # When a comment is made on a PR, the event still has the header
    # "issue_comment", but the payload has a "pull_request" key.
    # This is just a workaround to get the correct topic.
    if "pull_request" in payload["issue"]:
        return True
    return False


def get_topic_based_on_type(payload: WildValue, event: str) -> str:
    if "pull_request" in event:
        return TOPIC_WITH_PR_OR_ISSUE_INFO_TEMPLATE.format(
            repo=get_repository_name(payload),
            type="PR",
            id=payload["pull_request"]["number"].tame(check_int),
            title=payload["pull_request"]["title"].tame(check_string),
        )
    elif event.startswith("issue"):
        type_for_topic = "PR" if is_pull_request_comment_event(payload) else "issue"
        return TOPIC_WITH_PR_OR_ISSUE_INFO_TEMPLATE.format(
            repo=get_repository_name(payload),
            type=type_for_topic,
            id=payload["issue"]["number"].tame(check_int),
            title=payload["issue"]["title"].tame(check_string),
        )
    elif event.startswith("deployment"):
        return "{} / Deployment on {}".format(
            get_repository_name(payload),
            payload["deployment"]["environment"].tame(check_string),
        )
    elif event == "membership":
        return "{} organization".format(payload["organization"]["login"].tame(check_string))
    elif event == "team":
        return "team {}".format(payload["team"]["name"].tame(check_string))
    elif event == "push_commits":
        return TOPIC_WITH_BRANCH_TEMPLATE.format(
            repo=get_repository_name(payload),
            branch=get_branch_name_from_ref(payload["ref"].tame(check_string)),
        )
    elif event == "gollum":
        return TOPIC_WITH_BRANCH_TEMPLATE.format(
            repo=get_repository_name(payload),
            branch="wiki pages",
        )
    elif event == "ping":
        if "repository" in payload:
            return get_repository_name(payload)
        if "organization" in payload:
            return get_organization_name(payload)
        # GitHub Sponsors webhook pings include neither "repository" nor
        # "organization"; group them under the topic SPONSORS_EVENT_TYPES use.
        hook_type = payload["hook"]["type"].tame(check_string)
        if hook_type == "SponsorsListing":
            return "sponsors"
        raise UnsupportedWebhookEventTypeError(f"ping:{hook_type}")
    elif event == "check_run":
        return f"{get_repository_name(payload)} / checks"
    elif event.startswith("discussion"):
        return TOPIC_FOR_DISCUSSION.format(
            repo=get_repository_name(payload),
            number=payload["discussion"]["number"].tame(check_int),
            title=payload["discussion"]["title"].tame(check_string),
        )
    elif event in SPONSORS_EVENT_TYPES:
        return "sponsors"

    return get_repository_name(payload)


EVENT_FUNCTION_MAPPER: dict[str, Callable[[Helper], str]] = {
    "commit_comment": get_commit_comment_body,
    "closed_pull_request": get_closed_pull_request_body,
    "create": partial(get_create_or_delete_body, "created"),
    "check_run": get_check_run_body,
    "delete": partial(get_create_or_delete_body, "deleted"),
    "deployment": get_deployment_body,
    "deployment_status": get_change_deployment_status_body,
    "discussion": get_discussion_body,
    "discussion_comment": get_discussion_comment_body,
    "fork": get_fork_body,
    "gollum": get_wiki_pages_body,
    "issue_comment": get_issue_comment_body,
    "issue_labeled_or_unlabeled": get_issue_labeled_or_unlabeled_body,
    "issue_milestoned_or_demilestoned": get_issue_milestoned_or_demilestoned_body,
    "issues_opened_via_transfer": get_issue_opened_via_transfer_body,
    "issues_transferred": get_issue_transferred_body,
    "issues": get_issue_body,
    "member": get_member_body,
    "membership": get_membership_body,
    "opened_pull_request": get_opened_or_update_pull_request_body,
    "updated_pull_request": get_opened_or_update_pull_request_body,
    "assigned_or_unassigned_pull_request": get_assigned_or_unassigned_pull_request_body,
    "page_build": get_page_build_body,
    "ping": get_ping_body,
    "public": get_public_body,
    "pull_request_ready_for_review": get_pull_request_ready_for_review_body,
    "pull_request_review": get_pull_request_review_body,
    "pull_request_review_comment": get_pull_request_review_comment_body,
    "pull_request_review_requested": get_pull_request_review_requested_body,
    "pull_request_milestoned_or_demilestoned": get_pull_request_milestoned_or_demilestoned_body,
    "pull_request_enqueued_or_dequeued": get_pull_request_enqueued_or_dequeued_body,
    "pull_request_labeled_or_unlabeled": get_pull_request_labeled_or_unlabeled_body,
    "pull_request_converted_to_draft": get_pull_request_converted_to_draft_body,
    "pull_request_review_request_removed": get_pull_request_review_request_removed_body,
    "pull_request_auto_merge": get_pull_request_auto_merge_body,
    "locked_or_unlocked_pull_request": get_locked_or_unlocked_pull_request_body,
    "push_commits": get_push_commits_body,
    "push_tags": get_push_tags_body,
    "release": get_release_body,
    "repository_advisory": get_repository_advisory_body,
    "repository": get_repository_body,
    "star": get_star_body,
    "status": get_status_body,
    "team": get_team_body,
    "team_add": get_add_team_body,
    "watch": get_watch_body,
    "cancelled": get_cancelled_body,
    "created": get_created_body,
    "edited": get_edited_body,
    "pending_cancellation": get_pending_cancellation_body,
    "pending_tier_change": get_pending_tier_change_body,
    "tier_changed": get_tier_changed_body,
}

SPONSORS_EVENT_TYPES = [
    "cancelled",
    "created",
    "edited",
    "pending_cancellation",
    "pending_tier_change",
    "tier_changed",
]

IGNORED_EVENTS = [
    "check_suite",
    "label",
    "meta",
    "milestone",
    "organization",
    "project_card",
    "push__merge_queue",
    "repository_vulnerability_alert",
]

IGNORED_PULL_REQUEST_ACTIONS = [
    "approved",
]

IGNORED_TEAM_ACTIONS = [
    # These are actions that are well documented by github
    # (https://docs.github.com/en/developers/webhooks-and-events/webhook-events-and-payloads)
    # but we ignore them for now, possibly just due to laziness.
    # One curious example here is team/added_to_repository, which is
    # possibly the same as team_add.
    "added_to_repository",
    "created",
    "deleted",
    "removed_from_repository",
]

ALL_EVENT_TYPES = list(EVENT_FUNCTION_MAPPER.keys())


@webhook_view("GitHub", notify_bot_owner_on_invalid_json=True, all_event_types=ALL_EVENT_TYPES)
@typed_endpoint
def api_github_webhook(
    request: HttpRequest,
    user_profile: UserProfile,
    *,
    payload: JsonBodyPayload[WildValue],
    branches: str | None = None,
    user_specified_topic: OptionalUserSpecifiedTopicStr = None,
    ignore_private_repositories: Json[bool] = False,
    include_repository_name: Json[bool] = False,
    include_emoji_indicators: Json[bool] = True,
) -> HttpResponse:
    """
    GitHub sends the event as an HTTP header.  We have our
    own Zulip-specific concept of an event that often maps
    directly to the X-GitHub-Event header's event, but we sometimes
    refine it based on the payload.
    """
    header_event = get_event_header(request, "X-GitHub-Event", "GitHub")

    # Ignore events from private repositories if the URL option is set
    if (
        "repository" in payload
        and payload["repository"]["private"].tame(check_bool)
        and ignore_private_repositories
    ):
        # Ignore private repository events
        return json_success(request)

    # Ignore 'comment edited' events (except discussion comments)
    # if the comment body remains unchanged.
    if (
        "comment" in header_event
        and "discussion" not in header_event
        and payload.get("action", "").tame(check_string) == "edited"
        and payload["changes"]["body"]["from"].tame(check_string)
        == payload["comment"]["body"].tame(check_string)
    ):
        return json_success(request)

    event = get_zulip_event_name(header_event, payload, branches)
    if event is None:
        # This is nothing to worry about--get_event() returns None
        # for events that are valid but not yet handled by us.
        # See IGNORED_EVENTS, for example.
        return json_success(request)
    topic_name = get_topic_based_on_type(payload, event)

    body_function = EVENT_FUNCTION_MAPPER[event]

    helper = Helper(
        request=request,
        payload=payload,
        include_title=user_specified_topic is not None,
        include_repository_name=include_repository_name,
        include_emoji_indicators=include_emoji_indicators,
        user_profile=user_profile,
    )
    body = body_function(helper)

    check_send_webhook_message(request, user_profile, topic_name, body, event)
    return json_success(request)


def is_empty_pull_request_review_event(payload: WildValue) -> bool:
    # When submitting a review, GitHub has a bug where it'll send a
    # duplicate empty "edited" event for the main review body. We ignore
    # those, to avoid triggering duplicate notifications.
    action = payload["action"].tame(check_string)
    changes = payload.get("changes", {})
    return action == "edited" and len(changes) == 0


PULL_REQUEST_ACTION_TO_EVENT = {
    "opened": "opened_pull_request",
    "reopened": "opened_pull_request",
    "synchronize": "updated_pull_request",
    "edited": "updated_pull_request",
    "assigned": "assigned_or_unassigned_pull_request",
    "unassigned": "assigned_or_unassigned_pull_request",
    "closed": "closed_pull_request",
    "review_requested": "pull_request_review_requested",
    "review_request_removed": "pull_request_review_request_removed",
    "ready_for_review": "pull_request_ready_for_review",
    "locked": "locked_or_unlocked_pull_request",
    "unlocked": "locked_or_unlocked_pull_request",
    "auto_merge_enabled": "pull_request_auto_merge",
    "auto_merge_disabled": "pull_request_auto_merge",
    "milestoned": "pull_request_milestoned_or_demilestoned",
    "demilestoned": "pull_request_milestoned_or_demilestoned",
    "enqueued": "pull_request_enqueued_or_dequeued",
    "dequeued": "pull_request_enqueued_or_dequeued",
    "labeled": "pull_request_labeled_or_unlabeled",
    "unlabeled": "pull_request_labeled_or_unlabeled",
    "converted_to_draft": "pull_request_converted_to_draft",
}

ISSUE_ACTION_TO_EVENT = {
    "labeled": "issue_labeled_or_unlabeled",
    "unlabeled": "issue_labeled_or_unlabeled",
    "milestoned": "issue_milestoned_or_demilestoned",
    "demilestoned": "issue_milestoned_or_demilestoned",
    "transferred": "issues_transferred",
}


def get_push_event_name(payload: WildValue, branches: str | None) -> str | None:
    if is_merge_queue_push_event(payload):
        return None
    if not is_commit_push_event(payload):
        return "push_tags"
    if branches is not None:
        branch = get_branch_name_from_ref(payload["ref"].tame(check_string))
        if not is_branch_name_notifiable(branch, branches):
            return None
    return "push_commits"


def get_zulip_event_name(
    header_event: str,
    payload: WildValue,
    branches: str | None,
) -> str | None:
    action = payload.get("action", "").tame(check_string)
    match header_event:
        case "pull_request":
            if action in IGNORED_PULL_REQUEST_ACTIONS:
                return None  # nocoverage
            return PULL_REQUEST_ACTION_TO_EVENT.get(action)
        case "issues":
            if action == "opened" and payload.get("changes", {}).get("old_issue"):
                return "issues_opened_via_transfer"
            return ISSUE_ACTION_TO_EVENT.get(action, "issues")
        case "pull_request_review":
            return None if is_empty_pull_request_review_event(payload) else "pull_request_review"
        case "push":
            return get_push_event_name(payload, branches)
        case "check_run":
            status = payload["check_run"]["status"].tame(check_string)
            return "check_run" if status == "completed" else None
        case "team":
            if action == "edited":
                return "team"
            if action in IGNORED_TEAM_ACTIONS:
                return None
            raise UnsupportedWebhookEventTypeError(f"unsupported team action {action}")
        case _:
            if header_event in IGNORED_EVENTS:
                return None
            if header_event in EVENT_FUNCTION_MAPPER:
                return header_event
            raise UnsupportedWebhookEventTypeError(f"{header_event}:{action or '???'}")