diff --git a/.dist-sync/sources.json b/.dist-sync/sources.json new file mode 100644 index 0000000..8821a63 --- /dev/null +++ b/.dist-sync/sources.json @@ -0,0 +1,9 @@ +{ + "sources": [ + { + "name": "think-twice", + "repository": "agent-skills/think-twice", + "asset": "think-twice.zip" + } + ] +} diff --git a/.dist-sync/versions.json b/.dist-sync/versions.json new file mode 100644 index 0000000..0967ef4 --- /dev/null +++ b/.dist-sync/versions.json @@ -0,0 +1 @@ +{} diff --git a/.gitea/workflows/sync-releases.yml b/.gitea/workflows/sync-releases.yml new file mode 100644 index 0000000..1e9d0ce --- /dev/null +++ b/.gitea/workflows/sync-releases.yml @@ -0,0 +1,92 @@ +name: 同步 skill Releases + +on: + schedule: + - cron: "17 * * * *" + workflow_dispatch: + +concurrency: + group: sync-skill-releases + cancel-in-progress: false + +permissions: + contents: write + +jobs: + sync: + runs-on: ubuntu-latest + steps: + - name: 检出分发仓 + uses: actions/checkout@11bd71901bbe5b1630ceea73d27597364c9af683 # v4.2.2 + with: + fetch-depth: 0 + persist-credentials: false + + - name: 检查同步凭据 + env: + SOURCE_REPOS_TOKEN: ${{ secrets.AGENT_SKILL_BOT }} + run: | + set -eu + test -n "$SOURCE_REPOS_TOKEN" || { + echo "缺少组织 Actions secret：AGENT_SKILL_BOT" + exit 1 + } + + - name: 同步 latest Releases + env: + GITEA_SERVER_URL: ${{ github.server_url }} + SOURCE_REPOS_TOKEN: ${{ secrets.AGENT_SKILL_BOT }} + run: python3 scripts/sync-releases.py + + - name: 校验分发仓 + run: | + set -eu + python3 scripts/validate-dist.py + NO_COLOR=1 npx --yes skills@1.5.22 add . --list | tee /tmp/skills-list.txt + python3 - <<'PY' + from pathlib import Path + + expected = { + "baimiao-table-ocr", + "disk-master", + "everything-file-search", + "teach-me", + "think-twice", + } + output = Path("/tmp/skills-list.txt").read_text(encoding="utf-8") + lines = {line.strip(" │└├─") for line in output.splitlines()} + missing = sorted(expected - lines) + if missing: + raise SystemExit(f"skills CLI 未发现：{', '.join(missing)}") + print(f"skills CLI 已发现全部 {len(expected)} 个 skill") + PY + + - name: 提交同步结果 + env: + GITEA_TOKEN: ${{ secrets.GITEA_TOKEN }} + GITEA_SERVER_URL: ${{ github.server_url }} + run: | + set -eu + if git diff --quiet -- .dist-sync skills; then + echo "latest Releases 没有产生内容变化" + exit 0 + fi + + git config user.name "Agent Skills Bot" + git config user.email "agent-skills-bot@im0o.top" + git add .dist-sync skills + git commit -m "同步 latest skill releases" + + auth=$(printf 'x-access-token:%s' "$GITEA_TOKEN" | base64 | tr -d '\n') + server_url=${GITEA_SERVER_URL%/} + git_auth() { + git -c "http.${server_url}/.extraheader=Authorization: Basic $auth" "$@" + } + + git_auth fetch origin main + test "$(git rev-parse HEAD^)" = "$(git rev-parse origin/main)" || { + echo "origin/main 已前进，本次不自动变基或强制推送" + exit 1 + } + + git_auth push origin HEAD:main diff --git a/README.md b/README.md index 978dd5c..ed11a8d 100644 --- a/README.md +++ b/README.md @@ -1,13 +1,14 @@ # agent-skills -一次装齐 im0o 的 agent skill。 +从一个入口按用途选择并安装 im0o 的 agent skill。 ```bash npx skills add ssh://gitea/agent-skills/skills.git ``` -会出现一个多选列表,按用途分成三组,勾选需要的即可。仓库是私有的,`skills add` -用系统 git 克隆,所以只要你的 SSH key 能访问就能装,不需要额外配置凭据。 +会出现一个按用途分组的多选列表。可以选择整组，也可以只勾选其中几个 skill。仓库是 +私有的，`skills add` 使用系统 Git 克隆；本机的 SSH key 能访问该仓库即可，不需要另外 +配置安装凭据。 ## 目录里有什么 @@ -43,5 +44,7 @@ npx skills add ssh://gitea/agent-skills/skills.git | `everything-file-search` | `agent-skills/everything-skill` | `everything-file-search/` | | `baimiao-table-ocr` | `agent-skills/img2excel-skill` | `skills/baimiao-table-ocr/` | -改动在源码仓做,这个仓只承载分发副本。源码仓打 `v` 开头的 tag 发布一个版本, -分发仓再把内容取回来。自动同步还没接上,目前仍是手工。 +改动在源码仓完成，分发仓只承载可安装副本。源码仓发布正式 `v*` tag 并生成 Release +ZIP 后，分发仓按 `.dist-sync/sources.json` 白名单同步 latest Release。当前只自动同步 +`think-twice`；其余四个 skill 具备标准 Release 后再逐项加入。白名单来源尚无 latest +Release 时会跳过并保留现有副本。 diff --git a/scripts/sync-releases.py b/scripts/sync-releases.py new file mode 100644 index 0000000..625522f --- /dev/null +++ b/scripts/sync-releases.py @@ -0,0 +1,440 @@ +#!/usr/bin/env python3 +"""Synchronize allowlisted skills from their latest Gitea Release assets.""" + +from __future__ import annotations + +import argparse +import json +import os +import re +import shutil +import stat +import sys +import tempfile +import urllib.error +import urllib.parse +import urllib.request +import zipfile +from pathlib import Path, PurePosixPath +from typing import Any + +REPO_ROOT = Path(__file__).resolve().parents[1] +DEFAULT_SOURCES = REPO_ROOT / ".dist-sync" / "sources.json" +DEFAULT_VERSIONS = REPO_ROOT / ".dist-sync" / "versions.json" +MAX_ASSET_BYTES = 250 * 1024 * 1024 +MAX_API_BYTES = 5 * 1024 * 1024 +MAX_ARCHIVE_FILES = 10_000 +MAX_ARCHIVE_DEPTH = 32 +MAX_ARCHIVE_PATH_LENGTH = 1024 +TAG_PATTERN = re.compile(r"v[0-9]+(?:\.[0-9]+)*") + + +class SyncError(RuntimeError): + """A source exists but cannot be synchronized safely.""" + + +def log(message: str) -> None: + print(message, flush=True) + + +def read_json(path: Path) -> Any: + try: + return json.loads(path.read_text(encoding="utf-8")) + except (OSError, json.JSONDecodeError) as error: + raise SyncError(f"无法读取 JSON：{path}: {error}") from error + + +def write_json_atomic(path: Path, value: Any) -> None: + path.parent.mkdir(parents=True, exist_ok=True) + encoded = (json.dumps(value, ensure_ascii=False, indent=2, sort_keys=True) + "\n").encode("utf-8") + temporary: Path | None = None + try: + with tempfile.NamedTemporaryFile(prefix=f".{path.name}.", dir=path.parent, delete=False) as output: + temporary = Path(output.name) + output.write(encoded) + output.flush() + os.fsync(output.fileno()) + os.replace(temporary, path) + except OSError as error: + if temporary is not None: + temporary.unlink(missing_ok=True) + raise SyncError(f"无法原子更新 JSON：{path}: {error}") from error + + +def validate_source(raw: Any) -> dict[str, str]: + if not isinstance(raw, dict): + raise SyncError("sources.json 的每个来源必须是对象") + + source: dict[str, str] = {} + for field in ("name", "repository", "asset"): + value = raw.get(field) + if not isinstance(value, str) or not value.strip(): + raise SyncError(f"来源缺少有效字段：{field}") + source[field] = value.strip() + + name = source["name"] + repository = source["repository"] + asset = source["asset"] + if not re.fullmatch(r"[a-z0-9]+(?:-[a-z0-9]+)*", name): + raise SyncError(f"无效 skill 名称：{name}") + owner, repo = repository.split("/", 1) if "/" in repository else ("", "") + repository_part = re.compile(r"[A-Za-z0-9_.-]+") + if not owner or not repo or not repository_part.fullmatch(owner) or not repository_part.fullmatch(repo): + raise SyncError(f"无效仓库名称：{repository}") + if Path(asset).name != asset: + raise SyncError(f"附件名称不能包含路径：{asset}") + return source + + +def canonical_origin(url: str) -> tuple[str, str, int | None]: + parsed = urllib.parse.urlsplit(url) + scheme = parsed.scheme.lower() + hostname = parsed.hostname.lower() if parsed.hostname else "" + try: + port = parsed.port + except ValueError as error: + raise SyncError(f"URL 端口无效：{url}") from error + if scheme not in {"http", "https"} or not hostname: + raise SyncError(f"URL 必须是有效的 HTTP(S) 地址：{url}") + effective_port = port if port is not None else (443 if scheme == "https" else 80) + return scheme, hostname, effective_port + + +class SameOriginRedirectHandler(urllib.request.HTTPRedirectHandler): + def __init__(self, allowed_origin: tuple[str, str, int | None]) -> None: + super().__init__() + self.allowed_origin = allowed_origin + + def redirect_request( + self, + request: urllib.request.Request, + file_pointer: Any, + code: int, + message: str, + headers: Any, + new_url: str, + ) -> urllib.request.Request | None: + if canonical_origin(new_url) != self.allowed_origin: + raise SyncError(f"拒绝跨源重定向：{request.full_url} -> {new_url}") + return super().redirect_request(request, file_pointer, code, message, headers, new_url) + + +def build_opener(server_url: str) -> tuple[urllib.request.OpenerDirector, tuple[str, str, int | None]]: + origin = canonical_origin(server_url) + return urllib.request.build_opener(SameOriginRedirectHandler(origin)), origin + + +def request_json(url: str, token: str, opener: urllib.request.OpenerDirector) -> tuple[int, Any | None]: + request = urllib.request.Request( + url, + headers={ + "Accept": "application/json", + "Authorization": f"token {token}", + "User-Agent": "agent-skills-dist-sync", + }, + ) + try: + with opener.open(request, timeout=30) as response: + body = response.read(MAX_API_BYTES + 1) + if len(body) > MAX_API_BYTES: + raise SyncError(f"Gitea API 响应超过 {MAX_API_BYTES} bytes：{url}") + try: + return response.status, json.loads(body) + except (UnicodeDecodeError, json.JSONDecodeError) as error: + raise SyncError(f"Gitea API 返回无效 JSON：{url}: {error}") from error + except urllib.error.HTTPError as error: + if error.code == 404: + return error.code, None + detail = error.read(2048).decode("utf-8", errors="replace").strip() + raise SyncError(f"Gitea API 返回 HTTP {error.code}: {detail or url}") from error + except SyncError: + raise + except OSError as error: + raise SyncError(f"请求 Gitea API 失败：{url}: {error}") from error + + +def download( + url: str, + token: str, + destination: Path, + opener: urllib.request.OpenerDirector, + allowed_origin: tuple[str, str, int | None], +) -> None: + if canonical_origin(url) != allowed_origin: + raise SyncError(f"附件下载地址与 Gitea 不同源，拒绝发送 token：{url}") + request = urllib.request.Request( + url, + headers={ + "Accept": "application/octet-stream", + "Authorization": f"token {token}", + "User-Agent": "agent-skills-dist-sync", + }, + ) + try: + with opener.open(request, timeout=120) as response, destination.open("wb") as output: + declared_size = response.headers.get("Content-Length") + if declared_size: + try: + if int(declared_size) > MAX_ASSET_BYTES: + raise SyncError(f"附件超过 {MAX_ASSET_BYTES} bytes：{url}") + except ValueError as error: + raise SyncError(f"附件返回无效 Content-Length：{declared_size!r}") from error + + copied = 0 + while chunk := response.read(1024 * 1024): + copied += len(chunk) + if copied > MAX_ASSET_BYTES: + raise SyncError(f"附件超过 {MAX_ASSET_BYTES} bytes：{url}") + output.write(chunk) + except urllib.error.HTTPError as error: + detail = error.read(2048).decode("utf-8", errors="replace").strip() + raise SyncError(f"下载附件返回 HTTP {error.code}: {detail or url}") from error + except SyncError: + raise + except OSError as error: + raise SyncError(f"下载附件失败：{url}: {error}") from error + + +def safe_extract(archive_path: Path, destination: Path) -> None: + try: + with zipfile.ZipFile(archive_path) as archive: + files = 0 + total_size = 0 + for info in archive.infolist(): + normalized = info.filename.replace("\\", "/") + member = PurePosixPath(normalized) + if member.is_absolute() or ".." in member.parts: + raise SyncError(f"ZIP 包含不安全路径：{info.filename}") + if not member.parts: + continue + if len(normalized) > MAX_ARCHIVE_PATH_LENGTH or len(member.parts) > MAX_ARCHIVE_DEPTH: + raise SyncError(f"ZIP 路径过长或目录过深：{info.filename}") + + mode = info.external_attr >> 16 + if stat.S_ISLNK(mode): + raise SyncError(f"ZIP 不允许符号链接：{info.filename}") + + target = destination.joinpath(*member.parts) + if not info.is_dir(): + files += 1 + if files > MAX_ARCHIVE_FILES: + raise SyncError(f"ZIP 文件数量超过 {MAX_ARCHIVE_FILES}") + total_size += info.file_size + if total_size > MAX_ASSET_BYTES: + raise SyncError("ZIP 解压后内容过大") + target.parent.mkdir(parents=True, exist_ok=True) + with archive.open(info) as source, target.open("wb") as output: + shutil.copyfileobj(source, output) + if mode & 0o111: + target.chmod(target.stat().st_mode | 0o111) + else: + target.mkdir(parents=True, exist_ok=True) + + if files == 0: + raise SyncError("ZIP 中没有文件") + except zipfile.BadZipFile as error: + raise SyncError(f"附件不是有效 ZIP：{archive_path}") from error + + +def frontmatter_name(skill_file: Path) -> str: + try: + lines = skill_file.read_text(encoding="utf-8-sig").splitlines() + except (OSError, UnicodeDecodeError) as error: + raise SyncError(f"无法读取 {skill_file}: {error}") from error + + if not lines or lines[0].strip() != "---": + raise SyncError(f"SKILL.md 缺少 YAML frontmatter：{skill_file}") + + for line in lines[1:]: + if line.strip() == "---": + break + if line.startswith("name:"): + return line.split(":", 1)[1].strip().strip("'\"") + raise SyncError(f"SKILL.md frontmatter 缺少 name：{skill_file}") + + +def stage_source( + source: dict[str, str], + *, + server_url: str, + token: str, + current_versions: dict[str, str], + next_versions: dict[str, str], + temp_root: Path, + opener: urllib.request.OpenerDirector, + allowed_origin: tuple[str, str, int | None], +) -> tuple[str, Path | None]: + name = source["name"] + repository = source["repository"] + encoded_repository = "/".join(urllib.parse.quote(part, safe="") for part in repository.split("/")) + latest_url = f"{server_url}/api/v1/repos/{encoded_repository}/releases/latest" + + status, release = request_json(latest_url, token, opener) + if status == 404: + log(f"跳过 {name}：仓库不可见或尚无 latest Release") + return "skipped", None + if not isinstance(release, dict): + raise SyncError(f"{repository} 的 latest Release 响应无效") + + tag = release.get("tag_name") + assets = release.get("assets") + if not isinstance(tag, str) or not TAG_PATTERN.fullmatch(tag): + raise SyncError(f"{repository} 的 latest Release tag 不符合 v* 正式版本约定：{tag!r}") + if not isinstance(assets, list): + raise SyncError(f"{repository} 的 latest Release 缺少附件列表") + + matches = [asset for asset in assets if isinstance(asset, dict) and asset.get("name") == source["asset"]] + if len(matches) != 1: + raise SyncError(f"{repository}@{tag} 应恰好包含一个附件 {source['asset']}，实际为 {len(matches)} 个") + + target = REPO_ROOT / "skills" / name + if current_versions.get(name) == tag and target.is_dir(): + log(f"跳过 {name}：已经同步 {tag}") + return "unchanged", None + + download_url = matches[0].get("browser_download_url") + if not isinstance(download_url, str) or not download_url: + raise SyncError(f"{repository}@{tag} 的附件缺少 browser_download_url") + + work_dir = temp_root / name + extract_dir = work_dir / "content" + archive_path = work_dir / source["asset"] + extract_dir.mkdir(parents=True) + download(download_url, token, archive_path, opener, allowed_origin) + safe_extract(archive_path, extract_dir) + + skill_file = extract_dir / "SKILL.md" + if not skill_file.is_file(): + raise SyncError(f"{repository}@{tag} 的 ZIP 根目录缺少 SKILL.md") + actual_name = frontmatter_name(skill_file) + if actual_name != name: + raise SyncError(f"{repository}@{tag} 的 skill name 为 {actual_name!r}，预期 {name!r}") + + next_versions[name] = tag + log(f"已准备 {name}：{current_versions.get(name, '未记录')} -> {tag}") + return "updated", extract_dir + + +def apply_staged_updates( + staged: list[tuple[str, Path]], + current_versions: dict[str, str], + next_versions: dict[str, str], + versions_path: Path, +) -> None: + if not staged: + return + + backup_root = Path(tempfile.mkdtemp(prefix=".dist-sync-backup-", dir=REPO_ROOT)) + applied: list[tuple[Path, Path | None]] = [] + try: + for name, staged_dir in staged: + destination = REPO_ROOT / "skills" / name + replacement = destination.with_name(f".{name}.dist-sync-new") + if replacement.exists(): + shutil.rmtree(replacement) + shutil.copytree(staged_dir, replacement) + + backup: Path | None = None + if destination.exists(): + backup = backup_root / name + destination.rename(backup) + try: + replacement.rename(destination) + except Exception: + if backup is not None and backup.exists() and not destination.exists(): + backup.rename(destination) + raise + applied.append((destination, backup)) + + write_json_atomic(versions_path, next_versions) + except Exception as error: + for destination, backup in reversed(applied): + if destination.exists(): + shutil.rmtree(destination) + if backup is not None and backup.exists(): + backup.rename(destination) + try: + write_json_atomic(versions_path, current_versions) + except SyncError as rollback_error: + raise SyncError( + f"应用同步结果失败；skill 目录已回滚，但 versions.json 回滚失败：{rollback_error}" + ) from error + if isinstance(error, SyncError): + raise + raise SyncError(f"应用同步结果失败，已回滚 skill 目录：{error}") from error + finally: + shutil.rmtree(backup_root, ignore_errors=True) + for name, _ in staged: + replacement = REPO_ROOT / "skills" / f".{name}.dist-sync-new" + if replacement.exists(): + shutil.rmtree(replacement, ignore_errors=True) + + +def main() -> int: + parser = argparse.ArgumentParser() + parser.add_argument("--sources", type=Path, default=DEFAULT_SOURCES) + parser.add_argument("--versions", type=Path, default=DEFAULT_VERSIONS) + parser.add_argument("--server-url", default=os.environ.get("GITEA_SERVER_URL", "")) + parser.add_argument("--token", default=os.environ.get("SOURCE_REPOS_TOKEN", "")) + args = parser.parse_args() + + server_url = args.server_url.rstrip("/") + token = args.token.strip() + if not server_url: + raise SyncError("缺少 GITEA_SERVER_URL 或 --server-url") + if not token: + raise SyncError("缺少 SOURCE_REPOS_TOKEN 或 --token") + opener, allowed_origin = build_opener(server_url) + + config = read_json(args.sources) + raw_sources = config.get("sources") if isinstance(config, dict) else None + if not isinstance(raw_sources, list) or not raw_sources: + raise SyncError("sources.json 必须包含非空 sources 数组") + sources = [validate_source(raw) for raw in raw_sources] + names = [source["name"] for source in sources] + if len(names) != len(set(names)): + raise SyncError("sources.json 包含重复的 skill name") + + raw_versions = read_json(args.versions) if args.versions.exists() else {} + if not isinstance(raw_versions, dict) or not all( + isinstance(name, str) and isinstance(tag, str) and TAG_PATTERN.fullmatch(tag) + for name, tag in raw_versions.items() + ): + raise SyncError("versions.json 必须是 skill name 到正式 v* Release tag 的字符串映射") + current_versions: dict[str, str] = dict(raw_versions) + next_versions = dict(current_versions) + + counts = {"updated": 0, "unchanged": 0, "skipped": 0} + staged: list[tuple[str, Path]] = [] + with tempfile.TemporaryDirectory(prefix="agent-skills-dist-sync-") as temp: + for source in sources: + result, staged_dir = stage_source( + source, + server_url=server_url, + token=token, + current_versions=current_versions, + next_versions=next_versions, + temp_root=Path(temp), + opener=opener, + allowed_origin=allowed_origin, + ) + counts[result] += 1 + if staged_dir is not None: + staged.append((source["name"], staged_dir)) + + apply_staged_updates(staged, current_versions, next_versions, args.versions) + + log( + "同步完成：" + f"更新 {counts['updated']}，无变化 {counts['unchanged']}，" + f"无 latest Release {counts['skipped']}" + ) + return 0 + + +if __name__ == "__main__": + try: + raise SystemExit(main()) + except SyncError as error: + print(f"同步失败：{error}", file=sys.stderr) + raise SystemExit(1) diff --git a/scripts/validate-dist.py b/scripts/validate-dist.py new file mode 100644 index 0000000..0d94484 --- /dev/null +++ b/scripts/validate-dist.py @@ -0,0 +1,133 @@ +#!/usr/bin/env python3 +"""Validate the distribution repository and its plugin manifest.""" + +from __future__ import annotations + +import json +import sys +from pathlib import Path +from typing import Any + +REPO_ROOT = Path(__file__).resolve().parents[1] +SOURCES_PATH = REPO_ROOT / ".dist-sync" / "sources.json" +VERSIONS_PATH = REPO_ROOT / ".dist-sync" / "versions.json" +MARKETPLACE_PATH = REPO_ROOT / ".claude-plugin" / "marketplace.json" + + +class ValidationError(RuntimeError): + pass + + +def read_json(path: Path) -> Any: + try: + return json.loads(path.read_text(encoding="utf-8")) + except (OSError, json.JSONDecodeError) as error: + raise ValidationError(f"无法读取 JSON：{path}: {error}") from error + + +def frontmatter_name(skill_file: Path) -> str: + try: + lines = skill_file.read_text(encoding="utf-8-sig").splitlines() + except (OSError, UnicodeDecodeError) as error: + raise ValidationError(f"无法读取 {skill_file}: {error}") from error + if not lines or lines[0].strip() != "---": + raise ValidationError(f"SKILL.md 缺少 YAML frontmatter：{skill_file}") + for line in lines[1:]: + if line.strip() == "---": + break + if line.startswith("name:"): + return line.split(":", 1)[1].strip().strip("'\"") + raise ValidationError(f"SKILL.md frontmatter 缺少 name：{skill_file}") + + +def main() -> int: + sources_doc = read_json(SOURCES_PATH) + sources = sources_doc.get("sources") if isinstance(sources_doc, dict) else None + if not isinstance(sources, list) or not sources: + raise ValidationError("sources.json 必须包含非空 sources 数组") + + allowlisted: set[str] = set() + for source in sources: + if not isinstance(source, dict) or not all( + isinstance(source.get(field), str) and source[field] for field in ("name", "repository", "asset") + ): + raise ValidationError("sources.json 包含无效来源") + name = source["name"] + if name in allowlisted: + raise ValidationError(f"sources.json 包含重复名称：{name}") + allowlisted.add(name) + + versions = read_json(VERSIONS_PATH) + if not isinstance(versions, dict) or not all( + isinstance(name, str) and isinstance(tag, str) and tag for name, tag in versions.items() + ): + raise ValidationError("versions.json 必须是非空 tag 字符串映射") + unknown_versions = set(versions) - allowlisted + if unknown_versions: + raise ValidationError(f"versions.json 包含未列入同步白名单的 skill：{sorted(unknown_versions)}") + + marketplace = read_json(MARKETPLACE_PATH) + plugins = marketplace.get("plugins") if isinstance(marketplace, dict) else None + if not isinstance(plugins, list) or not plugins: + raise ValidationError("marketplace.json 必须包含非空 plugins 数组") + + manifest_names: list[str] = [] + manifest_paths: set[Path] = set() + for plugin in plugins: + skills = plugin.get("skills") if isinstance(plugin, dict) else None + if not isinstance(skills, list) or not skills: + raise ValidationError("marketplace.json 的每个 plugin 都必须列出 skills") + for raw_path in skills: + if not isinstance(raw_path, str) or not raw_path.startswith("./skills/"): + raise ValidationError(f"无效 marketplace skill 路径：{raw_path!r}") + relative = Path(raw_path[2:]) + skill_dir = (REPO_ROOT / relative).resolve() + skills_root = (REPO_ROOT / "skills").resolve() + if skill_dir.parent != skills_root: + raise ValidationError(f"skill 必须直接位于 skills/：{raw_path}") + if skill_dir in manifest_paths: + raise ValidationError(f"marketplace.json 重复列出路径：{raw_path}") + manifest_paths.add(skill_dir) + + skill_file = skill_dir / "SKILL.md" + if not skill_file.is_file(): + raise ValidationError(f"marketplace 路径缺少 SKILL.md：{raw_path}") + actual_name = frontmatter_name(skill_file) + if actual_name != skill_dir.name: + raise ValidationError( + f"{raw_path}/SKILL.md 的 name 为 {actual_name!r}，目录名为 {skill_dir.name!r}" + ) + manifest_names.append(actual_name) + + if len(manifest_names) != len(set(manifest_names)): + raise ValidationError("marketplace.json 中存在重复的 skill name") + + disk_dirs = { + path.resolve() + for path in (REPO_ROOT / "skills").iterdir() + if path.is_dir() and not path.name.startswith(".") + } + unlisted = disk_dirs - manifest_paths + missing = manifest_paths - disk_dirs + if unlisted: + raise ValidationError(f"skills/ 中存在未列入 marketplace 的目录：{sorted(path.name for path in unlisted)}") + if missing: + raise ValidationError(f"marketplace 指向不存在的目录：{sorted(path.name for path in missing)}") + + for skill_dir in sorted(manifest_paths): + for path in skill_dir.rglob("*"): + if not path.is_file() or path.stat().st_size > 1024: + continue + if path.read_bytes().startswith(b"version https://git-lfs.github.com/spec/v1"): + raise ValidationError(f"发现 Git LFS pointer，分发副本缺少真实文件：{path.relative_to(REPO_ROOT)}") + + print(f"分发仓校验通过：{len(manifest_names)} 个 skill，{len(allowlisted)} 个 Release 同步来源") + return 0 + + +if __name__ == "__main__": + try: + raise SystemExit(main()) + except ValidationError as error: + print(f"校验失败：{error}", file=sys.stderr) + raise SystemExit(1)