feat: 重构 GitHub webhook 处理逻辑，提取签名验证和仓库匹配功能到独立模块

Author: AptS-1547

app.py

@@ -1,12 +1,9 @@
-import hmac
-import hashlib
 import logging
-from typing import Optional
-
-from fastapi import FastAPI, Request, HTTPException, Header
+from fastapi import FastAPI, Request, HTTPException
 
 from settings import Settings
 from send_message import send_github_notification
+from hooks.github_webhook import verify_signature, find_matching_webhook, extract_push_data
 
 logging.basicConfig(level=logging.INFO)
 logger = logging.getLogger(__name__)
@@ -15,71 +12,6 @@
 
 settings = Settings().from_yaml()
 
-# 全局变量
-WS_URL = settings.WS_URL
-WS_ACCESS_TOKEN = settings.WS_ACCESS_TOKEN
-GITHUB_WEBHOOK = settings.GITHUB_WEBHOOK
-
-def match_repository(repo_name: str, pattern: str) -> bool:
-    """
-    检查仓库名是否匹配配置中的模式
-    支持大小写不敏感匹配和通配符（用户名/*）形式
-    
-    Args:
-        repo_name: 实际的仓库全名 (例如 'user/repo')
-        pattern: 配置中的仓库模式 (例如 'user/repo' 或 'user/*')
-    
-    Returns:
-        bool: 是否匹配
-    """
-    if not repo_name or not pattern:
-        return False
-
-    repo_name = repo_name.lower()
-    pattern = pattern.lower()
-
-    if pattern.endswith('/*'):
-        user = pattern[:-2]
-        return repo_name.startswith(f"{user}/")
-
-    return repo_name == pattern
-
-async def verify_signature(request: Request, x_hub_signature_256: Optional[str] = Header(None)):
-    """验证 GitHub webhook 签名"""
-
-    try:
-        body = await request.body()
-        payload = await request.json()
-        repo_name = payload.get("repository", {}).get("full_name")
-    except Exception as e:
-        logger.error("解析请求体失败: %s", str(e))
-        raise HTTPException(status_code=400, detail="Invalid JSON payload") from e
-
-    webhook_secret = None
-    for webhook in settings.GITHUB_WEBHOOK:
-        if any(match_repository(repo_name, repo_pattern) for repo_pattern in webhook.REPO):
-            webhook_secret = webhook.SECRET
-            break
-
-    if not webhook_secret:
-        logger.warning("仓库 %s 未配置 webhook 密钥，跳过签名验证", repo_name)
-        return True
-
-    if not x_hub_signature_256:
-        raise HTTPException(status_code=401, detail="Missing X-Hub-Signature-256 header")
-
-    signature = hmac.new(
-        key=webhook_secret.encode(),
-        msg=body,
-        digestmod=hashlib.sha256
-    ).hexdigest()
-
-    expected_signature = f"sha256={signature}"
-    if not hmac.compare_digest(expected_signature, x_hub_signature_256):
-        raise HTTPException(status_code=401, detail="Invalid signature")
-
-    return True
-
 @app.post("/github-webhook")
 async def github_webhook(request: Request):
     """处理 GitHub webhook 请求"""
@@ -89,13 +21,20 @@ async def github_webhook(request: Request):
         logger.info("收到非 JSON 格式的请求，忽略")
         return {"status": "ignored", "message": "只处理 application/json 格式的请求"}
 
-    event_type = request.headers.get("X-GitHub-Event")
+    event_type = request.headers.get("X-GitHub-Event", "")
+    if not event_type:
+        logger.info("缺少 X-GitHub-Event 头，忽略")
+        return {"status": "ignored", "message": "缺少 X-GitHub-Event 头"}
 
     try:
-        await verify_signature(request, request.headers.get("X-Hub-Signature-256"))
-    except HTTPException:
-        logger.info("签名验证失败")
-        return {"status": "ignored", "message": "签名验证失败"}
+        await verify_signature(
+            request, 
+            settings.GITHUB_WEBHOOK, 
+            request.headers.get("X-Hub-Signature-256")
+        )
+    except HTTPException as e:
+        logger.info(f"签名验证失败: {e.detail}")
+        return {"status": "ignored", "message": f"签名验证失败: {e.detail}"}
 
     payload = await request.json()
     if not payload:
@@ -105,41 +44,38 @@ async def github_webhook(request: Request):
     repo_name = payload.get("repository", {}).get("full_name")
     branch = payload.get("ref", "").replace("refs/heads/", "")
 
-    matched_webhook = None
-    for webhook in settings.GITHUB_WEBHOOK:
-
-        repo_matches = any(match_repository(repo_name, repo_pattern) for repo_pattern in webhook.REPO)
-
-        if (repo_matches and 
-            branch in webhook.BRANCH and 
-            event_type in webhook.EVENTS):
-            matched_webhook = webhook
-            break
+    matched_webhook = find_matching_webhook(
+        repo_name,
+        branch,
+        event_type,
+        settings.GITHUB_WEBHOOK
+    )
 
     if not matched_webhook:
         logger.info("找不到匹配的 webhook 配置: 仓库 %s, 分支 %s, 事件类型 %s", repo_name, branch, event_type)
         return {"status": "ignored", "message": "找不到匹配的 webhook 配置"}
 
     # 处理不同类型的事件，暂时只支持 push 事件
     if event_type == "push":
-        pusher = payload.get("pusher", {}).get("name")
-        commits = payload.get("commits", [])
-        commit_count = len(commits)
+        push_data = extract_push_data(payload)
 
-        logger.info("发现新的 push 事件，来自 %s 仓库", repo_name)
-        logger.info("分支: %s，推送者: %s，提交数量: %s", branch, pusher, commit_count)
+        logger.info("发现新的 push 事件，来自 %s 仓库", push_data["repo_name"])
+        logger.info("分支: %s，推送者: %s，提交数量: %s",
+                   push_data["branch"],
+                   push_data["pusher"],
+                   push_data["commit_count"])
 
         # 向配置的所有 OneBot 目标发送通知
         for target in matched_webhook.ONEBOT:
             logger.info("正在发送消息到 QQ 群 %s", target.id)
             await send_github_notification(
                 ws_url=settings.WS_URL,
                 access_token=settings.WS_ACCESS_TOKEN,
-                repo_name=repo_name,
-                branch=branch,
-                pusher=pusher,
-                commit_count=commit_count,
-                commits=commits,
+                repo_name=push_data["repo_name"],
+                branch=push_data["branch"],
+                pusher=push_data["pusher"],
+                commit_count=push_data["commit_count"],
+                commits=push_data["commits"],
                 onebot_type=target.type,
                 onebot_id=target.id
             )

hooks/github_webhook.py

@@ -0,0 +1,109 @@
+import hmac
+import hashlib
+import logging
+from typing import Optional, Dict, Any, List
+
+from fastapi import Request, HTTPException, Header
+
+logger = logging.getLogger(__name__)
+
+def match_repository(repo_name: str, pattern: str) -> bool:
+    """
+    检查仓库名是否匹配配置中的模式
+    支持大小写不敏感匹配和通配符（用户名/*）形式
+    
+    Args:
+        repo_name: 实际的仓库全名 (例如 'user/repo')
+        pattern: 配置中的仓库模式 (例如 'user/repo' 或 'user/*')
+    
+    Returns:
+        bool: 是否匹配
+    """
+    if not repo_name or not pattern:
+        return False
+
+    repo_name = repo_name.lower()
+    pattern = pattern.lower()
+
+    if pattern.endswith('/*'):
+        user = pattern[:-2]
+        return repo_name.startswith(f"{user}/")
+
+    return repo_name == pattern
+
+async def verify_signature(request: Request, webhooks: List, x_hub_signature_256: Optional[str] = Header(None)):
+    """验证 GitHub webhook 签名"""
+
+    try:
+        body = await request.body()
+        payload = await request.json()
+        repo_name = payload.get("repository", {}).get("full_name")
+    except Exception as e:
+        logger.error("解析请求体失败: %s", str(e))
+        raise HTTPException(status_code=400, detail="Invalid JSON payload") from e
+
+    webhook_secret = None
+    for webhook in webhooks:
+        if any(match_repository(repo_name, repo_pattern) for repo_pattern in webhook.REPO):
+            webhook_secret = webhook.SECRET
+            break
+
+    if not webhook_secret:
+        logger.warning("仓库 %s 未配置 webhook 密钥，跳过签名验证", repo_name)
+        return True
+
+    if not x_hub_signature_256:
+        raise HTTPException(status_code=401, detail="Missing X-Hub-Signature-256 header")
+
+    signature = hmac.new(
+        key=webhook_secret.encode(),
+        msg=body,
+        digestmod=hashlib.sha256
+    ).hexdigest()
+
+    expected_signature = f"sha256={signature}"
+    if not hmac.compare_digest(expected_signature, x_hub_signature_256):
+        raise HTTPException(status_code=401, detail="Invalid signature")
+
+    return True
+
+def find_matching_webhook(repo_name: str, branch: str, event_type: str, webhooks: List) -> Optional[Any]:
+    """
+    查找匹配的 webhook 配置
+    
+    Args:
+        repo_name: 仓库名
+        branch: 分支名
+        event_type: 事件类型
+        webhooks: webhook 配置列表
+        
+    Returns:
+        匹配的 webhook 配置或 None
+    """
+    for webhook in webhooks:
+        repo_matches = any(match_repository(repo_name, repo_pattern) for repo_pattern in webhook.REPO)
+
+        if (repo_matches and
+            branch in webhook.BRANCH and
+            event_type in webhook.EVENTS):
+            return webhook
+
+    return None
+
+def extract_push_data(payload: Dict[str, Any]) -> Dict[str, Any]:
+    """
+    从 push 事件中提取相关数据
+    
+    Args:
+        payload: GitHub webhook 负载
+        
+    Returns:
+        包含提取数据的字典
+    """
+    return {
+        "repo_name": payload.get("repository", {}).get("full_name"),
+        "branch": payload.get("ref", "").replace("refs/heads/", ""),
+        "pusher": payload.get("pusher", {}).get("name"),
+        "commits": payload.get("commits", []),
+        "commit_count": len(payload.get("commits", []))
+    }