Mirror
/
zulip
mirror of https://github.com/zulip/zulip.git
1
0
Fork 0
Code Issues Projects Releases Wiki Activity
zulip/zerver/lib/webhooks/git.py

298 lines
12 KiB
Python
Raw Blame History

import string
from typing import Optional, Any, Dict, List, Tuple
from collections import defaultdict
TOPIC_WITH_BRANCH_TEMPLATE = '{repo} / {branch}'
TOPIC_WITH_PR_OR_ISSUE_INFO_TEMPLATE = '{repo} / {type} #{id} {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_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}`"
SETUP_MESSAGE_TEMPLATE = "{integration} webhook has been successfully configured"
SETUP_MESSAGE_USER_PART = " by {user_name}"
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}"
def get_push_commits_event_message(user_name: str, compare_url: Optional[str],
branch_name: str, commits_data: List[Dict[str, Any]],
is_truncated: Optional[bool]=False,
deleted: Optional[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:
return PUSH_LOCAL_BRANCH_WITHOUT_COMMITS_MESSAGE_TEMPLATE.format(
user_name=user_name,
compare_url=compare_url,
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 = get_all_committers(commits_data) # type: List[Tuple[str, int]]
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 = "{}, {} ({})".format(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: Optional[str]='PR', title: Optional[str]=None) -> str:
kwargs = {
'user_name': user_name,
'action': action,
'type': type,
'url': url,
'id': ' #{}'.format(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 = "{} {}".format(main_message, assignee_info)
elif assignee:
assignee_info = PULL_REQUEST_OR_ISSUE_ASSIGNEE_INFO_TEMPLATE.format(
assignee=assignee)
main_message = "{} {}".format(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 = "{} {}".format(main_message, branch_info)
punctuation = ':' if message else '.'
if (assignees or assignee or (target_branch and base_branch) or (title is None)):
main_message = '{message}{punctuation}'.format(
message=main_message, punctuation=punctuation)
elif title is not None:
# Once we get here, we know that the message ends with a title
# which could already have punctuation at the end
if title[-1] not in string.punctuation:
main_message = '{message}{punctuation}'.format(
message=main_message, punctuation=punctuation)
if message:
main_message += '\n' + CONTENT_MESSAGE_TEMPLATE.format(message=message)
return main_message.rstrip()
def get_setup_webhook_message(integration: str, user_name: Optional[str]=None) -> str:
content = SETUP_MESSAGE_TEMPLATE.format(integration=integration)
if user_name:
content += SETUP_MESSAGE_USER_PART.format(user_name=user_name)
content = "{}.".format(content)
return content
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,
action,
url,
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: Optional[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 = '{}.'.format(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 = '{}{}'.format(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: Optional[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_short_sha(sha: str) -> str:
return sha[:7]
def get_all_committers(commits_data: List[Dict[str, Any]]) -> List[Tuple[str, int]]:
committers = defaultdict(int) # type: Dict[str, int]
for commit in commits_data:
committers[commit['name']] += 1
# Sort by commit count, breaking ties alphabetically.
committers_items = sorted(list(committers.items()),
key=lambda item: (-item[1], item[0])) # type: List[Tuple[str, int]]
committers_values = [c_i[1] for c_i in committers_items] # type: List[int]
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
Reference in New Issue View Git Blame Copy Permalink