bitbot-3.11-fork/modules/git_webhooks/gitlab.py

170 lines
5.9 KiB
Python
Raw Normal View History

from src import ModuleManager, utils
from . import colors
EVENT_CATEGORIES = {
"ping": [
"ping" # new webhook received
],
"code": ["push"],
"pr-minimal": [
"pull_request/opened", "pull_request/closed", "pull_request/reopened"
],
"pr": [
"pull_request/opened", "pull_request/closed", "pull_request/reopened",
"pull_request/edited", "pull_request/assigned",
"pull_request/unassigned"
],
"pr-all": ["pull_request"],
"issue-minimal": [
"issue/opened", "issue/closed", "issue/reopened", "issue/deleted"
],
"issues": [
"issue/opened", "issue/closed", "issue/reopened", "issue/deleted",
"issue/edited", "issue/assigned", "issue/unassigned", "issue_comment"
],
"issue-all": [
"issue", "issue_comment"
],
"issue-comment-minimal": [
"issue_comment/created", "issue_comment/deleted"
],
"repo": [
"create", # a repository, branch or tag has been created
"delete", # same as above but deleted
"release",
"fork",
"repository"
]
}
COMMENT_ACTIONS = {
"created": "commented",
"edited": "edited a comment",
"deleted": "deleted a comment"
}
class GitLab(object):
def is_private(self, data, headers):
if "project" in data:
return not data["project"]["visibility_level"] == 20
return False
def names(self, data, headers):
full_name = data["project"]["path_with_namespace"]
repo_username = data["project"]["namespace"]
repo_name = data["project"]["name"]
return full_name, repo_username, repo_name, None
def branch(self, data, headers):
if "ref" in data:
return data["ref"].rpartition("/")[2]
return None
def event(self, data, headers):
event = headers["X-GitLab-Event"].rsplit(" ", 1)[0].lower()
event = event.replace(" ", "_")
event_action = None
if ("object_attributes" in data and
"action" in data["object_attributes"]):
event_action = "%s/%s" % (
event, data["object_attributes"]["action"])
return event, event_action
def event_categories(self, event):
return EVENT_CATEGORIES.get(event, [event])
def webhook(self, full_name, event, data, headers):
if event == "push":
return self.push(full_name, data)
elif event == "merge_request":
return self.pull_request(full_name, data)
elif event == "issue":
return self.issues(full_name, data)
elif event == "note":
return self.note(full_name, data)
def _short_hash(self, hash):
return hash[:8]
def push(self, full_name, data):
outputs = []
branch = data["ref"].rpartition("/")[2]
branch = utils.irc.color(branch, colors.COLOR_BRANCH)
author = utils.irc.bold(data["user_username"])
if len(data["commits"]) <= 3:
for commit in data["commits"]:
hash = commit["id"]
hash_colored = utils.irc.color(self._short_hash(hash),
colors.COLOR_ID)
message = commit["message"].split("\n")[0].strip()
url = commit["url"]
outputs.append(
"%s pushed %s to %s: %s - %s"
% (author, hash_colored, branch, message, url))
else:
first_id = data["before"]
last_id = data["after"]
url = data["compare_url"]
outputs.append("%s pushed %d commits to %s"
% (author, len(data["commits"]), branch))
return outputs
def merge_request(self, full_name, data):
number = utils.irc.color("#%s" % data["object_attributes"]["iid"],
colors.COLOR_ID)
action = data["object_attributes"]["action"]
action_desc = "%s %s" % (action, number)
branch = data["object_attributes"]["target_branch"]
colored_branch = utils.irc.color(branch, colors.COLOR_BRANCH)
if action == "open":
action_desc = "requested %s merge into %s" % (number,
colored_branch)
elif action == "close":
if data["pull_request"]["merged"]:
action_desc = "%s %s into %s" % (
utils.irc.color("merged", colors.COLOR_POSITIVE), number,
colored_branch)
else:
action_desc = "%s %s" % (
utils.irc.color("closed", colors.COLOR_NEGATIVE), number)
pr_title = data["object_attributes"]["title"]
author = utils.irc.bold(data["user"]["username"])
url = data["object_attributes"]["url"]
return ["[PR] %s %s: %s - %s" % (
author, action_desc, pr_title, url)]
def issues(self, full_name, data):
number = utils.irc.color("#%s" % data["object_attributes"]["iid"],
colors.COLOR_ID)
action = data["object_attributes"]["action"]
issue_title = data["object_attributes"]["title"]
author = utils.irc.bold(data["user"]["username"])
url = data["object_attributes"]["url"]
return ["[issue] %s %s %s: %s - %s" %
(author, action, number, issue_title, url)]
def note(self, full_name, data):
type = data["object_attributes"]["noteable_type"]
if type in ["Issue", "MergeRequest"]:
return self.issue_comment(full_name, data)
def issue_note(self, full_name, data):
number = utils.irc.color("#%s" % data["object_attributes"]["iid"],
colors.COLOR_ID)
type = data["object_attributes"]["noteable_type"]
type == "issue" if type == "Issue" else "PR"
issue_title = data["issue"]["title"]
commenter = utils.irc.bold(data["user"]["username"])
url = data["object_attributes"]["url"]
return ["[%s] %s commented on %s: %s - %s" %
(type, commenter, number, issue_title, url)]