226 lines
8.0 KiB
Python
226 lines
8.0 KiB
Python
"""自研 skill 发现与解析。"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import re
|
||
from pathlib import Path
|
||
|
||
from skiff.paths import SKILLS_DIR, TEMPLATE_DIR, ensure_skills_home
|
||
from skiff.registry import external_skill_path, load_registry
|
||
from skiff.sources import list_source_skills, load_sources, source_skills_root
|
||
|
||
|
||
def list_owned_skills() -> list[str]:
|
||
ensure_skills_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 owned_skill_path(name: str) -> Path:
|
||
path = SKILLS_DIR / name
|
||
if not (path / "SKILL.md").is_file():
|
||
raise SystemExit(f"自研 skill 不存在: {name}")
|
||
return path
|
||
|
||
|
||
def split_skill_spec(spec: str, source: str | None = None) -> tuple[str, str | None]:
|
||
if "/" not in spec:
|
||
return spec, 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）")
|
||
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_skills_home()
|
||
name, source = split_skill_spec(name, source)
|
||
|
||
owned = SKILLS_DIR / name
|
||
if source == "owned":
|
||
if not (owned / "SKILL.md").is_file():
|
||
raise SystemExit(f"owned source 中找不到 skill: {name}")
|
||
return owned, "owned"
|
||
|
||
registry = load_registry()
|
||
if source == "registry":
|
||
if name not in registry:
|
||
raise SystemExit(f"registry 中找不到 skill: {name}")
|
||
path = external_skill_path(name, registry[name])
|
||
if not path.exists():
|
||
raise SystemExit(
|
||
f"外部 skill {name!r} 尚未 fetch。请先运行: skiff fetch {name}"
|
||
)
|
||
return path, "registry"
|
||
|
||
sources = load_sources()
|
||
if source:
|
||
if source not in sources:
|
||
raise SystemExit(
|
||
f"项目依赖 source {source!r}，但本机尚未配置。"
|
||
f"请运行: skiff source add {source} <repo>"
|
||
)
|
||
path = source_skills_root(source, sources[source]) / name
|
||
if not (path / "SKILL.md").is_file():
|
||
raise SystemExit(f"source {source!r} 中找不到 skill: {name}")
|
||
return path, source
|
||
|
||
candidates: list[tuple[Path, str]] = []
|
||
if (owned / "SKILL.md").is_file():
|
||
candidates.append((owned, "owned"))
|
||
if name in registry:
|
||
candidates.append((external_skill_path(name, registry[name]), "registry"))
|
||
for source_name, entry in sources.items():
|
||
path = source_skills_root(source_name, entry) / name
|
||
if (path / "SKILL.md").is_file():
|
||
candidates.append((path, source_name))
|
||
|
||
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 == "registry" and not path.exists():
|
||
raise SystemExit(f"外部 skill {name!r} 尚未 fetch。请先运行: skiff fetch {name}")
|
||
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"
|
||
if not skill_md.is_file():
|
||
return ["缺少 SKILL.md"]
|
||
|
||
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
|