126 lines
4.6 KiB
Python
126 lines
4.6 KiB
Python
import re
|
|
from collections.abc import Iterable
|
|
from collections.abc import Mapping
|
|
from dataclasses import dataclass
|
|
from importlib.resources.abc import Traversable
|
|
from itertools import groupby
|
|
from itertools import starmap
|
|
from pathlib import PurePosixPath
|
|
from typing import Self
|
|
|
|
from personal_mcp.registry.models.common import SKILL_ID_RE
|
|
from personal_mcp.registry.models.common import DocsPath
|
|
from personal_mcp.registry.models.common import ReferenceEntry
|
|
from personal_mcp.registry.models.skill import SkillFrontmatter
|
|
from personal_mcp.registry.models.skill import StoredSkill
|
|
from personal_mcp.registry.models.skill import StoredSkillReference
|
|
|
|
from .document import MarkdownDocument
|
|
|
|
|
|
@dataclass(frozen=True, slots=True)
|
|
class SkillFilesBundle:
|
|
"""Represents a skill and all of its associated markdown files."""
|
|
|
|
slug: str
|
|
skill: MarkdownDocument
|
|
references: tuple[MarkdownDocument, ...]
|
|
other: tuple[MarkdownDocument, ...]
|
|
|
|
@classmethod
|
|
def from_root(cls, root: Traversable) -> list[Self]:
|
|
# Should only be used for testing
|
|
return list(cls.from_docs(MarkdownDocument.from_root(root).values()))
|
|
|
|
@classmethod
|
|
def from_docs(cls, docs: Iterable[MarkdownDocument]) -> tuple[Self, ...]:
|
|
return tuple(starmap(cls.from_paths, group_skill_paths(docs).items()))
|
|
|
|
@classmethod
|
|
def from_paths(cls, slug: str, paths: set[MarkdownDocument]) -> Self:
|
|
skill = next(iter(p for p in paths if p.relpath.name == "SKILL.md"))
|
|
sorted_paths = tuple(sorted(paths, key=lambda p: p.relpath))
|
|
references_dir = PurePosixPath("skills", slug, "references")
|
|
references = tuple(p for p in sorted_paths if p.relpath.parent == references_dir)
|
|
other = tuple(p for p in sorted_paths if p not in references and p != skill)
|
|
return cls(
|
|
slug=slug,
|
|
skill=skill,
|
|
references=references,
|
|
other=other,
|
|
)
|
|
|
|
|
|
def group_skill_paths(docs: Iterable[MarkdownDocument]) -> dict[str, set[MarkdownDocument]]:
|
|
"""Group skills from a list of markdown documents by their skill slug."""
|
|
s = sorted(
|
|
filter(lambda d: d.skill_slug is not None, docs),
|
|
key=lambda d: (d.skill_slug or "", d.relpath.stem),
|
|
)
|
|
grouped = groupby(s, key=lambda doc: doc.skill_slug)
|
|
return {k: set(g) for k, g in grouped if k}
|
|
|
|
|
|
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 or not SKILL_ID_RE.fullmatch(normalized):
|
|
return None
|
|
return normalized
|
|
|
|
|
|
def _discover_reference_entries(bundle: SkillFilesBundle) -> dict[str, ReferenceEntry]:
|
|
discovered: dict[str, ReferenceEntry] = {}
|
|
for reference_doc in bundle.references:
|
|
ref_id = _reference_id_from_filename(reference_doc.relpath.name)
|
|
if ref_id is None:
|
|
continue
|
|
discovered[ref_id] = ReferenceEntry(
|
|
path=PurePosixPath("references", reference_doc.relpath.name),
|
|
title=_title_from_reference_filename(reference_doc.relpath.name),
|
|
)
|
|
return discovered
|
|
|
|
|
|
def build_stored_skill(
|
|
*,
|
|
bundle: SkillFilesBundle,
|
|
docs_by_relpath: Mapping[DocsPath, MarkdownDocument],
|
|
) -> StoredSkill:
|
|
frontmatter = SkillFrontmatter.from_raw_yaml(bundle.skill.frontmatter)
|
|
metadata = frontmatter.x_personal_mcp
|
|
merged_entries = _discover_reference_entries(bundle)
|
|
merged_entries.update(dict(metadata.references))
|
|
|
|
references: dict[str, StoredSkillReference] = {}
|
|
for ref_id, entry in sorted(merged_entries.items()):
|
|
ref_relpath = PurePosixPath("skills", bundle.slug, entry.path)
|
|
if ref_relpath not in docs_by_relpath:
|
|
raise KeyError(f"reference document not found for '{metadata.id}:{ref_id}' at {ref_relpath.as_posix()}")
|
|
ref_doc = docs_by_relpath[ref_relpath]
|
|
references[ref_id] = StoredSkillReference(
|
|
ref_id=ref_id,
|
|
relpath=ref_relpath,
|
|
content=ref_doc.content,
|
|
entry=entry,
|
|
)
|
|
|
|
return StoredSkill.model_validate(
|
|
{
|
|
"skill_id": metadata.id,
|
|
"relpath": bundle.skill.relpath,
|
|
"content": bundle.skill.content,
|
|
"frontmatter": frontmatter,
|
|
"references": references,
|
|
}
|
|
)
|