feat(reply): add reply module, implement reply factory and add Gitlab and Dingtalk reply handlers

Author: Robbings

reply_module/__init__.py

@@ -0,0 +1,55 @@
+import threading
+
+from reply_module.reply_target.reply_factory import ReplyFactory
+
+
+class Reply:
+    def __init__(self, project_id, merge_request_id):
+        self.replies = []
+        self.lock = threading.Lock()
+        self.project_id = project_id
+        self.merge_request_id = merge_request_id
+
+    def add_reply(self, reply):
+        # reply 格式检查：title, content 必选
+        if 'title' not in reply or 'content' not in reply:
+            raise Exception('Reply format error, title and content are required.')
+        if 'priority' in reply:
+            if not isinstance(reply['priority'], int):
+                raise Exception('Reply format error, priority should be an integer.')
+            elif reply['priority'] == 0:
+                self.send_single_message(reply)
+                return
+
+        with self.lock:  # 加锁
+            self.replies.append(reply)
+
+    def send(self):
+        markdown_message = ""
+        with self.lock:  # 加锁
+            # 发送所有消息的逻辑
+            for reply in self.replies:
+                markdown_message += f"## {reply['title']}\n\n{reply['content']}\n\n"
+            self.replies = []  # 清空已发送的消息
+        reply_target = ReplyFactory.get_reply_instance(reply['target'], self.project_id, self.merge_request_id)
+        return reply_target.send(markdown_message)
+
+    def send_single_message(self, reply):
+        """
+        实时发送消息
+        """
+        reply_target = ReplyFactory.get_reply_instance(reply['target'], self.project_id, self.merge_request_id)
+        return reply_target.send(reply['content'])
+
+
+if __name__ == '__main__':
+    reply = Reply(9885, 18)
+    threads = []
+    for i in range(10):
+        threads.append(threading.Thread(target=reply.add_reply, args=(
+            {'title': f'title{i}', 'content': f'content{i}', 'target': 'gitlab', 'priority': i % 3},)))
+    for thread in threads:
+        thread.start()
+    for thread in threads:
+        thread.join()
+    reply.send()

reply_module/reply.py

@@ -0,0 +1,21 @@
+from abc import ABC, abstractmethod
+
+class AbstractReply(ABC):
+    @abstractmethod
+    def __init__(self, project_id, merge_request_id):
+        self.project_id = project_id
+        self.merge_request_id = merge_request_id
+
+    @abstractmethod
+    def send(self, message):
+        pass
+
+    # # 发送失败调用
+    # @abstractmethod
+    # def send_failed(self, message):
+    #     pass
+    #
+    # # 发送成功调用
+    # @abstractmethod
+    # def send_success(self, message):
+    #     pass

reply_module/reply_target/__init__.py

@@ -0,0 +1,103 @@
+import base64
+import hashlib
+import hmac
+import time
+import urllib
+import requests
+import json
+from config.config import *
+from utils.logger import *
+from reply_module.reply_target.abstract_reply import AbstractReply
+
+
+class DingtalkReply(AbstractReply):
+    def __init__(self, project_id, merge_request_id):
+        super().__init__(project_id, merge_request_id)
+
+    def send(self, message):
+        return self.send_dingtalk_message_by_sign(message)
+
+    def send_dingtalk_message_by_sign(self, message_text):
+        """
+        使用签名方式发送消息通知到钉钉群
+
+        Args:
+            message_text (str): 消息文本内容
+
+        Returns:
+            bool: 消息是否发送成功
+        """
+        timestamp = str(round(time.time() * 1000))
+        sign = self.__get_sign(timestamp)
+        webhookurl = f"{dingding_bot_webhook}&timestamp={timestamp}&sign={sign}"
+        # 构建请求头
+        headers = {
+            "Content-Type": "application/json",
+        }
+
+        # 构建请求体
+        message = {
+            "msgtype": "markdown",
+            "markdown": {
+                "title": "Gitlab 通知",
+                "text": message_text
+            },
+            "timestamp": timestamp,
+            "sign": sign
+        }
+
+        # 发送HTTP POST请求
+        response = requests.post(
+            webhookurl,
+            headers=headers,
+            data=json.dumps(message)
+        )
+
+        # 检查响应
+        if response.status_code == 200 and response.json()["errcode"] == 0:
+            log.info(f"评论信息发送成功：project_id:{self.project_id}  merge_request_id:{self.merge_request_id}")
+            return True
+        else:
+            log.error(
+                f"评论信息发送失败：project_id:{self.project_id}  merge_request_id:{self.merge_request_id} response:{response}")
+            return False
+
+    def send_dingtalk_message_by_key_word(self, project_url):
+        """
+        通过关键词发送
+
+        """
+        # 设置钉钉机器人的 Webhook URL
+        webhook_url = dingding_bot_webhook
+
+        # 要发送的消息内容
+        message = f"新工程接入\nurl：{project_url}"
+
+        headers = {"Content-Type": "application/json"}
+        payload = {
+            "msgtype": "text",
+            "text": {
+                "content": message
+            }
+        }
+        response = requests.post(webhook_url, headers=headers, data=json.dumps(payload))
+        return response.json()
+
+    def __get_sign(self, timestamp):
+        '''
+        计算签名
+        :param timestamp: 时间戳
+        :return: 签名
+        '''
+
+        secret = dingding_secret
+        secret_enc = secret.encode('utf-8')
+        string_to_sign = '{}\n{}'.format(timestamp, secret)
+        string_to_sign_enc = string_to_sign.encode('utf-8')
+        hmac_code = hmac.new(secret_enc, string_to_sign_enc, digestmod=hashlib.sha256).digest()
+        sign = urllib.parse.quote_plus(base64.b64encode(hmac_code))
+        return sign
+
+if __name__ == '__main__':
+    dingtalk = DingtalkReply(1, 1)
+    dingtalk.send("test message")

reply_module/reply_target/abstract_reply.py

@@ -0,0 +1,37 @@
+import requests
+from retrying import retry
+from config.config import *
+from reply_module.reply_target.abstract_reply import AbstractReply
+from utils.logger import log
+
+# 继承AbstractReply类，实现send方法
+class GitlabReply(AbstractReply):
+    def __init__(self, project_id, merge_request_id):
+        super().__init__(project_id, merge_request_id)
+
+    @retry(stop_max_attempt_number=3, wait_fixed=2000)
+    def send(self, message):
+        headers = {
+            "Private-Token": gitlab_private_token,
+            "Content-Type": "application/json"
+        }
+        project_id = self.project_id
+        merge_request_id = self.merge_request_id
+        url = f"{gitlab_server_url}/api/v4/projects/{project_id}/merge_requests/{merge_request_id}/notes"
+        data = {
+            "body": message
+        }
+
+        response = requests.post(url, headers=headers, json=data)
+
+        if response.status_code == 201:
+            log.info(f"评论信息发送成功：project_id:{project_id}  merge_request_id:{merge_request_id}")
+            return True
+        else:
+            log.error(
+                f"评论信息发送成功：project_id:{project_id}  merge_request_id:{merge_request_id} response:{response}")
+            return False
+
+if __name__ == '__main__':
+    gitlab_reply = GitlabReply(9885, 18)
+    gitlab_reply.send("test")

reply_module/reply_target/dingtalk_reply.py

@@ -0,0 +1,13 @@
+from reply_module.reply_target.dingtalk_reply import DingtalkReply
+from reply_module.reply_target.gitlab_reply import GitlabReply
+
+
+class ReplyFactory:
+    @staticmethod
+    def get_reply_instance(target, project_id, merge_request_id):
+        if target == 'gitlab':
+            return GitlabReply(project_id, merge_request_id)
+        elif target == 'dingtalk':
+            return DingtalkReply(project_id, merge_request_id)
+        else:
+            raise ValueError(f"Unknown target: {target}")