Merge pull request #2 from GGHansome/github

feat: optimize code

Author: GGHansome

biz/event/event_manager.py

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

biz/utils/im/dingtalk.py

@@ -16,11 +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, gitlab_url_slug=None):
+    def _get_webhook_url(self, project_name=None, url_slug=None):
         """
         获取项目对应的 Webhook URL
         :param project_name: 项目名称
-        :param gitlab_url_slug: 由 gitlab 项目的 url 转换而来的 slug
+        :param url_slug: 由 gitlab 项目的 url 转换而来的 slug
         :return: Webhook URL
         :raises ValueError: 如果未找到 Webhook URL
         """
@@ -33,7 +33,7 @@ def _get_webhook_url(self, project_name=None, gitlab_url_slug=None):
 
         # 构造目标键
         target_key_project = f"DINGTALK_WEBHOOK_URL_{project_name.upper()}"
-        target_key_url_slug = f"DINGTALK_WEBHOOK_URL_{gitlab_url_slug.upper()}"
+        target_key_url_slug = f"DINGTALK_WEBHOOK_URL_{url_slug.upper()}"
 
         # 遍历环境变量
         for env_key, env_value in os.environ.items():
@@ -50,13 +50,13 @@ def _get_webhook_url(self, project_name=None, gitlab_url_slug=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, gitlab_url_slug = None):
+    def send_message(self, content: str, msg_type='text', title='通知', is_at_all=False, project_name=None, url_slug = None):
         if not self.enabled:
             logger.info("钉钉推送未启用")
             return
 
         try:
-            post_url = self._get_webhook_url(project_name=project_name, gitlab_url_slug=gitlab_url_slug)
+            post_url = self._get_webhook_url(project_name=project_name, url_slug=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, gitlab_url_slug=None):
+    def _get_webhook_url(self, project_name=None, url_slug=None):
         """
         获取项目对应的 Webhook URL
         :param project_name: 项目名称
@@ -30,7 +30,7 @@ def _get_webhook_url(self, project_name=None, gitlab_url_slug=None):
 
         # 构造目标键
         target_key_project = f"FEISHU_WEBHOOK_URL_{project_name.upper()}"
-        target_key_url_slug = f"FEISHU_WEBHOOK_URL_{gitlab_url_slug.upper()}"
+        target_key_url_slug = f"FEISHU_WEBHOOK_URL_{url_slug.upper()}"
 
         # 遍历环境变量
         for env_key, env_value in os.environ.items():
@@ -47,7 +47,7 @@ def _get_webhook_url(self, project_name=None, gitlab_url_slug=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, gitlab_url_slug=None):
+    def send_message(self, content, msg_type='text', title=None, is_at_all=False, project_name=None, 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, gitlab_url_slug=gitlab_url_slug)
+            post_url = self._get_webhook_url(project_name=project_name, url_slug=url_slug)
             if msg_type == 'markdown':
                 data = {
                     "msg_type": "interactive",

biz/utils/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, gitlab_url_slug=None):
+def send_notification(content, msg_type='text', title="通知", is_at_all=False, project_name=None, url_slug=None):
     """
     发送通知消息到配置的平台(钉钉和企业微信)
     :param content: 消息内容
     :param msg_type: 消息类型，支持text和markdown
     :param title: 消息标题(markdown类型时使用)
     :param is_at_all: 是否@所有人
-    :param gitlab_url_slug: 由gitlab服务器的url地址(如:http://www.gitlab.com)转换成的slug格式，如: www_gitlab_com
+    :param url_slug: 由gitlab服务器的url地址(如:http://www.gitlab.com)转换成的slug格式，如: www_gitlab_com
     """
     # 钉钉推送
     dingtalk_notifier = DingTalkNotifier()
     dingtalk_notifier.send_message(content=content, msg_type=msg_type, title=title, is_at_all=is_at_all,
-                          project_name=project_name, gitlab_url_slug=gitlab_url_slug)
+                          project_name=project_name, url_slug=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, gitlab_url_slug=gitlab_url_slug)
+                                project_name=project_name, url_slug=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, gitlab_url_slug=gitlab_url_slug)
+                                 project_name=project_name, url_slug=url_slug)

biz/utils/im/wecom.py

@@ -14,7 +14,7 @@ def __init__(self, webhook_url=None):
         self.default_webhook_url = webhook_url or os.environ.get('WECOM_WEBHOOK_URL', '')
         self.enabled = os.environ.get('WECOM_ENABLED', '0') == '1'
 
-    def _get_webhook_url(self, project_name=None, gitlab_url_slug=None):
+    def _get_webhook_url(self, project_name=None, url_slug=None):
         """
         获取项目对应的 Webhook URL
         :param project_name: 项目名称
@@ -30,7 +30,7 @@ def _get_webhook_url(self, project_name=None, gitlab_url_slug=None):
 
         # 构造目标键
         target_key_project = f"WECOM_WEBHOOK_URL_{project_name.upper()}"
-        target_key_url_slug = f"WECOM_WEBHOOK_URL_{gitlab_url_slug.upper()}"
+        target_key_url_slug = f"WECOM_WEBHOOK_URL_{url_slug.upper()}"
 
         # 遍历环境变量
         for env_key, env_value in os.environ.items():
@@ -67,22 +67,22 @@ def format_markdown_content(self, content, title=None):
         return formatted_content
 
     def send_message(self, content, msg_type='text', title=None, is_at_all=False, project_name=None,
-                     gitlab_url_slug=None):
+                     url_slug=None):
         """
         发送企业微信消息
         :param content: 消息内容
         :param msg_type: 消息类型，支持 text 和 markdown
         :param title: 消息标题 (markdown 类型时使用)
         :param is_at_all: 是否 @所有人
         :param project_name: 关联项目名称
-        :param gitlab_url_slug: GitLab URL Slug
+        :param url_slug: GitLab URL Slug
         """
         if not self.enabled:
             logger.info("企业微信推送未启用")
             return
 
         try:
-            post_url = self._get_webhook_url(project_name=project_name, gitlab_url_slug=gitlab_url_slug)
+            post_url = self._get_webhook_url(project_name=project_name, url_slug=url_slug)
             data = self._build_markdown_message(content, title) if msg_type == 'markdown' else self._build_text_message(
                 content, is_at_all)
 

biz/utils/queue.py

@@ -14,14 +14,14 @@
     queues = {}
 
 
-def handle_queue(function: callable, data: any, gitlab_token: str, gitlab_url: str, gitlab_url_slug: str):
+def handle_queue(function: callable, data: any, token: str, url: str, url_slug: str):
     if queue_driver == 'rq':
-        if gitlab_url_slug not in queues:
+        if url_slug not in queues:
             logger.info(f'REDIS_HOST: {os.getenv("REDIS_HOST", "127.0.0.1")}，REDIS_PORT: {os.getenv("REDIS_PORT", 6379)}')
-            queues[gitlab_url_slug] = Queue(gitlab_url_slug, connection=Redis(os.getenv('REDIS_HOST', '127.0.0.1'),
+            queues[url_slug] = Queue(url_slug, connection=Redis(os.getenv('REDIS_HOST', '127.0.0.1'),
                                                                               os.getenv('REDIS_PORT', 6379)))
 
-        queues[gitlab_url_slug].enqueue(function, data, gitlab_token, gitlab_url, gitlab_url_slug)
+        queues[url_slug].enqueue(function, data, token, url, url_slug)
     else:
-        process = Process(target=function, args=(data, gitlab_token, gitlab_url, gitlab_url_slug))
+        process = Process(target=function, args=(data, token, url, url_slug))
         process.start()