mirror of
https://github.com/zulip/zulip.git
synced 2025-10-29 11:03:54 +00:00
a2c578e84d
Previously, some call sites for the function provided optional arguments as positional arguments. These changes will allow the arguments to be passed as keyword arguments to the function and fix up the call sites of the function to pass keyword arguments instead.
356 lines
11 KiB
Python
356 lines
11 KiB
Python
import string
|
|
from collections import defaultdict
|
|
from typing import Any, Dict, List, Optional, Tuple
|
|
|
|
TOPIC_WITH_BRANCH_TEMPLATE = "{repo} / {branch}"
|
|
TOPIC_WITH_PR_OR_ISSUE_INFO_TEMPLATE = "{repo} / {type} #{id} {title}"
|
|
TOPIC_WITH_RELEASE_TEMPLATE = "{repo} / {tag} {title}"
|
|
|
|
EMPTY_SHA = "0000000000000000000000000000000000000000"
|
|
|
|
COMMITS_LIMIT = 20
|
|
COMMIT_ROW_TEMPLATE = "* {commit_msg} ([{commit_short_sha}]({commit_url}))\n"
|
|
COMMITS_MORE_THAN_LIMIT_TEMPLATE = "[and {commits_number} more commit(s)]"
|
|
COMMIT_OR_COMMITS = "commit{}"
|
|
|
|
PUSH_PUSHED_TEXT_WITH_URL = "[pushed]({compare_url}) {number_of_commits} {commit_or_commits}"
|
|
PUSH_PUSHED_TEXT_WITHOUT_URL = "pushed {number_of_commits} {commit_or_commits}"
|
|
|
|
PUSH_COMMITS_BASE = "{user_name} {pushed_text} to branch {branch_name}."
|
|
PUSH_COMMITS_MESSAGE_TEMPLATE_WITH_COMMITTERS = (
|
|
PUSH_COMMITS_BASE
|
|
+ """ {committers_details}.
|
|
|
|
{commits_data}
|
|
"""
|
|
)
|
|
PUSH_COMMITS_MESSAGE_TEMPLATE_WITHOUT_COMMITTERS = (
|
|
PUSH_COMMITS_BASE
|
|
+ """
|
|
|
|
{commits_data}
|
|
"""
|
|
)
|
|
PUSH_DELETE_BRANCH_MESSAGE_TEMPLATE = (
|
|
"{user_name} [deleted]({compare_url}) the branch {branch_name}."
|
|
)
|
|
PUSH_LOCAL_BRANCH_WITHOUT_COMMITS_MESSAGE_TEMPLATE = (
|
|
"{user_name} [pushed]({compare_url}) the branch {branch_name}."
|
|
)
|
|
PUSH_LOCAL_BRANCH_WITHOUT_COMMITS_MESSAGE_WITHOUT_URL_TEMPLATE = (
|
|
"{user_name} pushed the branch {branch_name}."
|
|
)
|
|
PUSH_COMMITS_MESSAGE_EXTENSION = "Commits by {}"
|
|
PUSH_COMMITTERS_LIMIT_INFO = 3
|
|
|
|
FORCE_PUSH_COMMITS_MESSAGE_TEMPLATE = (
|
|
"{user_name} [force pushed]({url}) to branch {branch_name}. Head is now {head}."
|
|
)
|
|
CREATE_BRANCH_MESSAGE_TEMPLATE = "{user_name} created [{branch_name}]({url}) branch."
|
|
CREATE_BRANCH_WITHOUT_URL_MESSAGE_TEMPLATE = "{user_name} created {branch_name} branch."
|
|
REMOVE_BRANCH_MESSAGE_TEMPLATE = "{user_name} deleted branch {branch_name}."
|
|
|
|
PULL_REQUEST_OR_ISSUE_MESSAGE_TEMPLATE = "{user_name} {action} [{type}{id}]({url})"
|
|
PULL_REQUEST_OR_ISSUE_MESSAGE_TEMPLATE_WITH_TITLE = (
|
|
"{user_name} {action} [{type}{id} {title}]({url})"
|
|
)
|
|
PULL_REQUEST_OR_ISSUE_ASSIGNEE_INFO_TEMPLATE = "(assigned to {assignee})"
|
|
PULL_REQUEST_BRANCH_INFO_TEMPLATE = "from `{target}` to `{base}`"
|
|
|
|
CONTENT_MESSAGE_TEMPLATE = "\n~~~ quote\n{message}\n~~~"
|
|
|
|
COMMITS_COMMENT_MESSAGE_TEMPLATE = "{user_name} {action} on [{sha}]({url})"
|
|
|
|
PUSH_TAGS_MESSAGE_TEMPLATE = """{user_name} {action} tag {tag}"""
|
|
TAG_WITH_URL_TEMPLATE = "[{tag_name}]({tag_url})"
|
|
TAG_WITHOUT_URL_TEMPLATE = "{tag_name}"
|
|
|
|
RELEASE_MESSAGE_TEMPLATE = "{user_name} {action} release [{release_name}]({url}) for tag {tagname}."
|
|
|
|
|
|
def get_push_commits_event_message(
|
|
user_name: str,
|
|
compare_url: Optional[str],
|
|
branch_name: str,
|
|
commits_data: List[Dict[str, Any]],
|
|
is_truncated: bool = False,
|
|
deleted: bool = False,
|
|
) -> str:
|
|
if not commits_data and deleted:
|
|
return PUSH_DELETE_BRANCH_MESSAGE_TEMPLATE.format(
|
|
user_name=user_name,
|
|
compare_url=compare_url,
|
|
branch_name=branch_name,
|
|
)
|
|
|
|
if not commits_data and not deleted:
|
|
if compare_url:
|
|
return PUSH_LOCAL_BRANCH_WITHOUT_COMMITS_MESSAGE_TEMPLATE.format(
|
|
user_name=user_name,
|
|
compare_url=compare_url,
|
|
branch_name=branch_name,
|
|
)
|
|
return PUSH_LOCAL_BRANCH_WITHOUT_COMMITS_MESSAGE_WITHOUT_URL_TEMPLATE.format(
|
|
user_name=user_name,
|
|
branch_name=branch_name,
|
|
)
|
|
|
|
pushed_message_template = (
|
|
PUSH_PUSHED_TEXT_WITH_URL if compare_url else PUSH_PUSHED_TEXT_WITHOUT_URL
|
|
)
|
|
|
|
pushed_text_message = pushed_message_template.format(
|
|
compare_url=compare_url,
|
|
number_of_commits=len(commits_data),
|
|
commit_or_commits=COMMIT_OR_COMMITS.format("s" if len(commits_data) > 1 else ""),
|
|
)
|
|
|
|
committers_items: List[Tuple[str, int]] = get_all_committers(commits_data)
|
|
if len(committers_items) == 1 and user_name == committers_items[0][0]:
|
|
return PUSH_COMMITS_MESSAGE_TEMPLATE_WITHOUT_COMMITTERS.format(
|
|
user_name=user_name,
|
|
pushed_text=pushed_text_message,
|
|
branch_name=branch_name,
|
|
commits_data=get_commits_content(commits_data, is_truncated),
|
|
).rstrip()
|
|
else:
|
|
committers_details = "{} ({})".format(*committers_items[0])
|
|
|
|
for name, number_of_commits in committers_items[1:-1]:
|
|
committers_details = f"{committers_details}, {name} ({number_of_commits})"
|
|
|
|
if len(committers_items) > 1:
|
|
committers_details = "{} and {} ({})".format(committers_details, *committers_items[-1])
|
|
|
|
return PUSH_COMMITS_MESSAGE_TEMPLATE_WITH_COMMITTERS.format(
|
|
user_name=user_name,
|
|
pushed_text=pushed_text_message,
|
|
branch_name=branch_name,
|
|
committers_details=PUSH_COMMITS_MESSAGE_EXTENSION.format(committers_details),
|
|
commits_data=get_commits_content(commits_data, is_truncated),
|
|
).rstrip()
|
|
|
|
|
|
def get_force_push_commits_event_message(
|
|
user_name: str, url: str, branch_name: str, head: str
|
|
) -> str:
|
|
return FORCE_PUSH_COMMITS_MESSAGE_TEMPLATE.format(
|
|
user_name=user_name,
|
|
url=url,
|
|
branch_name=branch_name,
|
|
head=head,
|
|
)
|
|
|
|
|
|
def get_create_branch_event_message(user_name: str, url: Optional[str], branch_name: str) -> str:
|
|
if url is None:
|
|
return CREATE_BRANCH_WITHOUT_URL_MESSAGE_TEMPLATE.format(
|
|
user_name=user_name,
|
|
branch_name=branch_name,
|
|
)
|
|
return CREATE_BRANCH_MESSAGE_TEMPLATE.format(
|
|
user_name=user_name,
|
|
url=url,
|
|
branch_name=branch_name,
|
|
)
|
|
|
|
|
|
def get_remove_branch_event_message(user_name: str, branch_name: str) -> str:
|
|
return REMOVE_BRANCH_MESSAGE_TEMPLATE.format(
|
|
user_name=user_name,
|
|
branch_name=branch_name,
|
|
)
|
|
|
|
|
|
def get_pull_request_event_message(
|
|
*,
|
|
user_name: str,
|
|
action: str,
|
|
url: str,
|
|
number: Optional[int] = None,
|
|
target_branch: Optional[str] = None,
|
|
base_branch: Optional[str] = None,
|
|
message: Optional[str] = None,
|
|
assignee: Optional[str] = None,
|
|
assignees: Optional[List[Dict[str, Any]]] = None,
|
|
type: str = "PR",
|
|
title: Optional[str] = None,
|
|
) -> str:
|
|
kwargs = {
|
|
"user_name": user_name,
|
|
"action": action,
|
|
"type": type,
|
|
"url": url,
|
|
"id": f" #{number}" if number is not None else "",
|
|
"title": title,
|
|
}
|
|
|
|
if title is not None:
|
|
main_message = PULL_REQUEST_OR_ISSUE_MESSAGE_TEMPLATE_WITH_TITLE.format(**kwargs)
|
|
else:
|
|
main_message = PULL_REQUEST_OR_ISSUE_MESSAGE_TEMPLATE.format(**kwargs)
|
|
|
|
if assignees:
|
|
assignees_string = ""
|
|
if len(assignees) == 1:
|
|
assignees_string = "{username}".format(**assignees[0])
|
|
else:
|
|
usernames = []
|
|
for a in assignees:
|
|
usernames.append(a["username"])
|
|
|
|
assignees_string = ", ".join(usernames[:-1]) + " and " + usernames[-1]
|
|
|
|
assignee_info = PULL_REQUEST_OR_ISSUE_ASSIGNEE_INFO_TEMPLATE.format(
|
|
assignee=assignees_string
|
|
)
|
|
main_message = f"{main_message} {assignee_info}"
|
|
|
|
elif assignee:
|
|
assignee_info = PULL_REQUEST_OR_ISSUE_ASSIGNEE_INFO_TEMPLATE.format(assignee=assignee)
|
|
main_message = f"{main_message} {assignee_info}"
|
|
|
|
if target_branch and base_branch:
|
|
branch_info = PULL_REQUEST_BRANCH_INFO_TEMPLATE.format(
|
|
target=target_branch,
|
|
base=base_branch,
|
|
)
|
|
main_message = f"{main_message} {branch_info}"
|
|
|
|
punctuation = ":" if message else "."
|
|
if (
|
|
assignees
|
|
or assignee
|
|
or (target_branch and base_branch)
|
|
or title is None
|
|
# Once we get here, we know that the message ends with a title
|
|
# which could already have punctuation at the end
|
|
or title[-1] not in string.punctuation
|
|
):
|
|
main_message = f"{main_message}{punctuation}"
|
|
|
|
if message:
|
|
main_message += "\n" + CONTENT_MESSAGE_TEMPLATE.format(message=message)
|
|
return main_message.rstrip()
|
|
|
|
|
|
def get_issue_event_message(
|
|
user_name: str,
|
|
action: str,
|
|
url: str,
|
|
number: Optional[int] = None,
|
|
message: Optional[str] = None,
|
|
assignee: Optional[str] = None,
|
|
assignees: Optional[List[Dict[str, Any]]] = None,
|
|
title: Optional[str] = None,
|
|
) -> str:
|
|
return get_pull_request_event_message(
|
|
user_name=user_name,
|
|
action=action,
|
|
url=url,
|
|
number=number,
|
|
message=message,
|
|
assignee=assignee,
|
|
assignees=assignees,
|
|
type="issue",
|
|
title=title,
|
|
)
|
|
|
|
|
|
def get_push_tag_event_message(
|
|
user_name: str, tag_name: str, tag_url: Optional[str] = None, action: str = "pushed"
|
|
) -> str:
|
|
if tag_url:
|
|
tag_part = TAG_WITH_URL_TEMPLATE.format(tag_name=tag_name, tag_url=tag_url)
|
|
else:
|
|
tag_part = TAG_WITHOUT_URL_TEMPLATE.format(tag_name=tag_name)
|
|
|
|
message = PUSH_TAGS_MESSAGE_TEMPLATE.format(
|
|
user_name=user_name,
|
|
action=action,
|
|
tag=tag_part,
|
|
)
|
|
|
|
if tag_name[-1] not in string.punctuation:
|
|
message = f"{message}."
|
|
|
|
return message
|
|
|
|
|
|
def get_commits_comment_action_message(
|
|
user_name: str, action: str, commit_url: str, sha: str, message: Optional[str] = None
|
|
) -> str:
|
|
content = COMMITS_COMMENT_MESSAGE_TEMPLATE.format(
|
|
user_name=user_name,
|
|
action=action,
|
|
sha=get_short_sha(sha),
|
|
url=commit_url,
|
|
)
|
|
punctuation = ":" if message else "."
|
|
content = f"{content}{punctuation}"
|
|
if message:
|
|
content += CONTENT_MESSAGE_TEMPLATE.format(
|
|
message=message,
|
|
)
|
|
|
|
return content
|
|
|
|
|
|
def get_commits_content(commits_data: List[Dict[str, Any]], is_truncated: bool = False) -> str:
|
|
commits_content = ""
|
|
for commit in commits_data[:COMMITS_LIMIT]:
|
|
commits_content += COMMIT_ROW_TEMPLATE.format(
|
|
commit_short_sha=get_short_sha(commit["sha"]),
|
|
commit_url=commit.get("url"),
|
|
commit_msg=commit["message"].partition("\n")[0],
|
|
)
|
|
|
|
if len(commits_data) > COMMITS_LIMIT:
|
|
commits_content += COMMITS_MORE_THAN_LIMIT_TEMPLATE.format(
|
|
commits_number=len(commits_data) - COMMITS_LIMIT,
|
|
)
|
|
elif is_truncated:
|
|
commits_content += COMMITS_MORE_THAN_LIMIT_TEMPLATE.format(
|
|
commits_number="",
|
|
).replace(" ", " ")
|
|
return commits_content.rstrip()
|
|
|
|
|
|
def get_release_event_message(
|
|
user_name: str, action: str, tagname: str, release_name: str, url: str
|
|
) -> str:
|
|
content = RELEASE_MESSAGE_TEMPLATE.format(
|
|
user_name=user_name,
|
|
action=action,
|
|
tagname=tagname,
|
|
release_name=release_name,
|
|
url=url,
|
|
)
|
|
|
|
return content
|
|
|
|
|
|
def get_short_sha(sha: str) -> str:
|
|
return sha[:11]
|
|
|
|
|
|
def get_all_committers(commits_data: List[Dict[str, Any]]) -> List[Tuple[str, int]]:
|
|
committers: Dict[str, int] = defaultdict(int)
|
|
|
|
for commit in commits_data:
|
|
committers[commit["name"]] += 1
|
|
|
|
# Sort by commit count, breaking ties alphabetically.
|
|
committers_items: List[Tuple[str, int]] = sorted(
|
|
committers.items(),
|
|
key=lambda item: (-item[1], item[0]),
|
|
)
|
|
committers_values: List[int] = [c_i[1] for c_i in committers_items]
|
|
|
|
if len(committers) > PUSH_COMMITTERS_LIMIT_INFO:
|
|
others_number_of_commits = sum(committers_values[PUSH_COMMITTERS_LIMIT_INFO:])
|
|
committers_items = committers_items[:PUSH_COMMITTERS_LIMIT_INFO]
|
|
committers_items.append(("others", others_number_of_commits))
|
|
|
|
return committers_items
|