Add expression intensity controls

prompt_builder.py

@@ -562,15 +562,15 @@ def load_category_library() -> list[dict[str, Any]]:
     return categories
 
 
-def load_scene_pool_library() -> dict[str, list[Any]]:
+def _load_named_pool_library(key: str) -> dict[str, list[Any]]:
     pools: dict[str, list[Any]] = {}
     for path in _json_files():
         data = _read_json(path)
-        raw_pools = data.get("scene_pools", {})
+        raw_pools = data.get(key, {})
         if not raw_pools:
             continue
         if not isinstance(raw_pools, dict):
-            raise ValueError(f"scene_pools in {path} must be an object")
+            raise ValueError(f"{key} in {path} must be an object")
         for name, entries in raw_pools.items():
             pool_name = str(name).strip()
             if not pool_name:
@@ -580,6 +580,18 @@ def load_scene_pool_library() -> dict[str, list[Any]]:
     return pools
 
 
+def load_scene_pool_library() -> dict[str, list[Any]]:
+    return _load_named_pool_library("scene_pools")
+
+
+def load_expression_pool_library() -> dict[str, list[Any]]:
+    return _load_named_pool_library("expression_pools")
+
+
+def load_composition_pool_library() -> dict[str, list[Any]]:
+    return _load_named_pool_library("composition_pools")
+
+
 def _extension_targets() -> dict[str, tuple[list[Any], bool]]:
     return {
         "women_clothes": (g.WOMEN_CLOTHES, False),
@@ -654,6 +666,14 @@ def _ratio_or_none(value: float) -> float | None:
     return max(0.0, min(1.0, ratio))
 
 
+def _clamped_float(value: Any, default: float = 0.5, min_value: float = 0.0, max_value: float = 1.0) -> float:
+    try:
+        number = float(value)
+    except (TypeError, ValueError):
+        return default
+    return max(min_value, min(max_value, number))
+
+
 def build_seed_config_json(
     category_seed: int = -1,
     subcategory_seed: int = -1,
@@ -1384,6 +1404,180 @@ def _scene_pool(category: dict[str, Any], subcategory: dict[str, Any], item: Any
     return scene_entries or fallback
 
 
+def _sources_with_inheritance(
+    category: dict[str, Any],
+    subcategory: dict[str, Any],
+    item: Any,
+    inherit_key: str,
+) -> tuple[Any, ...]:
+    item_source = item if isinstance(item, dict) else None
+    if item_source is not None and _is_false(item_source.get(inherit_key)):
+        return (item_source,)
+    if _is_false(subcategory.get(inherit_key)):
+        return (subcategory, item_source)
+    return (category, subcategory, item_source)
+
+
+def _configured_pool(
+    category: dict[str, Any],
+    subcategory: dict[str, Any],
+    item: Any,
+    direct_key: str,
+    pool_key: str,
+    pool_library: dict[str, list[Any]],
+    inherit_key: str,
+) -> list[Any]:
+    entries: list[Any] = []
+    singular_pool_key = pool_key[:-1] if pool_key.endswith("s") else pool_key
+    for source in _sources_with_inheritance(category, subcategory, item, inherit_key):
+        if not isinstance(source, dict):
+            continue
+        if direct_key in source:
+            _unique_extend(entries, _list_from(source[direct_key]))
+        refs = _list_from(source.get(singular_pool_key)) + _list_from(source.get(pool_key))
+        for ref in refs:
+            ref_name = str(ref).strip()
+            if ref_name not in pool_library:
+                raise ValueError(f"Unknown {singular_pool_key} '{ref_name}'")
+            _unique_extend(entries, pool_library[ref_name])
+    return entries
+
+
+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
+
+
+def _expression_intensity_hint(entry: Any) -> float:
+    if isinstance(entry, dict):
+        for key in ("expression_intensity", "intensity"):
+            if key in entry:
+                return _clamped_float(entry[key], 0.5)
+
+    text = _entry_text(entry).lower()
+    high_terms = (
+        "ahegao",
+        "orgasm",
+        "climax",
+        "drool",
+        "drooling",
+        "tongue out",
+        "eyes rolled",
+        "fucked-out",
+        "cum-smeared",
+        "saliva",
+        "gagging",
+        "slack jaw",
+        "jaw slack",
+        "slack-jawed",
+        "sex-drunk",
+        "overwhelmed",
+        "strained",
+        "messy",
+        "panting",
+        "trembling",
+        "shaking",
+        "wide open mouth",
+        "raw ",
+        "wild ",
+        "dazed",
+        "spent",
+    )
+    if any(term in text for term in high_terms):
+        return 0.9
+
+    medium_terms = (
+        "seductive",
+        "teasing",
+        "lustful",
+        "aroused",
+        "bedroom",
+        "dominant",
+        "predatory",
+        "control",
+        "stern",
+        "strict",
+        "smirk",
+        "parted lips",
+        "open-mouthed",
+        "heated",
+        "hungry",
+        "inviting",
+        "sensual",
+        "fetish",
+        "commanding",
+        "flushed",
+        "moan",
+    )
+    if any(term in text for term in medium_terms):
+        return 0.62
+
+    low_terms = (
+        "neutral",
+        "quiet",
+        "calm",
+        "reserved",
+        "relaxed",
+        "candid",
+        "closed-mouth",
+        "thoughtful",
+        "controlled",
+        "focused",
+        "steady",
+        "bitten-lip",
+        "braced",
+        "held breath",
+        "concentrated",
+        "aloof",
+        "bored",
+        "tired",
+        "unfocused",
+        "contented",
+        "fashion",
+        "soft",
+        "sleepy",
+        "fresh-faced",
+    )
+    if any(term in text for term in low_terms):
+        return 0.25
+    return 0.5
+
+
+def _expression_entries_for_intensity(entries: list[Any], expression_intensity: float) -> list[Any]:
+    target = _clamped_float(expression_intensity, 0.5)
+    weighted: list[Any] = []
+    for entry in entries:
+        entry_intensity = _expression_intensity_hint(entry)
+        distance = abs(target - entry_intensity)
+        if distance <= 0.18:
+            intensity_weight = 4.0
+        elif distance <= 0.35:
+            intensity_weight = 1.4
+        elif distance <= 0.55:
+            intensity_weight = 0.35
+        else:
+            intensity_weight = 0.05
+
+        if isinstance(entry, dict):
+            adjusted = dict(entry)
+            try:
+                base_weight = float(adjusted.get("weight", 1.0))
+            except (TypeError, ValueError):
+                base_weight = 1.0
+            adjusted["weight"] = max(0.0, base_weight) * intensity_weight
+            weighted.append(adjusted)
+        else:
+            weighted.append({"text": _entry_text(entry), "weight": intensity_weight})
+    return weighted or entries
+
+
 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:
@@ -1396,9 +1590,17 @@ def _pose_pool(category: dict[str, Any], subcategory: dict[str, Any], item: Any,
 
 
 def _composition_pool(category: dict[str, Any], subcategory: dict[str, Any], item: Any, subject_type: str) -> list[Any]:
-    configured = _merged_field(category, subcategory, item, "compositions")
+    configured = _configured_pool(
+        category,
+        subcategory,
+        item,
+        "compositions",
+        "composition_pools",
+        load_composition_pool_library(),
+        "inherit_compositions",
+    )
     if configured:
-        return _list_from(configured)
+        return configured
     if subject_type in ("group", "configured_cast"):
         return g.GROUP_COMPOSITIONS
     if subject_type in ("layout", "scene"):
@@ -1420,6 +1622,7 @@ def _build_custom_row(
     men_count: int,
     seed: int,
     seed_config: dict[str, int],
+    expression_intensity: float,
 ) -> dict[str, Any]:
     categories = load_category_library()
     category_rng = _axis_rng(seed_config, "category", seed, row_number)
@@ -1454,12 +1657,14 @@ def _build_custom_row(
     pose = str(_merged_field(category, subcategory, item, "pose", "") or context.get("fallback_pose") or _choose_text(
         pose_rng, _compatible_entries(_pose_pool(category, subcategory, item, subject_type, poses), women_count, men_count)
     ))
-    expression = _choose_text(
-        expression_rng,
-        _compatible_entries(_list_from(_merged_field(category, subcategory, item, "expressions", g.EXPRESSIONS)), women_count, men_count),
+    expression_entries = _compatible_entries(
+        _expression_entries_for_intensity(_expression_pool(category, subcategory, item), expression_intensity),
+        women_count,
+        men_count,
     )
+    expression = _choose_text(expression_rng, expression_entries)
     if subject_type in ("couple", "group") and ";" not in expression:
-        expression = f"{expression}; {_choose_text(expression_rng, g.EXPRESSIONS)}"
+        expression = f"{expression}; {_choose_text(expression_rng, expression_entries)}"
     composition = _choose_text(
         composition_rng,
         _compatible_entries(_composition_pool(category, subcategory, item, subject_type), women_count, men_count),
@@ -1492,6 +1697,7 @@ def _build_custom_row(
             "scene_slug": scene_slug,
             "pose": pose,
             "expression": expression,
+            "expression_intensity": expression_intensity,
             "composition": composition,
             "role_graph": role_graph,
             "positive_suffix": positive_suffix,
@@ -1588,6 +1794,7 @@ def build_prompt(
     women_count: int = 1,
     men_count: int = 1,
     camera_config: str | dict[str, Any] | None = None,
+    expression_intensity: float = 0.5,
 ) -> dict[str, Any]:
     apply_pool_extensions()
     row_number = max(1, int(row_number))
@@ -1599,6 +1806,7 @@ def build_prompt(
     figure = figure if figure in ("curvy", "balanced", "bombshell") else "curvy"
     minimal_ratio = _ratio_or_none(minimal_clothing_ratio)
     pose_ratio = _ratio_or_none(standard_pose_ratio)
+    expression_intensity = _clamped_float(expression_intensity, 0.5)
     parsed_seed_config = _parse_seed_config(seed_config)
 
     exact_custom_subcategory = bool(subcategory and subcategory != RANDOM_SUBCATEGORY and " / " in subcategory)
@@ -1649,6 +1857,7 @@ def build_prompt(
             int(men_count),
             seed,
             parsed_seed_config,
+            expression_intensity,
         )
 
     if extra_positive.strip():
@@ -1658,6 +1867,7 @@ def build_prompt(
     row["prompt"] = _prepend_trigger(row["prompt"], active_trigger, bool(prepend_trigger_to_prompt))
     row["negative_prompt"] = _combined_negative(row.get("negative_prompt", g.NEGATIVE_PROMPT), extra_negative)
     row["trigger"] = active_trigger
+    row["expression_intensity"] = expression_intensity
     return row
 
 
@@ -1700,6 +1910,8 @@ def build_insta_of_options_json(
     continuity: str = "same_creator_same_room",
     softcore_camera_mode: str = "handheld_selfie",
     hardcore_camera_mode: str = "same_as_softcore",
+    softcore_expression_intensity: float = 0.45,
+    hardcore_expression_intensity: float = 0.85,
 ) -> str:
     return json.dumps(
         {
@@ -1713,6 +1925,8 @@ def build_insta_of_options_json(
             "continuity": continuity,
             "softcore_camera_mode": softcore_camera_mode,
             "hardcore_camera_mode": hardcore_camera_mode,
+            "softcore_expression_intensity": _clamped_float(softcore_expression_intensity, 0.45),
+            "hardcore_expression_intensity": _clamped_float(hardcore_expression_intensity, 0.85),
         },
         ensure_ascii=True,
         sort_keys=True,
@@ -1731,6 +1945,8 @@ def _parse_insta_of_options(options_json: str | dict[str, Any] | None) -> dict[s
         "continuity": "same_creator_same_room",
         "softcore_camera_mode": "handheld_selfie",
         "hardcore_camera_mode": "same_as_softcore",
+        "softcore_expression_intensity": 0.45,
+        "hardcore_expression_intensity": 0.85,
     }
     if not options_json:
         return defaults
@@ -1753,6 +1969,14 @@ def _parse_insta_of_options(options_json: str | dict[str, Any] | None) -> dict[s
     parsed["softcore_camera_mode"] = parsed["softcore_camera_mode"] if parsed["softcore_camera_mode"] in CAMERA_MODE_PROMPTS else defaults["softcore_camera_mode"]
     if parsed["hardcore_camera_mode"] not in CAMERA_MODE_PROMPTS and parsed["hardcore_camera_mode"] != "same_as_softcore":
         parsed["hardcore_camera_mode"] = defaults["hardcore_camera_mode"]
+    parsed["softcore_expression_intensity"] = _clamped_float(
+        parsed.get("softcore_expression_intensity"),
+        defaults["softcore_expression_intensity"],
+    )
+    parsed["hardcore_expression_intensity"] = _clamped_float(
+        parsed.get("hardcore_expression_intensity"),
+        defaults["hardcore_expression_intensity"],
+    )
     for key in ("hardcore_women_count", "hardcore_men_count"):
         try:
             parsed[key] = max(0, min(12, int(parsed[key])))
@@ -1845,6 +2069,7 @@ def build_insta_of_pair(
         seed_config=parsed_seed_config,
         women_count=1,
         men_count=0,
+        expression_intensity=options["softcore_expression_intensity"],
     )
     hard_row = build_prompt(
         category="Hardcore sexual poses",
@@ -1868,6 +2093,7 @@ def build_insta_of_pair(
         seed_config=parsed_seed_config,
         women_count=hard_women_count,
         men_count=hard_men_count,
+        expression_intensity=options["hardcore_expression_intensity"],
     )
 
     descriptor = _insta_of_descriptor(soft_row)