sanguo/sanguo_moziplus_v2
9
0
Fork 0
Code Issues 3 Pull Requests Actions Packages Projects Releases Wiki Activity

feat(toolchain): §25 Gitea @mention end-to-end integration

Browse Source 
New files:
- src/api/mention_utils.py: extract_mentions(), infer_intent(),
  _build_response_guidance(), AGENT_ALIAS mapping
- templates/toolchain/mention.md: @mention notification template
  with context API + response guidance by intent type

Modified:
- src/api/toolchain_routes.py: S1-S5 handler changes
  - S1: Issue body @mention on opened
  - S2+S4: _handle_issue_comment control flow refactored
    (guard clause → positive if, CI + @mention independent paths)
  - S3: PR body @mention on opened
  - S5: Review body @mention on submitted
  - New _send_mention_mails() with auto-flow suppression
- src/daemon/toolchain_templates.py: register mention template

Design: docs/design/25-gitea-mention-toolchain.md (v2.0)
Tests: 405 passed, 3 skipped
This commit is contained in:
cfdaily
2026-06-12 18:44:50 +08:00
parent 866060e557
commit f25af64f00
5 changed files with 1201 additions and 26 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/api/toolchain_routes.py
+213 -26
View File
@@ -27,6 +27,13 @@ from src.blackboard.db import init_db
from src.blackboard.models import Task
from src.blackboard.operations import Blackboard
from src.config.agents import AGENT_IDS
from src.api.mention_utils import (
extract_mentions,
should_suppress_mention,
infer_intent,
_build_response_guidance,
GITEA_API_BASE,
)
from src.daemon.toolchain_templates import render_template
from src.utils import get_data_root
@@ -253,6 +260,76 @@ def _repo_fullname(payload: Dict[str, Any]) -> str:
return repo.get("full_name", "")
# ---------------------------------------------------------------------------
# @mention 通用发送函数
# ---------------------------------------------------------------------------
async def _send_mention_mails(
mentions: list[str],
auto_targets: list[str],
source_type: str,
mention_type: str,
source_url: str,
commenter: str,
content: str,
repo: str,
issue_number: int,
is_pr: bool,
) -> None:
"""通用 @mention Mail 发送函数。
自动抑制已在 auto_targets 中的 Agent，避免双重通知。
根据内容推断意图，生成不同的响应指引。
"""
# 确定 API 路径
if is_pr:
detail_api = f"pulls/{issue_number}"
comments_api = f"issues/{issue_number}/comments"
else:
detail_api = f"issues/{issue_number}"
comments_api = f"issues/{issue_number}/comments"
for agent_id in mentions:
if should_suppress_mention(agent_id, auto_targets):
logger.info(
"Mention suppressed for %s (already notified by auto flow)",
agent_id)
continue
# 从 api_path 提取编号用于标题，如 "issues/32" → "#32"
number_str = f"#{issue_number}" if issue_number else ""
intent = infer_intent(content)
intent_hint = {"help": "求助", "notify": "通知关注",
"collaborate": "协作请求", "assign": "分配子任务"}[intent]
# 生成响应指引
guidance = _build_response_guidance(
intent=intent,
gitea_api=GITEA_API_BASE,
repo=repo,
issue_number=issue_number,
commenter=commenter,
)
text = render_template("mention", {
"mention_type": mention_type,
"source_type": source_type,
"source_url": source_url,
"commenter": commenter,
"intent_hint": intent_hint,
"content_snippet": content[:500],
"gitea_api": GITEA_API_BASE,
"repo": repo,
"source_detail_api_path": detail_api,
"source_comments_api_path": comments_api,
"response_guidance": guidance,
})
title = f"@mention ({intent_hint}): {source_type} {number_str} ({repo})"
_send_mail(agent_id, title, text)
# ---------------------------------------------------------------------------
# 事件处理函数
# ---------------------------------------------------------------------------
@@ -302,6 +379,26 @@ async def _handle_pr_opened(payload: Dict[str, Any]) -> None:
title = f"Review 请求: {pr_title} ({repo}#{pr_number})"
_send_mail("simayi-challenger", title, text)
# S3: PR body @mention 通知
pr_body = pr.get("body", "") or ""
sender = pr.get("user", {}).get("login", "")
mentions = extract_mentions(pr_body, sender)
if mentions:
# 自动流转已通知 simayi-challenger（review_request）
auto_targets = ["simayi-challenger"]
await _send_mention_mails(
mentions=mentions,
auto_targets=auto_targets,
source_type="PR",
mention_type="PR @mention",
source_url=pr.get("html_url", ""),
commenter=sender,
content=pr_body,
repo=repo,
issue_number=pr_number,
is_pr=True,
)
async def _handle_pull_request_review(payload: Dict[str, Any]) -> None:
"""处理 pull_request_review 事件：非 COMMENTED → 通知 PR 作者。
@@ -364,6 +461,25 @@ async def _handle_pull_request_review(payload: Dict[str, Any]) -> None:
title = f"Review 评论: {pr_title} ({repo}#{pr_number})"
_send_mail(pr_author, title, text)
# S5: Review body @mention 通知（COMMENTED 路径）
mentions = extract_mentions(review_body, reviewer)
if mentions:
# 自动流转已通知 PR 作者（review_comment）
auto_targets = [pr_author]
await _send_mention_mails(
mentions=mentions,
auto_targets=auto_targets,
source_type="Review",
mention_type="Review @mention",
source_url=pr.get("html_url", ""),
commenter=reviewer,
content=review_body,
repo=repo,
issue_number=pr_number,
is_pr=True,
)
return
result_map = {"APPROVED": "通过 ✓", "REQUEST_CHANGES": "驳回 ✗"}
@@ -383,6 +499,24 @@ async def _handle_pull_request_review(payload: Dict[str, Any]) -> None:
title = f"Review {result}: {pr_title} ({repo}#{pr_number})"
_send_mail(pr_author, title, text)
# S5: Review body @mention 通知（非 COMMENTED 路径）
mentions = extract_mentions(review_body, reviewer)
if mentions:
# 自动流转已通知 PR 作者（review_result）
auto_targets = [pr_author]
await _send_mention_mails(
mentions=mentions,
auto_targets=auto_targets,
source_type="Review",
mention_type="Review @mention",
source_url=pr.get("html_url", ""),
commenter=reviewer,
content=review_body,
repo=repo,
issue_number=pr_number,
is_pr=True,
)
async def _fetch_latest_reviewer(repo: str, pr_number: int) -> str:
"""查询 PR 最近一次非 PENDING review 的提交者。
@@ -672,51 +806,104 @@ async def _handle_issues(payload: Dict[str, Any]) -> None:
for agent_id in ("jiangwei-infra", "pangtong-fujunshi"):
_send_mail(agent_id, title, text)
# S1: Issue body @mention（opened 时检查）
if action == "opened":
issue_body = issue.get("body", "") or ""
sender = payload.get("sender", {}).get("login", "")
mentions = extract_mentions(issue_body, sender)
if mentions:
# 自动流转已通知 assignee
assignees = issue.get("assignees") or []
if not assignees:
single = issue.get("assignee")
if single and isinstance(single, dict):
assignees = [single]
auto_targets = [a.get("login", "") for a in assignees if isinstance(a, dict)]
await _send_mention_mails(
mentions=mentions,
auto_targets=auto_targets,
source_type="Issue",
mention_type="Issue @mention",
source_url=issue.get("html_url", ""),
commenter=sender,
content=issue_body,
repo=repo,
issue_number=issue_number,
is_pr=False,
)
async def _handle_issue_comment(payload: Dict[str, Any]) -> None:
"""处理 issue_comment 事件：CI 失败关键词 → 通知 PR 作者。"""
"""处理 issue_comment 事件：CI 失败关键词 → 通知 PR 作者；@mention → 通知被提及者。"""
comment = payload.get("comment")
if not comment or not isinstance(comment, dict):
logger.warning("issue_comment event missing comment field, skipping")
return
body = comment.get("body", "")
# 检查是否包含 CI 失败关键词
if "[CI]" not in body and "CI 失败" not in body:
return
sender = comment.get("user", {}).get("login", "")
issue = payload.get("issue")
if not issue or not isinstance(issue, dict):
logger.warning("issue_comment event missing issue field, skipping")
return
# 已关闭的 Issue/PR 不再发送 CI 失败通知
if issue.get("state") == "closed":
logger.debug(
"Skipping CI failure notification for closed issue #%s",
issue.get("number"))
action = payload.get("action", "")
if action != "created":
return
repo = _repo_fullname(payload)
issue_number = issue.get("number", 0)
# === 路径 1：CI 失败通知（原有逻辑，改为正向 if） ===
if ("[CI]" in body or "CI 失败" in body) and issue.get("state") != "closed":
repo = _repo_fullname(payload)
issue_number = issue.get("number", 0)
# 尝试从关联 PR 获取信息
pr_author = issue.get("user", {}).get("login", "unknown")
branch_match = re.search(r"分支:\s*(\S+)", body)
branch = branch_match.group(1) if branch_match else "(未知)"
# 尝试从关联 PR 获取信息
pr_author = issue.get("user", {}).get("login", "unknown")
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 "(无错误信息)"
# 提取错误摘要（取 comment body 前 500 字符）
error_summary = body[:500] if body else "(无错误信息)"
text = render_template("ci_failure", {
"repo": repo,
"pr_number": str(issue_number),
"branch": branch,
"error_summary": error_summary,
})
text = render_template("ci_failure", {
"repo": repo,
"pr_number": str(issue_number),
"branch": branch,
"error_summary": error_summary,
})
title = f"CI 失败: {repo}#{issue_number}"
_send_mail(pr_author, title, text)
title = f"CI 失败: {repo}#{issue_number}"
_send_mail(pr_author, title, text)
# CI 处理完不 return，继续检查 @mention
# === 路径 2：@mention 通知（新增，独立路径） ===
# 注意：@mention 检测与 CI 检测是独立的，同一条评论可同时触发两者
mentions = extract_mentions(body, sender)
if mentions:
# 判断是 PR 还是 Issue（Gitea 中 PR 本质是特殊的 Issue）
is_pr = "pull_request" in issue
source_type = "PR" if is_pr else "Issue"
mention_type = "PR @mention" if is_pr else "Issue @mention"
issue_number = issue.get("number", 0)
repo = _repo_fullname(payload)
# 自动流转已通知的人（CI 失败通知的 PR 作者）
auto_targets: list[str] = []
if ("[CI]" in body or "CI 失败" in body) and issue.get("state") != "closed":
auto_targets.append(issue.get("user", {}).get("login", ""))
await _send_mention_mails(
mentions=mentions,
auto_targets=auto_targets,
source_type=source_type,
mention_type=mention_type,
source_url=issue.get("html_url", ""),
commenter=sender,
content=body,
repo=repo,
issue_number=issue_number,
is_pr=is_pr,
)
# ---------------------------------------------------------------------------