Files
character-browser/services/workflow.py
2026-03-15 17:45:17 +00:00

611 lines
23 KiB
Python

import json
import logging
import random
from flask import session
from models import Settings, Checkpoint
from utils import _resolve_lora_weight
from services.prompts import _cross_dedup_prompts
logger = logging.getLogger('gaze')
# Node IDs used by DetailerForEach in multi-char mode
_SEGS_DETAILER_NODES = ['46', '47', '53', '54']
# Node IDs for per-character CLIP prompts in multi-char mode
_SEGS_PROMPT_NODES = ['44', '45', '51', '52']
def _log_workflow_prompts(label, workflow):
"""Log the final assembled ComfyUI prompts in a consistent, readable block."""
sep = "=" * 72
active_loras = []
lora_details = []
# Collect detailed LoRA information
for node_id, label_str in [("16", "char/look"), ("17", "outfit"), ("18", "action"), ("19", "style/detail/scene"), ("20", "char_b")]:
if node_id in workflow:
name = workflow[node_id]["inputs"].get("lora_name", "")
if name:
strength_model = workflow[node_id]["inputs"].get("strength_model", "?")
strength_clip = workflow[node_id]["inputs"].get("strength_clip", "?")
# Short version for summary
if isinstance(strength_model, float):
active_loras.append(f"{label_str}:{name.split('/')[-1]}@{strength_model:.3f}")
else:
active_loras.append(f"{label_str}:{name.split('/')[-1]}@{strength_model}")
# Detailed version
lora_details.append(f" Node {node_id} ({label_str}): {name}")
lora_details.append(f" strength_model={strength_model}, strength_clip={strength_clip}")
# Extract VAE information
vae_info = "(integrated)"
if '21' in workflow:
vae_info = workflow['21']['inputs'].get('vae_name', '(custom)')
# Extract adetailer information
adetailer_info = []
# Single-char mode: FaceDetailer nodes 11 + 13
for node_id, node_name in [("11", "Face"), ("13", "Hand")]:
if node_id in workflow:
adetailer_info.append(f" {node_name} (Node {node_id}): steps={workflow[node_id]['inputs'].get('steps', '?')}, "
f"cfg={workflow[node_id]['inputs'].get('cfg', '?')}, "
f"denoise={workflow[node_id]['inputs'].get('denoise', '?')}")
# Multi-char mode: SEGS DetailerForEach nodes
for node_id, node_name in [("46", "Person A"), ("47", "Person B"), ("53", "Face A"), ("54", "Face B")]:
if node_id in workflow:
adetailer_info.append(f" {node_name} (Node {node_id}): steps={workflow[node_id]['inputs'].get('steps', '?')}, "
f"cfg={workflow[node_id]['inputs'].get('cfg', '?')}, "
f"denoise={workflow[node_id]['inputs'].get('denoise', '?')}")
face_text = workflow.get('14', {}).get('inputs', {}).get('text', '')
hand_text = workflow.get('15', {}).get('inputs', {}).get('text', '')
lines = [
sep,
f" WORKFLOW PROMPTS [{label}]",
sep,
" MODEL CONFIGURATION:",
f" Checkpoint : {workflow['4']['inputs'].get('ckpt_name', '(not set)')}",
f" VAE : {vae_info}",
"",
" GENERATION SETTINGS:",
f" Seed : {workflow['3']['inputs'].get('seed', '(not set)')}",
f" Resolution : {workflow['5']['inputs'].get('width', '?')} x {workflow['5']['inputs'].get('height', '?')}",
f" Sampler : {workflow['3']['inputs'].get('sampler_name', '?')} / {workflow['3']['inputs'].get('scheduler', '?')}",
f" Steps : {workflow['3']['inputs'].get('steps', '?')}",
f" CFG Scale : {workflow['3']['inputs'].get('cfg', '?')}",
f" Denoise : {workflow['3']['inputs'].get('denoise', '1.0')}",
]
# Add LoRA details
if active_loras:
lines.append("")
lines.append(" LORA CONFIGURATION:")
lines.extend(lora_details)
else:
lines.append("")
lines.append(" LORA CONFIGURATION: (none)")
# Add adetailer details
if adetailer_info:
lines.append("")
lines.append(" ADETAILER CONFIGURATION:")
lines.extend(adetailer_info)
# Add prompts
lines.extend([
"",
" PROMPTS:",
f" [+] Positive : {workflow['6']['inputs'].get('text', '')}",
f" [-] Negative : {workflow['7']['inputs'].get('text', '')}",
])
if face_text:
lines.append(f" [F] Face : {face_text}")
if hand_text:
lines.append(f" [H] Hand : {hand_text}")
# Multi-char per-character prompts
for node_id, lbl in [("44", "Person A"), ("45", "Person B"), ("51", "Face A"), ("52", "Face B")]:
txt = workflow.get(node_id, {}).get('inputs', {}).get('text', '')
if txt:
lines.append(f" [{lbl}] : {txt}")
lines.append(sep)
logger.info("\n%s", "\n".join(lines))
def _apply_checkpoint_settings(workflow, ckpt_data):
"""Apply checkpoint-specific sampler/prompt/VAE settings to the workflow."""
steps = ckpt_data.get('steps')
cfg = ckpt_data.get('cfg')
sampler_name = ckpt_data.get('sampler_name')
scheduler = ckpt_data.get('scheduler')
base_positive = ckpt_data.get('base_positive', '')
base_negative = ckpt_data.get('base_negative', '')
vae = ckpt_data.get('vae', 'integrated')
# KSampler (node 3)
if steps and '3' in workflow:
workflow['3']['inputs']['steps'] = int(steps)
if cfg and '3' in workflow:
workflow['3']['inputs']['cfg'] = float(cfg)
if sampler_name and '3' in workflow:
workflow['3']['inputs']['sampler_name'] = sampler_name
if scheduler and '3' in workflow:
workflow['3']['inputs']['scheduler'] = scheduler
# Face/hand detailers (nodes 11, 13) + multi-char SEGS detailers
for node_id in ['11', '13'] + _SEGS_DETAILER_NODES:
if node_id in workflow:
if steps:
workflow[node_id]['inputs']['steps'] = int(steps)
if cfg:
workflow[node_id]['inputs']['cfg'] = float(cfg)
if sampler_name:
workflow[node_id]['inputs']['sampler_name'] = sampler_name
if scheduler:
workflow[node_id]['inputs']['scheduler'] = scheduler
# Prepend base_positive to all positive prompt nodes
if base_positive:
for node_id in ['6', '14', '15'] + _SEGS_PROMPT_NODES:
if node_id in workflow:
workflow[node_id]['inputs']['text'] = f"{base_positive}, {workflow[node_id]['inputs']['text']}"
# Append base_negative to negative prompt (shared by main + detailers via node 7)
if base_negative and '7' in workflow:
workflow['7']['inputs']['text'] = f"{workflow['7']['inputs']['text']}, {base_negative}"
# VAE: if not integrated, inject a VAELoader node and rewire
if vae and vae != 'integrated':
workflow['21'] = {
'inputs': {'vae_name': vae},
'class_type': 'VAELoader'
}
if '8' in workflow:
workflow['8']['inputs']['vae'] = ['21', 0]
for node_id in ['11', '13'] + _SEGS_DETAILER_NODES:
if node_id in workflow:
workflow[node_id]['inputs']['vae'] = ['21', 0]
return workflow
def _get_default_checkpoint():
"""Return (checkpoint_path, checkpoint_data) from the database Settings, session, or fall back to workflow file."""
ckpt_path = session.get('default_checkpoint')
# If no session checkpoint, try to read from database Settings
if not ckpt_path:
settings = Settings.query.first()
if settings and settings.default_checkpoint:
ckpt_path = settings.default_checkpoint
logger.debug("Loaded default checkpoint from database: %s", ckpt_path)
# If still no checkpoint, try to read from the workflow file
if not ckpt_path:
try:
with open('comfy_workflow.json', 'r') as f:
workflow = json.load(f)
ckpt_path = workflow.get('4', {}).get('inputs', {}).get('ckpt_name')
logger.debug("Loaded default checkpoint from workflow file: %s", ckpt_path)
except Exception:
pass
if not ckpt_path:
return None, None
ckpt = Checkpoint.query.filter_by(checkpoint_path=ckpt_path).first()
if not ckpt:
# Checkpoint path exists but not in DB - return path with empty data
return ckpt_path, {}
return ckpt.checkpoint_path, ckpt.data or {}
def _inject_multi_char_detailers(workflow, prompts, model_source, clip_source):
"""Replace single FaceDetailer (node 11) with per-character SEGS-based detailers.
Injects person detection + face detection pipelines that order detections
left-to-right and apply character A's prompt to the left person/face and
character B's prompt to the right person/face.
Nodes added:
40 - Person detector (UltralyticsDetectorProvider)
41 - Person SEGS detection (BboxDetectorSEGS)
42 - Filter: left person (char A)
43 - Filter: right person (char B)
44 - CLIPTextEncode: char A body prompt
45 - CLIPTextEncode: char B body prompt
46 - DetailerForEach: person A
47 - DetailerForEach: person B
48 - Face SEGS detection (BboxDetectorSEGS, reuses face detector node 10)
49 - Filter: left face (char A)
50 - Filter: right face (char B)
51 - CLIPTextEncode: char A face prompt
52 - CLIPTextEncode: char B face prompt
53 - DetailerForEach: face A
54 - DetailerForEach: face B
Image flow: VAEDecode(8) → PersonA(46) → PersonB(47) → FaceA(53) → FaceB(54) → Hand(13)
"""
vae_source = ["4", 2]
# Remove old single face detailer and its prompt — we replace them
workflow.pop('11', None)
workflow.pop('14', None)
# --- Person detection ---
workflow['40'] = {
'inputs': {'model_name': 'segm/person_yolov8m-seg.pt'},
'class_type': 'UltralyticsDetectorProvider'
}
workflow['41'] = {
'inputs': {
'bbox_detector': ['40', 0],
'image': ['8', 0],
'threshold': 0.5,
'dilation': 10,
'crop_factor': 3.0,
'drop_size': 10,
'labels': 'all',
},
'class_type': 'BboxDetectorSEGS'
}
# Order by x1 ascending (left to right), pick index 0 = leftmost person
workflow['42'] = {
'inputs': {
'segs': ['41', 0],
'target': 'x1',
'order': False,
'take_start': 0,
'take_count': 1,
},
'class_type': 'ImpactSEGSOrderedFilter'
}
# Pick index 1 = rightmost person
workflow['43'] = {
'inputs': {
'segs': ['41', 0],
'target': 'x1',
'order': False,
'take_start': 1,
'take_count': 1,
},
'class_type': 'ImpactSEGSOrderedFilter'
}
# --- Per-character body prompts ---
workflow['44'] = {
'inputs': {'text': prompts.get('char_a_main', ''), 'clip': clip_source},
'class_type': 'CLIPTextEncode'
}
workflow['45'] = {
'inputs': {'text': prompts.get('char_b_main', ''), 'clip': clip_source},
'class_type': 'CLIPTextEncode'
}
# --- Person detailing (DetailerForEach) ---
_person_base = {
'guide_size': 512,
'guide_size_for': True,
'max_size': 1024,
'seed': 0, # overwritten by seed step
'steps': 20, # overwritten by checkpoint settings
'cfg': 3.5, # overwritten by checkpoint settings
'sampler_name': 'euler_ancestral',
'scheduler': 'normal',
'denoise': 0.4,
'feather': 5,
'noise_mask': True,
'force_inpaint': True,
'wildcard': '',
'cycle': 1,
'inpaint_model': False,
'noise_mask_feather': 20,
}
workflow['46'] = {
'inputs': {
**_person_base,
'image': ['8', 0],
'segs': ['42', 0],
'model': model_source,
'clip': clip_source,
'vae': vae_source,
'positive': ['44', 0],
'negative': ['7', 0],
},
'class_type': 'DetailerForEach'
}
workflow['47'] = {
'inputs': {
**_person_base,
'image': ['46', 0], # chains from person A output
'segs': ['43', 0],
'model': model_source,
'clip': clip_source,
'vae': vae_source,
'positive': ['45', 0],
'negative': ['7', 0],
},
'class_type': 'DetailerForEach'
}
# --- Face detection (on person-detailed image) ---
workflow['48'] = {
'inputs': {
'bbox_detector': ['10', 0], # reuse existing face YOLO detector
'image': ['47', 0],
'threshold': 0.5,
'dilation': 10,
'crop_factor': 3.0,
'drop_size': 10,
'labels': 'all',
},
'class_type': 'BboxDetectorSEGS'
}
workflow['49'] = {
'inputs': {
'segs': ['48', 0],
'target': 'x1',
'order': False,
'take_start': 0,
'take_count': 1,
},
'class_type': 'ImpactSEGSOrderedFilter'
}
workflow['50'] = {
'inputs': {
'segs': ['48', 0],
'target': 'x1',
'order': False,
'take_start': 1,
'take_count': 1,
},
'class_type': 'ImpactSEGSOrderedFilter'
}
# --- Per-character face prompts ---
workflow['51'] = {
'inputs': {'text': prompts.get('char_a_face', ''), 'clip': clip_source},
'class_type': 'CLIPTextEncode'
}
workflow['52'] = {
'inputs': {'text': prompts.get('char_b_face', ''), 'clip': clip_source},
'class_type': 'CLIPTextEncode'
}
# --- Face detailing (DetailerForEach) ---
_face_base = {
'guide_size': 384,
'guide_size_for': True,
'max_size': 1024,
'seed': 0,
'steps': 20,
'cfg': 3.5,
'sampler_name': 'euler_ancestral',
'scheduler': 'normal',
'denoise': 0.5,
'feather': 5,
'noise_mask': True,
'force_inpaint': True,
'wildcard': '',
'cycle': 1,
'inpaint_model': False,
'noise_mask_feather': 20,
}
workflow['53'] = {
'inputs': {
**_face_base,
'image': ['47', 0],
'segs': ['49', 0],
'model': model_source,
'clip': clip_source,
'vae': vae_source,
'positive': ['51', 0],
'negative': ['7', 0],
},
'class_type': 'DetailerForEach'
}
workflow['54'] = {
'inputs': {
**_face_base,
'image': ['53', 0], # chains from face A output
'segs': ['50', 0],
'model': model_source,
'clip': clip_source,
'vae': vae_source,
'positive': ['52', 0],
'negative': ['7', 0],
},
'class_type': 'DetailerForEach'
}
# Rewire hand detailer: image input from last face detailer instead of old node 11
if '13' in workflow:
workflow['13']['inputs']['image'] = ['54', 0]
logger.debug("Injected multi-char SEGS detailers (nodes 40-54)")
def _prepare_workflow(workflow, character, prompts, checkpoint=None, custom_negative=None, outfit=None, action=None, style=None, detailer=None, scene=None, width=None, height=None, checkpoint_data=None, look=None, fixed_seed=None, character_b=None):
# 1. Update prompts using replacement to preserve embeddings
workflow["6"]["inputs"]["text"] = workflow["6"]["inputs"]["text"].replace("{{POSITIVE_PROMPT}}", prompts["main"])
if custom_negative:
workflow["7"]["inputs"]["text"] = f"{custom_negative}, {workflow['7']['inputs']['text']}"
if "14" in workflow:
workflow["14"]["inputs"]["text"] = workflow["14"]["inputs"]["text"].replace("{{FACE_PROMPT}}", prompts["face"])
if "15" in workflow:
workflow["15"]["inputs"]["text"] = workflow["15"]["inputs"]["text"].replace("{{HAND_PROMPT}}", prompts["hand"])
# 2. Update Checkpoint - always set one, fall back to default if not provided
if not checkpoint:
default_ckpt, default_ckpt_data = _get_default_checkpoint()
checkpoint = default_ckpt
if not checkpoint_data:
checkpoint_data = default_ckpt_data
if checkpoint:
workflow["4"]["inputs"]["ckpt_name"] = checkpoint
else:
raise ValueError("No checkpoint specified and no default checkpoint configured")
# 3. Handle LoRAs - Node 16 for character, Node 17 for outfit, Node 18 for action, Node 19 for style/detailer
# Start with direct checkpoint connections
model_source = ["4", 0]
clip_source = ["4", 1]
# Look negative prompt (applied before character LoRA)
if look:
look_negative = look.data.get('negative', '')
if look_negative:
workflow["7"]["inputs"]["text"] = f"{look_negative}, {workflow['7']['inputs']['text']}"
# Character LoRA (Node 16) — look LoRA overrides character LoRA when present
if look:
char_lora_data = look.data.get('lora', {})
else:
char_lora_data = character.data.get('lora', {}) if character else {}
char_lora_name = char_lora_data.get('lora_name')
if char_lora_name and "16" in workflow:
_w16 = _resolve_lora_weight(char_lora_data)
workflow["16"]["inputs"]["lora_name"] = char_lora_name
workflow["16"]["inputs"]["strength_model"] = _w16
workflow["16"]["inputs"]["strength_clip"] = _w16
workflow["16"]["inputs"]["model"] = ["4", 0] # From checkpoint
workflow["16"]["inputs"]["clip"] = ["4", 1] # From checkpoint
model_source = ["16", 0]
clip_source = ["16", 1]
logger.debug("Character LoRA: %s @ %s", char_lora_name, _w16)
# Outfit LoRA (Node 17) - chains from character LoRA or checkpoint
outfit_lora_data = outfit.data.get('lora', {}) if outfit else {}
outfit_lora_name = outfit_lora_data.get('lora_name')
if outfit_lora_name and "17" in workflow:
_w17 = _resolve_lora_weight({**{'lora_weight': 0.8}, **outfit_lora_data})
workflow["17"]["inputs"]["lora_name"] = outfit_lora_name
workflow["17"]["inputs"]["strength_model"] = _w17
workflow["17"]["inputs"]["strength_clip"] = _w17
# Chain from character LoRA (node 16) or checkpoint (node 4)
workflow["17"]["inputs"]["model"] = model_source
workflow["17"]["inputs"]["clip"] = clip_source
model_source = ["17", 0]
clip_source = ["17", 1]
logger.debug("Outfit LoRA: %s @ %s", outfit_lora_name, _w17)
# Action LoRA (Node 18) - chains from previous LoRA or checkpoint
action_lora_data = action.data.get('lora', {}) if action else {}
action_lora_name = action_lora_data.get('lora_name')
if action_lora_name and "18" in workflow:
_w18 = _resolve_lora_weight(action_lora_data)
workflow["18"]["inputs"]["lora_name"] = action_lora_name
workflow["18"]["inputs"]["strength_model"] = _w18
workflow["18"]["inputs"]["strength_clip"] = _w18
# Chain from previous source
workflow["18"]["inputs"]["model"] = model_source
workflow["18"]["inputs"]["clip"] = clip_source
model_source = ["18", 0]
clip_source = ["18", 1]
logger.debug("Action LoRA: %s @ %s", action_lora_name, _w18)
# Style/Detailer/Scene LoRA (Node 19) - chains from previous LoRA or checkpoint
# Priority: Style > Detailer > Scene (Scene LoRAs are rare but supported)
target_obj = style or detailer or scene
style_lora_data = target_obj.data.get('lora', {}) if target_obj else {}
style_lora_name = style_lora_data.get('lora_name')
if style_lora_name and "19" in workflow:
_w19 = _resolve_lora_weight(style_lora_data)
workflow["19"]["inputs"]["lora_name"] = style_lora_name
workflow["19"]["inputs"]["strength_model"] = _w19
workflow["19"]["inputs"]["strength_clip"] = _w19
# Chain from previous source
workflow["19"]["inputs"]["model"] = model_source
workflow["19"]["inputs"]["clip"] = clip_source
model_source = ["19", 0]
clip_source = ["19", 1]
logger.debug("Style/Detailer LoRA: %s @ %s", style_lora_name, _w19)
# Second character LoRA (Node 20) - for multi-character generation
if character_b:
char_b_lora_data = character_b.data.get('lora', {})
char_b_lora_name = char_b_lora_data.get('lora_name')
if char_b_lora_name and "20" in workflow:
_w20 = _resolve_lora_weight(char_b_lora_data)
workflow["20"]["inputs"]["lora_name"] = char_b_lora_name
workflow["20"]["inputs"]["strength_model"] = _w20
workflow["20"]["inputs"]["strength_clip"] = _w20
workflow["20"]["inputs"]["model"] = model_source
workflow["20"]["inputs"]["clip"] = clip_source
model_source = ["20", 0]
clip_source = ["20", 1]
logger.debug("Character B LoRA: %s @ %s", char_b_lora_name, _w20)
# 3b. Multi-char: inject per-character SEGS detailers (replaces node 11/14)
if character_b:
_inject_multi_char_detailers(workflow, prompts, model_source, clip_source)
# Apply connections to all model/clip consumers (conditional on node existence)
for nid in ["3", "11", "13"] + _SEGS_DETAILER_NODES:
if nid in workflow:
workflow[nid]["inputs"]["model"] = model_source
for nid in ["6", "7", "11", "13", "14", "15"] + _SEGS_PROMPT_NODES:
if nid in workflow:
workflow[nid]["inputs"]["clip"] = clip_source
# 4. Randomize seeds (or use a fixed seed for reproducible batches like Strengths Gallery)
gen_seed = fixed_seed if fixed_seed is not None else random.randint(1, 10**15)
for nid in ["3", "11", "13"] + _SEGS_DETAILER_NODES:
if nid in workflow:
workflow[nid]["inputs"]["seed"] = gen_seed
# 5. Set image dimensions
if "5" in workflow:
if width:
workflow["5"]["inputs"]["width"] = int(width)
if height:
workflow["5"]["inputs"]["height"] = int(height)
# 6. Apply checkpoint-specific settings (steps, cfg, sampler, base prompts, VAE)
if checkpoint_data:
workflow = _apply_checkpoint_settings(workflow, checkpoint_data)
# 7. Sync sampler/scheduler from main KSampler to adetailer nodes
sampler_name = workflow["3"]["inputs"].get("sampler_name")
scheduler = workflow["3"]["inputs"].get("scheduler")
for node_id in ["11", "13"] + _SEGS_DETAILER_NODES:
if node_id in workflow:
if sampler_name:
workflow[node_id]["inputs"]["sampler_name"] = sampler_name
if scheduler:
workflow[node_id]["inputs"]["scheduler"] = scheduler
# 8. Cross-deduplicate: remove tags shared between positive and negative
pos_text, neg_text = _cross_dedup_prompts(
workflow["6"]["inputs"]["text"],
workflow["7"]["inputs"]["text"]
)
workflow["6"]["inputs"]["text"] = pos_text
workflow["7"]["inputs"]["text"] = neg_text
# 9. Final prompt debug — logged after all transformations are complete
_log_workflow_prompts("_prepare_workflow", workflow)
return workflow