Extract row pool routing policy

row_pools.py

@@ -0,0 +1,138 @@
+from __future__ import annotations
+
+import json
+from typing import Any
+
+try:
+    from . import category_library as category_policy
+    from . import generate_prompt_batches as g
+    from . import location_config as location_policy
+except ImportError:  # Allows local smoke tests with top-level imports.
+    import category_library as category_policy
+    import generate_prompt_batches as g
+    import location_config as location_policy
+
+
+def _list_from(value: Any) -> list[Any]:
+    if value is None:
+        return []
+    if isinstance(value, list):
+        return value
+    return [value]
+
+
+def _is_false(value: Any) -> bool:
+    if isinstance(value, bool):
+        return value is False
+    if isinstance(value, str):
+        return value.strip().lower() in ("false", "0", "no", "off")
+    return False
+
+
+def _unique_extend(target: list[Any], additions: list[Any]) -> None:
+    seen = set()
+    for item in target:
+        try:
+            seen.add(json.dumps(item, sort_keys=True))
+        except TypeError:
+            seen.add(repr(item))
+    for item in additions:
+        try:
+            marker = json.dumps(item, sort_keys=True)
+        except TypeError:
+            marker = repr(item)
+        if marker not in seen:
+            target.append(item)
+            seen.add(marker)
+
+
+def scene_pool(
+    category: dict[str, Any],
+    subcategory: dict[str, Any],
+    item: Any,
+    subject_type: str,
+    location_config: dict[str, Any] | None = None,
+) -> list[Any]:
+    location_config = location_config or {}
+    location_entries = _list_from(location_config.get("scene_entries"))
+    if location_policy.location_config_active(location_config) and location_config.get("apply_mode") == "replace":
+        return location_entries
+    fallback = g.GROUP_SCENES if subject_type in ("group", "configured_cast") else g.SCENES
+    scene_entries: list[Any] = []
+    scene_pools = category_policy.load_scene_pool_library()
+    item_source = item if isinstance(item, dict) else None
+    if item_source is not None and _is_false(item_source.get("inherit_scenes")):
+        sources = (item_source,)
+    elif _is_false(subcategory.get("inherit_scenes")):
+        sources = (subcategory, item_source)
+    else:
+        sources = (category, subcategory, item_source)
+    for source in sources:
+        if not isinstance(source, dict):
+            continue
+        if "scenes" in source:
+            _unique_extend(scene_entries, _list_from(source["scenes"]))
+        refs = _list_from(source.get("scene_pool")) + _list_from(source.get("scene_pools"))
+        for ref in refs:
+            ref_name = str(ref).strip()
+            if ref_name not in scene_pools:
+                raise ValueError(f"Unknown scene pool '{ref_name}'")
+            _unique_extend(scene_entries, scene_pools[ref_name])
+    if location_policy.location_config_active(location_config) and location_config.get("apply_mode") == "add":
+        _unique_extend(scene_entries, location_entries)
+    return scene_entries or fallback
+
+
+def expression_pool(category: dict[str, Any], subcategory: dict[str, Any], item: Any) -> list[Any]:
+    return category_policy.configured_pool(
+        category,
+        subcategory,
+        item,
+        "expressions",
+        "expression_pools",
+        category_policy.load_expression_pool_library(),
+        "inherit_expressions",
+    ) or g.EXPRESSIONS
+
+
+def pose_pool(category: dict[str, Any], subcategory: dict[str, Any], item: Any, subject_type: str, poses: str) -> list[Any]:
+    configured = category_policy.merged_field(category, subcategory, item, "poses")
+    if configured:
+        return _list_from(configured)
+    if subject_type == "couple":
+        return [entry[2] for entry in g.COUPLE_TYPES]
+    if subject_type in ("layout", "scene"):
+        return ["clean designed layout"]
+    return g.EVOCATIVE_ALL if poses == "evocative" else g.POSES
+
+
+def composition_pool(
+    category: dict[str, Any],
+    subcategory: dict[str, Any],
+    item: Any,
+    subject_type: str,
+    composition_config: dict[str, Any] | None = None,
+) -> list[Any]:
+    composition_config = composition_config or {}
+    composition_entries = _list_from(composition_config.get("composition_entries"))
+    if location_policy.composition_config_active(composition_config) and composition_config.get("apply_mode") == "replace":
+        return composition_entries
+    configured = category_policy.configured_pool(
+        category,
+        subcategory,
+        item,
+        "compositions",
+        "composition_pools",
+        category_policy.load_composition_pool_library(),
+        "inherit_compositions",
+    )
+    if location_policy.composition_config_active(composition_config) and composition_config.get("apply_mode") == "add":
+        configured = list(configured or [])
+        _unique_extend(configured, composition_entries)
+    if configured:
+        return configured
+    if subject_type in ("group", "configured_cast"):
+        return g.GROUP_COMPOSITIONS
+    if subject_type in ("layout", "scene"):
+        return ["designed illustration layout"]
+    return g.COMPOSITIONS