优化消息通知代码

Author: sunminghui

api.py

@@ -1,6 +1,7 @@
 import atexit
 import json
 import os
+import re
 import traceback
 from datetime import datetime
 from multiprocessing import Process
@@ -117,16 +118,16 @@ def handle_webhook():
         if not gitlab_url:
             repository = data.get('repository')
             if not repository:
-                return jsonify({'message': 'Missing GitLab URL'}), 400 
+                return jsonify({'message': 'Missing GitLab URL'}), 400
             homepage = repository.get("homepage")
             if not homepage:
-                return jsonify({'message': 'Missing GitLab URL'}), 400 
+                return jsonify({'message': 'Missing GitLab URL'}), 400
             try:
                 parsed_url = urlparse(homepage)
                 gitlab_url = f"{parsed_url.scheme}://{parsed_url.netloc}/"
             except Exception as e:
                 return jsonify({"error": f"Failed to parse homepage URL: {str(e)}"}), 400
-                
+
         # 优先从环境变量获取，如果没有，则从请求头获取
         gitlab_token = os.getenv('GITLAB_ACCESS_TOKEN') or request.headers.get('X-Gitlab-Token')
         # 如果gitlab_token为空，返回错误
@@ -143,34 +144,40 @@ def handle_webhook():
             process = Process(target=__handle_merge_request_event, args=(data, gitlab_token, gitlab_url))
             process.start()
             # 立马返回响应
-            return jsonify({'message': f'Request received(object_kind={object_kind}), will process asynchronously.'}), 200
+            return jsonify(
+                {'message': f'Request received(object_kind={object_kind}), will process asynchronously.'}), 200
         elif object_kind == "push":
             # 创建一个新进程进行异步处理
             process = Process(target=__handle_push_event, args=(data, gitlab_token, gitlab_url))
             process.start()
             # 立马返回响应
-            return jsonify({'message': f'Request received(object_kind={object_kind}), will process asynchronously.'}), 200
+            return jsonify(
+                {'message': f'Request received(object_kind={object_kind}), will process asynchronously.'}), 200
         else:
             error_message = f'Only merge_request and push events are supported (both Webhook and System Hook), but received: {object_kind}.'
             logger.error(error_message)
             return jsonify(error_message), 400
     else:
         return jsonify({'message': 'Invalid data format'}), 400
 
-def transform_gitlab_url(url):
-    # 去掉 http:// 或 https://
-    if url.startswith("http://"):
-        url = url[len("http://"):]
-    elif url.startswith("https://"):
-        url = url[len("https://"):]
-
-    if url.endswith('/'):
-        url = url[:-1]
-        
-    # 替换 . 和 / 为 _
-    transformed_url = url.replace('.', '_').replace('/', '_')
-    
-    return transformed_url
+
+def slugify_url(original_url: str) -> str:
+    """
+    将原始URL转换为适合作为文件名的字符串，其中非字母或数字的字符会被替换为下划线，举例：
+    slugify_url("http://example.com/path/to/repo/") => example_com_path_to_repo
+    slugify_url("https://gitlab.com/user/repo.git") => gitlab_com_user_repo_git
+    """
+    # Remove URL scheme (http, https, etc.) if present
+    original_url = re.sub(r'^https?://', '', original_url)
+
+    # Replace non-alphanumeric characters (except underscore) with underscores
+    target = re.sub(r'[^a-zA-Z0-9]', '_', original_url)
+
+    # Remove trailing underscore if present
+    target = target.rstrip('_')
+
+    return target
+
 
 def __handle_push_event(webhook_data: dict, gitlab_token: str, gitlab_url: str):
     try:
@@ -200,7 +207,6 @@ def __handle_push_event(webhook_data: dict, gitlab_token: str, gitlab_url: str):
             # 将review结果提交到Gitlab的 notes
             handler.add_push_notes(f'Auto Review Result: \n{review_result}')
 
-        url_base = transform_gitlab_url(gitlab_url)
 
         event_manager['push_reviewed'].send(PushReviewEntity(
             project_name=webhook_data['project']['name'],
@@ -210,7 +216,7 @@ def __handle_push_event(webhook_data: dict, gitlab_token: str, gitlab_url: str):
             commits=commits,
             score=score,
             review_result=review_result,
-            gitlab_url = url_base,
+            gitlab_url_slug = slugify_url(gitlab_url),
         ))
 
     except Exception as e:
@@ -258,7 +264,6 @@ def __handle_merge_request_event(webhook_data: dict, gitlab_token: str, gitlab_u
             # 将review结果提交到Gitlab的 notes
             handler.add_merge_request_notes(f'Auto Review Result: \n{review_result}')
 
-            url_base = transform_gitlab_url(gitlab_url)
             # dispatch merge_request_reviewed event
             event_manager['merge_request_reviewed'].send(
                 MergeRequestReviewEntity(
@@ -271,7 +276,7 @@ def __handle_merge_request_event(webhook_data: dict, gitlab_token: str, gitlab_u
                     score=CodeReviewer.parse_review_score(review_text=review_result),
                     url=webhook_data['object_attributes']['url'],
                     review_result=review_result,
-                    gitlab_url = url_base,
+                    gitlab_url_slug = slugify_url(gitlab_url),
                 )
             )
 
@@ -283,6 +288,7 @@ def __handle_merge_request_event(webhook_data: dict, gitlab_token: str, gitlab_u
         im_notifier.send_notification(content=error_message)
         logger.error('出现未知错误: %s', error_message)
 
+
 def filter_changes(changes: list):
     '''
     过滤数据，只保留支持的文件类型以及必要的字段信息

biz/entity/review_entity.py

@@ -1,6 +1,6 @@
 class MergeRequestReviewEntity:
     def __init__(self, project_name: str, author: str, source_branch: str, target_branch: str, updated_at: int,
-                 commits: list, score: float, url: str, review_result: str, gitlab_url: str):
+                 commits: list, score: float, url: str, review_result: str, gitlab_url_slug: str):
         self.project_name = project_name
         self.author = author
         self.source_branch = source_branch
@@ -10,7 +10,7 @@ def __init__(self, project_name: str, author: str, source_branch: str, target_br
         self.score = score
         self.url = url
         self.review_result = review_result
-        self.gitlab_url = gitlab_url
+        self.gitlab_url_slug = gitlab_url_slug
 
     @property
     def commit_messages(self):
@@ -20,15 +20,15 @@ def commit_messages(self):
 
 class PushReviewEntity:
     def __init__(self, project_name: str, author: str, branch: str, updated_at: int, commits: list, score: float,
-                 review_result: str, gitlab_url: str):
+                 review_result: str, gitlab_url_slug: str):
         self.project_name = project_name
         self.author = author
         self.branch = branch
         self.updated_at = updated_at
         self.commits = commits
         self.score = score
         self.review_result = review_result
-        self.gitlab_url = gitlab_url
+        self.gitlab_url_slug = gitlab_url_slug
 
     @property
     def commit_messages(self):

biz/event/event_manager.py

@@ -32,7 +32,8 @@ def on_merge_request_reviewed(mr_review_entity: MergeRequestReviewEntity):
 {mr_review_entity.review_result}
     """
     im_notifier.send_notification(content=im_msg, msg_type='markdown', title='Merge Request Review',
-                                  project_name=mr_review_entity.project_name, url_base=mr_review_entity.gitlab_url)
+                                  project_name=mr_review_entity.project_name,
+                                  gitlab_url_slug=mr_review_entity.gitlab_url_slug)
 
     # 记录到数据库
     ReviewService().insert_mr_review_log(mr_review_entity)
@@ -58,7 +59,8 @@ def on_push_reviewed(entity: PushReviewEntity):
     if entity.review_result:
         im_msg += f"#### AI Review 结果: \n {entity.review_result}\n\n"
     im_notifier.send_notification(content=im_msg, msg_type='markdown',
-                                  title=f"{entity.project_name} Push Event", project_name=entity.project_name, url_base=entity.gitlab_url)
+                                  title=f"{entity.project_name} Push Event", project_name=entity.project_name,
+                                  gitlab_url_slug=entity.gitlab_url_slug)
 
     # 记录到数据库
     ReviewService().insert_push_review_log(entity)

biz/utils/im/dingtalk.py

@@ -16,10 +16,11 @@ def __init__(self, webhook_url=None):
         self.enabled = os.environ.get('DINGTALK_ENABLED', '0') == '1'
         self.default_webhook_url = webhook_url or os.environ.get('DINGTALK_WEBHOOK_URL')
 
-    def _get_webhook_url(self, project_name=None, url_base=None):
+    def _get_webhook_url(self, project_name=None, gitlab_url_slug=None):
         """
         获取项目对应的 Webhook URL
         :param project_name: 项目名称
+        :param gitlab_url_slug: 由 gitlab 项目的 url 转换而来的 slug
         :return: Webhook URL
         :raises ValueError: 如果未找到 Webhook URL
         """
@@ -30,17 +31,17 @@ def _get_webhook_url(self, project_name=None, url_base=None):
             else:
                 raise ValueError("未提供项目名称，且未设置默认的钉钉 Webhook URL。")
 
-        # 遍历所有环境变量（忽略大小写），找到项目对应的 Webhook URL
-        target_key = f"DINGTALK_WEBHOOK_URL_{project_name.upper()}"
-        for env_key, env_value in os.environ.items():
-            if env_key.upper() == target_key:
-                return env_value  # 找到匹配项，直接返回
-            
-        # url_base 优先级次之
-        target_key_url_base = f"WECOM_WEBHOOK_URL_{url_base.upper()}"
+        # 构造目标键
+        target_key_project = f"DINGTALK_WEBHOOK_URL_{project_name.upper()}"
+        target_key_url_slug = f"DINGTALK_WEBHOOK_URL_{gitlab_url_slug.upper()}"
+
+        # 遍历环境变量
         for env_key, env_value in os.environ.items():
-            if target_key_url_base !=None and  env_key.upper() == target_key_url_base:
-                return env_value  # 找到匹配项，直接返回
+            env_key_upper = env_key.upper()
+            if env_key_upper == target_key_project:
+                return env_value  # 找到项目名称对应的 Webhook URL，直接返回
+            if env_key_upper == target_key_url_slug:
+                return env_value  # 找到 GitLab URL 对应的 Webhook URL，直接返回
 
         # 如果未找到匹配的环境变量，降级使用全局的 Webhook URL
         if self.default_webhook_url:
@@ -49,13 +50,13 @@ def _get_webhook_url(self, project_name=None, url_base=None):
         # 如果既未找到匹配项，也没有默认值，抛出异常
         raise ValueError(f"未找到项目 '{project_name}' 对应的钉钉Webhook URL，且未设置默认的 Webhook URL。")
 
-    def send_message(self, content: str, msg_type='text', title='通知', is_at_all=False, project_name=None, url_base = None):
+    def send_message(self, content: str, msg_type='text', title='通知', is_at_all=False, project_name=None, gitlab_url_slug = None):
         if not self.enabled:
             logger.info("钉钉推送未启用")
             return
 
         try:
-            post_url = self._get_webhook_url(project_name=project_name, url_base=url_base)
+            post_url = self._get_webhook_url(project_name=project_name, gitlab_url_slug=gitlab_url_slug)
             headers = {
                 "Content-Type": "application/json",
                 "Charset": "UTF-8"

biz/utils/im/feishu.py

@@ -14,7 +14,7 @@ def __init__(self, webhook_url=None):
         self.default_webhook_url = webhook_url or os.environ.get('FEISHU_WEBHOOK_URL', '')
         self.enabled = os.environ.get('FEISHU_ENABLED', '0') == '1'
 
-    def _get_webhook_url(self, project_name=None, url_base=None):
+    def _get_webhook_url(self, project_name=None, gitlab_url_slug=None):
         """
         获取项目对应的 Webhook URL
         :param project_name: 项目名称
@@ -28,17 +28,17 @@ def _get_webhook_url(self, project_name=None, url_base=None):
             else:
                 raise ValueError("未提供项目名称，且未设置默认的 飞书 Webhook URL。")
 
-        # 遍历所有环境变量（忽略大小写），找到项目对应的 Webhook URL
-        target_key = f"FEISHU_WEBHOOK_URL_{project_name.upper()}"
-        for env_key, env_value in os.environ.items():
-            if env_key.upper() == target_key:
-                return env_value  # 找到匹配项，直接返回
-            
-        # url_base 优先级次之
-        target_key_url_base = f"WECOM_WEBHOOK_URL_{url_base.upper()}"
+        # 构造目标键
+        target_key_project = f"FEISHU_WEBHOOK_URL_{project_name.upper()}"
+        target_key_url_slug = f"FEISHU_WEBHOOK_URL_{gitlab_url_slug.upper()}"
+
+        # 遍历环境变量
         for env_key, env_value in os.environ.items():
-            if target_key_url_base !=None and  env_key.upper() == target_key_url_base:
-                return env_value  # 找到匹配项，直接返回
+            env_key_upper = env_key.upper()
+            if env_key_upper == target_key_project:
+                return env_value  # 找到项目名称对应的 Webhook URL，直接返回
+            if env_key_upper == target_key_url_slug:
+                return env_value  # 找到 GitLab URL 对应的 Webhook URL，直接返回
 
         # 如果未找到匹配的环境变量，降级使用全局的 Webhook URL
         if self.default_webhook_url:
@@ -47,7 +47,7 @@ def _get_webhook_url(self, project_name=None, url_base=None):
         # 如果既未找到匹配项，也没有默认值，抛出异常
         raise ValueError(f"未找到项目 '{project_name}' 对应的 Feishu Webhook URL，且未设置默认的 Webhook URL。")
 
-    def send_message(self, content, msg_type='text', title=None, is_at_all=False, project_name=None, url_base=None):
+    def send_message(self, content, msg_type='text', title=None, is_at_all=False, project_name=None, gitlab_url_slug=None):
         """
         发送飞书消息
         :param content: 消息内容
@@ -61,7 +61,7 @@ def send_message(self, content, msg_type='text', title=None, is_at_all=False, pr
             return
 
         try:
-            post_url = self._get_webhook_url(project_name=project_name, url_base=url_base)
+            post_url = self._get_webhook_url(project_name=project_name, gitlab_url_slug=gitlab_url_slug)
             if msg_type == 'markdown':
                 data = {
                     "msg_type": "interactive",

biz/utils/im/im_notifier.py

@@ -3,26 +3,26 @@
 from biz.utils.im.wecom import WeComNotifier
 
 
-def send_notification(content, msg_type='text', title="通知", is_at_all=False, project_name=None, url_base=None):
+def send_notification(content, msg_type='text', title="通知", is_at_all=False, project_name=None, gitlab_url_slug=None):
     """
     发送通知消息到配置的平台(钉钉和企业微信)
     :param content: 消息内容
     :param msg_type: 消息类型，支持text和markdown
     :param title: 消息标题(markdown类型时使用)
     :param is_at_all: 是否@所有人
-    :param url_base: gitlab服务器的url地址 http://www.gitlab.com 传递进来自动移除http和https，转换成 www_gitlab_com
+    :param gitlab_url_slug: 由gitlab服务器的url地址(如:http://www.gitlab.com)转换成的slug格式，如: www_gitlab_com
     """
     # 钉钉推送
     notifier = DingTalkNotifier()
     notifier.send_message(content=content, msg_type=msg_type, title=title, is_at_all=is_at_all,
-                          project_name=project_name, url_base=url_base)
+                          project_name=project_name, gitlab_url_slug=gitlab_url_slug)
 
     # 企业微信推送
     wecom_notifier = WeComNotifier()
     wecom_notifier.send_message(content=content, msg_type=msg_type, title=title, is_at_all=is_at_all,
-                                project_name=project_name, url_base=url_base)
+                                project_name=project_name, gitlab_url_slug=gitlab_url_slug)
 
     # 飞书推送
     feishu_notifier = FeishuNotifier()
     feishu_notifier.send_message(content=content, msg_type=msg_type, title=title, is_at_all=is_at_all,
-                                 project_name=project_name, url_base=url_base)
+                                 project_name=project_name, gitlab_url_slug=gitlab_url_slug)