feat: implement Actions Gallery with character integration and triple LoRA chaining
- Added Actions gallery with CRUD and JSON sync - Implemented Triple LoRA workflow (Character -> Outfit -> Action) - Added character-integrated previews for Actions with style matching - Implemented granular prompt selection and default persistence for Actions - Added detailed development guide for extending gallery features
This commit is contained in:
Aodhan Collins
653
app.py
653
app.py
@@ -6,7 +6,7 @@ import requests
|
||||
import random
|
||||
from flask import Flask, render_template, request, redirect, url_for, flash, session
|
||||
from werkzeug.utils import secure_filename
|
||||
from models import db, Character, Settings, Outfit
|
||||
from models import db, Character, Settings, Outfit, Action
|
||||
|
||||
app = Flask(__name__)
|
||||
app.config['SQLALCHEMY_DATABASE_URI'] = 'sqlite:///database.db'
|
||||
@@ -15,6 +15,7 @@ app.config['UPLOAD_FOLDER'] = 'static/uploads'
|
||||
app.config['SECRET_KEY'] = 'dev-key-123'
|
||||
app.config['CHARACTERS_DIR'] = 'data/characters'
|
||||
app.config['CLOTHING_DIR'] = 'data/clothing'
|
||||
app.config['ACTIONS_DIR'] = 'data/actions'
|
||||
app.config['COMFYUI_URL'] = 'http://127.0.0.1:8188'
|
||||
app.config['ILLUSTRIOUS_MODELS_DIR'] = '/mnt/alexander/AITools/Image Models/Stable-diffusion/Illustrious/'
|
||||
app.config['NOOB_MODELS_DIR'] = '/mnt/alexander/AITools/Image Models/Stable-diffusion/Noob/'
|
||||
@@ -43,6 +44,16 @@ def get_available_clothing_loras():
|
||||
loras.append(f"Illustrious/Clothing/{f}")
|
||||
return sorted(loras)
|
||||
|
||||
def get_available_action_loras():
|
||||
"""Get LoRAs from the Poses directory for action LoRAs."""
|
||||
poses_lora_dir = '/mnt/alexander/AITools/Image Models/lora/Illustrious/Poses/'
|
||||
loras = []
|
||||
if os.path.exists(poses_lora_dir):
|
||||
for f in os.listdir(poses_lora_dir):
|
||||
if f.endswith('.safetensors'):
|
||||
loras.append(f"Illustrious/Poses/{f}")
|
||||
return sorted(loras)
|
||||
|
||||
def get_available_checkpoints():
|
||||
checkpoints = []
|
||||
|
||||
@@ -87,6 +98,7 @@ def build_prompt(data, selected_fields=None, default_fields=None, active_outfit=
|
||||
wardrobe = wardrobe_data
|
||||
|
||||
defaults = data.get('defaults', {})
|
||||
action_data = data.get('action', {})
|
||||
|
||||
# Pre-calculate Hand/Glove priority
|
||||
# Priority: wardrobe gloves > wardrobe hands (outfit) > identity hands (character)
|
||||
@@ -137,15 +149,21 @@ def build_prompt(data, selected_fields=None, default_fields=None, active_outfit=
|
||||
if lora.get('lora_triggers') and is_selected('lora', 'lora_triggers'):
|
||||
parts.append(lora.get('lora_triggers'))
|
||||
|
||||
# 2. Face Prompt: Tag, Eyes, Expression, Headwear
|
||||
# 2. Face Prompt: Tag, Eyes, Expression, Headwear, Action details
|
||||
face_parts = []
|
||||
if char_tag and is_selected('special', 'name'): face_parts.append(char_tag)
|
||||
if identity.get('eyes') and is_selected('identity', 'eyes'): face_parts.append(identity.get('eyes'))
|
||||
if defaults.get('expression') and is_selected('defaults', 'expression'): face_parts.append(defaults.get('expression'))
|
||||
if wardrobe.get('headwear') and is_selected('wardrobe', 'headwear'): face_parts.append(wardrobe.get('headwear'))
|
||||
|
||||
# 3. Hand Prompt: Hand value (Gloves or Hands)
|
||||
# Add specific Action expression details if available
|
||||
if action_data.get('head') and is_selected('action', 'head'): face_parts.append(action_data.get('head'))
|
||||
if action_data.get('eyes') and is_selected('action', 'eyes'): face_parts.append(action_data.get('eyes'))
|
||||
|
||||
# 3. Hand Prompt: Hand value (Gloves or Hands), Action details
|
||||
hand_parts = [hand_val] if hand_val else []
|
||||
if action_data.get('arms') and is_selected('action', 'arms'): hand_parts.append(action_data.get('arms'))
|
||||
if action_data.get('hands') and is_selected('action', 'hands'): hand_parts.append(action_data.get('hands'))
|
||||
|
||||
return {
|
||||
"main": ", ".join(parts),
|
||||
@@ -290,6 +308,63 @@ def sync_outfits():
|
||||
|
||||
db.session.commit()
|
||||
|
||||
def sync_actions():
|
||||
if not os.path.exists(app.config['ACTIONS_DIR']):
|
||||
return
|
||||
|
||||
current_ids = []
|
||||
|
||||
for filename in os.listdir(app.config['ACTIONS_DIR']):
|
||||
if filename.endswith('.json'):
|
||||
file_path = os.path.join(app.config['ACTIONS_DIR'], filename)
|
||||
try:
|
||||
with open(file_path, 'r') as f:
|
||||
data = json.load(f)
|
||||
action_id = data.get('action_id') or filename.replace('.json', '')
|
||||
|
||||
current_ids.append(action_id)
|
||||
|
||||
# Generate URL-safe slug
|
||||
slug = re.sub(r'[^a-zA-Z0-9_]', '', action_id)
|
||||
|
||||
# Check if action already exists
|
||||
action = Action.query.filter_by(action_id=action_id).first()
|
||||
name = data.get('action_name', action_id.replace('_', ' ').title())
|
||||
|
||||
if action:
|
||||
action.data = data
|
||||
action.name = name
|
||||
action.slug = slug
|
||||
action.filename = filename
|
||||
|
||||
# Check if cover image still exists
|
||||
if action.image_path:
|
||||
full_img_path = os.path.join(app.config['UPLOAD_FOLDER'], action.image_path)
|
||||
if not os.path.exists(full_img_path):
|
||||
print(f"Image missing for {action.name}, clearing path.")
|
||||
action.image_path = None
|
||||
|
||||
flag_modified(action, "data")
|
||||
else:
|
||||
new_action = Action(
|
||||
action_id=action_id,
|
||||
slug=slug,
|
||||
filename=filename,
|
||||
name=name,
|
||||
data=data
|
||||
)
|
||||
db.session.add(new_action)
|
||||
except Exception as e:
|
||||
print(f"Error importing action {filename}: {e}")
|
||||
|
||||
# Remove actions that are no longer in the folder
|
||||
all_actions = Action.query.all()
|
||||
for action in all_actions:
|
||||
if action.action_id not in current_ids:
|
||||
db.session.delete(action)
|
||||
|
||||
db.session.commit()
|
||||
|
||||
def call_llm(prompt, system_prompt="You are a creative assistant."):
|
||||
settings = Settings.query.first()
|
||||
if not settings or not settings.openrouter_api_key:
|
||||
@@ -907,7 +982,7 @@ def replace_cover_from_preview(slug):
|
||||
|
||||
return redirect(url_for('detail', slug=slug))
|
||||
|
||||
def _prepare_workflow(workflow, character, prompts, checkpoint=None, custom_negative=None, outfit=None):
|
||||
def _prepare_workflow(workflow, character, prompts, checkpoint=None, custom_negative=None, outfit=None, action=None):
|
||||
# 1. Update prompts using replacement to preserve embeddings
|
||||
workflow["6"]["inputs"]["text"] = workflow["6"]["inputs"]["text"].replace("{{POSITIVE_PROMPT}}", prompts["main"])
|
||||
|
||||
@@ -932,7 +1007,7 @@ def _prepare_workflow(workflow, character, prompts, checkpoint=None, custom_nega
|
||||
if checkpoint:
|
||||
workflow["4"]["inputs"]["ckpt_name"] = checkpoint
|
||||
|
||||
# 3. Handle LoRAs - Node 16 for character, Node 17 for outfit
|
||||
# 3. Handle LoRAs - Node 16 for character, Node 17 for outfit, Node 18 for action
|
||||
# Start with direct checkpoint connections
|
||||
model_source = ["4", 0]
|
||||
clip_source = ["4", 1]
|
||||
@@ -960,16 +1035,27 @@ def _prepare_workflow(workflow, character, prompts, checkpoint=None, custom_nega
|
||||
workflow["17"]["inputs"]["strength_model"] = outfit_lora_data.get('lora_weight', 0.8)
|
||||
workflow["17"]["inputs"]["strength_clip"] = outfit_lora_data.get('lora_weight', 0.8)
|
||||
# Chain from character LoRA (node 16) or checkpoint (node 4)
|
||||
if char_lora_name and "16" in workflow:
|
||||
workflow["17"]["inputs"]["model"] = ["16", 0]
|
||||
workflow["17"]["inputs"]["clip"] = ["16", 1]
|
||||
else:
|
||||
workflow["17"]["inputs"]["model"] = ["4", 0]
|
||||
workflow["17"]["inputs"]["clip"] = ["4", 1]
|
||||
workflow["17"]["inputs"]["model"] = model_source
|
||||
workflow["17"]["inputs"]["clip"] = clip_source
|
||||
model_source = ["17", 0]
|
||||
clip_source = ["17", 1]
|
||||
print(f"Outfit LoRA: {outfit_lora_name} @ {outfit_lora_data.get('lora_weight', 0.8)}")
|
||||
|
||||
# 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:
|
||||
workflow["18"]["inputs"]["lora_name"] = action_lora_name
|
||||
workflow["18"]["inputs"]["strength_model"] = action_lora_data.get('lora_weight', 1.0)
|
||||
workflow["18"]["inputs"]["strength_clip"] = action_lora_data.get('lora_weight', 1.0)
|
||||
# Chain from previous source
|
||||
workflow["18"]["inputs"]["model"] = model_source
|
||||
workflow["18"]["inputs"]["clip"] = clip_source
|
||||
model_source = ["18", 0]
|
||||
clip_source = ["18", 1]
|
||||
print(f"Action LoRA: {action_lora_name} @ {action_lora_data.get('lora_weight', 1.0)}")
|
||||
|
||||
# Apply connections to all model/clip consumers
|
||||
workflow["3"]["inputs"]["model"] = model_source
|
||||
workflow["11"]["inputs"]["model"] = model_source
|
||||
@@ -1619,6 +1705,538 @@ def clone_outfit(slug):
|
||||
flash(f'Outfit cloned as "{new_id}"!')
|
||||
return redirect(url_for('outfit_detail', slug=new_slug))
|
||||
|
||||
def sync_actions():
|
||||
if not os.path.exists(app.config['ACTIONS_DIR']):
|
||||
return
|
||||
|
||||
current_ids = []
|
||||
|
||||
for filename in os.listdir(app.config['ACTIONS_DIR']):
|
||||
if filename.endswith('.json'):
|
||||
file_path = os.path.join(app.config['ACTIONS_DIR'], filename)
|
||||
try:
|
||||
with open(file_path, 'r') as f:
|
||||
data = json.load(f)
|
||||
action_id = data.get('action_id') or filename.replace('.json', '')
|
||||
|
||||
current_ids.append(action_id)
|
||||
|
||||
# Generate URL-safe slug
|
||||
slug = re.sub(r'[^a-zA-Z0-9_]', '', action_id)
|
||||
|
||||
# Check if action already exists
|
||||
action = Action.query.filter_by(action_id=action_id).first()
|
||||
name = data.get('action_name', action_id.replace('_', ' ').title())
|
||||
|
||||
if action:
|
||||
action.data = data
|
||||
action.name = name
|
||||
action.slug = slug
|
||||
action.filename = filename
|
||||
|
||||
# Check if cover image still exists
|
||||
if action.image_path:
|
||||
full_img_path = os.path.join(app.config['UPLOAD_FOLDER'], action.image_path)
|
||||
if not os.path.exists(full_img_path):
|
||||
print(f"Image missing for {action.name}, clearing path.")
|
||||
action.image_path = None
|
||||
|
||||
flag_modified(action, "data")
|
||||
else:
|
||||
new_action = Action(
|
||||
action_id=action_id,
|
||||
slug=slug,
|
||||
filename=filename,
|
||||
name=name,
|
||||
data=data
|
||||
)
|
||||
db.session.add(new_action)
|
||||
except Exception as e:
|
||||
print(f"Error importing action {filename}: {e}")
|
||||
|
||||
# Remove actions that are no longer in the folder
|
||||
all_actions = Action.query.all()
|
||||
for action in all_actions:
|
||||
if action.action_id not in current_ids:
|
||||
db.session.delete(action)
|
||||
|
||||
db.session.commit()
|
||||
|
||||
# ============ ACTION ROUTES ============
|
||||
|
||||
@app.route('/actions')
|
||||
def actions_index():
|
||||
actions = Action.query.order_by(Action.name).all()
|
||||
return render_template('actions/index.html', actions=actions)
|
||||
|
||||
@app.route('/actions/rescan', methods=['POST'])
|
||||
def rescan_actions():
|
||||
sync_actions()
|
||||
flash('Database synced with action files.')
|
||||
return redirect(url_for('actions_index'))
|
||||
|
||||
@app.route('/action/<path:slug>')
|
||||
def action_detail(slug):
|
||||
action = Action.query.filter_by(slug=slug).first_or_404()
|
||||
characters = Character.query.order_by(Character.name).all()
|
||||
|
||||
# Load state from session
|
||||
preferences = session.get(f'prefs_action_{slug}')
|
||||
preview_image = session.get(f'preview_action_{slug}')
|
||||
selected_character = session.get(f'char_action_{slug}')
|
||||
|
||||
return render_template('actions/detail.html', action=action, characters=characters,
|
||||
preferences=preferences, preview_image=preview_image,
|
||||
selected_character=selected_character)
|
||||
|
||||
@app.route('/action/<path:slug>/edit', methods=['GET', 'POST'])
|
||||
def edit_action(slug):
|
||||
action = Action.query.filter_by(slug=slug).first_or_404()
|
||||
loras = get_available_action_loras()
|
||||
|
||||
if request.method == 'POST':
|
||||
try:
|
||||
# 1. Update basic fields
|
||||
action.name = request.form.get('action_name')
|
||||
|
||||
# 2. Rebuild the data dictionary
|
||||
new_data = action.data.copy()
|
||||
new_data['action_name'] = action.name
|
||||
|
||||
# Update action_id if provided
|
||||
new_action_id = request.form.get('action_id', action.action_id)
|
||||
new_data['action_id'] = new_action_id
|
||||
|
||||
# Update action section
|
||||
if 'action' in new_data:
|
||||
for key in new_data['action'].keys():
|
||||
form_key = f"action_{key}"
|
||||
if form_key in request.form:
|
||||
new_data['action'][key] = request.form.get(form_key)
|
||||
|
||||
# Update lora section
|
||||
if 'lora' in new_data:
|
||||
for key in new_data['lora'].keys():
|
||||
form_key = f"lora_{key}"
|
||||
if form_key in request.form:
|
||||
val = request.form.get(form_key)
|
||||
if key == 'lora_weight':
|
||||
try: val = float(val)
|
||||
except: val = 1.0
|
||||
new_data['lora'][key] = val
|
||||
|
||||
# Update Tags (comma separated string to list)
|
||||
tags_raw = request.form.get('tags', '')
|
||||
new_data['tags'] = [t.strip() for f in tags_raw.split(',') for t in [f.strip()] if t]
|
||||
|
||||
action.data = new_data
|
||||
flag_modified(action, "data")
|
||||
|
||||
# 3. Write back to JSON file
|
||||
action_file = action.filename or f"{re.sub(r'[^a-zA-Z0-9_]', '', action.action_id)}.json"
|
||||
file_path = os.path.join(app.config['ACTIONS_DIR'], action_file)
|
||||
|
||||
with open(file_path, 'w') as f:
|
||||
json.dump(new_data, f, indent=2)
|
||||
|
||||
db.session.commit()
|
||||
flash('Action profile updated successfully!')
|
||||
return redirect(url_for('action_detail', slug=slug))
|
||||
|
||||
except Exception as e:
|
||||
print(f"Edit error: {e}")
|
||||
flash(f"Error saving changes: {str(e)}")
|
||||
|
||||
return render_template('actions/edit.html', action=action, loras=loras)
|
||||
|
||||
@app.route('/action/<path:slug>/upload', methods=['POST'])
|
||||
def upload_action_image(slug):
|
||||
action = Action.query.filter_by(slug=slug).first_or_404()
|
||||
|
||||
if 'image' not in request.files:
|
||||
flash('No file part')
|
||||
return redirect(request.url)
|
||||
|
||||
file = request.files['image']
|
||||
if file.filename == '':
|
||||
flash('No selected file')
|
||||
return redirect(request.url)
|
||||
|
||||
if file and allowed_file(file.filename):
|
||||
# Create action subfolder
|
||||
action_folder = os.path.join(app.config['UPLOAD_FOLDER'], f"actions/{slug}")
|
||||
os.makedirs(action_folder, exist_ok=True)
|
||||
|
||||
filename = secure_filename(file.filename)
|
||||
file_path = os.path.join(action_folder, filename)
|
||||
file.save(file_path)
|
||||
|
||||
# Store relative path in DB
|
||||
action.image_path = f"actions/{slug}/{filename}"
|
||||
db.session.commit()
|
||||
flash('Image uploaded successfully!')
|
||||
|
||||
return redirect(url_for('action_detail', slug=slug))
|
||||
|
||||
@app.route('/action/<path:slug>/generate', methods=['POST'])
|
||||
def generate_action_image(slug):
|
||||
action_obj = Action.query.filter_by(slug=slug).first_or_404()
|
||||
|
||||
try:
|
||||
# Get action type
|
||||
action_type = request.form.get('action', 'preview')
|
||||
client_id = request.form.get('client_id')
|
||||
|
||||
# Get selected fields
|
||||
selected_fields = request.form.getlist('include_field')
|
||||
|
||||
# Get selected character (if any)
|
||||
character_slug = request.form.get('character_slug', '')
|
||||
character = None
|
||||
|
||||
# Handle random character selection
|
||||
if character_slug == '__random__':
|
||||
all_characters = Character.query.all()
|
||||
if all_characters:
|
||||
character = random.choice(all_characters)
|
||||
character_slug = character.slug
|
||||
elif character_slug:
|
||||
character = Character.query.filter_by(slug=character_slug).first()
|
||||
|
||||
# Save preferences
|
||||
session[f'char_action_{slug}'] = character_slug
|
||||
session[f'prefs_action_{slug}'] = selected_fields
|
||||
|
||||
# Build combined data for prompt building
|
||||
if character:
|
||||
# Combine character identity/wardrobe with action details
|
||||
# Action details replace character's 'defaults' (pose, etc.)
|
||||
combined_data = character.data.copy()
|
||||
|
||||
# Update 'defaults' with action details
|
||||
action_data = action_obj.data.get('action', {})
|
||||
combined_data['action'] = action_data # Ensure action section is present for routing
|
||||
|
||||
# Aggregate pose-related fields into 'pose'
|
||||
pose_fields = ['full_body', 'arms', 'hands', 'torso', 'pelvis', 'legs', 'feet']
|
||||
pose_parts = [action_data.get(k) for k in pose_fields if action_data.get(k)]
|
||||
|
||||
# Aggregate expression-related fields into 'expression'
|
||||
expression_parts = [action_data.get(k) for k in ['head', 'eyes'] if action_data.get(k)]
|
||||
|
||||
combined_data['defaults'] = {
|
||||
'pose': ", ".join(pose_parts),
|
||||
'expression': ", ".join(expression_parts),
|
||||
'scene': action_data.get('additional', '')
|
||||
}
|
||||
|
||||
# Merge lora triggers if present
|
||||
action_lora = action_obj.data.get('lora', {})
|
||||
if action_lora.get('lora_triggers'):
|
||||
if 'lora' not in combined_data: combined_data['lora'] = {}
|
||||
combined_data['lora']['lora_triggers'] = f"{combined_data['lora'].get('lora_triggers', '')}, {action_lora['lora_triggers']}"
|
||||
|
||||
# Merge tags
|
||||
combined_data['tags'] = list(set(combined_data.get('tags', []) + action_obj.data.get('tags', [])))
|
||||
|
||||
# Use action's defaults if no manual selection
|
||||
if not selected_fields:
|
||||
selected_fields = list(action_obj.default_fields) if action_obj.default_fields else []
|
||||
|
||||
# Auto-include essential character fields if a character is selected
|
||||
if selected_fields:
|
||||
# Add character identity fields to selection if not already present
|
||||
for key in ['base_specs', 'hair', 'eyes', 'hands', 'arms', 'torso', 'pelvis', 'legs', 'feet', 'extra']:
|
||||
if character.data.get('identity', {}).get(key):
|
||||
field_key = f'identity::{key}'
|
||||
if field_key not in selected_fields:
|
||||
selected_fields.append(field_key)
|
||||
|
||||
# Always include character name
|
||||
if 'special::name' not in selected_fields:
|
||||
selected_fields.append('special::name')
|
||||
|
||||
# Add active wardrobe fields
|
||||
wardrobe = character.get_active_wardrobe()
|
||||
for key in ['full_body', 'headwear', 'top', 'bottom', 'legwear', 'footwear', 'hands', 'gloves', 'accessories']:
|
||||
if wardrobe.get(key):
|
||||
field_key = f'wardrobe::{key}'
|
||||
if field_key not in selected_fields:
|
||||
selected_fields.append(field_key)
|
||||
else:
|
||||
# Fallback to sensible defaults if still empty (no checkboxes and no action defaults)
|
||||
selected_fields = ['special::name', 'defaults::pose', 'defaults::expression']
|
||||
# Add identity fields
|
||||
for key in ['base_specs', 'hair', 'eyes']:
|
||||
if character.data.get('identity', {}).get(key):
|
||||
selected_fields.append(f'identity::{key}')
|
||||
# Add wardrobe fields
|
||||
wardrobe = character.get_active_wardrobe()
|
||||
for key in ['full_body', 'top', 'bottom']:
|
||||
if wardrobe.get(key):
|
||||
selected_fields.append(f'wardrobe::{key}')
|
||||
|
||||
default_fields = action_obj.default_fields
|
||||
active_outfit = character.active_outfit
|
||||
else:
|
||||
# Action only - no character (rarely makes sense for actions but let's handle it)
|
||||
action_data = action_obj.data.get('action', {})
|
||||
|
||||
# Aggregate pose-related fields into 'pose'
|
||||
pose_fields = ['full_body', 'arms', 'hands', 'torso', 'pelvis', 'legs', 'feet']
|
||||
pose_parts = [action_data.get(k) for k in pose_fields if action_data.get(k)]
|
||||
|
||||
# Aggregate expression-related fields into 'expression'
|
||||
expression_parts = [action_data.get(k) for k in ['head', 'eyes'] if action_data.get(k)]
|
||||
|
||||
combined_data = {
|
||||
'character_id': action_obj.action_id,
|
||||
'defaults': {
|
||||
'pose': ", ".join(pose_parts),
|
||||
'expression': ", ".join(expression_parts),
|
||||
'scene': action_data.get('additional', '')
|
||||
},
|
||||
'lora': action_obj.data.get('lora', {}),
|
||||
'tags': action_obj.data.get('tags', [])
|
||||
}
|
||||
if not selected_fields:
|
||||
selected_fields = ['defaults::pose', 'defaults::expression', 'defaults::scene', 'lora::lora_triggers', 'special::tags']
|
||||
default_fields = action_obj.default_fields
|
||||
active_outfit = 'default'
|
||||
|
||||
# Queue generation
|
||||
with open('comfy_workflow.json', 'r') as f:
|
||||
workflow = json.load(f)
|
||||
|
||||
# Build prompts for combined data
|
||||
prompts = build_prompt(combined_data, selected_fields, default_fields, active_outfit=active_outfit)
|
||||
|
||||
# Add colored simple background to the main prompt for action previews
|
||||
if character:
|
||||
primary_color = character.data.get('styles', {}).get('primary_color', '')
|
||||
if primary_color:
|
||||
prompts["main"] = f"{prompts['main']}, {primary_color} simple background"
|
||||
else:
|
||||
prompts["main"] = f"{prompts['main']}, simple background"
|
||||
else:
|
||||
prompts["main"] = f"{prompts['main']}, simple background"
|
||||
|
||||
# Prepare workflow
|
||||
workflow = _prepare_workflow(workflow, character, prompts, action=action_obj)
|
||||
|
||||
prompt_response = queue_prompt(workflow, client_id=client_id)
|
||||
|
||||
if 'prompt_id' not in prompt_response:
|
||||
raise Exception(f"ComfyUI failed: {prompt_response.get('error', 'Unknown error')}")
|
||||
|
||||
prompt_id = prompt_response['prompt_id']
|
||||
|
||||
if request.headers.get('X-Requested-With') == 'XMLHttpRequest':
|
||||
return {'status': 'queued', 'prompt_id': prompt_id}
|
||||
|
||||
return redirect(url_for('action_detail', slug=slug))
|
||||
|
||||
except Exception as e:
|
||||
print(f"Generation error: {e}")
|
||||
if request.headers.get('X-Requested-With') == 'XMLHttpRequest':
|
||||
return {'error': str(e)}, 500
|
||||
flash(f"Error during generation: {str(e)}")
|
||||
return redirect(url_for('action_detail', slug=slug))
|
||||
|
||||
@app.route('/action/<path:slug>/finalize_generation/<prompt_id>', methods=['POST'])
|
||||
def finalize_action_generation(slug, prompt_id):
|
||||
action_obj = Action.query.filter_by(slug=slug).first_or_404()
|
||||
|
||||
try:
|
||||
history = get_history(prompt_id)
|
||||
if prompt_id not in history:
|
||||
return {'error': 'History not found'}, 404
|
||||
|
||||
outputs = history[prompt_id]['outputs']
|
||||
for node_id in outputs:
|
||||
if 'images' in outputs[node_id]:
|
||||
image_info = outputs[node_id]['images'][0]
|
||||
image_data = get_image(image_info['filename'], image_info['subfolder'], image_info['type'])
|
||||
|
||||
# Create action subfolder
|
||||
action_folder = os.path.join(app.config['UPLOAD_FOLDER'], f"actions/{slug}")
|
||||
os.makedirs(action_folder, exist_ok=True)
|
||||
|
||||
filename = f"gen_{int(time.time())}.png"
|
||||
file_path = os.path.join(action_folder, filename)
|
||||
with open(file_path, 'wb') as f:
|
||||
f.write(image_data)
|
||||
|
||||
# Always save as preview
|
||||
relative_path = f"actions/{slug}/{filename}"
|
||||
session[f'preview_action_{slug}'] = relative_path
|
||||
|
||||
return {'success': True, 'image_url': url_for('static', filename=f'uploads/{relative_path}')}
|
||||
|
||||
return {'error': 'No image found in output'}, 404
|
||||
except Exception as e:
|
||||
print(f"Finalize error: {e}")
|
||||
return {'error': str(e)}, 500
|
||||
|
||||
@app.route('/action/<path:slug>/replace_cover_from_preview', methods=['POST'])
|
||||
def replace_action_cover_from_preview(slug):
|
||||
action = Action.query.filter_by(slug=slug).first_or_404()
|
||||
preview_path = session.get(f'preview_action_{slug}')
|
||||
|
||||
if preview_path:
|
||||
action.image_path = preview_path
|
||||
db.session.commit()
|
||||
flash('Cover image updated from preview!')
|
||||
else:
|
||||
flash('No preview image available', 'error')
|
||||
|
||||
return redirect(url_for('action_detail', slug=slug))
|
||||
|
||||
@app.route('/action/<path:slug>/save_defaults', methods=['POST'])
|
||||
def save_action_defaults(slug):
|
||||
action = Action.query.filter_by(slug=slug).first_or_404()
|
||||
selected_fields = request.form.getlist('include_field')
|
||||
action.default_fields = selected_fields
|
||||
db.session.commit()
|
||||
flash('Default prompt selection saved for this action!')
|
||||
return redirect(url_for('action_detail', slug=slug))
|
||||
|
||||
@app.route('/action/create', methods=['GET', 'POST'])
|
||||
def create_action():
|
||||
if request.method == 'POST':
|
||||
name = request.form.get('name')
|
||||
slug = request.form.get('filename', '').strip()
|
||||
prompt = request.form.get('prompt', '')
|
||||
use_llm = request.form.get('use_llm') == 'on'
|
||||
|
||||
if not slug:
|
||||
slug = re.sub(r'[^a-zA-Z0-9]+', '_', name.lower()).strip('_')
|
||||
|
||||
safe_slug = re.sub(r'[^a-zA-Z0-9_]', '', slug)
|
||||
if not safe_slug:
|
||||
safe_slug = 'action'
|
||||
|
||||
base_slug = safe_slug
|
||||
counter = 1
|
||||
while os.path.exists(os.path.join(app.config['ACTIONS_DIR'], f"{safe_slug}.json")):
|
||||
safe_slug = f"{base_slug}_{counter}"
|
||||
counter += 1
|
||||
|
||||
if use_llm:
|
||||
if not prompt:
|
||||
flash("Description is required when AI generation is enabled.")
|
||||
return redirect(request.url)
|
||||
|
||||
system_prompt = """You are a JSON generator. output ONLY valid JSON matching this exact structure. Do not wrap in markdown blocks.
|
||||
Structure:
|
||||
{
|
||||
"action_id": "WILL_BE_REPLACED",
|
||||
"action_name": "WILL_BE_REPLACED",
|
||||
"action": {
|
||||
"full_body": "string (pose description)",
|
||||
"head": "string (expression/head position)",
|
||||
"eyes": "string",
|
||||
"arms": "string",
|
||||
"hands": "string",
|
||||
"torso": "string",
|
||||
"pelvis": "string",
|
||||
"legs": "string",
|
||||
"feet": "string",
|
||||
"additional": "string"
|
||||
},
|
||||
"lora": {
|
||||
"lora_name": "",
|
||||
"lora_weight": 1.0,
|
||||
"lora_triggers": ""
|
||||
},
|
||||
"tags": ["string", "string"]
|
||||
}"""
|
||||
|
||||
try:
|
||||
llm_response = call_llm(f"Create an action profile for '{name}' based on this description: {prompt}", system_prompt)
|
||||
clean_json = llm_response.replace('```json', '').replace('```', '').strip()
|
||||
action_data = json.loads(clean_json)
|
||||
action_data['action_id'] = safe_slug
|
||||
action_data['action_name'] = name
|
||||
except Exception as e:
|
||||
print(f"LLM error: {e}")
|
||||
flash(f"Failed to generate action profile: {e}")
|
||||
return redirect(request.url)
|
||||
else:
|
||||
action_data = {
|
||||
"action_id": safe_slug,
|
||||
"action_name": name,
|
||||
"action": {
|
||||
"full_body": "", "head": "", "eyes": "", "arms": "", "hands": "",
|
||||
"torso": "", "pelvis": "", "legs": "", "feet": "", "additional": ""
|
||||
},
|
||||
"lora": {"lora_name": "", "lora_weight": 1.0, "lora_triggers": ""},
|
||||
"tags": []
|
||||
}
|
||||
|
||||
try:
|
||||
file_path = os.path.join(app.config['ACTIONS_DIR'], f"{safe_slug}.json")
|
||||
with open(file_path, 'w') as f:
|
||||
json.dump(action_data, f, indent=2)
|
||||
|
||||
new_action = Action(
|
||||
action_id=safe_slug, slug=safe_slug, filename=f"{safe_slug}.json",
|
||||
name=name, data=action_data
|
||||
)
|
||||
db.session.add(new_action)
|
||||
db.session.commit()
|
||||
|
||||
flash('Action created successfully!')
|
||||
return redirect(url_for('action_detail', slug=safe_slug))
|
||||
except Exception as e:
|
||||
print(f"Save error: {e}")
|
||||
flash(f"Failed to create action: {e}")
|
||||
return redirect(request.url)
|
||||
|
||||
return render_template('actions/create.html')
|
||||
|
||||
@app.route('/action/<path:slug>/clone', methods=['POST'])
|
||||
def clone_action(slug):
|
||||
action = Action.query.filter_by(slug=slug).first_or_404()
|
||||
|
||||
# Find the next available number for the clone
|
||||
base_id = action.action_id
|
||||
import re
|
||||
match = re.match(r'^(.+?)_(\d+)$', base_id)
|
||||
if match:
|
||||
base_name = match.group(1)
|
||||
current_num = int(match.group(2))
|
||||
else:
|
||||
base_name = base_id
|
||||
current_num = 1
|
||||
|
||||
next_num = current_num + 1
|
||||
while True:
|
||||
new_id = f"{base_name}_{next_num:02d}"
|
||||
new_filename = f"{new_id}.json"
|
||||
new_path = os.path.join(app.config['ACTIONS_DIR'], new_filename)
|
||||
if not os.path.exists(new_path):
|
||||
break
|
||||
next_num += 1
|
||||
|
||||
new_data = action.data.copy()
|
||||
new_data['action_id'] = new_id
|
||||
new_data['action_name'] = f"{action.name} (Copy)"
|
||||
|
||||
with open(new_path, 'w') as f:
|
||||
json.dump(new_data, f, indent=2)
|
||||
|
||||
new_slug = re.sub(r'[^a-zA-Z0-9_]', '', new_id)
|
||||
new_action = Action(
|
||||
action_id=new_id, slug=new_slug, filename=new_filename,
|
||||
name=new_data['action_name'], data=new_data
|
||||
)
|
||||
db.session.add(new_action)
|
||||
db.session.commit()
|
||||
|
||||
flash(f'Action cloned as "{new_id}"!')
|
||||
return redirect(url_for('action_detail', slug=new_slug))
|
||||
|
||||
if __name__ == '__main__':
|
||||
with app.app_context():
|
||||
os.makedirs(app.config['UPLOAD_FOLDER'], exist_ok=True)
|
||||
@@ -1636,6 +2254,19 @@ if __name__ == '__main__':
|
||||
else:
|
||||
print(f"Migration note: {e}")
|
||||
|
||||
# Migration: Add default_fields column to action table if it doesn't exist
|
||||
try:
|
||||
from sqlalchemy import text
|
||||
db.session.execute(text('ALTER TABLE action ADD COLUMN default_fields JSON'))
|
||||
db.session.commit()
|
||||
print("Added default_fields column to action table")
|
||||
except Exception as e:
|
||||
if 'duplicate column name' in str(e).lower() or 'already exists' in str(e).lower():
|
||||
print("default_fields column already exists in action table")
|
||||
else:
|
||||
print(f"Migration action note: {e}")
|
||||
|
||||
sync_characters()
|
||||
sync_outfits()
|
||||
sync_actions()
|
||||
app.run(debug=True, port=5000)
|
||||
|
||||
Reference in New Issue
Block a user