gitlab2feishu.py

# MIT License
#
# Copyright (c) 2024 GiraKoo
#
# Permission is hereby granted, free of charge, to any person obtaining a copy
# of this software and associated documentation files (the "Software"), to deal
# in the Software without restriction, including without limitation the rights
# to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
# copies of the Software, and to permit persons to whom the Software is
# furnished to do so, subject to the following conditions:
#
# The above copyright notice and this permission notice shall be included in all
# copies or substantial portions of the Software.
#
# THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
# IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
# FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
# AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
# LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
# OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
# SOFTWARE.

# 作者声明
# 作者: GiraKoo
# 联系方式: girakoo@163.com
# 项目地址:
# https://github.com/i-zmj/gitlab2feishu
# https://gitee.com/izmj/gitlab2feishu

import json
import os
import re
import requests


# 将日志输出到文件
def file_log_print(msg):
    import datetime

    date = datetime.datetime.now().strftime("%Y-%m-%d")
    log_file = f"logs/{date}.log"
    if not os.path.exists("logs"):
        os.makedirs("logs", exist_ok=True)

    time = datetime.datetime.now().strftime("%H:%M:%S")
    print(f"[{time}]{msg}")

    with open(log_file, "a") as f:
        f.write(f"[{time}]{msg}\n")


# 读取配置文件
def read_config():
    if not os.path.isfile("config.json"):
        file_log_print("Config file not found.")
        exit()

    # 读取config.json文件
    with open("config.json", "r") as f:
        try:
            config = json.load(f)
        except ValueError:
            file_log_print("Invalid JSON format in config.json")
            raise
    return config


def notify_feishu_text(webhook, msg):
    # 构建飞书消息
    feishu_message = {"msg_type": "text", "content": {"text": msg}}

    print("\nSend:")
    print(feishu_message)

    # 将消息发送到飞书机器人
    return requests.post(webhook, json=feishu_message)
    # return


def notify_feishu_interactive(
    webhook,
    mr_template_id,
    mr_template_ver,
    mr_title_bg,
    mr_action,
    mr_title,
    mr_brief,
    mr_gitlab_url,
    mr_jira_url
):

    # 构建飞书消息
    feishu_message = {
        "msg_type": "interactive",
        "card": {
            "type": "template",
            "data": {
                "template_id": mr_template_id,
                "template_version_name": mr_template_ver,
                "template_variable": {
                    "mr_action": mr_action,
                    "mr_title": mr_title,
                    "mr_brief": mr_brief,
                    "mr_gitlab_url": mr_gitlab_url,
                    "mr_jira_url": mr_jira_url,
                    "mr_title_bg": mr_title_bg,
                    "mr_jira_disable": (mr_jira_url == '')
                },
            },
        },
    }

    print("\nSend:")
    print(feishu_message)

    # 将消息发送到飞书机器人
    return requests.post(webhook, json=feishu_message)
    # return ""


def get_change_content(changes):
    # 提取变更内容
    message = "变更: \n"

    if "labels" in changes.keys():
        message += "- 标签: "
        if "previous" in changes["labels"].keys():
            for label in changes["labels"]["previous"]:
                if label != None:
                    message += f"{label['title']},"
                else:
                    message += "无,"
            message = message[:-1]
        else:
            message += "无"
        message += " --> "

        if "current" in changes["labels"].keys():
            for label in changes["labels"]["current"]:
                if label != None:
                    message += f"{label['title']},"
                else:
                    message += "无,"
            message = message[:-1]
        else:
            message += "无"
        message += "\n"

    if "target_branch" in changes.keys():
        message += "- 分支: "
        if "previous" in changes["target_branch"].keys():
            if changes["target_branch"]["previous"] != None:
                message += f"{changes['target_branch']['previous']}"
            else:
                message += "无"
        else:
            message += "无"
        message += " --> "

        if "current" in changes["target_branch"].keys():
            if changes["target_branch"]["current"] != None:
                message += f"{changes['target_branch']['current']}"
            else:
                message += "无"
        else:
            message += "无"
        message += "\n"

    if "assignees" in changes.keys():
        message += "- 指派: "
        if "previous" in changes["assignees"].keys():
            for assignee in changes["assignees"]["previous"]:
                if assignee != None:
                    message += f"{assignee['name']},"
                else:
                    message += "无,"
            message = message[:-1]
        else:
            message += "无"
        message += " --> "

        if "current" in changes["assignees"].keys():
            for assignee in changes["assignees"]["current"]:
                if assignee != None:
                    message += f"{assignee['name']},"
                else:
                    message += "无,"
            message = message[:-1]
        else:
            message += "无"
        message += "\n"

    return message


def get_change_content_interactive(config, changes):
    # 提取变更内容
    change_list = []

    if "labels" in changes.keys():
        message = "- **标签**: "
        if (
            "previous" in changes["labels"].keys()
            and len(changes["labels"]["previous"]) > 0
        ):
            message += "".join(
                f"<text_tag color=\"{get_label_color(config, label['title'])}\">{label['title']}</text_tag>"
                for label in changes["labels"]["previous"]
            )
        else:
            message += "无"
        message += " --> "
        if (
            "current" in changes["labels"].keys()
            and len(changes["labels"]["current"]) > 0
        ):
            message += "".join(
                f"<text_tag color=\"{get_label_color(config, label['title'])}\">{label['title']}</text_tag>"
                for label in changes["labels"]["current"]
            )
        else:
            message += "无"
        change_list.append(message)

    if "target_branch" in changes.keys():
        message = "- **分支**: "
        if (
            "previous" in changes["target_branch"].keys()
            and changes["target_branch"]["previous"] != None
        ):
            message += changes["target_branch"]["previous"]
        else:
            message += "无"
        message += " --> "
        if (
            "current" in changes["target_branch"].keys()
            and changes["target_branch"]["current"] != None
        ):
            message += changes["target_branch"]["current"]
        else:
            message += "无"
        change_list.append(message)

    if "assignees" in changes.keys():
        message = "- **指派**: "
        if (
            "previous" in changes["assignees"].keys()
            and len(changes["assignees"]["previous"]) > 0
        ):
            message += " | ".join(
                assignee["name"] for assignee in changes["assignees"]["previous"]
            )
        else:
            message += "无"
        message += " --> "
        if (
            "current" in changes["assignees"].keys()
            and len(changes["assignees"]["current"]) > 0
        ):
            message += " | ".join(
                assignee["name"] for assignee in changes["assignees"]["current"]
            )
        else:
            message += "无"
        change_list.append(message)

    if "state" in changes.keys():
        message = "- **状态**: "
        if (
            "previous" in changes["state"].keys()
            and changes["state"]["previous"] != None
        ):
            message += changes["state"]["previous"]
        else:
            message += "无"
        message += " --> "
        if "current" in changes["state"].keys() and changes["state"]["current"] != None:
            message += changes["state"]["current"]
        else:
            message += "无"
        change_list.append(message)

    return change_list


def notify_merge_request(config, gitlab_data):

    # 读取基本filter
    target_branch_filter = config["gitlab"]["merge_request"]["target_branch_filter"]
    change_filter = config["gitlab"]["merge_request"]["change_filter"]

    # 读取gitlab基本信息
    username = gitlab_data["user"]["name"]
    merge_request_id = gitlab_data["object_attributes"]["iid"]
    project_name = gitlab_data["project"]["name"]
    target_branch = gitlab_data["object_attributes"]["target_branch"]
    title = gitlab_data["object_attributes"]["title"]
    assignees = ""
    if "assignees" in gitlab_data.keys():
        assignees = gitlab_data["assignees"]

    changes = gitlab_data["changes"]
    url = gitlab_data["object_attributes"]["url"]
    action = gitlab_data["object_attributes"]["action"]
    if action == "open":
        action = "[创建]"
    elif action == "update":
        action = "[更新]"
    elif action == "reopen":
        action = "[重开]"
    elif action == "approval":
        action = "[接受]"
    elif action == "unapprove":
        action = "[拒绝]"
    elif action == "approval":
        action = "[批准]"
    elif action == "unapproval":
        action = "[取消]"
    elif action == "close":
        action = "[关闭]"
    elif action == "merge":
        action = "[合并]"
    else:
        action = "[未知]"
    labels = gitlab_data["labels"]
    last_commit_user = f"{gitlab_data['object_attributes']['last_commit']['author']['name']}({gitlab_data['object_attributes']['last_commit']['author']['email']})"

    if len(target_branch_filter) > 0:
        if not target_branch in target_branch_filter:
            file_log_print(
                f"[warn] Target branch[{target_branch}] do not match{target_branch_filter}."
            )
            return

    # 判断changes中的元素key值是否在change_filter中
    if len(change_filter) > 0:
        if not any(change in change_filter for change in changes):
            file_log_print(f"[warn] Change filter do not match.")
            return

    # 构建飞书消息
    message = f"{action} {title}\n"
    message += "-----------\n"
    message += f"提交人: {username}\n"

    if len(assignees) > 0:
        message += "指派给: "
        for assignee in assignees:
            message += f"{assignee['name']},"
        message = message[:-1] + "\n"  # 去掉最后一个逗号
    message += f"最新提交人: {last_commit_user}\n"
    message += f"链接: {url}\n"
    message += f"分支: {project_name}({target_branch})\n"

    # 遍历所有的labels数组，将[labels]的[title]都添加到$message中
    if len(labels) > 0:
        message += "标签: "
        for label in labels:
            message += f"{label['title']},"
        message = message[:-1] + "\n"  # 去掉最后一个逗号
    message += "-----------\n"

    message += get_change_content(changes)

    # 发送普通消息
    response = notify_feishu_text(config["feishu"]["webhook"], message)

    if response != None:
        http_code = response.status_code
        file_log_print(f"[info] feishu response: {http_code}")


def check_change_add_new_label(changes):
    # 如果current的label全部都在previous中，则说明是删除label
    if "labels" in changes.keys():
        if (
            "current" in changes["labels"].keys()
            and len(changes["labels"]["current"]) > 0
            and len(changes["labels"]["previous"]) > 0
        ):
            current_labels = [label["title"] for label in changes["labels"]["current"]]
            previous_labels = [
                label["title"] for label in changes["labels"]["previous"]
            ]
            print("current_labels: ", current_labels)
            print("previous_labels: ", previous_labels)
            return all(item in current_labels for item in previous_labels)
    return False

def get_label_color(config, label):
    if label in config['gitlab']['merge_request']['label_bg_color']:
        return config['gitlab']['merge_request']['label_bg_color'][label]
    return 'neutral'

def notify_merge_request_interactive(config, gitlab_data):

    # 读取基本filter
    target_branch_filter = config["gitlab"]["merge_request"]["target_branch_filter"]
    change_filter = config["gitlab"]["merge_request"]["change_filter"]

    # 读取gitlab基本信息
    username = f"<at email={gitlab_data['user']['username']}{config['gitlab']['default_mail_postfix']}></at>"
    merge_request_id = gitlab_data["object_attributes"]["iid"]
    project_name = gitlab_data["project"]["name"]
    target_branch = gitlab_data["object_attributes"]["target_branch"]
    title = gitlab_data["object_attributes"]["title"]

    # 从title中，尝试提取以[字母-数字]格式的JIRA编号
    jira_base_url = config["feishu"]["mr_jira_base_url"]
    default_jira_url = ""
    jira_matches = []
    try:
        matches = re.findall(r"\[([^\]]+)\]", title)
        # 提取第2个方括号内的数据
        if len(matches) >= 2:
            result_matches = matches[1].split("/")

            # 判定jira_matches中的结果是否符合字母-数字的格式
            for item in result_matches:
                if re.match(r"\w+-\d+", item):
                    jira_matches.append(item)

        if len(jira_matches) > 0:
            default_jira_url = jira_base_url + jira_matches[0]

    except AttributeError:
        file_log_print("[error] Invalid MR Title")

    assignees = ""
    if "assignees" in gitlab_data.keys():
        assignees = gitlab_data["assignees"]

    changes = gitlab_data["changes"]
    url = gitlab_data["object_attributes"]["url"]
    action = gitlab_data["object_attributes"]["action"]
    if action == "open":
        action = "创建"
    elif action == "update":
        action = "更新"
    elif action == "reopen":
        action = "重开"
    elif action == "approval":
        action = "接受"
    elif action == "unapprove":
        action = "拒绝"
    elif action == "approval":
        action = "批准"
    elif action == "unapproval":
        action = "取消"
    elif action == "close":
        action = "关闭"
    elif action == "merge":
        action = "合并"
    else:
        action = "未知"
    labels = gitlab_data["labels"]
    last_commit_user = f"<at email={gitlab_data['object_attributes']['last_commit']['author']['email']}></at>"

    if len(target_branch_filter) > 0:
        if not target_branch in target_branch_filter:
            file_log_print(
                f"[warn] Target branch[{target_branch}] do not match{target_branch_filter}."
            )
            return

    # 判断changes中的元素key值是否在change_filter中
    if len(change_filter) > 0:
        if not any(change in change_filter for change in changes):
            file_log_print(f"[warn] Change filter do not match.")
            return

    # 合并是绿色,创建是红色,其他是黄色
    if action == "合并":
        title_bg = "green"
    elif action == "创建":
        title_bg = "red"
    else:
        title_bg = "yellow"

    if (
        "add_new_label" in config["gitlab"]["merge_request"]["skip"]
    ) and check_change_add_new_label(changes):
        file_log_print("[warn] Skip add new label")
        return

    if ("no_assignee" in config["gitlab"]["merge_request"]["skip"]) and len(
        assignees
    ) == 0:
        file_log_print("[warn] Skip no assignee")
        return

    if ("no_label" in config["gitlab"]["merge_request"]["skip"]) and len(labels) == 0:
        file_log_print("[warn] Skip no label")
        return

    message_content = "\n".join(
        [
            ": ".join(["**提交人**", username]),
            ": ".join(
                [
                    "**指派给**",
                    ", ".join(f"<at email={assignee['username']}{config['gitlab']['default_mail_postfix']}></at>" for assignee in assignees),
                ]
            ),
            ": ".join(["**最新提交人**", last_commit_user]),
            ": ".join(["**分支**", "".join([target_branch, "(", project_name, ")"])]),
            ": ".join(
                [
                    "**标签**",
                    "".join(
                        f"<text_tag color=\"{get_label_color(config, label['title'])}\">{label['title']}</text_tag>"
                        for label in labels
                    ),
                ]
            ),
        ]
    )
    message_content += "\n"

    if action != "合并":
        if 'oldrev' in gitlab_data['object_attributes']:
            file_log_print("[warn] Skip add force push")
            return
        else:
            # 在合并通知中，变更内容会出现一些奇怪的现象。
            # 例如：Asignee的变更和label的变更。暂时移除
            message_content += ": \n".join(
                ["**变更**", "\n".join(get_change_content_interactive(config, changes))]
            )
        message_content += "\n"

    if action == "合并":
        message_content += "<hr/>**检测到正式合并通知，恭喜！**\n"
        message_content += "请及时更新 [ReleaseNote](https://navinfo.feishu.cn/wiki/QPoawNOsoi7TDqkomCwcq3PYnYc)"
        if len(jira_matches) > 0:
            for jira in jira_matches:
                message_content += "\nJIRA: " + jira_base_url + jira
        else:
            message_content += "\nJIRA: 未检测到符合命名规则的名称"

    # 发送普通消息
    response = notify_feishu_interactive(
        config["feishu"]["webhook"],
        config["feishu"]["mr_message_template_id"],
        config["feishu"]["mr_message_template_ver"],
        title_bg,
        action,
        title,
        message_content,
        url,
        default_jira_url,
    )

    if response != None:
        http_code = response.status_code
        file_log_print(f"[info] feishu response: {http_code}")


def notify_note(config, gitlab_data):

    target_branch_filter = config["gitlab"]["merge_request"]["target_branch_filter"]

    username = gitlab_data["user"]["name"]
    # merge_request_id = gitlab_data['merge_request']['iid']
    # project_name = gitlab_data['project']['name']
    # target_branch = gitlab_data['merge_request']['target_branch']./
    title = gitlab_data["merge_request"]["title"]
    url = gitlab_data["merge_request"]["url"]
    note = gitlab_data["object_attributes"]["note"]
    last_commit_user = gitlab_data["merge_request"]["last_commit"]["author"]["name"]

    # 构建飞书消息
    message = f"[评论] {title}\n"
    message += "-----------\n"
    message += f"评论人: {username}\n"
    message += f"最新提交人: {last_commit_user}\n"
    at_users = ""
    for match in re.finditer(r"@(\w+)", note):
        user = match.group(1)
        if user not in at_users:
            at_users += f"{user},"
    if len(at_users) > 0:
        message += "提及人: "
        for at_user in at_users:
            message += f"{at_user['name']},"
        message = message[:-1] + "\n"  # 去掉最后一个逗号
    message += f"URL: {url}\n"
    message += "-----------\n"
    message += f"评论: \n{note}"

    # 发送普通消息
    response = notify_feishu_text(config["feishu"]["webhook"], message)
    http_code = response.status_code
    file_log_print(f"[info] feishu response: {http_code}")

    print(message)


def notify_changelog(webhook):
    message = "[公告][Gitlab2Feishu] 版本更新通知: v1.0.3\n"
    message += "-----------\n"
    message += "变更内容:\n"
    message += "- [feishu] 没有标签，指派人的通知将不发送\n"
    message += "- [feishu] 标签从1个变更2个，且旧标签包含在新标签的情况下，将不发送。\n"
    message += "- [feishu] 卡片背景支持合并（绿色），创建（红色），其他（黄色）\n"
    message += "- [feishu] 监测到合并通知时，将额外发送ReleaseNote和JIRA链接。\n"

    notify_feishu_text(webhook, message)


if __name__ == "__main__":
    config = read_config()

    import sys

    if "--notify-changelog" in sys.argv:
        notify_changelog(config["feishu"]["webhook"])
        exit(0)

    # 获取GitLab的Webhook通知
    in_data = input()
    gitlab_data = json.loads(in_data)
    file_log_print(in_data)

    # 判断通知类型
    if gitlab_data["object_kind"] in config["gitlab"]["object_kind_filter"]:
        if "merge_request" == gitlab_data["object_kind"]:
            if config["feishu"]["mr_message_type"] == "template":
                notify_merge_request_interactive(config, gitlab_data)
            else:
                notify_merge_request(config, gitlab_data)
        elif "note" == gitlab_data["object_kind"]:
            notify_note(config, gitlab_data)
        else:
            file_log_print(
                f"[info] Not supported object_kind[{gitlab_data['object_kind']}]."
            )
    else:
        file_log_print(
            f"[info] Not supported object_kind[{gitlab_data['object_kind']}]."
        )