From d9d3bbc30cf34954f1815cacd3b80a620e2dff34 Mon Sep 17 00:00:00 2001 From: wauxing Date: Tue, 7 Jul 2026 10:43:08 +0800 Subject: [PATCH] chore: move community-ops-sweep workflow to project root workflows/ Move examples/workflows/community-ops-sweep to workflows/community-ops-sweep to promote it from an example to a first-class project workflow. Also includes pending org shortcuts modifications. --- shortcuts/org/org.go | 109 +++ shortcuts/org/org_test.go | 162 ++++ workflows/community-ops-sweep/README.md | 83 ++ .../community-ops-sweep.wf.js | 164 ++++ .../routing-rules.example.yaml | 37 + .../scripts/community_ops_sweep.py | 910 ++++++++++++++++++ .../community-ops-sweep/tests/test_sweep.py | 149 +++ 7 files changed, 1614 insertions(+) create mode 100644 workflows/community-ops-sweep/README.md create mode 100644 workflows/community-ops-sweep/community-ops-sweep.wf.js create mode 100644 workflows/community-ops-sweep/routing-rules.example.yaml create mode 100644 workflows/community-ops-sweep/scripts/community_ops_sweep.py create mode 100644 workflows/community-ops-sweep/tests/test_sweep.py diff --git a/shortcuts/org/org.go b/shortcuts/org/org.go index 57d8b4e..4e45561 100644 --- a/shortcuts/org/org.go +++ b/shortcuts/org/org.go @@ -88,6 +88,115 @@ func Shortcuts(translators ...*i18n.Translator) []*common.Shortcut { return ctx.Output(env) }, }, + { + Name: "teams", + Description: "List teams in an organization", + Flags: []common.Flag{ + {Name: "id", Short: "i", Usage: tr.T("flag.org.id"), Required: true}, + {Name: "page", Short: "p", Usage: tr.T("flag.page"), Default: "1"}, + {Name: "limit", Short: "l", Usage: tr.T("flag.limit"), Default: "20"}, + }, + Run: func(ctx *common.RuntimeContext) error { + id, _ := ctx.RequireArg("id") + q := url.Values{} + q.Set("page", ctx.Arg("page")) + q.Set("limit", ctx.Arg("limit")) + env, err := ctx.CallAPIWithQuery("GET", fmt.Sprintf("/organizations/%s/teams", id), q) + if err != nil { + return err + } + return ctx.Output(env) + }, + }, + { + Name: "create-team", + Description: "Create a team in an organization", + Flags: []common.Flag{ + {Name: "id", Short: "i", Usage: tr.T("flag.org.id"), Required: true}, + {Name: "name", Short: "n", Usage: tr.T("flag.org.name"), Required: true}, + {Name: "description", Short: "d", Usage: tr.T("flag.description")}, + }, + Run: func(ctx *common.RuntimeContext) error { + id, _ := ctx.RequireArg("id") + name, _ := ctx.RequireArg("name") + payload := map[string]interface{}{ + "name": name, + "nickname": name, + } + if d := ctx.Arg("description"); d != "" { + payload["description"] = d + } + env, err := ctx.CallAPI("POST", fmt.Sprintf("/organizations/%s/teams", id), payload) + if err != nil { + return err + } + return ctx.Output(env) + }, + }, + { + Name: "remove-member", + Description: "Remove a member from an organization", + Flags: []common.Flag{ + {Name: "id", Short: "i", Usage: tr.T("flag.org.id"), Required: true}, + {Name: "uid", Short: "u", Usage: "User ID", Required: true}, + }, + Run: func(ctx *common.RuntimeContext) error { + id, _ := ctx.RequireArg("id") + uid, _ := ctx.RequireArg("uid") + env, err := ctx.CallAPI("DELETE", fmt.Sprintf("/organizations/%s/organization_users/%s", id, uid), nil) + if err != nil { + return err + } + return ctx.Output(env) + }, + }, + { + Name: "nickname", + Description: "Set or view a member's nickname in an organization", + Flags: []common.Flag{ + {Name: "id", Short: "i", Usage: tr.T("flag.org.id"), Required: true}, + {Name: "uid", Short: "u", Usage: "User ID", Required: true}, + {Name: "nickname", Short: "n", Usage: "New nickname (omit to view current)"}, + }, + Run: func(ctx *common.RuntimeContext) error { + id, _ := ctx.RequireArg("id") + uid, _ := ctx.RequireArg("uid") + nickname := ctx.Arg("nickname") + if nickname != "" { + payload := map[string]interface{}{ + "nickname": nickname, + } + env, err := ctx.CallAPI("PUT", fmt.Sprintf("/organizations/%s/organization_users/%s", id, uid), payload) + if err != nil { + return err + } + return ctx.Output(env) + } + env, err := ctx.CallAPI("GET", fmt.Sprintf("/organizations/%s/organization_users/%s", id, uid), nil) + if err != nil { + return err + } + return ctx.Output(env) + }, + }, + { + Name: "uid", + Description: "Look up a user's numeric ID by login name", + Flags: []common.Flag{ + {Name: "login", Short: "l", Usage: "User login name", Required: true}, + }, + Run: func(ctx *common.RuntimeContext) error { + login, err := ctx.RequireArg("login") + if err != nil { + return err + } + env, err := ctx.CallAPI("GET", fmt.Sprintf("/users/%s", login), nil) + if err != nil { + return err + } + return ctx.Output(env) + }, + }, } } diff --git a/shortcuts/org/org_test.go b/shortcuts/org/org_test.go index 98594bd..d501099 100644 --- a/shortcuts/org/org_test.go +++ b/shortcuts/org/org_test.go @@ -112,3 +112,165 @@ func TestOrgCreate(t *testing.T) { t.Errorf("expected POST, got %s", requestMethod) } } + +// --- teams --- + +func TestOrgTeams(t *testing.T) { + server := common.NewTestServer(t, func(w http.ResponseWriter, r *http.Request) { + if r.Method == "GET" && r.URL.Path == "/organizations/5/teams.json" { + common.WriteJSON(t, w, map[string]interface{}{ + "total_count": float64(1), + "teams": []interface{}{ + map[string]interface{}{ + "id": float64(1), + "name": "dev-team", + }, + }, + }) + } else { + t.Fatalf("unexpected request: %s %s", r.Method, r.URL.Path) + } + }) + defer server.Close() + + ctx := common.NewTestContext(t, server, "", "", map[string]string{ + "id": "5", + }) + err := common.RunShortcut(t, Shortcuts(), "teams", ctx) + if err != nil { + t.Fatalf("teams failed: %v", err) + } +} + +// --- create-team --- + +func TestOrgCreateTeam(t *testing.T) { + var requestMethod string + server := common.NewTestServer(t, func(w http.ResponseWriter, r *http.Request) { + requestMethod = r.Method + payload := common.DecodeJSON(t, r) + if payload["name"] != "new-team" { + t.Fatalf("expected name=new-team, got %v", payload["name"]) + } + common.WriteJSON(t, w, map[string]interface{}{ + "id": float64(1), + "name": "new-team", + }) + }) + defer server.Close() + + ctx := common.NewTestContext(t, server, "", "", map[string]string{ + "id": "5", + "name": "new-team", + }) + err := common.RunShortcut(t, Shortcuts(), "create-team", ctx) + if err != nil { + t.Fatalf("create-team failed: %v", err) + } + if requestMethod != "POST" { + t.Errorf("expected POST, got %s", requestMethod) + } +} + +// --- remove-member --- + +func TestOrgRemoveMember(t *testing.T) { + var requestMethod string + server := common.NewTestServer(t, func(w http.ResponseWriter, r *http.Request) { + requestMethod = r.Method + common.WriteJSON(t, w, map[string]interface{}{ + "ok": true, + }) + }) + defer server.Close() + + ctx := common.NewTestContext(t, server, "", "", map[string]string{ + "id": "5", + "uid": "42", + }) + err := common.RunShortcut(t, Shortcuts(), "remove-member", ctx) + if err != nil { + t.Fatalf("remove-member failed: %v", err) + } + if requestMethod != "DELETE" { + t.Errorf("expected DELETE, got %s", requestMethod) + } +} + +// --- nickname (view) --- + +func TestOrgNicknameView(t *testing.T) { + server := common.NewTestServer(t, func(w http.ResponseWriter, r *http.Request) { + if r.Method == "GET" && r.URL.Path == "/organizations/5/organization_users/42.json" { + common.WriteJSON(t, w, map[string]interface{}{ + "nickname": "thename", + }) + } else { + t.Fatalf("unexpected request: %s %s", r.Method, r.URL.Path) + } + }) + defer server.Close() + + ctx := common.NewTestContext(t, server, "", "", map[string]string{ + "id": "5", + "uid": "42", + }) + err := common.RunShortcut(t, Shortcuts(), "nickname", ctx) + if err != nil { + t.Fatalf("nickname failed: %v", err) + } +} + +// --- nickname (set) --- + +func TestOrgNicknameSet(t *testing.T) { + var requestMethod string + server := common.NewTestServer(t, func(w http.ResponseWriter, r *http.Request) { + requestMethod = r.Method + payload := common.DecodeJSON(t, r) + if payload["nickname"] != "newname" { + t.Fatalf("expected nickname=newname, got %v", payload["nickname"]) + } + common.WriteJSON(t, w, map[string]interface{}{ + "nickname": "newname", + }) + }) + defer server.Close() + + ctx := common.NewTestContext(t, server, "", "", map[string]string{ + "id": "5", + "uid": "42", + "nickname": "newname", + }) + err := common.RunShortcut(t, Shortcuts(), "nickname", ctx) + if err != nil { + t.Fatalf("nickname set failed: %v", err) + } + if requestMethod != "PUT" { + t.Errorf("expected PUT, got %s", requestMethod) + } +} + +// --- uid --- + +func TestOrgUID(t *testing.T) { + server := common.NewTestServer(t, func(w http.ResponseWriter, r *http.Request) { + if r.Method == "GET" && r.URL.Path == "/users/baoerjun.json" { + common.WriteJSON(t, w, map[string]interface{}{ + "id": float64(148287), + "login": "baoerjun", + }) + } else { + t.Fatalf("unexpected request: %s %s", r.Method, r.URL.Path) + } + }) + defer server.Close() + + ctx := common.NewTestContext(t, server, "", "", map[string]string{ + "login": "baoerjun", + }) + err := common.RunShortcut(t, Shortcuts(), "uid", ctx) + if err != nil { + t.Fatalf("uid failed: %v", err) + } +} diff --git a/workflows/community-ops-sweep/README.md b/workflows/community-ops-sweep/README.md new file mode 100644 index 0000000..1e16400 --- /dev/null +++ b/workflows/community-ops-sweep/README.md @@ -0,0 +1,83 @@ +# community-ops-sweep —— 社区运营增量对账 sweep + +## 概述 + +一个 [Claude Code Workflow](../../../README.md) 实现的**增量对账式**社区运营自动化： +运行时给定一个 `since` 时间点（或默认读上次 checkpoint），处理该时间点后的所有**新增 Issue** 与**合并 PR**，按需更新社区周报与 Release Notes。 + +**一次调用、时间窗界定、5 项事件一网打尽**：R1 新 Issue triage → R2 建议 owner → R3 周报 → R4 Release Notes → R5 PR→Issue 关联闭环。 + +> 完整设计见 [`docs/superpowers/specs/2026-07-01-community-ops-sweep-design.md`](/docs/superpowers/specs/2026-07-01-community-ops-sweep-design.md) + +## 架构 + +三层，JSON 文件交接（判断与计算分离）： + +``` +CC Workflow (JS) ……… 编排：按 phase 调度 + ├── LLM agent 层 …… 只做判断：分类 / PR 关联 / owner 建议 / 散文增强 + │ ↕ JSON 文件 + └── Python 引擎 …… 只做计算：采集 / 归一化 / id 解析 / 合并 / 守卫 / 渲染 / 写回 +``` + +确定性引擎（`scripts/community_ops_sweep.py`）复用 `community-ops-automation` 的采集/归一化/渲染管线，真实字段名实测自 `baoerjun/gitlink-cli`。 + +## 前置条件 + +- **gitlink-cli** ≥ 0.2.0（`npm install -g @gitlink-ai/cli`） +- **Python** 3.9+（标准库，无第三方依赖） +- **gitlink-cli auth login**（认证） +- **Claude Code**（运行 workflow） + +## 运行方式 + +### 在 Claude Code 中 + +```bash +# 作为 slash 命令（若已放入 .claude/workflows/） +/community-ops-sweep owner=baoerjun repo=gitlink-cli + +# 或通过 Workflow 工具 +# 带入参数：since 可选（ISO 时间或留空读 .last-sweep）、apply 默认 false +``` + +### 单独跑确定性引擎 + +```bash +# collect：采集 since 后的新 issue + 合并 PR +python scripts/community_ops_sweep.py collect --owner baoerjun --repo gitlink-cli --since 2026-06-01T00:00:00Z + +# plan：读 candidates + LLM decisions → 写计划 +python scripts/community_ops_sweep.py plan --candidates candidates.json --triage triage.json --owners owners.json --links links.json + +# apply：dry-run 预览，--apply 才真写 +python scripts/community_ops_sweep.py apply --plan plan.json --owner baoerjun --repo gitlink-cli --apply + +# checkpoint：推进 .last-sweep + 追加 triage-log +python scripts/community_ops_sweep.py checkpoint --owner baoerjun --repo gitlink-cli +``` + +### 跑单测 + +```bash +python3 tests/test_sweep.py # 18 个确定性逻辑回归 +``` + +## 文件布局 + +``` +community-ops-sweep/ + community-ops-sweep.wf.js # CC workflow (JS) + scripts/community_ops_sweep.py # 确定性引擎 + routing-rules.example.yaml # R2 路由规则（opt-in auto_assign） + tests/test_sweep.py # 引擎单测 + README.md +``` + +## 安全 + +- 默认 **dry-run**，`--apply` 才写 +- 标签**整体替换**发期望全集（增删统一，不丢标签） +- 关 issue 前自动守卫（已关跳过） +- **绝不自动合并 PR** +- 可逆写（关可 reopen、release 可 edit）→ 无二次确认 diff --git a/workflows/community-ops-sweep/community-ops-sweep.wf.js b/workflows/community-ops-sweep/community-ops-sweep.wf.js new file mode 100644 index 0000000..aa92893 --- /dev/null +++ b/workflows/community-ops-sweep/community-ops-sweep.wf.js @@ -0,0 +1,164 @@ +export const meta = { + name: 'community-ops-sweep', + description: 'Incremental community ops sweep — triage new issues (R1), suggest owners (R2), link merged PRs→issues (R5), generate weekly report + release notes (R3/R4), upload report to Wiki. Default dry-run; --apply to write.', + phases: [ + { title: 'Collect', detail: 'Python engine gathers new issues + merged/open PRs + contributor stats since checkpoint' }, + { title: 'Triage', detail: 'LLM agents classify, suggest owners, link PRs — in parallel' }, + { title: 'Summarize', detail: 'LLM generates executive summary from collected data' }, + { title: 'Plan', detail: 'Python engine builds deterministic write plan + generates seven-section report' }, + { title: 'Apply', detail: 'Execute (or dry-run) write plan + publish to Wiki + advance checkpoint' }, + ], +} + +const cfg = args || {} +const OWNER = cfg.owner +const REPO = cfg.repo +const APPLY = cfg.apply === true +const REPORT_ISSUE = cfg.reportIssue || '' +const RELEASE_ID = cfg.releaseId || '' +const SINCE = cfg.since || '' +const PUBLISH_WIKI = cfg.publishWiki === true +const WIKI_DIR = cfg.wikiDir || '周报专区' +const REPORT_FLAG = REPORT_ISSUE ? `--report-issue ${REPORT_ISSUE}` : '' +const RELEASE_FLAG = RELEASE_ID ? `--release-id ${RELEASE_ID}` : '' +const WIKI_FLAG = PUBLISH_WIKI ? `--publish-wiki --wiki-dir ${WIKI_DIR}` : '' + +const ENGINE = 'examples/workflows/community-ops-sweep/scripts/community_ops_sweep.py' +const CANDIDATES = '/tmp/sweep-candidates.json' +const TRIAGE = '/tmp/sweep-triage.json' +const OWNERS = '/tmp/sweep-owners.json' +const LINKS = '/tmp/sweep-links.json' +const PLAN = '/tmp/sweep-plan.json' +const SUMMARY = '/tmp/sweep-summary.json' +const SUMMARY_ARG = `--summary ${SUMMARY}` + +const SINCE_ARG = SINCE ? `--since ${SINCE}` : '' + +if (!OWNER || !REPO) { + throw new Error('Missing required args: owner, repo') +} + +phase('Collect') +log(`Collecting since=${SINCE || 'checkpoint'} for ${OWNER}/${REPO}`) +const collectOut = await agent( + `Run the sweep engine collect:\n` + + ` python3 ${ENGINE} collect --owner ${OWNER} --repo ${REPO} ${SINCE_ARG} --out ${CANDIDATES}\n` + + `Then Read ${CANDIDATES} and return a compact summary: how many new_issues, merged_prs, open_issues, and the tag_map keys. Also list each issue's number, title and current labels.`, + { label: 'collect', phase: 'Collect' } +) +log(`Collect result: ${collectOut}`) + +// ----------------------------------------------------------------------- +// Phase Triage — three LLM agents, independent → parallel +// ----------------------------------------------------------------------- +phase('Triage') +log('R1 classify + R2 suggest owners + R5 link PRs — in parallel') +const triagePrompts = [ + // R1: classify new issues + () => agent( + `## Task: Classify new issues for triage (R1)\n\n` + + `1. Read the file ${CANDIDATES}. It contains:\n` + + ` - "issues": new issues with fields number, title, description, tags, labels\n` + + ` - "tag_map": available label names → IDs. ONLY use labels from tag_map keys.\n\n` + + `2. For EACH issue, determine:\n` + + ` - tag_names: labels to ADD (pick from tag_map keys, e.g. "bug","feature","enhancement","question","docs","P1","P2","P3")\n` + + ` - remove_tag_names: labels to REMOVE if any current ones are wrong\n` + + ` - recommended_action: one of "schedule_fix","duplicate","answered","spam","invalid". Use "schedule_fix" for normal issues.\n` + + ` (duplicate/answered/spam/invalid → will be auto-closed)\n\n` + + `3. Write your decision as a single-line JSON array to ${TRIAGE} using Bash:\n` + + ` cat > ${TRIAGE} << 'EOD'\n [{"issue":"","tag_names":[""],"remove_tag_names":[""],"recommended_action":""}]\n EOD\n` + + ` Make sure each issue has an entry if there are decisions. Skip issues that need no changes.\n\n` + + `4. Confirm: cat ${TRIAGE}`, + { label: 'triage:R1-classify', phase: 'Triage' } + ), + // R2: suggest owners + () => agent( + `## Task: Suggest owners for new issues (R2)\n\n` + + `1. Read the file ${CANDIDATES}.\n\n` + + `2. For EACH issue, suggest 1-3 likely owners based on:\n` + + ` - The issue content (title, description)\n` + + ` - Author logins of recent merged PRs (in "merged_prs")\n` + + ` - General domain familiarity\n\n` + + `3. Output: write a JSON array to ${OWNERS} using Bash:\n` + + ` cat > ${OWNERS} << 'EOD'\n [{"issue":"","suggested_owners":[""],"evidence":""}]\n EOD\n` + + ` If you cannot suggest anyone for an issue, omit it.\n\n` + + `4. Confirm: cat ${OWNERS}`, + { label: 'triage:R2-owners', phase: 'Triage' } + ), + // R5: link merged PRs → issues + () => agent( + `## Task: Link merged PRs to issues (R5)\n\n` + + `1. Read the file ${CANDIDATES}. Focus on the "merged_prs" list.\n\n` + + `2. For EACH merged PR, parse its "description" (body) for patterns like:\n` + + ` - "fixes #N", "closes #N", "resolves #N"\n` + + ` - Chinese equivalents: "修复 #N", "关闭 #N", "解决 #N"\n` + + ` - Semantic references that clearly identify a specific issue number\n\n` + + `3. Output: write a JSON array to ${LINKS} using Bash:\n` + + ` cat > ${LINKS} << 'EOD'\n [{"pr":"","linked_issue_numbers":[""]}]\n EOD\n` + + ` If no PR references any issue, write an empty array [].\n\n` + + `4. Confirm: cat ${LINKS}`, + { label: 'triage:R5-link', phase: 'Triage' } + ), +] + +await Promise.all(triagePrompts.map(fn => fn())) +log('Triage complete — decision files written') + +// ----------------------------------------------------------------------- +// Phase Summarize — LLM generates executive summary +// ----------------------------------------------------------------------- +phase('Summarize') +log('Generating executive summary from collected data') +const summaryOut = await agent( + `## Task: Generate weekly report executive summary\n\n` + + `1. Read the file ${CANDIDATES}. It contains:\n` + + ` - "issues": new issues this period (number, title, labels, author_login, state)\n` + + ` - "merged_prs": merged PRs (number, title, author_login)\n` + + ` - "open_prs": pending PRs\n` + + ` - "current_counts": issue/PR counts\n` + + ` - "contributors": top contributors with issue/PR counts\n` + + ` - "trends": trend arrows (↑↓→) compared to last period\n\n` + + `2. Write a 3-5 sentence executive summary in Chinese to ${SUMMARY} using Bash:\n` + + ` cat > ${SUMMARY} << 'EOD'\n {"summary": "本周社区新增 N 个 Issue，合并 M 个 PR。……"}\n EOD\n\n` + + `3. The summary should cover:\n` + + ` - Overall activity level (busy/calm/normal)\n` + + ` - Key highlights (major features merged, important bugs fixed)\n` + + ` - Notable trends (↑ or ↓ compared to last week)\n` + + ` - Any concerns or action items\n\n` + + `4. Confirm: cat ${SUMMARY}`, + { label: 'summarize', phase: 'Summarize' } +) +log(`Summary: ${summaryOut}`) + +// ----------------------------------------------------------------------- +// Phase Plan — deterministic engine assembles write plan + generates report +// ----------------------------------------------------------------------- +phase('Plan') +const planOut = await agent( + `Run the sweep engine plan:\n` + + ` python3 ${ENGINE} plan --candidates ${CANDIDATES} --triage ${TRIAGE} --owners ${OWNERS} --links ${LINKS} ${SUMMARY_ARG} --out ${PLAN}\n` + + `Then Read ${PLAN} and summarize: how many writes, by op (update_tags/update_status/comment/update_assigner), which issues affected. Show the writes table.`, + { label: 'plan', phase: 'Plan' } +) +log(`Plan: ${planOut}`) + +// ----------------------------------------------------------------------- +// Phase Apply — execute (or dry-run) + checkpoint +// ----------------------------------------------------------------------- +phase('Apply') +const mode = APPLY ? '--apply' : '' +const applyOut = await agent( + APPLY + ? `Execute the write plan (--apply):\n python3 ${ENGINE} apply --plan ${PLAN} --owner ${OWNER} --repo ${REPO} --apply ${REPORT_FLAG} ${RELEASE_FLAG} ${WIKI_FLAG}\nThen: python3 ${ENGINE} checkpoint --owner ${OWNER} --repo ${REPO} --plan ${PLAN}\nReturn a summary of writes executed and where the report was published.` + : `Dry-run the write plan (no writes executed):\n python3 ${ENGINE} apply --plan ${PLAN} --owner ${OWNER} --repo ${REPO} ${REPORT_FLAG} ${RELEASE_FLAG} ${WIKI_FLAG}\nSummarize what WOULD be written.`, + { label: 'apply', phase: 'Apply' } +) +log(`Apply: ${applyOut}`) + +return { + mode: APPLY ? 'applied' : 'dry-run', + owner: OWNER, repo: REPO, + collect: collectOut, + plan: planOut, + apply: applyOut, +} diff --git a/workflows/community-ops-sweep/routing-rules.example.yaml b/workflows/community-ops-sweep/routing-rules.example.yaml new file mode 100644 index 0000000..50c8ea8 --- /dev/null +++ b/workflows/community-ops-sweep/routing-rules.example.yaml @@ -0,0 +1,37 @@ +# 路由规则（R2 owner 建议 + 可选 auto_assign） + +# rules：首次命中生效，顺序即优先级 +# 默认用途：R2 建议 owner 的依据（LLM 参考 + 确定性兜底） +# auto_assign: true → 该规则命中的 issue 自动指派（内部团队仓，opt-in） + +rules: + - match: + labels: [bug] + any_keyword: [auth, 认证, 登录, token, oauth, session] + assigner: auth-owner + auto_assign: false + + - match: + labels: [bug] + any_keyword: [ci, build, 构建, pipeline, deploy] + assigner: ci-owner + auto_assign: false + + - match: + labels: [docs] + assigner: doc-maintainer + auto_assign: false + + - match: + labels: [enhancement, feature] + assigner: product-owner + auto_assign: false + + - match: + labels: [security] + assigner: security-team + auto_assign: false + +# 所有未命中规则 → 兜底 +default_assigner: maintainer +auto_assign: false diff --git a/workflows/community-ops-sweep/scripts/community_ops_sweep.py b/workflows/community-ops-sweep/scripts/community_ops_sweep.py new file mode 100644 index 0000000..357dcd2 --- /dev/null +++ b/workflows/community-ops-sweep/scripts/community_ops_sweep.py @@ -0,0 +1,910 @@ +#!/usr/bin/env python3 +"""community_ops_sweep — 社区运营增量对账 sweep 的确定性引擎。 + +三层架构里的「Python 引擎层」：只做事实性计算（采集 / 归一化 / id 解析 / 合并 / +守卫 / 渲染 / 写回）；判断（分类 / PR 关联 / owner 建议）由 CC workflow 的 LLM +agent 产出 JSON 喂入。设计见 +docs/superpowers/specs/2026-07-01-community-ops-sweep-design.md。 + +子命令： + collect 采集 since 之后的新 issue + 合并 PR + 标签/分配人映射 → candidates.json + plan 读 candidates + LLM decision JSON → plan.json（写计划） + apply 执行 plan.json（默认 dry-run，--apply 才真写） + checkpoint 推进 .last-sweep + 追加 docs/issue-triage-log.md + +字段名实测自 baoerjun/gitlink-cli：issue 在 data.issues[]，PR 在 data.pulls[]。 +""" +from __future__ import annotations + +import argparse +import importlib.util +import json +import subprocess +import sys +from datetime import datetime, timedelta, timezone +from pathlib import Path +from typing import Any, Iterable + +# --------------------------------------------------------------------------- # +# 复用 community-ops-automation 的 CLI 封装与通用 helpers（best-effort：缺失 +# 也不影响纯逻辑导入与单测） +# --------------------------------------------------------------------------- # +_GW_PATH = ( + Path(__file__).resolve().parents[2] + / "community-ops-automation" + / "scripts" + / "gitlink_workflow.py" +) +try: + _spec = importlib.util.spec_from_file_location("_gw", _GW_PATH) + _gw = importlib.util.module_from_spec(_spec) # type: ignore[union-attr] + sys.modules["_gw"] = _gw + _spec.loader.exec_module(_gw) # type: ignore[union-attr] + run_gitlink_cli = _gw.run_gitlink_cli + first_value = _gw.first_value + extract_first_list = _gw.extract_first_list + parse_datetime = _gw.parse_datetime + WorkflowError = _gw.WorkflowError +except Exception: # pragma: no cover - 纯逻辑单测不应依赖 _gw + _gw = None + run_gitlink_cli = None # type: ignore[assignment] + first_value = lambda item, keys, default=None: next( # noqa: E731 + (item[k] for k in keys if isinstance(item, dict) and k in item and item[k] not in (None, "", [])), default + ) + extract_first_list = lambda payload, keys: payload if isinstance(payload, list) else [] # noqa: E731 + parse_datetime = lambda v: None # type: ignore[assignment] # noqa: E731 + + class WorkflowError(RuntimeError): + pass + + +DEFAULT_POLICY = { + "status_map": { + "duplicate": "closed", + "answered": "closed", + "spam": "closed", + "invalid": "closed", + } +} + + +# --------------------------------------------------------------------------- # +# 纯逻辑（确定性、可单测）—— TDD 覆盖 +# --------------------------------------------------------------------------- # +def _id_sort_key(x: Any) -> tuple: + try: + return (0, int(x)) + except (TypeError, ValueError): + return (1, str(x)) + + +def merge_tag_ids(current_ids, add_names, remove_names, tag_map): + """整体替换语义：返回 (current ∪ add) \\ remove；白名单过滤、去重、按数值排序。""" + result = {str(x) for x in (current_ids or []) if x is not None and str(x) != ""} + for name in add_names or []: + tid = (tag_map or {}).get(name) + if tid is not None: + result.add(str(tid)) + for name in remove_names or []: + tid = (tag_map or {}).get(name) + if tid is not None: + result.discard(str(tid)) + return sorted(result, key=_id_sort_key) + + +def map_status(recommended_action, policy): + """recommended_action → 目标状态（如 duplicate→closed）；未映射或 None → None（不改）。""" + if not recommended_action: + return None + sm = (policy or {}).get("status_map", {}) or {} + return sm.get(recommended_action) + + +def _issue_current_tag_ids(issue: dict) -> list[str]: + out = [] + for t in issue.get("tags") or []: + if isinstance(t, dict) and t.get("id") is not None: + out.append(str(t["id"])) + return out + + +def coerce_dates(item, keys=("updated_at", "created_at")): + """把 item 中字符串日期字段解析回 datetime（磁盘 round-trip 后的修复）。 + + datetime/None 原样保留；只解析 str。缺失的键补齐为 None。非 dict 原样返回。""" + if not isinstance(item, dict): + return item + out = dict(item) + for k in keys: + v = out.get(k) + if isinstance(v, str): + out[k] = parse_datetime(v) + elif k not in out: + out[k] = None + return out + + +def _render_owner_comment(evidence, suggested) -> str: + names = "、".join(f"@{s}" for s in suggested) + ev = f"（依据：{evidence}）" if evidence else "" + return ( + f"🤖 建议负责人：{names}{ev}。\n" + "如非本人负责，请自行重新指派或在下方留言说明。" + ) + + +def build_plan(candidates, triage, owners, links, routing, policy, tag_map): + """把 LLM 判断 + 确定性规则汇编成写计划。纯函数：同输入 → 同输出。""" + policy = policy or {} + routing = routing or {} + tag_map = tag_map or {} + writes: list[dict] = [] + + issues = candidates.get("issues", []) or [] + by_num = {str(i.get("number")): i for i in issues if i.get("number") is not None} + open_issues = {str(x) for x in (candidates.get("open_issues") or [])} + + # R1 triage：标签增删 + 状态映射 + for t in triage or []: + num = str(t.get("issue")) + issue = by_num.get(num) + if issue is None: + continue # 校验：候选里没有 → 跳过，不盲写 + current = _issue_current_tag_ids(issue) + new_ids = merge_tag_ids( + current, t.get("tag_names", []), t.get("remove_tag_names", []), tag_map + ) + if sorted(new_ids, key=_id_sort_key) != sorted(current, key=_id_sort_key): + writes.append({ + "op": "update_tags", "issue": num, "tag_ids": new_ids, + "tag_names": t.get("tag_names", []) or [], + "remove_tag_names": t.get("remove_tag_names", []) or [], + }) + state = map_status(t.get("recommended_action"), policy) + if state: + writes.append({ + "op": "update_status", "issue": num, "state": state, + "reason": f"triage: {t.get('recommended_action')}", + }) + + # R2 owner 建议（默认评论；routing.auto_assign=true 才真指派） + auto_assign = bool(routing.get("auto_assign")) + assigner_map = candidates.get("assigner_map") or {} + for o in owners or []: + num = str(o.get("issue")) + if num not in by_num: + continue + suggested = o.get("suggested_owners") or [] + if not suggested: + continue + writes.append({ + "op": "comment", "issue": num, "kind": "owner_suggest", + "body": _render_owner_comment(o.get("evidence"), suggested), + }) + if auto_assign: + aids = [str(assigner_map[s]) for s in suggested if s in assigner_map] + if aids: + writes.append({"op": "update_assigner", "issue": num, "assigner_ids": aids}) + + # R5 PR→Issue 关联闭环：只关「存在且 open」的（幂等 + 防幻觉编号） + for l in links or []: + pr = l.get("pr") + for inum in l.get("linked_issue_numbers", []) or []: + inum = str(inum) + if inum not in open_issues: + continue # 校验：不存在或非 open → 跳过 + writes.append({ + "op": "update_status", "issue": inum, "state": "closed", + "reason": f"fixed by PR#{pr}", + }) + writes.append({ + "op": "comment", "issue": inum, "kind": "link_close", + "body": f"🤖 PR#{pr} 已合并且语义关联本 issue，自动关闭（可 reopen）。", + }) + + return {"writes": writes} + + +# --------------------------------------------------------------------------- # +# 归一化（实测字段名） +# --------------------------------------------------------------------------- # +def _issue_state(item: dict) -> str: + sid = first_value(item, ("status_id", "state_id"), None) + if sid in (5, "5", "closed", "close"): + return "closed" + return "open" + + +def normalize_issue(item: dict) -> dict: + author = item.get("author") or {} + tags = item.get("tags") or [] + return { + "id": str(first_value(item, ("number", "project_issues_index", "iid"), "")), + "number": str(first_value(item, ("number", "project_issues_index", "iid"), "")), + "title": str(first_value(item, ("subject", "title", "name"), "(untitled)")), + "description": str(first_value(item, ("description", "body", "content"), "")), + "state": _issue_state(item), + "labels": [str(t.get("name")) for t in tags if isinstance(t, dict) and t.get("name")], + "tags": tags, + "author_login": author.get("login") or author.get("name") or "", + "author_id": author.get("id"), + "created_at": parse_datetime(first_value(item, ("created_at", "createdAt"), None)), + "updated_at": parse_datetime(first_value(item, ("updated_at", "updatedAt"), None)), + } + + +def normalize_pr(pull: dict) -> dict: + inner = pull.get("issue") or {} + author = inner.get("author") or pull.get("author") or {} + state = str(first_value(pull, ("status", "state"), "open")).lower() + merged = state == "merged" + return { + "id": str(first_value(pull, ("index", "number", "pull_request_number"), "")), + "number": str(first_value(pull, ("index", "number", "pull_request_number"), "")), + "title": str(first_value(pull, ("title", "name"), "(untitled)")), + "description": str(first_value(pull, ("body", "description", "content"), "")), + "state": state, + "merged": merged, + "base": pull.get("base"), + "head": pull.get("head"), + "author_login": author.get("login") or author.get("name") or "", + "author_id": author.get("id"), + "merged_at": parse_datetime(first_value(pull, ("merged_at", "merge_time"), None)), + "created_at": parse_datetime(first_value(pull, ("created_at", "pr_created_unix"), None)), + "updated_at": parse_datetime(first_value(pull, ("updated_at", "created_at", "pr_created_unix"), None)), + "labels": [str(t.get("name")) for t in (inner.get("issue_tags") or []) if isinstance(t, dict) and t.get("name")], + } + + +def _load_tag_map(owner, repo) -> dict[str, str]: + if run_gitlink_cli is None: + return {} + payload = run_gitlink_cli(["issue", "+tags", "--only-name"], owner, repo) + items = extract_first_list(payload, ("issue_tags", "tags", "items", "list")) + return {str(t.get("name")): str(t.get("id")) for t in items if isinstance(t, dict) and t.get("name") and t.get("id") is not None} + + +def _load_assigner_map(owner, repo) -> dict[str, str]: + if run_gitlink_cli is None: + return {} + payload = run_gitlink_cli(["issue", "+assigners"], owner, repo) + items = extract_first_list(payload, ("assigners", "issue_assigners", "users", "items", "list")) + out = {} + for u in items: + if not isinstance(u, dict): + continue + login = u.get("login") or u.get("name") or u.get("username") + if login and u.get("id") is not None: + out[str(login)] = str(u["id"]) + return out + + +# --------------------------------------------------------------------------- # +# I/O 子命令 +# --------------------------------------------------------------------------- # +def _read_json(path: Path) -> Any: + return json.loads(path.read_text(encoding="utf-8")) + + +def _write_json(path: Path, data: Any) -> None: + path.parent.mkdir(parents=True, exist_ok=True) + path.write_text(json.dumps(data, ensure_ascii=False, indent=2, default=str), encoding="utf-8") + + +def _parse_since(value: str | None, checkpoint: Path | None) -> datetime: + if value: + dt = parse_datetime(value) + if dt is None: + raise WorkflowError(f"无法解析 --since: {value}") + return dt + if checkpoint and checkpoint.exists(): + dt = parse_datetime(checkpoint.read_text(encoding="utf-8").strip()) + if dt is not None: + return dt + return datetime.now(timezone.utc) - timedelta(days=7) + + +def _collect_all_prs(owner: str, repo: str): + """采集所有 PR（open + merged + closed），返回 (open_prs, merged_prs, all_pr_authors)。""" + all_prs: list[dict] = [] + for state in ("open", "merged", "closed"): + raw = run_gitlink_cli(["pr", "+list", "--state", state, "--limit", "200"], owner, repo) + items = [ + normalize_pr(p) + for p in extract_first_list(raw, ("pulls", "pull_requests", "issues", "items", "list")) + if isinstance(p, dict) + ] + all_prs.extend(items) + # 去重 (同一条 PR 可能出现在多个 state 查询中) + seen: set[str] = set() + deduped: list[dict] = [] + for p in all_prs: + if p["number"] not in seen: + seen.add(p["number"]) + deduped.append(p) + open_prs = [p for p in deduped if p["state"] == "open"] + merged_prs = [p for p in deduped if p["merged"] or p["state"] == "merged"] + authors = sorted({p["author_login"] for p in deduped if p["author_login"]}) + return open_prs, merged_prs, authors + + +def _contributor_ranking(issues: list[dict], merged_prs: list[dict]) -> list[dict]: + """统计贡献者活跃度：合并 PR 数 + 提交 Issue 数。""" + scores: dict[str, dict] = {} + for i in issues: + login = i.get("author_login", "") + if not login: + continue + if login not in scores: + scores[login] = {"login": login, "issues": 0, "prs": 0} + scores[login]["issues"] += 1 + for p in merged_prs: + login = p.get("author_login", "") + if not login: + continue + if login not in scores: + scores[login] = {"login": login, "issues": 0, "prs": 0} + scores[login]["prs"] += 1 + ranked = sorted(scores.values(), key=lambda x: x["prs"] + x["issues"], reverse=True) + return ranked[:10] + + +def _load_previous_summary(checkpoint_dir: Path) -> dict | None: + """读取上一周期的汇总数据用于趋势对比。""" + path = checkpoint_dir / ".last-sweep-summary.json" + if not path.exists(): + return None + try: + return json.loads(path.read_text(encoding="utf-8")) + except Exception: + return None + + +def _save_current_summary(checkpoint_dir: Path, summary: dict) -> None: + """保存本周期的汇总数据供下一次趋势对比。""" + checkpoint_dir.mkdir(parents=True, exist_ok=True) + (checkpoint_dir / ".last-sweep-summary.json").write_text( + json.dumps(summary, ensure_ascii=False, indent=2, default=str), + encoding="utf-8", + ) + + +def cmd_collect(args) -> int: + owner, repo = args.owner, args.repo + checkpoint = Path(args.checkpoint) + since = _parse_since(args.since, checkpoint) + + raw_issues = run_gitlink_cli(["issue", "+list", "--limit", "200"], owner, repo) + all_issues = [ + normalize_issue(i) + for i in extract_first_list(raw_issues, ("issues", "issue_list", "items", "list")) + if isinstance(i, dict) + ] + new_issues = [i for i in all_issues if i["created_at"] and i["created_at"] >= since] + open_issue_numbers = sorted({i["number"] for i in all_issues if i["state"] == "open"}) + closed_issues = [i for i in all_issues if i["state"] == "closed"] + + open_prs, merged_prs, pr_authors = _collect_all_prs(owner, repo) + + # 趋势对比：加载上次汇总 + previous = _load_previous_summary(checkpoint.parent) + current_counts = { + "issues_total": len(all_issues), "issues_open": len(open_issue_numbers), + "issues_closed": len(closed_issues), "prs_open": len(open_prs), + "prs_merged": len(merged_prs), "contributors": len( + {i.get("author_login", "") for i in all_issues if i.get("author_login")} | set(pr_authors) + ), + } + trends = _compute_trends(current_counts, previous) + + # 贡献者排行 + contributors = _contributor_ranking(all_issues, merged_prs) + + # 保存当前汇总供下一次对比 + _save_current_summary(checkpoint.parent, current_counts) + + candidates = { + "owner": owner, "repo": repo, "since": since.isoformat(), + "issues": new_issues, "open_issues": open_issue_numbers, + "closed_issues": [c["number"] for c in closed_issues], + "merged_prs": merged_prs, "open_prs": open_prs, + "trends": trends, "contributors": contributors, + "current_counts": current_counts, "previous_counts": previous, + "tag_map": _load_tag_map(owner, repo), + "assigner_map": _load_assigner_map(owner, repo), + } + _write_json(Path(args.out), candidates) + print(json.dumps({ + "since": since.isoformat(), "new_issues": len(new_issues), + "merged_prs": len(merged_prs), "open_issues": len(open_issue_numbers), + "open_prs": len(open_prs), "contributors": len(contributors), + }, ensure_ascii=False)) + return 0 + + +def _compute_trends(current: dict, previous: dict | None) -> dict[str, str]: + """对比上周期数据，返回趋势箭头。""" + if not previous: + return {} + trends = {} + for key in ("issues_total", "issues_open", "issues_closed", "prs_open", "prs_merged", "contributors"): + prev = previous.get(key, 0) + curr = current.get(key, 0) + if isinstance(prev, (int, float)) and isinstance(curr, (int, float)): + if curr > prev: + trends[key] = f"↑{curr - prev}" + elif curr < prev: + trends[key] = f"↓{prev - curr}" + else: + trends[key] = "→" + return trends + + +def cmd_plan(args) -> int: + candidates = _read_json(Path(args.candidates)) + tag_map = candidates.get("tag_map") or {} + policy = _read_json(Path(args.policy)) if args.policy else DEFAULT_POLICY + routing = _read_json(Path(args.routing)) if args.routing else {} + triage = _read_json(Path(args.triage)) if args.triage and Path(args.triage).exists() else [] + owners = _read_json(Path(args.owners)) if args.owners and Path(args.owners).exists() else [] + links = _read_json(Path(args.links)) if args.links and Path(args.links).exists() else [] + + # 读取 LLM 生成的执行摘要 + summary_path = getattr(args, "summary", None) + if summary_path and Path(summary_path).exists(): + summary_data = _read_json(Path(summary_path)) + candidates["executive_summary"] = summary_data.get("summary", "") if isinstance(summary_data, dict) else "" + + plan = build_plan(candidates, triage, owners, links, routing, policy, tag_map) + # R3: 周报 Markdown 生成（七段式）；R4: Release Notes + plan["report_md"] = _generate_report(candidates, plan) + plan["release_notes_md"] = _render_release_notes(candidates) + _write_json(Path(args.out), plan) + print(json.dumps({"writes": len(plan["writes"])}, ensure_ascii=False)) + return 0 + + +def _build_command(w: dict) -> list[str] | None: + op = w.get("op") + num = str(w.get("issue")) + if op == "update_tags": + return ["issue", "+update", "--number", num, "--tag-ids", ",".join(w.get("tag_ids") or [])] + if op == "update_status": + return ["issue", "+update", "--number", num, "--state", str(w.get("state"))] + if op == "comment": + return ["issue", "+comment", "--number", num, "--body", str(w.get("body"))] + if op == "update_assigner": + return ["issue", "+update", "--number", num, "--assigner-ids", ",".join(w.get("assigner_ids") or [])] + return None + + +def _exec_or_dry(args, cmd, label) -> bool: + """dry-run 打印；--apply 才真执行。返回是否执行成功。""" + if not args.apply: + preview = " ".join(cmd[:4]) + (" …" if len(cmd) > 4 else "") + print(f"[dry-run] {label}: {preview}") + return False + try: + run_gitlink_cli(cmd, args.owner, args.repo) + return True + except Exception as exc: # noqa: BLE001 + print(f"[fail] {label}: {exc}", file=sys.stderr) + return False + + +def _publish_report(args, plan) -> None: + """R3：把 report_md 作为评论发到指定 issue。""" + body = (plan.get("report_md") or "").strip() + issue = getattr(args, "report_issue", None) + if not body or not issue: + return + _exec_or_dry(args, ["issue", "+comment", "--number", str(issue), "--body", body], + f"report→issue#{issue}") + + +def _publish_release_notes(args, plan) -> None: + """R4：把 release_notes_md 合并进指定 release 的 body（release +update --body 整体替换，先读后拼）。""" + body = (plan.get("release_notes_md") or "").strip() + rid = getattr(args, "release_id", None) + if not body or not rid: + return + existing = "" + if args.apply: + try: + rv = run_gitlink_cli(["release", "+view", "--id", str(rid)], args.owner, args.repo) + data = rv.get("data", rv) if isinstance(rv, dict) else {} + existing = str(first_value(data, ("body", "description", "content", "version_content"), "") or "").rstrip() + except Exception as exc: # noqa: BLE001 + print(f"[warn] read release body failed: {exc}", file=sys.stderr) + merged = (existing + "\n\n" + body) if existing else body + _exec_or_dry(args, ["release", "+update", "--id", str(rid), "--body", merged], + f"notes→release#{rid}") + + +def cmd_apply(args) -> int: + plan = _read_json(Path(args.plan)) + writes = plan.get("writes", []) + done = 0 + for w in writes: + cmd = _build_command(w) + if cmd is None: + continue + if _exec_or_dry(args, cmd, f"#{w.get('issue')} {w.get('op')}"): + done += 1 + _publish_report(args, plan) + _publish_release_notes(args, plan) + _publish_report_to_wiki(args, plan) + n = done if args.apply else len(writes) + print(f"{'applied' if args.apply else 'dry-run'}: {n} write(s)") + return 0 + + +# --------------------------------------------------------------------------- # +# Wiki 发布（R3-wiki：周报上传到 Wiki 周报目录） +# --------------------------------------------------------------------------- # +_WIKI_CLI_BIN = "gitlink-cli" + + +def _find_wiki_cli() -> str: + """找到 gitlink-cli 可执行文件；优先复用 _gw 的 which 逻辑。""" + if _gw is not None and hasattr(_gw, "shutil_which"): + path = _gw.shutil_which(_WIKI_CLI_BIN) + if path: + return path + from shutil import which + return which(_WIKI_CLI_BIN) or _WIKI_CLI_BIN + + +def _run_wiki(args, command: list[str], label: str) -> bool: + """执行 gitlink-cli wiki 子命令；dry-run 打印，--apply 真写。返回是否实际执行。""" + cli = _find_wiki_cli() + cmd = [cli, "wiki", *command, "--owner", args.owner, "--repo", args.repo] + if not args.apply: + print(f"[dry-run] {label}: {' '.join(cmd)}") + return False + try: + proc = subprocess.run(cmd, capture_output=True, text=True, timeout=30) + if proc.returncode != 0: + err = proc.stderr.strip() or proc.stdout.strip() or "未知错误" + print(f"[fail] {label}: {err}", file=sys.stderr) + return False + print(f"[ok] {label}") + return True + except Exception as exc: # noqa: BLE001 + print(f"[fail] {label}: {exc}", file=sys.stderr) + return False + + +def _ensure_wiki_dir(args, dir_name: str) -> bool: + """确保 Wiki 侧边栏中存在目录；不存在则创建。""" + # 先检查目录是否已存在（通过读 Sidebar 判断） + cli = _find_wiki_cli() + check_cmd = [cli, "wiki", "+view", "--name", "_Sidebar", + "--owner", args.owner, "--repo", args.repo] + try: + proc = subprocess.run(check_cmd, capture_output=True, text=True, timeout=15) + if proc.returncode == 0 and f"- {dir_name}" not in proc.stdout: + return _run_wiki(args, ["+mkdir", "--name", dir_name], + f"wiki mkdir {dir_name}") + except Exception: + pass + return True # 无法判断时跳过，让 create --dir 自行报错 + + +def _publish_report_to_wiki(args, plan) -> None: + """R3-wiki：把周报上传到 Wiki 周报目录（支持幂等：同日期运行更新而非报错）。""" + body = (plan.get("report_md") or "").strip() + if not body: + return + if not getattr(args, "publish_wiki", False): + return + + wiki_dir = getattr(args, "wiki_dir", None) or "周报专区" + now = datetime.now(timezone.utc) + page_name = f"周报-{now.strftime('%Y-%m-%d-%H%M')}" + commit_msg = f"自动生成周报 {now.strftime('%Y-%m-%d %H:%M')}" + + # Step 1: 确保目录存在 + _ensure_wiki_dir(args, wiki_dir) + + # Step 2: 尝试创建 → 失败则更新（幂等） + ok = _run_wiki(args, [ + "+create", "--name", page_name, + "--content", body, + "--message", commit_msg, + "--dir", wiki_dir, + ], f"wiki create {page_name} → {wiki_dir}") + if not ok: + _run_wiki(args, [ + "+update", "--name", page_name, + "--content", body, + "--message", commit_msg, + ], f"wiki update {page_name} (fallback)") + + +def cmd_checkpoint(args) -> int: + now = datetime.now(timezone.utc) + Path(args.checkpoint).parent.mkdir(parents=True, exist_ok=True) + Path(args.checkpoint).write_text(now.isoformat(), encoding="utf-8") + log = Path(args.triage_log) + entry = _read_json(Path(args.plan)) if args.plan and Path(args.plan).exists() else {"writes": []} + batch = [ + f"\n## 批次 {now.isoformat()}（{args.owner}/{args.repo}）\n", + f"- writes: {len(entry.get('writes', []))}\n", + ] + log.parent.mkdir(parents=True, exist_ok=True) + with log.open("a", encoding="utf-8") as fh: + fh.writelines(batch) + print(f"checkpoint → {args.checkpoint} ({now.isoformat()})") + return 0 + + +def _generate_report(candidates: dict, plan: dict | None = None) -> str: + """生成七段式社区运营周报（纯 Markdown 文本拼接，非 HTML 渲染）。 + + Wiki 平台负责将 Markdown 转为页面展示，我们只负责数据→内容的编排。 + """ + repo = candidates.get("repo", "") + owner = candidates.get("owner", "") + since_str = candidates.get("since", "")[:10] + now = datetime.now(timezone.utc) + week_end = now.strftime("%Y-%m-%d") + week_range = f"{since_str} ~ {week_end}" + + trends = candidates.get("trends", {}) or {} + current = candidates.get("current_counts", {}) or {} + contributors = candidates.get("contributors", []) or [] + writes = (plan or {}).get("writes", []) or [] + + issues = candidates.get("issues", []) or [] + merged_prs = candidates.get("merged_prs", []) or [] + open_prs = candidates.get("open_prs", []) or [] + + lines: list[str] = [] + + # ── 标题 ── + lines.append(f"# {repo} 社区运营周报") + lines.append("") + lines.append(f"> 📅 {week_range} | 🤖 由 community-ops-sweep 自动生成") + lines.append("") + + # ── 一、执行摘要 ── + lines.append("## 一、📊 执行摘要") + lines.append("") + summary_placeholder = candidates.get("executive_summary", "").strip() + if summary_placeholder: + lines.append(summary_placeholder) + else: + lines.append("> ⚠️ 执行摘要未生成。运行 Workflow 时由 LLM Agent 自动填充。") + lines.append("") + + # ── 二、核心指标 ── + lines.append("## 二、📈 核心指标") + lines.append("") + lines.append("| 指标 | 本周数值 | 趋势 |") + lines.append("|------|:--------:|:----:|") + _metric_row(lines, "Issue 总数", current, "issues_total", trends) + _metric_row(lines, "打开 Issue", current, "issues_open", trends) + _metric_row(lines, "已关闭 Issue", current, "issues_closed", trends) + _metric_row(lines, "待审 PR", current, "prs_open", trends) + _metric_row(lines, "已合并 PR", current, "prs_merged", trends) + _metric_row(lines, "活跃贡献者", current, "contributors", trends) + lines.append("") + + # ── 三、新增 Issue 清单 ── + lines.append("## 三、🆕 新增 Issue 清单") + lines.append("") + if issues: + lines.append("| 编号 | 标题 | 标签 | 提交者 |") + lines.append("|------|------|------|--------|") + for i in issues: + num = i.get("number", "-") + title = (i.get("title") or "(untitled)")[:60] + labels = ", ".join(i.get("labels", [])[:4]) or "-" + author = i.get("author_login", "-") + lines.append(f"| #{num} | {title} | {labels} | @{author} |") + else: + lines.append("本周无新增 Issue。") + lines.append("") + + # ── 四、PR 活动 ── + lines.append("## 四、🔀 PR 活动") + lines.append("") + + lines.append(f"### 已合并 PR（{len(merged_prs)} 个）") + if merged_prs: + lines.append("") + lines.append("| 编号 | 标题 | 作者 | 合并时间 |") + lines.append("|------|------|------|----------|") + for p in merged_prs[:15]: + num = p.get("number", "-") + title = (p.get("title") or "(untitled)")[:60] + author = p.get("author_login", "-") + merged_at = _fmt_dt(p.get("merged_at") or p.get("updated_at")) + lines.append(f"| !{num} | {title} | @{author} | {merged_at} |") + else: + lines.append("") + lines.append("本周无合并 PR。") + lines.append("") + + lines.append(f"### 待审 PR（{len(open_prs)} 个）") + if open_prs: + lines.append("") + lines.append("| 编号 | 标题 | 作者 | 创建时间 |") + lines.append("|------|------|------|----------|") + for p in open_prs[:15]: + num = p.get("number", "-") + title = (p.get("title") or "(untitled)")[:60] + author = p.get("author_login", "-") + created = _fmt_dt(p.get("created_at")) + lines.append(f"| !{num} | {title} | @{author} | {created} |") + else: + lines.append("") + lines.append("无待审 PR 🎉") + lines.append("") + + # ── 五、贡献者活跃榜 ── + lines.append("## 五、👥 贡献者活跃榜") + lines.append("") + if contributors: + lines.append("| 排名 | 贡献者 | Issue 提交 | PR 合并 | 活跃度 |") + lines.append("|:----:|--------|:----------:|:-------:|:------:|") + for idx, c in enumerate(contributors, 1): + score = c["issues"] + c["prs"] + bar = "█" * min(score, 10) + lines.append(f"| {idx} | @{c['login']} | {c['issues']} | {c['prs']} | {bar} |") + else: + lines.append("本周无活跃贡献者。") + lines.append("") + + # ── 六、自动 Triage 日志 ── + lines.append("## 六、🤖 自动 Triage 日志") + lines.append("") + if writes: + lines.append("| Issue | 操作 | 详情 |") + lines.append("|-------|------|------|") + for w in writes: + issue_num = f"#{w.get('issue', '-')}" + op = w.get("op", "-") + detail = _triage_detail(w) + lines.append(f"| {issue_num} | {op} | {detail} |") + else: + lines.append("本次 sweep 未执行写操作（无变更或 dry-run）。") + lines.append("") + + # ── 七、待办与风险 ── + lines.append("## 七、⚠️ 待办与风险") + lines.append("") + stale_issues = [i for i in issues if _is_stale(i, since_str)] + stale_prs = [p for p in open_prs if _is_stale_pr(p, since_str)] + if stale_issues: + lines.append(f"- 🔴 **{len(stale_issues)} 个 Issue** 超过窗口期未更新，建议优先处理") + if stale_prs: + lines.append(f"- 🟡 **{len(stale_prs)} 个 PR** 等待 Review 超过窗口期，建议推进") + if open_prs: + lines.append(f"- 📋 **{len(open_prs)} 个 PR** 待 Review，建议分配 Reviewer") + if not stale_issues and not stale_prs: + lines.append("- ✅ 当前无超期未处理的 Issue 或 PR。") + lines.append("") + + lines.append("---") + lines.append("") + lines.append(f"*报告生成时间：{now.strftime('%Y-%m-%d %H:%M:%S UTC')} | 数据来源：[{owner}/{repo}](https://gitlink.org.cn/{owner}/{repo})*") + lines.append("") + + return "\n".join(lines) + + +def _metric_row(lines: list[str], label: str, counts: dict, key: str, trends: dict): + val = counts.get(key, "-") + trend = trends.get(key, "→") + lines.append(f"| {label} | {val} | {trend} |") + + +def _fmt_dt(dt) -> str: + if isinstance(dt, datetime): + return dt.strftime("%m-%d") + if isinstance(dt, str): + return dt[:10] + return "-" + + +def _is_stale(issue: dict, since_str: str) -> bool: + updated = issue.get("updated_at") + if isinstance(updated, datetime): + return updated.strftime("%Y-%m-%d") < since_str + if isinstance(updated, str): + return updated[:10] < since_str + return False + + +def _is_stale_pr(pr: dict, since_str: str) -> bool: + updated = pr.get("updated_at") + if isinstance(updated, datetime): + return updated.strftime("%Y-%m-%d") < since_str + if isinstance(updated, str): + return updated[:10] < since_str + return False + + +def _triage_detail(w: dict) -> str: + op = w.get("op", "") + if op == "update_tags": + added = ", ".join(w.get("tag_names", [])) + removed = ", ".join(w.get("remove_tag_names", [])) + parts = [] + if added: + parts.append(f"+{added}") + if removed: + parts.append(f"-{removed}") + return " ".join(parts) if parts else "标签调整" + if op == "update_status": + return f"→ {w.get('state', '?')} ({w.get('reason', '')})" + if op == "comment": + kind = w.get("kind", "") + return {"owner_suggest": "建议负责人", "link_close": "自动关闭通知"}.get(kind, "评论") + if op == "update_assigner": + return f"指派 {', '.join(w.get('assigner_ids', []))}" + return "-" + + +def _render_release_notes(candidates: dict) -> str: + if _gw is None: + return "" + try: + repo_info = {"name": candidates.get("repo", ""), "description": ""} + issues = [coerce_dates(i) for i in candidates.get("issues", [])] + prs = [coerce_dates(p, ("updated_at", "created_at", "merged_at")) for p in candidates.get("merged_prs", [])] + summary = _gw.summarize_workflow( + repo_info, issues, prs, [], + datetime.now(timezone.utc), 7, + ) + return _gw.render_release_notes(summary) + except Exception as exc: # noqa: BLE001 + print(f"[warn] render_release_notes failed: {exc}", file=sys.stderr) + return "" + + +def main(argv: list[str] | None = None) -> int: + p = argparse.ArgumentParser(description="community-ops-sweep 确定性引擎") + sub = p.add_subparsers(dest="cmd", required=True) + + pc = sub.add_parser("collect", help="采集 → candidates.json") + pc.add_argument("--owner", required=True); pc.add_argument("--repo", required=True) + pc.add_argument("--since"); pc.add_argument("--out", default="candidates.json") + pc.add_argument("--checkpoint", default=".last-sweep") + pc.set_defaults(func=cmd_collect) + + pp = sub.add_parser("plan", help="candidates + decisions → plan.json") + pp.add_argument("--candidates", default="candidates.json") + pp.add_argument("--triage"); pp.add_argument("--owners"); pp.add_argument("--links") + pp.add_argument("--summary", help="LLM 生成的执行摘要 JSON 文件") + pp.add_argument("--routing"); pp.add_argument("--policy") + pp.add_argument("--out", default="plan.json") + pp.set_defaults(func=cmd_plan) + + pa = sub.add_parser("apply", help="执行 plan.json（默认 dry-run）") + pa.add_argument("--plan", default="plan.json") + pa.add_argument("--owner", required=True); pa.add_argument("--repo", required=True) + pa.add_argument("--apply", action="store_true") + pa.add_argument("--report-issue", dest="report_issue", help="R3：把周报作为评论发到此 issue") + pa.add_argument("--release-id", dest="release_id", help="R4：把 release notes 合并进此 release body") + pa.add_argument("--publish-wiki", action="store_true", dest="publish_wiki", + help="R3-wiki：把周报上传到 Wiki 周报目录") + pa.add_argument("--wiki-dir", dest="wiki_dir", default="周报专区", + help="Wiki 周报目录名（默认：周报专区）") + pa.set_defaults(func=cmd_apply) + + pk = sub.add_parser("checkpoint", help="推进 .last-sweep + 追加 triage-log") + pk.add_argument("--owner", required=True); pk.add_argument("--repo", required=True) + pk.add_argument("--checkpoint", default=".last-sweep") + pk.add_argument("--triage-log", default="docs/issue-triage-log.md") + pk.add_argument("--plan") + pk.set_defaults(func=cmd_checkpoint) + + args = p.parse_args(argv) + return args.func(args) + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/workflows/community-ops-sweep/tests/test_sweep.py b/workflows/community-ops-sweep/tests/test_sweep.py new file mode 100644 index 0000000..3fdd637 --- /dev/null +++ b/workflows/community-ops-sweep/tests/test_sweep.py @@ -0,0 +1,149 @@ +#!/usr/bin/env python3 +"""community_ops_sweep 引擎纯逻辑单测 —— 把"同输入→同 plan"钉成可回归事实。 + +只测确定性纯函数（标签增删 / 状态映射 / 链接校验 / plan 组装），不碰 gitlink-cli I/O。 +纯标准库 unittest（与 gatekeeper test_scoring.py 同风格）。 + +运行： + python3 examples/workflows/community-ops-sweep/tests/test_sweep.py +""" + +from __future__ import annotations + +import importlib.util +import sys +import unittest +from datetime import datetime, timezone +from pathlib import Path + +# 以绝对路径加载被测脚本（非包，按文件载入；先注册 sys.modules 以兼容 dataclass+annotations） +_SCRIPT = ( + Path(__file__).resolve().parent.parent / "scripts" / "community_ops_sweep.py" +) +_spec = importlib.util.spec_from_file_location("community_ops_sweep", _SCRIPT) +assert _spec and _spec.loader, f"无法定位被测脚本：{_SCRIPT}" +sw = importlib.util.module_from_spec(_spec) +sys.modules["community_ops_sweep"] = sw +_spec.loader.exec_module(sw) # type: ignore[union-attr] + +TAG_MAP = {"bug": "3", "feature": "5", "docs": "7", "question": "9", "enhancement": "11"} +POLICY = { + "status_map": {"duplicate": "closed", "answered": "closed", "spam": "closed"}, +} + + +class TestMergeTagIds(unittest.TestCase): + def test_add_to_empty(self): + self.assertEqual(sw.merge_tag_ids([], ["bug"], [], TAG_MAP), ["3"]) + + def test_add_preserves_existing_union(self): + # 整体替换语义：现有 ∪ 新增 + self.assertEqual(sw.merge_tag_ids(["3", "7"], ["feature"], [], TAG_MAP), ["3", "5", "7"]) + + def test_remove_drops_tag(self): + self.assertEqual(sw.merge_tag_ids(["3", "7"], [], ["bug"], TAG_MAP), ["7"]) + + def test_add_and_remove_together(self): + # current[3,7] ∪ {5} \ {7} = {3,5} + self.assertEqual(sw.merge_tag_ids(["3", "7"], ["feature"], ["docs"], TAG_MAP), ["3", "5"]) + + def test_unknown_add_filtered_by_whitelist(self): + self.assertEqual(sw.merge_tag_ids([], ["nonexistent"], [], TAG_MAP), []) + + def test_result_sorted_deduped(self): + self.assertEqual(sw.merge_tag_ids(["7", "3"], ["bug"], [], TAG_MAP), ["3", "7"]) + + +class TestMapStatus(unittest.TestCase): + def test_duplicate_closes(self): + self.assertEqual(sw.map_status("duplicate", POLICY), "closed") + + def test_unknown_action_no_change(self): + self.assertIsNone(sw.map_status("schedule_fix", POLICY)) + + def test_none_action_no_change(self): + self.assertIsNone(sw.map_status(None, POLICY)) + + +class TestBuildPlanTriage(unittest.TestCase): + def _candidates(self, issues): + return {"issues": issues, "merged_prs": [], "open_issues": {i["number"] for i in issues}} + + def test_plan_adds_tag(self): + candidates = self._candidates([{"number": "42", "title": "登录失败", "description": "", "tags": [], "state": "open"}]) + triage = [{"issue": "42", "tag_names": ["bug"], "remove_tag_names": [], "recommended_action": "schedule_fix"}] + plan = sw.build_plan(candidates, triage, [], [], {}, POLICY, TAG_MAP) + tag_writes = [w for w in plan["writes"] if w.get("issue") == "42" and w["op"] == "update_tags"] + self.assertTrue(any(w["tag_ids"] == ["3"] for w in tag_writes), plan["writes"]) + + def test_plan_removes_tag(self): + candidates = self._candidates([{"number": "42", "title": "x", "description": "", "tags": [{"id": "3", "name": "bug"}], "state": "open"}]) + triage = [{"issue": "42", "tag_names": [], "remove_tag_names": ["bug"], "recommended_action": "schedule_fix"}] + plan = sw.build_plan(candidates, triage, [], [], {}, POLICY, TAG_MAP) + tag_writes = [w for w in plan["writes"] if w.get("issue") == "42" and w["op"] == "update_tags"] + # 期望全集为空 → tag_ids == [] + self.assertTrue(any(w["tag_ids"] == [] for w in tag_writes), plan["writes"]) + + def test_plan_closes_duplicate(self): + candidates = self._candidates([{"number": "43", "title": "dup", "description": "", "tags": [], "state": "open"}]) + triage = [{"issue": "43", "tag_names": [], "remove_tag_names": [], "recommended_action": "duplicate"}] + plan = sw.build_plan(candidates, triage, [], [], {}, POLICY, TAG_MAP) + status_writes = [w for w in plan["writes"] if w.get("issue") == "43" and w["op"] == "update_status"] + self.assertTrue(any(w["state"] == "closed" for w in status_writes), plan["writes"]) + + +class TestBuildPlanLinks(unittest.TestCase): + def test_link_closes_only_open_existing_issue(self): + candidates = { + "issues": [], + "merged_prs": [{"number": "128", "title": "fix login", "description": "fixes #45", "author": "alice"}], + "open_issues": {"45"}, # 45 开着；99 不在 → 不存在/未开 + } + links = [{"pr": "128", "linked_issue_numbers": ["45", "99"]}] + plan = sw.build_plan(candidates, [], [], links, {}, POLICY, TAG_MAP) + closes = [w for w in plan["writes"] if w["op"] == "update_status" and w.get("state") == "closed"] + closed = {w["issue"] for w in closes} + self.assertIn("45", closed) + self.assertNotIn("99", closed) + + def test_link_no_match_no_write(self): + candidates = {"issues": [], "merged_prs": [{"number": "1", "title": "x", "description": "", "author": "a"}], "open_issues": set()} + links = [{"pr": "1", "linked_issue_numbers": []}] + plan = sw.build_plan(candidates, [], [], links, {}, POLICY, TAG_MAP) + self.assertFalse(any(w["op"] == "update_status" for w in plan["writes"])) + + +class TestBuildPlanOwnerSuggest(unittest.TestCase): + def test_owner_suggestion_emits_comment_not_assign(self): + # R2 默认建议：写 comment，不写 assign + candidates = {"issues": [{"number": "50", "title": "t", "description": "", "tags": [], "state": "open"}], "merged_prs": [], "open_issues": {"50"}} + owners = [{"issue": "50", "suggested_owners": ["alice", "bob"], "evidence": "近期 PR#128"}] + plan = sw.build_plan(candidates, [], owners, [], {}, POLICY, TAG_MAP) + comments = [w for w in plan["writes"] if w.get("issue") == "50" and w["op"] == "comment"] + assigns = [w for w in plan["writes"] if w.get("issue") == "50" and w["op"] == "update_assigner"] + self.assertTrue(comments, "应有建议评论") + self.assertFalse(assigns, "默认不应自动指派") + + +class TestCoerceDates(unittest.TestCase): + """candidates.json 经磁盘 round-trip 后日期变字符串；渲染前要 coerce 回 datetime。""" + + def test_parses_string_dates(self): + out = sw.coerce_dates({"updated_at": "2026-06-15T10:00:00+00:00", "created_at": "2026-06-10T00:00:00+00:00", "x": 1}) + self.assertIsInstance(out["updated_at"], datetime) + self.assertIsInstance(out["created_at"], datetime) + self.assertEqual(out["x"], 1) + + def test_datetime_and_none_passthrough(self): + dt = datetime(2026, 6, 1, tzinfo=timezone.utc) + out = sw.coerce_dates({"updated_at": dt, "created_at": None}) + self.assertIs(out["updated_at"], dt) + self.assertIsNone(out["created_at"]) + + def test_custom_keys(self): + out = sw.coerce_dates({"merged_at": "2026-06-15T10:00:00+00:00"}, keys=("merged_at",)) + self.assertIsInstance(out["merged_at"], datetime) + + +if __name__ == "__main__": + unittest.main(verbosity=2)