190 lines
7.6 KiB
Python
190 lines
7.6 KiB
Python
"""gitlink-flow：社区运营自动化端到端工作流编排器。
|
||
|
||
串联多个步骤，对一个真实 GitLink 仓库执行完整的社区运营自动化：
|
||
|
||
采集（glapi）
|
||
→ ① Issue 自动分拣（triage）
|
||
→ ② PR Review 汇总（pr-review）
|
||
→ ③ Release Notes 生成（release-notes）
|
||
→ ④ 社区健康体检（复用 gitlink-scaffold）
|
||
→ ⑤ 贡献者致谢（复用 gitlink-contributor）
|
||
→ ⑥ 生成社区运营周报（汇总以上全部）
|
||
|
||
串联 6 个步骤、复用 5 个自研 Skill 的能力，远超"≥3 个调用"的要求。
|
||
数据来自 GitLink 公开 API（只读），无需登录。
|
||
|
||
用法：
|
||
python flow.py --owner Gitlink --repo gitlink-cli
|
||
python flow.py --owner Gitlink --repo gitlink-cli --format json
|
||
python flow.py --config examples/config.json
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import argparse
|
||
import json
|
||
import sys
|
||
from datetime import datetime
|
||
from pathlib import Path
|
||
from typing import Any
|
||
|
||
sys.path.insert(0, str(Path(__file__).resolve().parent))
|
||
from glapi import GitLinkClient, GitLinkError, split_owner_repo
|
||
import cli
|
||
import steps
|
||
import report as report_mod
|
||
|
||
|
||
def collect(owner: str, repo: str, client: GitLinkClient,
|
||
commit_pages: int = 4, use_cli: bool | None = None) -> dict[str, Any]:
|
||
"""采集阶段：优先走 gitlink-cli 命令（主调用链），失败回退直连 API。
|
||
|
||
赛题要求工作流组合 gitlink-cli 已有命令。因此 repo/issue/pr/release
|
||
四类数据优先调用 `gitlink-cli +list/+info`；当本机未装 gitlink-cli 或
|
||
某命令调用失败时，回退到 glapi 直连，保证工作流不因依赖缺失而中断。
|
||
|
||
commits 与目录树（list_dir）gitlink-cli 暂无对应只读命令，沿用 glapi。
|
||
|
||
use_cli 为 None 时自动探测本机是否安装 gitlink-cli；显式传 False 可强制
|
||
走 glapi 直连（供离线单元测试使用）。
|
||
"""
|
||
if use_cli is None:
|
||
use_cli = cli.cli_available()
|
||
source = "gitlink-cli 命令" if use_cli else "直连 API（未检测到 gitlink-cli，已回退）"
|
||
|
||
def via_cli(cli_fn, fallback_fn):
|
||
"""单项数据：优先 cli，任何失败回退 glapi。"""
|
||
if use_cli:
|
||
try:
|
||
return cli_fn()
|
||
except cli.CliError:
|
||
pass
|
||
return fallback_fn()
|
||
|
||
info = via_cli(lambda: cli.repo_info(owner, repo),
|
||
lambda: client.repo_info(owner, repo))
|
||
issues = via_cli(lambda: cli.issues(owner, repo, limit=50),
|
||
lambda: client.issues(owner, repo, limit=50))
|
||
pulls = via_cli(lambda: cli.pulls(owner, repo, limit=50),
|
||
lambda: client.pulls(owner, repo, limit=50))
|
||
releases = via_cli(lambda: cli.releases(owner, repo),
|
||
lambda: client.releases(owner, repo))
|
||
# commits / 目录树：gitlink-cli 无对应只读命令，直接用 glapi
|
||
commits = client.commits(owner, repo, max_pages=commit_pages)
|
||
contributors = client.contributors(owner, repo)
|
||
try:
|
||
root_entries = client.list_dir(owner, repo, "", "master")
|
||
root_files = [str(e.get("name", "")) for e in root_entries]
|
||
except GitLinkError:
|
||
root_files = []
|
||
|
||
return {"source": source, "info": info, "issues": issues, "pulls": pulls,
|
||
"commits": commits, "contributors": contributors,
|
||
"releases": releases, "root_files": root_files}
|
||
|
||
|
||
def run_flow(owner: str, repo: str, client: GitLinkClient | None = None,
|
||
commit_pages: int = 4, use_cli: bool | None = None) -> dict[str, Any]:
|
||
"""对单个仓库执行完整工作流，返回各步骤结果。"""
|
||
client = client or GitLinkClient()
|
||
|
||
# ===== 采集阶段（gitlink-cli 主调用链 + glapi fallback）=====
|
||
bundle = collect(owner, repo, client, commit_pages=commit_pages, use_cli=use_cli)
|
||
info = bundle["info"]
|
||
issues = bundle["issues"]
|
||
pulls = bundle["pulls"]
|
||
commits = bundle["commits"]
|
||
contributors = bundle["contributors"]
|
||
releases = bundle["releases"]
|
||
root_files = bundle["root_files"]
|
||
|
||
# ===== 分析阶段（6 步）=====
|
||
latest_version = "Unreleased"
|
||
if releases:
|
||
latest_version = releases[0].get("tag_name") or releases[0].get("name") or "Unreleased"
|
||
|
||
result = {
|
||
"owner": owner,
|
||
"repo": repo,
|
||
"generated_at": datetime.now().strftime("%Y-%m-%d %H:%M"),
|
||
"data_source": bundle["source"],
|
||
"repo_info": {
|
||
"name": info.get("name"),
|
||
"issues_count": info.get("issues_count"),
|
||
"pull_requests_count": info.get("pull_requests_count"),
|
||
"praises_count": info.get("praises_count"),
|
||
"forked_count": info.get("forked_count"),
|
||
},
|
||
"step1_triage": steps.triage_issues(issues),
|
||
"step2_pr_review": steps.pr_review_summary(pulls),
|
||
"step3_release_notes": steps.release_notes(commits, version=latest_version),
|
||
"step4_health": steps.health_check(root_files),
|
||
"step5_contributors": steps.contributor_highlights(contributors),
|
||
}
|
||
# 第 6 步：汇总周报
|
||
result["step6_weekly_report"] = report_mod.render_weekly(result)
|
||
return result
|
||
|
||
|
||
def main(argv: list[str] | None = None) -> int:
|
||
p = argparse.ArgumentParser(prog="gitlink-flow",
|
||
description="社区运营自动化端到端工作流")
|
||
p.add_argument("--owner", help="仓库所有者")
|
||
p.add_argument("--repo", help="仓库名称")
|
||
p.add_argument("--slug", help="owner/repo 或完整 URL")
|
||
p.add_argument("--config", type=Path, help="批量配置 JSON（含 repos 列表）")
|
||
p.add_argument("--commit-pages", type=int, default=4, help="提交采集页数（每页 50）")
|
||
p.add_argument("--format", choices=["markdown", "json"], default="markdown")
|
||
p.add_argument("--output", type=Path, help="输出文件")
|
||
p.add_argument("--output-dir", type=Path, help="批量模式输出目录")
|
||
args = p.parse_args(argv)
|
||
|
||
targets: list[tuple[str, str]] = []
|
||
if args.config:
|
||
cfg = json.loads(args.config.read_text(encoding="utf-8"))
|
||
for item in cfg.get("repos", []):
|
||
if item.get("owner") and item.get("repo"):
|
||
targets.append((item["owner"], item["repo"]))
|
||
if args.slug:
|
||
targets.append(split_owner_repo(args.slug))
|
||
elif args.owner and args.repo:
|
||
targets.append((args.owner, args.repo))
|
||
|
||
if not targets:
|
||
print("错误：请用 --owner/--repo 或 --slug 或 --config 指定仓库。", file=sys.stderr)
|
||
return 2
|
||
|
||
client = GitLinkClient()
|
||
exit_code = 0
|
||
for owner, repo in targets:
|
||
print(f"[工作流] 处理 {owner}/{repo} ...", flush=True)
|
||
try:
|
||
result = run_flow(owner, repo, client=client, commit_pages=args.commit_pages)
|
||
except GitLinkError as exc:
|
||
print(f" 失败：{exc}", file=sys.stderr)
|
||
exit_code = 1
|
||
continue
|
||
|
||
if args.format == "json":
|
||
out = json.dumps(result, ensure_ascii=False, indent=2)
|
||
else:
|
||
out = result["step6_weekly_report"]
|
||
|
||
if args.output_dir:
|
||
args.output_dir.mkdir(parents=True, exist_ok=True)
|
||
ext = "json" if args.format == "json" else "md"
|
||
fp = args.output_dir / f"{owner}_{repo}_flow.{ext}"
|
||
fp.write_text(out, encoding="utf-8")
|
||
print(f" 已写入 {fp}", flush=True)
|
||
elif args.output:
|
||
args.output.parent.mkdir(parents=True, exist_ok=True)
|
||
args.output.write_text(out, encoding="utf-8")
|
||
print(f" 已写入 {args.output}", flush=True)
|
||
else:
|
||
print(out)
|
||
return exit_code
|
||
|
||
|
||
if __name__ == "__main__":
|
||
raise SystemExit(main())
|