auto-sync: 2026-06-07 13:50:55

src/api/toolchain_routes.py

@@ -13,6 +13,7 @@ import hmac
 import json
 import logging
 import os
+import re
 import time
 from datetime import datetime
 from pathlib import Path, PurePath
@@ -34,6 +35,8 @@ router = APIRouter(tags=["toolchain"])
 # ---------------------------------------------------------------------------
 # 幂等检查：内存 set，保留最近 7 天
 # ---------------------------------------------------------------------------
+# 使用内存 set 而非 SQLite（设计文档原计划 SQLite，简化实现：daemon 重启不频繁，
+# 重启后丢失可接受，Webhook 重试窗口内不会重复）
 
 _delivery_cache: Set[str] = set()
 _delivery_timestamps: List[Tuple[float, str]] = []
@@ -79,9 +82,7 @@ def _verify_signature(body: bytes, signature: Optional[str]) -> bool:
 # Gitea API 调用
 # ---------------------------------------------------------------------------
 
-_GITEA_TOKEN: str = os.environ.get(
-    "GITEA_TOKEN", "a6d596b826f4bfeaf983ef4d25ac25dab95bbc4e"
-)
+_GITEA_TOKEN: str = os.environ.get("GITEA_TOKEN", "")
 _GITEA_BASE = "http://192.168.2.154:3000/api/v1"
 
 
@@ -95,6 +96,10 @@ async def _fetch_pr_files(repo: str, pr_number: int) -> List[str]:
     Returns:
         文件路径列表
     """
+    if not _GITEA_TOKEN:
+        logger.warning("GITEA_TOKEN not set, cannot fetch PR files")
+        return []
+
     url = f"{_GITEA_BASE}/repos/{repo}/pulls/{pr_number}/files"
     headers = {"Authorization": f"token {_GITEA_TOKEN}"}
     try:
@@ -131,6 +136,11 @@ def _calc_risk_level(changed_files: List[str]) -> str:
 # Mail 创建
 # ---------------------------------------------------------------------------
 
+KNOWN_AGENTS = {
+    "pangtong-fujunshi", "simayi-challenger", "zhangfei-dev",
+    "guanyu-dev", "zhaoyun-data", "jiangwei-infra",
+}
+
 MAIL_PROJECT_ID = "_mail"
 
 
@@ -163,6 +173,10 @@ def _send_mail(
     Raises:
         Exception: 数据库写入失败
     """
+    if to_agent not in KNOWN_AGENTS:
+        logger.warning("Unknown agent: %s, skipping mail", to_agent)
+        return ""
+
     mail_id = f"mail-{int(datetime.now().timestamp() * 1000)}"
     notify_meta = {
         "type": "inform",
@@ -253,8 +267,10 @@ async def _handle_pull_request_review(payload: Dict[str, Any]) -> None:
     reviewer = review.get("user", {}).get("login", "unknown")
     review_body = review.get("body", "(无评论)")
 
-    result_map = {"APPROVED": "通过 ✓", "REJECTED": "驳回 ✗", "PENDING": "待定"}
-    result = result_map.get(state, state)
+    result_map = {"APPROVED": "通过 ✓", "REQUEST_CHANGES": "驳回 ✗"}
+    if state not in result_map:
+        return
+    result = result_map[state]
 
     text = render_template("review_result", {
         "repo": repo,
@@ -304,7 +320,9 @@ async def _handle_issues(payload: Dict[str, Any]) -> None:
         _send_mail(assignee, title, text)
 
     elif action == "opened" and "部署失败" in issue_title:
-        commit_sha = issue.get("body", "")[:40] if issue.get("body") else ""
+        # 从 Issue body 提取 commit hash（Gitea deploy workflow 格式）
+        sha_match = re.search(r'[0-9a-f]{40}', issue.get("body", ""))
+        commit_sha = sha_match.group(0) if sha_match else "(未知)"
 
         text = render_template("deploy_failure", {
             "repo": repo,
@@ -331,7 +349,8 @@ async def _handle_issue_comment(payload: Dict[str, Any]) -> None:
 
     # 尝试从关联 PR 获取信息
     pr_author = issue.get("user", {}).get("login", "unknown")
-    branch = "(未知)"
+    branch_match = re.search(r"分支:\s*(\S+)", body)
+    branch = branch_match.group(1) if branch_match else "(未知)"
 
     # 提取错误摘要（取 comment body 前 500 字符）
     error_summary = body[:500] if body else "(无错误信息)"
@@ -384,7 +403,7 @@ async def gitea_webhook(
     # 1. 签名验证
     if not _verify_signature(body, x_gitea_signature):
         logger.warning("Webhook signature verification failed")
-        return Response(status_code=200, content="signature verification failed")
+        return Response(status_code=403, content="signature verification failed")
 
     # 2. 幂等检查
     if x_gitea_event and x_gitea_delivery:

src/daemon/toolchain_templates.py

@@ -56,6 +56,11 @@ def _load_template(name: str) -> str:
     return content
 
 
+def _escape_braces(value: str) -> str:
+    """转义花括号防止 format_map 报错"""
+    return str(value).replace("{", "{{").replace("}", "}}")
+
+
 def render_template(name: str, variables: Dict[str, str]) -> str:
     """渲染模板，将 {variable} 占位符替换为实际值。
 
@@ -69,7 +74,9 @@ def render_template(name: str, variables: Dict[str, str]) -> str:
         渲染后的文本
     """
     template_text = _load_template(name)
-    safe_vars: Dict[str, str] = defaultdict(str, variables)
+    # 先对所有变量值转义花括号，防止 format_map 报错
+    escaped_vars = {k: _escape_braces(v) for k, v in variables.items()}
+    safe_vars: Dict[str, str] = defaultdict(str, escaped_vars)
     return template_text.format_map(safe_vars)