"""§22.4a SLICE-3 — faceted catalog filtering + counts (read). Pure functions over already-mirrored entries; no I/O, no DB. An "entry" here is a plain dict carrying at least: - "state": the lifecycle state column, - "metadata_malformed": bool, - "meta": the per-entry metadata mapping (from cached_rfcs.meta_json). Facetable fields (§5.1, plan decision 1): a collection's declared `enum` and `tags` fields, in declaration order, plus the built-in `state` facet appended last — but only when the collection declares a schema (INV-5: a no-`fields:` collection has no facets at all, so the frontend keeps its legacy chips). `text` fields are not faceted in v1 (they get a detail control in SLICE-4). Counts use drill-down semantics (plan decision 2): the count for a value of field F is taken over entries matching every OTHER field's selection (and the malformed toggle), not F's own — so within-field values stay switchable (OR within a field, AND across fields). The returned items list applies ALL selections. """ from __future__ import annotations from typing import Any # enum + tags are facetable; text is rendered as a detail control (SLICE-4). FACETABLE_TYPES = {"enum", "tags"} def facet_fields(fields: dict[str, dict] | None) -> list[tuple[str, str]]: """Ordered `[(name, type), ...]` facetable from the schema, `state` last. Empty when the collection declares no schema (INV-5).""" if not fields: return [] out = [ (name, spec.get("type")) for name, spec in fields.items() if spec.get("type") in FACETABLE_TYPES ] out.append(("state", "enum")) return out def allowed_filter_keys(fields: dict[str, dict] | None) -> set[str]: """Query-param keys the collection-scoped list accepts (plan decision 6).""" keys = {name for name, _ in facet_fields(fields)} keys.update({"unreviewed", "malformed"}) return keys def _values_for(entry: dict[str, Any], name: str, ftype: str) -> list[str]: """The facet value(s) an entry contributes for field `name` (str-cast).""" if name == "state": v = entry.get("state") return [str(v)] if v else [] meta = entry.get("meta") or {} v = meta.get(name) if v is None: return [] if ftype == "tags": return [str(x) for x in v] if isinstance(v, list) else [] return [str(v)] def _matches(entry: dict[str, Any], name: str, ftype: str, selected: set[str]) -> bool: if not selected: return True return bool(set(_values_for(entry, name, ftype)) & selected) # OR within field def filter_and_count( entries: list[dict[str, Any]], fields: dict[str, dict] | None, selections: dict[str, set[str]], only_malformed: bool = False, ) -> tuple[list[dict[str, Any]], dict[str, dict[str, int]]]: """Filter `entries` by `selections` and compute drill-down facet counts. `selections` maps a facet field name → the set of selected values (OR within the field; AND across fields). `only_malformed` narrows items and counts to entries flagged malformed (INV-3). Returns `(items, facets)` where `facets = {field: {value: count}}`. With no schema → `([all passing], {})`. """ facetable = facet_fields(fields) def passes_malformed(e: dict[str, Any]) -> bool: return (not only_malformed) or bool(e.get("metadata_malformed")) items = [ e for e in entries if passes_malformed(e) and all(_matches(e, n, t, selections.get(n, set())) for n, t in facetable) ] facets: dict[str, dict[str, int]] = {} for name, ftype in facetable: counts: dict[str, int] = {} for e in entries: if not passes_malformed(e): continue if not all( _matches(e, on, ot, selections.get(on, set())) for on, ot in facetable if on != name ): continue for val in _values_for(e, name, ftype): counts[val] = counts.get(val, 0) + 1 facets[name] = counts return items, facets