增加 Release 自动同步 workflow
This commit is contained in:
@@ -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)
|
||||
@@ -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/<name>：{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)
|
||||
Reference in New Issue
Block a user