126 lines
4.5 KiB
Python
126 lines
4.5 KiB
Python
from __future__ import annotations
|
|
|
|
import re
|
|
from importlib.resources.abc import Traversable
|
|
from pathlib import PurePosixPath
|
|
from typing import Any
|
|
|
|
import yaml
|
|
|
|
from ..registry.models.common import SKILL_ID_RE
|
|
from ..registry.models.common import ReferenceEntry
|
|
from ..registry.models.prompt import PromptFrontmatter
|
|
from ..registry.models.skill import SkillFrontmatter
|
|
|
|
|
|
def _parse_frontmatter(markdown: str, *, path: str) -> tuple[dict[str, Any], str]:
|
|
if not markdown.startswith("---"):
|
|
raise ValueError(f"missing YAML frontmatter: {path}")
|
|
|
|
lines = markdown.splitlines()
|
|
if len(lines) < 3 or lines[0].strip() != "---":
|
|
raise ValueError(f"invalid YAML frontmatter start: {path}")
|
|
|
|
end_index: int | None = None
|
|
for i in range(1, len(lines)):
|
|
if lines[i].strip() == "---":
|
|
end_index = i
|
|
break
|
|
|
|
if end_index is None:
|
|
raise ValueError(f"missing YAML frontmatter terminator: {path}")
|
|
|
|
raw_yaml = "\n".join(lines[1:end_index])
|
|
body = "\n".join(lines[end_index + 1 :])
|
|
parsed = yaml.safe_load(raw_yaml)
|
|
if not isinstance(parsed, dict):
|
|
raise TypeError(f"frontmatter must parse to an object: {path}")
|
|
return parsed, body
|
|
|
|
|
|
def _walk_markdown(
|
|
node: Traversable,
|
|
*,
|
|
prefix: PurePosixPath | None = None,
|
|
) -> list[tuple[str, Traversable]]:
|
|
prefix = PurePosixPath() if prefix is None else prefix
|
|
results: list[tuple[str, Traversable]] = []
|
|
for child in sorted(node.iterdir(), key=lambda item: item.name):
|
|
relpath = prefix.joinpath(child.name)
|
|
if child.is_dir():
|
|
results.extend(_walk_markdown(child, prefix=relpath))
|
|
continue
|
|
if not child.is_file() or not child.name.lower().endswith(".md"):
|
|
continue
|
|
results.append((relpath.as_posix(), child))
|
|
return results
|
|
|
|
|
|
def _validate_skill_frontmatter(raw: dict[str, Any], *, skill_dir_name: str) -> SkillFrontmatter:
|
|
model = SkillFrontmatter.model_validate(raw)
|
|
if model.name != skill_dir_name:
|
|
raise ValueError("frontmatter name must exactly match skill directory name")
|
|
if model.x_personal_mcp.id != model.name:
|
|
raise ValueError("x-personal-mcp.id must exactly match name")
|
|
expected_capability = f"resource://skills/{model.name}/document"
|
|
if expected_capability not in model.x_personal_mcp.capabilities:
|
|
raise ValueError(f"capabilities must include {expected_capability}")
|
|
return model
|
|
|
|
|
|
def _validate_prompt_frontmatter(raw: dict[str, Any], *, prompt_dir_name: str) -> PromptFrontmatter:
|
|
model = PromptFrontmatter.model_validate(raw)
|
|
if model.name != prompt_dir_name:
|
|
raise ValueError("frontmatter name must exactly match prompt directory name")
|
|
if model.x_personal_mcp.id != model.name:
|
|
raise ValueError("x-personal-mcp.id must exactly match name")
|
|
expected_capability = f"resource://prompts/{model.name}/document"
|
|
if expected_capability not in model.x_personal_mcp.capabilities:
|
|
raise ValueError(f"capabilities must include {expected_capability}")
|
|
return model
|
|
|
|
|
|
def _title_from_reference_filename(filename: str) -> str:
|
|
stem = PurePosixPath(filename).stem
|
|
normalized = stem.replace("-", " ").replace("_", " ").split()
|
|
if not normalized:
|
|
return stem
|
|
return " ".join(token.capitalize() for token in normalized)
|
|
|
|
|
|
def _reference_id_from_filename(filename: str) -> str | None:
|
|
stem = PurePosixPath(filename).stem.strip().lower().replace("_", "-")
|
|
normalized = re.sub(r"[^a-z0-9-]+", "-", stem)
|
|
normalized = re.sub(r"-+", "-", normalized).strip("-")
|
|
if not normalized:
|
|
return None
|
|
if not SKILL_ID_RE.fullmatch(normalized):
|
|
return None
|
|
return normalized
|
|
|
|
|
|
def _discover_top_level_references(
|
|
*,
|
|
skill_dir: Traversable,
|
|
) -> dict[str, ReferenceEntry]:
|
|
references_dir = skill_dir.joinpath("references")
|
|
if not references_dir.is_dir():
|
|
return {}
|
|
|
|
discovered: dict[str, ReferenceEntry] = {}
|
|
for child in sorted(references_dir.iterdir(), key=lambda item: item.name):
|
|
if child.is_dir() or not child.is_file():
|
|
continue
|
|
if not child.name.lower().endswith(".md"):
|
|
continue
|
|
|
|
ref_id = _reference_id_from_filename(child.name)
|
|
if ref_id is None:
|
|
continue
|
|
|
|
discovered[ref_id] = ReferenceEntry(
|
|
path=PurePosixPath("references").joinpath(child.name).as_posix(),
|
|
title=_title_from_reference_filename(child.name),
|
|
)
|
|
return discovered
|