#!/usr/bin/env python3 """Collect local evidence for a WeCom project-group feedback message.""" from __future__ import annotations import argparse import datetime as dt import os import re import shutil import subprocess from pathlib import Path DEFAULT_TERMS = [ "订单", "小票", "打印", "支付", "退款", "充值", "导出", "报错", "异常", "接口", "权限", "登录", "设备", "出货", "售卖柜", "菜品", "档口", "餐别", "消费", "扣款", "餐补", "补贴", "余额", "现金", "小程序", "金斯瑞", "叫号", "叫号屏", "叫号大屏", "大屏", "轮播", "取餐", "待取餐", "备餐", "屏幕", "机场", "负值", "负值餐补", "餐补扣款", "补贴扣款", "补贴余额", "营养", "定时", "cron", "redis", "mysql", ] SEARCH_DIRS = [ "zhctprompt/work", "zhctprompt/work_store", "zhctprompt/work_ai_api", "zhctprompt/work_ai_store", "zhctprompt/work_ai_app", "zhctprompt/control", "zhctprompt/standards-stack/prompt", "zhctproject/store/application", "zhctproject/store/public/screen/src", "zhctproject/store/app", "zhctproject/store/cron", "zhctproject/ai_api/app", "zhctproject/ai_api/config", "zhctproject/ai_api/route", "zhctproject/ai_store/src", "zhctproject/ai_app", "zhctlogs/log", "zhctlogs/wwwlogs", ] RG_GLOBS = [ "!zhctprompt/work/2026-05-09-wecom-project-agent/**", "!vendor/**", "!node_modules/**", "!runtime/**", "!public/uploads/**", "!*.min.js", "!*.map", "!*.png", "!*.jpg", "!*.jpeg", "!*.gif", "!*.zip", "!*.tar", "!*.gz", ] SKIP_PREFIXES = [ "zhctprompt/work/2026-05-09-wecom-project-agent/", ] SKIP_FILES = { "zhctprompt/control/scripts/wecom_project_triage.py", } SKIP_DIR_NAMES = { ".git", "vendor", "node_modules", "runtime", "public/uploads", } SKIP_SUFFIXES = { ".png", ".jpg", ".jpeg", ".gif", ".zip", ".tar", ".gz", ".map", } STOP_TERMS = { "tec", "arobot", "回答一下这个问题", "帮忙看下原因", "测试", } def read_message(args: argparse.Namespace) -> str: if args.message: return args.message.strip() if args.message_file: return Path(args.message_file).read_text(encoding="utf-8").strip() raise SystemExit("missing --message or --message-file") def project_root(script_path: Path) -> Path: return script_path.resolve().parents[3] def extract_keywords(message: str, extra: list[str]) -> list[str]: terms: list[str] = [] cleaned_message = re.sub(r"@[A-Za-z0-9_\-\u4e00-\u9fff]+", " ", message) for pattern in [ r"[A-Za-z][A-Za-z0-9_./?=&:-]{2,}", r"\d{4,}", r"[\u4e00-\u9fffA-Za-z0-9_]+/[A-Za-z0-9_./-]+", ]: terms.extend(re.findall(pattern, cleaned_message)) terms.extend(term for term in DEFAULT_TERMS if term.lower() in cleaned_message.lower()) if re.search(r"(叫号|大屏|轮播|取餐)", cleaned_message): terms.extend(["pickupPageSize", "pickupPageCount", "turnQueuePage", "visiblePickupNumbers"]) terms.extend(extra) normalized: list[str] = [] seen: set[str] = set() for term in terms: value = term.strip(" \t\r\n，。；：、,.!?()（）[]【】'\"") if len(value) < 2 or value in STOP_TERMS or value in seen: continue seen.add(value) normalized.append(value) return normalized[:24] def existing_search_dirs(root: Path) -> list[Path]: return [root / path for path in SEARCH_DIRS if (root / path).exists()] def run_rg(root: Path, dirs: list[Path], keyword: str, limit: int) -> list[tuple[str, int, str]]: if not shutil.which("rg"): return run_python_search(root, dirs, keyword, limit) cmd = [ "rg", "-n", "-i", "--no-heading", "--color", "never", "--fixed-strings", ] for glob in RG_GLOBS: cmd.extend(["--glob", glob]) cmd.extend([keyword]) cmd.extend(str(path) for path in dirs) try: proc = subprocess.run( cmd, cwd=root, text=True, stdout=subprocess.PIPE, stderr=subprocess.DEVNULL, timeout=12, check=False, ) except (subprocess.TimeoutExpired, FileNotFoundError): return [] matches: list[tuple[str, int, str]] = [] for raw in proc.stdout.splitlines()[:limit]: parts = raw.split(":", 2) if len(parts) != 3 or not parts[1].isdigit(): continue path = os.path.relpath(parts[0], root) if any(path.startswith(prefix) for prefix in SKIP_PREFIXES): continue if path in SKIP_FILES: continue matches.append((path, int(parts[1]), parts[2].strip())) return matches def should_skip_file(root: Path, file: Path) -> bool: rel = os.path.relpath(file, root) if any(rel.startswith(prefix) for prefix in SKIP_PREFIXES): return True if rel in SKIP_FILES: return True if any(part in SKIP_DIR_NAMES for part in file.parts): return True return file.suffix.lower() in SKIP_SUFFIXES def run_python_search(root: Path, dirs: list[Path], keyword: str, limit: int) -> list[tuple[str, int, str]]: matches: list[tuple[str, int, str]] = [] needle = keyword.lower() for directory in dirs: for file in directory.rglob("*"): if len(matches) >= limit: return matches if not file.is_file() or should_skip_file(root, file): continue try: with file.open("r", encoding="utf-8", errors="ignore") as handle: for line_no, line in enumerate(handle, start=1): if needle in line.lower(): path = os.path.relpath(file, root) matches.append((path, line_no, line.strip())) if len(matches) >= limit: return matches except OSError: continue return matches def score_matches( keyword_matches: dict[str, list[tuple[str, int, str]]], keywords: list[str], ) -> list[tuple[int, str, int, str, list[str]]]: by_location: dict[tuple[str, int, str], set[str]] = {} for keyword, matches in keyword_matches.items(): for path, line_no, text in matches: by_location.setdefault((path, line_no, text), set()).add(keyword) scored: list[tuple[int, str, int, str, list[str]]] = [] for (path, line_no, text), terms in by_location.items(): score = len(terms) * 10 score += sum(1 for keyword in keywords if keyword.lower() in text.lower()) if path.startswith("zhctlogs/"): score += 3 if "zhctproject/store/public/screen/src" in path: score += 25 if re.search(r"(pickupPageSize|pickupPageCount|turnQueuePage|visiblePickupNumbers)", text): score += 35 if path.startswith("zhctprompt/control/") and "deployment-versions" in path: score += 30 if "apifox/openapi" in path or path.endswith("openapi20260415.json"): score -= 20 if "/work" in path or path.endswith(".md"): score += 2 if path.endswith("summary.md"): score += 8 if re.search(r"(原因|修复|结论|链路|补偿)", text): score += 5 if re.search(r"(直接原因|原因已经明确|初步结论|根因)", text): score += 10 if re.search(r"(已完成修复|修复：|默认 `?print_status|缺少 .*cron)", text): score += 10 if re.search(r"(修复负值餐补扣款异常|负值餐补|扣款异常|现金不动)", text): score += 40 if any(term.isdigit() and len(term) >= 8 for term in terms): score += 8 scored.append((score, path, line_no, text, sorted(terms))) scored.sort(key=lambda item: (-item[0], item[1], item[2])) return scored def render_markdown( message: str, keywords: list[str], matches: list[tuple[int, str, int, str, list[str]]], root: Path, max_evidence: int, ) -> str: now = dt.datetime.now().strftime("%Y-%m-%d %H:%M:%S") evidence = matches[:max_evidence] lines = [ "# 企业微信项目群反馈分析草稿", "", f"- 生成时间：{now}", f"- 本地根目录：`{root}`", "", "## 群反馈原文", "", message, "", "## 检索关键词", "", ", ".join(f"`{item}`" for item in keywords) if keywords else "未提取到有效关键词。", "", "## 高相关本地证据", "", ] if evidence: for _, path, line_no, text, terms in evidence: safe_text = text[:240].replace("|", "\\|") lines.append(f"- `{path}:{line_no}` 命中 {', '.join(f'`{term}`' for term in terms)}：{safe_text}") else: lines.append("- 暂未在本地资料、代码和日志中检索到直接命中。建议补充订单号、接口路径、报错文案、门店/设备编号或发生时间。") lines.extend( [ "", "## 分析结论模板", "", "- 初步结论：待结合上方证据和运行环境确认。", "- 可能原因：优先从命中的接口、服务、定时任务、日志错误和历史排障记录中收敛。", "- 下一步动作：如涉及线上数据，先查订单号/设备编号/trace_id；如涉及代码缺陷，再按项目规则新建分支修复。", "", "## 可发群回复草稿", "", "收到，我先按本地项目资料、代码和日志做了初步排查。", ] ) if evidence: top = evidence[0] joined_evidence = "\n".join(item[3] for item in evidence[:5]) if re.search(r"(餐补|补贴|现金|余额)", message) and "修复负值餐补扣款异常" in joined_evidence: lines.extend( [ "初步看，这和已处理过的“负值餐补扣款异常”同类：餐补余额不足/为负时仍优先扣餐补，现金余额没有按预期扣减。", f"本地部署记录显示 `store` 已发布 `V2.38.1`，发布内容为“修复负值餐补扣款异常”，线索在 `{top[1]}:{top[2]}`。", "建议先确认当前问题环境是否已经更新到该版本并重启消费 `work`；如果仍复现，请补一笔订单号/人员/发生时间，我继续追扣款链路和流水。", ] ) elif re.search(r"(叫号|大屏|轮播|取餐)", message) and "pickupPageSize" in joined_evidence: code_evidence = next( (item for item in evidence if "zhctproject/store/public/screen/src/view/callScreen.vue" in item[1]), top, ) lines.extend( [ "看代码逻辑，右侧“请取餐”区域不是一直滚动，而是分页数大于 1 时才轮播。", f"当前右侧每页 `pickupPageSize=6`，`turnQueuePage()` 里只有 `pickupPageCount > 1` 才切页，所以待取餐号超过 6 个后右侧才会轮播；6 个以内会停留在当前页。线索在 `{code_evidence[1]}:{code_evidence[2]}`。", "如果金斯瑞现场希望即使未满 6 个也周期刷新/动画轮播，需要改前端叫号屏交互逻辑。", ] ) else: lines.extend( [ f"目前最相关的线索在 `{top[1]}:{top[2]}`，命中了 {', '.join(f'`{term}`' for term in top[4])}。", "我会继续按订单/接口/日志链路确认实际原因，确认后同步具体结论和处理动作。", ] ) else: lines.extend( [ "当前信息还不够定位到具体代码或日志，请补充发生时间、订单号/设备编号、页面入口或完整报错截图。", "拿到这些信息后我会继续追接口日志和相关代码链路。", ] ) return "\n".join(lines) + "\n" def parse_args() -> argparse.Namespace: parser = argparse.ArgumentParser( description="Search zhct local knowledge/code/logs and draft a WeCom project-group reply." ) parser.add_argument("--message", help="Project group feedback text.") parser.add_argument("--message-file", help="Path to a file containing project group feedback.") parser.add_argument("--keyword", action="append", default=[], help="Extra keyword to search. Can be repeated.") parser.add_argument("--max-per-keyword", type=int, default=25, help="Maximum rg lines kept for each keyword.") parser.add_argument("--max-evidence", type=int, default=20, help="Maximum evidence lines rendered.") parser.add_argument("--output", help="Optional Markdown output path.") return parser.parse_args() def main() -> None: args = parse_args() message = read_message(args) root = project_root(Path(__file__)) dirs = existing_search_dirs(root) keywords = extract_keywords(message, args.keyword) keyword_matches = { keyword: run_rg(root, dirs, keyword, args.max_per_keyword) for keyword in keywords } matches = score_matches(keyword_matches, keywords) markdown = render_markdown(message, keywords, matches, root, args.max_evidence) if args.output: output = Path(args.output) output.parent.mkdir(parents=True, exist_ok=True) output.write_text(markdown, encoding="utf-8") else: print(markdown) if __name__ == "__main__": main()