diff --git a/src/api/toolchain_routes.py b/src/api/toolchain_routes.py index 5faed9a..2bd8a84 100644 --- a/src/api/toolchain_routes.py +++ b/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: diff --git a/src/daemon/toolchain_templates.py b/src/daemon/toolchain_templates.py index ce6a1e2..c6f0af8 100644 --- a/src/daemon/toolchain_templates.py +++ b/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)