增加Merge Request目标分支为受保护分支时才Review的开关：MERGE_REVIEW_ONLY_PROTECTED_BRANCHES_ENABLED

Author: quanbisen

biz/github/webhook_handler.py

@@ -1,10 +1,9 @@
-import json
 import os
 import re
 import time
 
 import requests
-
+import fnmatch
 from biz.utils.log import logger
 
 
@@ -173,6 +172,22 @@ def add_pull_request_notes(self, review_result):
             logger.error(f"Failed to add comment: {response.status_code}")
             logger.error(response.text)
 
+    def target_branch_protected(self) -> bool:
+        url = f"https://api.github.com/repos/{self.repo_full_name}/branches?protected=true"
+        headers = {
+            'Authorization': f'token {self.github_token}',
+            'Accept': 'application/vnd.github.v3+json'
+        }
+
+        response = requests.get(url, headers=headers)
+        if response.status_code == 200:
+            data = response.json()
+            target_branch = self.webhook_data['pull_request']['base']['ref']
+            return any(fnmatch.fnmatch(target_branch, item['name']) for item in data)
+        else:
+            logger.warn(f"Failed to get protected branches: {response.status_code}, {response.text}")
+            return False
+
 
 class PushHandler:
     def __init__(self, webhook_data: dict, github_token: str, github_url: str):

biz/gitlab/webhook_handler.py

@@ -2,7 +2,7 @@
 import re
 import time
 from urllib.parse import urljoin
-
+import fnmatch
 import requests
 
 from biz.utils.log import logger
@@ -15,7 +15,7 @@ def filter_changes(changes: list):
     # 从环境变量中获取支持的文件扩展名
     supported_extensions = os.getenv('SUPPORTED_EXTENSIONS', '.java,.py,.php').split(',')
 
-    filter_deleted_files_changes = [change for change in changes if change.get("deleted_file") == False]
+    filter_deleted_files_changes = [change for change in changes if not change.get("deleted_file")]
 
     # 过滤 `new_path` 以支持的扩展名结尾的元素, 仅保留diff和new_path字段
     filtered_changes = [
@@ -47,7 +47,6 @@ def slugify_url(original_url: str) -> str:
     return target
 
 
-
 class MergeRequestHandler:
     def __init__(self, webhook_data: dict, gitlab_token: str, gitlab_url: str):
         self.merge_request_iid = None
@@ -146,6 +145,24 @@ def add_merge_request_notes(self, review_result):
             logger.error(f"Failed to add note: {response.status_code}")
             logger.error(response.text)
 
+    def target_branch_protected(self) -> bool:
+        url = urljoin(f"{self.gitlab_url}/",
+                      f"api/v4/projects/{self.project_id}/protected_branches")
+        headers = {
+            'Private-Token': self.gitlab_token,
+            'Content-Type': 'application/json'
+        }
+        response = requests.get(url, headers=headers, verify=False)
+        logger.debug(f"Get protected branches response from gitlab: {response.status_code}, {response.text}")
+        # 检查请求是否成功
+        if response.status_code == 200:
+            data = response.json()
+            target_branch = self.webhook_data['object_attributes']['target_branch']
+            return any(fnmatch.fnmatch(target_branch, item['name']) for item in data)
+        else:
+            logger.warn(f"Failed to get protected branches: {response.status_code}, {response.text}")
+            return False
+
 
 class PushHandler:
     def __init__(self, webhook_data: dict, gitlab_token: str, gitlab_url: str):

biz/queue/worker.py

@@ -66,10 +66,15 @@ def handle_merge_request_event(webhook_data: dict, gitlab_token: str, gitlab_url
     :param gitlab_url_slug:
     :return:
     '''
+    merge_review_only_protected_branches = os.environ.get('MERGE_REVIEW_ONLY_PROTECTED_BRANCHES_ENABLED', '0') == '1'
     try:
         # 解析Webhook数据
         handler = MergeRequestHandler(webhook_data, gitlab_token, gitlab_url)
         logger.info('Merge Request Hook event received')
+        # 如果开启了仅review projected branches的，判断当前目标分支是否为projected branches
+        if merge_review_only_protected_branches and not handler.target_branch_protected():
+            logger.info("Merge Request target branch not match protected branches, ignored.")
+            return
 
         if handler.action not in ['open', 'update']:
             logger.info(f"Merge Request Hook event, action={handler.action}, ignored.")
@@ -172,10 +177,15 @@ def handle_github_pull_request_event(webhook_data: dict, github_token: str, gith
     :param github_url_slug:
     :return:
     '''
+    merge_review_only_protected_branches = os.environ.get('MERGE_REVIEW_ONLY_PROTECTED_BRANCHES_ENABLED', '0') == '1'
     try:
         # 解析Webhook数据
         handler = GithubPullRequestHandler(webhook_data, github_token, github_url)
         logger.info('GitHub Pull Request event received')
+        # 如果开启了仅review projected branches的，判断当前目标分支是否为projected branches
+        if merge_review_only_protected_branches and not handler.target_branch_protected():
+            logger.info("Merge Request target branch not match protected branches, ignored.")
+            return
 
         if handler.action not in ['opened', 'synchronize']:
             logger.info(f"Pull Request Hook event, action={handler.action}, ignored.")

biz/utils/code_reviewer.py

@@ -16,7 +16,7 @@ class BaseReviewer(abc.ABC):
 
     def __init__(self, prompt_key: str):
         self.client = Factory().getClient()
-        self.prompts = self._load_prompts(prompt_key,os.getenv("REVIEW_STYLE", "professional"))
+        self.prompts = self._load_prompts(prompt_key, os.getenv("REVIEW_STYLE", "professional"))
 
     def _load_prompts(self, prompt_key: str, style="professional") -> Dict[str, Any]:
         """加载提示词配置"""

conf/.env.dist

@@ -68,6 +68,8 @@ REPORT_CRONTAB_EXPRESSION=0 18 * * 1-5
 
 # 开启Push Review功能(如果不需要push事件触发Code Review，设置为0)
 PUSH_REVIEW_ENABLED=1
+# 开启Merge请求过滤，过滤仅当合并目标分支是受保护分支时才Review(开启此选项请确保仓库已配置受保护分支protected branches)
+MERGE_REVIEW_ONLY_PROTECTED_BRANCHES_ENABLED=0
 
 # Dashboard登录用户名和密码
 DASHBOARD_USER=admin