Add REST API for preset-based generation and fallback cover images
REST API (routes/api.py): Three endpoints behind API key auth for programmatic image generation via presets — list presets, queue generation with optional overrides, and poll job status. Shared generation logic extracted from routes/presets.py into services/generation.py so both web UI and API use the same code path. Fallback covers: library index pages now show a random generated image at reduced opacity when no cover is assigned, instead of "No Image". Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
This commit is contained in:
Aodhan Collins
@@ -17,6 +17,7 @@ def register_routes(app):
|
||||
from routes import gallery
|
||||
from routes import strengths
|
||||
from routes import transfer
|
||||
from routes import api
|
||||
|
||||
queue_api.register_routes(app)
|
||||
settings.register_routes(app)
|
||||
@@ -35,3 +36,4 @@ def register_routes(app):
|
||||
gallery.register_routes(app)
|
||||
strengths.register_routes(app)
|
||||
transfer.register_routes(app)
|
||||
api.register_routes(app)
|
||||
|
||||
105
routes/api.py
Normal file
105
routes/api.py
Normal file
@@ -0,0 +1,105 @@
|
||||
import logging
|
||||
from functools import wraps
|
||||
from flask import request, jsonify
|
||||
from models import Preset, Settings
|
||||
from services.generation import generate_from_preset
|
||||
from services.job_queue import _job_queue_lock, _job_history
|
||||
|
||||
logger = logging.getLogger('gaze')
|
||||
|
||||
|
||||
def register_routes(app):
|
||||
|
||||
def require_api_key(f):
|
||||
@wraps(f)
|
||||
def decorated(*args, **kwargs):
|
||||
api_key = request.headers.get('X-API-Key') or request.args.get('api_key')
|
||||
if not api_key:
|
||||
return jsonify({'error': 'Missing API key. Provide X-API-Key header or api_key query parameter.'}), 401
|
||||
settings = Settings.query.first()
|
||||
if not settings or not settings.api_key or api_key != settings.api_key:
|
||||
return jsonify({'error': 'Invalid API key.'}), 403
|
||||
return f(*args, **kwargs)
|
||||
return decorated
|
||||
|
||||
@app.route('/api/v1/presets')
|
||||
@require_api_key
|
||||
def api_v1_presets():
|
||||
presets = Preset.query.order_by(Preset.name).all()
|
||||
return jsonify({
|
||||
'presets': [
|
||||
{
|
||||
'preset_id': p.preset_id,
|
||||
'slug': p.slug,
|
||||
'name': p.name,
|
||||
'has_cover': bool(p.image_path),
|
||||
}
|
||||
for p in presets
|
||||
]
|
||||
})
|
||||
|
||||
@app.route('/api/v1/generate/<path:preset_slug>', methods=['POST'])
|
||||
@require_api_key
|
||||
def api_v1_generate(preset_slug):
|
||||
preset = Preset.query.filter_by(slug=preset_slug).first()
|
||||
if not preset:
|
||||
return jsonify({'error': f'Preset not found: {preset_slug}'}), 404
|
||||
|
||||
body = request.get_json(silent=True) or {}
|
||||
|
||||
count = body.get('count', 1)
|
||||
try:
|
||||
count = int(count)
|
||||
except (TypeError, ValueError):
|
||||
return jsonify({'error': 'count must be an integer'}), 400
|
||||
if count < 1 or count > 20:
|
||||
return jsonify({'error': 'count must be between 1 and 20'}), 400
|
||||
|
||||
seed = body.get('seed')
|
||||
if seed is not None:
|
||||
try:
|
||||
seed = int(seed)
|
||||
except (TypeError, ValueError):
|
||||
return jsonify({'error': 'seed must be an integer'}), 400
|
||||
|
||||
width = body.get('width')
|
||||
height = body.get('height')
|
||||
if (width is None) != (height is None):
|
||||
return jsonify({'error': 'width and height must both be provided or both omitted'}), 400
|
||||
|
||||
overrides = {
|
||||
'action': 'preview',
|
||||
'checkpoint': body.get('checkpoint', ''),
|
||||
'extra_positive': body.get('extra_positive', ''),
|
||||
'extra_negative': body.get('extra_negative', ''),
|
||||
'seed': seed,
|
||||
'width': int(width) if width else None,
|
||||
'height': int(height) if height else None,
|
||||
}
|
||||
|
||||
try:
|
||||
jobs = []
|
||||
for _ in range(count):
|
||||
job = generate_from_preset(preset, overrides)
|
||||
jobs.append({'job_id': job['id'], 'status': 'queued'})
|
||||
return jsonify({'jobs': jobs}), 202
|
||||
except Exception as e:
|
||||
logger.exception("API generation error (preset %s): %s", preset_slug, e)
|
||||
return jsonify({'error': str(e)}), 500
|
||||
|
||||
@app.route('/api/v1/job/<job_id>')
|
||||
@require_api_key
|
||||
def api_v1_job_status(job_id):
|
||||
with _job_queue_lock:
|
||||
job = _job_history.get(job_id)
|
||||
if not job:
|
||||
return jsonify({'error': 'Job not found'}), 404
|
||||
resp = {
|
||||
'id': job['id'],
|
||||
'label': job['label'],
|
||||
'status': job['status'],
|
||||
'error': job['error'],
|
||||
}
|
||||
if job.get('result'):
|
||||
resp['result'] = job['result']
|
||||
return jsonify(resp)
|
||||
@@ -11,6 +11,7 @@ from services.prompts import build_prompt, _dedup_tags, _resolve_character, _ens
|
||||
from services.workflow import _prepare_workflow, _get_default_checkpoint
|
||||
from services.job_queue import _enqueue_job, _make_finalize
|
||||
from services.sync import sync_presets, _resolve_preset_entity, _resolve_preset_fields, _PRESET_ENTITY_MAP
|
||||
from services.generation import generate_from_preset
|
||||
from services.llm import load_prompt, call_llm
|
||||
from utils import allowed_file
|
||||
|
||||
@@ -39,180 +40,30 @@ def register_routes(app):
|
||||
|
||||
try:
|
||||
action = request.form.get('action', 'preview')
|
||||
|
||||
# Get additional prompts
|
||||
extra_positive = request.form.get('extra_positive', '').strip()
|
||||
extra_negative = request.form.get('extra_negative', '').strip()
|
||||
|
||||
# Save to session (web UI state)
|
||||
session[f'extra_pos_preset_{slug}'] = extra_positive
|
||||
session[f'extra_neg_preset_{slug}'] = extra_negative
|
||||
session.modified = True
|
||||
|
||||
data = preset.data
|
||||
|
||||
# Resolve entities
|
||||
char_cfg = data.get('character', {})
|
||||
character = _resolve_preset_entity('character', char_cfg.get('character_id'))
|
||||
if not character:
|
||||
character = Character.query.order_by(db.func.random()).first()
|
||||
|
||||
outfit_cfg = data.get('outfit', {})
|
||||
action_cfg = data.get('action', {})
|
||||
style_cfg = data.get('style', {})
|
||||
scene_cfg = data.get('scene', {})
|
||||
detailer_cfg = data.get('detailer', {})
|
||||
look_cfg = data.get('look', {})
|
||||
ckpt_cfg = data.get('checkpoint', {})
|
||||
|
||||
outfit = _resolve_preset_entity('outfit', outfit_cfg.get('outfit_id'))
|
||||
action_obj = _resolve_preset_entity('action', action_cfg.get('action_id'))
|
||||
style_obj = _resolve_preset_entity('style', style_cfg.get('style_id'))
|
||||
scene_obj = _resolve_preset_entity('scene', scene_cfg.get('scene_id'))
|
||||
detailer_obj = _resolve_preset_entity('detailer', detailer_cfg.get('detailer_id'))
|
||||
look_obj = _resolve_preset_entity('look', look_cfg.get('look_id'))
|
||||
|
||||
# Checkpoint: form override > preset config > session default
|
||||
checkpoint_override = request.form.get('checkpoint_override', '').strip()
|
||||
if checkpoint_override:
|
||||
ckpt_obj = Checkpoint.query.filter_by(checkpoint_path=checkpoint_override).first()
|
||||
ckpt_path = checkpoint_override
|
||||
ckpt_data = ckpt_obj.data if ckpt_obj else None
|
||||
else:
|
||||
preset_ckpt = ckpt_cfg.get('checkpoint_path')
|
||||
if preset_ckpt == 'random':
|
||||
ckpt_obj = Checkpoint.query.order_by(db.func.random()).first()
|
||||
ckpt_path = ckpt_obj.checkpoint_path if ckpt_obj else None
|
||||
ckpt_data = ckpt_obj.data if ckpt_obj else None
|
||||
elif preset_ckpt:
|
||||
ckpt_obj = Checkpoint.query.filter_by(checkpoint_path=preset_ckpt).first()
|
||||
ckpt_path = preset_ckpt
|
||||
ckpt_data = ckpt_obj.data if ckpt_obj else None
|
||||
else:
|
||||
ckpt_path, ckpt_data = _get_default_checkpoint()
|
||||
|
||||
# Resolve selected fields from preset toggles
|
||||
selected_fields = _resolve_preset_fields(data)
|
||||
|
||||
# Build combined data for prompt building
|
||||
active_wardrobe = char_cfg.get('fields', {}).get('wardrobe', {}).get('outfit', 'default')
|
||||
wardrobe_source = outfit.data.get('wardrobe', {}) if outfit else None
|
||||
if wardrobe_source is None:
|
||||
wardrobe_source = character.get_active_wardrobe() if character else {}
|
||||
|
||||
combined_data = {
|
||||
'character_id': character.character_id if character else 'unknown',
|
||||
'identity': character.data.get('identity', {}) if character else {},
|
||||
'defaults': character.data.get('defaults', {}) if character else {},
|
||||
'wardrobe': wardrobe_source,
|
||||
'styles': character.data.get('styles', {}) if character else {},
|
||||
'lora': (look_obj.data.get('lora', {}) if look_obj
|
||||
else (character.data.get('lora', {}) if character else {})),
|
||||
'tags': (character.data.get('tags', []) if character else []) + data.get('tags', []),
|
||||
}
|
||||
|
||||
# Build extras prompt from secondary resources
|
||||
extras_parts = []
|
||||
if action_obj:
|
||||
action_fields = action_cfg.get('fields', {})
|
||||
from utils import _BODY_GROUP_KEYS
|
||||
for key in _BODY_GROUP_KEYS:
|
||||
val_cfg = action_fields.get(key, True)
|
||||
if val_cfg == 'random':
|
||||
val_cfg = random.choice([True, False])
|
||||
if val_cfg:
|
||||
val = action_obj.data.get('action', {}).get(key, '')
|
||||
if val:
|
||||
extras_parts.append(val)
|
||||
if action_cfg.get('use_lora', True):
|
||||
trg = action_obj.data.get('lora', {}).get('lora_triggers', '')
|
||||
if trg:
|
||||
extras_parts.append(trg)
|
||||
extras_parts.extend(action_obj.data.get('tags', []))
|
||||
if style_obj:
|
||||
s = style_obj.data.get('style', {})
|
||||
if s.get('artist_name'):
|
||||
extras_parts.append(f"by {s['artist_name']}")
|
||||
if s.get('artistic_style'):
|
||||
extras_parts.append(s['artistic_style'])
|
||||
if style_cfg.get('use_lora', True):
|
||||
trg = style_obj.data.get('lora', {}).get('lora_triggers', '')
|
||||
if trg:
|
||||
extras_parts.append(trg)
|
||||
if scene_obj:
|
||||
scene_fields = scene_cfg.get('fields', {})
|
||||
for key in ['background', 'foreground', 'furniture', 'colors', 'lighting', 'theme']:
|
||||
val_cfg = scene_fields.get(key, True)
|
||||
if val_cfg == 'random':
|
||||
val_cfg = random.choice([True, False])
|
||||
if val_cfg:
|
||||
val = scene_obj.data.get('scene', {}).get(key, '')
|
||||
if val:
|
||||
extras_parts.append(val)
|
||||
if scene_cfg.get('use_lora', True):
|
||||
trg = scene_obj.data.get('lora', {}).get('lora_triggers', '')
|
||||
if trg:
|
||||
extras_parts.append(trg)
|
||||
extras_parts.extend(scene_obj.data.get('tags', []))
|
||||
if detailer_obj:
|
||||
prompt_val = detailer_obj.data.get('prompt', '')
|
||||
if isinstance(prompt_val, list):
|
||||
extras_parts.extend(p for p in prompt_val if p)
|
||||
elif prompt_val:
|
||||
extras_parts.append(prompt_val)
|
||||
if detailer_cfg.get('use_lora', True):
|
||||
trg = detailer_obj.data.get('lora', {}).get('lora_triggers', '')
|
||||
if trg:
|
||||
extras_parts.append(trg)
|
||||
|
||||
with open('comfy_workflow.json', 'r') as f:
|
||||
workflow = json.load(f)
|
||||
|
||||
prompts = build_prompt(combined_data, selected_fields, default_fields=None,
|
||||
active_outfit=active_wardrobe)
|
||||
if extras_parts:
|
||||
extra_str = ', '.join(filter(None, extras_parts))
|
||||
prompts['main'] = _dedup_tags(f"{prompts['main']}, {extra_str}" if prompts['main'] else extra_str)
|
||||
|
||||
if extra_positive:
|
||||
prompts["main"] = f"{prompts['main']}, {extra_positive}"
|
||||
|
||||
# Parse optional seed
|
||||
# Build overrides from form
|
||||
seed_val = request.form.get('seed', '').strip()
|
||||
fixed_seed = int(seed_val) if seed_val else None
|
||||
|
||||
# Resolution: form override > preset config > workflow default
|
||||
res_cfg = data.get('resolution', {})
|
||||
form_width = request.form.get('width', '').strip()
|
||||
form_height = request.form.get('height', '').strip()
|
||||
if form_width and form_height:
|
||||
gen_width = int(form_width)
|
||||
gen_height = int(form_height)
|
||||
elif res_cfg.get('random', False):
|
||||
_RES_OPTIONS = [
|
||||
(1024, 1024), (1152, 896), (896, 1152), (1344, 768),
|
||||
(768, 1344), (1280, 800), (800, 1280),
|
||||
]
|
||||
gen_width, gen_height = random.choice(_RES_OPTIONS)
|
||||
else:
|
||||
gen_width = res_cfg.get('width') or None
|
||||
gen_height = res_cfg.get('height') or None
|
||||
|
||||
workflow = _prepare_workflow(
|
||||
workflow, character, prompts,
|
||||
checkpoint=ckpt_path, checkpoint_data=ckpt_data,
|
||||
custom_negative=extra_negative or None,
|
||||
outfit=outfit if outfit_cfg.get('use_lora', True) else None,
|
||||
action=action_obj if action_cfg.get('use_lora', True) else None,
|
||||
style=style_obj if style_cfg.get('use_lora', True) else None,
|
||||
scene=scene_obj if scene_cfg.get('use_lora', True) else None,
|
||||
detailer=detailer_obj if detailer_cfg.get('use_lora', True) else None,
|
||||
look=look_obj,
|
||||
fixed_seed=fixed_seed,
|
||||
width=gen_width,
|
||||
height=gen_height,
|
||||
)
|
||||
overrides = {
|
||||
'action': action,
|
||||
'extra_positive': extra_positive,
|
||||
'extra_negative': extra_negative,
|
||||
'checkpoint': request.form.get('checkpoint_override', '').strip(),
|
||||
'seed': int(seed_val) if seed_val else None,
|
||||
'width': int(form_width) if form_width else None,
|
||||
'height': int(form_height) if form_height else None,
|
||||
}
|
||||
|
||||
label = f"Preset: {preset.name} – {action}"
|
||||
job = _enqueue_job(label, workflow, _make_finalize('presets', slug, Preset, action))
|
||||
job = generate_from_preset(preset, overrides)
|
||||
|
||||
session[f'preview_preset_{slug}'] = None
|
||||
session.modified = True
|
||||
|
||||
@@ -1,9 +1,12 @@
|
||||
import json
|
||||
import logging
|
||||
import os
|
||||
import random
|
||||
import secrets
|
||||
import subprocess
|
||||
|
||||
import requests
|
||||
from flask import flash, redirect, render_template, request, session, url_for
|
||||
from flask import flash, jsonify, redirect, render_template, request, session, url_for
|
||||
|
||||
from models import Checkpoint, Settings, db
|
||||
|
||||
@@ -12,6 +15,19 @@ logger = logging.getLogger('gaze')
|
||||
|
||||
def register_routes(app):
|
||||
|
||||
@app.template_global()
|
||||
def random_gen_image(category, slug):
|
||||
"""Return a random generated image path for a resource, or None."""
|
||||
folder = os.path.join(app.config['UPLOAD_FOLDER'], category, slug)
|
||||
try:
|
||||
images = [f for f in os.listdir(folder)
|
||||
if f.lower().endswith(('.png', '.jpg', '.jpeg', '.webp'))]
|
||||
except FileNotFoundError:
|
||||
return None
|
||||
if not images:
|
||||
return None
|
||||
return f"{category}/{slug}/{random.choice(images)}"
|
||||
|
||||
@app.context_processor
|
||||
def inject_comfyui_ws_url():
|
||||
url = app.config.get('COMFYUI_URL', 'http://127.0.0.1:8188')
|
||||
@@ -227,3 +243,14 @@ def register_routes(app):
|
||||
return redirect(url_for('settings'))
|
||||
|
||||
return render_template('settings.html', settings=settings)
|
||||
|
||||
@app.route('/api/key/regenerate', methods=['POST'])
|
||||
def regenerate_api_key():
|
||||
settings = Settings.query.first()
|
||||
if not settings:
|
||||
settings = Settings()
|
||||
db.session.add(settings)
|
||||
settings.api_key = secrets.token_hex(32)
|
||||
db.session.commit()
|
||||
logger.info("API key regenerated")
|
||||
return jsonify({'api_key': settings.api_key})
|
||||
|
||||
Reference in New Issue
Block a user