registry ingest test scaffolding

This commit is contained in:
John Lancaster
2026-06-21 12:58:42 -05:00
parent aec3500370
commit 29130c3a0c
5 changed files with 127 additions and 0 deletions
+57
View File
@@ -0,0 +1,57 @@
from collections.abc import Iterable
from dataclasses import dataclass
from fnmatch import fnmatch
from importlib.resources.abc import Traversable
from itertools import groupby
from itertools import starmap
from typing import Self
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"))
groups = {
k: tuple(v)
for k, v in groupby(
paths,
key=lambda p: fnmatch(p.relpath.as_posix(), f"skills/{slug}/references/*.md"),
)
}
references = groups.get(True, ())
other = tuple(p for p in groups.get(False, ()) if 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}