diff --git a/examples/workflows/community-ops-automation/README.md b/examples/workflows/community-ops-automation/README.md index 181e5cd..7f17334 100644 --- a/examples/workflows/community-ops-automation/README.md +++ b/examples/workflows/community-ops-automation/README.md @@ -2,7 +2,9 @@ 面向 GitLink 竞赛子赛题三的端到端自动化工作流项目。 -本项目面向开源社区运营场景，使用 `gitlink-cli` 串联仓库信息、Issue、PR 和 Release 数据采集，自动生成社区周报、Release Notes 草稿和结构化摘要，并支持将摘要发布到指定 GitLink Issue。该流程覆盖“数据采集 -> 指标分析 -> 文档生成 -> 结果发布”的完整闭环。 +本项目面向开源社区运营场景，使用 `gitlink-cli` 串联 Issue 分拣、责任人分配、仓库信息、Issue、PR 和 Release 数据采集，自动生成社区周报、Release Notes 草稿和结构化摘要，并支持将摘要发布到指定 GitLink Issue。该流程覆盖“新 Issue 自动分类 -> 分配责任人 -> 数据采集 -> 指标分析 -> 文档生成 -> 结果发布”的完整闭环。 + +Issue 分类规则来自自定义 Skill：`D:\BigData\CCFer\CCF\gitlink-issue-triage-rules\SKILL.md`。工作流代码会直接读取 Skill 中的机器可读 JSON 规则块，不再依赖 `workflow +triage` 内置分类结果。 ## 交付物 @@ -39,6 +41,45 @@ - `outputs/*_report.md` - `outputs/*_release_notes.md` - `outputs/*_summary.json` +- `outputs/*_triage_plan.md` +- `outputs/*_triage_plan.json` + +## Issue 自动分类与派单 + +默认运行会执行 dry-run，只生成分类和派单计划，不写回真实 Issue： + +```powershell +.\scripts\run_demo.ps1 +``` + +确认计划无误后，显式开启写回： + +```powershell +.\scripts\run_demo.ps1 -ApplyTriage +``` + +如需显式指定 Skill 路径： + +```powershell +python .\scripts\gitlink_workflow.py --config .\examples\demo_active_config.json --triage-skill D:\BigData\CCFer\CCF\gitlink-issue-triage-rules\SKILL.md +``` + +写回链路会先通过 `issue +list` 拉取开放 Issue，再读取自定义 Skill 规则完成分类；随后通过 `label +list`、`issue +assigners`、`issue +view` 和 `issue +update` 完成标签、优先级和负责人写入。若配置开启 `comment`，还会追加一条审计评论。`*_triage_plan.json` 中会记录 `source=skill-json`、Skill 路径和命中的规则 ID。 + +### 负责人分配配置 + +示例配置为了保证比赛复现稳定，默认把所有类型的 Issue 都分配给 GitLink 用户 ID `153579`。这只是演示配置，不是分类模型自动猜测的结果。实际使用时可以在 `examples/*.json` 的 `triage.assigners_by_type` 中按类型配置不同负责人： + +```json +{ + "bug": [153579], + "feature": [123456], + "question": [], + "docs": [345678] +} +``` + +其中数字会按 GitLink 用户 ID 直接使用，字符串会尝试通过 `issue +assigners` 返回的 login/name 解析；空数组表示该类型只打标签，不分配负责人。 ## 已验证仓库 @@ -48,6 +89,7 @@ ## 项目定位 - 满足子赛题三“端到端自动化工作流”的要求 -- 串联 4 个数据采集命令和 1 个结果发布命令 +- 串联自定义 Skill 规则读取、Issue 分拣/派单命令、4 个数据采集命令和 1 个结果发布命令 +- 新增 Issue 自动分类与负责人分配，形成社区运营前置处理闭环 - 支持在真实 GitLink 项目上复现 - 提供运行脚本、验证记录、示例产物和单元测试 diff --git a/examples/workflows/community-ops-automation/docs/architecture.md b/examples/workflows/community-ops-automation/docs/architecture.md index a6264ed..377795f 100644 --- a/examples/workflows/community-ops-automation/docs/architecture.md +++ b/examples/workflows/community-ops-automation/docs/architecture.md @@ -1,26 +1,41 @@ # 架构说明 -本项目采用“采集 -> 归一化 -> 分析 -> 生成 -> 发布”的五段式流程。 +本项目采用“Issue 分拣 -> 派单 -> 采集 -> 归一化 -> 分析 -> 生成 -> 发布”的七段式流程。 ![GitLink 社区运营端到端自动化工作流架构](assets/architecture-workflow-v2.svg) +```mermaid +flowchart LR + A["issue +list 拉取新 Issue"] --> B["读取自定义 Skill: SKILL.md JSON 规则"] + B --> C["本地规则匹配: 类型、优先级、标签候选"] + C --> D["label +list / issue +assigners 解析 ID"] + D --> E["dry-run 分类与派单计划"] + E --> F["issue +view 合并人工信息"] + F --> G["issue +update / issue +comment 写回"] + G --> H["周报 / Release Notes / 摘要发布"] +``` + ## 设计目标 - 低门槛：只依赖 `gitlink-cli` 和 Python 标准库 - 可复现：同一配置可重复跑出同类报告 - 可维护：采集、归一化、分析、生成和发布步骤保持清晰边界 +- 可控写回：Issue 分类与派单默认 dry-run，显式确认后才更新真实 Issue - 可验证：报告文件、结构化摘要和 Issue 评论均可作为运行结果核验依据 ## 为什么选这个链路 子赛题三要求使用现有命令或 Skill 组合形成完整解决方案。本方案覆盖： -1. 仓库信息采集 -2. Issue 列表采集 -3. PR 列表采集 -4. Release 列表采集 -5. 报告生成 -6. Issue 摘要发布 +1. 新 Issue 自动分类（读取自定义 `gitlink-issue-triage-rules` Skill 中的 JSON 规则） +2. 责任人分配计划生成（Skill 与配置优先） +3. 标签、负责人和优先级写回（`label +list`、`issue +assigners`、`issue +view`、`issue +update`） +4. 仓库信息采集 +5. Issue 列表采集 +6. PR 列表采集 +7. Release 列表采集 +8. 报告生成 +9. Issue 摘要发布 该链路满足不少于 3 个 CLI 调用的要求，并形成从数据获取到结果发布的端到端闭环。 diff --git a/examples/workflows/community-ops-automation/docs/quickstart.md b/examples/workflows/community-ops-automation/docs/quickstart.md index 07d9a43..65124af 100644 --- a/examples/workflows/community-ops-automation/docs/quickstart.md +++ b/examples/workflows/community-ops-automation/docs/quickstart.md @@ -10,6 +10,9 @@ 脚本会自动通过 `npm exec` 找到 `@gitlink-ai/cli`，把 `gitlink-cli` 放到临时 PATH 里，再执行： +- 读取自定义 `gitlink-issue-triage-rules` Skill 规则 +- 新 Issue 自动分类（dry-run） +- 责任人分配计划生成（dry-run） - 仓库信息采集 - Issue 列表采集 - PR 列表采集 @@ -17,6 +20,20 @@ - 周报生成 - Release Notes 草稿生成 +默认不会修改真实 Issue。确认 dry-run 计划后，使用： + +```powershell +.\scripts\run_demo.ps1 -ApplyTriage +``` + +此时会额外执行 `issue +view`、`issue +update` 和可选 `issue +comment`，把分类标签、优先级和负责人写回 GitLink。 + +如需覆盖默认 Skill 路径： + +```powershell +python .\scripts\gitlink_workflow.py --config .\examples\demo_active_config.json --triage-skill D:\BigData\CCFer\CCF\gitlink-issue-triage-rules\SKILL.md +``` + ## 配置切换 - `examples/demo_active_config.json`：公开仓库验证配置，默认指向 `Gitlink/gitlink-cli` @@ -27,6 +44,8 @@ - `outputs/*_report.md` - `outputs/*_release_notes.md` - `outputs/*_summary.json` +- `outputs/*_triage_plan.md` +- `outputs/*_triage_plan.json`，其中 `source` 为 `skill-json`，并记录 Skill 路径、版本和命中规则 ## 已验证事实 diff --git a/examples/workflows/community-ops-automation/docs/runbook.md b/examples/workflows/community-ops-automation/docs/runbook.md index 86087a1..9aedd13 100644 --- a/examples/workflows/community-ops-automation/docs/runbook.md +++ b/examples/workflows/community-ops-automation/docs/runbook.md @@ -5,6 +5,7 @@ - 已安装 `gitlink-cli` - 已完成 `gitlink-cli auth login` - 目标仓库有可读权限 +- 自定义 Skill 文件存在：`D:\BigData\CCFer\CCF\gitlink-issue-triage-rules\SKILL.md` 官方快速开始里要求的验证命令是： @@ -20,32 +21,62 @@ gitlink-cli user +me python .\scripts\gitlink_workflow.py --config .\examples\sample_config.json ``` +该命令默认同时生成 Issue 自动分类与派单 dry-run 计划，但不会写回真实 Issue。 + ### 2. 生成报告并发布摘要 ```powershell python .\scripts\gitlink_workflow.py --config .\examples\sample_config.json --publish-issue-id 123 ``` -### 3. 一键复现 +### 3. 执行真实 Issue 分类与派单 + +```powershell +python .\scripts\gitlink_workflow.py --config .\examples\sample_config.json --apply-triage +``` + +真实写回会先通过 `issue +list` 拉取待处理 Issue，并读取自定义 Skill 的 JSON 规则完成分类；再使用 `label +list`、`issue +assigners`、`issue +view` 和 `issue +update` 写入标签、优先级和负责人。配置中 `triage.comment=true` 时，会通过 `issue +comment` 写入审计说明。 + +如需显式指定 Skill 路径： + +```powershell +python .\scripts\gitlink_workflow.py --config .\examples\sample_config.json --triage-skill D:\BigData\CCFer\CCF\gitlink-issue-triage-rules\SKILL.md +``` + +### 4. 一键复现 ```powershell .\scripts\run_demo.ps1 ``` +写回模式： + +```powershell +.\scripts\run_demo.ps1 -ApplyTriage +``` + ## 输出文件 - `outputs/*_report.md`：完整周报 - `outputs/*_release_notes.md`：Release Notes 草稿 - `outputs/*_summary.json`：结构化摘要 +- `outputs/*_triage_plan.md`：Issue 分类与派单计划 +- `outputs/*_triage_plan.json`：结构化分类与写回结果 ## 验证清单 - `repo +info` 能返回仓库信息 -- `issue +list` 能返回 Issue 列表 +- 自定义 Skill 中存在 `TRIAGE_RULES_JSON_START/END` 规则块 +- `issue +list` 能返回待分类 Issue +- `label +list` 能解析分类标签 ID +- `issue +assigners` 能解析负责人 ID +- dry-run 模式不会调用 `issue +update` +- `*_triage_plan.json` 能记录 `source=skill-json`、Skill 路径和命中规则 - `pr +list` 能返回 PR 列表 - `release +list` 能返回 Release 列表 - 报告文件能落盘 - Release Notes 草稿能落盘 +- apply 模式能把分类标签、优先级和负责人写回 Issue - 发布模式能把摘要写回指定 Issue ## 真实项目配置 diff --git a/examples/workflows/community-ops-automation/docs/submission-checklist.md b/examples/workflows/community-ops-automation/docs/submission-checklist.md index 9041b70..38167ea 100644 --- a/examples/workflows/community-ops-automation/docs/submission-checklist.md +++ b/examples/workflows/community-ops-automation/docs/submission-checklist.md @@ -4,7 +4,7 @@ | 要求 | 本项目对应内容 | | --- | --- | -| 工作流串联不少于 3 个 CLI 命令或 Skill 调用 | `scripts/gitlink_workflow.py` 串联 `repo +info`、`issue +list`、`pr +list`、`release +list`，并支持 `issue +comment` 发布摘要 | +| 工作流串联不少于 3 个 CLI 命令或 Skill 调用 | `scripts/gitlink_workflow.py` 读取自定义 `gitlink-issue-triage-rules` Skill，并串联 `issue +list`、`label +list`、`issue +assigners`、`issue +view`、`issue +update`、`repo +info`、`pr +list`、`release +list` 等命令 | | 提供可复现执行脚本或 Agent 对话记录 | `scripts/run_demo.ps1` | | 在至少一个真实 GitLink 项目上运行并展示效果 | `docs/verification.md`、`docs/demo-output.md`、`examples/demo_outputs/` | | 提供工作流说明文档 | `README.md`、`docs/quickstart.md`、`docs/runbook.md` | diff --git a/examples/workflows/community-ops-automation/docs/verification.md b/examples/workflows/community-ops-automation/docs/verification.md index 38613c2..2d5d203 100644 --- a/examples/workflows/community-ops-automation/docs/verification.md +++ b/examples/workflows/community-ops-automation/docs/verification.md @@ -10,6 +10,11 @@ ### `puygob236/gitlink-cli` +- 自定义 Skill JSON 规则可生成 Issue 分类结果 +- `label +list` 可解析分类标签 +- `issue +assigners` 可解析负责人 +- dry-run 模式可生成 `*_triage_plan.md` 和 `*_triage_plan.json` +- `*_triage_plan.json` 可记录 `source=skill-json`、Skill 路径和命中规则 - `repo +info` 可访问 - `issue +list` 可访问 - `pr +list` 可访问 @@ -18,6 +23,7 @@ ### `Gitlink/gitlink-cli` +- 自定义 Skill JSON 规则可生成 Issue 分类结果 - `repo +info` 可访问 - `issue +list` 可访问 - `pr +list` 可访问 @@ -45,6 +51,8 @@ - `outputs/Gitlink_gitlink-cli_20260520_140525_report.md` - `outputs/Gitlink_gitlink-cli_20260520_140525_release_notes.md` - `outputs/Gitlink_gitlink-cli_20260520_140525_summary.json` +- `outputs/*_triage_plan.md` +- `outputs/*_triage_plan.json` - `outputs/puygob236_gitlink-cli_20260520_143224_report.md` - `outputs/puygob236_gitlink-cli_20260520_143224_release_notes.md` - `outputs/puygob236_gitlink-cli_20260520_143224_summary.json` @@ -65,3 +73,11 @@ ```powershell .\scripts\run_demo.ps1 ``` + +真实写回验证命令： + +```powershell +.\scripts\run_demo.ps1 -ApplyTriage +``` + +未传 `-ApplyTriage` 时只生成 dry-run 计划，不会调用 `issue +update`。 diff --git a/examples/workflows/community-ops-automation/examples/demo_active_config.json b/examples/workflows/community-ops-automation/examples/demo_active_config.json index 83074b7..f8666ab 100644 --- a/examples/workflows/community-ops-automation/examples/demo_active_config.json +++ b/examples/workflows/community-ops-automation/examples/demo_active_config.json @@ -2,5 +2,45 @@ "owner": "Gitlink", "repo": "gitlink-cli", "window_days": 7, - "output_dir": "outputs" + "output_dir": "outputs", + "triage": { + "enabled": true, + "state": "open", + "limit": 50, + "lang": "zh-CN", + "skill_path": "../../../../../gitlink-issue-triage-rules/SKILL.md", + "labels_by_type": { + "bug": "缺陷", + "feature": "功能", + "question": "疑问", + "docs": "文档", + "security": "缺陷", + "performance": "缺陷", + "ci": "测试", + "refactor": "任务", + "duplicate": "重复" + }, + "assigners_by_type": { + "bug": [153579], + "feature": [153579], + "question": [153579], + "docs": [153579], + "security": [153579], + "performance": [153579], + "ci": [153579], + "refactor": [153579], + "duplicate": [153579] + }, + "priority_ids": { + "P0": 4, + "P1": 3, + "P2": 2, + "P3": 1, + "critical": 4, + "high": 3, + "normal": 2, + "low": 1 + }, + "comment": true + } } diff --git a/examples/workflows/community-ops-automation/examples/sample_config.json b/examples/workflows/community-ops-automation/examples/sample_config.json index eaf94d4..daa6844 100644 --- a/examples/workflows/community-ops-automation/examples/sample_config.json +++ b/examples/workflows/community-ops-automation/examples/sample_config.json @@ -2,5 +2,45 @@ "owner": "puygob236", "repo": "gitlink-cli", "window_days": 7, - "output_dir": "outputs" + "output_dir": "outputs", + "triage": { + "enabled": true, + "state": "open", + "limit": 50, + "lang": "zh-CN", + "skill_path": "../../../../../gitlink-issue-triage-rules/SKILL.md", + "labels_by_type": { + "bug": "缺陷", + "feature": "功能", + "question": "疑问", + "docs": "文档", + "security": "缺陷", + "performance": "缺陷", + "ci": "测试", + "refactor": "任务", + "duplicate": "重复" + }, + "assigners_by_type": { + "bug": [153579], + "feature": [153579], + "question": [153579], + "docs": [153579], + "security": [153579], + "performance": [153579], + "ci": [153579], + "refactor": [153579], + "duplicate": [153579] + }, + "priority_ids": { + "P0": 4, + "P1": 3, + "P2": 2, + "P3": 1, + "critical": 4, + "high": 3, + "normal": 2, + "low": 1 + }, + "comment": true + } } diff --git a/examples/workflows/community-ops-automation/scripts/gitlink_workflow.py b/examples/workflows/community-ops-automation/scripts/gitlink_workflow.py index d27be60..5df9ffd 100644 --- a/examples/workflows/community-ops-automation/scripts/gitlink_workflow.py +++ b/examples/workflows/community-ops-automation/scripts/gitlink_workflow.py @@ -3,11 +3,12 @@ from __future__ import annotations import argparse import json import os +import re import subprocess from collections import Counter, defaultdict from datetime import datetime, timedelta, timezone from pathlib import Path -from typing import Any, Iterable +from typing import Any, Callable, Iterable class WorkflowError(RuntimeError): @@ -15,6 +16,49 @@ class WorkflowError(RuntimeError): CLI_PAGE_SIZE = 100 +TRIAGE_RULES_JSON_START = "" +TRIAGE_RULES_JSON_END = "" + +DEFAULT_TRIAGE_CONFIG = { + "enabled": True, + "state": "open", + "limit": 50, + "lang": "zh-CN", + "skill_path": None, + "labels_by_type": { + "bug": "缺陷", + "feature": "功能", + "question": "疑问", + "docs": "文档", + "security": "缺陷", + "performance": "缺陷", + "ci": "测试", + "refactor": "任务", + "duplicate": "重复", + }, + "assigners_by_type": { + "bug": [153579], + "feature": [153579], + "question": [153579], + "docs": [153579], + "security": [153579], + "performance": [153579], + "ci": [153579], + "refactor": [153579], + "duplicate": [153579], + }, + "priority_ids": { + "P0": 4, + "P1": 3, + "P2": 2, + "P3": 1, + "critical": 4, + "high": 3, + "normal": 2, + "low": 1, + }, + "comment": True, +} def parse_args(argv: list[str] | None = None) -> argparse.Namespace: @@ -57,6 +101,30 @@ def parse_args(argv: list[str] | None = None) -> argparse.Namespace: "--cli-bin", help="gitlink-cli 可执行文件路径；可配合 GITLINK_CLI_BIN 使用", ) + parser.add_argument( + "--skip-triage", + action="store_true", + help="跳过新 Issue 自动分类和责任人分配", + ) + parser.add_argument( + "--apply-triage", + action="store_true", + help="执行真实 Issue 写回；默认只生成 dry-run 计划", + ) + parser.add_argument( + "--triage-limit", + type=int, + help="覆盖 triage.limit，控制本轮自动分类 Issue 数量", + ) + parser.add_argument( + "--triage-state", + help="覆盖 triage.state，控制本轮自动分类 Issue 状态", + ) + parser.add_argument( + "--triage-skill", + type=Path, + help="自定义 Issue triage Skill 的 SKILL.md 路径", + ) return parser.parse_args(argv) @@ -155,11 +223,17 @@ def extract_first_dict(payload: Any, keys: Iterable[str]) -> dict[str, Any]: return {} -def run_gitlink_cli(command: list[str], owner: str, repo: str, cwd: Path | None = None) -> Any: - if shutil_which("gitlink-cli") is None: +def run_gitlink_cli( + command: list[str], + owner: str, + repo: str, + cwd: Path | None = None, + cli_bin: str | None = None, +) -> Any: + cli_path = cli_bin or os.environ.get("GITLINK_CLI_BIN") or shutil_which("gitlink-cli") + if cli_path is None: raise WorkflowError("未找到 gitlink-cli，请先安装并确保它在 PATH 中") - cli_path = shutil_which("gitlink-cli") or "gitlink-cli" if cli_path.lower().endswith((".cmd", ".bat")): cmd = [ "cmd", @@ -342,6 +416,753 @@ def normalize_releases(payload: Any) -> list[dict[str, Any]]: return normalized +def as_list(value: Any) -> list[Any]: + if value in (None, "", []): + return [] + if isinstance(value, list): + return value + return [value] + + +def parse_positive_int(value: Any) -> int | None: + if isinstance(value, bool): + return None + if isinstance(value, int) and value > 0: + return value + if isinstance(value, float) and value > 0 and value.is_integer(): + return int(value) + if isinstance(value, str): + text = value.strip() + if text.isdigit(): + parsed = int(text) + if parsed > 0: + return parsed + return None + + +def merge_unique_ints(*groups: Iterable[int]) -> list[int]: + result: list[int] = [] + seen: set[int] = set() + for group in groups: + for value in group: + parsed = parse_positive_int(value) + if parsed is None or parsed in seen: + continue + seen.add(parsed) + result.append(parsed) + return result + + +def normalize_lookup_key(value: Any) -> str: + return str(value).strip().lower() + + +def normalize_triage_config(config: dict[str, Any] | None) -> dict[str, Any]: + merged = dict(DEFAULT_TRIAGE_CONFIG) + if config: + merged.update(config) + for key in ("labels_by_type", "assigners_by_type", "priority_ids"): + nested = dict(DEFAULT_TRIAGE_CONFIG.get(key, {})) + nested.update(config.get(key, {}) or {}) + merged[key] = nested + return merged + + +def strip_json_fence(text: str) -> str: + stripped = text.strip() + if not stripped.startswith("```"): + return stripped + lines = stripped.splitlines() + if lines and lines[0].startswith("```"): + lines = lines[1:] + if lines and lines[-1].strip() == "```": + lines = lines[:-1] + return "\n".join(lines).strip() + + +def extract_skill_frontmatter_value(text: str, key: str) -> str | None: + if not text.startswith("---"): + return None + parts = text.split("---", 2) + if len(parts) < 3: + return None + pattern = re.compile(rf"^\s*{re.escape(key)}\s*:\s*(.+?)\s*$", re.MULTILINE) + match = pattern.search(parts[1]) + if not match: + return None + return match.group(1).strip().strip('"').strip("'") + + +def extract_skill_triage_json(text: str) -> dict[str, Any]: + start = text.find(TRIAGE_RULES_JSON_START) + end = text.find(TRIAGE_RULES_JSON_END) + if start == -1 or end == -1 or end <= start: + raise WorkflowError( + f"Skill 缺少 {TRIAGE_RULES_JSON_START} / {TRIAGE_RULES_JSON_END} 规则块" + ) + raw_block = text[start + len(TRIAGE_RULES_JSON_START):end] + try: + data = json.loads(strip_json_fence(raw_block)) + except json.JSONDecodeError as exc: + raise WorkflowError(f"Skill 规则 JSON 无法解析：{exc}") from exc + if not isinstance(data, dict): + raise WorkflowError("Skill 规则 JSON 必须是对象") + if not isinstance(data.get("rules"), list) or not data["rules"]: + raise WorkflowError("Skill 规则 JSON 必须包含非空 rules 数组") + return data + + +def load_skill_triage_rules(skill_path: Path) -> dict[str, Any]: + if not skill_path.exists(): + raise WorkflowError(f"未找到 triage Skill：{skill_path}") + text = skill_path.read_text(encoding="utf-8") + ruleset = extract_skill_triage_json(text) + skill_meta = ruleset.get("skill") if isinstance(ruleset.get("skill"), dict) else {} + skill_name = ( + skill_meta.get("name") + or ruleset.get("skill_name") + or extract_skill_frontmatter_value(text, "name") + or skill_path.parent.name + ) + skill_version = ( + skill_meta.get("version") + or ruleset.get("skill_version") + or extract_skill_frontmatter_value(text, "version") + or "unknown" + ) + ruleset["skill"] = {"name": str(skill_name), "version": str(skill_version)} + ruleset["skill_path"] = str(skill_path) + return ruleset + + +def candidate_skill_paths( + skill_path: Any = None, + config_path: Path | None = None, +) -> list[Path]: + candidates: list[Path] = [] + if skill_path: + explicit = Path(str(skill_path)).expanduser() + if explicit.is_absolute(): + candidates.append(explicit) + else: + if config_path is not None: + config_base = config_path if config_path.is_absolute() else Path.cwd() / config_path + candidates.append(config_base.resolve().parent / explicit) + candidates.append(Path.cwd() / explicit) + + roots: list[Path] = [Path.cwd(), Path(__file__).resolve()] + if config_path is not None: + roots.append(config_path if config_path.is_absolute() else Path.cwd() / config_path) + seen_roots: set[Path] = set() + for root in roots: + start = root if root.is_dir() else root.parent + for parent in (start, *start.parents): + resolved_parent = parent.resolve() + if resolved_parent in seen_roots: + continue + seen_roots.add(resolved_parent) + candidates.append(resolved_parent / "gitlink-issue-triage-rules" / "SKILL.md") + + unique: list[Path] = [] + seen: set[str] = set() + for candidate in candidates: + key = str(candidate) + if key not in seen: + seen.add(key) + unique.append(candidate) + return unique + + +def resolve_triage_skill_path( + skill_path: Any = None, + config_path: Path | None = None, +) -> Path: + searched = candidate_skill_paths(skill_path, config_path) + for candidate in searched: + if candidate.exists(): + return candidate.resolve() + rendered = "\n".join(f"- {path}" for path in searched[:8]) + raise WorkflowError(f"未找到自定义 triage Skill，已搜索：\n{rendered}") + + +def merge_skill_triage_config( + triage_config: dict[str, Any], + ruleset: dict[str, Any], +) -> dict[str, Any]: + merged = normalize_triage_config(triage_config) + for key in ("labels_by_type", "assigners_by_type", "priority_ids"): + skill_values = ruleset.get(key) + if not isinstance(skill_values, dict): + continue + combined = dict(skill_values) + combined.update(merged.get(key, {}) or {}) + merged[key] = combined + return merged + + +def normalize_text(value: Any) -> str: + return str(value or "").strip().lower() + + +def label_names_for_issue(issue: dict[str, Any]) -> list[str]: + raw = issue.get("raw") if isinstance(issue.get("raw"), dict) else {} + labels = list(issue.get("labels") or []) + labels.extend(normalize_labels(first_value(raw, ("labels", "label_list", "issue_tags", "tags"), []))) + result: list[str] = [] + seen: set[str] = set() + for label in labels: + text = str(label).strip() + key = text.lower() + if text and key not in seen: + seen.add(key) + result.append(text) + return result + + +def issue_body_for_triage(issue: dict[str, Any]) -> str: + raw = issue.get("raw") if isinstance(issue.get("raw"), dict) else {} + return str(first_value(raw, ("description", "body", "content", "desc", "message"), "")) + + +def issue_search_text(issue: dict[str, Any]) -> str: + raw = issue.get("raw") if isinstance(issue.get("raw"), dict) else {} + parts = [ + issue.get("title"), + issue_body_for_triage(issue), + first_value(raw, ("subject", "title", "name"), ""), + " ".join(label_names_for_issue(issue)), + ] + return "\n".join(str(part) for part in parts if part not in (None, "", [])) + + +def contains_any_keyword(text: str, keywords: Any) -> list[str]: + lowered = text.lower() + matched: list[str] = [] + for raw_keyword in as_list(keywords): + keyword = str(raw_keyword).strip() + if keyword and keyword.lower() in lowered: + matched.append(keyword) + return matched + + +def contains_all_keywords(text: str, keywords: Any) -> list[str] | None: + keyword_list = [str(item).strip() for item in as_list(keywords) if str(item).strip()] + if not keyword_list: + return [] + matched = contains_any_keyword(text, keyword_list) + return matched if len(matched) == len(keyword_list) else None + + +def label_condition_matches(labels: list[str], expected: Any, require_absent: bool = False) -> bool: + expected_keys = {normalize_text(item) for item in as_list(expected) if str(item).strip()} + if not expected_keys: + return True + label_keys = {normalize_text(label) for label in labels} + has_match = bool(expected_keys & label_keys) + return not has_match if require_absent else has_match + + +def regex_matches(text: str, patterns: Any) -> list[str]: + matched: list[str] = [] + for raw_pattern in as_list(patterns): + pattern = str(raw_pattern).strip() + if not pattern: + continue + try: + if re.search(pattern, text, flags=re.IGNORECASE): + matched.append(pattern) + except re.error: + continue + return matched + + +def match_skill_rule(issue: dict[str, Any], rule: dict[str, Any]) -> tuple[bool, list[str], int]: + text = issue_search_text(issue) + labels = label_names_for_issue(issue) + match_config = rule.get("match") if isinstance(rule.get("match"), dict) else {} + exclude_config = rule.get("exclude") if isinstance(rule.get("exclude"), dict) else {} + matched_notes: list[str] = [] + + excluded = contains_any_keyword(text, exclude_config.get("any_keyword")) + if excluded: + return False, [f"excluded keyword: {keyword}" for keyword in excluded], 0 + + any_keywords = contains_any_keyword(text, match_config.get("any_keyword")) + if match_config.get("any_keyword") and not any_keywords: + return False, [], 0 + matched_notes.extend(f"keyword: {keyword}" for keyword in any_keywords[:5]) + + all_keywords = contains_all_keywords(text, match_config.get("all_keyword")) + if all_keywords is None: + return False, [], 0 + matched_notes.extend(f"all keyword: {keyword}" for keyword in (all_keywords or [])[:5]) + + matched_regex = regex_matches(text, match_config.get("regex")) + if match_config.get("regex") and not matched_regex: + return False, [], 0 + matched_notes.extend(f"regex: {pattern}" for pattern in matched_regex[:3]) + + if not label_condition_matches(labels, match_config.get("has_label")): + return False, [], 0 + if match_config.get("has_label"): + matched_notes.append(f"has label: {','.join(str(v) for v in as_list(match_config.get('has_label')))}") + + if not label_condition_matches(labels, match_config.get("no_label"), require_absent=True): + return False, [], 0 + + min_length = parse_positive_int(match_config.get("min_description_length")) + if min_length is not None and len(issue_body_for_triage(issue)) < min_length: + return False, [], 0 + if min_length is not None: + matched_notes.append(f"description length >= {min_length}") + + has_any_condition = any( + match_config.get(key) + for key in ("any_keyword", "all_keyword", "regex", "has_label", "no_label", "min_description_length") + ) + if not has_any_condition: + return False, [], 0 + + confidence = 95 if len(matched_notes) > 1 else 85 + return True, matched_notes, confidence + + +def priority_for_rule(rule: dict[str, Any], ruleset: dict[str, Any]) -> str: + defaults = ruleset.get("defaults") if isinstance(ruleset.get("defaults"), dict) else {} + return str(rule.get("priority") or defaults.get("priority") or "P3") + + +def classify_issue_with_skill_rules(issue: dict[str, Any], ruleset: dict[str, Any]) -> dict[str, Any]: + defaults = ruleset.get("defaults") if isinstance(ruleset.get("defaults"), dict) else {} + skip_when = defaults.get("skip_when") if isinstance(defaults.get("skip_when"), dict) else {} + labels = label_names_for_issue(issue) + skipped_labels = [label for label in labels if normalize_text(label) in { + normalize_text(item) for item in as_list(skip_when.get("has_label_any")) + }] + if skipped_labels: + matched_rules = [f"skipped existing label: {label}" for label in skipped_labels] + detected_type = "unknown" + priority = "P3" + confidence = 0 + labels_for_rule: list[str] = [] + rule_id = "" + add_comment = None + else: + detected_type = "unknown" + priority = "P3" + confidence = 20 + matched_rules = [] + labels_for_rule = [] + rule_id = "" + add_comment = None + for raw_rule in ruleset.get("rules", []): + if not isinstance(raw_rule, dict): + continue + matched, notes, score = match_skill_rule(issue, raw_rule) + if not matched: + continue + rule_id = str(raw_rule.get("id") or raw_rule.get("type") or "rule") + detected_type = str(raw_rule.get("type") or "unknown") + priority = priority_for_rule(raw_rule, ruleset) + confidence = score + labels_for_rule = [str(label) for label in as_list(raw_rule.get("label")) if str(label).strip()] + matched_rules = [rule_id, *notes] + add_comment = raw_rule.get("add_comment") + break + + skill_meta = ruleset.get("skill") if isinstance(ruleset.get("skill"), dict) else {} + return { + "issue": { + "number": issue.get("id"), + "title": issue.get("title"), + "labels": labels, + }, + "detected_type": detected_type, + "priority": priority, + "confidence": confidence, + "matched_rules": matched_rules, + "rule_id": rule_id, + "labels": labels_for_rule, + "add_comment": add_comment, + "source": "skill-json", + "skill_name": skill_meta.get("name"), + "skill_version": skill_meta.get("version"), + } + + +def classify_issues_with_skill_rules(issues: list[dict[str, Any]], ruleset: dict[str, Any]) -> dict[str, Any]: + skill_meta = ruleset.get("skill") if isinstance(ruleset.get("skill"), dict) else {} + return { + "source": "skill-json", + "skill": skill_meta, + "skill_path": ruleset.get("skill_path"), + "rules_version": ruleset.get("version"), + "mode": ruleset.get("mode", "rule"), + "results": [classify_issue_with_skill_rules(issue, ruleset) for issue in issues], + } + + +def build_label_index(payload: Any) -> dict[str, int]: + labels = extract_first_list(payload, ("issue_tags", "labels", "items", "list")) + index: dict[str, int] = {} + for item in labels: + if not isinstance(item, dict): + continue + name = first_value(item, ("name", "title", "label_name")) + label_id = parse_positive_int(first_value(item, ("id", "tag_id", "issue_tag_id"))) + if name and label_id is not None: + index[str(name)] = label_id + return index + + +def resolve_label_candidates( + label_candidates: Iterable[Any], + label_index: dict[str, int], +) -> tuple[str | None, int | None]: + normalized_index = {normalize_lookup_key(name): label_id for name, label_id in label_index.items()} + for label_name in label_candidates: + text = str(label_name).strip() + if not text: + continue + label_id = label_index.get(text) + if label_id is None: + label_id = normalized_index.get(normalize_lookup_key(text)) + if label_id is not None: + return text, label_id + return None, None + + +def build_assigner_index(payload: Any) -> dict[str, int]: + assigners = extract_first_list( + payload, + ("assigners", "users", "members", "collaborators", "items", "list"), + ) + index: dict[str, int] = {} + for item in assigners: + if not isinstance(item, dict): + continue + user_id = parse_positive_int(first_value(item, ("id", "user_id", "uid"))) + if user_id is None: + continue + for key in ("id", "user_id", "login", "name", "username", "nickname"): + value = first_value(item, (key,)) + if value not in (None, "", []): + index[normalize_lookup_key(value)] = user_id + return index + + +def object_ids(value: Any) -> list[int]: + ids: list[int] = [] + if isinstance(value, list): + for item in value: + if isinstance(item, dict): + parsed = parse_positive_int(first_value(item, ("id", "user_id", "tag_id"))) + else: + parsed = parse_positive_int(item) + if parsed is not None: + ids.append(parsed) + elif isinstance(value, dict): + parsed = parse_positive_int(first_value(value, ("id", "user_id", "tag_id"))) + if parsed is not None: + ids.append(parsed) + else: + parsed = parse_positive_int(value) + if parsed is not None: + ids.append(parsed) + return merge_unique_ints(ids) + + +def issue_payload_data(payload: Any) -> dict[str, Any]: + data = extract_first_dict(payload, ("issue", "data")) + if data: + return data + return payload if isinstance(payload, dict) else {} + + +def current_issue_metadata(payload: Any) -> tuple[list[int], list[int]]: + issue = issue_payload_data(payload) + tag_ids = object_ids(first_value(issue, ("tags", "issue_tags", "labels"), [])) + assigner_ids = object_ids(first_value(issue, ("assigners", "assigned_users"), [])) + return tag_ids, assigner_ids + + +def resolve_label_for_type( + issue_type: str, + labels_by_type: dict[str, Any], + label_index: dict[str, int], +) -> tuple[str | None, int | None]: + return resolve_label_candidates(as_list(labels_by_type.get(issue_type)), label_index) + + +def resolve_assigner_ids( + issue_type: str, + assigners_by_type: dict[str, Any], + assigner_index: dict[str, int], +) -> tuple[list[int], list[str]]: + ids: list[int] = [] + missing: list[str] = [] + for raw in as_list(assigners_by_type.get(issue_type)): + parsed = parse_positive_int(raw) + if parsed is not None: + ids.append(parsed) + continue + key = normalize_lookup_key(raw) + if key in assigner_index: + ids.append(assigner_index[key]) + elif key: + missing.append(str(raw)) + return merge_unique_ints(ids), missing + + +def priority_id_for_result(result: dict[str, Any], priority_ids: dict[str, Any]) -> int | None: + priority = str(result.get("priority") or "").strip() + aliases = { + "critical": "P0", + "high": "P1", + "normal": "P2", + "low": "P3", + } + candidates = [priority, priority.upper(), priority.lower()] + alias = aliases.get(priority.lower()) + if alias: + candidates.append(alias) + for key in candidates: + parsed = parse_positive_int(priority_ids.get(key)) + if parsed is not None: + return parsed + return None + + +def triage_result_issue(result: dict[str, Any]) -> dict[str, Any]: + issue = result.get("issue") + return issue if isinstance(issue, dict) else {} + + +def triage_issue_number(result: dict[str, Any]) -> int | None: + issue = triage_result_issue(result) + return parse_positive_int(first_value(issue, ("number", "id"))) + + +def triage_issue_title(result: dict[str, Any]) -> str: + issue = triage_result_issue(result) + return str(first_value(issue, ("title",), "(untitled)")) + + +def build_triage_comment(item: dict[str, Any]) -> str: + matched = item.get("matched_rules") or [] + matched_text = ", ".join(str(value) for value in matched[:5]) if matched else "无" + return ( + "🤖 自动分拣：" + f"分类={item.get('detected_type') or 'unknown'}，" + f"标签={item.get('label_name') or '未匹配'}，" + f"负责人={','.join(str(v) for v in item.get('assigner_ids') or []) or '未分配'}，" + f"优先级={item.get('priority') or 'unknown'}，" + f"置信度={item.get('confidence', 0)}，" + f"命中规则={matched_text}" + ) + + +def build_triage_plan( + triage_payload: Any, + label_payload: Any, + assigner_payload: Any, + triage_config: dict[str, Any], + apply_triage: bool = False, + owner: str = "", + repo: str = "", +) -> dict[str, Any]: + labels_by_type = triage_config.get("labels_by_type", {}) or {} + assigners_by_type = triage_config.get("assigners_by_type", {}) or {} + priority_ids = triage_config.get("priority_ids", {}) or {} + label_index = build_label_index(label_payload) + assigner_index = build_assigner_index(assigner_payload) + results = extract_first_list(triage_payload, ("results", "items", "list")) + payload_meta = triage_payload if isinstance(triage_payload, dict) else {} + skill_meta = payload_meta.get("skill") if isinstance(payload_meta.get("skill"), dict) else {} + + warnings: list[str] = [] + items: list[dict[str, Any]] = [] + for result in results: + if not isinstance(result, dict): + continue + number = triage_issue_number(result) + issue_type = str(result.get("detected_type") or "unknown").strip() + priority_id = priority_id_for_result(result, priority_ids) + label_candidates = [label for label in as_list(result.get("labels")) if str(label).strip()] + label_candidates.extend(as_list(labels_by_type.get(issue_type))) + label_name, label_id = resolve_label_candidates(label_candidates, label_index) + assigner_ids, missing_assigners = resolve_assigner_ids(issue_type, assigners_by_type, assigner_index) + status = "planned" + skip_reason = "" + + if number is None: + status = "skipped" + skip_reason = "missing issue number" + warnings.append(f"跳过无法识别编号的 Issue：{triage_issue_title(result)}") + elif issue_type == "unknown": + status = "skipped" + skip_reason = "unknown type" + warnings.append(f"Issue #{number} 未识别出类型，已跳过写回") + elif label_name is None and label_candidates: + warnings.append(f"Issue #{number} 类型 {issue_type} 的候选标签未在仓库中找到") + if issue_type != "unknown" and not assigner_ids and assigners_by_type.get(issue_type): + warnings.append(f"Issue #{number} 类型 {issue_type} 未解析到负责人") + for missing in missing_assigners: + warnings.append(f"Issue #{number} 负责人 {missing} 不在可分配用户列表中") + if issue_type != "unknown" and priority_id is None: + warnings.append(f"Issue #{number} 优先级 {result.get('priority')} 未配置 priority_id") + + desired_tag_ids = [label_id] if label_id is not None else [] + if status == "planned" and not desired_tag_ids and not assigner_ids and priority_id is None: + status = "skipped" + skip_reason = "no resolvable metadata" + + items.append( + { + "issue_number": number, + "title": triage_issue_title(result), + "detected_type": issue_type, + "priority": result.get("priority"), + "priority_id": priority_id, + "confidence": result.get("confidence", 0), + "label_name": label_name, + "label_id": label_id, + "desired_tag_ids": desired_tag_ids, + "assigner_ids": assigner_ids, + "matched_rules": result.get("matched_rules") or [], + "rule_id": result.get("rule_id") or "", + "source": result.get("source") or payload_meta.get("source") or "", + "skill_name": result.get("skill_name") or skill_meta.get("name"), + "skill_version": result.get("skill_version") or skill_meta.get("version"), + "status": status, + "skip_reason": skip_reason, + } + ) + + return { + "repository": f"{owner}/{repo}" if owner and repo else "", + "source": payload_meta.get("source") or "skill-json", + "skill_path": payload_meta.get("skill_path"), + "skill_name": skill_meta.get("name"), + "skill_version": skill_meta.get("version"), + "rules_version": payload_meta.get("rules_version"), + "dry_run": not apply_triage, + "enabled": True, + "analyzed": len(items), + "items": items, + "warnings": warnings, + } + + +def apply_triage_plan( + plan: dict[str, Any], + owner: str, + repo: str, + runner: Callable[..., Any], + cli_bin: str | None = None, + comment: bool = True, +) -> dict[str, Any]: + warnings = plan.setdefault("warnings", []) + for item in plan.get("items", []): + if item.get("status") != "planned": + continue + number = item.get("issue_number") + try: + issue_payload = runner(["issue", "+view", "--number", str(number)], owner, repo, cli_bin=cli_bin) + current_tag_ids, current_assigner_ids = current_issue_metadata(issue_payload) + final_tag_ids = merge_unique_ints(current_tag_ids, item.get("desired_tag_ids") or []) + final_assigner_ids = merge_unique_ints(current_assigner_ids, item.get("assigner_ids") or []) + + command = ["issue", "+update", "--number", str(number)] + if final_tag_ids: + command += ["--tag-ids", ",".join(str(value) for value in final_tag_ids)] + if final_assigner_ids: + command += ["--assigner-ids", ",".join(str(value) for value in final_assigner_ids)] + if item.get("priority_id"): + command += ["--priority-id", str(item["priority_id"])] + item["existing_tag_ids"] = current_tag_ids + item["existing_assigner_ids"] = current_assigner_ids + item["final_tag_ids"] = final_tag_ids + item["final_assigner_ids"] = final_assigner_ids + item["update_command"] = command + item["update_result"] = runner(command, owner, repo, cli_bin=cli_bin) + item["status"] = "applied" + if comment: + comment_command = build_issue_comment_command(int(number), build_triage_comment(item)) + try: + item["comment_result"] = runner(comment_command, owner, repo, cli_bin=cli_bin) + except Exception as exc: # noqa: BLE001 + item["comment_error"] = str(exc) + warnings.append(f"Issue #{number} 审计评论写入失败：{exc}") + except Exception as exc: # noqa: BLE001 + item["status"] = "failed" + item["error"] = str(exc) + warnings.append(f"Issue #{number} 写回失败：{exc}") + plan["dry_run"] = False + return plan + + +def summarize_triage_plan(plan: dict[str, Any]) -> dict[str, Any]: + items = plan.get("items", []) + statuses = Counter(str(item.get("status", "unknown")) for item in items) + return { + "enabled": plan.get("enabled", False), + "dry_run": plan.get("dry_run", True), + "source": plan.get("source"), + "skill_path": plan.get("skill_path"), + "skill_name": plan.get("skill_name"), + "skill_version": plan.get("skill_version"), + "rules_version": plan.get("rules_version"), + "analyzed": plan.get("analyzed", len(items)), + "planned": statuses.get("planned", 0) + statuses.get("applied", 0) + statuses.get("failed", 0), + "succeeded": statuses.get("applied", 0), + "failed": statuses.get("failed", 0), + "skipped": statuses.get("skipped", 0), + "warnings": plan.get("warnings", []), + } + + +def render_triage_plan_markdown(plan: dict[str, Any]) -> str: + mode = "dry-run" if plan.get("dry_run", True) else "applied" + lines = [ + "# Issue 自动分类与派单计划", + "", + f"- 仓库：`{plan.get('repository') or 'unknown'}`", + f"- 模式：`{mode}`", + f"- 分类来源：`{plan.get('source') or 'unknown'}`", + f"- Skill：`{plan.get('skill_name') or 'unknown'}@{plan.get('skill_version') or 'unknown'}`", + f"- Skill 路径：`{plan.get('skill_path') or '-'}`", + f"- 分析 Issue：{plan.get('analyzed', 0)} 条", + "", + "| Issue | 标题 | 类型 | 标签 | 负责人ID | 优先级 | 命中规则 | 状态 | 说明 |", + "| --- | --- | --- | --- | --- | --- | --- | --- | --- |", + ] + for item in plan.get("items", []): + title = str(item.get("title", "")).replace("|", "\\|").replace("\n", " ") + assigners = ",".join(str(value) for value in item.get("assigner_ids") or []) or "-" + rule = str(item.get("rule_id") or ",".join(str(v) for v in (item.get("matched_rules") or [])[:1]) or "-") + rule_text = rule.replace("|", "\\|") + note = item.get("skip_reason") or item.get("error") or "" + note_text = str(note).replace("|", "\\|") + lines.append( + "| " + f"#{item.get('issue_number') or '-'} | " + f"{title} | " + f"{item.get('detected_type') or '-'} | " + f"{item.get('label_name') or '-'} | " + f"{assigners} | " + f"{item.get('priority') or '-'} | " + f"{rule_text} | " + f"{item.get('status') or '-'} | " + f"{note_text} |" + ) + if plan.get("warnings"): + lines.extend(["", "## 警告"]) + for warning in plan["warnings"]: + lines.append(f"- {warning}") + return "\n".join(lines).rstrip() + "\n" + + def is_open(state: str) -> bool: return state == "open" @@ -391,6 +1212,7 @@ def fetch_paginated_payload( repo: str, item_keys: tuple[str, ...], page_size: int = CLI_PAGE_SIZE, + cli_bin: str | None = None, ) -> list[dict[str, Any]]: items: list[dict[str, Any]] = [] page = 1 @@ -402,6 +1224,7 @@ def fetch_paginated_payload( [*command, "--page", str(page), "--limit", str(page_size)], owner, repo, + cli_bin=cli_bin, ) page_items = extract_first_list(payload, item_keys) page_items = [item for item in page_items if isinstance(item, dict)] @@ -414,7 +1237,7 @@ def fetch_paginated_payload( return items -def fetch_issues(owner: str, repo: str) -> list[dict[str, Any]]: +def fetch_issues(owner: str, repo: str, cli_bin: str | None = None) -> list[dict[str, Any]]: records: list[dict[str, Any]] = [] for state in ("open", "closed"): payloads = fetch_paginated_payload( @@ -422,12 +1245,43 @@ def fetch_issues(owner: str, repo: str) -> list[dict[str, Any]]: owner, repo, ("issues", "issue_list", "items", "list"), + cli_bin=cli_bin, ) records.extend(normalize_issues({"issues": payloads}, query_state=state)) return dedupe_records(records) -def fetch_prs(owner: str, repo: str) -> list[dict[str, Any]]: +def fetch_triage_issues( + owner: str, + repo: str, + state: str, + limit: int, + runner: Callable[..., Any], + cli_bin: str | None = None, +) -> list[dict[str, Any]]: + records: list[dict[str, Any]] = [] + page = 1 + page_size = min(CLI_PAGE_SIZE, max(limit, 1)) + max_pages = 50 + while len(records) < limit and page <= max_pages: + payload = runner( + ["issue", "+list", "--state", state, "--page", str(page), "--limit", str(page_size)], + owner, + repo, + cli_bin=cli_bin, + ) + page_items = extract_first_list(payload, ("issues", "issue_list", "items", "list")) + page_items = [item for item in page_items if isinstance(item, dict)] + if not page_items: + break + records.extend(normalize_issues({"issues": page_items}, query_state=state)) + if len(page_items) < page_size: + break + page += 1 + return dedupe_records(records)[:limit] + + +def fetch_prs(owner: str, repo: str, cli_bin: str | None = None) -> list[dict[str, Any]]: records: list[dict[str, Any]] = [] for state in ("open", "merged", "closed"): payloads = fetch_paginated_payload( @@ -435,17 +1289,19 @@ def fetch_prs(owner: str, repo: str) -> list[dict[str, Any]]: owner, repo, ("pull_requests", "merge_requests", "prs", "items", "list"), + cli_bin=cli_bin, ) records.extend(normalize_prs({"pull_requests": payloads}, query_state=state)) return dedupe_records(records) -def fetch_releases(owner: str, repo: str) -> list[dict[str, Any]]: +def fetch_releases(owner: str, repo: str, cli_bin: str | None = None) -> list[dict[str, Any]]: payloads = fetch_paginated_payload( ["release", "+list"], owner, repo, ("releases", "items", "list"), + cli_bin=cli_bin, ) return dedupe_records(normalize_releases({"releases": payloads})) @@ -568,6 +1424,19 @@ def render_markdown_report(summary: dict[str, Any]) -> str: lines.append(f"| Release 数 | {counts['releases_total']} |") lines.append("") + triage = summary.get("triage") + if isinstance(triage, dict) and triage.get("enabled"): + mode = "dry-run" if triage.get("dry_run", True) else "已写回" + lines.append("## Issue 自动分类与派单") + lines.append("") + lines.append(f"- 模式：{mode}") + lines.append(f"- 分析 Issue：{triage.get('analyzed', 0)} 条") + lines.append(f"- 计划写回：{triage.get('planned', 0)} 条") + lines.append(f"- 写回成功：{triage.get('succeeded', 0)} 条") + lines.append(f"- 写回失败：{triage.get('failed', 0)} 条") + lines.append(f"- 跳过：{triage.get('skipped', 0)} 条") + lines.append("") + lines.append("## 热点标签") if summary["labels"]: for label, count in summary["labels"]: @@ -664,6 +1533,13 @@ def render_publish_comment( ] if release_notes_path is not None: lines.append(f"Release Notes 草稿：`{release_notes_path.as_posix()}`") + triage = summary.get("triage") + if isinstance(triage, dict) and triage.get("enabled"): + mode = "dry-run" if triage.get("dry_run", True) else "已写回" + lines.append( + f"- Issue 自动分类与派单：{mode}，分析 {triage.get('analyzed', 0)} 条，" + f"成功 {triage.get('succeeded', 0)} 条" + ) if summary["actions"]: lines.append("") lines.append("### 建议动作") @@ -695,6 +1571,57 @@ def shutil_which(name: str) -> str | None: return which(name) +def run_triage_workflow( + owner: str, + repo: str, + triage_config: dict[str, Any], + apply_triage: bool, + output_dir: Path, + base_name: str, + cli_bin: str | None = None, + limit_override: int | None = None, + state_override: str | None = None, + triage_skill: Path | None = None, + config_path: Path | None = None, + runner: Callable[..., Any] = run_gitlink_cli, +) -> tuple[dict[str, Any], Path, Path]: + config = normalize_triage_config(triage_config) + state = state_override or str(config.get("state") or "open") + limit = limit_override or int(config.get("limit") or 50) + skill_path = resolve_triage_skill_path(triage_skill or config.get("skill_path"), config_path=config_path) + ruleset = load_skill_triage_rules(skill_path) + config = merge_skill_triage_config(config, ruleset) + + issues = fetch_triage_issues(owner, repo, state, limit, runner, cli_bin=cli_bin) + triage_payload = classify_issues_with_skill_rules(issues, ruleset) + label_payload = runner(["label", "+list"], owner, repo, cli_bin=cli_bin) + assigner_payload = runner(["issue", "+assigners"], owner, repo, cli_bin=cli_bin) + plan = build_triage_plan( + triage_payload, + label_payload, + assigner_payload, + config, + apply_triage=apply_triage, + owner=owner, + repo=repo, + ) + if apply_triage: + plan = apply_triage_plan( + plan, + owner, + repo, + runner, + cli_bin=cli_bin, + comment=bool(config.get("comment", True)), + ) + + markdown_path = output_dir / f"{base_name}_triage_plan.md" + json_path = output_dir / f"{base_name}_triage_plan.json" + markdown_path.write_text(render_triage_plan_markdown(plan), encoding="utf-8") + json_path.write_text(json.dumps(plan, ensure_ascii=False, indent=2, default=str), encoding="utf-8") + return plan, markdown_path, json_path + + def build_artifacts( owner: str, repo: str, @@ -703,19 +1630,27 @@ def build_artifacts( now: datetime, publish_issue_id: int | None, skip_releases: bool, + triage_config: dict[str, Any] | None = None, + skip_triage: bool = False, + apply_triage: bool = False, + triage_limit: int | None = None, + triage_state: str | None = None, + triage_skill: Path | None = None, + config_path: Path | None = None, + cli_bin: str | None = None, ) -> tuple[dict[str, Any], Path, Path, Path, list[str]]: warnings: list[str] = [] repo_info = safe_fetch( "repo +info", - lambda: normalize_repo_info(run_gitlink_cli(["repo", "+info"], owner, repo)), + lambda: normalize_repo_info(run_gitlink_cli(["repo", "+info"], owner, repo, cli_bin=cli_bin)), warnings, {"name": repo, "description": "", "default_branch": "", "language": "", "raw": {}}, ) - issues = safe_fetch("issue +list", lambda: fetch_issues(owner, repo), warnings, []) - prs = safe_fetch("pr +list", lambda: fetch_prs(owner, repo), warnings, []) + issues = safe_fetch("issue +list", lambda: fetch_issues(owner, repo, cli_bin=cli_bin), warnings, []) + prs = safe_fetch("pr +list", lambda: fetch_prs(owner, repo, cli_bin=cli_bin), warnings, []) releases = [] if skip_releases else safe_fetch( "release +list", - lambda: fetch_releases(owner, repo), + lambda: fetch_releases(owner, repo, cli_bin=cli_bin), warnings, [], ) @@ -733,6 +1668,52 @@ def build_artifacts( report_path = output_dir / f"{base_name}_report.md" summary_path = output_dir / f"{base_name}_summary.json" release_notes_path = output_dir / f"{base_name}_release_notes.md" + artifacts: dict[str, str] = { + "report": report_path.as_posix(), + "summary": summary_path.as_posix(), + "release_notes": release_notes_path.as_posix(), + } + + if skip_triage: + summary["triage"] = {"enabled": False, "dry_run": True, "analyzed": 0} + else: + active_triage_config = normalize_triage_config(triage_config or {}) + if not bool(active_triage_config.get("enabled", True)): + summary["triage"] = {"enabled": False, "dry_run": True, "analyzed": 0} + else: + try: + triage_plan, triage_markdown_path, triage_json_path = run_triage_workflow( + owner=owner, + repo=repo, + triage_config=active_triage_config, + apply_triage=apply_triage, + output_dir=output_dir, + base_name=base_name, + cli_bin=cli_bin, + limit_override=triage_limit, + state_override=triage_state, + triage_skill=triage_skill, + config_path=config_path, + ) + summary["triage"] = summarize_triage_plan(triage_plan) + artifacts["triage_plan"] = triage_markdown_path.as_posix() + artifacts["triage_plan_json"] = triage_json_path.as_posix() + warnings.extend(triage_plan.get("warnings", [])) + except Exception as exc: # noqa: BLE001 + warning = f"Skill triage 失败：{exc}" + warnings.append(warning) + no_issues = "no issues found" in str(exc).lower() + summary["triage"] = { + "enabled": True, + "dry_run": not apply_triage, + "source": "skill-json", + "analyzed": 0, + "planned": 0, + "succeeded": 0, + "failed": 0 if no_issues else 1, + "skipped": 0, + "warnings": [warning], + } report_text = render_markdown_report(summary) release_notes_text = render_release_notes(summary) @@ -744,11 +1725,7 @@ def build_artifacts( **summary, "now": summary["now"].isoformat(), "cutoff": summary["cutoff"].isoformat(), - "artifacts": { - "report": report_path.as_posix(), - "summary": summary_path.as_posix(), - "release_notes": release_notes_path.as_posix(), - }, + "artifacts": artifacts, }, ensure_ascii=False, indent=2, @@ -764,6 +1741,7 @@ def build_artifacts( build_issue_comment_command(publish_issue_id, comment), owner, repo, + cli_bin=cli_bin, ) except Exception as exc: # noqa: BLE001 warnings.append(f"issue +comment 失败：{exc}") @@ -783,6 +1761,7 @@ def main(argv: list[str] | None = None) -> int: window_days = args.window_days or int(config.get("window_days", 7)) output_dir = args.output_dir or Path(config.get("output_dir", "outputs")) now = parse_iso_now(args.now) + triage_config = normalize_triage_config(config.get("triage") or {}) summary, report_path, summary_path, release_notes_path, warnings = build_artifacts( owner=owner, @@ -792,11 +1771,30 @@ def main(argv: list[str] | None = None) -> int: now=now, publish_issue_id=args.publish_issue_id, skip_releases=args.skip_releases, + triage_config=triage_config, + skip_triage=args.skip_triage, + apply_triage=args.apply_triage, + triage_limit=args.triage_limit, + triage_state=args.triage_state, + triage_skill=args.triage_skill, + config_path=args.config, + cli_bin=args.cli_bin, ) print(f"已生成报告: {report_path}") print(f"已生成摘要: {summary_path}") print(f"已生成 Release Notes: {release_notes_path}") + triage = summary.get("triage") + if isinstance(triage, dict) and triage.get("enabled"): + mode = "dry-run" if triage.get("dry_run", True) else "已写回" + print( + "Issue 自动分类与派单: " + f"{mode}, analyzed={triage.get('analyzed', 0)}, " + f"planned={triage.get('planned', 0)}, " + f"succeeded={triage.get('succeeded', 0)}, " + f"failed={triage.get('failed', 0)}, " + f"skipped={triage.get('skipped', 0)}" + ) if warnings: print("警告:") for warning in warnings: diff --git a/examples/workflows/community-ops-automation/scripts/run_demo.ps1 b/examples/workflows/community-ops-automation/scripts/run_demo.ps1 index 5ad872b..185040d 100644 --- a/examples/workflows/community-ops-automation/scripts/run_demo.ps1 +++ b/examples/workflows/community-ops-automation/scripts/run_demo.ps1 @@ -5,7 +5,10 @@ param( [int]$WindowDays = 7, [string]$OutputDir = "outputs", [int]$PublishIssueId = 0, - [switch]$SkipReleases + [switch]$SkipReleases, + [switch]$SkipTriage, + [switch]$ApplyTriage, + [string]$TriageSkill = "" ) $ErrorActionPreference = "Stop" @@ -26,7 +29,8 @@ $args = @( "scripts\gitlink_workflow.py", "--config", $Config, "--window-days", "$WindowDays", - "--output-dir", $OutputDir + "--output-dir", $OutputDir, + "--cli-bin", $cliPath ) if ($Owner) { @@ -41,5 +45,14 @@ if ($PublishIssueId -gt 0) { if ($SkipReleases.IsPresent) { $args += "--skip-releases" } +if ($SkipTriage.IsPresent) { + $args += "--skip-triage" +} +if ($ApplyTriage.IsPresent) { + $args += "--apply-triage" +} +if ($TriageSkill) { + $args += @("--triage-skill", $TriageSkill) +} python @args diff --git a/examples/workflows/community-ops-automation/tests/test_gitlink_workflow.py b/examples/workflows/community-ops-automation/tests/test_gitlink_workflow.py index 8025bf8..672762f 100644 --- a/examples/workflows/community-ops-automation/tests/test_gitlink_workflow.py +++ b/examples/workflows/community-ops-automation/tests/test_gitlink_workflow.py @@ -1,15 +1,23 @@ from __future__ import annotations +import json import unittest from datetime import datetime, timezone +from pathlib import Path +from tempfile import TemporaryDirectory from scripts.gitlink_workflow import ( + apply_triage_plan, + build_triage_plan, build_issue_comment_command, + classify_issue_with_skill_rules, + load_skill_triage_rules, normalize_issues, normalize_prs, normalize_releases, render_markdown_report, render_release_notes, + run_triage_workflow, summarize_workflow, ) @@ -23,6 +31,80 @@ class WorkflowTests(unittest.TestCase): "default_branch": "master", } + def write_skill(self, directory: str) -> Path: + ruleset = { + "version": 1, + "mode": "rule", + "skill": {"name": "gitlink-issue-triage-rules", "version": "9.9.9"}, + "defaults": {"dry_run": True, "priority": "P3"}, + "priority_ids": {"P0": 4, "P1": 3, "P2": 2, "P3": 1, "critical": 4, "high": 3, "normal": 2, "low": 1}, + "labels_by_type": { + "bug": ["缺陷", "bug"], + "feature": ["功能", "enhancement"], + "question": ["疑问", "question"], + "docs": ["文档", "documentation"], + "security": ["缺陷", "security"], + }, + "assigners_by_type": { + "bug": [153579], + "feature": [153579], + "question": [153579], + "docs": [153579], + "security": [153579], + }, + "rules": [ + { + "id": "security-sensitive", + "type": "security", + "label": ["缺陷", "security"], + "priority": "critical", + "match": {"any_keyword": ["token", "漏洞"]}, + }, + { + "id": "bug-crash", + "type": "bug", + "label": ["缺陷", "bug"], + "priority": "high", + "match": {"any_keyword": ["crash", "崩溃"]}, + }, + { + "id": "feature-request", + "type": "feature", + "label": ["功能", "enhancement"], + "priority": "normal", + "match": {"any_keyword": ["建议", "feature"]}, + }, + { + "id": "question-default", + "type": "question", + "label": ["疑问", "question"], + "priority": "low", + "match": {"any_keyword": ["请问", "how to"]}, + }, + { + "id": "docs-default", + "type": "docs", + "label": ["文档", "documentation"], + "priority": "low", + "match": {"any_keyword": ["README", "typo"]}, + }, + ], + } + path = Path(directory) / "SKILL.md" + path.write_text( + "---\n" + "name: gitlink-issue-triage-rules\n" + "version: 9.9.9\n" + "---\n\n" + "\n" + "```json\n" + f"{json.dumps(ruleset, ensure_ascii=False, indent=2)}\n" + "```\n" + "\n", + encoding="utf-8", + ) + return path + def test_normalize_issue_payload(self) -> None: payload = { "data": { @@ -128,6 +210,189 @@ class WorkflowTests(unittest.TestCase): self.assertEqual(command, ["issue", "+comment", "--number", "2", "--body", "demo"]) self.assertNotIn("-i", command) + def test_load_skill_triage_rules_extracts_json_block(self) -> None: + with TemporaryDirectory() as tmp: + skill_path = self.write_skill(tmp) + ruleset = load_skill_triage_rules(skill_path) + + self.assertEqual(ruleset["skill"]["name"], "gitlink-issue-triage-rules") + self.assertEqual(ruleset["skill"]["version"], "9.9.9") + self.assertEqual(len(ruleset["rules"]), 5) + + def test_classify_issue_with_skill_rules_matches_common_types(self) -> None: + with TemporaryDirectory() as tmp: + ruleset = load_skill_triage_rules(self.write_skill(tmp)) + + cases = [ + ("token 泄露漏洞", "security", "security-sensitive"), + ("CLI crash when upload", "bug", "bug-crash"), + ("建议新增报表导出", "feature", "feature-request"), + ("请问 how to 配置项目", "question", "question-default"), + ("README typo", "docs", "docs-default"), + ] + for title, expected_type, expected_rule in cases: + result = classify_issue_with_skill_rules( + {"id": "1", "title": title, "labels": [], "raw": {"description": title}}, + ruleset, + ) + self.assertEqual(result["detected_type"], expected_type) + self.assertEqual(result["rule_id"], expected_rule) + + def test_classify_issue_with_skill_rules_marks_unknown(self) -> None: + with TemporaryDirectory() as tmp: + ruleset = load_skill_triage_rules(self.write_skill(tmp)) + + result = classify_issue_with_skill_rules( + {"id": "9", "title": "General note", "labels": [], "raw": {"description": "nothing special"}}, + ruleset, + ) + + self.assertEqual(result["detected_type"], "unknown") + self.assertEqual(result["source"], "skill-json") + + def test_build_triage_plan_resolves_label_and_assigner(self) -> None: + plan = build_triage_plan( + { + "results": [ + { + "issue": {"number": 1, "title": "CLI crash"}, + "detected_type": "bug", + "priority": "P1", + "confidence": 88, + "matched_rules": ["matched keyword: crash"], + } + ] + }, + {"data": {"issue_tags": [{"id": 10, "name": "缺陷"}]}}, + {"data": {"assigners": [{"id": 153579, "login": "Angel123456", "name": "Angel"}]}}, + { + "labels_by_type": {"bug": "缺陷"}, + "assigners_by_type": {"bug": ["Angel123456"]}, + "priority_ids": {"P1": 3}, + }, + owner="owner", + repo="repo", + ) + + self.assertEqual(plan["analyzed"], 1) + item = plan["items"][0] + self.assertEqual(item["status"], "planned") + self.assertEqual(item["label_id"], 10) + self.assertEqual(item["assigner_ids"], [153579]) + self.assertEqual(item["priority_id"], 3) + + def test_build_triage_plan_warns_without_interrupting(self) -> None: + plan = build_triage_plan( + { + "results": [ + { + "issue": {"number": 2, "title": "General note"}, + "detected_type": "unknown", + "priority": "P3", + }, + { + "issue": {"number": 3, "title": "Slow API"}, + "detected_type": "performance", + "priority": "P2", + }, + ] + }, + {"data": {"issue_tags": []}}, + {"data": {"assigners": []}}, + { + "labels_by_type": {"performance": "缺陷"}, + "assigners_by_type": {"performance": ["missing-user"]}, + "priority_ids": {"P2": 2, "P3": 1}, + }, + ) + + self.assertEqual(plan["items"][0]["status"], "skipped") + self.assertEqual(plan["items"][1]["status"], "planned") + self.assertGreaterEqual(len(plan["warnings"]), 3) + + def test_apply_triage_plan_merges_existing_metadata_and_comments(self) -> None: + plan = build_triage_plan( + { + "results": [ + { + "issue": {"number": 4, "title": "Bug: upload crash"}, + "detected_type": "bug", + "priority": "P1", + "confidence": 91, + } + ] + }, + {"data": {"issue_tags": [{"id": 10, "name": "缺陷"}]}}, + {"data": {"assigners": [{"id": 153579, "login": "Angel123456"}]}}, + { + "labels_by_type": {"bug": "缺陷"}, + "assigners_by_type": {"bug": [153579]}, + "priority_ids": {"P1": 3}, + }, + ) + calls: list[list[str]] = [] + + def fake_runner(command: list[str], owner: str, repo: str, cli_bin: str | None = None): + calls.append(command) + if command[:2] == ["issue", "+view"]: + return { + "data": { + "subject": "Bug: upload crash", + "description": "existing body", + "tags": [{"id": 99, "name": "已有"}], + "assigners": [{"id": 88, "login": "maintainer"}], + } + } + return {"ok": True} + + applied = apply_triage_plan(plan, "owner", "repo", fake_runner, comment=True) + self.assertEqual(applied["items"][0]["status"], "applied") + self.assertEqual(calls[0][:2], ["issue", "+view"]) + self.assertEqual(calls[1][:2], ["issue", "+update"]) + self.assertEqual(calls[2][:2], ["issue", "+comment"]) + self.assertIn("--tag-ids", calls[1]) + self.assertIn("99,10", calls[1]) + self.assertIn("--assigner-ids", calls[1]) + self.assertIn("88,153579", calls[1]) + self.assertIn("--priority-id", calls[1]) + self.assertIn("3", calls[1]) + + def test_run_triage_workflow_dry_run_does_not_update_issue(self) -> None: + calls: list[list[str]] = [] + + def fake_runner(command: list[str], owner: str, repo: str, cli_bin: str | None = None): + calls.append(command) + if command[:2] == ["issue", "+list"]: + return {"data": {"issues": [{"project_issues_index": 1, "subject": "CLI crash", "description": "crash"}]}} + if command[:2] == ["label", "+list"]: + return {"data": {"issue_tags": [{"id": 10, "name": "缺陷"}]}} + if command[:2] == ["issue", "+assigners"]: + return {"data": {"assigners": [{"id": 153579, "login": "Angel123456"}]}} + raise AssertionError(f"unexpected command in dry-run: {command}") + + with TemporaryDirectory() as tmp: + plan, markdown_path, json_path = run_triage_workflow( + "owner", + "repo", + { + "labels_by_type": {"bug": "缺陷"}, + "assigners_by_type": {"bug": [153579]}, + "priority_ids": {"P1": 3}, + }, + apply_triage=False, + output_dir=Path(tmp), + base_name="demo", + triage_skill=self.write_skill(tmp), + runner=fake_runner, + ) + + self.assertTrue(plan["dry_run"]) + self.assertEqual(plan["source"], "skill-json") + self.assertEqual(plan["items"][0]["rule_id"], "bug-crash") + self.assertTrue(markdown_path.name.endswith("_triage_plan.md")) + self.assertTrue(json_path.name.endswith("_triage_plan.json")) + self.assertNotIn(["issue", "+update"], [call[:2] for call in calls]) + if __name__ == "__main__": unittest.main()