feat: add skiff

2026-07-04 10:50:03 +08:00
parent 8b5ac01a9d
commit c2acbc3809
16 changed files with 1501 additions and 73 deletions
@@ -0,0 +1,429 @@
+"""skiff CLI 入口。"""
+
+from __future__ import annotations
+
+import argparse
+import re
+import subprocess
+import sys
+from pathlib import Path
+from typing import Any
+
+from skiff import __version__
+from skiff.paths import (
+    ALL_TARGETS,
+    EXTERNALS_DIR,
+    REGISTRY_FILE,
+    SKILLS_DIR,
+    SKILLS_HOME,
+    TEMPLATE_DIR,
+    agent_skill_dir,
+    ensure_skills_home,
+    resolve_targets,
+)
+from skiff.project import (
+    add_skill_to_manifest,
+    iter_manifest_skills,
+    load_manifest,
+    remove_skill_from_manifest,
+    resolve_manifest_skill,
+    save_manifest,
+)
+from skiff.registry import external_skill_path, load_registry, save_registry
+from skiff.skills import list_owned_skills, owned_skill_path, resolve_skill_source, validate_skill_name
+from skiff.symlinks import check_link, copy_template, create_link, find_repo_root, remove_link
+
+
+def _print(msg: str = "") -> None:
+    print(msg, file=sys.stdout)
+
+
+def _err(msg: str) -> None:
+    print(msg, file=sys.stderr)
+
+
+def cmd_setup(args: argparse.Namespace) -> None:
+    repo = Path(args.path).resolve()
+    if not (repo / "skills").is_dir():
+        raise SystemExit(f"不是有效的 skills 仓库（缺少 skills/）: {repo}")
+
+    # 开发场景：仓库本身就在 ~/.skills
+    if repo == SKILLS_HOME.resolve() and SKILLS_HOME.is_dir() and not SKILLS_HOME.is_symlink():
+        _print(f"skills 仓库已在 ~/.skills: {repo}")
+        return
+
+    if SKILLS_HOME.is_symlink():
+        current = SKILLS_HOME.resolve()
+        if current == repo:
+            _print(f"已关联: ~/.skills -> {repo}")
+            return
+        SKILLS_HOME.unlink()
+    elif SKILLS_HOME.exists():
+        raise SystemExit(f"~/.skills 已存在且不是软链: {SKILLS_HOME}")
+
+    SKILLS_HOME.parent.mkdir(parents=True, exist_ok=True)
+    SKILLS_HOME.symlink_to(repo)
+    _print(f"已关联: ~/.skills -> {repo}")
+
+
+def cmd_list(args: argparse.Namespace) -> None:
+    ensure_skills_home()
+    owned = list_owned_skills()
+    registry = load_registry()
+
+    _print("自研 (owned):")
+    for name in owned:
+        _print(f"  {name}")
+
+    _print("\n外部 (registry):")
+    if not registry:
+        _print("  (无)")
+    else:
+        for name, entry in registry.items():
+            repo = entry.get("repo", "?")
+            _print(f"  {name}  ({repo})")
+
+
+def _installed_links(name: str, targets: list[str], project_root: Path | None = None) -> list[tuple[str, Path, Path]]:
+    skill_path, _ = resolve_skill_source(name)
+    rows: list[tuple[str, Path, Path]] = []
+    for target in targets:
+        link = agent_skill_dir(target, project_root=project_root) / name
+        rows.append((target, link, skill_path))
+    return rows
+
+
+def cmd_status(args: argparse.Namespace) -> None:
+    ensure_skills_home()
+    targets = resolve_targets(args.target)
+    owned = list_owned_skills()
+    registry = load_registry()
+    all_names = owned + [n for n in registry if n not in owned]
+
+    _print(f"skills 仓库: {SKILLS_HOME.resolve()}")
+    _print(f"targets: {', '.join(targets)}\n")
+
+    for name in all_names:
+        kind = "owned" if name in owned else "external"
+        _print(f"[{kind}] {name}")
+        try:
+            rows = _installed_links(name, targets)
+        except SystemExit:
+            _print("  (未 fetch)")
+            continue
+        for target, link, expected in rows:
+            status = check_link(link, expected)
+            mark = "✓" if status.ok else "✗"
+            detail = "" if status.ok else f" — {status.issue}"
+            _print(f"  {mark} {target}: {link}{detail}")
+        _print("")
+
+
+def _install_skill(name: str, targets: list[str], project_root: Path | None = None) -> None:
+    skill_path, kind = resolve_skill_source(name)
+    for target in targets:
+        link = agent_skill_dir(target, project_root=project_root) / name
+        create_link(link, skill_path)
+        scope = "项目" if project_root else "全局"
+        _print(f"已安装 ({scope}/{target}): {name} -> {skill_path}")
+
+
+def cmd_install(args: argparse.Namespace) -> None:
+    ensure_skills_home()
+    validate_skill_name(args.name)
+    targets = resolve_targets(args.target)
+    _install_skill(args.name, targets)
+
+
+def cmd_uninstall(args: argparse.Namespace) -> None:
+    ensure_skills_home()
+    targets = resolve_targets(args.target)
+    for target in targets:
+        link = agent_skill_dir(target) / args.name
+        if remove_link(link):
+            _print(f"已移除 ({target}): {link}")
+
+
+def cmd_add(args: argparse.Namespace) -> None:
+    ensure_skills_home()
+    validate_skill_name(args.name)
+    registry = load_registry()
+    if args.name in registry:
+        raise SystemExit(f"registry 中已存在: {args.name}")
+
+    registry[args.name] = {
+        "repo": args.repo,
+        "ref": args.ref,
+        "path": args.path,
+    }
+    save_registry(registry)
+    _print(f"已添加 registry 条目: {args.name}")
+
+
+def cmd_fetch(args: argparse.Namespace) -> None:
+    ensure_skills_home()
+    registry = load_registry()
+    if args.name not in registry:
+        raise SystemExit(f"registry 中不存在: {args.name}")
+
+    entry = registry[args.name]
+    repo = entry["repo"]
+    ref = entry.get("ref", "main")
+    dest = EXTERNALS_DIR / args.name
+
+    EXTERNALS_DIR.mkdir(parents=True, exist_ok=True)
+
+    if dest.exists():
+        _print(f"更新: {dest}")
+        subprocess.run(["git", "-C", str(dest), "fetch", "--all", "--tags"], check=True)
+        subprocess.run(["git", "-C", str(dest), "checkout", ref], check=True)
+        subprocess.run(["git", "-C", str(dest), "pull", "--ff-only"], check=True)
+    else:
+        _print(f"克隆: {repo} -> {dest}")
+        subprocess.run(
+            ["git", "clone", "--branch", ref, "--", repo, str(dest)],
+            check=True,
+        )
+
+
+def cmd_install_external(args: argparse.Namespace) -> None:
+    ensure_skills_home()
+    registry = load_registry()
+    if args.name not in registry:
+        raise SystemExit(f"registry 中不存在: {args.name}")
+
+    path = external_skill_path(args.name, registry[args.name])
+    if not path.exists():
+        raise SystemExit(f"请先 fetch: skiff fetch {args.name}")
+
+    targets = resolve_targets(args.target)
+    for target in targets:
+        link = agent_skill_dir(target) / args.name
+        create_link(link, path)
+        _print(f"已安装外部 ({target}): {args.name} -> {path}")
+
+
+def _project_root(args: argparse.Namespace) -> Path:
+    if args.project:
+        root = Path(args.project).resolve()
+    else:
+        root = find_repo_root() or Path.cwd()
+    return root
+
+
+def cmd_enable(args: argparse.Namespace) -> None:
+    ensure_skills_home()
+    validate_skill_name(args.name)
+    root = _project_root(args)
+    manifest_path = root / ".skills.yaml"
+
+    # 判断来源
+    registry = load_registry()
+    if args.name in registry:
+        source = "registry"
+        extra = {"source": "registry", "ref": registry[args.name].get("ref", "main")}
+        add_skill_to_manifest(manifest_path, args.name, source="registry", extra=extra)
+    else:
+        owned_skill_path(args.name)  # 验证存在
+        add_skill_to_manifest(manifest_path, args.name, source="owned")
+
+    targets = resolve_targets(args.target)
+    _, data = load_manifest(manifest_path)
+    manifest_targets = data.get("targets")
+    if manifest_targets:
+        targets = [t for t in targets if t in manifest_targets]
+
+    _install_skill(args.name, targets, project_root=root)
+    _print(f"已启用项目 skill: {args.name} @ {root}")
+
+
+def cmd_disable(args: argparse.Namespace) -> None:
+    root = _project_root(args)
+    manifest_path = root / ".skills.yaml"
+    if not remove_skill_from_manifest(manifest_path, args.name):
+        _print(f"manifest 中不存在: {args.name}")
+        return
+
+    targets = resolve_targets(args.target)
+    for target in targets:
+        link = agent_skill_dir(target, project_root=root) / args.name
+        if remove_link(link):
+            _print(f"已禁用 ({target}): {link}")
+
+
+def cmd_sync(args: argparse.Namespace) -> None:
+    ensure_skills_home()
+    root = _project_root(args)
+    manifest_path = root / ".skills.yaml"
+    if not manifest_path.is_file():
+        raise SystemExit(f"未找到 {manifest_path}")
+
+    _, data = load_manifest(manifest_path)
+    targets = resolve_targets(args.target)
+    manifest_targets = data.get("targets")
+    if manifest_targets:
+        targets = [t for t in targets if t in manifest_targets]
+
+    for entry in iter_manifest_skills(data):
+        name = entry["name"]
+        skill_path, _ = resolve_manifest_skill(entry)
+        for target in targets:
+            link = agent_skill_dir(target, project_root=root) / name
+            create_link(link, skill_path)
+            _print(f"已同步: {link} -> {skill_path}")
+
+
+def cmd_create(args: argparse.Namespace) -> None:
+    ensure_skills_home()
+    validate_skill_name(args.name)
+    if not TEMPLATE_DIR.is_dir():
+        raise SystemExit(f"模板目录不存在: {TEMPLATE_DIR}")
+
+    dst = SKILLS_DIR / args.name
+    copy_template(TEMPLATE_DIR, dst)
+
+    skill_md = dst / "SKILL.md"
+    content = skill_md.read_text(encoding="utf-8")
+    content = re.sub(
+        r"(^name:\s*)skill-name\s*$",
+        rf"\g<1>{args.name}",
+        content,
+        count=1,
+        flags=re.MULTILINE,
+    )
+    skill_md.write_text(content, encoding="utf-8")
+    _print(f"已创建 skill: {dst}")
+    _print(f"下一步: 编辑 {skill_md}，然后 skiff install {args.name}")
+
+
+def cmd_doctor(args: argparse.Namespace) -> None:
+    ensure_skills_home()
+    targets = resolve_targets(args.target)
+    issues = 0
+
+    _print(f"skills 仓库: {SKILLS_HOME.resolve()}")
+    if SKILLS_HOME.is_symlink():
+        if not SKILLS_HOME.resolve().is_dir():
+            _err(f"✗ ~/.skills 指向无效路径: {SKILLS_HOME.resolve()}")
+            issues += 1
+        else:
+            _print("✓ ~/.skills 软链正常")
+    elif SKILLS_HOME.is_dir() and (SKILLS_HOME / "skills").is_dir():
+        _print("✓ ~/.skills 为本地仓库目录")
+    else:
+        _err("✗ ~/.skills 未正确配置")
+        issues += 1
+
+    for name in list_owned_skills():
+        for target, link, expected in _installed_links(name, targets):
+            status = check_link(link, expected)
+            if status.ok:
+                continue
+            _err(f"✗ [{name}/{target}] {status.issue}: {link}")
+            issues += 1
+            if args.fix:
+                try:
+                    create_link(link, expected)
+                    _print(f"  已修复: {link}")
+                except Exception as exc:  # noqa: BLE001
+                    _err(f"  修复失败: {exc}")
+
+    registry = load_registry()
+    for name in registry:
+        ext = external_skill_path(name, registry[name])
+        if not ext.exists():
+            _err(f"✗ 外部 skill 未 fetch: {name}")
+            issues += 1
+
+    if issues == 0:
+        _print("\n全部正常")
+    else:
+        _print(f"\n发现 {issues} 个问题")
+        if not args.fix:
+            _print("提示: 使用 skiff doctor --fix 尝试自动修复软链")
+        sys.exit(1)
+
+
+def build_parser() -> argparse.ArgumentParser:
+    parser = argparse.ArgumentParser(
+        prog="skiff",
+        description="Agent Skills 安装与管理 CLI",
+    )
+    parser.add_argument("--version", action="version", version=f"skiff {__version__}")
+
+    sub = parser.add_subparsers(dest="command", required=True)
+
+    p_setup = sub.add_parser("setup", help="关联 ~/.skills 到本仓库")
+    p_setup.add_argument("path", help="skills 仓库路径")
+    p_setup.set_defaults(func=cmd_setup)
+
+    p_list = sub.add_parser("list", help="列出所有 skill")
+    p_list.set_defaults(func=cmd_list)
+
+    p_status = sub.add_parser("status", help="安装状态总览")
+    p_status.add_argument("--target", choices=[*ALL_TARGETS, "all"], default="all")
+    p_status.set_defaults(func=cmd_status)
+
+    p_install = sub.add_parser("install", help="全局安装 skill（软链）")
+    p_install.add_argument("name", help="skill 名称")
+    p_install.add_argument("--target", choices=[*ALL_TARGETS, "all"], default="all")
+    p_install.set_defaults(func=cmd_install)
+
+    p_uninstall = sub.add_parser("uninstall", help="移除全局软链")
+    p_uninstall.add_argument("name", help="skill 名称")
+    p_uninstall.add_argument("--target", choices=[*ALL_TARGETS, "all"], default="all")
+    p_uninstall.set_defaults(func=cmd_uninstall)
+
+    p_add = sub.add_parser("add", help="添加外部 skill 到 registry")
+    p_add.add_argument("name", help="registry 名称")
+    p_add.add_argument("repo", help="Git 仓库 URL")
+    p_add.add_argument("--ref", default="main", help="分支或 tag（默认 main）")
+    p_add.add_argument("--path", default=".", help="仓库内子路径（默认 .）")
+    p_add.set_defaults(func=cmd_add)
+
+    p_fetch = sub.add_parser("fetch", help="拉取外部 Git skill")
+    p_fetch.add_argument("name", help="registry 名称")
+    p_fetch.set_defaults(func=cmd_fetch)
+
+    p_inst_ext = sub.add_parser("install-external", help="安装外部 skill 到 Agent 目录")
+    p_inst_ext.add_argument("name", help="registry 名称")
+    p_inst_ext.add_argument("--target", choices=[*ALL_TARGETS, "all"], default="all")
+    p_inst_ext.set_defaults(func=cmd_install_external)
+
+    p_enable = sub.add_parser("enable", help="项目级启用 skill")
+    p_enable.add_argument("name", help="skill 名称")
+    p_enable.add_argument("--target", choices=[*ALL_TARGETS, "all"], default="all")
+    p_enable.add_argument("--project", help="项目根目录（默认自动检测）")
+    p_enable.set_defaults(func=cmd_enable)
+
+    p_disable = sub.add_parser("disable", help="项目级禁用 skill")
+    p_disable.add_argument("name", help="skill 名称")
+    p_disable.add_argument("--target", choices=[*ALL_TARGETS, "all"], default="all")
+    p_disable.add_argument("--project", help="项目根目录（默认自动检测）")
+    p_disable.set_defaults(func=cmd_disable)
+
+    p_sync = sub.add_parser("sync", help="按 .skills.yaml 重建项目软链")
+    p_sync.add_argument("--target", choices=[*ALL_TARGETS, "all"], default="all")
+    p_sync.add_argument("--project", help="项目根目录（默认自动检测）")
+    p_sync.set_defaults(func=cmd_sync)
+
+    p_create = sub.add_parser("create", help="从 _template 创建自研 skill")
+    p_create.add_argument("name", help="skill 名称")
+    p_create.set_defaults(func=cmd_create)
+
+    p_doctor = sub.add_parser("doctor", help="软链健康检查")
+    p_doctor.add_argument("--target", choices=[*ALL_TARGETS, "all"], default="all")
+    p_doctor.add_argument("--fix", action="store_true", help="自动修复可修复的软链")
+    p_doctor.set_defaults(func=cmd_doctor)
+
+    return parser
+
+
+def main(argv: list[str] | None = None) -> None:
+    parser = build_parser()
+    args = parser.parse_args(argv)
+    args.func(args)
+
+
+if __name__ == "__main__":
+    main()