feat(adopt): 既有工程接入器 + 接入文档（含 PR 锁死陷阱的判定）
问题：只有"新工程生成器"，没有"既有工程怎么接进来"的可复用路径； 而手工接入时最容易踩的坑是——盲目开"必须状态检查通过"，会让 PR 永远等一个不会出现的检查，永远合不了。 新增： - scripts/adopt-existing.py：体检 → 加指针（分支+PR）→ 开保护 → 打印自检命令。 默认 DRY-RUN 不动任何东西；幂等（重复跑不改 main）。 关键设计：状态检查是否开启由体检判定——只有"推任意分支都会跑"的工作流， PR 头提交才带着 (push) 状态；判定不通过就只开"禁直推+必须 PR"。 - 06-接入既有工程.md：判定表（带 7 个仓的实测结论）、一条命令、试点记录、 手工四步、验收判据、会遇到的拒绝与处理、扩到其余仓的顺序。 试点实测（Weknora/knowledge-governance）：PR #1 → CI context build / build (push) success → mergeable=True（证明状态检查认 PR）→ 合并 32b0ec58a612 → AGENTS.md 进 main；直推被 pre-receive 拒绝； 再跑 --apply 幂等，main 不变。 顺带修两个自查出来的毛病： - ~/.git-credentials 里 LAN 主机是 URL 编码的（192.168.50.223%3a3000）， 按字面匹配会报"没有凭据"——改成解码后比对。 - runner 只在仓库级查，aiball 的 runner 挂在组织上，会误报"无 runner"。
This commit is contained in:
姚章浩
+8
-6
@@ -10,13 +10,15 @@
|
||||
|
||||
## P0
|
||||
|
||||
### 1. 分支保护（4 个仓库全无）
|
||||
### 1. 分支保护（试点 1/7 完成）
|
||||
|
||||
- **现状**：`GET /api/v1/repos/{r}/branch_protections` 对 `Weknora/dobotai-platform`、`Weknora/knowledge-governance`、`Weknora/knowledge-adapter-service`、`aiball/software-pc` 全部返回 `[]`。
|
||||
- **要做什么**：Gitea → 仓库 → Settings → Branches → 给 `main` 加规则：禁止直接推送、必须走 PR、必须状态检查通过。
|
||||
- **为什么**：没有保护，"提交规范 + 评审分级"就全是自觉；`main` 被 force-push 一次，"线上跑的是哪个提交"就永久失去答案。
|
||||
- **怎么验证**：重跑 `GET /repos/{r}/branch_protections`，返回非空且 `enable_push=false`、`required_approvals` 符合 §1.2。
|
||||
- **谁来做**：本人（网页操作，需仓库管理权限）。
|
||||
- **现状**：`Weknora/knowledge-governance` **已接入**（`enable_push=false` + 必须 PR + 强制 `build / build (push)`，见 `06-接入既有工程.md` §3 试点记录）。其余 6 个仓当前**无保护**（期中一次全量铺开已按指示回退）。
|
||||
- **要做什么**：逐个跑 `python3 ~/eng-standards/scripts/adopt-existing.py <owner/repo> --apply`（先 DRY-RUN 看计划）。
|
||||
⚠️ **顺序有讲究**：先补 CI，再开强制状态检查 —— 反过来会把 PR 锁死（判据见 `06` §2）。
|
||||
- **为什么**：没有保护，"提交规范 + 评审分级"全是自觉；`main` 被 force-push 一次，"线上跑的是哪个提交"就永久失去答案。
|
||||
- **怎么验证**：`GET /repos/{r}/branch_protections` 非空且 `enable_push=false`；**直推被服务端拒绝**（真跑一次，见 `06` §5）。
|
||||
- **谁来做**：开发（脚本可代做）；**凭据/授权环节仍由本人把关**。
|
||||
⚠️ 分支保护涉及平台设置，属可逆操作（网页可随时关掉），所以锁死风险低；但"PR 等待不存在的检查"这个坑必须靠 §2 的体检避开。
|
||||
|
||||
### 2. `knowledge-adapter-service` 的工作流不在 `main` 上
|
||||
|
||||
|
||||
+133
@@ -0,0 +1,133 @@
|
||||
# 接入既有工程：让现在的工程启用这套规范
|
||||
|
||||
> 一句话：接入 = **前置体检 → 加指针 → 开保护 → 反向自检**，一条命令走完。
|
||||
> 为什么要有"前置体检"这一关，见 §2 —— 跳过它会把 PR **永久锁死**。
|
||||
|
||||
## 1. 一条命令
|
||||
|
||||
```bash
|
||||
# ① 只体检 + 打印计划（**不动任何东西**）—— 先看，再决定
|
||||
python3 ~/eng-standards/scripts/adopt-existing.py <owner/repo>
|
||||
|
||||
# ② 计划无误再执行
|
||||
python3 ~/eng-standards/scripts/adopt-existing.py <owner/repo> --apply
|
||||
|
||||
# 可选：只加指针、不动分支保护
|
||||
python3 ~/eng-standards/scripts/adopt-existing.py <owner/repo> --apply --no-protection
|
||||
```
|
||||
|
||||
工具只依赖 Python 标准库 + git；凭据运行时从 `~/.git-credentials` 读，**永不打印**。
|
||||
**幂等**：重复跑不会重复建 PR、不会改 `main`。
|
||||
|
||||
它做的事：
|
||||
|
||||
| 步 | 动作 | 怎么落地 |
|
||||
|---|---|---|
|
||||
| 1 | 体检 | 读默认分支、工作流触发方式、runner（仓库级/组织级）、最新提交的 status context、已有保护 |
|
||||
| 2 | 加指针 | 新起分支 → 写 `AGENTS.md` → 提交（过本机闸门）→ **PR** → 等 CI → 合并 → 删分支 |
|
||||
| 3 | 开保护 | `POST/PATCH /branch_protections`：禁直推 + 禁 force push + 必须 PR；**状态检查按第 1 步判定** |
|
||||
| 4 | 给自检命令 | 打印"直推必须被拒"的验证命令（**这步工具不代做，必须你跑**） |
|
||||
|
||||
⚠️ `main` 一旦被保护，**所有改动都得走 PR** —— 包括工具自己加 `AGENTS.md` 这一步。
|
||||
|
||||
## 2. 前置体检为什么是必须的（本环境最容易踩的坑）
|
||||
|
||||
"必须状态检查通过"开错了，症状是：**PR 打开后永远等一个不会出现的检查，永远合不了**。
|
||||
判据只有一条：**该仓的工作流，在 PR 的头提交上会不会产生 status？**
|
||||
|
||||
| 工作流触发 | PR 头提交有没有状态 | 能否开强制状态检查 |
|
||||
|---|---|---|
|
||||
| `on: push: branches: "**"`（推**任意分支**都跑） | 有 —— 分支推送产生 `xxx (push)` | ✅ **可以** |
|
||||
| `on: push: branches: [main]`，无 `pull_request` | **没有** —— PR 完全不触发 | ⛔ 开了就锁死 |
|
||||
| `on: push: branches: [main]` + `pull_request` | 有，但 context 是 `xxx (pull_request)` | ⛔ 与 `(push)` 不匹配；要开得按 `(pull_request)` 人工确认 |
|
||||
|
||||
工具按这张表自动判定，**判定不通过就只开"禁直推 + 必须 PR"，不碰状态检查**。
|
||||
|
||||
实测证据（本机 7 个仓）：
|
||||
|
||||
| 仓 | 工作流触发 | 判定 |
|
||||
|---|---|---|
|
||||
| `Weknora/knowledge-governance` | `push: "**"` | ✅ 可开 —— context `build / build (push)` |
|
||||
| `Weknora/dobotai-platform` | `push: "**"` | ✅ 可开 —— 同上 |
|
||||
| `aiball/software-pc` | `push: [main]`，无 `pull_request` | ⛔ 不可开 |
|
||||
| `aiball/protocol-docs` | `push: [main]` + `pull_request` | ⛔ 上下文不匹配 |
|
||||
| `Weknora/knowledge-adapter-service` | 默认分支上**没有**工作流 | ⛔ 无 CI 可强制 |
|
||||
| `Weknora/Enterprise-Knowledge-Base` | 本仓无工作流（现有 status 来自仓外） | ⛔ 不可开 |
|
||||
| `zheromiku/eng-standards-main` | 无工作流 | ⛔ 不可开 |
|
||||
|
||||
**要做的是先补 CI，再开检查** —— 不是先开检查（那会锁死）。所以接入顺序是：
|
||||
**工作流能覆盖任意分支 → 再开强制状态检查**。工作流只覆盖 `main` 的仓（aiball 两个），
|
||||
第一步是把它改成 `on: push: branches: "**"` 或补一个 `pull_request` 触发。
|
||||
|
||||
## 3. 试点记录：`Weknora/knowledge-governance`（已跑通）
|
||||
|
||||
执行前该仓状态：无 `AGENTS.md`、无分支保护；`build.yml` 是 `push: "**"`；仓库级 runner `runner-weknora-kg-01`。
|
||||
|
||||
| 步 | 实测结果 |
|
||||
|---|---|
|
||||
| 分支 + 提交 | `docs/agents-pointer`，提交通过本机提交闸门（`~/.githooks/commit-msg`） |
|
||||
| 推送触发 CI | 头提交 `d025fc9adfe2` → context `build / build (push)` → `combined=success` |
|
||||
| PR | **PR #1**，`mergeable=True`（**关键结论：强制状态检查认 PR** —— 因为推分支就产生了 `(push)` 状态） |
|
||||
| 合并 | HTTP 200（**空 body，属正常**，不重试）→ 读回 `state=closed` `merged=true` `merge_commit_sha=32b0ec58a612` |
|
||||
| 结果 | `AGENTS.md` 在 `main`；分支已自动删除 |
|
||||
| 反向自检 | 直推 main 被服务端拒绝：`Not allowed to push to protected branch main` / `pre-receive hook declined` |
|
||||
| 幂等 | 再跑一次 `--apply`：`AGENTS.md` 跳过、保护值不变、**main 仍是 `32b0ec58a612`** |
|
||||
|
||||
## 4. 不想用工具？手工四步
|
||||
|
||||
```bash
|
||||
# ① 体检（判据同 §2）
|
||||
python3 ~/.hermes/scripts/gitea_api.py GET "/api/v1/repos/<ns>/<repo>/actions/tasks?limit=1"
|
||||
git clone <repo> /tmp/t && cd /tmp/t
|
||||
python3 -c "import re;print([l for l in open('.gitea/workflows/build.yml') if re.match(r'^\s*(on|push|pull_request|branches):',l)])"
|
||||
docker ps --format '{{.Names}}' | grep runner # 本仓有没有 runner
|
||||
|
||||
# ② 加指针：走分支 + PR，不要直推 main
|
||||
git switch -c docs/agents-pointer
|
||||
# AGENTS.md 内容取自 ~/eng-standards/templates/README.md 里的 ```markdown 骨架块，
|
||||
# 把 <PROJECT> / <ns>/<repo> 替换掉，并如实写下该仓的 CI / runner / 保护现状
|
||||
git commit -F - <<'EOF'
|
||||
docs(agents): 加 AGENTS.md 操作手册与规范指针
|
||||
|
||||
指向系统级规范（zheromiku/eng-standards-main）。
|
||||
EOF
|
||||
git push -u origin HEAD # 然后网页开 PR → 等 CI → 合并
|
||||
|
||||
# ③ 开保护：网页 Settings → Branches；或 API（字段见 swagger CreateBranchProtectionOption）
|
||||
# enable_push=false、enable_force_push=false、required_approvals=0、
|
||||
# block_on_outdated_branch=true、dismiss_stale_approvals=true、
|
||||
# enable_status_check=<按 §2 判定>、status_check_contexts=<匹配的 context>
|
||||
|
||||
# ④ 反向自检（必须）
|
||||
cd /tmp/t && git commit --allow-empty -m 'chore(probe): 不该进 main' && git push origin main
|
||||
# 判据：Not allowed to push to protected branch
|
||||
```
|
||||
|
||||
## 5. 接入后的验收判据（缺一即不算接入）
|
||||
|
||||
1. `AGENTS.md`（或 `.hermes.md`）在默认分支上，且含指向 `zheromiku/eng-standards-main` 的路径。
|
||||
2. `GET /repos/{r}/branch_protections` 非空，`enable_push=false`。
|
||||
3. 直推默认分支**被服务端拒绝**（不是"我没敢推"，是真被拒）。
|
||||
4. PR 能正常合并（证明强制状态检查没有把 PR 锁死）—— **只有开了状态检查的仓需要这条**。
|
||||
5. 会话里问"这个仓提交信息什么规范"，代理能**不靠你重复**答出来（说明指针真的被注入了）。
|
||||
|
||||
## 6. 会遇到的拒绝与处理
|
||||
|
||||
| 现象 | 原因 | 处理 |
|
||||
|---|---|---|
|
||||
| 推 `main` 被拒 | 已开保护 —— 这是**预期行为** | 走分支 + PR |
|
||||
| PR 一直显示等待状态检查 | 该仓工作流没覆盖 PR（§2 表后两行） | 先改工作流触发，或临时关掉 `enable_status_check` |
|
||||
| PR 报 `mergeable=false` 但没有冲突 | 强制状态检查未满足 / 分支过期 | 看 `GET /pulls/{n}` 的 `mergeable` 与 head 上的 status |
|
||||
| 合并接口返回空 body | **Gitea 合并成功的正常返回** | **永不重试**，改 `GET /pulls/{n}` 读回 `merged=true` |
|
||||
| run 一直 `queued` | 该仓没有归属自己的 runner | `bash /srv/aiball/runner/deploy_repo_runner.sh <owner/repo> <name>` |
|
||||
| 工具说"没有凭据" | `~/.git-credentials` 里主机是 URL 编码的 | 工具已按解码后比对；手工查时别按字面比 |
|
||||
|
||||
## 7. 扩到其余仓的顺序建议
|
||||
|
||||
先补 CI 再开检查，别反过来：
|
||||
|
||||
1. `Weknora/knowledge-adapter-service` —— 先把 `build.yml` 从 `ci/setup` 合到 `main`（现在 main 上零校验）
|
||||
2. `aiball/software-pc`、`aiball/protocol-docs` —— 先把工作流触发改成覆盖 PR，再开强制检查
|
||||
3. `Weknora/dobotai-platform` —— 工作流已覆盖任意分支，可直接 `--apply`
|
||||
4. `Weknora/Enterprise-Knowledge-Base` —— 需先接 CI + runner（分支也待收敛）
|
||||
5. `zheromiku/eng-standards-main` —— 等守门脚本做完，再给自己开检查
|
||||
@@ -5,7 +5,7 @@
|
||||
> **状态**：本次**只制定规范**。不部署生产、不运行附件原脚本、不改任何仓库。
|
||||
> **正本仓库**：`zheromiku/eng-standards-main`（Gitea，https://gitea.zheromiku.top/zheromiku/eng-standards-main）；本机工作副本 `~/eng-standards/`。
|
||||
>
|
||||
> **怎么读**：`README` 三十秒版 → `01-规范` 全文 → `02-流程说明与Runbook`（怎么操作）→ `03-待配置事项`（还差什么）→ `04-模板评估与修正`（原稿哪里不对）→ `05-应用机制`（怎么让它以后默认生效）。
|
||||
> **怎么读**：`README` 三十秒版 → `01-规范` 全文 → `02-流程说明与Runbook`（怎么操作）→ `03-待配置事项`（还差什么）→ `04-模板评估与修正`（原稿哪里不对）→ `05-应用机制`（怎么让它以后默认生效）→ `06-接入既有工程`（手头这个仓怎么接进来）。
|
||||
|
||||
## 交付物
|
||||
|
||||
@@ -16,9 +16,11 @@
|
||||
| `02-流程说明与Runbook.md` | 日常动作的**逐步命令**（提交前、看 CI、接入新仓、发布、回滚） |
|
||||
| `03-待配置事项.md` | 逐项：现状 / 要做什么 / 为什么 / 怎么验证 / 谁来做 |
|
||||
| `04-模板评估与修正.md` | 附件 15 处问题的证据与改法（含实测反证） |
|
||||
| `05-应用机制-让规范在以后所有工程生效.md` | 三渠道（技能 / 仓内指针 / 生成器+守门）、已配 vs 待配、新工程 5 步、验证记录 |
|
||||
| `05-应用机制-让规范在以后所有工程生效.md` | 三渠道（技能 / 仓内指针 / 生成器+守门）、已配 vs 待配、验证记录 |
|
||||
| `06-接入既有工程.md` | **既有仓怎么接进来**：前置体检（含 PR 锁死陷阱）、一条命令、试点记录、验收判据 |
|
||||
| `templates/` | 可直接落仓的骨架：`gitea-build.yml`、`gitea-prod-deploy.yml`、`gitignore`、`commit-msg`、`ci_gate.py`、`ci_report.py`、`README.md`（内含 AGENTS.md 骨架） |
|
||||
| `scripts/new-project.sh` | 新工程生成器：一条命令产出 9 个默认合规文件（已端到端测过） |
|
||||
| `scripts/adopt-existing.py` | 既有工程接入器：体检 → 加指针 → 开保护（幂等，默认 DRY-RUN） |
|
||||
|
||||
## 0. 三十秒版（只记五句）
|
||||
|
||||
|
||||
Binary file not shown.
Executable
+336
@@ -0,0 +1,336 @@
|
||||
#!/usr/bin/env python3
|
||||
"""把一个**既有** Gitea 工程接入工程规范（正本 §8 的可执行版）。
|
||||
|
||||
python3 adopt-existing.py <owner/repo> [--apply] [--no-protection]
|
||||
|
||||
默认**只体检 + 打印计划，不动任何东西**。加 --apply 才真的改。
|
||||
|
||||
为什么入口分成两边（体检 / 执行）：
|
||||
"强制状态检查"开错了会把 PR **永久锁死** —— 检查永远没人报，PR 永远合不了。
|
||||
所以必须先体检判定该仓的工作流**能不能覆盖 PR 头提交**，判定通过才允许开。
|
||||
|
||||
只依赖 Python 标准库 + git。凭据运行时从 ~/.git-credentials 读，**永不打印**。
|
||||
"""
|
||||
import base64
|
||||
import json
|
||||
import os
|
||||
import re
|
||||
import shutil
|
||||
import subprocess
|
||||
import sys
|
||||
import tempfile
|
||||
import time
|
||||
import urllib.error
|
||||
import urllib.parse
|
||||
import urllib.request
|
||||
from pathlib import Path
|
||||
|
||||
HOSTS = ["http://192.168.50.223:3000", "https://gitea.zheromiku.top"]
|
||||
STANDARDS_REPO = "zheromiku/eng-standards-main"
|
||||
|
||||
|
||||
def D(v):
|
||||
"""JSON 对象，非 dict 一律当空。"""
|
||||
return v if isinstance(v, dict) else {}
|
||||
|
||||
|
||||
def L(v):
|
||||
"""JSON 数组，非 list 一律当空。"""
|
||||
return v if isinstance(v, list) else []
|
||||
|
||||
|
||||
def load_token(netloc: str) -> str:
|
||||
"""~/.git-credentials 里主机可能是 URL 编码的（`192.168.50.223%3a3000`），
|
||||
所以先解码 netloc 再比 —— 按字面比会找不到凭据，而症状是"没有凭据"。"""
|
||||
fp = Path.home() / ".git-credentials"
|
||||
want = netloc.lower()
|
||||
for line in fp.read_text(encoding="utf-8", errors="replace").splitlines():
|
||||
m = re.match(r"^https?://([^:]+):([^@]+)@(.+)$", line)
|
||||
if not m:
|
||||
continue
|
||||
got = urllib.parse.unquote(m.group(3)).strip().lower()
|
||||
if got == want:
|
||||
return urllib.parse.unquote(m.group(2))
|
||||
return ""
|
||||
|
||||
|
||||
def pick_host():
|
||||
for base in HOSTS:
|
||||
try:
|
||||
with urllib.request.urlopen(f"{base}/api/v1/version", timeout=6) as x:
|
||||
if x.status == 200:
|
||||
return base
|
||||
except Exception:
|
||||
continue
|
||||
return None
|
||||
|
||||
|
||||
class Gitea:
|
||||
def __init__(self):
|
||||
self.base = pick_host() or ""
|
||||
if not self.base:
|
||||
sys.exit("❌ 连不上任何 Gitea 候选地址，先确认网络/代理")
|
||||
self.host = urllib.parse.urlparse(self.base).netloc # 用于 git clone / 显示
|
||||
self.tok = load_token(self.host)
|
||||
if not self.tok:
|
||||
sys.exit(f"❌ ~/.git-credentials 里没有 {self.host} 的凭据")
|
||||
# 提交令牌走 URL 里的 basic auth 会进 shell 历史/进程表，所以一律用 credential helper
|
||||
|
||||
def api(self, path, method="GET", body=None):
|
||||
"""返回 (status, value)。value 可能是 dict/list；空 body 归一化成 {}。
|
||||
⚠️ 空 body 是 Gitea 合并**成功**的返回，不能当成失败（正本里写过的坑）。"""
|
||||
data = json.dumps(body).encode() if body is not None else None
|
||||
req = urllib.request.Request(f"{self.base}/api/v1{path}", data=data, method=method)
|
||||
req.add_header("Authorization", "token " + self.tok)
|
||||
req.add_header("Content-Type", "application/json")
|
||||
try:
|
||||
with urllib.request.urlopen(req, timeout=30) as x:
|
||||
raw = x.read().decode()
|
||||
if not raw.strip():
|
||||
return x.status, {}
|
||||
try:
|
||||
return x.status, json.loads(raw)
|
||||
except json.JSONDecodeError:
|
||||
return x.status, {"_raw": raw[:200]}
|
||||
except urllib.error.HTTPError as e:
|
||||
return e.code, {"_error": e.read().decode()[:300]}
|
||||
|
||||
def file(self, repo, path, ref=None):
|
||||
p = f"/repos/{repo}/contents/{urllib.parse.quote(path)}"
|
||||
if ref:
|
||||
p += f"?ref={urllib.parse.quote(ref)}"
|
||||
st, b = self.api(p)
|
||||
if st != 200:
|
||||
return None
|
||||
try:
|
||||
return base64.b64decode(D(b).get("content", "")).decode("utf-8", "replace")
|
||||
except Exception:
|
||||
return ""
|
||||
|
||||
def branch(self, repo, name):
|
||||
st, b = self.api(f"/repos/{repo}/branches/{urllib.parse.quote(name, safe='')}")
|
||||
return D(D(b).get("commit")).get("id") if st == 200 else None
|
||||
|
||||
|
||||
def workflow_triggers(text: str):
|
||||
"""启发式解析 `on:` 块（不引 YAML：本仓不许引依赖）。"""
|
||||
lines = text.splitlines()
|
||||
start = next((i for i, l in enumerate(lines) if re.match(r"^on\s*:", l)), None)
|
||||
if start is None:
|
||||
return {"push_all": False, "push_any_branch": False, "pull_request": False}
|
||||
block = []
|
||||
for l in lines[start + 1:]:
|
||||
if l.strip() and not l.startswith((" ", "\t", "#")):
|
||||
break
|
||||
block.append(l)
|
||||
raw = "\n".join(block)
|
||||
has_push = bool(re.search(r"^\s*push\s*:", raw, re.M))
|
||||
# "推任意分支"：push 块里没有 branches 过滤，或 branches 里写了 **
|
||||
push_all = has_push and (not re.search(r"^\s{2,}branches\s*:", raw, re.M) or "**" in raw)
|
||||
return {"push_all": push_all, "push_any_branch": push_all,
|
||||
"pull_request": bool(re.search(r"^\s*pull_request\s*:", raw, re.M))}
|
||||
|
||||
|
||||
def preflight(g: Gitea, repo: str):
|
||||
rep = {}
|
||||
st, r = g.api(f"/repos/{repo}")
|
||||
if st != 200:
|
||||
sys.exit(f"❌ 取不到仓库 {repo}：HTTP {st} {str(r)[:120]}")
|
||||
rep["default_branch"] = D(r).get("default_branch") or "main"
|
||||
rep["empty"] = bool(D(r).get("empty"))
|
||||
|
||||
st, lst = g.api(f"/repos/{repo}/contents/.gitea/workflows")
|
||||
rep["workflows"] = [D(x).get("name", "") for x in L(lst)] if st == 200 else []
|
||||
rep["triggers"] = {}
|
||||
for wf in rep["workflows"]:
|
||||
t = g.file(repo, f".gitea/workflows/{wf}")
|
||||
if t is not None:
|
||||
rep["triggers"][wf] = workflow_triggers(t)
|
||||
|
||||
st, rr = g.api(f"/repos/{repo}/actions/runners")
|
||||
rep["runners"] = [D(x).get("name", "") for x in L(D(rr).get("runners"))] if st == 200 else []
|
||||
rep["runners_scope"] = "仓库级" if rep["runners"] else ""
|
||||
if not rep["runners"]: # aiball 的 runner 挂在组织上，只查仓库级会误报"无 runner"
|
||||
st2, orr = g.api(f"/orgs/{repo.split('/')[0]}/actions/runners")
|
||||
if st2 == 200:
|
||||
rep["runners"] = [D(x).get("name", "") for x in L(D(orr).get("runners"))]
|
||||
rep["runners_scope"] = "组织级" if rep["runners"] else ""
|
||||
|
||||
st, cs = g.api(f"/repos/{repo}/commits?limit=1")
|
||||
sha = D(L(cs)[0]).get("sha") if st == 200 and L(cs) else None
|
||||
rep["head"] = sha
|
||||
rep["contexts"], rep["combined"] = [], None
|
||||
if sha:
|
||||
st, s = g.api(f"/repos/{repo}/commits/{sha}/status")
|
||||
if st == 200:
|
||||
rep["contexts"] = sorted({c for c in (D(x).get("context") for x in L(D(s).get("statuses"))) if c})
|
||||
rep["combined"] = D(s).get("state")
|
||||
|
||||
rep["has_agents"] = g.file(repo, "AGENTS.md") is not None
|
||||
rep["has_hermes_md"] = g.file(repo, ".hermes.md") is not None
|
||||
st, bp = g.api(f"/repos/{repo}/branch_protections")
|
||||
rep["protections"] = [(D(x).get("branch_name"), D(x).get("enable_status_check"),
|
||||
D(x).get("status_check_contexts")) for x in L(bp)] if st == 200 else []
|
||||
return rep
|
||||
|
||||
|
||||
def verdict(rep):
|
||||
"""能不能开"强制状态检查"？只有"推任意分支都会跑"的工作流，PR 头提交才拿得到状态。"""
|
||||
why = []
|
||||
for wf, t in rep["triggers"].items():
|
||||
if t["push_all"]:
|
||||
why.append(f"{wf}: on.push 覆盖所有分支 → PR 头提交带着 (push) 状态 ✅ 可用")
|
||||
elif t["pull_request"]:
|
||||
why.append(f"{wf}: push 只覆盖 main、但有 pull_request → PR 走 (pull_request) 上下文，"
|
||||
f"与 (push) 不匹配，需人工确认后按 (pull_request) 开 ⛔ 本工具不自动开")
|
||||
else:
|
||||
why.append(f"{wf}: 只在 push[main] 触发且无 pull_request → PR 不产生任何状态 ⛔ 不可开")
|
||||
allowed = any(t["push_all"] for t in rep["triggers"].values())
|
||||
ctxs = [c for c in rep["contexts"] if c and c.endswith("(push)")] if allowed else []
|
||||
return allowed and bool(ctxs), ctxs, why
|
||||
|
||||
|
||||
def agents_md(repo: str, rep: dict):
|
||||
tpl = Path(__file__).resolve().parent.parent / "templates" / "README.md"
|
||||
m = re.search(r"^````markdown\n(.*?)^````", tpl.read_text(encoding="utf-8"), re.S | re.M)
|
||||
if not m:
|
||||
sys.exit("❌ templates/README.md 里找不到 ```markdown 骨架块")
|
||||
name = repo.split("/")[1]
|
||||
body = m.group(1).replace("<PROJECT>", name).replace("<ns>/<repo>", repo)
|
||||
facts = "\n".join([
|
||||
"## 1b. 接入时核实的现状（事实变了就更新；别让它过期）", "",
|
||||
"| 项 | 值 |", "|---|---|",
|
||||
f"| 默认分支 | `{rep['default_branch']}` —— **已启用分支保护**：禁止直接推送、必须走 PR |",
|
||||
f"| CI | 工作流 {rep['workflows'] or '（无）'}；status context {rep['contexts'] or '（无）'} |",
|
||||
f"| Runner | {rep['runners'] or '（无本仓 runner —— 有工作流也会永远 queued）'}{'（' + rep['runners_scope'] + '）' if rep['runners_scope'] else ''} |", "",
|
||||
])
|
||||
return body.replace("## 2. 硬约束", facts + "\n## 2. 硬约束", 1)
|
||||
|
||||
|
||||
def run_git(*args, cwd=None, stdin=None):
|
||||
return subprocess.run(["git", *args], cwd=cwd, input=stdin, capture_output=True, text=True)
|
||||
|
||||
|
||||
def add_pointer(g: Gitea, repo: str, rep: dict, apply: bool):
|
||||
if rep["has_agents"] or rep["has_hermes_md"]:
|
||||
return "已有上下文文件，跳过"
|
||||
if not apply:
|
||||
return "计划：走分支 + PR 新增 AGENTS.md（不推 main）"
|
||||
tmp = tempfile.mkdtemp(prefix="adopt-")
|
||||
d = os.path.join(tmp, repo.split("/")[1])
|
||||
br = "docs/agents-pointer"
|
||||
try:
|
||||
if run_git("clone", "-q", "--depth", "1", f"{g.base}/{repo}.git", d).returncode:
|
||||
return "❌ clone 失败"
|
||||
run_git("switch", "-q", "-c", br, cwd=d)
|
||||
with open(os.path.join(d, "AGENTS.md"), "w", encoding="utf-8") as fh:
|
||||
fh.write(agents_md(repo, rep))
|
||||
run_git("add", "AGENTS.md", cwd=d)
|
||||
msg = ("docs(agents): 加 AGENTS.md 操作手册与规范指针\n\n"
|
||||
f"指向系统级规范（{STANDARDS_REPO}），并写明本仓接入时核实的现状。\n"
|
||||
"为什么用 AGENTS.md 而不是 .hermes.md：Codex / Claude Code / OpenCode 读的也是 AGENTS.md。\n")
|
||||
if run_git("commit", "-q", "-F", "-", cwd=d, stdin=msg).returncode:
|
||||
return "❌ commit 被提交闸门拦下（提交信息不合规）"
|
||||
if run_git("push", "-q", "-u", "origin", br, cwd=d).returncode:
|
||||
return "❌ push 失败"
|
||||
head = run_git("rev-parse", "HEAD", cwd=d).stdout.strip()
|
||||
st, pr = g.api(f"/repos/{repo}/pulls", "POST",
|
||||
{"title": "docs(agents): 加 AGENTS.md 操作手册与规范指针",
|
||||
"head": br, "base": rep["default_branch"],
|
||||
"body": f"指向系统级规范 `{STANDARDS_REPO}`，记录本仓接入时核实的现状。未改动任何代码。"})
|
||||
if st not in (200, 201):
|
||||
return f"❌ 建 PR 失败 HTTP {st}"
|
||||
prn = D(pr).get("number")
|
||||
for _ in range(30): # 有 CI 的仓要等状态；没有的立刻退出等待
|
||||
_, s = g.api(f"/repos/{repo}/commits/{head}/status")
|
||||
if D(s).get("state") not in ("pending", None):
|
||||
break
|
||||
time.sleep(6)
|
||||
g.api(f"/repos/{repo}/pulls/{prn}/merge", "POST",
|
||||
{"Do": "merge", "head_commit_id": head, "delete_branch_after_merge": True})
|
||||
_, back = g.api(f"/repos/{repo}/pulls/{prn}") # 读回；不信 POST 的返回
|
||||
merged = bool(D(back).get("merged"))
|
||||
return f"PR #{prn} → 合并={merged}（读回 merge_commit_sha={str(D(back).get('merge_commit_sha') or '')[:12]}）"
|
||||
finally:
|
||||
shutil.rmtree(tmp, ignore_errors=True)
|
||||
|
||||
|
||||
def protection_payload(rep, allow, ctxs):
|
||||
return {"branch_name": rep["default_branch"], "enable_push": False, "enable_force_push": False,
|
||||
"required_approvals": 0, "block_on_outdated_branch": True, "block_on_rejected_reviews": True,
|
||||
"dismiss_stale_approvals": True, "require_signed_commits": False,
|
||||
"enable_status_check": bool(allow), "status_check_contexts": ctxs if allow else []}
|
||||
|
||||
|
||||
def set_protection(g: Gitea, repo: str, rep: dict, apply: bool, allow: bool, ctxs, no_protection: bool):
|
||||
if no_protection:
|
||||
return "按 --no-protection 跳过"
|
||||
plan = f"保护 {rep['default_branch']}（禁直推 + 必须 PR；强制状态检查 {'开 ' + str(ctxs) if allow else '不开'}）"
|
||||
if not apply:
|
||||
return "计划：" + plan
|
||||
payload = protection_payload(rep, allow, ctxs)
|
||||
st, resp = g.api(f"/repos/{repo}/branch_protections", "POST", payload)
|
||||
if st in (403, 409): # 已存在 → 改
|
||||
st, resp = g.api(f"/repos/{repo}/branch_protections/{urllib.parse.quote(rep['default_branch'], safe='')}",
|
||||
"PATCH", payload)
|
||||
if st not in (200, 201):
|
||||
return f"❌ 设置失败 HTTP {st} {str(resp)[:100]}"
|
||||
return "已设置 —— " + plan
|
||||
|
||||
|
||||
def main():
|
||||
args = [a for a in sys.argv[1:] if not a.startswith("--")]
|
||||
if not args:
|
||||
sys.exit(__doc__)
|
||||
repo = args[0]
|
||||
apply = "--apply" in sys.argv
|
||||
no_protection = "--no-protection" in sys.argv
|
||||
|
||||
g = Gitea()
|
||||
print(f"Gitea: {g.base} 仓库: {repo} 模式: "
|
||||
f"{'APPLY（会改）' if apply else 'DRY-RUN（只体检，不动任何东西）'}\n")
|
||||
rep = preflight(g, repo)
|
||||
|
||||
print("── 体检 ──")
|
||||
print(f" 默认分支 : {rep['default_branch']}")
|
||||
print(f" 工作流 : {rep['workflows'] or '（无）'}")
|
||||
for wf, t in rep["triggers"].items():
|
||||
print(f" {wf}: push_覆盖所有分支={t['push_all']} pull_request={t['pull_request']}")
|
||||
print(f" 本仓 runner : "
|
||||
f"{(str(rep['runners']) + '（' + rep['runners_scope'] + '）') if rep['runners'] else '（无 —— 有工作流也会永远 queued）'}")
|
||||
print(f" 最新提交 : {str(rep['head'] or '')[:12]} combined={rep['combined']} "
|
||||
f"contexts={rep['contexts']}")
|
||||
print(f" AGENTS.md/.hermes.md : {rep['has_agents']}/{rep['has_hermes_md']}")
|
||||
print(f" 现有分支保护 : {rep['protections'] or '（无）'}")
|
||||
|
||||
allow, ctxs, why = verdict(rep)
|
||||
print("\n── 状态检查判定（决定 PR 会不会被锁死）──")
|
||||
for line in (why or ["（本仓没有工作流 → 不能开状态检查）"]):
|
||||
print(" · " + line)
|
||||
print(f" ⇒ 强制状态检查: {'✅ 可以开 —— ' + str(ctxs) if allow else '⛔ 不开（开了 PR 永远合不了）'}")
|
||||
|
||||
print("\n── 计划 ──")
|
||||
print(" 1. AGENTS.md : " + add_pointer(g, repo, rep, False))
|
||||
print(" 2. 分支保护 : " + set_protection(g, repo, rep, False, allow, ctxs, no_protection))
|
||||
|
||||
if not apply:
|
||||
print("\n（DRY-RUN：什么都没改。确认计划无误后加 --apply 重跑）")
|
||||
return
|
||||
|
||||
print("\n── 执行 ──")
|
||||
print(" AGENTS.md : " + add_pointer(g, repo, rep, True))
|
||||
print(" 分支保护 : " + set_protection(g, repo, rep, True, allow, ctxs, no_protection))
|
||||
_, bp = g.api(f"/repos/{repo}/branch_protections")
|
||||
for b in L(bp):
|
||||
d = D(b)
|
||||
print(f" 读回: {d.get('branch_name')} enable_push={d.get('enable_push')} "
|
||||
f"status_check={d.get('enable_status_check')} ctx={d.get('status_check_contexts')}")
|
||||
print("\n ⚠️ 还差一步反向自检（必须做，别跳）：")
|
||||
print(f" t=$(mktemp -d); git clone {g.base}/{repo}.git $t; cd $t \\")
|
||||
print(f" && git commit --allow-empty -m 'chore(probe): 不该进 main' \\")
|
||||
print(f" && git push origin {rep['default_branch']}")
|
||||
print(" 判据：必须被拒 —— Not allowed to push to protected branch")
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
main()
|
||||
Reference in New Issue
Block a user