fix(plugins): 插件市场流水线按审查意见加固

- 插件公钥移入 module/plugin/，随在线更新包分发，旧镜像更新后也能验签目录
- 合并后每个插件在独立的 job 中构建；签名 job 自己下载复用的 zip 并校验
  sha256，逐个核对新 zip 的 id 与来源，单个插件的构建篡改不到别的插件
- 发布只从 4.0-dev 运行；PR 检查使用 base 分支的脚本与 ab_sdk
- 复用判断加入 path，手动触发可全部重新打包
- 打包前拒绝符号链接；摘要转义作者控制的输出；单个条目出错不影响整份摘要
- 已上架版本以已发布的 catalog 为准，不再拉取旧 commit
- 修复 git diff 把提交范围当作路径的问题

.github/workflows/plugin-registry.yml

@@ -1,18 +1,25 @@
 name: Plugin Registry
 
 # 插件市场：plugins/registry/<id>.toml 指向作者仓库的固定 commit。
-# - check：PR 上运行作者的代码（测试、pnpm 构建），所以只读权限、不读任何 secret。
-# - build：合并后打包，同样没有密钥；commit 未变的条目复用已发布的 zip。
-# - sign：只有这个 job 读 PLUGIN_SIGNING_KEY，它不运行作者的任何代码，
-#   只用 build_plugin_catalog.py 读 zip 里的 plugin.toml 并签名。
+# - check：PR 上运行作者的代码（测试、pnpm 构建），只读权限、不读 secret。
+#   检查脚本与 ab_sdk 取自 base 分支，PR 不能改掉检查规则。
+# - plan / build：合并后。build 每个插件一个 job（会运行作者的前端构建，没有密钥），
+#   一个插件的构建篡改不到别的插件。
+# - sign：唯一读 PLUGIN_SIGNING_KEY 的 job，不运行作者代码。它自己下载复用的 zip
+#   并校验 sha256，核对每个新 zip 的 id 与来源，再签名发布。
 on:
   pull_request:
     branches: ["4.0-dev"]
-    paths: ["plugins/**", "scripts/plugin_registry.py", ".github/workflows/plugin-registry.yml"]
+    paths: ["plugins/**"]
   push:
     branches: ["4.0-dev"]
     paths: ["plugins/**"]
   workflow_dispatch:
+    inputs:
+      rebuild:
+        description: "忽略已发布的 zip，全部重新打包（打包逻辑变更后使用）"
+        type: boolean
+        default: false
 
 permissions:
   contents: read
@@ -21,6 +28,9 @@ concurrency:
   group: plugin-registry-${{ github.event_name == 'pull_request' && github.ref || 'publish' }}
   cancel-in-progress: false
 
+env:
+  REBUILD: ${{ inputs.rebuild && 'true' || '' }}
+
 jobs:
   check:
     if: github.event_name == 'pull_request'
@@ -28,7 +38,17 @@ jobs:
     steps:
       - uses: actions/checkout@v4
         with:
-          fetch-depth: 0 # 需要 base 分支上的旧条目
+          ref: ${{ github.event.pull_request.base.sha }}
+          fetch-depth: 0
+          persist-credentials: false
+      - name: Fetch PR head
+        run: |
+          git fetch -q origin "refs/pull/${{ github.event.number }}/head"
+          # 引入本脚本的那个 PR 时 base 上还没有它，只能用 PR 自己的版本
+          if [ ! -f scripts/plugin_registry.py ]; then
+            echo "::warning::base 分支没有 scripts/plugin_registry.py，使用 PR 中的版本"
+            git checkout FETCH_HEAD -- scripts/plugin_registry.py
+          fi
       - &python
         uses: actions/setup-python@v5
         with:
@@ -49,48 +69,85 @@ jobs:
       - &deps
         name: Install backend
         run: mkdir -p backend/config && cd backend && uv sync --group dev
+      - &catalog
+        name: Download published catalog
+        env:
+          GH_TOKEN: ${{ github.token }}
+        run: gh release download plugins -p catalog.json -D old || echo "release plugins 还不存在"
       - name: Check changed entries
         run: |
           set -o pipefail
           uv run --project backend python scripts/plugin_registry.py check \
-            --base "origin/${{ github.base_ref }}" --out dist | tee -a "$GITHUB_STEP_SUMMARY"
+            --base "${{ github.event.pull_request.base.sha }}" --head FETCH_HEAD \
+            --catalog old/catalog.json --out dist | tee -a "$GITHUB_STEP_SUMMARY"
 
-  build:
-    if: github.event_name != 'pull_request'
+  plan:
+    # 带着签名密钥的发布只从 4.0-dev 运行
+    if: github.event_name != 'pull_request' && github.ref == 'refs/heads/4.0-dev'
     runs-on: ubuntu-latest
+    outputs:
+      build: ${{ steps.plan.outputs.build }}
     steps:
       - uses: actions/checkout@v4
+        with:
+          persist-credentials: false
+      - *python
+      - *uv
+      - *deps
+      - *catalog
+      - id: plan
+        run: |
+          uv run --project backend python scripts/plugin_registry.py plan --catalog old/catalog.json ${REBUILD:+--rebuild} > plan.json
+          cat plan.json
+          echo "build=$(jq -c .build plan.json)" >> "$GITHUB_OUTPUT"
+
+  build:
+    needs: plan
+    if: needs.plan.outputs.build != '[]'
+    runs-on: ubuntu-latest
+    strategy:
+      fail-fast: true
+      matrix:
+        id: ${{ fromJSON(needs.plan.outputs.build) }}
+    steps:
+      - uses: actions/checkout@v4
+        with:
+          persist-credentials: false
       - *python
       - *uv
       - *pnpm
       - *node
       - *deps
-      - name: Download current catalog
-        env:
-          GH_TOKEN: ${{ github.token }}
-        run: gh release download plugins -p catalog.json -D old || echo "release plugins 还不存在，全部重新打包"
-      - name: Build
-        run: uv run --project backend python scripts/plugin_registry.py build --old old/catalog.json --out dist
+      - run: uv run --project backend python scripts/plugin_registry.py build-one "${{ matrix.id }}" --out dist
       - uses: actions/upload-artifact@v4
         with:
-          name: plugins-dist
+          name: plugin-${{ matrix.id }}
           path: dist/
           retention-days: 7
 
   sign:
-    needs: build
+    needs: [plan, build]
+    if: always() && needs.plan.result == 'success' && (needs.build.result == 'success' || needs.build.result == 'skipped')
     runs-on: ubuntu-latest
     permissions:
       contents: write
     steps:
       - uses: actions/checkout@v4
-      - uses: astral-sh/setup-uv@v4
         with:
-          version: "latest"
+          persist-credentials: false
+      - *python
+      - *uv
+      - *deps
+      - *catalog
       - uses: actions/download-artifact@v4
         with:
-          name: plugins-dist
+          pattern: plugin-*
           path: dist
+      - name: Collect
+        run: |
+          mkdir -p dist
+          uv run --project backend python scripts/plugin_registry.py collect \
+            --catalog old/catalog.json ${REBUILD:+--rebuild} --dist dist --out stage
       - name: Sign catalog
         env:
           PLUGIN_SIGNING_KEY: ${{ secrets.PLUGIN_SIGNING_KEY }}
@@ -102,7 +159,7 @@ jobs:
           umask 077
           printf '%s\n' "$PLUGIN_SIGNING_KEY" > "$RUNNER_TEMP/key.pem"
           uv run --no-project --with cryptography python scripts/build_plugin_catalog.py \
-            --key "$RUNNER_TEMP/key.pem" --sources dist/sources.json --out release-assets dist/*.zip
+            --key "$RUNNER_TEMP/key.pem" --sources stage/sources.json --out release-assets stage/*.zip
           rm -f "$RUNNER_TEMP/key.pem"
       - name: Publish
         env:

backend/src/module/plugin/installer.py

@@ -40,8 +40,9 @@ GITHUB_REPO = "Auto_Bangumi"
 _DL_HEADERS = {"User-Agent": "AutoBangumi-Plugins"}
 DEFAULT_PLUGINS_ROOT = Path("config") / "plugins"
 # 通用插件目录的公钥（与在线更新分开的一把密钥，私钥在 CI 的 PLUGIN_SIGNING_KEY）：
-# 插件签名只代表“维护者审查过这份代码”，不能用来批准更新包
-PLUGIN_PUBKEY_PATH = Path("ab_plugin_pubkey.pem")
+# 插件签名只代表“维护者审查过这份代码”，不能用来批准更新包。放在 module/ 里，
+# 随在线更新包一起分发（更新包本身由更新密钥验签），旧镜像更新后也能验签目录
+PLUGIN_PUBKEY_PATH = Path(__file__).with_name("ab_plugin_pubkey.pem")
 
 
 def release_base(tag: str) -> str:

backend/src/test/test_plugin_catalog.py

@@ -231,6 +231,8 @@ class TestInstall:
     def test_default_pubkey_is_the_plugin_key_not_the_update_key(self):
         assert PluginInstaller().pubkey_path == PLUGIN_PUBKEY_PATH
         assert PLUGIN_PUBKEY_PATH != DEFAULT_PUBKEY_PATH
+        # 公钥随 module/ 进入在线更新包，旧镜像更新后也有它
+        assert PLUGIN_PUBKEY_PATH.is_file()
 
     async def test_uninstall_leaves_llm_provider_plugins_alone(self, tmp_path, keypair):
         installer = make_installer(tmp_path, keypair, build_zip(tmp_path))

backend/src/test/test_plugin_registry.py

@@ -2,6 +2,7 @@
 
 import hashlib
 import importlib.util
+import json
 from pathlib import Path
 
 import pytest
@@ -115,7 +116,40 @@ def test_committed_web_without_web_src_is_rejected(reg, tmp_path):
         reg.build_web(src)
 
 
-class TestBuild:
+def test_symlink_in_packed_files_is_rejected(reg, tmp_path):
+    src = scaffold(tmp_path)
+    (src / "leak").symlink_to("/etc/hosts")
+    entry = reg.Entry.parse(PLUGIN_ID, entry_text())
+    with pytest.raises(reg.Reject, match="符号链接"):
+        reg.verify(entry, src, None, None)
+
+
+def test_summary_escapes_author_output(reg):
+    result = reg.Result(PLUGIN_ID, False, [("拒绝", "x |\n| tests | ✓ fake |")])
+    text = reg.summary([result], [], [])
+    assert "| tests | ✓ fake |" not in text
+    assert "|" in text
+
+
+def test_unexpected_error_is_reported_per_entry(reg, tmp_path, monkeypatch):
+    monkeypatch.setattr(reg, "_git", lambda *a: (_ for _ in ()).throw(OSError("boom")))
+    result = reg.check_entry(PLUGIN_ID, "base", "head", {}, tmp_path, tmp_path)
+    assert not result.ok and result.lines == [("错误", "OSError: boom")]
+
+
+def published_catalog(**overrides):
+    prev = {
+        "id": PLUGIN_ID,
+        "version": "0.1.0",
+        "repo": "alice/ab-demo",
+        "commit": SHA,
+        "asset": f"{PLUGIN_ID}-0.1.0.zip",
+        "sha256": "0" * 64,
+    }
+    return {PLUGIN_ID: prev | overrides}
+
+
+class TestPublish:
     @pytest.fixture
     def registry(self, reg, tmp_path, monkeypatch):
         root = tmp_path / "registry"
@@ -124,64 +158,94 @@ class TestBuild:
         monkeypatch.setattr(reg, "REGISTRY", root)
         return root
 
-    def test_unchanged_commit_reuses_released_zip(
+    @pytest.mark.parametrize(
+        ("published", "rebuild", "expected"),
+        [
+            ({}, False, "build"),
+            (published_catalog(), False, "reuse"),
+            (published_catalog(), True, "build"),
+            (published_catalog(commit="b" * 40), False, "build"),
+            (published_catalog(path="other"), False, "build"),
+        ],
+    )
+    def test_plan_reuses_only_unchanged_sources(
+        self, reg, registry, published, rebuild, expected
+    ):
+        assert reg.plan(published, rebuild)[expected] == [PLUGIN_ID]
+
+    def test_collect_reuses_released_zip(self, reg, registry, tmp_path, monkeypatch):
+        src = scaffold(tmp_path / "src")
+        data = pack(src, tmp_path / "old", load_manifest(src)).read_bytes()
+        published = published_catalog(
+            sha256=hashlib.sha256(data).hexdigest(), readme="hi"
+        )
+        monkeypatch.setattr(reg, "_download", lambda url: data)
+
+        reg.collect(published, False, tmp_path / "dist", tmp_path / "stage")
+
+        assert (tmp_path / "stage" / f"{PLUGIN_ID}-0.1.0.zip").read_bytes() == data
+        sources = json.loads((tmp_path / "stage" / "sources.json").read_text())
+        assert sources[PLUGIN_ID]["readme"] == "hi"
+
+    def test_collect_rejects_reused_zip_with_wrong_sha256(
         self, reg, registry, tmp_path, monkeypatch
     ):
+        monkeypatch.setattr(reg, "_download", lambda url: b"tampered")
+        with pytest.raises(SystemExit, match="sha256"):
+            reg.collect(
+                published_catalog(), False, tmp_path / "dist", tmp_path / "stage"
+            )
+
+    def test_build_one_then_collect(self, reg, registry, tmp_path, monkeypatch):
         src = scaffold(tmp_path / "src")
-        archive = pack(src, tmp_path / "old", load_manifest(src))
-        data = archive.read_bytes()
-        old = {
-            "plugins": [
-                {
-                    "id": PLUGIN_ID,
-                    "version": "0.1.0",
-                    "repo": "alice/ab-demo",
-                    "commit": SHA,
-                    "asset": archive.name,
-                    "readme": "hi",
-                    "sha256": hashlib.sha256(data).hexdigest(),
-                }
-            ]
-        }
-        monkeypatch.setattr(reg, "_download", lambda url: data)
-        monkeypatch.setattr(reg, "fetch", lambda *a: pytest.fail("must not fetch"))
+        (src / "README.md").write_text("# demo\n")
+        monkeypatch.setattr(reg, "fetch", lambda entry, workdir: src)
+        reg.build_one(PLUGIN_ID, tmp_path / "dist" / f"plugin-{PLUGIN_ID}")
 
-        sources = reg.build(old, tmp_path / "out", tmp_path / "tmp")
+        reg.collect({}, False, tmp_path / "dist", tmp_path / "stage")
 
-        assert (tmp_path / "out" / archive.name).read_bytes() == data
+        assert (tmp_path / "stage" / f"{PLUGIN_ID}-0.1.0.zip").is_file()
+        sources = json.loads((tmp_path / "stage" / "sources.json").read_text())
         assert sources[PLUGIN_ID] == {
             "repo": "alice/ab-demo",
             "commit": SHA,
-            "readme": "hi",
+            "path": ".",
+            "readme": "# demo\n",
         }
 
-    def test_reused_zip_with_wrong_sha256_fails(
-        self, reg, registry, tmp_path, monkeypatch
-    ):
-        old = {
-            "plugins": [
-                {
-                    "id": PLUGIN_ID,
-                    "version": "0.1.0",
-                    "repo": "alice/ab-demo",
-                    "commit": SHA,
-                    "asset": "x.zip",
-                    "sha256": "0" * 64,
-                }
-            ]
-        }
-        monkeypatch.setattr(reg, "_download", lambda url: b"tampered")
-        with pytest.raises(SystemExit, match="sha256"):
-            reg.build(old, tmp_path / "out", tmp_path / "tmp")
-
-    def test_changed_commit_is_fetched_and_packed(
-        self, reg, registry, tmp_path, monkeypatch
+    @pytest.mark.parametrize("tamper", ["extra_zip", "wrong_source"])
+    def test_collect_rejects_tampered_build_output(
+        self, reg, registry, tmp_path, monkeypatch, tamper
     ):
         src = scaffold(tmp_path / "src")
-        (src / "README.md").write_text("# demo\n")
-        monkeypatch.setattr(reg, "fetch", lambda repo, commit, path, workdir: src)
+        monkeypatch.setattr(reg, "fetch", lambda entry, workdir: src)
+        built = tmp_path / "dist" / f"plugin-{PLUGIN_ID}"
+        reg.build_one(PLUGIN_ID, built)
+        if tamper == "extra_zip":
+            other = scaffold(tmp_path / "other", "other-plugin")
+            pack(other, built, load_manifest(other))
+        else:
+            (built / "source.json").write_text(
+                json.dumps({"repo": "mallory/x", "commit": SHA, "path": "."})
+            )
+        with pytest.raises(SystemExit):
+            reg.collect({}, False, tmp_path / "dist", tmp_path / "stage")
 
-        sources = reg.build({}, tmp_path / "out", tmp_path / "tmp")
 
-        assert (tmp_path / "out" / f"{PLUGIN_ID}-0.1.0.zip").is_file()
-        assert sources[PLUGIN_ID]["readme"] == "# demo\n"
+def test_diff_lists_registry_changes_between_commits(reg, tmp_path, monkeypatch):
+    def git(*args: str) -> None:
+        reg._run(["git", "-c", "user.name=t", "-c", "user.email=t@t", *args], tmp_path)
+
+    git("init", "-q")
+    (tmp_path / "README.md").write_text("x\n")
+    git("add", "-A")
+    git("commit", "-qm", "base")
+    (tmp_path / "plugins" / "registry").mkdir(parents=True)
+    (tmp_path / "plugins" / "registry" / f"{PLUGIN_ID}.toml").write_text(entry_text())
+    git("add", "-A")
+    git("commit", "-qm", "add")
+    monkeypatch.setattr(reg, "ROOT", tmp_path)
+
+    changed = reg._diff("HEAD~1", "HEAD", "--diff-filter=AMR", "--", "plugins/registry")
+
+    assert changed == [f"plugins/registry/{PLUGIN_ID}.toml"]

scripts/build_plugin_catalog.py

@@ -47,6 +47,7 @@ def _entry(archive: Path, min_ab: str, sources: dict[str, dict[str, str]]) -> di
         "has_web": bool(plugin.get("ui")),
         "repo": source.get("repo", ""),
         "commit": source.get("commit", ""),
+        "path": source.get("path", "."),
         "readme": source.get("readme", ""),
         "asset": archive.name,
         "sha256": hashlib.sha256(archive.read_bytes()).hexdigest(),

scripts/plugin_registry.py

@@ -1,4 +1,4 @@
-"""插件市场的索引工具：``plugins/registry/<id>.toml`` → 校验、测试、打包。
+"""插件市场的索引工具：``plugins/registry/<id>.toml`` → 校验、测试、打包、汇总。
 
 条目只有三个字段，代码留在作者自己的仓库，用完整 SHA 固定（tag 可以移动）::
 
@@ -6,21 +6,24 @@
     commit = "<40 位 SHA>"
     path = "."            # 可选，插件在仓库中的子目录
 
-用法（在仓库根目录运行；backend 的虚拟环境里有 ab_sdk）::
+在仓库根目录运行（backend 的虚拟环境里有 ab_sdk）::
 
-    uv run --project backend python scripts/plugin_registry.py check --base origin/4.0-dev --out dist
-    uv run --project backend python scripts/plugin_registry.py build --old old/catalog.json --out dist
+    uv run --project backend python scripts/plugin_registry.py <命令> ...
 
-- ``check``：PR 上运行（没有 secret、只读权限）。只处理本 PR 改动的条目，运行作者的
-  测试，把审查摘要（Markdown）写到 stdout，有条目不合格时以 1 退出。
-- ``build``：合并后运行（同样没有密钥）。commit 未变的条目复用 release 中的 zip（按旧
-  catalog 的 sha256 校验），其余拉取并打包，再写出 ``sources.json``。签名 job 只用
-  ``build_plugin_catalog.py`` 读这些 zip，不运行作者的任何代码。
+- ``check``：PR 上运行（只读、没有 secret），用 base 分支的本脚本。只处理本 PR 改动
+  的条目，运行作者的测试，把审查摘要（Markdown）写到 stdout；有条目不合格时以 1 退出。
+- ``plan``：合并后列出要重新打包的条目（其余复用已发布的 zip）。不运行作者代码。
+- ``build-one``：在独立的 job 里打包一个条目（会运行作者的前端构建，没有密钥）。
+- ``collect``：签名 job 中运行，不运行作者代码。复用的 zip 自己下载并按旧 catalog 的
+  sha256 校验；新打的 zip 逐个核对 id 与来源，再交给 ``build_plugin_catalog.py`` 签名。
+  这样某个插件的构建即使被篡改，也只能影响它自己的 zip。
 """
 
 import argparse
 import hashlib
+import html
 import json
+import os
 import re
 import shutil
 import subprocess
@@ -28,12 +31,13 @@ import sys
 import tempfile
 import tomllib
 import urllib.request
+import zipfile
 from dataclasses import dataclass
 from pathlib import Path, PurePosixPath
 
 from packaging.version import Version
 
-from ab_sdk.cli import pack
+from ab_sdk.cli import _PACK_EXCLUDE_DIRS, pack
 from ab_sdk.manifest import ID_RE, MANIFEST_NAME, RESERVED_IDS, PluginManifest, check
 
 ROOT = Path(__file__).resolve().parent.parent
@@ -85,6 +89,17 @@ class Entry:
             raise Reject("path 必须是仓库内的相对路径")
         return entry
 
+    @property
+    def source(self) -> dict[str, str]:
+        return {"repo": self.repo, "commit": self.commit, "path": self.path}
+
+
+def registry_entries() -> list[Entry]:
+    return [
+        Entry.parse(p.stem, p.read_text("utf-8"))
+        for p in sorted(REGISTRY.glob("*.toml"))
+    ]
+
 
 def _run(cmd: list[str], cwd: Path) -> subprocess.CompletedProcess[str]:
     return subprocess.run(cmd, cwd=cwd, capture_output=True, text=True)
@@ -94,7 +109,7 @@ def _tail(proc: subprocess.CompletedProcess[str], lines: int = 15) -> str:
     return "\n".join((proc.stdout + proc.stderr).strip().splitlines()[-lines:])
 
 
-def fetch(repo: str, commit: str, path: str, workdir: Path) -> Path:
+def fetch(entry: Entry, workdir: Path) -> Path:
     """浅拉取固定 commit，返回插件目录。不执行仓库里的任何代码。"""
     workdir.mkdir(parents=True, exist_ok=True)
     for cmd in (
@@ -105,17 +120,21 @@ def fetch(repo: str, commit: str, path: str, workdir: Path) -> Path:
             "-q",
             "--depth",
             "1",
-            f"https://github.com/{repo}.git",
-            commit,
+            f"https://github.com/{entry.repo}.git",
+            entry.commit,
         ],
         ["git", "checkout", "-q", "FETCH_HEAD"],
     ):
         proc = _run(cmd, workdir)
         if proc.returncode != 0:
-            raise Reject(f"无法拉取 {repo}@{commit[:7]}：{_tail(proc, 3)}")
-    src = workdir / path
+            raise Reject(f"无法拉取 {entry.repo}@{entry.commit[:7]}：{_tail(proc, 3)}")
+    src = workdir / entry.path
+    if not src.resolve().is_relative_to(workdir.resolve()):
+        raise Reject("path 经符号链接指向了仓库之外")
     if not (src / MANIFEST_NAME).is_file():
-        raise Reject(f"{repo}@{commit[:7]} 的 {path} 下没有 {MANIFEST_NAME}")
+        raise Reject(
+            f"{entry.repo}@{entry.commit[:7]} 的 {entry.path} 下没有 {MANIFEST_NAME}"
+        )
     return src
 
 
@@ -145,10 +164,21 @@ def build_web(src: Path) -> None:
         raise Reject(f"pnpm build 失败：\n{_tail(proc)}")
 
 
+def reject_symlinks(src: Path) -> None:
+    """pack 会跟随符号链接读文件：要打包的部分里不允许有符号链接，
+    否则作者可以把 runner 上的文件（如带 token 的 .git/config）打进签名包。"""
+    for root, dirs, names in os.walk(src):
+        dirs[:] = [d for d in dirs if d not in _PACK_EXCLUDE_DIRS]
+        for name in dirs + names:
+            if (Path(root) / name).is_symlink():
+                raise Reject(f"不允许符号链接：{(Path(root) / name).relative_to(src)}")
+
+
 def verify(
     entry: Entry, src: Path, old: Entry | None, old_version: str | None
 ) -> PluginManifest:
     """清单与条目的一致性规则；不执行插件代码。"""
+    reject_symlinks(src)
     manifest, problems = check(src)
     if manifest is None or problems:
         raise Reject("ab-plugin validate 失败：" + "；".join(problems))
@@ -186,21 +216,25 @@ def run_tests(src: Path) -> str:
         raise Reject("tests/ 中没有收集到测试：至少要有 1 个测试")
     if proc.returncode != 0:
         raise Reject(f"测试失败：\n{_tail(proc)}")
-    return proc.stdout.strip().splitlines()[-1]
+    return _tail(proc, 1)
 
 
 def readme(src: Path) -> str:
     for name in ("README.md", "readme.md", "README"):
         if (src / name).is_file():
-            data = (src / name).read_bytes()[:README_LIMIT]
-            return data.decode("utf-8", errors="ignore")
+            return (
+                (src / name)
+                .read_bytes()[:README_LIMIT]
+                .decode("utf-8", errors="ignore")
+            )
     return ""
 
 
-def _manifest_version(repo: str, commit: str, path: str, workdir: Path) -> str:
-    """已上架版本的 version：只读旧 commit 的 plugin.toml。"""
-    src = fetch(repo, commit, path, workdir)
-    return tomllib.loads((src / MANIFEST_NAME).read_text("utf-8"))["plugin"]["version"]
+def load_catalog(path: Path | None) -> dict[str, dict]:
+    """已发布的 catalog：id → 条目。release 还不存在时为空。"""
+    if path is None or not path.is_file():
+        return {}
+    return {p["id"]: p for p in json.loads(path.read_text("utf-8")).get("plugins", [])}
 
 
 def _git(*args: str) -> subprocess.CompletedProcess[str]:
@@ -217,21 +251,27 @@ class Result:
     lines: list[tuple[str, str]]
 
 
-def check_entry(plugin_id: str, base: str, out: Path, tmp: Path) -> Result:
+def check_entry(
+    plugin_id: str,
+    base: str,
+    head: str,
+    published: dict[str, dict],
+    out: Path,
+    tmp: Path,
+) -> Result:
     rel = f"plugins/registry/{plugin_id}.toml"
     lines: list[tuple[str, str]] = []
     try:
-        entry = Entry.parse(plugin_id, (ROOT / rel).read_text("utf-8"))
+        entry = Entry.parse(plugin_id, _git("show", f"{head}:{rel}").stdout)
         old_text = _git("show", f"{base}:{rel}")
         old = (
             Entry.parse(plugin_id, old_text.stdout)
             if old_text.returncode == 0
             else None
         )
+        # 已上架版本以已发布的 catalog 为准：旧 commit 可能已被作者删掉
         old_version = (
-            _manifest_version(old.repo, old.commit, old.path, tmp / "old")
-            if old is not None
-            else None
+            published[plugin_id]["version"] if plugin_id in published else None
         )
         url = f"https://github.com/{entry.repo}"
         tree = f"{url}/tree/{entry.commit}" + (
@@ -240,7 +280,7 @@ def check_entry(plugin_id: str, base: str, out: Path, tmp: Path) -> Result:
         lines.append(
             ("审查", f"{url}/compare/{old.commit}...{entry.commit}" if old else tree)
         )
-        src = fetch(entry.repo, entry.commit, entry.path, tmp / "new")
+        src = fetch(entry, tmp)
         has_web_src = (src / "web-src" / "package.json").is_file()
         build_web(src)
         manifest = verify(entry, src, old, old_version)
@@ -255,34 +295,34 @@ def check_entry(plugin_id: str, base: str, out: Path, tmp: Path) -> Result:
         lines.append(("前端", "web-src/（CI 构建）" if has_web_src else "无"))
         archive = pack(src, out, manifest)
         digest = hashlib.sha256(archive.read_bytes()).hexdigest()
-        lines.append(("zip", f"`{archive.name}` · sha256 `{digest}`"))
+        lines.append(("zip", f"{archive.name} · sha256 {digest}"))
         return Result(plugin_id, True, lines)
     except Reject as e:
         lines.append(("拒绝", str(e)))
-        return Result(plugin_id, False, lines)
+    except Exception as e:  # noqa: BLE001 - 一个条目出错不能让整份摘要丢失
+        lines.append(("错误", f"{type(e).__name__}: {e}"))
+    return Result(plugin_id, False, lines)
 
 
-def changed_ids(base: str) -> tuple[list[str], list[str]]:
-    """本 PR 新增 / 修改与删除的条目 id。"""
-
-    def ids(diff_filter: str) -> list[str]:
-        proc = _git(
-            "diff",
-            "--name-only",
-            f"--diff-filter={diff_filter}",
-            f"{base}...HEAD",
-            "--",
-            "plugins/registry",
-        )
-        if proc.returncode != 0:
-            raise SystemExit(proc.stderr)
-        return sorted(PurePosixPath(p).stem for p in proc.stdout.split())
-
-    return ids("AMR"), ids("D")
+def _diff(base: str, head: str, *args: str) -> list[str]:
+    proc = _git("diff", "--name-only", f"{base}...{head}", *args)
+    if proc.returncode != 0:
+        raise SystemExit(proc.stderr)
+    return proc.stdout.split()
 
 
-def summary(results: list[Result], removed: list[str]) -> str:
+def _cell(value: str) -> str:
+    """作者控制的文本（测试输出等）进表格前转义，不能伪造行或链接。"""
+    return html.escape(value).replace("|", "&#124;").replace("\n", "<br>")
+
+
+def summary(results: list[Result], removed: list[str], others: list[str]) -> str:
     out = ["## 插件索引检查", ""]
+    if others:
+        out += [
+            f"> **注意** 本 PR 还改动了 {len(others)} 个 `plugins/registry/` 以外的文件，需要另行审查。",
+            "",
+        ]
     for r in results:
         out += [
             f"### {'✓' if r.ok else '✗'} {r.id}",
@@ -290,7 +330,7 @@ def summary(results: list[Result], removed: list[str]) -> str:
             "| 项目 | 结果 |",
             "| --- | --- |",
         ]
-        out += [f"| {k} | {v.replace(chr(10), '<br>')} |" for k, v in r.lines]
+        out += [f"| {k} | {_cell(v)} |" for k, v in r.lines]
         out.append("")
     if removed:
         out += ["### 下架", "", *[f"- {i}" for i in removed], ""]
@@ -299,15 +339,54 @@ def summary(results: list[Result], removed: list[str]) -> str:
     return "\n".join(out)
 
 
-def cmd_check(base: str, out: Path) -> int:
-    changed, removed = changed_ids(base)
+def cmd_check(base: str, head: str, catalog: Path | None, out: Path) -> int:
+    changed = sorted(
+        PurePosixPath(p).stem
+        for p in _diff(base, head, "--diff-filter=AMR", "--", "plugins/registry")
+    )
+    removed = sorted(
+        PurePosixPath(p).stem
+        for p in _diff(base, head, "--diff-filter=D", "--", "plugins/registry")
+    )
+    others = [p for p in _diff(base, head) if not p.startswith("plugins/registry/")]
+    published = load_catalog(catalog)
     with tempfile.TemporaryDirectory() as tmp:
-        results = [check_entry(i, base, out, Path(tmp) / i) for i in changed]
-    print(summary(results, removed))
+        results = [
+            check_entry(i, base, head, published, out, Path(tmp) / i) for i in changed
+        ]
+    print(summary(results, removed, others))
     return 0 if all(r.ok for r in results) else 1
 
 
-# --------------------------------------------------------------------- build
+# --------------------------------------------------------------------- publish
+
+
+def plan(published: dict[str, dict], rebuild: bool) -> dict[str, list[str]]:
+    """来源（repo、commit、path）未变的条目复用已发布的 zip，其余重新打包。"""
+    build: list[str] = []
+    reuse: list[str] = []
+    for entry in registry_entries():
+        prev = published.get(entry.id, {})
+        same = {
+            k: prev.get(k, "." if k == "path" else None) for k in entry.source
+        } == entry.source
+        (reuse if same and not rebuild else build).append(entry.id)
+    return {"build": build, "reuse": reuse}
+
+
+def build_one(plugin_id: str, out: Path) -> None:
+    """打包一个条目，并写出 source.json（repo、commit、path、readme）。"""
+    entry = Entry.parse(plugin_id, (REGISTRY / f"{plugin_id}.toml").read_text("utf-8"))
+    out.mkdir(parents=True, exist_ok=True)
+    with tempfile.TemporaryDirectory() as tmp:
+        src = fetch(entry, Path(tmp))
+        build_web(src)
+        manifest = verify(entry, src, None, None)
+        pack(src, out, manifest)
+        source = {**entry.source, "readme": readme(src)}
+    (out / "source.json").write_text(
+        json.dumps(source, ensure_ascii=False), encoding="utf-8"
+    )
 
 
 def _download(url: str) -> bytes:
@@ -315,61 +394,80 @@ def _download(url: str) -> bytes:
         return resp.read()
 
 
-def build(old_catalog: dict, out: Path, tmp: Path) -> dict[str, dict[str, str]]:
-    """重建全部条目的 zip，返回 sources（id → repo、commit、readme）。"""
-    out.mkdir(parents=True, exist_ok=True)
-    previous = {p["id"]: p for p in old_catalog.get("plugins", [])}
+def _zip_id(archive: Path) -> str:
+    with zipfile.ZipFile(archive) as zf:
+        return tomllib.loads(zf.read(MANIFEST_NAME).decode("utf-8"))["plugin"]["id"]
+
+
+def collect(published: dict[str, dict], rebuild: bool, dist: Path, stage: Path) -> None:
+    """签名前的汇总：只放进 registry 中的条目，每个 zip 都核对过来源。
+
+    ``dist/plugin-<id>/`` 是 build-one 各自的产物。复用的 zip 在这里下载，不经过
+    运行作者代码的 job。
+    """
+    stage.mkdir(parents=True, exist_ok=True)
+    todo = plan(published, rebuild)
+    entries = {e.id: e for e in registry_entries()}
     sources: dict[str, dict[str, str]] = {}
-    for path in sorted(REGISTRY.glob("*.toml")):
-        entry = Entry.parse(path.stem, path.read_text("utf-8"))
-        prev = previous.get(entry.id)
-        if prev and (prev.get("repo"), prev.get("commit")) == (
-            entry.repo,
-            entry.commit,
-        ):
-            # commit 未变：复用已发布的 zip，作者删库或 force-push 不影响发布
-            data = _download(f"{RELEASE_BASE}/{prev['asset']}")
-            if hashlib.sha256(data).hexdigest() != prev["sha256"]:
-                raise SystemExit(f"{prev['asset']} 与旧 catalog 的 sha256 不一致")
-            (out / prev["asset"]).write_bytes(data)
-            source = {"readme": prev.get("readme", "")}
-            print(f"✓ {entry.id} {prev['version']} 复用", file=sys.stderr)
-        else:
-            src = fetch(entry.repo, entry.commit, entry.path, tmp / entry.id)
-            build_web(src)
-            manifest = verify(entry, src, None, None)
-            pack(src, out, manifest)
-            source = {"readme": readme(src)}
-            print(f"✓ {entry.id} {manifest.version} 打包", file=sys.stderr)
-        sources[entry.id] = {"repo": entry.repo, "commit": entry.commit, **source}
-    return sources
-
-
-def cmd_build(old: Path | None, out: Path) -> int:
-    old_catalog = json.loads(old.read_text("utf-8")) if old and old.is_file() else {}
-    with tempfile.TemporaryDirectory() as tmp:
-        sources = build(old_catalog, out, Path(tmp))
-    (out / "sources.json").write_text(
+    for plugin_id in todo["reuse"]:
+        prev = published[plugin_id]
+        data = _download(f"{RELEASE_BASE}/{prev['asset']}")
+        if hashlib.sha256(data).hexdigest() != prev["sha256"]:
+            raise SystemExit(f"{prev['asset']} 与已发布 catalog 的 sha256 不一致")
+        (stage / prev["asset"]).write_bytes(data)
+        sources[plugin_id] = {
+            **entries[plugin_id].source,
+            "readme": prev.get("readme", ""),
+        }
+    for plugin_id in todo["build"]:
+        built = dist / f"plugin-{plugin_id}"
+        zips = sorted(built.glob("*.zip"))
+        if len(zips) != 1 or _zip_id(zips[0]) != plugin_id:
+            raise SystemExit(f"{plugin_id} 的构建产物不是恰好一个 id 相符的 zip")
+        source = json.loads((built / "source.json").read_text("utf-8"))
+        if {k: source.get(k) for k in entries[plugin_id].source} != entries[
+            plugin_id
+        ].source:
+            raise SystemExit(f"{plugin_id} 的构建产物来源与 registry 不一致")
+        shutil.copy2(zips[0], stage / zips[0].name)
+        sources[plugin_id] = {
+            **entries[plugin_id].source,
+            "readme": str(source.get("readme", ""))[:README_LIMIT],
+        }
+    (stage / "sources.json").write_text(
         json.dumps(sources, ensure_ascii=False, indent=2) + "\n", encoding="utf-8"
     )
-    return 0
 
 
 if __name__ == "__main__":
     parser = argparse.ArgumentParser(description=__doc__.splitlines()[0])
     sub = parser.add_subparsers(dest="cmd", required=True)
-    p_check = sub.add_parser("check", help="检查本 PR 改动的条目")
-    p_check.add_argument("--base", required=True)
-    p_check.add_argument("--out", required=True, type=Path)
-    p_build = sub.add_parser("build", help="重建全部条目的 zip 与 sources.json")
-    p_build.add_argument("--old", type=Path, help="当前 release 的 catalog.json")
-    p_build.add_argument("--out", required=True, type=Path)
+    p = sub.add_parser("check", help="检查 PR 改动的条目")
+    p.add_argument("--base", required=True)
+    p.add_argument("--head", required=True)
+    p.add_argument("--catalog", type=Path, help="已发布的 catalog.json")
+    p.add_argument("--out", required=True, type=Path)
+    for name in ("plan", "collect"):
+        p = sub.add_parser(name)
+        p.add_argument("--catalog", type=Path)
+        p.add_argument(
+            "--rebuild", action="store_true", help="忽略已发布的 zip，全部重新打包"
+        )
+        if name == "collect":
+            p.add_argument("--dist", required=True, type=Path)
+            p.add_argument("--out", required=True, type=Path)
+    p = sub.add_parser("build-one")
+    p.add_argument("id")
+    p.add_argument("--out", required=True, type=Path)
     args = parser.parse_args()
     try:
-        sys.exit(
-            cmd_check(args.base, args.out)
-            if args.cmd == "check"
-            else cmd_build(args.old, args.out)
-        )
+        if args.cmd == "check":
+            sys.exit(cmd_check(args.base, args.head, args.catalog, args.out))
+        elif args.cmd == "plan":
+            print(json.dumps(plan(load_catalog(args.catalog), args.rebuild)))
+        elif args.cmd == "build-one":
+            build_one(args.id, args.out)
+        else:
+            collect(load_catalog(args.catalog), args.rebuild, args.dist, args.out)
     except Reject as e:
         sys.exit(f"错误：{e}")