feat(splitter): implement the Category Splitter (#13) (#57)

Add compiler/splitter/category_splitter.py: split_into_categories(tags,
logger=None) -> CompilerResult (data = CategoryMap), per MASTER_SPEC §18.

- Each Resolved Tag is placed in the category recorded on it by the Resolver
  (its Knowledge Base Entry category); tags are never modified, generated, or
  removed (original objects kept verbatim).
- Tags within a category preserve Resolver order; category ordering follows the
  shared canonical 19-category order; empty categories are omitted (§18.4).
- Defensive: a tag whose category is outside the 19 stops with SC0008.

Tests: assignment, within-category ordering, canonical category ordering,
empty-category omission, tag identity preserved, empty input, and SC0008.

Co-authored-by: kunail0804 <kunail0804@gmail.com>
Co-authored-by: Claude Opus 4.8 <noreply@anthropic.com>
This commit is contained in:
Crespin Alexandre
2026-07-07 20:00:54 +02:00
committed by GitHub
co-authored by kunail0804 Claude Opus 4.8
parent 76d440fbaf
commit 499a2490ca
2 changed files with 127 additions and 0 deletions
+58
View File
@@ -0,0 +1,58 @@
"""The Category Splitter (MASTER_SPEC §18).
Organizes Resolved Tags into the 19 canonical categories, preserving order and
without modifying, generating, or removing any tag. Each tag is placed in the
category recorded on it by the Resolver (which is the category of its Knowledge
Base Entry).
Category ordering is deterministic (the shared canonical order); tags within a
category keep Resolver order; empty categories are omitted from the Category Map
(§18.4).
"""
from __future__ import annotations
from types import MappingProxyType
from compiler.common.categories import CANONICAL_CATEGORIES, is_valid_category
from compiler.common.log import StructuredLogger
from compiler.common.message_codes import message
from compiler.common.result import CompilerResult
from schemas.models import CategoryMap, ResolvedTag
def split_into_categories(
tags: tuple[ResolvedTag, ...],
logger: StructuredLogger | None = None,
) -> CompilerResult:
"""Group Resolved Tags into a Category Map.
Returns:
A CompilerResult whose ``data`` is the :class:`CategoryMap` on success,
or ``None`` with an SC0008 error if any tag carries a category outside
the 19 canonical categories.
"""
buckets: dict[str, list[ResolvedTag]] = {}
for resolved in tags:
if not is_valid_category(resolved.category):
return CompilerResult().add_error(
message(
"SC0008",
f"Tag '{resolved.tag}' has invalid category '{resolved.category}'.",
tag=resolved.tag,
category=resolved.category,
)
)
buckets.setdefault(resolved.category, []).append(resolved)
# Deterministic category order; empty categories omitted. Original ResolvedTag
# objects are kept verbatim (tags are never modified).
ordered = {
category: tuple(buckets[category])
for category in CANONICAL_CATEGORIES
if category in buckets
}
category_map = CategoryMap(categories=MappingProxyType(ordered))
if logger is not None:
logger.basic("categories_split", categories=len(ordered), tags=len(tags))
return CompilerResult(data=category_map)
+69
View File
@@ -0,0 +1,69 @@
"""Tests for the Category Splitter (issue #13, MASTER_SPEC §18)."""
from __future__ import annotations
from compiler.common.categories import CANONICAL_CATEGORIES
from compiler.splitter.category_splitter import split_into_categories
from schemas.models import CategoryMap, ResolvedTag
def tag(name: str, category: str) -> ResolvedTag:
return ResolvedTag(tag=name, category=category, source_concept=name, knowledge_base_entry=name)
def codes(messages) -> list[str]:
return [m.code for m in messages]
def test_output_is_a_category_map() -> None:
result = split_into_categories((tag("1girl", "character"),))
assert result.success
assert isinstance(result.data, CategoryMap)
def test_tags_land_in_their_entry_category() -> None:
tags = (tag("1girl", "character"), tag("blue eyes", "eyes"), tag("long hair", "hair"))
result = split_into_categories(tags)
assert [t.tag for t in result.data.tags_for("character")] == ["1girl"]
assert [t.tag for t in result.data.tags_for("eyes")] == ["blue eyes"]
assert [t.tag for t in result.data.tags_for("hair")] == ["long hair"]
def test_order_within_category_preserves_resolver_order() -> None:
tags = (tag("long hair", "hair"), tag("blonde hair", "hair"), tag("bangs", "hair"))
result = split_into_categories(tags)
assert [t.tag for t in result.data.tags_for("hair")] == ["long hair", "blonde hair", "bangs"]
def test_empty_categories_are_omitted() -> None:
result = split_into_categories((tag("1girl", "character"),))
present = list(result.data.categories.keys())
assert present == ["character"] # only the non-empty category
def test_category_ordering_is_canonical() -> None:
# Provide tags out of canonical order; the map must follow CANONICAL_CATEGORIES.
tags = (tag("sunset", "lighting"), tag("1girl", "character"), tag("long hair", "hair"))
result = split_into_categories(tags)
present = list(result.data.categories.keys())
assert present == ["character", "hair", "lighting"]
assert present == [c for c in CANONICAL_CATEGORIES if c in present]
def test_tags_are_not_modified() -> None:
original = tag("1girl", "character")
result = split_into_categories((original,))
assert result.data.tags_for("character")[0] is original
def test_empty_input_produces_empty_map() -> None:
result = split_into_categories(())
assert result.success
assert len(result.data.categories) == 0
def test_invalid_category_stops_with_sc0008() -> None:
result = split_into_categories((tag("x", "not_a_category"),))
assert not result.success
assert result.data is None
assert "SC0008" in codes(result.errors)