Merge pull request #63 from sunmh207/mashb1t-feature/queue-driver-rq

Mashb1t feature/queue driver rq

Author: sunmh207

.github/workflows/build_image.yml renamed to .github/workflows/build_images.yml

@@ -45,6 +45,17 @@ jobs:
             type=semver,pattern={{major}}
             type=edge,branch=main
 
+      - name: Extract metadata (tags, labels) for Docker (worker)
+        id: meta_worker
+        uses: docker/metadata-action@v5
+        with:
+          images: ghcr.io/${{ github.repository_owner }}/${{ github.event.repository.name }}
+          tags: |
+            type=semver,pattern={{version}},suffix=-worker
+            type=semver,pattern={{major}}.{{minor}},suffix=-worker
+            type=semver,pattern={{major}},suffix=-worker
+            type=edge,branch=main,suffix=-worker
+
       # Build and push multi-arch prod image
       - name: Build and push prod Docker image (Multi-Arch)
         uses: docker/build-push-action@v6
@@ -56,3 +67,15 @@ jobs:
           tags: ${{ steps.meta_prod.outputs.tags }}
           labels: ${{ steps.meta_prod.outputs.labels }}
           target: prod
+
+      # Build and push multi-arch worker image
+      - name: Build and push worker Docker image (Multi-Arch)
+        uses: docker/build-push-action@v6
+        with:
+          context: .
+          file: ./Dockerfile
+          push: true
+          platforms: linux/amd64,linux/arm64/v8
+          tags: ${{ steps.meta_worker.outputs.tags }}
+          labels: ${{ steps.meta_worker.outputs.labels }}
+          target: worker

Dockerfile

@@ -19,15 +19,18 @@ COPY api.py ./api.py
 COPY ui.py ./ui.py
 COPY conf/prompt_templates.yml ./conf/prompt_templates.yml
 
-
-# 暴露 Flask 和 Streamlit 的端口
-EXPOSE 5001 5002
-
 # 使用 supervisord 作为启动命令
 CMD ["/usr/bin/supervisord", "-c", "/etc/supervisor/conf.d/supervisord.conf"]
 
 FROM base AS dev
 COPY ./conf/supervisord.dev.conf /etc/supervisor/conf.d/supervisord.conf
+# 暴露 Flask 和 Streamlit 的端口
+EXPOSE 5001 5002
 
 FROM base AS prod
-COPY ./conf/supervisord.prod.conf /etc/supervisor/conf.d/supervisord.conf
+COPY ./conf/supervisord.prod.conf /etc/supervisor/conf.d/supervisord.conf
+# 暴露 Flask 和 Streamlit 的端口
+EXPOSE 5001 5002
+
+FROM base AS worker
+COPY ./conf/supervisord.worker.conf /etc/supervisor/conf.d/supervisord.conf

api.py

@@ -1,25 +1,21 @@
 import atexit
 import json
 import os
-import re
 import traceback
 from datetime import datetime
-from multiprocessing import Process
 from urllib.parse import urlparse
 
-from biz.utils.token_util import count_tokens, truncate_text_by_tokens
 from apscheduler.schedulers.background import BackgroundScheduler
 from apscheduler.triggers.cron import CronTrigger
 from dotenv import load_dotenv
 from flask import Flask, request, jsonify
 
-from biz.entity.review_entity import MergeRequestReviewEntity, PushReviewEntity
-from biz.event.event_manager import event_manager
-from biz.gitlab.webhook_handler import MergeRequestHandler, PushHandler
+from biz.gitlab.webhook_handler import slugify_url
+from biz.queue.worker import handle_merge_request_event, handle_push_event
 from biz.service.review_service import ReviewService
-from biz.utils.code_reviewer import CodeReviewer
 from biz.utils.im import notifier
 from biz.utils.log import logger
+from biz.utils.queue import handle_queue
 from biz.utils.reporter import Reporter
 
 load_dotenv("conf/.env")
@@ -135,22 +131,23 @@ def handle_webhook():
         if not gitlab_token:
             return jsonify({'message': 'Missing GitLab access token'}), 400
 
+        gitlab_url_slug = slugify_url(gitlab_url)
+
         # 打印整个payload数据，或根据需求进行处理
         logger.info(f'Received event: {object_kind}')
         logger.info(f'Payload: {json.dumps(data)}')
 
         # 处理Merge Request Hook
         if object_kind == "merge_request":
             # 创建一个新进程进行异步处理
-            process = Process(target=__handle_merge_request_event, args=(data, gitlab_token, gitlab_url))
-            process.start()
+            handle_queue(handle_merge_request_event, data, gitlab_token, gitlab_url, gitlab_url_slug)
             # 立马返回响应
             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()
+            # TODO check if PUSH_REVIEW_ENABLED is needed here
+            handle_queue(handle_push_event, data, gitlab_token, gitlab_url, gitlab_url_slug)
             # 立马返回响应
             return jsonify(
                 {'message': f'Request received(object_kind={object_kind}), will process asynchronously.'}), 200
@@ -162,171 +159,6 @@ def handle_webhook():
         return jsonify({'message': 'Invalid data format'}), 400
 
 
-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:
-        handler = PushHandler(webhook_data, gitlab_token, gitlab_url)
-        logger.info('Push Hook event received')
-        commits = handler.get_push_commits()
-        if not commits:
-            logger.error('Failed to get commits')
-            return
-
-        review_result = None
-        score = 0
-        if PUSH_REVIEW_ENABLED:
-            # 获取PUSH的changes
-            changes = handler.get_push_changes()
-            logger.info('changes: %s', changes)
-            changes = filter_changes(changes)
-            if not changes:
-                logger.info('未检测到PUSH代码的修改,修改文件可能不满足SUPPORTED_EXTENSIONS。')
-                return
-            review_result = "关注的文件没有修改"
-
-            if len(changes) > 0:
-                commits_text = ';'.join(commit.get('message', '').strip() for commit in commits)
-                review_result = review_code(str(changes), commits_text)
-                score = CodeReviewer.parse_review_score(review_text=review_result)
-            # 将review结果提交到Gitlab的 notes
-            handler.add_push_notes(f'Auto Review Result: \n{review_result}')
-
-        event_manager['push_reviewed'].send(PushReviewEntity(
-            project_name=webhook_data['project']['name'],
-            author=webhook_data['user_username'],
-            branch=webhook_data['project']['default_branch'],
-            updated_at=int(datetime.now().timestamp()),  # 当前时间
-            commits=commits,
-            score=score,
-            review_result=review_result,
-            gitlab_url_slug=slugify_url(gitlab_url),
-        ))
-
-    except Exception as e:
-        error_message = f'服务出现未知错误: {str(e)}\n{traceback.format_exc()}'
-        notifier.send_notification(content=error_message)
-        logger.error('出现未知错误: %s', error_message)
-
-
-def __handle_merge_request_event(webhook_data: dict, gitlab_token: str, gitlab_url: str):
-    '''
-    处理Merge Request Hook事件
-    :param webhook_data:
-    :param gitlab_token:
-    :param gitlab_url:
-    :return:
-    '''
-    try:
-        # 解析Webhook数据
-        handler = MergeRequestHandler(webhook_data, gitlab_token, gitlab_url)
-        logger.info('Merge Request Hook event received')
-
-        if (handler.action in ['open', 'update']):  # 仅仅在MR创建或更新时进行Code Review
-            # 获取Merge Request的changes
-            changes = handler.get_merge_request_changes()
-            logger.info('changes: %s', changes)
-            changes = filter_changes(changes)
-            if not changes:
-                logger.info('未检测到有关代码的修改,修改文件可能不满足SUPPORTED_EXTENSIONS。')
-                return
-
-            # 获取Merge Request的commits
-            commits = handler.get_merge_request_commits()
-            if not commits:
-                logger.error('Failed to get commits')
-                return
-
-            # review 代码
-            commits_text = ';'.join(commit['title'] for commit in commits)
-            review_result = review_code(str(changes), commits_text)
-
-            if "COT ABORT!" in review_result:
-                logger.error('COT ABORT!')
-                return
-
-            # 将review结果提交到Gitlab的 notes
-            handler.add_merge_request_notes(f'Auto Review Result: \n{review_result}')
-
-            # dispatch merge_request_reviewed event
-            event_manager['merge_request_reviewed'].send(
-                MergeRequestReviewEntity(
-                    project_name=webhook_data['project']['name'],
-                    author=webhook_data['user']['username'],
-                    source_branch=webhook_data['object_attributes']['source_branch'],
-                    target_branch=webhook_data['object_attributes']['target_branch'],
-                    updated_at=int(datetime.now().timestamp()),
-                    commits=commits,
-                    score=CodeReviewer.parse_review_score(review_text=review_result),
-                    url=webhook_data['object_attributes']['url'],
-                    review_result=review_result,
-                    gitlab_url_slug=slugify_url(gitlab_url),
-                )
-            )
-
-        else:
-            logger.info(f"Merge Request Hook event, action={handler.action}, ignored.")
-
-    except Exception as e:
-        error_message = f'AI Code Review 服务出现未知错误: {str(e)}\n{traceback.format_exc()}'
-        notifier.send_notification(content=error_message)
-        logger.error('出现未知错误: %s', error_message)
-
-
-def filter_changes(changes: list):
-    '''
-    过滤数据，只保留支持的文件类型以及必要的字段信息
-    '''
-    filter_deleted_files_changes = [change for change in changes if change.get("deleted_file") == False]
-    # 从环境变量中获取支持的文件扩展名
-    SUPPORTED_EXTENSIONS = os.getenv('SUPPORTED_EXTENSIONS', '.java,.py,.php').split(',')
-    # 过滤 `new_path` 以支持的扩展名结尾的元素, 仅保留diff和new_path字段
-    filtered_changes = [
-        {
-            'diff': item.get('diff', ''),
-            'new_path': item['new_path']
-        }
-        for item in filter_deleted_files_changes
-        if any(item.get('new_path', '').endswith(ext) for ext in SUPPORTED_EXTENSIONS)
-    ]
-    return filtered_changes
-
-
-def review_code(changes_text: str, commits_text: str = '') -> str:
-    # 如果超长，取前REVIEW_MAX_TOKENS个token
-    review_max_tokens = int(os.getenv('REVIEW_MAX_TOKENS', 10000))
-    # 如果changes为空,打印日志
-    if not changes_text:
-        logger.info('代码为空, diffs_text = %', str(changes_text))
-        return '代码为空'
-
-    # 计算tokens数量，如果超过REVIEW_MAX_TOKENS，截断changes_text
-    tokens_count = count_tokens(changes_text)
-    if tokens_count > review_max_tokens:
-        changes_text = truncate_text_by_tokens(changes_text, review_max_tokens)
-
-    review_result = CodeReviewer().review_code(changes_text, commits_text).strip()
-    if review_result.startswith("```markdown") and review_result.endswith("```"):
-        return review_result[11:-3].strip()
-    return review_result
-
-
 if __name__ == '__main__':
     # 启动定时任务调度器
     setup_scheduler()

biz/gitlab/test_webhook_handler.py

@@ -2,7 +2,6 @@
 # -*- coding: utf-8 -*-
 # @Time    : 2025/3/18 17:58
 # @Author  : Arrow
-import os
 from unittest import TestCase, main
 
 from biz.gitlab.webhook_handler import PushHandler
@@ -34,4 +33,4 @@ def test_get_parent_commit_id(self):
 
 
 if __name__ == '__main__':
-    main()
+    main()

biz/gitlab/webhook_handler.py

@@ -1,11 +1,54 @@
-import json
+import os
+import re
 import time
 from urllib.parse import urljoin
 
 import requests
+from dotenv import load_dotenv
 
 from biz.utils.log import logger
 
+load_dotenv()
+# 从环境变量中获取支持的文件扩展名
+SUPPORTED_EXTENSIONS = os.getenv('SUPPORTED_EXTENSIONS', '.java,.py,.php').split(',')
+
+
+def filter_changes(changes: list):
+    '''
+    过滤数据，只保留支持的文件类型以及必要的字段信息
+    '''
+    filter_deleted_files_changes = [change for change in changes if change.get("deleted_file") == False]
+
+    # 过滤 `new_path` 以支持的扩展名结尾的元素, 仅保留diff和new_path字段
+    filtered_changes = [
+        {
+            'diff': item.get('diff', ''),
+            'new_path': item['new_path']
+        }
+        for item in filter_deleted_files_changes
+        if any(item.get('new_path', '').endswith(ext) for ext in SUPPORTED_EXTENSIONS)
+    ]
+    return filtered_changes
+
+
+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
+
+
 
 class MergeRequestHandler:
     def __init__(self, webhook_data: dict, gitlab_token: str, gitlab_url: str):
@@ -179,7 +222,8 @@ def add_push_notes(self, message: str):
             logger.error(f"Failed to add comment: {response.status_code}")
             logger.error(response.text)
 
-    def __repository_commits(self, ref_name: str = "", since: str = "", until: str = "", pre_page: int = 100, page: int = 1):
+    def __repository_commits(self, ref_name: str = "", since: str = "", until: str = "", pre_page: int = 100,
+                             page: int = 1):
         # 获取仓库提交信息
         url = f"{urljoin(f'{self.gitlab_url}/', f'api/v4/projects/{self.project_id}/repository/commits')}?ref_name={ref_name}&since={since}&until={until}&per_page={pre_page}&page={page}"
         headers = {