WIP simplifying load/startup
This commit is contained in:
@@ -1,125 +0,0 @@
|
||||
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
|
||||
Reference in New Issue
Block a user