Files
.pouch/skiff/cli.py
T
2026-07-27 14:03:19 +08:00

639 lines
21 KiB
Python
Raw Blame History

This file contains ambiguous Unicode characters
This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.
"""skiff CLI 入口。"""
from __future__ import annotations
import argparse
import re
import subprocess
import sys
from datetime import datetime
from pathlib import Path
from skiff import __version__
from skiff.agents import flatten_agent_args, resolve_agent_args
from skiff.gitops import publish as git_publish
from skiff.paths import (
ALL_TARGETS,
DRAFTS_DIR,
EXTERNALS_DIR,
SKILLS_DIR,
SKILLS_HOME,
TEMPLATE_DIR,
agent_skill_dir,
ensure_skills_home,
)
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,
skill_description,
validate_skill_dir,
validate_skill_name,
)
from skiff.yaml_io import safe_dump
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 _warn_deprecated(old: str, new: str) -> None:
_err(f"警告: `{old}` 已弃用,请改用 `{new}`")
def _project_root(explicit: str | None = None) -> Path:
if explicit:
return Path(explicit).resolve()
return find_repo_root() or Path.cwd()
def _collect_skill_names(positional: list[str] | None, flagged: list[str] | None) -> list[str]:
names = list(positional or [])
if flagged:
names.extend(flagged)
if "*" in names:
return list_owned_skills()
return names
def _ensure_external_fetched(name: str) -> None:
registry = load_registry()
if name not in registry:
return
entry = registry[name]
path = external_skill_path(name, entry)
if path.exists():
return
repo = entry["repo"]
ref = entry.get("ref", "main")
dest = EXTERNALS_DIR / name
EXTERNALS_DIR.mkdir(parents=True, exist_ok=True)
_print(f"拉取外部 skill: {name}")
subprocess.run(
["git", "clone", "--branch", ref, "--", repo, str(dest)],
check=True,
)
def _install_skill(name: str, targets: list[str], project_root: Path | None = None) -> None:
_ensure_external_fetched(name)
skill_path, _ = 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 is None else "项目"
_print(f"已安装 ({scope}/{target}): {name} -> {skill_path}")
def _list_installed_names(project_root: Path | None, targets: list[str]) -> list[str]:
names: set[str] = set()
for target in targets:
skill_dir = agent_skill_dir(target, project_root=project_root)
if not skill_dir.is_dir():
continue
for entry in skill_dir.iterdir():
if entry.name.startswith("."):
continue
if entry.is_symlink() or entry.is_dir():
names.add(entry.name)
return sorted(names)
def _remove_skill(
name: str,
targets: list[str],
*,
project_root: Path | None,
) -> int:
removed = 0
for target in targets:
link = agent_skill_dir(target, project_root=project_root) / name
try:
if remove_link(link):
scope = "全局" if project_root is None else "项目"
_print(f"已移除 ({scope}/{target}): {name}")
removed += 1
except FileExistsError as exc:
_err(f"跳过 {link}: {exc}")
return removed
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 cmd_bootstrap(args: argparse.Namespace) -> None:
del args
ensure_skills_home()
project_skill = "skiff"
owned_skill_path(project_skill)
_install_skill(project_skill, list(ALL_TARGETS), project_root=None)
_print("已安装项目 skill 到所有 agent")
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_agent_args(flatten_agent_args(args.agents))
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"agents: {', '.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 _print_available_skills() -> None:
ensure_skills_home()
owned = list_owned_skills()
if not owned:
_print("~/.skills/skills/ 中没有自研 skill")
return
_print(f"来源: {SKILLS_DIR}\n")
for name in owned:
desc = skill_description(name)
_print(f" {name}")
if desc:
for line in desc.splitlines():
_print(f" {line}")
_print("")
def cmd_add(args: argparse.Namespace) -> None:
ensure_skills_home()
if args.list_available:
_print_available_skills()
_print("使用 skiff add <name> 安装,或 skiff add <name> -g 全局安装")
return
if args.all:
names = list_owned_skills()
targets = resolve_agent_args(["*"])
else:
names = _collect_skill_names(args.skills, args.skills_flag)
if not names:
raise SystemExit("请指定 skill 名称,或使用 skiff add --list 查看可用 skill")
targets = resolve_agent_args(flatten_agent_args(args.agents))
project_root = None if args.global_scope else _project_root(args.project)
for name in names:
validate_skill_name(name)
_install_skill(name, targets, project_root=project_root)
def cmd_remove(args: argparse.Namespace) -> None:
ensure_skills_home()
project_root = None if args.global_scope else _project_root(args.project)
targets = resolve_agent_args(flatten_agent_args(args.agents))
if args.all:
names = _list_installed_names(project_root, targets)
else:
names = _collect_skill_names(args.skills, args.skills_flag)
if not names:
raise SystemExit("请指定 skill 名称,或使用 skiff remove --all")
total = 0
for name in names:
total += _remove_skill(name, targets, project_root=project_root)
if total == 0:
_print("没有移除任何 skill")
def cmd_publish(args: argparse.Namespace) -> None:
ensure_skills_home()
paths = args.paths or ["."]
git_publish(
paths=paths,
message=args.message,
push=args.push,
no_commit=args.no_commit,
)
def cmd_registry_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_enable(args: argparse.Namespace) -> None:
_warn_deprecated("skiff enable", "skiff add <name>")
ensure_skills_home()
validate_skill_name(args.name)
root = _project_root(args.project)
manifest_path = root / ".skills.yaml"
registry = load_registry()
if args.name in 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_agent_args(flatten_agent_args(args.agents))
_, 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:
_warn_deprecated("skiff disable", "skiff remove <name>")
root = _project_root(args.project)
manifest_path = root / ".skills.yaml"
if not remove_skill_from_manifest(manifest_path, args.name):
_print(f"manifest 中不存在: {args.name}")
return
targets = resolve_agent_args(flatten_agent_args(args.agents))
_remove_skill(args.name, targets, project_root=root)
def cmd_sync(args: argparse.Namespace) -> None:
ensure_skills_home()
root = _project_root(args.project)
manifest_path = root / ".skills.yaml"
if not manifest_path.is_file():
raise SystemExit(f"未找到 {manifest_path}")
_, data = load_manifest(manifest_path)
targets = resolve_agent_args(flatten_agent_args(args.agents))
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 = DRAFTS_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")
brief = {
"name": args.name,
"idea": args.idea or "",
"source_project": str(Path(args.from_project).resolve()) if args.from_project else "",
"status": "draft",
"created_at": datetime.now().astimezone().isoformat(timespec="seconds"),
}
(dst / "brief.yaml").write_text(safe_dump(brief), encoding="utf-8")
_print(f"草稿已创建: {dst}")
_print(f"下一步: 请完善 skiff 草稿 {args.name}")
_print(f"完成后运行: skiff check {args.name} && skiff finalize {args.name}")
def _draft_or_owned_path(name: str) -> tuple[Path, str]:
draft = DRAFTS_DIR / name
if draft.is_dir():
return draft, "草稿"
owned = SKILLS_DIR / name
if owned.is_dir():
return owned, "正式 skill"
raise SystemExit(f"找不到草稿或正式 skill: {name}")
def cmd_check(args: argparse.Namespace) -> None:
ensure_skills_home()
validate_skill_name(args.name)
path, kind = _draft_or_owned_path(args.name)
issues = validate_skill_dir(path, args.name)
if kind == "草稿":
issues = [issue for issue in issues if "草稿文件: brief.yaml" not in issue]
if issues:
for issue in issues:
_err(f"✗ {issue}")
raise SystemExit(1)
_print(f"✓ 校验通过 ({kind}): {path}")
def cmd_finalize(args: argparse.Namespace) -> None:
ensure_skills_home()
validate_skill_name(args.name)
draft = DRAFTS_DIR / args.name
if not draft.is_dir():
raise SystemExit(f"草稿不存在: {args.name}")
final = SKILLS_DIR / args.name
if final.exists():
raise SystemExit(f"正式 skill 已存在: {final}")
issues = validate_skill_dir(draft, args.name)
issues = [issue for issue in issues if "草稿文件: brief.yaml" not in issue]
if issues:
for issue in issues:
_err(f"✗ {issue}")
raise SystemExit(1)
final.parent.mkdir(parents=True, exist_ok=True)
draft.replace(final)
brief = final / "brief.yaml"
if brief.exists():
brief.unlink()
_print(f"已完成 skill: {final}")
_print(f"下一步: skiff publish skills/{args.name} -m \"add {args.name}\" --push")
def cmd_doctor(args: argparse.Namespace) -> None:
ensure_skills_home()
targets = resolve_agent_args(flatten_agent_args(args.agents))
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 _add_common_flags(parser: argparse.ArgumentParser) -> None:
parser.add_argument(
"-a",
"--agent",
dest="agents",
nargs="+",
action="append",
metavar="AGENT",
help="目标 agentcursor、claude、claude-code、codex、*",
)
parser.add_argument(
"-g",
"--global",
dest="global_scope",
action="store_true",
help="全局安装/卸载(默认当前项目)",
)
parser.add_argument("-y", "--yes", action="store_true", help="跳过确认(兼容 Vercel skills")
parser.add_argument("--project", help="项目根目录(默认自动检测或当前目录)")
def build_parser() -> argparse.ArgumentParser:
parser = argparse.ArgumentParser(
prog="skiff",
description="自研 Agent Skills 安装与管理 CLI(接口对齐 Vercel skills",
)
parser.add_argument("--version", action="version", version=f"skiff {__version__}")
sub = parser.add_subparsers(dest="command", required=True)
p_bootstrap = sub.add_parser(
"bootstrap",
help="将本项目 skiff skill 全局安装到所有 agent",
)
p_bootstrap.set_defaults(func=cmd_bootstrap)
p_list = sub.add_parser("list", help="列出 ~/.skills 中的 skill 目录")
p_list.set_defaults(func=cmd_list)
p_status = sub.add_parser("status", help="安装状态总览")
p_status.add_argument("-a", "--agent", dest="agents", nargs="+", action="append", metavar="AGENT")
p_status.set_defaults(func=cmd_status)
p_add = sub.add_parser(
"add",
help="安装 skill 到 agent(自研或 registry",
description="安装 ~/.skills 中的自研 skill,或 registry 中的外部 skill",
)
p_add.add_argument("skills", nargs="*", metavar="skill", help="skill 名称(可多个)")
p_add.add_argument("-s", "--skill", dest="skills_flag", action="append", metavar="SKILL")
p_add.add_argument("--list", dest="list_available", action="store_true", help="列出可用自研 skill,不安装")
p_add.add_argument("--all", action="store_true", help="安装全部自研 skill 到全部 agent")
_add_common_flags(p_add)
p_add.set_defaults(func=cmd_add)
p_remove = sub.add_parser(
"remove",
aliases=["rm", "r"],
help="从 agent 移除 skill",
)
p_remove.add_argument("skills", nargs="*", metavar="skill", help="skill 名称(可多个)")
p_remove.add_argument("-s", "--skill", dest="skills_flag", action="append", metavar="SKILL")
p_remove.add_argument("--all", action="store_true", help="移除当前范围内全部 skill")
_add_common_flags(p_remove)
p_remove.set_defaults(func=cmd_remove)
p_publish = sub.add_parser(
"publish",
help="在 ~/.skills 内 git add / commit / push",
)
p_publish.add_argument("paths", nargs="*", help="要提交的路径(默认 .")
p_publish.add_argument("-m", "--message", help="commit 说明")
p_publish.add_argument("--push", action="store_true", help="commit 后 push")
p_publish.add_argument("--no-commit", action="store_true", help="只 git add,不 commit")
p_publish.set_defaults(func=cmd_publish)
p_registry = sub.add_parser("registry", help="管理 registry.yaml 中的外部 skill")
registry_sub = p_registry.add_subparsers(dest="registry_command", required=True)
p_reg_add = registry_sub.add_parser("add", help="注册外部 Git skill")
p_reg_add.add_argument("name", help="registry 名称")
p_reg_add.add_argument("repo", help="Git 仓库 URL")
p_reg_add.add_argument("--ref", default="main", help="分支或 tag(默认 main")
p_reg_add.add_argument("--path", default=".", help="仓库内子路径(默认 .")
p_reg_add.set_defaults(func=cmd_registry_add)
p_fetch = sub.add_parser("fetch", help="拉取/更新 registry 中的外部 skill")
p_fetch.add_argument("name", help="registry 名称")
p_fetch.set_defaults(func=cmd_fetch)
p_enable = sub.add_parser("enable", help=argparse.SUPPRESS)
p_enable.add_argument("name")
p_enable.add_argument("-a", "--agent", dest="agents", nargs="+", action="append")
p_enable.add_argument("--project")
p_enable.set_defaults(func=cmd_enable)
p_disable = sub.add_parser("disable", help=argparse.SUPPRESS)
p_disable.add_argument("name")
p_disable.add_argument("-a", "--agent", dest="agents", nargs="+", action="append")
p_disable.add_argument("--project")
p_disable.set_defaults(func=cmd_disable)
p_sync = sub.add_parser("sync", help="按 .skills.yaml 重建项目软链")
p_sync.add_argument("-a", "--agent", dest="agents", nargs="+", action="append")
p_sync.add_argument("--project")
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.add_argument("--idea", help="创建 skill 的原始想法")
p_create.add_argument("--from-project", help="想法来源项目(仅记录上下文)")
p_create.set_defaults(func=cmd_create)
p_check = sub.add_parser("check", help="校验草稿或正式 skill")
p_check.add_argument("name", help="skill 名称")
p_check.set_defaults(func=cmd_check)
p_finalize = sub.add_parser("finalize", help="校验草稿并转为正式 skill")
p_finalize.add_argument("name", help="skill 名称")
p_finalize.set_defaults(func=cmd_finalize)
p_doctor = sub.add_parser("doctor", help="软链健康检查")
p_doctor.add_argument("-a", "--agent", dest="agents", nargs="+", action="append")
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()