"""@mention 解析工具模块。供所有 toolchain handler 复用。""" import re import logging from typing import List, Set from src.config.agents import AGENT_IDS logger = logging.getLogger(__name__) # Gitea API 基地址常量(避免硬编码) GITEA_API_BASE = "http://192.168.2.154:3000/api/v1" GITEA_WEB_BASE = "http://192.168.2.154:3000" # Agent 别名映射 # 规则: # 1. 中文名(如"张飞")→ 完整 Agent ID # 2. 英文短名(如"zhangfei")→ 完整 Agent ID # 3. 前缀模糊匹配需唯一匹配(见 extract_mentions 假设 A2) AGENT_ALIAS: dict[str, str] = { # 中文名 "张飞": "zhangfei-dev", "关羽": "guanyu-dev", "赵云": "zhaoyun-data", "姜维": "jiangwei-infra", "司马懿": "simayi-challenger", "庞统": "pangtong-fujunshi", # 字+号(常见写法) "翼德": "zhangfei-dev", "云长": "guanyu-dev", "子龙": "zhaoyun-data", "伯约": "jiangwei-infra", "仲达": "simayi-challenger", "士元": "pangtong-fujunshi", # 英文短名 "zhangfei": "zhangfei-dev", "guanyu": "guanyu-dev", "zhaoyun": "zhaoyun-data", "jiangwei": "jiangwei-infra", "simayi": "simayi-challenger", "pangtong": "pangtong-fujunshi", } # 正则:匹配 @后面跟着的合法 Agent 名(英文字母/中文/数字/连字符) _MENTION_PATTERN = re.compile(r"@([a-zA-Z\u4e00-\u9fa5][a-zA-Z0-9\u4e00-\u9fff-]*)") def extract_mentions(body: str, sender: str) -> list[str]: """从文本中提取 @mention 的 Agent ID 列表。 Args: body: 评论文本 sender: 评论者 Gitea 用户名(用于排除自己 @自己) Returns: 去重后的 Agent ID 列表 匹配优先级:精确 > 别名 > 前缀模糊(需唯一匹配,多候选则跳过) """ candidates = _MENTION_PATTERN.findall(body) result: Set[str] = set() for c in candidates: # 1. 精确匹配(@zhangfei-dev) if c in AGENT_IDS: result.add(c) # 2. 别名匹配(@张飞、@zhangfei) elif c in AGENT_ALIAS: result.add(AGENT_ALIAS[c]) else: # 3. 前缀模糊匹配(@zhangf → zhangfei-dev) # 假设 A2:多个候选时不匹配,只 log warning matches = [aid for aid in AGENT_IDS if aid.startswith(c)] if len(matches) == 1: result.add(matches[0]) elif len(matches) > 1: logger.warning( "Prefix '%s' matched %d agents (%s), skipping ambiguous mention", c, len(matches), matches) # 排除自己 @自己(假设 A1:Gitea login = Agent ID) result.discard(sender) return list(result) def should_suppress_mention( mentioned_agent: str, auto_notify_targets: List[str], ) -> bool: """判断 @mention 通知是否应被抑制(因为自动流转已通知同一人)。 Args: mentioned_agent: 被 @的 Agent ID auto_notify_targets: 本次事件自动流转已通知的目标列表 Returns: True 表示应抑制(不发 @mention Mail) """ return mentioned_agent in auto_notify_targets def infer_intent(body: str) -> str: """从 @mention 内容推断意图。 Returns: "help" | "notify" | "collaborate" | "assign" """ # 分配子任务关键词 assign_keywords = ["交给", "分配", "负责", "认领", "做一下", "帮忙做", "implement"] if any(kw in body for kw in assign_keywords): return "assign" # 求助关键词(注意:"帮忙"已由 assign_keywords 的"帮忙做"覆盖,"请帮忙"由 collab_keywords 覆盖) help_keywords = ["怎么", "如何", "?", "?", "什么", "哪个", "能否"] if any(kw in body for kw in help_keywords): return "help" # 协作请求关键词 collab_keywords = ["请帮忙", "请协助", "请澄清", "请review", "请审查", "评估"] if any(kw in body for kw in collab_keywords): return "collaborate" # 默认为通知关注 return "notify" def _build_response_guidance( intent: str, gitea_api: str, repo: str, issue_number: int, commenter: str, ) -> str: """根据意图类型生成响应指引文本。""" if intent == "help": return ( f"这是一条求助,请到 Gitea 评论回复:\n" f"1. 获取评论上下文(上方 API)\n" f"2. 组织回答\n" f"3. 在 Gitea 评论回复: POST {gitea_api}/repos/{repo}/issues/{issue_number}/comments\n" f' Body: {{"body": "你的回答内容"}}' ) elif intent == "notify": return ( f"这是一条通知,请查看并知晓。如有意见,可到 Gitea 评论:\n" f"- 查看 Issue/PR 详情(上方 API)\n" f"- 如有意见,评论回复: POST {gitea_api}/repos/{repo}/issues/{issue_number}/comments" ) elif intent == "collaborate": return ( f"这是一条协作请求,请评估后回复(评论或 Mail):\n" f"1. 获取详情(上方 API)\n" f"2. 评估可行性\n" f"3a. 评论回复: POST {gitea_api}/repos/{repo}/issues/{issue_number}/comments\n" f' Body: {{"body": "你的回复"}}\n' f"3b. 或通过 Mail 回复评论者: {commenter}" ) elif intent == "assign": return ( f"这是一条任务分配,请认领并执行:\n" f"1. 获取 Issue 详情(上方 API)\n" f"2. 评估可行性\n" f"3. 认领 Issue: POST {gitea_api}/repos/{repo}/issues/{issue_number}/assignees\n" f' Body: {{"assignees": ["{{your_agent_id}}"]}}\n' f"4. 执行任务\n" f"5. 完成后更新 Issue 状态: PATCH {gitea_api}/repos/{repo}/issues/{issue_number}\n" f' Body: {{"state": "closed"}}' ) return "请查看详情(上方 API)并按需回复。"