119 lines
4.9 KiB
Python
119 lines
4.9 KiB
Python
"""Skills Scanner — Scan skills/ directory and generate SKILLS_SNAPSHOT.md."""
|
|
|
|
from hashlib import sha256
|
|
from pathlib import Path
|
|
import re
|
|
|
|
import yaml
|
|
|
|
|
|
MAX_SKILL_BYTES = 64 * 1024
|
|
|
|
|
|
def parse_skill(content: str) -> dict:
|
|
"""Validate the small, declarative SKILL.md format used by the workbench."""
|
|
if not isinstance(content, str) or not content.startswith("---") or len(content.encode("utf-8")) > MAX_SKILL_BYTES:
|
|
raise ValueError("SKILL.md 缺少 YAML 头或超过 64 KB")
|
|
parts = content.split("---", 2)
|
|
meta = yaml.safe_load(parts[1]) if len(parts) == 3 else None
|
|
if not isinstance(meta, dict) or not isinstance(meta.get("name"), str) or not meta["name"].strip():
|
|
raise ValueError("SKILL.md 必须声明 name")
|
|
if not isinstance(meta.get("description"), str) or not meta["description"].strip():
|
|
raise ValueError("SKILL.md 必须声明 description")
|
|
details = meta.get("metadata", {})
|
|
if not isinstance(details, dict):
|
|
raise ValueError("SKILL.md 的 metadata 必须是对象")
|
|
for field in ("inputs", "audience"):
|
|
value = meta.get(field, details.get(field))
|
|
if value is not None and (not isinstance(value, list) or not all(isinstance(item, str) for item in value)):
|
|
raise ValueError(f"SKILL.md 的 {field} 必须是文本列表")
|
|
return meta
|
|
|
|
|
|
def skill_attributes(meta: dict) -> dict:
|
|
return {**meta, **meta.get("metadata", {})}
|
|
|
|
|
|
def skill_slug(meta: dict) -> str:
|
|
slug = re.sub(r"[^a-z0-9-]+", "-", meta["name"].strip().lower().replace("_", "-")).strip("-")
|
|
if not slug or len(slug) > 64:
|
|
raise ValueError("Skill 名称需使用 1–64 位英文小写字母、数字或连字符")
|
|
return slug
|
|
|
|
|
|
def install_skill(base_dir: Path, content: str) -> dict:
|
|
"""Install one uploaded SKILL.md under the managed imported-skills directory."""
|
|
meta = parse_skill(content)
|
|
slug = skill_slug(meta)
|
|
root = (base_dir / "skills").resolve()
|
|
target = (root / "imported" / slug / "SKILL.md").resolve()
|
|
if not target.is_relative_to(root):
|
|
raise ValueError("Skill 目录不在允许范围内")
|
|
if target.exists():
|
|
raise FileExistsError("同名 Skill 已存在，请先归档旧版本或修改 name")
|
|
target.parent.mkdir(parents=True, exist_ok=False)
|
|
temporary = target.with_suffix(".tmp")
|
|
temporary.write_text(content, encoding="utf-8")
|
|
temporary.replace(target)
|
|
legacy_id = target.relative_to(root).as_posix()
|
|
return {"id": legacy_id, "legacy_id": legacy_id, "name": meta["name"], "description": meta["description"], "meta": skill_attributes(meta)}
|
|
|
|
|
|
def discover_skills(base_dir: Path) -> list[dict]:
|
|
"""Read the native Skills catalog without changing its generated snapshot."""
|
|
root = (base_dir / "skills").resolve()
|
|
entries = []
|
|
for file in sorted(root.rglob("SKILL.md")):
|
|
resolved = file.resolve()
|
|
if not resolved.is_relative_to(root):
|
|
continue
|
|
try:
|
|
content = file.read_text(encoding="utf-8")
|
|
meta = parse_skill(content)
|
|
attributes = skill_attributes(meta)
|
|
if str(attributes.get("status", "published")) == "archived":
|
|
continue
|
|
legacy_id = file.relative_to(root).as_posix()
|
|
entries.append({
|
|
"id": legacy_id,
|
|
"component_id": str(attributes.get("id", legacy_id)),
|
|
"legacy_id": legacy_id,
|
|
"kind": str(attributes.get("kind", "skill")),
|
|
"version": str(attributes.get("version", "legacy")),
|
|
"schema_version": int(attributes.get("schema_version", 1)),
|
|
"status": str(attributes.get("status", "published")),
|
|
"name": str(meta.get("name", file.parent.name)),
|
|
"description": str(meta.get("description", "")),
|
|
"content_hash": sha256(content.encode("utf-8")).hexdigest(),
|
|
"content": content,
|
|
})
|
|
except (OSError, ValueError, TypeError, yaml.YAMLError):
|
|
continue
|
|
return entries
|
|
|
|
|
|
def scan_skills(base_dir: Path) -> str:
|
|
"""Scan all SKILL.md files and generate SKILLS_SNAPSHOT.md."""
|
|
skills_dir = base_dir / "skills"
|
|
snapshot_path = base_dir / "SKILLS_SNAPSHOT.md"
|
|
|
|
if not skills_dir.exists():
|
|
skills_dir.mkdir(parents=True)
|
|
|
|
skills = [{**s, "location":f"./backend/skills/{s['legacy_id']}"} for s in discover_skills(base_dir)]
|
|
|
|
# Build XML-style snapshot
|
|
lines = ["<available_skills>"]
|
|
for s in skills:
|
|
lines.append(" <skill>")
|
|
lines.append(f" <name>{s['name']}</name>")
|
|
lines.append(f" <description>{s['description']}</description>")
|
|
lines.append(f" <location>{s['location']}</location>")
|
|
lines.append(" </skill>")
|
|
lines.append("</available_skills>")
|
|
|
|
snapshot = "\n".join(lines)
|
|
snapshot_path.write_text(snapshot, encoding="utf-8")
|
|
print(f"📋 Skills snapshot: {len(skills)} skills found")
|
|
return snapshot
|