mirror of
https://github.com/zhayujie/chatgpt-on-wechat.git
synced 2026-06-02 00:57:41 +08:00
1142 lines
39 KiB
Python
1142 lines
39 KiB
Python
"""cow skill - Skill management commands."""
|
|
|
|
import os
|
|
import re
|
|
import sys
|
|
import json
|
|
import hashlib
|
|
import shutil
|
|
import zipfile
|
|
import tempfile
|
|
|
|
from urllib.parse import urlparse
|
|
|
|
import click
|
|
import requests
|
|
|
|
from cli.utils import (
|
|
get_project_root,
|
|
get_skills_dir,
|
|
get_builtin_skills_dir,
|
|
load_skills_config,
|
|
SKILL_HUB_API,
|
|
)
|
|
|
|
_SAFE_NAME_RE = re.compile(r"^[a-zA-Z0-9][a-zA-Z0-9_\-]{0,63}$")
|
|
_GITHUB_URL_RE = re.compile(
|
|
r"^https?://github\.com/([^/]+)/([^/]+?)(?:\.git)?(?:/(?:tree|blob)/([^/]+)(?:/(.+))?)?/?$"
|
|
)
|
|
_GITLAB_URL_RE = re.compile(
|
|
r"^https?://gitlab\.com/([^/]+)/([^/]+?)(?:\.git)?(?:/-/tree/([^/]+)(?:/(.+))?)?/?$"
|
|
)
|
|
_GIT_SSH_RE = re.compile(
|
|
r"^git@([^:]+):([^/]+)/([^/]+?)(?:\.git)?$"
|
|
)
|
|
|
|
|
|
def _parse_github_url(url: str):
|
|
"""Parse a full GitHub URL into (owner, repo, branch, subpath).
|
|
|
|
Returns None if the URL doesn't match.
|
|
Supported formats:
|
|
https://github.com/owner/repo
|
|
https://github.com/owner/repo/tree/branch
|
|
https://github.com/owner/repo/tree/branch/path/to/skill
|
|
https://github.com/owner/repo/blob/branch/path/to/skill
|
|
"""
|
|
m = _GITHUB_URL_RE.match(url.strip())
|
|
if not m:
|
|
return None
|
|
owner, repo, branch, subpath = m.groups()
|
|
return owner, repo, branch or "main", subpath
|
|
|
|
|
|
def _parse_gitlab_url(url: str):
|
|
"""Parse a GitLab URL into (owner, repo, branch, subpath).
|
|
|
|
Returns None if the URL doesn't match.
|
|
Supported formats:
|
|
https://gitlab.com/owner/repo
|
|
https://gitlab.com/owner/repo/-/tree/branch
|
|
https://gitlab.com/owner/repo/-/tree/branch/path/to/skill
|
|
"""
|
|
m = _GITLAB_URL_RE.match(url.strip())
|
|
if not m:
|
|
return None
|
|
owner, repo, branch, subpath = m.groups()
|
|
return owner, repo, branch or "main", subpath
|
|
|
|
|
|
def _parse_git_ssh_url(url: str):
|
|
"""Parse a git@ SSH URL into (host, owner, repo).
|
|
|
|
Returns None if the URL doesn't match.
|
|
Supported format: git@github.com:owner/repo.git
|
|
"""
|
|
m = _GIT_SSH_RE.match(url.strip())
|
|
if not m:
|
|
return None
|
|
host, owner, repo = m.groups()
|
|
return host, owner, repo
|
|
|
|
|
|
def _clone_repo(git_url: str):
|
|
"""Shallow-clone a git repo and return (tmp_dir, repo_root).
|
|
|
|
Requires git to be installed. The caller must clean up tmp_dir.
|
|
"""
|
|
tmp_dir = tempfile.mkdtemp(prefix="cow-skill-")
|
|
repo_dir = os.path.join(tmp_dir, "repo")
|
|
try:
|
|
import subprocess
|
|
subprocess.run(
|
|
["git", "clone", "--depth", "1", git_url, repo_dir],
|
|
check=True, capture_output=True, timeout=120,
|
|
)
|
|
except FileNotFoundError:
|
|
shutil.rmtree(tmp_dir, ignore_errors=True)
|
|
raise RuntimeError("git is not installed")
|
|
except Exception as e:
|
|
shutil.rmtree(tmp_dir, ignore_errors=True)
|
|
raise RuntimeError(f"git clone failed: {e}")
|
|
return tmp_dir, repo_dir
|
|
|
|
|
|
def _download_repo_zip(spec: str, branch: str = "main", host: str = "github"):
|
|
"""Download a GitHub/GitLab repo as zip and extract it.
|
|
|
|
Returns (tmp_dir, repo_root) where tmp_dir is the temp directory to clean up
|
|
and repo_root is the extracted repository root path.
|
|
"""
|
|
if host == "gitlab":
|
|
zip_url = f"https://gitlab.com/{spec}/-/archive/{branch}/{spec.split('/')[-1]}-{branch}.zip"
|
|
else:
|
|
zip_url = f"https://github.com/{spec}/archive/refs/heads/{branch}.zip"
|
|
resp = requests.get(zip_url, timeout=120, allow_redirects=True)
|
|
resp.raise_for_status()
|
|
|
|
tmp_dir = tempfile.mkdtemp(prefix="cow-skill-")
|
|
zip_path = os.path.join(tmp_dir, "repo.zip")
|
|
with open(zip_path, "wb") as f:
|
|
f.write(resp.content)
|
|
|
|
extract_dir = os.path.join(tmp_dir, "extracted")
|
|
with zipfile.ZipFile(zip_path, "r") as zf:
|
|
_safe_extractall(zf, extract_dir)
|
|
|
|
# GitHub zips have a single top-level dir like "repo-main/"
|
|
top_items = [d for d in os.listdir(extract_dir) if not d.startswith(".")]
|
|
if len(top_items) == 1 and os.path.isdir(os.path.join(extract_dir, top_items[0])):
|
|
return tmp_dir, os.path.join(extract_dir, top_items[0])
|
|
return tmp_dir, extract_dir
|
|
|
|
|
|
def _download_github_dir(owner, repo, branch, subpath, dest_dir):
|
|
"""Download a subdirectory from GitHub using the Contents API.
|
|
|
|
Recursively fetches all files under the given subpath and writes them
|
|
to dest_dir. Used as a fallback when zip download fails.
|
|
Costs one API request per directory (60/hr unauthenticated).
|
|
"""
|
|
api_url = f"https://api.github.com/repos/{owner}/{repo}/contents/{subpath}?ref={branch}"
|
|
resp = requests.get(api_url, timeout=30, headers={"Accept": "application/vnd.github.v3+json"})
|
|
resp.raise_for_status()
|
|
items = resp.json()
|
|
|
|
if isinstance(items, dict):
|
|
items = [items]
|
|
|
|
for item in items:
|
|
rel_path = item["path"]
|
|
if subpath:
|
|
rel_path = rel_path[len(subpath.strip("/")):].lstrip("/")
|
|
local_path = os.path.join(dest_dir, rel_path)
|
|
|
|
if item["type"] == "file":
|
|
os.makedirs(os.path.dirname(local_path), exist_ok=True)
|
|
dl_url = item.get("download_url")
|
|
if not dl_url:
|
|
continue
|
|
file_resp = requests.get(dl_url, timeout=30)
|
|
file_resp.raise_for_status()
|
|
with open(local_path, "wb") as f:
|
|
f.write(file_resp.content)
|
|
elif item["type"] == "dir":
|
|
os.makedirs(local_path, exist_ok=True)
|
|
child_subpath = item["path"]
|
|
_download_github_dir(owner, repo, branch, child_subpath, dest_dir)
|
|
|
|
|
|
# Directories to search for skills following the Agent Skills convention
|
|
_SKILL_SCAN_DIRS = [
|
|
"skills",
|
|
"skills/.curated",
|
|
"skills/.experimental",
|
|
]
|
|
|
|
_SKILL_SCAN_SKIP = {
|
|
"node_modules", "__pycache__", ".git", ".github", "venv", ".venv",
|
|
}
|
|
|
|
|
|
def _scan_skills_in_repo(repo_root: str) -> list:
|
|
"""Scan a repo for skill directories containing SKILL.md.
|
|
|
|
Searches in conventional locations (skills/, skills/.curated/, etc.)
|
|
and also checks the repo root itself.
|
|
|
|
Returns a list of (skill_name, skill_dir_path) tuples.
|
|
"""
|
|
found = []
|
|
|
|
# Check repo root for a SKILL.md (single-skill repo)
|
|
if os.path.isfile(os.path.join(repo_root, "SKILL.md")):
|
|
fm = _parse_skill_frontmatter(_read_file_text(os.path.join(repo_root, "SKILL.md")))
|
|
name = fm.get("name") or os.path.basename(repo_root)
|
|
found.append((name, repo_root))
|
|
return found
|
|
|
|
for scan_dir in _SKILL_SCAN_DIRS:
|
|
search_root = os.path.join(repo_root, scan_dir)
|
|
if not os.path.isdir(search_root):
|
|
continue
|
|
for entry in os.listdir(search_root):
|
|
if entry.startswith(".") and entry not in (".curated", ".experimental"):
|
|
continue
|
|
if entry in _SKILL_SCAN_SKIP:
|
|
continue
|
|
entry_path = os.path.join(search_root, entry)
|
|
if os.path.isdir(entry_path) and os.path.isfile(os.path.join(entry_path, "SKILL.md")):
|
|
fm = _parse_skill_frontmatter(
|
|
_read_file_text(os.path.join(entry_path, "SKILL.md"))
|
|
)
|
|
name = fm.get("name") or entry
|
|
found.append((name, entry_path))
|
|
|
|
return found
|
|
|
|
|
|
def _read_file_text(path: str) -> str:
|
|
"""Read a file as UTF-8 text, returning empty string on failure."""
|
|
try:
|
|
with open(path, "r", encoding="utf-8") as f:
|
|
return f.read()
|
|
except Exception:
|
|
return ""
|
|
|
|
|
|
def _install_local(path: str):
|
|
"""Install skill(s) from a local directory.
|
|
|
|
If the path contains SKILL.md directly, install it as a single skill.
|
|
Otherwise scan for skills/ subdirectories with SKILL.md.
|
|
"""
|
|
path = os.path.abspath(os.path.expanduser(path))
|
|
if not os.path.isdir(path):
|
|
click.echo(f"Error: '{path}' is not a directory.", err=True)
|
|
sys.exit(1)
|
|
|
|
skills_dir = get_skills_dir()
|
|
os.makedirs(skills_dir, exist_ok=True)
|
|
|
|
if os.path.isfile(os.path.join(path, "SKILL.md")):
|
|
fm = _parse_skill_frontmatter(_read_file_text(os.path.join(path, "SKILL.md")))
|
|
skill_name = fm.get("name") or os.path.basename(path)
|
|
skill_name = re.sub(r'[^a-zA-Z0-9_\-]', '-', skill_name)[:64]
|
|
_validate_skill_name(skill_name)
|
|
target_dir = os.path.join(skills_dir, skill_name)
|
|
if os.path.exists(target_dir):
|
|
shutil.rmtree(target_dir)
|
|
shutil.copytree(path, target_dir)
|
|
_register_installed_skill(skill_name, source="local")
|
|
_print_install_success(skill_name, "local")
|
|
return
|
|
|
|
discovered = _scan_skills_in_repo(path)
|
|
if not discovered:
|
|
click.echo(f"Error: No skills found in '{path}'.", err=True)
|
|
sys.exit(1)
|
|
|
|
click.echo(f"Found {len(discovered)} skill(s) in {path}:\n")
|
|
installed_names = []
|
|
for sname, sdir in discovered:
|
|
safe_name = re.sub(r'[^a-zA-Z0-9_\-]', '-', sname)[:64]
|
|
if not _SAFE_NAME_RE.match(safe_name):
|
|
click.echo(click.style(f" ✗ Skipping '{sname}' (invalid name)", fg="yellow"))
|
|
continue
|
|
target_dir = os.path.join(skills_dir, safe_name)
|
|
if os.path.exists(target_dir):
|
|
shutil.rmtree(target_dir)
|
|
shutil.copytree(sdir, target_dir)
|
|
_register_installed_skill(safe_name, source="local")
|
|
installed_names.append(safe_name)
|
|
|
|
if installed_names:
|
|
click.echo("")
|
|
for n in installed_names:
|
|
_print_install_success(n, "local")
|
|
click.echo(f"\n {len(installed_names)} skill(s) installed from local path.")
|
|
else:
|
|
click.echo("No valid skills found in directory.")
|
|
|
|
|
|
def _register_installed_skill(name: str, source: str = "cowhub"):
|
|
"""Register a newly installed skill into skills_config.json.
|
|
|
|
source values: builtin, cow, github, clawhub, linkai, local, url
|
|
"""
|
|
skills_dir = get_skills_dir()
|
|
config_path = os.path.join(skills_dir, "skills_config.json")
|
|
|
|
config = {}
|
|
if os.path.exists(config_path):
|
|
try:
|
|
with open(config_path, "r", encoding="utf-8") as f:
|
|
config = json.load(f)
|
|
except Exception:
|
|
config = {}
|
|
|
|
if name in config:
|
|
return
|
|
|
|
skill_dir = os.path.join(skills_dir, name)
|
|
description = _read_skill_description(skill_dir) or ""
|
|
|
|
config[name] = {
|
|
"name": name,
|
|
"description": description,
|
|
"source": source,
|
|
"enabled": True,
|
|
"category": "skill",
|
|
}
|
|
|
|
try:
|
|
with open(config_path, "w", encoding="utf-8") as f:
|
|
json.dump(config, f, indent=4, ensure_ascii=False)
|
|
except Exception:
|
|
pass
|
|
|
|
|
|
def _parse_skill_frontmatter(content: str) -> dict:
|
|
"""Parse YAML frontmatter from SKILL.md content and return a dict with name/description."""
|
|
result = {}
|
|
match = re.match(r'^---\s*\n(.*?)\n---\s*\n', content, re.DOTALL)
|
|
if not match:
|
|
return result
|
|
for line in match.group(1).split('\n'):
|
|
line = line.strip()
|
|
for key in ('name', 'description'):
|
|
if line.startswith(f'{key}:'):
|
|
val = line[len(key) + 1:].strip()
|
|
result[key] = val.strip('"').strip("'")
|
|
return result
|
|
|
|
|
|
def _read_skill_description(skill_dir: str) -> str:
|
|
"""Read the description from a skill's SKILL.md frontmatter."""
|
|
skill_md = os.path.join(skill_dir, "SKILL.md")
|
|
if not os.path.exists(skill_md):
|
|
return ""
|
|
try:
|
|
with open(skill_md, "r", encoding="utf-8") as f:
|
|
content = f.read()
|
|
return _parse_skill_frontmatter(content).get("description", "")
|
|
except Exception:
|
|
return ""
|
|
|
|
|
|
def _install_url(url: str):
|
|
"""Install a skill from a direct SKILL.md URL."""
|
|
click.echo(f"Downloading SKILL.md from {url} ...")
|
|
try:
|
|
resp = requests.get(url, timeout=30)
|
|
resp.raise_for_status()
|
|
except Exception as e:
|
|
click.echo(f"Error: Failed to download SKILL.md: {e}", err=True)
|
|
sys.exit(1)
|
|
|
|
content = resp.text
|
|
fm = _parse_skill_frontmatter(content)
|
|
skill_name = fm.get("name")
|
|
if not skill_name:
|
|
click.echo("Error: SKILL.md missing 'name' field in frontmatter.", err=True)
|
|
sys.exit(1)
|
|
|
|
skill_name = skill_name.strip()
|
|
_validate_skill_name(skill_name)
|
|
|
|
skills_dir = get_skills_dir()
|
|
os.makedirs(skills_dir, exist_ok=True)
|
|
skill_dir = os.path.join(skills_dir, skill_name)
|
|
|
|
if os.path.isdir(skill_dir):
|
|
click.echo(f"Skill '{skill_name}' already exists. Overwriting SKILL.md ...")
|
|
os.makedirs(skill_dir, exist_ok=True)
|
|
|
|
with open(os.path.join(skill_dir, "SKILL.md"), "w", encoding="utf-8") as f:
|
|
f.write(content)
|
|
|
|
_register_installed_skill(skill_name, source="url")
|
|
_print_install_success(skill_name, "url")
|
|
|
|
|
|
def _print_install_success(name: str, source: str):
|
|
"""Print a unified install success message with description and source."""
|
|
skills_dir = get_skills_dir()
|
|
desc = _read_skill_description(os.path.join(skills_dir, name))
|
|
click.echo(click.style(f"✓ {name}", fg="green"))
|
|
if desc:
|
|
if len(desc) > 60:
|
|
desc = desc[:57] + "…"
|
|
click.echo(f" {desc}")
|
|
click.echo(f" 来源: {source}")
|
|
|
|
|
|
def _validate_skill_name(name: str):
|
|
"""Reject names that contain path traversal or special characters."""
|
|
if not _SAFE_NAME_RE.match(name):
|
|
click.echo(
|
|
f"Error: Invalid skill name '{name}'. "
|
|
"Use only letters, digits, hyphens, and underscores.",
|
|
err=True,
|
|
)
|
|
sys.exit(1)
|
|
|
|
|
|
def _validate_github_spec(spec: str):
|
|
"""Reject specs that don't look like owner/repo."""
|
|
if not re.match(r"^[a-zA-Z0-9_\-]+/[a-zA-Z0-9_.\-]+$", spec):
|
|
click.echo(f"Error: Invalid GitHub spec '{spec}'. Expected format: owner/repo", err=True)
|
|
sys.exit(1)
|
|
|
|
|
|
def _safe_extractall(zf: zipfile.ZipFile, dest: str):
|
|
"""Extract zip while guarding against Zip Slip (path traversal)."""
|
|
dest = os.path.realpath(dest)
|
|
for member in zf.infolist():
|
|
target = os.path.realpath(os.path.join(dest, member.filename))
|
|
if not target.startswith(dest + os.sep) and target != dest:
|
|
raise ValueError(f"Unsafe zip entry detected: {member.filename}")
|
|
zf.extractall(dest)
|
|
|
|
|
|
def _verify_checksum(content: bytes, expected: str):
|
|
"""Verify SHA-256 checksum of downloaded content.
|
|
|
|
Returns True if checksum matches or no expected value provided.
|
|
Exits with error if mismatch.
|
|
"""
|
|
if not expected:
|
|
return True
|
|
actual = hashlib.sha256(content).hexdigest()
|
|
if actual != expected.lower():
|
|
click.echo(
|
|
f"Error: Checksum mismatch!\n"
|
|
f" Expected: {expected}\n"
|
|
f" Actual: {actual}\n"
|
|
f"The downloaded package may have been tampered with.",
|
|
err=True,
|
|
)
|
|
sys.exit(1)
|
|
return True
|
|
|
|
|
|
@click.group()
|
|
def skill():
|
|
"""Manage CowAgent skills."""
|
|
pass
|
|
|
|
|
|
# ------------------------------------------------------------------
|
|
# cow skill list
|
|
# ------------------------------------------------------------------
|
|
@skill.command("list")
|
|
@click.option("--remote", is_flag=True, help="Browse skills on Skill Hub")
|
|
@click.option("--page", default=1, type=int, help="Page number for remote listing")
|
|
def skill_list(remote, page):
|
|
"""List installed skills or browse Skill Hub."""
|
|
if remote:
|
|
_list_remote(page=page)
|
|
else:
|
|
_list_local()
|
|
|
|
|
|
def _list_local():
|
|
"""List locally installed skills."""
|
|
config = load_skills_config()
|
|
skills_dir = get_skills_dir()
|
|
builtin_dir = get_builtin_skills_dir()
|
|
|
|
if not config:
|
|
# Fallback: scan directories directly
|
|
entries = []
|
|
for d in [builtin_dir, skills_dir]:
|
|
if not os.path.isdir(d):
|
|
continue
|
|
source = "builtin" if d == builtin_dir else "custom"
|
|
for name in sorted(os.listdir(d)):
|
|
skill_path = os.path.join(d, name)
|
|
if os.path.isdir(skill_path) and not name.startswith("."):
|
|
has_skill_md = os.path.exists(os.path.join(skill_path, "SKILL.md"))
|
|
if has_skill_md:
|
|
entries.append({"name": name, "source": source, "enabled": True, "description": ""})
|
|
if not entries:
|
|
click.echo("No skills installed.")
|
|
return
|
|
_print_skill_table(entries)
|
|
return
|
|
|
|
entries = sorted(config.values(), key=lambda x: x.get("name", ""))
|
|
if not entries:
|
|
click.echo("No skills installed.")
|
|
return
|
|
_print_skill_table(entries)
|
|
|
|
|
|
def _print_skill_table(entries):
|
|
"""Print skills as a formatted table."""
|
|
name_w = max(len(e.get("name", "")) for e in entries)
|
|
name_w = max(name_w, 4) + 2
|
|
desc_w = 40
|
|
|
|
header = f"{'Name':<{name_w}} {'Status':<10} {'Source':<10} {'Description'}"
|
|
click.echo(f"\n Installed skills ({len(entries)})\n")
|
|
click.echo(f" {header}")
|
|
click.echo(f" {'─' * (name_w + 10 + 10 + desc_w)}")
|
|
|
|
for e in entries:
|
|
name = e.get("name", "")
|
|
enabled = e.get("enabled", True)
|
|
source = e.get("source", "")
|
|
desc = e.get("description", "") or ""
|
|
if len(desc) > desc_w:
|
|
desc = desc[:desc_w - 3] + "..."
|
|
|
|
status_icon = click.style("✓ on ", fg="green") if enabled else click.style("✗ off", fg="red")
|
|
click.echo(f" {name:<{name_w}} {status_icon} {source:<10} {desc}")
|
|
|
|
click.echo()
|
|
|
|
|
|
_REMOTE_PAGE_SIZE = 10
|
|
|
|
|
|
def _list_remote(page: int = 1):
|
|
"""List skills from remote Skill Hub with server-side pagination."""
|
|
try:
|
|
resp = requests.get(
|
|
f"{SKILL_HUB_API}/skills",
|
|
params={"page": page, "limit": _REMOTE_PAGE_SIZE},
|
|
timeout=10,
|
|
)
|
|
resp.raise_for_status()
|
|
data = resp.json()
|
|
except Exception as e:
|
|
click.echo(f"Error: Failed to fetch from Skill Hub: {e}", err=True)
|
|
sys.exit(1)
|
|
|
|
skills = data.get("skills", [])
|
|
total = data.get("total", len(skills))
|
|
|
|
if not skills and page == 1:
|
|
click.echo("No skills available on Skill Hub.")
|
|
return
|
|
|
|
total_pages = max(1, (total + _REMOTE_PAGE_SIZE - 1) // _REMOTE_PAGE_SIZE)
|
|
page = min(page, total_pages)
|
|
installed = set(load_skills_config().keys())
|
|
|
|
name_w = max((len(s.get("name", "")) for s in skills), default=4)
|
|
name_w = max(name_w, 4) + 2
|
|
|
|
click.echo(f"\n Skill Hub ({total} available) — page {page}/{total_pages}\n")
|
|
click.echo(f" {'Name':<{name_w}} {'Status':<12} {'Description'}")
|
|
click.echo(f" {'─' * (name_w + 12 + 50)}")
|
|
|
|
for s in skills:
|
|
name = s.get("name", "")
|
|
desc = s.get("description", "") or s.get("display_name", "")
|
|
if len(desc) > 50:
|
|
desc = desc[:47] + "..."
|
|
status = click.style("installed", fg="green") if name in installed else "—"
|
|
click.echo(f" {name:<{name_w}} {status:<12} {desc}")
|
|
|
|
click.echo()
|
|
nav_parts = []
|
|
if page > 1:
|
|
nav_parts.append(f"cow skill list --remote --page {page - 1}")
|
|
if page < total_pages:
|
|
nav_parts.append(f"cow skill list --remote --page {page + 1}")
|
|
if nav_parts:
|
|
click.echo(f" Navigate: {' | '.join(nav_parts)}")
|
|
click.echo(f" Install: cow skill install <name>\n")
|
|
|
|
|
|
# ------------------------------------------------------------------
|
|
# cow skill search
|
|
# ------------------------------------------------------------------
|
|
@skill.command()
|
|
@click.argument("query")
|
|
def search(query):
|
|
"""Search skills on Skill Hub."""
|
|
try:
|
|
resp = requests.get(f"{SKILL_HUB_API}/skills/search", params={"q": query}, timeout=10)
|
|
resp.raise_for_status()
|
|
data = resp.json()
|
|
except Exception as e:
|
|
click.echo(f"Error: Failed to search Skill Hub: {e}", err=True)
|
|
sys.exit(1)
|
|
|
|
skills = data.get("skills", [])
|
|
if not skills:
|
|
click.echo(f'No skills found for "{query}".')
|
|
return
|
|
|
|
installed = set(load_skills_config().keys())
|
|
name_w = max(len(s.get("name", "")) for s in skills)
|
|
name_w = max(name_w, 4) + 2
|
|
|
|
click.echo(f'\n Search results for "{query}" ({len(skills)} found)\n')
|
|
click.echo(f" {'Name':<{name_w}} {'Status':<12} {'Description'}")
|
|
click.echo(f" {'─' * (name_w + 12 + 50)}")
|
|
|
|
for s in skills:
|
|
name = s.get("name", "")
|
|
desc = s.get("description", "") or s.get("display_name", "")
|
|
if len(desc) > 50:
|
|
desc = desc[:47] + "..."
|
|
status = click.style("installed", fg="green") if name in installed else "—"
|
|
click.echo(f" {name:<{name_w}} {status:<12} {desc}")
|
|
|
|
click.echo(f"\n Install with: cow skill install <name>\n")
|
|
|
|
|
|
# ------------------------------------------------------------------
|
|
# cow skill install
|
|
# ------------------------------------------------------------------
|
|
@skill.command()
|
|
@click.argument("name")
|
|
def install(name):
|
|
"""Install skill(s) from Skill Hub, GitHub, GitLab, git URL, or local path.
|
|
|
|
When given an owner/repo (or full URL), downloads the repo and
|
|
auto-discovers all skills/ subdirectories containing SKILL.md,
|
|
installing them in batch. Use a subpath to install a single skill.
|
|
|
|
Examples:
|
|
|
|
cow skill install pptx (from Skill Hub)
|
|
|
|
cow skill install larksuite/cli (GitHub shorthand, all skills)
|
|
|
|
cow skill install larksuite/cli#skills/lark-im (single skill by subpath)
|
|
|
|
cow skill install https://github.com/owner/repo
|
|
|
|
cow skill install https://gitlab.com/org/repo
|
|
|
|
cow skill install git@github.com:owner/repo.git
|
|
|
|
cow skill install ./my-local-skills (local directory)
|
|
|
|
cow skill install https://example.com/path/to/SKILL.md
|
|
"""
|
|
# --- Local path ---
|
|
if name.startswith(("./", "../", "/", "~/")):
|
|
_install_local(name)
|
|
return
|
|
|
|
# --- Direct SKILL.md URL ---
|
|
if name.startswith(("http://", "https://")) and name.rstrip("/").endswith("SKILL.md"):
|
|
dir_url = re.sub(r'/SKILL\.md/?$', '', name)
|
|
gh = _parse_github_url(dir_url)
|
|
if gh:
|
|
owner, repo, branch, subpath = gh
|
|
_install_github(f"{owner}/{repo}", subpath=subpath, skill_name=(
|
|
subpath.rstrip("/").split("/")[-1] if subpath else repo
|
|
), branch=branch)
|
|
return
|
|
_install_url(name)
|
|
return
|
|
|
|
# --- Full GitHub URL ---
|
|
parsed = _parse_github_url(name)
|
|
if parsed:
|
|
owner, repo, branch, subpath = parsed
|
|
_install_github(f"{owner}/{repo}", subpath=subpath, branch=branch)
|
|
return
|
|
|
|
# --- Full GitLab URL ---
|
|
gl = _parse_gitlab_url(name)
|
|
if gl:
|
|
owner, repo, branch, subpath = gl
|
|
_install_gitlab(f"{owner}/{repo}", subpath=subpath, branch=branch)
|
|
return
|
|
|
|
# --- git@host:owner/repo.git SSH URL ---
|
|
ssh = _parse_git_ssh_url(name)
|
|
if ssh:
|
|
host, owner, repo = ssh
|
|
_install_git_clone(name, f"{owner}/{repo}")
|
|
return
|
|
|
|
# --- github: prefix ---
|
|
if name.startswith("github:"):
|
|
raw = name[7:]
|
|
subpath = None
|
|
if "#" in raw:
|
|
raw, subpath = raw.split("#", 1)
|
|
_validate_github_spec(raw)
|
|
_install_github(raw, subpath=subpath)
|
|
return
|
|
|
|
# --- clawhub: prefix ---
|
|
if name.startswith("clawhub:"):
|
|
skill_name = name[8:]
|
|
_validate_skill_name(skill_name)
|
|
_install_hub(skill_name, provider="clawhub")
|
|
return
|
|
|
|
# --- owner/repo or owner/repo#subpath shorthand ---
|
|
if re.match(r"^[a-zA-Z0-9_\-]+/[a-zA-Z0-9_.\-]+(?:#.+)?$", name):
|
|
subpath = None
|
|
spec = name
|
|
if "#" in spec:
|
|
spec, subpath = spec.split("#", 1)
|
|
_install_github(spec, subpath=subpath)
|
|
return
|
|
|
|
# --- Fallback: Skill Hub by name ---
|
|
_validate_skill_name(name)
|
|
_install_hub(name)
|
|
|
|
|
|
def _install_hub(name, provider=None):
|
|
"""Install a skill from Skill Hub."""
|
|
skills_dir = get_skills_dir()
|
|
os.makedirs(skills_dir, exist_ok=True)
|
|
|
|
click.echo(f"Fetching skill info for '{name}'...")
|
|
|
|
try:
|
|
body = {}
|
|
if provider:
|
|
body["provider"] = provider
|
|
resp = requests.post(
|
|
f"{SKILL_HUB_API}/skills/{name}/download",
|
|
json=body,
|
|
timeout=15,
|
|
)
|
|
resp.raise_for_status()
|
|
except requests.HTTPError as e:
|
|
if e.response is not None and e.response.status_code == 404:
|
|
click.echo(f"Error: Skill '{name}' not found on Skill Hub.", err=True)
|
|
else:
|
|
click.echo(f"Error: Failed to fetch skill: {e}", err=True)
|
|
sys.exit(1)
|
|
except Exception as e:
|
|
click.echo(f"Error: Failed to connect to Skill Hub: {e}", err=True)
|
|
sys.exit(1)
|
|
|
|
content_type = resp.headers.get("Content-Type", "")
|
|
|
|
if "application/json" in content_type:
|
|
data = resp.json()
|
|
source_type = data.get("source_type")
|
|
|
|
if source_type == "github":
|
|
source_url = data.get("source_url", "")
|
|
parsed_url = _parse_github_url(source_url)
|
|
if parsed_url:
|
|
owner, repo, branch, subpath = parsed_url
|
|
click.echo(f"Source: GitHub ({source_url})")
|
|
_install_github(f"{owner}/{repo}", subpath=subpath, skill_name=name, branch=branch)
|
|
else:
|
|
_validate_github_spec(source_url)
|
|
click.echo(f"Source: GitHub ({source_url})")
|
|
_install_github(source_url, skill_name=name)
|
|
return
|
|
|
|
if source_type == "registry":
|
|
download_url = data.get("download_url")
|
|
if download_url:
|
|
parsed = urlparse(download_url)
|
|
if parsed.scheme != "https":
|
|
click.echo(f"Error: Refusing to download from non-HTTPS URL.", err=True)
|
|
sys.exit(1)
|
|
provider = data.get("source_provider", "registry")
|
|
expected_checksum = data.get("checksum") or data.get("sha256")
|
|
click.echo(f"Source: {provider}")
|
|
click.echo("Downloading skill package...")
|
|
try:
|
|
dl_resp = requests.get(download_url, timeout=60, allow_redirects=True)
|
|
dl_resp.raise_for_status()
|
|
except Exception as e:
|
|
click.echo(f"Error: Failed to download from {provider}: {e}", err=True)
|
|
sys.exit(1)
|
|
_verify_checksum(dl_resp.content, expected_checksum)
|
|
_install_zip_bytes(dl_resp.content, name, skills_dir)
|
|
_register_installed_skill(name, source=provider)
|
|
_print_install_success(name, provider)
|
|
else:
|
|
click.echo(f"Error: Unsupported registry provider.", err=True)
|
|
sys.exit(1)
|
|
return
|
|
|
|
if "redirect" in data:
|
|
source_url = data.get("source_url", "")
|
|
parsed_url = _parse_github_url(source_url)
|
|
if parsed_url:
|
|
owner, repo, branch, subpath = parsed_url
|
|
click.echo(f"Source: GitHub ({source_url})")
|
|
_install_github(f"{owner}/{repo}", subpath=subpath, skill_name=name, branch=branch)
|
|
else:
|
|
_validate_github_spec(source_url)
|
|
click.echo(f"Source: GitHub ({source_url})")
|
|
_install_github(source_url, skill_name=name)
|
|
return
|
|
|
|
elif "application/zip" in content_type:
|
|
click.echo("Downloading skill package...")
|
|
expected_checksum = resp.headers.get("X-Checksum-Sha256")
|
|
_verify_checksum(resp.content, expected_checksum)
|
|
_install_zip_bytes(resp.content, name, skills_dir)
|
|
_register_installed_skill(name)
|
|
_print_install_success(name, "cowhub")
|
|
return
|
|
|
|
click.echo(f"Error: Unexpected response from Skill Hub.", err=True)
|
|
sys.exit(1)
|
|
|
|
|
|
def _install_github(spec, subpath=None, skill_name=None, branch="main", source="github"):
|
|
"""Install skill(s) from a GitHub repo.
|
|
|
|
Strategy: zip download first (no API rate limit), Contents API as fallback.
|
|
|
|
When subpath is given, install that single skill directory.
|
|
When subpath is None, scan the repo for all skills/ subdirectories containing
|
|
SKILL.md and batch-install them.
|
|
|
|
spec format: owner/repo or owner/repo#path
|
|
"""
|
|
if "#" in spec and not subpath:
|
|
spec, subpath = spec.split("#", 1)
|
|
|
|
_validate_github_spec(spec)
|
|
|
|
skills_dir = get_skills_dir()
|
|
os.makedirs(skills_dir, exist_ok=True)
|
|
owner, repo = spec.split("/", 1)
|
|
|
|
# --- Primary: zip download (no rate limit) ---
|
|
click.echo(f"Downloading from GitHub: {spec} (branch: {branch})...")
|
|
|
|
tmp_dir = None
|
|
repo_root = None
|
|
try:
|
|
tmp_dir, repo_root = _download_repo_zip(spec, branch)
|
|
except Exception:
|
|
click.echo("Zip download failed, falling back to Contents API...")
|
|
|
|
if repo_root:
|
|
try:
|
|
_install_from_repo_root(repo_root, spec, subpath, skill_name, skills_dir, source)
|
|
return
|
|
except SystemExit:
|
|
raise
|
|
except Exception as e:
|
|
click.echo(f"Error processing zip: {e}", err=True)
|
|
click.echo("Falling back to Contents API...")
|
|
finally:
|
|
if tmp_dir:
|
|
shutil.rmtree(tmp_dir, ignore_errors=True)
|
|
|
|
# --- Fallback: Contents API (rate-limited but works for single skill) ---
|
|
if not subpath:
|
|
click.echo("Error: Zip download failed and batch install requires zip.", err=True)
|
|
click.echo(f" Try again or specify a subpath: cow skill install {spec}#skills/<name>", err=True)
|
|
sys.exit(1)
|
|
|
|
if not skill_name:
|
|
skill_name = subpath.rstrip("/").split("/")[-1]
|
|
_validate_skill_name(skill_name)
|
|
|
|
click.echo(f"Downloading via Contents API: {spec}/{subpath} ...")
|
|
target_dir = os.path.join(skills_dir, skill_name)
|
|
try:
|
|
with tempfile.TemporaryDirectory() as api_tmp:
|
|
api_dest = os.path.join(api_tmp, skill_name)
|
|
os.makedirs(api_dest)
|
|
_download_github_dir(owner, repo, branch, subpath.strip("/"), api_dest)
|
|
if os.path.exists(target_dir):
|
|
shutil.rmtree(target_dir)
|
|
shutil.copytree(api_dest, target_dir)
|
|
_register_installed_skill(skill_name, source=source)
|
|
_print_install_success(skill_name, source)
|
|
except Exception as e:
|
|
click.echo(f"Error: Contents API also failed: {e}", err=True)
|
|
sys.exit(1)
|
|
|
|
|
|
def _install_from_repo_root(repo_root, spec, subpath, skill_name, skills_dir, source):
|
|
"""Install skill(s) from an already-extracted repo root directory."""
|
|
if subpath:
|
|
source_dir = os.path.join(repo_root, subpath.strip("/"))
|
|
if not os.path.isdir(source_dir):
|
|
click.echo(f"Error: Path '{subpath}' not found in repository.", err=True)
|
|
sys.exit(1)
|
|
|
|
if not skill_name:
|
|
fm = _parse_skill_frontmatter(
|
|
_read_file_text(os.path.join(source_dir, "SKILL.md"))
|
|
)
|
|
skill_name = fm.get("name") or subpath.rstrip("/").split("/")[-1]
|
|
_validate_skill_name(skill_name)
|
|
|
|
target_dir = os.path.join(skills_dir, skill_name)
|
|
if os.path.exists(target_dir):
|
|
shutil.rmtree(target_dir)
|
|
shutil.copytree(source_dir, target_dir)
|
|
_register_installed_skill(skill_name, source=source)
|
|
_print_install_success(skill_name, source)
|
|
else:
|
|
# Auto-discover all skills in the repo
|
|
discovered = _scan_skills_in_repo(repo_root)
|
|
|
|
if not discovered:
|
|
if skill_name:
|
|
_validate_skill_name(skill_name)
|
|
else:
|
|
skill_name = spec.split("/")[-1]
|
|
_validate_skill_name(skill_name)
|
|
target_dir = os.path.join(skills_dir, skill_name)
|
|
if os.path.exists(target_dir):
|
|
shutil.rmtree(target_dir)
|
|
shutil.copytree(repo_root, target_dir)
|
|
_register_installed_skill(skill_name, source=source)
|
|
_print_install_success(skill_name, source)
|
|
return
|
|
|
|
click.echo(f"Found {len(discovered)} skill(s) in {spec}:\n")
|
|
installed_names = []
|
|
for sname, sdir in discovered:
|
|
safe_name = re.sub(r'[^a-zA-Z0-9_\-]', '-', sname)[:64]
|
|
if not _SAFE_NAME_RE.match(safe_name):
|
|
click.echo(click.style(f" ✗ Skipping '{sname}' (invalid name)", fg="yellow"))
|
|
continue
|
|
target_dir = os.path.join(skills_dir, safe_name)
|
|
if os.path.exists(target_dir):
|
|
shutil.rmtree(target_dir)
|
|
shutil.copytree(sdir, target_dir)
|
|
_register_installed_skill(safe_name, source=source)
|
|
installed_names.append(safe_name)
|
|
|
|
if installed_names:
|
|
click.echo("")
|
|
for n in installed_names:
|
|
_print_install_success(n, source)
|
|
click.echo(f"\n {len(installed_names)} skill(s) installed from {spec}.")
|
|
else:
|
|
click.echo("No valid skills found in repository.")
|
|
|
|
|
|
def _install_gitlab(spec, subpath=None, branch="main"):
|
|
"""Install skill(s) from a GitLab repo via zip download."""
|
|
_validate_github_spec(spec)
|
|
|
|
skills_dir = get_skills_dir()
|
|
os.makedirs(skills_dir, exist_ok=True)
|
|
|
|
click.echo(f"Downloading from GitLab: {spec} (branch: {branch})...")
|
|
|
|
try:
|
|
tmp_dir, repo_root = _download_repo_zip(spec, branch, host="gitlab")
|
|
except Exception as e:
|
|
click.echo(f"Error: Failed to download from GitLab: {e}", err=True)
|
|
sys.exit(1)
|
|
|
|
try:
|
|
_install_from_repo_root(repo_root, spec, subpath, None, skills_dir, "gitlab")
|
|
finally:
|
|
shutil.rmtree(tmp_dir, ignore_errors=True)
|
|
|
|
|
|
def _install_git_clone(git_url: str, display_name: str = ""):
|
|
"""Install skill(s) from any git URL via shallow clone."""
|
|
skills_dir = get_skills_dir()
|
|
os.makedirs(skills_dir, exist_ok=True)
|
|
|
|
click.echo(f"Cloning {display_name or git_url} ...")
|
|
|
|
try:
|
|
tmp_dir, repo_root = _clone_repo(git_url)
|
|
except RuntimeError as e:
|
|
click.echo(f"Error: {e}", err=True)
|
|
sys.exit(1)
|
|
|
|
try:
|
|
_install_from_repo_root(repo_root, display_name or git_url, None, None, skills_dir, "git")
|
|
finally:
|
|
shutil.rmtree(tmp_dir, ignore_errors=True)
|
|
|
|
|
|
def _install_zip_bytes(content, name, skills_dir):
|
|
"""Extract a zip archive into the skills directory."""
|
|
with tempfile.TemporaryDirectory() as tmp_dir:
|
|
zip_path = os.path.join(tmp_dir, "package.zip")
|
|
with open(zip_path, "wb") as f:
|
|
f.write(content)
|
|
|
|
extract_dir = os.path.join(tmp_dir, "extracted")
|
|
with zipfile.ZipFile(zip_path, "r") as zf:
|
|
_safe_extractall(zf, extract_dir)
|
|
|
|
top_items = [d for d in os.listdir(extract_dir) if not d.startswith(".")]
|
|
source = extract_dir
|
|
if len(top_items) == 1 and os.path.isdir(os.path.join(extract_dir, top_items[0])):
|
|
source = os.path.join(extract_dir, top_items[0])
|
|
|
|
target = os.path.join(skills_dir, name)
|
|
if os.path.exists(target):
|
|
shutil.rmtree(target)
|
|
shutil.copytree(source, target)
|
|
|
|
|
|
|
|
|
|
# ------------------------------------------------------------------
|
|
# cow skill uninstall
|
|
# ------------------------------------------------------------------
|
|
@skill.command()
|
|
@click.argument("name")
|
|
@click.option("--yes", "-y", is_flag=True, help="Skip confirmation")
|
|
def uninstall(name, yes):
|
|
"""Uninstall a skill."""
|
|
_validate_skill_name(name)
|
|
skills_dir = get_skills_dir()
|
|
skill_dir = os.path.join(skills_dir, name)
|
|
|
|
if not os.path.exists(skill_dir):
|
|
click.echo(f"Error: Skill '{name}' is not installed.", err=True)
|
|
sys.exit(1)
|
|
|
|
if not yes:
|
|
click.confirm(f"Uninstall skill '{name}'?", abort=True)
|
|
|
|
shutil.rmtree(skill_dir)
|
|
|
|
config_path = os.path.join(skills_dir, "skills_config.json")
|
|
if os.path.exists(config_path):
|
|
try:
|
|
with open(config_path, "r", encoding="utf-8") as f:
|
|
config = json.load(f)
|
|
config.pop(name, None)
|
|
with open(config_path, "w", encoding="utf-8") as f:
|
|
json.dump(config, f, indent=4, ensure_ascii=False)
|
|
except Exception:
|
|
pass
|
|
|
|
click.echo(click.style(f"✓ Skill '{name}' uninstalled.", fg="green"))
|
|
|
|
|
|
# ------------------------------------------------------------------
|
|
# cow skill enable / disable
|
|
# ------------------------------------------------------------------
|
|
@skill.command()
|
|
@click.argument("name")
|
|
def enable(name):
|
|
"""Enable a skill."""
|
|
_set_enabled(name, True)
|
|
|
|
|
|
@skill.command()
|
|
@click.argument("name")
|
|
def disable(name):
|
|
"""Disable a skill."""
|
|
_set_enabled(name, False)
|
|
|
|
|
|
def _set_enabled(name, enabled):
|
|
_validate_skill_name(name)
|
|
skills_dir = get_skills_dir()
|
|
config_path = os.path.join(skills_dir, "skills_config.json")
|
|
|
|
if not os.path.exists(config_path):
|
|
click.echo(f"Error: No skills config found.", err=True)
|
|
sys.exit(1)
|
|
|
|
try:
|
|
with open(config_path, "r", encoding="utf-8") as f:
|
|
config = json.load(f)
|
|
except Exception as e:
|
|
click.echo(f"Error: Failed to read skills config: {e}", err=True)
|
|
sys.exit(1)
|
|
|
|
if name not in config:
|
|
click.echo(f"Error: Skill '{name}' not found in config.", err=True)
|
|
sys.exit(1)
|
|
|
|
config[name]["enabled"] = enabled
|
|
with open(config_path, "w", encoding="utf-8") as f:
|
|
json.dump(config, f, indent=4, ensure_ascii=False)
|
|
|
|
state = "enabled" if enabled else "disabled"
|
|
icon = "✓" if enabled else "✗"
|
|
color = "green" if enabled else "yellow"
|
|
click.echo(click.style(f"{icon} Skill '{name}' {state}.", fg=color))
|
|
|
|
|
|
# ------------------------------------------------------------------
|
|
# cow skill info
|
|
# ------------------------------------------------------------------
|
|
@skill.command()
|
|
@click.argument("name")
|
|
def info(name):
|
|
"""Show details about an installed skill."""
|
|
_validate_skill_name(name)
|
|
skills_dir = get_skills_dir()
|
|
builtin_dir = get_builtin_skills_dir()
|
|
|
|
skill_dir = None
|
|
source = None
|
|
config = load_skills_config()
|
|
for d, src in [(skills_dir, "custom"), (builtin_dir, "builtin")]:
|
|
candidate = os.path.join(d, name)
|
|
if os.path.isdir(candidate):
|
|
skill_dir = candidate
|
|
source = config.get(name, {}).get("source") or src
|
|
break
|
|
|
|
if not skill_dir:
|
|
click.echo(f"Error: Skill '{name}' not found.", err=True)
|
|
sys.exit(1)
|
|
|
|
skill_md = os.path.join(skill_dir, "SKILL.md")
|
|
if not os.path.exists(skill_md):
|
|
click.echo(f"Skill directory: {skill_dir}")
|
|
click.echo("No SKILL.md found.")
|
|
return
|
|
|
|
with open(skill_md, "r", encoding="utf-8") as f:
|
|
content = f.read()
|
|
|
|
config = load_skills_config()
|
|
entry = config.get(name, {})
|
|
enabled = entry.get("enabled", True)
|
|
status_str = click.style("✓ enabled", fg="green") if enabled else click.style("✗ disabled", fg="red")
|
|
|
|
click.echo(f"\n Skill: {name}")
|
|
click.echo(f" Source: {source}")
|
|
click.echo(f" Status: {status_str}")
|
|
click.echo(f" Path: {skill_dir}")
|
|
click.echo(f"\n{'─' * 60}")
|
|
|
|
# Show first ~30 lines of SKILL.md as a preview
|
|
lines = content.split("\n")
|
|
preview = "\n".join(lines[:30])
|
|
click.echo(preview)
|
|
if len(lines) > 30:
|
|
click.echo(f"\n ... ({len(lines) - 30} more lines, see {skill_md})")
|
|
click.echo()
|