WIP loading

This commit is contained in:
John Lancaster
2026-06-21 11:34:07 -05:00
parent 993dc6a879
commit caa4a5079a
2 changed files with 63 additions and 38 deletions
+21 -10
View File
@@ -1,5 +1,6 @@
from collections.abc import Generator from collections.abc import Generator
from dataclasses import dataclass from dataclasses import dataclass
from dataclasses import field
from importlib.abc import Traversable from importlib.abc import Traversable
from itertools import starmap from itertools import starmap
from pathlib import PurePosixPath from pathlib import PurePosixPath
@@ -7,20 +8,21 @@ from typing import Self
@dataclass(frozen=True, slots=True) @dataclass(frozen=True, slots=True)
class LoadedMarkdownDocument: class MarkdownDocument:
"""Represents a loaded markdown document with its content and frontmatter.""" """Represents a loaded markdown document with its content and frontmatter."""
relpath: PurePosixPath relpath: PurePosixPath
"""The relative path of the document within the package resources.""" """The relative path of the document within the package resources."""
content: str content: str = field(repr=False)
"""The raw markdown content of the document.""" """The raw markdown content of the document."""
frontmatter: str frontmatter: str | None = field(repr=False, default=None)
"""The raw YAML frontmatter of the document, if present.""" """The raw YAML frontmatter of the document, if present."""
@classmethod @classmethod
def from_root(cls, root: Traversable) -> set[Self]: def from_root(cls, root: Traversable):
"""Recursively load all markdown documents from the root resource.""" """Recursively load all markdown documents from the root resource."""
return set(starmap(cls.from_resource, walk_resources(root))) mapped = starmap(cls.from_resource, walk_resources(root))
return {d.relpath: d for d in mapped}
@classmethod @classmethod
def from_resource(cls, relpath: PurePosixPath, resource: Traversable) -> Self: def from_resource(cls, relpath: PurePosixPath, resource: Traversable) -> Self:
@@ -29,13 +31,19 @@ class LoadedMarkdownDocument:
frontmatter = get_raw_frontmatter(raw) frontmatter = get_raw_frontmatter(raw)
return cls(relpath=relpath, content=raw, frontmatter=frontmatter) return cls(relpath=relpath, content=raw, frontmatter=frontmatter)
@property
def skill_slug(self) -> str | None:
parts = self.relpath.parts
if parts[0] == "skills" and len(parts) >= 3:
return parts[1]
def walk_resources( def walk_resources(
node: Traversable, node: Traversable,
*, *,
suffix: str = ".md", suffix: str = ".md",
prefix: PurePosixPath | None = None, prefix: PurePosixPath | None = None,
) -> Generator[tuple[str, Traversable]]: ) -> Generator[tuple[PurePosixPath, Traversable]]:
"""Recursively yield all resources in node, with their full path.""" """Recursively yield all resources in node, with their full path."""
prefix = prefix if prefix is not None else PurePosixPath() prefix = prefix if prefix is not None else PurePosixPath()
for child in sorted(node.iterdir(), key=lambda item: item.name): for child in sorted(node.iterdir(), key=lambda item: item.name):
@@ -45,13 +53,16 @@ def walk_resources(
continue continue
if not child.is_file() or not child.name.lower().endswith(suffix): if not child.is_file() or not child.name.lower().endswith(suffix):
continue continue
yield relpath.as_posix(), child yield relpath, child
def get_raw_frontmatter(raw: str) -> str: def get_raw_frontmatter(raw: str) -> str | None:
delimiter = iter(get_frontmatter_delim_idx(raw, delimiter="---")) delimiter = iter(get_frontmatter_delim_idx(raw, delimiter="---"))
start = next(delimiter) + 1 try:
end = next(delimiter) start = next(delimiter) + 1
end = next(delimiter)
except StopIteration:
return None
return "\n".join(raw.splitlines()[start:end]) return "\n".join(raw.splitlines()[start:end])
+42 -28
View File
@@ -1,43 +1,57 @@
from collections.abc import Generator from collections.abc import Iterable
from dataclasses import dataclass from dataclasses import dataclass
from fnmatch import fnmatch from fnmatch import fnmatch
from functools import partial from importlib.resources.abc import Traversable
from itertools import groupby from itertools import groupby
from itertools import starmap
from typing import Self
from .file import MarkdownDocument
def get_skill_filemap(content: dict[str, str]) -> dict[str, list[str]]:
"""Get the map of skill slugs to their associated files."""
def get_skill_name(relpath: str) -> str | None:
if relpath.startswith("skills/"):
return relpath.split("/")[1]
grouped = groupby(content.keys(), key=get_skill_name)
return {k: list(v) for k, v in grouped if k is not None}
@dataclass(frozen=True, slots=True) @dataclass(frozen=True, slots=True)
class SkillBundle: class SkillFilesBundle:
"""Represents a skill and all of its associated markdown files."""
slug: str slug: str
skill: str skill: MarkdownDocument
references: list[str] references: tuple[MarkdownDocument, ...]
other: list[str] 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()))
def gen_skill_bundles(content: dict[str, str]) -> Generator[SkillBundle]: @classmethod
"""Generate the skill bundles from the map of raw markdown content.""" def from_docs(cls, docs: Iterable[MarkdownDocument]) -> tuple[Self, ...]:
matcher = partial(fnmatch, pat="skills/*/references/*.md") return tuple(starmap(cls.from_paths, group_skill_paths(docs).items()))
for slug, paths in get_skill_filemap(content).items():
skill = next(iter(p for p in paths if fnmatch(p, "skills/*/SKILL.md"))) @classmethod
groups = {k: list(v) for k, v in groupby(paths, key=matcher)} def from_paths(cls, slug: str, paths: set[MarkdownDocument]) -> Self:
yield SkillBundle( 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, slug=slug,
skill=skill, skill=skill,
references=list(groups.get(True, [])), references=references,
other=[f for f in groups.get(False, []) if f != skill], other=other,
) )
def get_all_skill_bundles(content: dict[str, str]) -> list[SkillBundle]: def group_skill_paths(docs: Iterable[MarkdownDocument]) -> dict[str, set[MarkdownDocument]]:
"""Get all skill bundles from the map of raw markdown content.""" """Group skills from a list of markdown documents by their skill slug."""
return list(gen_skill_bundles(content)) 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}