diff --git a/examples/workflows/issue-triage-automation/README.md b/examples/workflows/issue-triage-automation/README.md new file mode 100644 index 0000000..eb559b7 --- /dev/null +++ b/examples/workflows/issue-triage-automation/README.md @@ -0,0 +1,47 @@ +# Issue 自动分拣工作流(issue-triage-automation) + +补齐起步资源包承诺的三个参考工作流之一「Issue 自动分拣」: + +> **采集 → 分类 → 报告 →(可选)回写**:拉取仓库全部 open issue(自动翻页合并),按规则文件做**确定性分类**(关键词 → 建议优先级/标签/负责人;超龄未更新 → stale),产出 markdown + json 分拣报告;仅在 `--apply` 时把优先级更新与分拣评论真实回写到 GitLink。 + +## 架构 + +```mermaid +flowchart LR + A["issue +list(自动翻页)"] --> B["规则引擎(rules.example.json)"] + B --> C["triage.md / triage.json 报告"] + B -->|"--apply"| D["issue +update --priority-id"] + B -->|"--apply"| E["issue +comment 分拣说明"] +``` + +## 交付物 + +- `scripts/issue_triage.py`:单文件工作流(纯标准库,Python ≥ 3.9,零第三方依赖) +- `rules.example.json`:规则样例(bug / feature / docs / question 四类,含平台优先级 ID 映射:1 低 / 2 正常 / 3 高 / 4 紧急) +- `tests/test_triage.py`:10 个确定性单测(规则命中、stale 判定、首个命中规则生效、同输入同输出、报告排序) + +## 快速运行(默认 dry-run,不写远端) + +```bash +gitlink-cli auth login + +python3 scripts/issue_triage.py --owner --repo --output-dir triage-out +# 产出 triage-out/triage.md(人读)与 triage.json(机读) + +# 真实回写(优先级 + 分拣评论;请先在自有仓库演练) +python3 scripts/issue_triage.py --owner --repo --apply +``` + +## 已在真实平台验证(2026-07-10) + +- **只读分拣**:`Gitlink/forgeplus` 35 个 open issue,命中规则 13 个、stale 32 个,报告确定性可复现 +- **回写闭环**:自有仓库探针 issue(标题含"报错 bug")→ 命中 bug 规则 → `issue +update --priority-id 4` 优先级变为「紧急」+ 分拣评论落盘 → `issue +view` 验证 `priority.name == 紧急`、评论数 +1 → 探针清理 + +## 定制规则 + +编辑 `rules.example.json`:每条规则含 `keywords_any`(标题/描述任一命中,大小写不敏感)、`set_priority` / `set_priority_id`、`add_tags`、`route_to`。规则**从上到下首个命中生效**,顺序即优先级。 + +## CI / Agent 集成 + +- 定时任务:`cron` 每日 dry-run 产出报告,人工确认后 `--apply` +- Agent:本工作流与 `skills/gitlink-issue-triage` Skill 同源互补——Skill 供 Agent 交互式分拣,本脚本供确定性批量闭环 diff --git a/examples/workflows/issue-triage-automation/rules.example.json b/examples/workflows/issue-triage-automation/rules.example.json new file mode 100644 index 0000000..b3bed38 --- /dev/null +++ b/examples/workflows/issue-triage-automation/rules.example.json @@ -0,0 +1,72 @@ +{ + "rules": [ + { + "name": "bug", + "keywords_any": [ + "bug", + "崩溃", + "报错", + "错误", + "失败", + "404", + "500", + "panic" + ], + "set_priority": "紧急", + "add_tags": [ + "bug" + ], + "route_to": null, + "set_priority_id": 4 + }, + { + "name": "feature", + "keywords_any": [ + "feature", + "新增", + "希望", + "建议", + "支持", + "功能请求" + ], + "set_priority": "正常", + "add_tags": [ + "enhancement" + ], + "route_to": null, + "set_priority_id": 2 + }, + { + "name": "docs", + "keywords_any": [ + "文档", + "readme", + "typo", + "错别字", + "翻译" + ], + "set_priority": "低", + "add_tags": [ + "documentation" + ], + "route_to": null, + "set_priority_id": 1 + }, + { + "name": "question", + "keywords_any": [ + "怎么", + "如何", + "请问", + "question", + "how to" + ], + "set_priority": "低", + "add_tags": [ + "question" + ], + "route_to": null, + "set_priority_id": 1 + } + ] +} \ No newline at end of file diff --git a/examples/workflows/issue-triage-automation/scripts/__pycache__/issue_triage.cpython-310.pyc b/examples/workflows/issue-triage-automation/scripts/__pycache__/issue_triage.cpython-310.pyc new file mode 100644 index 0000000..80be13e Binary files /dev/null and b/examples/workflows/issue-triage-automation/scripts/__pycache__/issue_triage.cpython-310.pyc differ diff --git a/examples/workflows/issue-triage-automation/scripts/issue_triage.py b/examples/workflows/issue-triage-automation/scripts/issue_triage.py new file mode 100644 index 0000000..380b7aa --- /dev/null +++ b/examples/workflows/issue-triage-automation/scripts/issue_triage.py @@ -0,0 +1,170 @@ +#!/usr/bin/env python3 +"""Issue 自动分拣端到端工作流(deterministic issue triage)。 + +采集 → 分类 → 报告 →(可选 --apply)回写: +1. 通过 gitlink-cli 拉取仓库 open issue 列表(含分页合并) +2. 按规则文件(JSON,纯标准库解析)做确定性分类: + - 关键词命中 → 建议优先级 / 建议标签 / 建议负责人 + - 超龄未更新 → 标记 stale +3. 产出 triage 报告(markdown + json,同输入必得同输出) +4. 仅在 --apply 时通过 gitlink-cli 真实回写(优先级更新 + 分拣评论) + +零第三方依赖,Python >= 3.9。 +""" + +from __future__ import annotations + +import argparse +import json +import subprocess +import sys +from datetime import datetime, timedelta +from pathlib import Path + +CLI = "gitlink-cli" +DATE_FMT = "%Y-%m-%d %H:%M" + + +# ---------- 纯函数核心(单测覆盖,无网络) ---------- + +def match_rule(issue: dict, rule: dict) -> bool: + """规则命中判定:keywords_any 任一关键词出现在标题或描述(大小写不敏感)。""" + hay = ((issue.get("subject") or "") + "\n" + (issue.get("description") or "")).lower() + return any(kw.lower() in hay for kw in rule.get("keywords_any", [])) + + +def is_stale(issue: dict, now: datetime, stale_days: int) -> bool: + raw = issue.get("updated_at") or issue.get("created_at") or "" + try: + updated = datetime.strptime(raw, DATE_FMT) + except ValueError: + return False + return now - updated > timedelta(days=stale_days) + + +def triage_issue(issue: dict, rules: list[dict], now: datetime, stale_days: int) -> dict: + """对单个 issue 产出确定性分拣决定(首个命中规则生效)。""" + decision = { + "number": issue.get("project_issues_index") or issue.get("number"), + "subject": issue.get("subject", ""), + "rule": None, + "priority": None, + "priority_id": None, + "tags": [], + "assignee": None, + "stale": is_stale(issue, now, stale_days), + } + for rule in rules: + if match_rule(issue, rule): + decision["rule"] = rule.get("name") + decision["priority"] = rule.get("set_priority") + decision["priority_id"] = rule.get("set_priority_id") + decision["tags"] = list(rule.get("add_tags", [])) + decision["assignee"] = rule.get("route_to") + break + return decision + + +def render_report(decisions: list[dict], owner: str, repo: str) -> str: + lines = [ + f"# Issue 分拣报告:{owner}/{repo}", + "", + f"共 {len(decisions)} 个 open issue," + f"命中规则 {sum(1 for d in decisions if d['rule'])} 个," + f"stale {sum(1 for d in decisions if d['stale'])} 个。", + "", + "| # | 标题 | 命中规则 | 建议优先级 | 建议标签 | 建议负责人 | stale |", + "|---|------|----------|------------|----------|------------|-------|", + ] + for d in sorted(decisions, key=lambda x: (x["number"] is None, x["number"])): + lines.append( + "| {number} | {subject} | {rule} | {priority} | {tags} | {assignee} | {stale} |".format( + number=d["number"], + subject=(d["subject"][:40] or "-"), + rule=d["rule"] or "-", + priority=d["priority"] or "-", + tags=",".join(d["tags"]) or "-", + assignee=d["assignee"] or "-", + stale="是" if d["stale"] else "-", + ) + ) + return "\n".join(lines) + "\n" + + +# ---------- CLI 采集与回写 ---------- + +def run_cli(args: list[str]) -> dict: + out = subprocess.run([CLI, *args, "--format", "json"], capture_output=True, text=True) + if out.returncode != 0: + raise RuntimeError(f"{CLI} {' '.join(args)} failed: {out.stderr.strip() or out.stdout.strip()}") + payload = json.loads(out.stdout) + if not payload.get("ok", False): + raise RuntimeError(f"{CLI} {' '.join(args)} returned error: {payload}") + return payload.get("data") or {} + + +def fetch_open_issues(owner: str, repo: str) -> list[dict]: + issues: list[dict] = [] + page = 1 + while True: + data = run_cli([ + "issue", "+list", "--owner", owner, "--repo", repo, + "--state", "open", "--page", str(page), "--limit", "50", + ]) + batch = data.get("issues") or [] + issues.extend(batch) + if len(batch) < 50: + return issues + page += 1 + + +def apply_decision(owner: str, repo: str, decision: dict) -> None: + number = str(decision["number"]) + if decision["priority_id"]: + run_cli([ + "issue", "+update", "--owner", owner, "--repo", repo, + "--number", number, "--priority-id", str(decision["priority_id"]), + ]) + note = f"[issue-triage] 规则「{decision['rule']}」命中:建议标签 {','.join(decision['tags']) or '无'},建议负责人 {decision['assignee'] or '无'}。" + run_cli([ + "issue", "+comment", "--owner", owner, "--repo", repo, + "--number", number, "--body", note, + ]) + + +def main() -> int: + ap = argparse.ArgumentParser(description="GitLink issue 自动分拣工作流") + ap.add_argument("--owner", required=True) + ap.add_argument("--repo", required=True) + ap.add_argument("--rules", default=str(Path(__file__).resolve().parent.parent / "rules.example.json")) + ap.add_argument("--stale-days", type=int, default=30) + ap.add_argument("--output-dir", default="triage-out") + ap.add_argument("--apply", action="store_true", help="真实回写优先级与分拣评论(默认 dry-run)") + args = ap.parse_args() + + rules = json.loads(Path(args.rules).read_text(encoding="utf-8"))["rules"] + issues = fetch_open_issues(args.owner, args.repo) + now = datetime.utcnow() + decisions = [triage_issue(i, rules, now, args.stale_days) for i in issues] + + outdir = Path(args.output_dir) + outdir.mkdir(parents=True, exist_ok=True) + (outdir / "triage.json").write_text( + json.dumps(decisions, ensure_ascii=False, indent=2), encoding="utf-8") + (outdir / "triage.md").write_text(render_report(decisions, args.owner, args.repo), encoding="utf-8") + print(f"triaged {len(decisions)} issues -> {outdir}/triage.md") + + if args.apply: + applied = 0 + for d in decisions: + if d["rule"]: + apply_decision(args.owner, args.repo, d) + applied += 1 + print(f"applied {applied} decisions") + else: + print("dry-run(未写远端);加 --apply 真实回写") + return 0 + + +if __name__ == "__main__": + sys.exit(main()) diff --git a/examples/workflows/issue-triage-automation/tests/__pycache__/test_triage.cpython-310.pyc b/examples/workflows/issue-triage-automation/tests/__pycache__/test_triage.cpython-310.pyc new file mode 100644 index 0000000..6005082 Binary files /dev/null and b/examples/workflows/issue-triage-automation/tests/__pycache__/test_triage.cpython-310.pyc differ diff --git a/examples/workflows/issue-triage-automation/tests/test_triage.py b/examples/workflows/issue-triage-automation/tests/test_triage.py new file mode 100644 index 0000000..56709eb --- /dev/null +++ b/examples/workflows/issue-triage-automation/tests/test_triage.py @@ -0,0 +1,74 @@ +import sys +import unittest +from datetime import datetime +from pathlib import Path + +sys.path.insert(0, str(Path(__file__).resolve().parent.parent / "scripts")) + +from issue_triage import is_stale, match_rule, render_report, triage_issue + +NOW = datetime(2026, 7, 10, 12, 0) +RULES = [ + {"name": "bug", "keywords_any": ["bug", "崩溃"], "set_priority": "紧急", "add_tags": ["bug"], "route_to": "alice"}, + {"name": "docs", "keywords_any": ["文档"], "set_priority": "低", "add_tags": ["documentation"], "route_to": None}, +] + + +class TestMatchRule(unittest.TestCase): + def test_keyword_in_subject_case_insensitive(self): + self.assertTrue(match_rule({"subject": "发现一个 BUG"}, RULES[0])) + + def test_keyword_in_description(self): + self.assertTrue(match_rule({"subject": "问题", "description": "程序崩溃了"}, RULES[0])) + + def test_no_match(self): + self.assertFalse(match_rule({"subject": "hello", "description": "world"}, RULES[0])) + + +class TestStale(unittest.TestCase): + def test_old_issue_is_stale(self): + self.assertTrue(is_stale({"updated_at": "2026-05-01 10:00"}, NOW, 30)) + + def test_fresh_issue_not_stale(self): + self.assertFalse(is_stale({"updated_at": "2026-07-01 10:00"}, NOW, 30)) + + def test_unparseable_date_not_stale(self): + self.assertFalse(is_stale({"updated_at": "n/a"}, NOW, 30)) + + +class TestTriageIssue(unittest.TestCase): + def test_first_matching_rule_wins(self): + issue = {"project_issues_index": 7, "subject": "文档里的 bug", "updated_at": "2026-07-09 10:00"} + d = triage_issue(issue, RULES, NOW, 30) + self.assertEqual(d["rule"], "bug") + self.assertEqual(d["priority"], "紧急") + self.assertEqual(d["tags"], ["bug"]) + self.assertEqual(d["assignee"], "alice") + self.assertFalse(d["stale"]) + + def test_no_rule_matched(self): + d = triage_issue({"project_issues_index": 8, "subject": "hello"}, RULES, NOW, 30) + self.assertIsNone(d["rule"]) + self.assertEqual(d["tags"], []) + + def test_deterministic(self): + issue = {"project_issues_index": 9, "subject": "崩溃", "updated_at": "2026-01-01 00:00"} + self.assertEqual( + triage_issue(issue, RULES, NOW, 30), + triage_issue(issue, RULES, NOW, 30), + ) + + +class TestReport(unittest.TestCase): + def test_report_contains_counts_and_rows(self): + decisions = [ + {"number": 2, "subject": "b", "rule": "bug", "priority": "紧急", "tags": ["bug"], "assignee": None, "stale": True}, + {"number": 1, "subject": "a", "rule": None, "priority": None, "tags": [], "assignee": None, "stale": False}, + ] + md = render_report(decisions, "o", "r") + self.assertIn("共 2 个 open issue,命中规则 1 个,stale 1 个。", md) + self.assertLess(md.index("| 1 |"), md.index("| 2 |")) # 按编号排序 + + +if __name__ == "__main__": + unittest.main()