GitBot/webhook_gitlab.py

"""GitLab webhook payload parser."""

from irc_format import (
    color, bold,
    COLOR_BRANCH, COLOR_ID, COLOR_POSITIVE, COLOR_NEGATIVE,
)

EVENT_CATEGORIES = {
    "ping":  ["ping"],
    "code":  ["push"],
    "pr-minimal": [
        "merge_request/open", "merge_request/close",
        "merge_request/reopen", "merge_request/merge",
    ],
    "pr": [
        "merge_request/open", "merge_request/close",
        "merge_request/reopen", "merge_request/update", "merge_request/merge",
        "note+mergerequest", "confidential_note+mergerequest",
    ],
    "pr-all": ["merge_request", "note+mergerequest", "confidential_note+mergerequest"],
    "issue-minimal": [
        "issue/open", "issue/close", "issue/reopen",
        "confidential_issue/open", "confidential_issue/close", "confidential_issue/reopen",
    ],
    "issue": [
        "issue/open", "issue/close", "issue/reopen", "issue/update",
        "confidential_issue/open", "confidential_issue/close",
        "confidential_issue/reopen", "confidential_issue/update",
        "note+issue", "confidential_note+issue",
    ],
    "issue-all": ["issue", "confidential_issue", "note+issue", "confidential_note+issue"],
    "repo": ["tag_push"],
}

ISSUE_ACTIONS = {
    "open": "opened", "close": "closed",
    "reopen": "reopened", "update": "updated", "merge": "merged",
}
WIKI_ACTIONS = {
    "create": "created", "update": "updated", "delete": "deleted",
}


def _short(h):
    return h[:7]


def names(data, headers):
    if "project" in data:
        full_name = data["project"]["path_with_namespace"]
    else:
        full_name = data.get("project_name", "").replace(" ", "")
    parts = full_name.split("/", 1)
    repo_user = parts[0] if parts else ""
    repo_name = parts[1] if len(parts) > 1 else ""
    organisation = None
    if full_name.count("/") == 2:
        organisation = repo_user
        repo_user = full_name.rsplit("/", 1)[0]
    return full_name, repo_user, repo_name, organisation


def branch(data, headers):
    if "ref" in data:
        return data["ref"].rpartition("/")[2]
    return None


def is_private(data, headers):
    project = data.get("project", {})
    return project.get("visibility_level", 0) != 20


def event(data, headers):
    ev = headers.get("X-GitLab-Event", "").rsplit(" ", 1)[0].lower().replace(" ", "_")
    action = None
    category = None
    oa = data.get("object_attributes", {})
    if "action" in oa:
        action = oa["action"]
    if "noteable_type" in oa:
        nt = oa["noteable_type"].lower()
        category = f"{ev}+{nt}"
    parts = [ev]
    if action:
        parts.append(f"{ev}/{action}")
    if category:
        parts.append(category)
        if action:
            parts.append(f"{category}/{action}")
    return parts


def event_categories(ev):
    return EVENT_CATEGORIES.get(ev, [ev])


def parse(full_name, ev, data, headers, commit_limit=3):
    dispatch = {
        "push": lambda fn, d: _push(fn, d, commit_limit),
        "tag_push":          _tag_push,
        "merge_request":     _merge_request,
        "issue":             _issues,
        "confidential_issue": _issues,
        "note":              _note,
        "confidential_note": _note,
        "wiki_page":         lambda fn, d: _wiki(d),
    }
    fn = dispatch.get(ev)
    if fn:
        return fn(full_name, data)
    return []


def _push(full_name, data, commit_limit=3):
    branch_str = color(data["ref"].rpartition("/")[2], COLOR_BRANCH)
    author     = bold(data["user_username"])
    commits    = data.get("commits", [])
    n          = len(commits)

    if not commits:
        return [(f"{author} pushed to {branch_str}", None)]

    # Single commit: one clean line
    if n == 1:
        c   = commits[0]
        h   = color(_short(c["id"]), COLOR_ID)
        msg = c["message"].split("\n")[0].strip()
        return [(f"{author} pushed {h} to {branch_str}: {msg}", c.get("url"))]

    # Multiple commits (GitLab has no compare URL in push payloads)
    outputs = [(f"{author} pushed {n} commits to {branch_str}", None)]
    shown   = commits[:commit_limit]
    for c in shown:
        msg = c["message"].split("\n")[0].strip()
        outputs.append((f"{author} {_short(c['id'])} - {msg}", c.get("url")))
    hidden = n - len(shown)
    if hidden > 0:
        outputs.append((f"(+{hidden} hidden commit{'s' if hidden != 1 else ''})", None))
    return outputs


def _tag_push(full_name, data):
    after = data.get("after", "")
    create = bool(after.strip("0"))
    tag = color(data["ref"].rsplit("/", 1)[-1], COLOR_BRANCH)
    author = bold(data["user_username"])
    action = "created" if create else "deleted"
    return [(f"{author} {action} a tag: {tag}", None)]


def _merge_request(full_name, data):
    oa = data["object_attributes"]
    num = color(f"!{oa['iid']}", COLOR_ID)
    action = oa["action"]
    branch_str = color(oa["target_branch"], COLOR_BRANCH)
    author = bold(data["user"]["username"])
    title = oa["title"]
    url = oa["url"]

    if action == "open":
        desc = f"requested {num} merge into {branch_str}"
    elif action == "close":
        desc = f"{color('closed', COLOR_NEGATIVE)} {num}"
    elif action == "merge":
        desc = f"{color('merged', COLOR_POSITIVE)} {num} into {branch_str}"
    else:
        desc = f"{ISSUE_ACTIONS.get(action, action)} {num}"

    return [(f"[MR] {author} {desc}: {title}", url)]


def _issues(full_name, data):
    oa = data["object_attributes"]
    if "action" not in oa:
        return []
    num = color(f"#{oa['iid']}", COLOR_ID)
    action = ISSUE_ACTIONS.get(oa["action"], oa["action"])
    title = oa["title"]
    author = bold(data["user"]["username"])
    url = oa["url"]
    return [(f"[issue] {author} {action} {num}: {title}", url)]


def _note(full_name, data):
    oa = data["object_attributes"]
    type_ = oa.get("noteable_type", "")
    if type_ == "Issue" and "issue" in data:
        obj = data["issue"]
        label = "issue"
    elif type_ == "MergeRequest" and "merge_request" in data:
        obj = data["merge_request"]
        label = "MR"
    else:
        return []
    num = color(f"#{obj['iid']}", COLOR_ID)
    title = obj["title"]
    commenter = bold(data["user"]["username"])
    url = oa["url"]
    return [(f"[{label}] {commenter} commented on {num}: {title}", url)]


def _wiki(data):
    oa = data["object_attributes"]
    author = bold(data["user"]["username"])
    action = WIKI_ACTIONS.get(oa["action"], oa["action"])
    title = oa["title"]
    url = oa["url"]
    return [(f"{author} {action} a wiki page: {title}", url)]