From e5822e42f84ac14f33670127f5eb9252f9998dba Mon Sep 17 00:00:00 2001 From: Ethanfel Date: Sat, 27 Jun 2026 08:47:22 +0200 Subject: [PATCH] Extract row pool routing policy --- docs/prompt-architecture-improvement-plan.md | 4 + docs/prompt-pool-routing-map.md | 1 + prompt_builder.py | 82 +---------- row_pools.py | 138 +++++++++++++++++++ tools/prompt_smoke.py | 21 +++ 5 files changed, 170 insertions(+), 76 deletions(-) create mode 100644 row_pools.py diff --git a/docs/prompt-architecture-improvement-plan.md b/docs/prompt-architecture-improvement-plan.md index 0e3aa70..96d058d 100644 --- a/docs/prompt-architecture-improvement-plan.md +++ b/docs/prompt-architecture-improvement-plan.md @@ -165,6 +165,10 @@ Already isolated: in `location_config.py`; built-in row location/composition config application, source metadata, and prompt/caption rewrites live in `row_location.py`. +- row scene/expression/pose/composition pool routing, category inheritance, + runtime location/composition pool overrides, and generator fallback pool + selection live in `row_pools.py`; `prompt_builder.py` keeps public delegate + wrappers. - hardcore position/action-filter choices, selected-position normalization, config JSON builders/parsers, focus-policy toggles, subcategory allow-list policy, position-key detection, category filtering, and item-template/axis diff --git a/docs/prompt-pool-routing-map.md b/docs/prompt-pool-routing-map.md index 086ffbf..7f61848 100644 --- a/docs/prompt-pool-routing-map.md +++ b/docs/prompt-pool-routing-map.md @@ -82,6 +82,7 @@ Core helper ownership: | `subject_context.py` | Row subject-context routing for single, couple, configured-cast, group, and layout subjects, combining appearance policy, cast metadata, and generator subject pools. | | `location_config.py` | Location/composition preset schemas, themed location packs, custom location/composition parsing, pool merge behavior, and location/composition config parsing. | | `row_location.py` | Built-in row location/composition config application, deterministic scene/composition choice, source metadata, and legacy prompt/caption rewrites. | +| `row_pools.py` | Row scene/expression/pose/composition pool routing, category inheritance handling, runtime location/composition pool overrides, and generator fallback pools. | | `hardcore_position_config.py` | Hardcore position/action-filter choices, selected-position normalization, config JSON builders/parsers, focus-policy toggles, subcategory allow-list policy, position-key detection, and category/template/axis filtering. | | `pair_options.py` | Insta/OF option schema/defaults, softcore category/outfit/pose pools, partner outfit pools, clothing-continuity labels, negatives, hardcore cast count policy, and hardcore detail-density directives. | | `pair_rows.py` | Insta/OF soft/hard row creation, softcore expression override resolution, Woman A slot context application, soft outfit/pose overrides, and POV row fields. | diff --git a/prompt_builder.py b/prompt_builder.py index 5f31e3c..941e671 100644 --- a/prompt_builder.py +++ b/prompt_builder.py @@ -12,12 +12,8 @@ try: category_json_files as _json_files, compatible_entries as _compatible_entries, compatible_entry as _compatible_entry, - configured_pool as _configured_pool, find_subcategory as _find_subcategory, load_category_library, - load_composition_pool_library, - load_expression_pool_library, - load_scene_pool_library, merged_axes as _merged_axes, merged_field as _merged_field, read_category_json as _read_json, @@ -46,6 +42,7 @@ try: from . import row_normalization as row_policy from . import row_camera as row_camera_policy from . import row_location as row_location_policy + from . import row_pools as row_pool_policy from . import seed_config as seed_policy from . import subject_context as subject_context_policy from .hardcore_text_cleanup import ( @@ -59,12 +56,8 @@ except ImportError: # Allows local smoke tests with `python -c`. category_json_files as _json_files, compatible_entries as _compatible_entries, compatible_entry as _compatible_entry, - configured_pool as _configured_pool, find_subcategory as _find_subcategory, load_category_library, - load_composition_pool_library, - load_expression_pool_library, - load_scene_pool_library, merged_axes as _merged_axes, merged_field as _merged_field, read_category_json as _read_json, @@ -93,6 +86,7 @@ except ImportError: # Allows local smoke tests with `python -c`. import row_normalization as row_policy import row_camera as row_camera_policy import row_location as row_location_policy + import row_pools as row_pool_policy import seed_config as seed_policy import subject_context as subject_context_policy from hardcore_text_cleanup import ( @@ -2519,46 +2513,11 @@ def _scene_pool( 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_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 = 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_config_active(location_config) and location_config.get("apply_mode") == "add": - _unique_extend(scene_entries, location_entries) - return scene_entries or fallback + return row_pool_policy.scene_pool(category, subcategory, item, subject_type, location_config) def _expression_pool(category: dict[str, Any], subcategory: dict[str, Any], item: Any) -> list[Any]: - return _configured_pool( - category, - subcategory, - item, - "expressions", - "expression_pools", - load_expression_pool_library(), - "inherit_expressions", - ) or g.EXPRESSIONS + return row_pool_policy.expression_pool(category, subcategory, item) def _expression_intensity_hint(entry: Any) -> float: @@ -2685,14 +2644,7 @@ def _expression_entries_for_intensity(entries: list[Any], expression_intensity: def _pose_pool(category: dict[str, Any], subcategory: dict[str, Any], item: Any, subject_type: str, poses: str) -> list[Any]: - configured = _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 + return row_pool_policy.pose_pool(category, subcategory, item, subject_type, poses) def _composition_pool( @@ -2702,29 +2654,7 @@ def _composition_pool( 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 _composition_config_active(composition_config) and composition_config.get("apply_mode") == "replace": - return composition_entries - configured = _configured_pool( - category, - subcategory, - item, - "compositions", - "composition_pools", - load_composition_pool_library(), - "inherit_compositions", - ) - if _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 + return row_pool_policy.composition_pool(category, subcategory, item, subject_type, composition_config) def _build_custom_row( diff --git a/row_pools.py b/row_pools.py new file mode 100644 index 0000000..5407989 --- /dev/null +++ b/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 diff --git a/tools/prompt_smoke.py b/tools/prompt_smoke.py index e9526cb..e440be4 100644 --- a/tools/prompt_smoke.py +++ b/tools/prompt_smoke.py @@ -52,6 +52,7 @@ import row_normalization # noqa: E402 import route_metadata # noqa: E402 import row_camera # noqa: E402 import row_location # noqa: E402 +import row_pools # noqa: E402 import server_routes # noqa: E402 import sdxl_formatter # noqa: E402 import sdxl_presets # noqa: E402 @@ -1662,6 +1663,26 @@ def smoke_category_library_route() -> None: _expect(expressions, "category inheritance did not resolve expressions") _expect(compositions, "category inheritance did not resolve compositions") _expect(any("oral" in _clean_key(entry.get("prompt") if isinstance(entry, dict) else entry) for entry in scenes), "oral scene pool did not contribute") + location_override = {"enabled": True, "apply_mode": "replace", "scene_entries": ["custom scene"]} + composition_override = {"enabled": True, "apply_mode": "replace", "composition_entries": ["custom composition"]} + _expect( + pb._scene_pool(category, subcategory, item, "configured_cast", location_override) + == row_pools.scene_pool(category, subcategory, item, "configured_cast", location_override), + "Prompt builder scene pool should delegate to row_pools", + ) + _expect( + pb._expression_pool(category, subcategory, item) == row_pools.expression_pool(category, subcategory, item), + "Prompt builder expression pool should delegate to row_pools", + ) + _expect( + pb._pose_pool(category, subcategory, item, "couple", "standard") == row_pools.pose_pool(category, subcategory, item, "couple", "standard"), + "Prompt builder pose pool should delegate to row_pools", + ) + _expect( + pb._composition_pool(category, subcategory, item, "configured_cast", composition_override) + == row_pools.composition_pool(category, subcategory, item, "configured_cast", composition_override), + "Prompt builder composition pool should delegate to row_pools", + ) def smoke_hardcore_category_routes() -> None: