"""registry.yaml 读写。""" from __future__ import annotations import hashlib from pathlib import Path from typing import Any from skiff import yaml_io from skiff.paths import REGISTRY_FILE def load_registry(path: Path | None = None) -> dict[str, dict[str, Any]]: path = path or REGISTRY_FILE if not path.is_file(): return {} data = yaml_io.safe_load(path.read_text(encoding="utf-8")) or {} if not isinstance(data, dict): raise SystemExit(f"registry 格式错误: {path}") return {k: v for k, v in data.items() if isinstance(v, dict) and not k.startswith("#")} def save_registry(data: dict[str, dict[str, Any]], path: Path | None = None) -> None: path = path or REGISTRY_FILE path.parent.mkdir(parents=True, exist_ok=True) path.write_text(yaml_io.safe_dump(data, allow_unicode=True, sort_keys=False), encoding="utf-8") def external_repo_path(entry: dict[str, Any]) -> Path: """Return the shared checkout path for a repo/ref pair.""" from skiff.paths import EXTERNALS_DIR repo = str(entry.get("repo", "")) ref = str(entry.get("ref", "main")) digest = hashlib.sha256(f"{repo}\0{ref}".encode()).hexdigest()[:16] return EXTERNALS_DIR / "_repos" / digest def external_checkout_path(name: str, entry: dict[str, Any]) -> Path: """Prefer an existing pre-shared-cache checkout for compatibility.""" from skiff.paths import EXTERNALS_DIR legacy = EXTERNALS_DIR / name return legacy if legacy.is_dir() else external_repo_path(entry) def external_skill_path(name: str, entry: dict[str, Any] | None = None) -> Path: entry = entry or load_registry().get(name, {}) subpath = entry.get("path", ".") or "." checkout = external_checkout_path(name, entry).resolve() skill_path = (checkout / subpath).resolve() try: skill_path.relative_to(checkout) except ValueError as exc: raise SystemExit( f"registry 条目 {name!r} 的 path 超出外部仓库: {subpath!r}" ) from exc return skill_path