laily
f3cd56b78e
Use ~/.pouch, the pouch CLI, and .pouch.yaml as the SSOT container. Keep the inner skills/ packages, and store ACK project state in .pouch/ack instead of docs/ack.
294 lines
11 KiB
Python
294 lines
11 KiB
Python
"""自研 skill 发现与解析。"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import re
|
||
from pathlib import Path
|
||
|
||
from pouch.paths import SKILLS_DIR, TEMPLATE_DIR, ensure_pouch_home
|
||
from pouch.catalog import (
|
||
catalog_skill_path,
|
||
discover_catalog_skills,
|
||
load_catalog,
|
||
)
|
||
from pouch.sources import (
|
||
discover_source_skills,
|
||
list_source_skills,
|
||
load_sources,
|
||
)
|
||
|
||
|
||
def list_builtin_skills() -> list[str]:
|
||
ensure_pouch_home()
|
||
if not SKILLS_DIR.is_dir():
|
||
return []
|
||
names: list[str] = []
|
||
for entry in sorted(SKILLS_DIR.iterdir()):
|
||
if not entry.is_dir():
|
||
continue
|
||
if entry.name.startswith("_"):
|
||
continue
|
||
if (entry / "SKILL.md").is_file():
|
||
names.append(entry.name)
|
||
return names
|
||
|
||
|
||
def builtin_skill_path(name: str) -> Path:
|
||
path = SKILLS_DIR / name
|
||
if not (path / "SKILL.md").is_file():
|
||
raise SystemExit(f"builtin skill 不存在: {name}")
|
||
return path
|
||
|
||
|
||
def normalize_source(source: str | None) -> str | None:
|
||
if source == "owned":
|
||
return "builtin"
|
||
if source == "registry":
|
||
return "catalog"
|
||
if source and source.startswith("registry:"):
|
||
return f"catalog:{source.split(':', 1)[1]}"
|
||
return source
|
||
|
||
|
||
def split_skill_spec(spec: str, source: str | None = None) -> tuple[str, str | None]:
|
||
if "/" not in spec:
|
||
return spec, normalize_source(source)
|
||
qualified_source, name = spec.split("/", 1)
|
||
if not qualified_source or not name or "/" in name:
|
||
raise SystemExit(f"skill 限定名称无效: {spec!r}（应为 source/name）")
|
||
qualified_source = normalize_source(qualified_source)
|
||
source = normalize_source(source)
|
||
if source and source != qualified_source:
|
||
raise SystemExit(
|
||
f"skill 来源冲突: {spec!r} 与 --source {source!r} 不一致"
|
||
)
|
||
return name, qualified_source
|
||
|
||
|
||
def list_custom_skills(source: str | None = None) -> dict[str, list[str]]:
|
||
sources = load_sources()
|
||
if source:
|
||
if source not in sources:
|
||
raise SystemExit(f"未配置 source: {source}")
|
||
return {source: list_source_skills(source, sources[source])}
|
||
return {name: list_source_skills(name, entry) for name, entry in sources.items()}
|
||
|
||
|
||
def resolve_skill_source(name: str, *, source: str | None = None) -> tuple[Path, str]:
|
||
"""返回 (skill_path, source)，未指定来源时拒绝同名歧义。"""
|
||
ensure_pouch_home()
|
||
name, source = split_skill_spec(name, source)
|
||
|
||
builtin = SKILLS_DIR / name
|
||
if source == "builtin":
|
||
if not (builtin / "SKILL.md").is_file():
|
||
raise SystemExit(f"builtin source 中找不到 skill: {name}")
|
||
return builtin, "builtin"
|
||
|
||
catalog = load_catalog()
|
||
sources = load_sources()
|
||
if source and source.startswith("catalog:"):
|
||
provider = source.split(":", 1)[1]
|
||
if provider not in catalog:
|
||
raise SystemExit(f"catalog 中找不到 source: {provider}")
|
||
skills = discover_catalog_skills(provider, catalog[provider])
|
||
if name not in skills:
|
||
available = ", ".join(skills) or "(无)"
|
||
raise SystemExit(
|
||
f"catalog source {provider!r} 中找不到 skill {name!r}。可用: {available}"
|
||
)
|
||
return skills[name], f"catalog:{provider}"
|
||
|
||
if source == "catalog":
|
||
if name not in catalog:
|
||
raise SystemExit(f"catalog 中找不到 skill source: {name}")
|
||
path = catalog_skill_path(name, catalog[name])
|
||
if not (path / "SKILL.md").is_file():
|
||
raise SystemExit(
|
||
f"catalog skill {name!r} 尚未 fetch 或 path 中缺少 SKILL.md。"
|
||
f"请运行: pouch fetch {name}"
|
||
)
|
||
return path, f"catalog:{name}"
|
||
|
||
if source:
|
||
if source not in sources:
|
||
raise SystemExit(
|
||
f"项目依赖 source {source!r}，但本机尚未配置。"
|
||
f"请运行: pouch source add {source} <repo>"
|
||
)
|
||
skills = discover_source_skills(source, sources[source])
|
||
if name not in skills:
|
||
raise SystemExit(f"source {source!r} 中找不到 skill: {name}")
|
||
return skills[name], source
|
||
|
||
candidates: list[tuple[Path, str]] = []
|
||
if (builtin / "SKILL.md").is_file():
|
||
candidates.append((builtin, "builtin"))
|
||
if name in catalog:
|
||
path = catalog_skill_path(name, catalog[name])
|
||
if (path / "SKILL.md").is_file() or not path.exists():
|
||
candidates.append((path, f"catalog:{name}"))
|
||
for provider, entry in catalog.items():
|
||
if provider == name:
|
||
continue
|
||
skills = discover_catalog_skills(provider, entry)
|
||
if name in skills:
|
||
candidates.append((skills[name], f"catalog:{provider}"))
|
||
for source_name, entry in sources.items():
|
||
skills = discover_source_skills(source_name, entry)
|
||
if name in skills:
|
||
candidates.append((skills[name], source_name))
|
||
|
||
unique_candidates: list[tuple[Path, str]] = []
|
||
seen_paths: set[Path] = set()
|
||
for path, candidate_source in candidates:
|
||
resolved = path.resolve()
|
||
if resolved in seen_paths:
|
||
continue
|
||
seen_paths.add(resolved)
|
||
unique_candidates.append((path, candidate_source))
|
||
candidates = unique_candidates
|
||
|
||
if len(candidates) > 1:
|
||
choices = ", ".join(f"{candidate_source}/{name}" for _, candidate_source in candidates)
|
||
raise SystemExit(f"skill 名称存在多个来源，请明确指定: {choices}")
|
||
if candidates:
|
||
path, resolved_source = candidates[0]
|
||
if resolved_source.startswith("catalog:") and not path.exists():
|
||
provider = resolved_source.split(":", 1)[1]
|
||
raise SystemExit(f"catalog source {provider!r} 尚未 fetch。请先运行: pouch fetch {provider}")
|
||
return path, resolved_source
|
||
|
||
raise SystemExit(f"找不到 skill: {name}")
|
||
|
||
|
||
def read_skill_meta(skill_dir: Path) -> dict[str, str]:
|
||
skill_md = skill_dir / "SKILL.md"
|
||
if not skill_md.is_file():
|
||
return {}
|
||
text = skill_md.read_text(encoding="utf-8")
|
||
if not text.startswith("---"):
|
||
return {}
|
||
end = text.find("\n---", 3)
|
||
if end == -1:
|
||
return {}
|
||
frontmatter = text[3:end]
|
||
meta: dict[str, str] = {}
|
||
|
||
name_match = re.search(r"^name:\s*(.+)$", frontmatter, re.MULTILINE)
|
||
if name_match:
|
||
meta["name"] = name_match.group(1).strip().strip("\"'")
|
||
|
||
desc_match = re.search(
|
||
r"^description:\s*(?:>-|>\||>|-)?\s*\n((?:[ \t].+\n?)+)",
|
||
frontmatter,
|
||
re.MULTILINE,
|
||
)
|
||
if desc_match:
|
||
lines = [line.strip() for line in desc_match.group(1).splitlines() if line.strip()]
|
||
meta["description"] = " ".join(lines)
|
||
else:
|
||
inline = re.search(r"^description:\s*(.+)$", frontmatter, re.MULTILINE)
|
||
if inline:
|
||
meta["description"] = inline.group(1).strip().strip("\"'")
|
||
|
||
return meta
|
||
|
||
|
||
def skill_description(name: str) -> str | None:
|
||
meta = read_skill_meta(SKILLS_DIR / name)
|
||
desc = meta.get("description")
|
||
return desc.strip() if desc else None
|
||
|
||
|
||
def validate_skill_name(name: str) -> None:
|
||
import re
|
||
|
||
if not re.fullmatch(r"[a-z0-9]+(?:-[a-z0-9]+)*", name):
|
||
raise SystemExit(
|
||
f"skill 名称无效: {name!r}（小写 + 连字符，如 security-review）"
|
||
)
|
||
if name == "_template":
|
||
raise SystemExit("不能使用保留名 _template")
|
||
|
||
|
||
_PLACEHOLDER_PATTERNS = (
|
||
(r"^name:\s*skill-name\s*$", "skill-name"),
|
||
(r"^#\s+Skill 名称\s*$", "Skill 名称"),
|
||
(r"简要描述 skill 做什么、何时触发", "简要描述 skill 做什么"),
|
||
(r"^-\s*触发场景 1\s*$", "触发场景 1"),
|
||
(r"^\d+\.\s*第一步\s*$", "第一步"),
|
||
)
|
||
|
||
|
||
def validate_skill_dir(skill_dir: Path, expected_name: str) -> list[str]:
|
||
"""返回 skill 目录中的校验问题；空列表表示通过。"""
|
||
issues: list[str] = []
|
||
skill_md = skill_dir / "SKILL.md"
|
||
readme = skill_dir / "README.md"
|
||
if not skill_md.is_file():
|
||
return ["缺少 SKILL.md"]
|
||
if not readme.is_file():
|
||
issues.append("缺少 README.md（面向人类的使用说明）")
|
||
else:
|
||
readme_text = readme.read_text(encoding="utf-8")
|
||
if not readme_text.strip():
|
||
issues.append("README.md 为空")
|
||
if re.search(r"^#\s+skill-name\s*$", readme_text, re.MULTILINE):
|
||
issues.append("README.md 存在模板占位内容: skill-name")
|
||
for placeholder in (
|
||
"用一句话告诉使用者这个 skill 能解决什么问题",
|
||
"给出用户可以直接说出的典型请求",
|
||
"给出一条可以直接交给 Agent 的示例请求",
|
||
):
|
||
if placeholder in readme_text:
|
||
issues.append(f"README.md 存在模板占位内容: {placeholder}")
|
||
|
||
text = skill_md.read_text(encoding="utf-8")
|
||
if not text.strip():
|
||
return ["SKILL.md 为空"]
|
||
if not text.startswith("---\n"):
|
||
return ["SKILL.md 缺少 YAML frontmatter"]
|
||
|
||
end = text.find("\n---", 4)
|
||
if end == -1:
|
||
return ["SKILL.md frontmatter 未闭合"]
|
||
frontmatter = text[4:end]
|
||
keys = re.findall(r"^([A-Za-z0-9_-]+):", frontmatter, re.MULTILINE)
|
||
unexpected = sorted(set(keys) - {"name", "description"})
|
||
missing = sorted({"name", "description"} - set(keys))
|
||
if missing:
|
||
issues.append(f"frontmatter 缺少字段: {', '.join(missing)}")
|
||
if unexpected:
|
||
issues.append(f"frontmatter 只允许 name、description，发现: {', '.join(unexpected)}")
|
||
|
||
meta = read_skill_meta(skill_dir)
|
||
if meta.get("name") != expected_name:
|
||
issues.append(
|
||
f"目录名与 frontmatter name 不一致: {expected_name} != {meta.get('name', '(缺失)')}"
|
||
)
|
||
description = meta.get("description", "").strip()
|
||
if not description:
|
||
issues.append("description 不能为空")
|
||
|
||
found_placeholders = [
|
||
label
|
||
for pattern, label in _PLACEHOLDER_PATTERNS
|
||
if re.search(pattern, text, re.MULTILINE | re.IGNORECASE)
|
||
]
|
||
if found_placeholders:
|
||
issues.append(f"存在模板占位内容: {', '.join(found_placeholders)}")
|
||
|
||
link_pattern = re.compile(r"!?\[[^\]]*\]\(([^)]+)\)")
|
||
for target in link_pattern.findall(text):
|
||
target = target.strip().split("#", 1)[0]
|
||
if not target or "://" in target or target.startswith(("mailto:", "/")):
|
||
continue
|
||
if not (skill_dir / target).resolve().is_file():
|
||
issues.append(f"引用文件不存在: {target}")
|
||
|
||
for draft_file in ("brief.yaml",):
|
||
if (skill_dir / draft_file).exists():
|
||
issues.append(f"正式 skill 不应包含草稿文件: {draft_file}")
|
||
return issues
|