688 lines
27 KiB
Python
688 lines
27 KiB
Python
import os
|
||
import random
|
||
import json
|
||
import requests
|
||
from datetime import datetime
|
||
|
||
class PromptCreatorNode:
|
||
@classmethod
|
||
def INPUT_TYPES(cls):
|
||
json_dir = os.path.join(os.path.dirname(__file__), "JSON_DATA")
|
||
json_files = [f for f in os.listdir(json_dir) if f.endswith(".json")]
|
||
|
||
# camera angles (optional) from camera_angles.json
|
||
camera_angle_list = ["none"]
|
||
camera_angles_path = os.path.join(os.path.dirname(__file__), "camera_angles.json")
|
||
if os.path.exists(camera_angles_path):
|
||
try:
|
||
with open(camera_angles_path, "r", encoding="utf-8") as f:
|
||
camera_data = json.load(f)
|
||
ids = camera_data.get("ids", [])
|
||
if isinstance(ids, list):
|
||
camera_angle_list += [str(x) for x in ids if str(x).strip()]
|
||
except Exception:
|
||
pass
|
||
|
||
# carico le chiavi dal file esterno system_prompt.json
|
||
system_prompt_path = os.path.join(os.path.dirname(__file__), "system_prompt.json")
|
||
if os.path.exists(system_prompt_path):
|
||
with open(system_prompt_path, "r", encoding="utf-8") as f:
|
||
system_prompts = json.load(f)
|
||
enhancer_modes = list(system_prompts.keys())
|
||
else:
|
||
enhancer_modes = ["standard"]
|
||
|
||
# carico le chiavi dal file esterno identities.json (volto consistente)
|
||
identities_path = os.path.join(os.path.dirname(__file__), "identities.json")
|
||
if os.path.exists(identities_path):
|
||
try:
|
||
with open(identities_path, "r", encoding="utf-8") as f:
|
||
identities = json.load(f)
|
||
identity_profiles = ["none", "external"] + list(identities.keys())
|
||
except Exception:
|
||
identity_profiles = ["none", "external"]
|
||
else:
|
||
identity_profiles = ["none", "external"]
|
||
|
||
return {
|
||
"required": {
|
||
"json_name": (sorted(json_files),),
|
||
"camera_angle": (camera_angle_list,),
|
||
"use_enhancer": (["none", "ollama","llamacpp", "openai", "cohere", "gemini"],),
|
||
"enhancer_mode": (enhancer_modes,),
|
||
"system_prompt_lock": (["auto", "external"], {"default": "auto"}),
|
||
"add_symbols": (["no", "yes"],),
|
||
"seed": ("INT", {"default": 0}),
|
||
"gender": (["neutral", "female", "2 female", "3 female", "female vampire", "anime woman", "male", "custom"],),
|
||
"identity_profile": (identity_profiles,),
|
||
"external_identity": ("STRING", {"default": "", "multiline": True, "forceInput": True}),
|
||
"lock_identity": (["no", "yes"], {"default": "yes"}),
|
||
"custom_intro": ("STRING", {"default": "", "multiline": True}),
|
||
"custom_intro_id": (["Random","0","1","2","3","4","5","6"], {"default":"Random"}),
|
||
"horror_intensity": (["auto"] + [str(i) for i in range(11)],),
|
||
"sensuality_level": (["auto", "0", "1", "2", "3"],),
|
||
"pose_mode": (["random", "world_pick", "lock"], {"default": "random"}),
|
||
"pose_index": ("INT", {"default": 0, "min": 0, "max": 200, "step": 1}),
|
||
"lora_triggers": ("STRING", {"default": ""}),
|
||
"subject_count": (["1", "2", "3"],),
|
||
"lock_last_prompt": (["no", "yes"], {"default": "no"}),
|
||
"multi_object_count": ("INT", {"default": 1, "min": 1, "max": 5}),
|
||
# Ollama settings
|
||
"ollama_host": ("STRING", {"default": "http://10.10.10.2:11434"}),
|
||
"ollama_model": ("STRING", {"default": "qwen3:8b"})
|
||
}
|
||
}
|
||
|
||
RETURN_TYPES = ("STRING",)
|
||
RETURN_NAMES = ("prompt",)
|
||
FUNCTION = "generate_prompt"
|
||
CATEGORY = "Prompt Creator"
|
||
|
||
def _system_prompts(self):
|
||
system_prompt_path = os.path.join(os.path.dirname(__file__), "system_prompt.json")
|
||
if os.path.exists(system_prompt_path):
|
||
with open(system_prompt_path, "r", encoding="utf-8") as f:
|
||
return json.load(f)
|
||
else:
|
||
# fallback se manca il file
|
||
return {
|
||
"standard": (
|
||
"You are a professional prompt enhancer for AI image generation. "
|
||
"Return ONE long, multi-clause sentence (~120–160 words), richly descriptive. "
|
||
"Cover: genre/style; era; foreground/midground/background; materials & textures; color palette (3–5 tones); "
|
||
"lighting (type/direction/intensity); atmosphere; camera angle & lens (mm); composition rule; 3–5 post-process keywords. "
|
||
"No story, no dialogue, no lists, no headings."
|
||
)
|
||
}
|
||
|
||
def _identity_profiles(self):
|
||
"""Carica i profili identità dal file identities.json (se presente)."""
|
||
identities_path = os.path.join(os.path.dirname(__file__), "identities.json")
|
||
if os.path.exists(identities_path):
|
||
try:
|
||
with open(identities_path, "r", encoding="utf-8") as f:
|
||
data = json.load(f)
|
||
return data if isinstance(data, dict) else {}
|
||
except Exception:
|
||
return {}
|
||
return {}
|
||
|
||
def _identity_to_text(self, identity_obj):
|
||
"""Converte un profilo identità in una stringa descrittiva coerente (prompt-only)."""
|
||
if not isinstance(identity_obj, dict):
|
||
return ""
|
||
order = ["age", "face_type", "eyes", "nose", "mouth", "hair", "skin", "expression_base"]
|
||
chunks = []
|
||
for k in order:
|
||
v = identity_obj.get(k) or identity_obj.get(k.upper())
|
||
if isinstance(v, str) and v.strip():
|
||
chunks.append(v.strip())
|
||
return ", ".join(chunks)
|
||
|
||
|
||
|
||
def _camera_angles(self):
|
||
path = os.path.join(os.path.dirname(__file__), "camera_angles.json")
|
||
if os.path.exists(path):
|
||
try:
|
||
with open(path, "r", encoding="utf-8") as f:
|
||
return json.load(f)
|
||
except Exception:
|
||
pass
|
||
return {"ids": ["front"], "map": {}}
|
||
|
||
@staticmethod
|
||
def _is_none(v):
|
||
return (not v) or (str(v).strip().lower() == "none")
|
||
|
||
def _resolve_camera_angle(self, ui_angle, world_data, angle_dict):
|
||
"""
|
||
Se UI ha un valore (≠ none) → usa quello e ignora il mondo.
|
||
Se UI è none → eredita da world_data["camera_angle"] o ["CAMERA_ANGLE"].
|
||
Mappa l'id in camera_angles.map se presente, altrimenti lascia il testo.
|
||
"""
|
||
if not self._is_none(ui_angle):
|
||
chosen = ui_angle
|
||
else:
|
||
chosen = world_data.get("camera_angle") or world_data.get("CAMERA_ANGLE") or ""
|
||
if not chosen:
|
||
return ""
|
||
mapped = (angle_dict or {}).get("map", {}).get(chosen)
|
||
return mapped or chosen
|
||
|
||
def _sanitize_world_camera(self, world_data, ui_angle):
|
||
if not self._is_none(ui_angle) and world_data:
|
||
wd = dict(world_data)
|
||
wd.pop("camera_angle", None)
|
||
wd.pop("CAMERA_ANGLE", None)
|
||
return wd
|
||
return world_data
|
||
|
||
def _build_prompt_from_json(
|
||
self,
|
||
data,
|
||
gender,
|
||
custom_intro,
|
||
custom_intro_id,
|
||
horror_intensity,
|
||
sensuality_level,
|
||
subject_count,
|
||
multi_object_count,
|
||
camera_angle_txt="",
|
||
):
|
||
parts = []
|
||
|
||
# 1) Camera angle (se presente) in testa
|
||
if isinstance(camera_angle_txt, str) and camera_angle_txt.strip():
|
||
parts.append(camera_angle_txt.strip())
|
||
|
||
# 2) Gender / subject intro
|
||
# Se gender=custom:
|
||
# - se custom_intro (textarea) non è vuoto -> usa quello (override manuale)
|
||
# - altrimenti usa custom_intro_id (0..6 o Random) dal world JSON (CUSTOM_INTRO come dict o list)
|
||
if gender == "custom":
|
||
ci = (custom_intro or "").strip()
|
||
|
||
def _get_custom_intro_pool(world_data):
|
||
pool = {}
|
||
raw = None
|
||
if isinstance(world_data, dict):
|
||
raw = world_data.get("CUSTOM_INTRO") or world_data.get("custom_intro")
|
||
|
||
# dict -> keys "0".."6"
|
||
if isinstance(raw, dict):
|
||
for k in [str(i) for i in range(7)]:
|
||
v = raw.get(k)
|
||
if isinstance(v, str) and v.strip():
|
||
pool[k] = v.strip()
|
||
|
||
# list -> indices 0..6
|
||
elif isinstance(raw, list):
|
||
for i in range(min(7, len(raw))):
|
||
v = raw[i]
|
||
if isinstance(v, str) and v.strip():
|
||
pool[str(i)] = v.strip()
|
||
|
||
return pool
|
||
|
||
if not ci:
|
||
pool = _get_custom_intro_pool(data)
|
||
|
||
if custom_intro_id and custom_intro_id != "Random":
|
||
chosen = pool.get(str(custom_intro_id))
|
||
if chosen:
|
||
ci = chosen
|
||
|
||
if not ci and pool:
|
||
ci = random.choice(list(pool.values()))
|
||
|
||
if ci:
|
||
parts.append(ci)
|
||
|
||
else:
|
||
gender_inserted = False
|
||
if isinstance(data, dict) and gender in data:
|
||
values = data[gender]
|
||
if isinstance(values, list) and values:
|
||
parts.append(random.choice(values))
|
||
gender_inserted = True
|
||
|
||
if not gender_inserted:
|
||
gender_defaults = {
|
||
"male": "a mysterious man",
|
||
"female": "a beautiful woman",
|
||
"2 female": "two beautiful womans",
|
||
"3 female": "three unique beautiful womans",
|
||
"female vampire": "a beautiful vampire woman",
|
||
"anime woman": "a beautiful anime woman",
|
||
}
|
||
parts.append(gender_defaults.get(gender, "a striking figure"))
|
||
|
||
# Optional color realm
|
||
|
||
|
||
color_realm_value = None
|
||
if "COLOR_REALM" in data:
|
||
possible_realms = data["COLOR_REALM"]
|
||
if isinstance(possible_realms, list) and possible_realms:
|
||
color_realm_value = random.choice(possible_realms)
|
||
print(f"[PromptCreator] Using COLOR_REALM: {color_realm_value}")
|
||
parts.append(color_realm_value)
|
||
|
||
multi_keys = ["OBJECTS", "ACCESSORIES"]
|
||
|
||
# Pick values according to realm or flat lists
|
||
if color_realm_value:
|
||
for key in ["OUTFITS", "LIGHTING", "BACKGROUNDS", "OBJECTS", "ACCESSORIES", "ATMOSPHERES"]:
|
||
values_by_realm = data.get(key, {}).get(color_realm_value, [])
|
||
if isinstance(values_by_realm, list) and values_by_realm:
|
||
if key in multi_keys:
|
||
sampled = random.sample(values_by_realm, min(multi_object_count, len(values_by_realm)))
|
||
parts.extend(sampled)
|
||
else:
|
||
parts.append(random.choice(values_by_realm))
|
||
else:
|
||
for key, values in data.items():
|
||
if key.lower() in [
|
||
"male", "female", "neutral",
|
||
"horror_intensity", "sensuality_level",
|
||
"color_realm",
|
||
"system_prompt", "world_name",
|
||
"custom_intro",
|
||
"camera_angles", "camera_angle",
|
||
"poses"
|
||
]:
|
||
continue
|
||
if isinstance(values, list) and values:
|
||
if key in multi_keys:
|
||
sampled = random.sample(values, min(multi_object_count, len(values)))
|
||
parts.extend(sampled)
|
||
else:
|
||
parts.append(random.choice(values))
|
||
|
||
# Horror intensity
|
||
if horror_intensity != "auto" and "HORROR_INTENSITY" in data:
|
||
try:
|
||
horror_level = int(horror_intensity)
|
||
intensity_entries = data["HORROR_INTENSITY"]
|
||
if isinstance(intensity_entries, dict):
|
||
matching = intensity_entries.get(str(horror_level))
|
||
if matching:
|
||
parts.append(matching)
|
||
except ValueError:
|
||
pass
|
||
|
||
# Sensuality level (modulatore simile a HORROR_INTENSITY)
|
||
if sensuality_level != "auto" and "SENSUALITY_LEVEL" in data:
|
||
try:
|
||
s_level = int(sensuality_level)
|
||
s_entries = data["SENSUALITY_LEVEL"]
|
||
if isinstance(s_entries, dict):
|
||
matching = s_entries.get(str(s_level))
|
||
if matching:
|
||
parts.append(matching)
|
||
except ValueError:
|
||
pass
|
||
|
||
|
||
if subject_count != "1":
|
||
parts.append(f"{subject_count} subjects present")
|
||
|
||
return ", ".join(parts)
|
||
|
||
def _read_api_keys(self, base_path):
|
||
keys_path = os.path.join(base_path, "api_keys.txt")
|
||
keys = {}
|
||
if os.path.exists(keys_path):
|
||
with open(keys_path, "r", encoding="utf-8") as f:
|
||
for line in f:
|
||
if "=" in line:
|
||
k, v = line.strip().split("=", 1)
|
||
keys[k.strip()] = v.strip()
|
||
return keys
|
||
|
||
def _enhance_with_ollama(self, host, model, system_prompt, user_prompt):
|
||
# Prefer chat to support system role
|
||
r = requests.post(f"{host}/api/chat", json={
|
||
"model": model,
|
||
"messages": [
|
||
{"role": "system", "content": system_prompt},
|
||
{"role": "user", "content": user_prompt}
|
||
],
|
||
"stream": False
|
||
}, timeout=120)
|
||
r.raise_for_status()
|
||
return r.json()["message"]["content"].strip()
|
||
|
||
def _enhance_with_openai(self, base_path, system_prompt, user_prompt):
|
||
import openai
|
||
keys = self._read_api_keys(base_path)
|
||
openai.api_key = keys.get("openai", "")
|
||
resp = openai.ChatCompletion.create(
|
||
model="gpt-3.5-turbo",
|
||
messages=[
|
||
{"role": "system", "content": system_prompt},
|
||
{"role": "user", "content": user_prompt}
|
||
]
|
||
)
|
||
return resp.choices[0].message.content.strip()
|
||
|
||
import requests
|
||
|
||
def _enhance_with_llamacpp(self, host, system_prompt, user_prompt, temperature=0.7, top_p=0.9, n_predict=220):
|
||
"""
|
||
llama.cpp server backend.
|
||
Tries OpenAI-compatible endpoint first (/v1/chat/completions), then falls back to (/completion).
|
||
host example: http://127.0.0.1:11434
|
||
"""
|
||
|
||
host = host.rstrip("/")
|
||
|
||
# 1) Try OpenAI-compatible chat endpoint
|
||
try:
|
||
r = requests.post(
|
||
f"{host}/v1/chat/completions",
|
||
json={
|
||
"model": "llama", # ignored by llama.cpp in many builds; kept for compatibility
|
||
"messages": [
|
||
{"role": "system", "content": system_prompt},
|
||
{"role": "user", "content": user_prompt},
|
||
],
|
||
"temperature": float(temperature),
|
||
"top_p": float(top_p),
|
||
"max_tokens": int(n_predict),
|
||
"stream": False,
|
||
},
|
||
timeout=120,
|
||
)
|
||
|
||
if r.status_code == 200:
|
||
j = r.json()
|
||
# OpenAI-like: choices[0].message.content
|
||
content = (
|
||
j.get("choices", [{}])[0]
|
||
.get("message", {})
|
||
.get("content", "")
|
||
)
|
||
content = (content or "").strip()
|
||
if content:
|
||
return content
|
||
|
||
# if non-200, fall through to fallback
|
||
except Exception:
|
||
pass
|
||
|
||
# 2) Fallback to llama.cpp native /completion
|
||
r = requests.post(
|
||
f"{host}/completion",
|
||
json={
|
||
"prompt": f"{system_prompt}\n\n{user_prompt}",
|
||
"temperature": float(temperature),
|
||
"top_p": float(top_p),
|
||
"n_predict": int(n_predict),
|
||
"stream": False,
|
||
},
|
||
timeout=120,
|
||
)
|
||
r.raise_for_status()
|
||
j = r.json()
|
||
|
||
# llama.cpp /completion commonly returns: {"content": "..."} or {"completion": "..."} depending on version
|
||
content = (j.get("content") or j.get("completion") or "").strip()
|
||
if not content:
|
||
raise RuntimeError(f"llama.cpp returned no content. Keys: {list(j.keys())}")
|
||
return content
|
||
|
||
|
||
def _enhance_with_cohere(self, base_path, system_prompt, user_prompt):
|
||
import re
|
||
try:
|
||
import cohere
|
||
except ImportError:
|
||
print("[PromptCreator] Cohere SDK not installed, skipping enhance.")
|
||
return user_prompt
|
||
|
||
keys = self._read_api_keys(base_path)
|
||
api_key = keys.get("cohere", "").strip()
|
||
if not api_key:
|
||
print("[PromptCreator] No Cohere API key found, skipping enhance.")
|
||
return user_prompt
|
||
|
||
# Usa il client V2 (nuova API)
|
||
try:
|
||
co = cohere.ClientV2(api_key=api_key)
|
||
except Exception as e:
|
||
print(f"[PromptCreator] Cohere ClientV2 not available: {e}")
|
||
return user_prompt
|
||
|
||
# Messaggi Chat v2
|
||
system_msg = (
|
||
f"{system_prompt}\n\n"
|
||
"You are a prompt enhancer for image generation. "
|
||
"Expand the user seed into ONE single, long, flowing sentence "
|
||
"with multiple clauses separated by commas and semicolons. "
|
||
"Do NOT use bullet points or numbered lists. "
|
||
"Keep the original mood and style."
|
||
)
|
||
|
||
user_msg = (
|
||
f"Seed: {user_prompt}\n\n"
|
||
"Return exactly one long sentence (around 100 to 150 words), no line breaks."
|
||
)
|
||
|
||
# === BLOCCO TRY PRINCIPALE ===
|
||
try:
|
||
resp = co.chat(
|
||
model="command-r-08-2024",
|
||
messages=[
|
||
{"role": "system", "content": system_prompt},
|
||
{"role": "user", "content": user_msg},
|
||
],
|
||
max_tokens=700,
|
||
temperature=0.9,
|
||
)
|
||
|
||
content = ""
|
||
if hasattr(resp, "message") and getattr(resp.message, "content", None):
|
||
parts = resp.message.content
|
||
texts = [getattr(p, "text", "") for p in parts if getattr(p, "text", None)]
|
||
content = " ".join(texts).strip()
|
||
|
||
if not content:
|
||
print("[PromptCreator] Cohere returned empty content.")
|
||
return user_prompt
|
||
|
||
content = re.sub(r"\s*\n+\s*", " ", content).strip()
|
||
return content
|
||
|
||
# === ECCO IL BLOCCO EXCEPT CHE ORA È ALLINEATO PERFETTAMENTE ===
|
||
except Exception as e:
|
||
print(f"[PromptCreator] Error calling Cohere chat: {e}")
|
||
return user_prompt
|
||
|
||
|
||
def _enhance_with_gemini(self, base_path, system_prompt, user_prompt):
|
||
import google.generativeai as genai
|
||
keys = self._read_api_keys(base_path)
|
||
genai.configure(api_key=keys.get("gemini", ""))
|
||
# In Gemini, passiamo tutto come testo
|
||
model = genai.GenerativeModel("models/gemini-2.5-pro")
|
||
txt = (
|
||
f"SYSTEM: {system_prompt}\n"
|
||
f"USER: {user_prompt}\n"
|
||
"Return only the final prompt as a single paragraph."
|
||
)
|
||
r = model.generate_content(txt)
|
||
return r.candidates[0].content.parts[0].text.strip()
|
||
|
||
@staticmethod
|
||
def log_prompt_run(
|
||
json_name,
|
||
enhancer_mode,
|
||
gender,
|
||
custom_intro,
|
||
lora_triggers,
|
||
final_prompt,
|
||
source="generated",
|
||
node_version="1.5.0",
|
||
):
|
||
"""Append one JSONL entry per run to ./logs/prompt_history.jsonl."""
|
||
log_dir = os.path.join(os.path.dirname(__file__), "logs")
|
||
os.makedirs(log_dir, exist_ok=True)
|
||
|
||
log_path = os.path.join(log_dir, "prompt_history.jsonl")
|
||
|
||
entry = {
|
||
"timestamp": datetime.now().isoformat(timespec="seconds"),
|
||
"json_world": json_name,
|
||
"system_prompt": enhancer_mode,
|
||
"gender": gender,
|
||
"custom_intro": custom_intro.strip() if (gender == "custom" and isinstance(custom_intro, str) and custom_intro.strip()) else None,
|
||
"lora_triggers": lora_triggers,
|
||
"final_prompt": final_prompt,
|
||
"source": source,
|
||
"node_version": node_version
|
||
}
|
||
|
||
with open(log_path, "a", encoding="utf-8") as f:
|
||
f.write(json.dumps(entry, ensure_ascii=False) + "\n")
|
||
|
||
|
||
|
||
|
||
def generate_prompt(self, json_name, camera_angle, use_enhancer, enhancer_mode, system_prompt_lock, add_symbols, seed, gender, identity_profile, external_identity, lock_identity, custom_intro, custom_intro_id, horror_intensity, sensuality_level, pose_mode, pose_index, lora_triggers, subject_count, lock_last_prompt, multi_object_count, ollama_host, ollama_model):
|
||
base_path = os.path.dirname(__file__)
|
||
json_path = os.path.join(base_path, "JSON_DATA", json_name)
|
||
|
||
# history lock
|
||
history_path = os.path.join(base_path, "history", f"last_prompt_{json_name}.txt")
|
||
if lock_last_prompt == "yes" and os.path.exists(history_path):
|
||
with open(history_path, "r", encoding="utf-8") as f:
|
||
prompt = f.read().strip()
|
||
self.log_prompt_run(
|
||
json_name=json_name,
|
||
enhancer_mode=enhancer_mode,
|
||
gender=gender,
|
||
custom_intro=custom_intro,
|
||
lora_triggers=lora_triggers,
|
||
final_prompt=prompt,
|
||
source="history_lock",
|
||
node_version="1.5.0"
|
||
)
|
||
print("[PromptCreator] Prompt locked and loaded from history")
|
||
return (prompt,)
|
||
|
||
try:
|
||
with open(json_path, "r", encoding="utf-8") as f:
|
||
data = json.load(f)
|
||
except Exception as e:
|
||
print(f"[PromptCreator] Errore nel caricamento del JSON: {e}")
|
||
return ("",)
|
||
|
||
# world-level SYSTEM_PROMPT (unified JSON) if present
|
||
world_system_prompt = None
|
||
if isinstance(data, dict):
|
||
wsp = data.get("SYSTEM_PROMPT") or data.get("system_prompt")
|
||
if isinstance(wsp, str) and wsp.strip():
|
||
world_system_prompt = wsp.strip()
|
||
# Camera angle resolution (UI overrides world)
|
||
angle_dict = self._camera_angles()
|
||
camera_angle_txt = self._resolve_camera_angle(camera_angle, data, angle_dict)
|
||
data = self._sanitize_world_camera(data, camera_angle)
|
||
|
||
# Seed opzionale per coerenza
|
||
if seed:
|
||
random.seed(seed)
|
||
|
||
# Identity profile (prompt-only consistency)
|
||
identity_txt = ""
|
||
if lock_identity == "yes":
|
||
if identity_profile == "external":
|
||
identity_txt = (external_identity or "").strip()
|
||
elif identity_profile != "none":
|
||
identities = self._identity_profiles()
|
||
identity_txt = self._identity_to_text(identities.get(identity_profile, {}))
|
||
print("[PromptCreator][DEBUG] identity_profile:", identity_profile)
|
||
print("[PromptCreator][DEBUG] lock_identity:", lock_identity)
|
||
print("[PromptCreator][DEBUG] identity_txt:", repr(identity_txt))
|
||
|
||
|
||
|
||
# Costruisci prompt base
|
||
prompt = self._build_prompt_from_json(
|
||
data, gender, custom_intro, custom_intro_id, horror_intensity, sensuality_level, subject_count, multi_object_count, camera_angle_txt=camera_angle_txt
|
||
)
|
||
# --- POSE CONTROL ---
|
||
poses = data.get("POSES") or data.get("poses") or []
|
||
poses = poses if isinstance(poses, list) else []
|
||
|
||
pose_path = os.path.join(base_path, "history", f"last_pose_{json_name}.txt")
|
||
|
||
def pick_pose_by_index(i):
|
||
if not poses:
|
||
return ""
|
||
i = max(0, min(int(i), len(poses) - 1))
|
||
return str(poses[i]).strip()
|
||
|
||
chosen_pose = ""
|
||
|
||
if pose_mode == "world_pick":
|
||
chosen_pose = pick_pose_by_index(pose_index)
|
||
|
||
elif pose_mode == "lock":
|
||
if os.path.exists(pose_path):
|
||
with open(pose_path, "r", encoding="utf-8") as f:
|
||
chosen_pose = f.read().strip()
|
||
if not chosen_pose:
|
||
chosen_pose = pick_pose_by_index(pose_index)
|
||
|
||
else: # random
|
||
if poses:
|
||
chosen_pose = random.choice(poses).strip()
|
||
|
||
if chosen_pose:
|
||
prompt = prompt + ", " + chosen_pose
|
||
os.makedirs(os.path.dirname(pose_path), exist_ok=True)
|
||
with open(pose_path, "w", encoding="utf-8") as f:
|
||
f.write(chosen_pose)
|
||
# --- END POSE CONTROL ---
|
||
|
||
if identity_txt:
|
||
prompt = prompt + ", " + identity_txt
|
||
print("[PromptCreator][DEBUG] prompt_pre_enhancer:", prompt)
|
||
|
||
|
||
# System prompt condiviso per tutti i backend
|
||
system_prompts = self._system_prompts()
|
||
effective_world_system_prompt = None if system_prompt_lock == "external" else world_system_prompt
|
||
system_prompt = effective_world_system_prompt or system_prompts.get(enhancer_mode, system_prompts["standard"])
|
||
user_prompt = f"Enhance this prompt: {prompt}"
|
||
|
||
# Enhancer backends
|
||
if use_enhancer != "none":
|
||
print("[PromptCreator][DEBUG] use_enhancer:", use_enhancer)
|
||
print("[PromptCreator][DEBUG] enhancer_mode:", enhancer_mode)
|
||
print("[PromptCreator][DEBUG] system_prompt:", system_prompt)
|
||
print("[PromptCreator][DEBUG] user_prompt:", user_prompt)
|
||
|
||
try:
|
||
if use_enhancer == "ollama":
|
||
prompt = self._enhance_with_ollama(ollama_host, ollama_model, system_prompt, user_prompt)
|
||
elif use_enhancer == "llamacpp":
|
||
prompt = self._enhance_with_llamacpp(ollama_host, system_prompt, user_prompt)
|
||
elif use_enhancer == "openai":
|
||
prompt = self._enhance_with_openai(base_path, system_prompt, user_prompt)
|
||
elif use_enhancer == "cohere":
|
||
prompt = self._enhance_with_cohere(base_path, system_prompt, user_prompt)
|
||
elif use_enhancer == "gemini":
|
||
prompt = self._enhance_with_gemini(base_path, system_prompt, user_prompt)
|
||
except Exception as e:
|
||
print(f"[PromptCreator] Errore nell'enhancer ({use_enhancer}): {e}")
|
||
|
||
# LoRA triggers
|
||
if lora_triggers.strip():
|
||
lts = [x.strip() for x in lora_triggers.split(",") if x.strip()]
|
||
if lts:
|
||
prompt += ", " + ", ".join(lts)
|
||
print(f"[PromptCreator] LoRA triggers aggiunti: {lts}")
|
||
|
||
if add_symbols == "yes":
|
||
prompt = f"[{prompt}]"
|
||
|
||
os.makedirs(os.path.join(base_path, "history"), exist_ok=True)
|
||
with open(history_path, "w", encoding="utf-8") as f:
|
||
f.write(prompt.strip())
|
||
|
||
print(f"[PromptCreator] Prompt finale: {prompt}")
|
||
self.log_prompt_run(
|
||
json_name=json_name,
|
||
enhancer_mode=enhancer_mode,
|
||
gender=gender,
|
||
custom_intro=custom_intro,
|
||
lora_triggers=lora_triggers,
|
||
final_prompt=prompt,
|
||
source="generated",
|
||
node_version="1.5.0"
|
||
)
|
||
return (prompt,)
|