119 lines
3.9 KiB
Python
119 lines
3.9 KiB
Python
"""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 registry_repo(entry: dict[str, Any]) -> str:
|
|
"""Expand a local home-relative repo while leaving remote URLs unchanged."""
|
|
repo = str(entry.get("repo", ""))
|
|
return str(Path(repo).expanduser()) if repo.startswith("~") else repo
|
|
|
|
|
|
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:
|
|
"""Use a local repo directly, otherwise return its external checkout."""
|
|
from skiff.paths import EXTERNALS_DIR
|
|
|
|
configured_repo = str(entry.get("repo", ""))
|
|
local_repo = Path(registry_repo(entry))
|
|
if configured_repo.startswith("~") and local_repo.is_dir():
|
|
return local_repo.resolve()
|
|
|
|
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
|
|
|
|
|
|
def discover_external_skills(
|
|
name: str,
|
|
entry: dict[str, Any] | None = None,
|
|
) -> dict[str, Path]:
|
|
"""Discover a single registry skill or a collection of sibling skills."""
|
|
entry = entry or load_registry().get(name, {})
|
|
root = external_skill_path(name, entry)
|
|
if (root / "SKILL.md").is_file():
|
|
return {name: root}
|
|
if not root.is_dir():
|
|
return {}
|
|
resolved_root = root.resolve()
|
|
skills: dict[str, Path] = {}
|
|
for item in sorted(root.iterdir()):
|
|
skill_md = item / "SKILL.md"
|
|
if (
|
|
item.name.startswith("_")
|
|
or item.is_symlink()
|
|
or not item.is_dir()
|
|
or skill_md.is_symlink()
|
|
or not skill_md.is_file()
|
|
):
|
|
continue
|
|
resolved_item = item.resolve()
|
|
resolved_skill_md = skill_md.resolve()
|
|
try:
|
|
resolved_item.relative_to(resolved_root)
|
|
resolved_skill_md.relative_to(resolved_item)
|
|
except ValueError:
|
|
continue
|
|
skills[item.name] = item
|
|
return skills
|
|
|
|
|
|
def external_collection_skill_path(
|
|
collection: str,
|
|
skill_name: str,
|
|
entry: dict[str, Any] | None = None,
|
|
) -> Path:
|
|
skills = discover_external_skills(collection, entry)
|
|
if skill_name not in skills:
|
|
available = ", ".join(skills) or "(无)"
|
|
raise SystemExit(
|
|
f"registry collection {collection!r} 中找不到 skill {skill_name!r}。"
|
|
f"可用: {available}"
|
|
)
|
|
return skills[skill_name]
|