appneta

#!/usr/bin/env python3
"""
=====================================================================
🌟 NETA-LUMINA BACKEND v1.0.4 — OFFICIAL RECIPE
=====================================================================
Model       : neta-art/Neta-Lumina v1.0
Base        : Alpha-VLLM/Lumina-Image-2.0
Pipeline    : diffusers.Lumina2Pipeline (v0.39.0)
GPU         : ZeroGPU RTX 6000 Pro Blackwell (96GB GDDR7)

v1.0.4 — OFFICIAL RECIPE (validated against lumina_workflow.json,
       Neta Prompt Book, and pipeline_lumina2.py source):
  🔧 PREFIX EXACT: "You are an assistant designed to generate anime
     images based on textual prompts. \n" — TANPA <Prompt Start>
  🔧 NEGATIVE PREFIX: "...generate low-quality images..." (Prompt Book)
  🔧 4 prompt style variants resmi (Anime/Danbooru/NL/Structured)
  🔧 cfg_trunc_ratio default 1.0 (workflow resmi: CFG semua step)
  🔧 CFG default 5.5 (workflow resmi)
  🔧 Dynamic Shift jadi default (mu-based, pipeline design)
  🔧 + Flow Heun sampler (FlowMatchHeunDiscreteScheduler) — padanan
     terdekat resmi res_multistep
  🔧 max_sequence_length 512 (prompt tag+NL panjang tidak terpotong)

v1.0.3: system prompt via pipeline mechanism, negative plain,
        CFG Decay dihapus (incompatible Lumina2)
v1.0.2: parameter order fix
v1.0.1: low_cpu_mem_usage (OOM fix)

PARAMETER ORDER (15 params) — frontend data[] HARUS match:
   1. prompt                 9. width
   2. negative_prompt       10. height
   3. sampler               11. enable_cfg_normalization
   4. schedule              12. cfg_trunc_ratio
   5. steps                 13. prompt_style   ← BARU
   6. cfg_scale             14. token_budget
   7. seed                  15. hf_token
   8. batch_size
=====================================================================
"""

# =====================================================================
# 🌍 SECTION 0: ENVIRONMENT VARIABLES (BEFORE ANY IMPORT!)
# =====================================================================
import os
os.environ["PYTORCH_CUDA_ALLOC_CONF"] = "expandable_segments:True"
os.environ["SAFETENSORS_FAST_GPU"] = "1"
os.environ["HF_HUB_ENABLE_HF_TRANSFER"] = "0"
os.environ["TOKENIZERS_PARALLELISM"] = "false"
os.environ["TRANSFORMERS_VERBOSITY"] = "error"
os.environ["HF_HUB_DISABLE_PROGRESS_BARS"] = "0"

# =====================================================================
# 🖥️ SECTION 1: ZERO-GPU DETECTION
# =====================================================================
try:
    import spaces
    ZERO_GPU = True
    print("🖥️ ZeroGPU Runtime Detected")
except ImportError:
    spaces = None
    ZERO_GPU = False
    print("🖥️ Dedicated GPU Runtime")

# =====================================================================
# 📦 SECTION 2: IMPORTS
# =====================================================================
import re
import gc
import math
import json
import time
import random
import traceback
import base64
import sys
import uuid
import threading
import urllib.request
import warnings
from io import BytesIO
from urllib.parse import unquote

warnings.filterwarnings("ignore", message=".*torchao.*")
warnings.filterwarnings("ignore", message=".*Tensor objects.*")

from PIL import Image
import torch
import gradio as gr
from huggingface_hub import hf_hub_download

_orig_torch_load = torch.load
def _patched_torch_load(*a, **k):
    k.setdefault("weights_only", False)
    return _orig_torch_load(*a, **k)
torch.load = _patched_torch_load

import diffusers as _diffusers
print(f"📦 diffusers version: {_diffusers.__version__}")
_vp = _diffusers.__version__.split(".")
if int(_vp[0]) == 0 and int(_vp[1]) < 33:
    raise ImportError(f"❌ diffusers >= 0.33.0 required! Got {_diffusers.__version__}")

from diffusers import (
    Lumina2Pipeline,
    Lumina2Transformer2DModel,
    FlowMatchEulerDiscreteScheduler,
    FlowMatchHeunDiscreteScheduler,
    AutoencoderKL,
)

from fastapi import Request
from fastapi.middleware.cors import CORSMiddleware
from starlette.responses import JSONResponse
from starlette.exceptions import HTTPException as StarletteHTTPException

print("✅ All imports successful")
print("   Pipeline  : Lumina2Pipeline (Flow Matching)")
print("   Backbone  : Lumina2Transformer2DModel (DiT)")
print("   Schedulers: FlowMatchEuler + FlowMatchHeun")
print("   Text Enc  : Gemma-2-2B (built-in)")
print("   VAE       : FLUX 16-ch AutoencoderKL (built-in)")

# =====================================================================
# 📊 SECTION 3: MEMORY MONITORING + GPU OPTIMIZATION
# =====================================================================
def _log_memory(stage: str):
    try:
        with open("/proc/meminfo") as f:
            for line in f:
                if line.startswith("MemAvailable:"):
                    print(f"📊 [{stage}] RAM Available: {int(line.split()[1]) / (1024*1024):.1f} GB")
                    break
    except Exception:
        pass
    try:
        if torch.cuda.is_available():
            print(f"📊 [{stage}] GPU: {torch.cuda.memory_allocated()/(1024**2):.0f}MB alloc / "
                  f"{torch.cuda.memory_reserved()/(1024**2):.0f}MB reserved")
    except Exception:
        pass

_log_memory("Init")

torch.backends.cuda.matmul.allow_tf32 = True
torch.backends.cudnn.allow_tf32 = True
torch.backends.cudnn.benchmark = True
torch.set_float32_matmul_precision("high")
try:
    torch.backends.cuda.enable_flash_sdp(True)
    print("⚡ Flash SDP enabled")
except AttributeError:
    pass
print("⚡ TF32=ON | bf16=ON | SDPA=FlashAttention")

# =====================================================================
# 🎯 SECTION 4: MODEL CONFIG
# =====================================================================
BASE_MODEL_REPO = "Alpha-VLLM/Lumina-Image-2.0"
NETA_MODEL_REPO = "neta-art/Neta-Lumina"
NETA_TRANSFORMER_FILE = "Unet/neta-lumina-v1.0.safetensors"
DISPLAY_NAME = "Neta-Lumina v1.0"
MODEL_ARCHITECTURE = "Lumina2 DiT (Flow Matching)"
TEXT_ENCODER_NAME = "Gemma-2-2B"
VAE_NAME = "FLUX 16-channel"
DEFAULT_BATCH = 1

# =====================================================================
# 📝 SECTION 5: OFFICIAL PROMPT PREFIXES (v1.0.4)
# =====================================================================
# Sumber: lumina_workflow.json + neta.art/blog/neta_lumina_prompt_book
#
# Positive: pipeline auto-prepend via param system_prompt:
#   hasil akhir = prefix + "  " + prompt  (mekanisme resmi pipeline)
# Negative: pipeline TIDAK pernah prepend system_prompt ke negative
#   (konfirmasi source encode_prompt) → prepend MANUAL ke string negative.
#
# CATATAN: <Prompt Start> TIDAK ADA di format resmi — sudah dihapus.
# =====================================================================
POSITIVE_PREFIXES = {
    "Anime (Standard)": (
        "You are an assistant designed to generate anime images "
        "based on textual prompts. \n"
    ),
    "Danbooru Tags": (
        "You are an assistant designed to generate anime images with the "
        "highest degree of image-text alignment based on danbooru tags. \n"
    ),
    "Natural Language": (
        "You are an assistant designed to generate high-quality images with "
        "the highest degree of image-text alignment based on textual prompts. \n"
    ),
    "Structured": (
        "You are an assistant designed to generate high-quality images with "
        "the highest degree of image-text alignment based on structural summary. \n"
    ),
}
DEFAULT_PROMPT_STYLE = "Anime (Standard)"

NEGATIVE_PREFIX = (
    "You are an assistant designed to generate low-quality images "
    "based on textual prompts.\n"
)

# =====================================================================
# 🔒 SECTION 6: VALIDATION LIMITS
# =====================================================================
MAX_STEPS = 60
MAX_BATCH_SIZE = 8
MAX_WIDTH = 2048
MAX_HEIGHT = 2048
MIN_WIDTH = 768
MIN_HEIGHT = 768
MAX_CFG = 12.0
MIN_CFG = 1.0

def validate_params(steps, batch_size, width, height):
    errors = []
    if steps < 1 or steps > MAX_STEPS:
        errors.append(f"Steps must be 1-{MAX_STEPS}, got {steps}")
    if batch_size < 1 or batch_size > MAX_BATCH_SIZE:
        errors.append(f"Batch size must be 1-{MAX_BATCH_SIZE}, got {batch_size}")
    if width < MIN_WIDTH or width > MAX_WIDTH:
        errors.append(f"Width must be {MIN_WIDTH}-{MAX_WIDTH}, got {width}")
    if height < MIN_HEIGHT or height > MAX_HEIGHT:
        errors.append(f"Height must be {MIN_HEIGHT}-{MAX_HEIGHT}, got {height}")
    if width % 16 != 0:
        errors.append(f"Width must be divisible by 16, got {width}")
    if height % 16 != 0:
        errors.append(f"Height must be divisible by 16, got {height}")
    return errors

# =====================================================================
# 📊 SECTION 7: ERROR DETECTION
# =====================================================================
def is_quota_error(error_msg):
    msg = str(error_msg).lower()
    return any(kw in msg for kw in ["exceeded", "quota", "rate limit", "too many"])

def is_gpu_cold_start_error(error_msg):
    msg = str(error_msg).lower()
    return any(kw in msg for kw in [
        "no gpu ", "gpu was available ", "available after ", "no gpu was ",
        "gpu is not available ", "gpu not available ", "not ready ",
    ])

# =====================================================================
# 🫀 SECTION 8: KEEP-ALIVE ENGINE
# =====================================================================
KEEP_ALIVE_INTERVAL = 180

def _send_keep_alive_ping():
    for endpoint in ["http://localhost:7860/", "http://localhost:7860/health"]:
        try:
            req = urllib.request.Request(endpoint,
                headers={"User-Agent": "NetaLumina-KeepAlive/1.0"}, method="GET")
            urllib.request.urlopen(req, timeout=10)
        except Exception:
            pass

def keep_alive_worker():
    time.sleep(30)
    print(f"🫀 Keep-alive started (interval: {KEEP_ALIVE_INTERVAL}s)")
    while True:
        time.sleep(KEEP_ALIVE_INTERVAL)
        _send_keep_alive_ping()
        print(f"🫀 Keep-alive ping @ {time.strftime('%H:%M:%S')}")

def start_keep_alive():
    t = threading.Thread(target=keep_alive_worker, daemon=True, name="keep-alive")
    t.start()
    return t

# =====================================================================
# 📥 SECTION 9: MODEL LOADING — MEMORY-EFFICIENT
# =====================================================================
_log_memory("Before Download")

print(f"📥 Step 1/6: Downloading Neta-Lumina transformer...")
print(f"   Repo: {NETA_MODEL_REPO} | File: {NETA_TRANSFORMER_FILE}")

neta_transformer_path = None
MAX_DOWNLOAD_RETRIES = 3
for attempt in range(1, MAX_DOWNLOAD_RETRIES + 1):
    try:
        print(f"   Attempt {attempt}/{MAX_DOWNLOAD_RETRIES}...")
        neta_transformer_path = hf_hub_download(
            repo_id=NETA_MODEL_REPO, filename=NETA_TRANSFORMER_FILE)
        print(f"✅ Downloaded (anonymous): {neta_transformer_path}")
        break
    except Exception as e:
        print(f"⚠️ Attempt {attempt} failed: {str(e)[:120]}")
        if attempt < MAX_DOWNLOAD_RETRIES:
            time.sleep(attempt * 10)

if neta_transformer_path is None:
    hf_env = os.environ.get("HF_TOKEN")
    if hf_env:
        try:
            neta_transformer_path = hf_hub_download(
                repo_id=NETA_MODEL_REPO, filename=NETA_TRANSFORMER_FILE, token=hf_env)
            print(f"✅ Downloaded (HF_TOKEN): {neta_transformer_path}")
        except Exception as e2:
            print(f"⚠️ HF_TOKEN attempt failed: {str(e2)[:120]}")

if neta_transformer_path is None:
    raise RuntimeError("❌ Failed to download Neta transformer. Set HF_TOKEN in Space secrets.")

_log_memory("After Download")

print("📥 Step 2/6: Loading Neta Transformer (low_cpu_mem_usage=True)...")
neta_transformer = Lumina2Transformer2DModel.from_single_file(
    neta_transformer_path, torch_dtype=torch.bfloat16, low_cpu_mem_usage=True)
print("✅ Neta Transformer loaded!")
print(f"   Parameters: {sum(p.numel() for p in neta_transformer.parameters())/1e6:.1f}M")
print(f"   Layers: {neta_transformer.config.num_layers} | Hidden: {neta_transformer.config.hidden_size}")
print(f"   Heads: {neta_transformer.config.num_attention_heads} | In-ch: {neta_transformer.config.in_channels}")
gc.collect()
_log_memory("After Transformer Load")

print(f"📥 Step 3/6: Loading Pipeline from {BASE_MODEL_REPO} (low_cpu_mem_usage=True)...")
pipe = Lumina2Pipeline.from_pretrained(
    BASE_MODEL_REPO, transformer=neta_transformer,
    torch_dtype=torch.bfloat16, low_cpu_mem_usage=True)
del neta_transformer
gc.collect()
print("✅ Pipeline loaded!")
_log_memory("After Pipeline Load")

print("📥 Step 4/6: Cleanup downloaded file...")
try:
    if neta_transformer_path and os.path.exists(neta_transformer_path):
        sz = os.path.getsize(neta_transformer_path) / (1024**3)
        os.remove(neta_transformer_path)
        print(f"✅ Deleted ({sz:.1f}GB freed)")
except Exception as e:
    print(f"⚠️ {e}")
gc.collect()
_log_memory("After Cleanup")

print("📥 Step 5/6: Moving to GPU (component by component)...")
pipe.text_encoder.to("cuda"); torch.cuda.empty_cache(); _log_memory("text_encoder → GPU")
pipe.transformer.to("cuda"); torch.cuda.empty_cache(); _log_memory("transformer → GPU")
pipe.vae.to("cuda"); torch.cuda.empty_cache(); _log_memory("VAE → GPU")
print("✅ All components on GPU!")

print("📥 Step 6/6: Post-load optimizations...")
pipe.transformer.to(memory_format=torch.channels_last)
pipe.vae.to(memory_format=torch.channels_last)
pipe.vae.to(dtype=torch.bfloat16)
try:
    pipe.vae.config.force_upcast = False
except AttributeError:
    pass
pipe.vae.enable_tiling()
print("   📐 Channels Last + VAE Tiling + Strict BF16 aktif")

_orig_vae_decode = pipe.vae.decode
_orig_vae_encode = pipe.vae.encode
def _patched_vae_decode(z, return_dict=True, **kw):
    torch.backends.cudnn.benchmark = False
    try:
        return _orig_vae_decode(z, return_dict=return_dict, **kw)
    finally:
        torch.backends.cudnn.benchmark = True
def _patched_vae_encode(x, return_dict=True, **kw):
    torch.backends.cudnn.benchmark = False
    try:
        return _orig_vae_encode(x, return_dict=return_dict, **kw)
    finally:
        torch.backends.cudnn.benchmark = True
pipe.vae.decode = _patched_vae_decode
pipe.vae.encode = _patched_vae_encode
print("   🛡️ cuDNN Benchmark Isolation aktif")

gc.collect(); torch.cuda.empty_cache()
_log_memory("Final (Ready)")
print(f"📊 GPU after load: {torch.cuda.memory_allocated()/(1024**3):.2f} GB")

# =====================================================================
# 🎯 SECTION 10: SCHEDULER SYSTEM (v1.0.4 — + Flow Heun)
# =====================================================================
# Official Neta: res_multistep / euler_ancestral + linear_quadratic.
# Padanan terdekat di diffusers:
#   - Flow Euler  : deterministik orde-1 (baseline)
#   - Flow Heun   : multi-step orde-2 → paling dekat ke res_multistep
# Schedule:
#   - Dynamic Shift : mu-based (pipeline hitung mu otomatis) → DEFAULT
#     (empiris lebih bagus; shift efektif ~1.58 vs Linear yang 6.0)
#   - Linear        : fixed shift=6.0
# =====================================================================
BASE_SCHEDULER_CONFIG = {
    "_class_name": "FlowMatchEulerDiscreteScheduler",
    "_diffusers_version": "0.33.0",
    "base_image_seq_len": 256,
    "base_shift": 0.5,
    "invert_sigmas": False,
    "max_image_seq_len": 4096,
    "max_shift": 1.15,
    "num_train_timesteps": 1000,
    "shift": 6.0,
    "shift_terminal": None,
    "use_beta_sigmas": False,
    "use_dynamic_shifting": False,
    "use_exponential_sigmas": False,
    "use_karras_sigmas": False,
}

SAMPLER_REGISTRY = {
    "Flow Euler": {
        "class": FlowMatchEulerDiscreteScheduler,
        "params": {},
        "schedules": ["Dynamic Shift", "Linear"],
        "tag": "flow",
    },
    "Flow Heun": {
        "class": FlowMatchHeunDiscreteScheduler,
        "params": {},
        "schedules": ["Dynamic Shift", "Linear"],
        "tag": "flow",
    },
}

SCHEDULE_REGISTRY = {
    "Dynamic Shift": {
        "params": {"use_dynamic_shifting": True},
        "description": "mu-based shift (pipeline calculate_shift) — recommended",
    },
    "Linear": {
        "params": {"shift": 6.0, "use_dynamic_shifting": False},
        "description": "Fixed shift=6.0",
    },
}

def build_scheduler(sampler_name, schedule_name):
    if sampler_name not in SAMPLER_REGISTRY:
        return None, f"Sampler '{sampler_name}' tidak ditemukan!"
    if schedule_name not in SCHEDULE_REGISTRY:
        return None, f"Schedule '{schedule_name}' tidak ditemukan!"
    info = SAMPLER_REGISTRY[sampler_name]
    if schedule_name not in info["schedules"]:
        return None, f"'{schedule_name}' tidak compatible. Compatible: {', '.join(info['schedules'])}"
    cfg = dict(BASE_SCHEDULER_CONFIG)
    cfg.update(info["params"])
    cfg.update(SCHEDULE_REGISTRY[schedule_name]["params"])
    cfg["_class_name"] = info["class"].__name__   # pastikan kelas scheduler benar
    try:
        return info["class"].from_config(cfg), None
    except Exception as e:
        return None, f"Gagal membangun scheduler: {str(e)}"

default_scheduler, _ = build_scheduler("Flow Euler", "Dynamic Shift")
if default_scheduler is not None:
    pipe.scheduler = default_scheduler
    print("✅ Default scheduler: Flow Euler + Dynamic Shift (mu-based)")
TOTAL_COMBOS = sum(len(i["schedules"]) for i in SAMPLER_REGISTRY.values())
print(f"✅ SAMPLER SYSTEM: {len(SAMPLER_REGISTRY)} samplers × {len(SCHEDULE_REGISTRY)} schedules = {TOTAL_COMBOS}")

# =====================================================================
# ⏱️ SECTION 11: GPU TOKEN BUDGET
# =====================================================================
GPU_DURATION_TIERS = [15, 30, 60, 120, 180]
GPU_WRAPPED_FNS = {}

def estimate_gpu_duration(steps, batch_size, width, height, sampler):
    megapixels = (width * height) / (1024 * 1024)
    base_time = megapixels * steps * 0.20 * batch_size
    if "Heun" in str(sampler):
        base_time *= 2.0    # Heun = 2 NFE per step
    estimated = base_time * 1.3
    for dur in GPU_DURATION_TIERS:
        if estimated <= dur:
            return dur, f"~{base_time:.1f}s est → {dur}s tier"
    return GPU_DURATION_TIERS[-1], f"~{base_time:.1f}s est → {GPU_DURATION_TIERS[-1]}s (MAX)"

# =====================================================================
# 🖼️ SECTION 12: IMAGE ENCODING
# =====================================================================
def encode_images_to_base64(images, seed, quality=92):
    results = []
    for i, img in enumerate(images):
        buf = BytesIO()
        img.save(buf, format="JPEG", quality=quality)
        img_bytes = buf.getvalue()
        b64 = base64.b64encode(img_bytes).decode("utf-8")
        results.append({
            "data": f"data:image/jpeg;base64,{b64}",
            "name": f"neta_{seed}_{i}.jpg", "seed": seed, "index": i,
            "width": img.width, "height": img.height, "bytes": len(img_bytes),
        })
    return results

# =====================================================================
# 🏞️ SECTION 13: RAM GALLERY SYSTEM v1.1
# =====================================================================
RAM_GALLERIES = {}
RAM_GALLERY_LOCK = threading.RLock()
RAM_GALLERY_START_TIME = time.time()
RAM_GALLERY_MAX_GALLERIES = 100
RAM_GALLERY_MAX_IMAGES_PER_GALLERY = 1000
RAM_GALLERY_MAX_TOTAL_IMAGES = 10000
RAM_GALLERY_MAX_NAME_LENGTH = 80
RAM_GALLERY_MAX_DESC_LENGTH = 500

def _sanitize_gallery_name(name):
    name = "" if name is None else str(name)
    name = name.strip()
    if not name:
        return ""
    name = name.replace("/", "-").replace("\\", "-")
    name = re.sub(r'[<>:"|?#&%*\x00-\x1f]', "", name).strip()
    return re.sub(r"\s+", " ", name)

def _estimate_base64_bytes(data_url):
    if not data_url or "," not in data_url:
        return 0
    return int(len(data_url.split(",", 1)[-1]) * 3 / 4)

def _get_total_image_count():
    return sum(len(g["images"]) for g in RAM_GALLERIES.values())

def _get_total_ram_bytes():
    return sum(img.get("bytes", 0) for g in RAM_GALLERIES.values() for img in g["images"])

def _gallery_summary(gallery, include_images=False):
    total_bytes = sum(img.get("bytes", 0) for img in gallery["images"])
    s = {
        "id": gallery["id"], "name": gallery["name"],
        "description": gallery["description"],
        "image_count": len(gallery["images"]),
        "total_bytes": total_bytes,
        "total_mb": round(total_bytes / (1024*1024), 2),
        "created_at": gallery["created_at"],
        "last_modified": gallery["last_modified"],
    }
    if include_images:
        s["images"] = gallery["images"]
    return s

def ram_gallery_create(name, description=""):
    name = _sanitize_gallery_name(name)
    description = ("" if description is None else str(description)).strip()
    if not name:
        return False, "Gallery name cannot be empty", None
    if len(name) > RAM_GALLERY_MAX_NAME_LENGTH:
        return False, f"Name too long (max {RAM_GALLERY_MAX_NAME_LENGTH})", None
    if len(description) > RAM_GALLERY_MAX_DESC_LENGTH:
        description = description[:RAM_GALLERY_MAX_DESC_LENGTH]
    with RAM_GALLERY_LOCK:
        if name in RAM_GALLERIES:
            return False, f"Gallery '{name}' already exists", None
        if len(RAM_GALLERIES) >= RAM_GALLERY_MAX_GALLERIES:
            return False, f"Max galleries reached ({RAM_GALLERY_MAX_GALLERIES})", None
        gid = str(uuid.uuid4())[:8]
        now = time.strftime("%Y-%m-%d %H:%M:%S")
        RAM_GALLERIES[name] = {"id": gid, "name": name, "description": description,
                               "images": [], "created_at": now, "last_modified": now}
        print(f"🏞️ Gallery created: '{name}' (id: {gid})")
        return True, f"Gallery '{name}' created", gid

def ram_gallery_add_images(gallery_name, images):
    gallery_name = _sanitize_gallery_name(gallery_name)
    with RAM_GALLERY_LOCK:
        if gallery_name not in RAM_GALLERIES:
            return False, f"Gallery '{gallery_name}' not found", 0
        gallery = RAM_GALLERIES[gallery_name]
        cur = len(gallery["images"]); tot = _get_total_image_count(); new = len(images)
        if cur + new > RAM_GALLERY_MAX_IMAGES_PER_GALLERY:
            allowed = RAM_GALLERY_MAX_IMAGES_PER_GALLERY - cur
            if allowed <= 0:
                return False, "Gallery full", 0
            images = images[:allowed]; new = allowed
        if tot + new > RAM_GALLERY_MAX_TOTAL_IMAGES:
            allowed = RAM_GALLERY_MAX_TOTAL_IMAGES - tot
            if allowed <= 0:
                return False, "Total limit reached", 0
            images = images[:allowed]; new = allowed
        added = 0
        for d in images:
            entry = {
                "id": str(uuid.uuid4())[:12], "data": d.get("data", ""),
                "seed": d.get("seed", "N/A"), "sampler": d.get("sampler", "N/A"),
                "schedule": d.get("schedule", "N/A"), "steps": d.get("steps", 0),
                "cfg_scale": d.get("cfg_scale", 0), "width": d.get("width", 0),
                "height": d.get("height", 0), "prompt": d.get("prompt", ""),
                "negative_prompt": d.get("negative_prompt", ""),
                "duration": d.get("duration", 0), "pipeline": d.get("pipeline", "N/A"),
                "guidance": d.get("guidance", []),
                "bytes": _estimate_base64_bytes(d.get("data", "")),
                "added_at": time.strftime("%Y-%m-%d %H:%M:%S"),
            }
            gallery["images"].append(entry); added += 1
        gallery["last_modified"] = time.strftime("%Y-%m-%d %H:%M:%S")
        tb = sum(img.get("bytes", 0) for img in gallery["images"])
        print(f"🏞️ +{added} → '{gallery_name}' ({len(gallery['images'])} total, {tb//(1024*1024)}MB)")
        return True, f"Added {added} images to '{gallery_name}'", added

def ram_gallery_list():
    with RAM_GALLERY_LOCK:
        return [_gallery_summary(g) for g in RAM_GALLERIES.values()]

def ram_gallery_get_images(gallery_name):
    gallery_name = _sanitize_gallery_name(gallery_name)
    with RAM_GALLERY_LOCK:
        if gallery_name not in RAM_GALLERIES:
            return None, f"Gallery '{gallery_name}' not found"
        return _gallery_summary(RAM_GALLERIES[gallery_name], include_images=True), None

def ram_gallery_delete(gallery_name):
    gallery_name = _sanitize_gallery_name(gallery_name)
    with RAM_GALLERY_LOCK:
        if gallery_name not in RAM_GALLERIES:
            return False, f"Gallery '{gallery_name}' not found"
        c = len(RAM_GALLERIES[gallery_name]["images"])
        del RAM_GALLERIES[gallery_name]
        return True, f"Gallery '{gallery_name}' deleted ({c} images removed)"

def ram_gallery_delete_image(gallery_name, image_id):
    gallery_name = _sanitize_gallery_name(gallery_name)
    image_id = str(image_id).strip()
    with RAM_GALLERY_LOCK:
        if gallery_name not in RAM_GALLERIES:
            return False, f"Gallery '{gallery_name}' not found"
        g = RAM_GALLERIES[gallery_name]
        orig = len(g["images"])
        g["images"] = [i for i in g["images"] if i["id"] != image_id]
        if len(g["images"]) == orig:
            return False, f"Image '{image_id}' not found"
        g["last_modified"] = time.strftime("%Y-%m-%d %H:%M:%S")
        return True, f"Image removed from '{gallery_name}'"

def ram_gallery_rename(old_name, new_name):
    old_name = _sanitize_gallery_name(unquote(str(old_name)))
    new_name = _sanitize_gallery_name(new_name)
    if not new_name:
        return False, "New name cannot be empty"
    if len(new_name) > RAM_GALLERY_MAX_NAME_LENGTH:
        return False, f"Name too long (max {RAM_GALLERY_MAX_NAME_LENGTH})"
    with RAM_GALLERY_LOCK:
        if old_name not in RAM_GALLERIES:
            return False, f"Gallery '{old_name}' not found"
        if new_name in RAM_GALLERIES:
            return False, f"Gallery '{new_name}' already exists"
        g = RAM_GALLERIES.pop(old_name)
        g["name"] = new_name
        g["last_modified"] = time.strftime("%Y-%m-%d %H:%M:%S")
        RAM_GALLERIES[new_name] = g
        return True, f"Gallery renamed to '{new_name}'"

def ram_gallery_stats():
    with RAM_GALLERY_LOCK:
        tb = _get_total_ram_bytes()
        uptime = time.time() - RAM_GALLERY_START_TIME
        sys_ram = 0
        try:
            import resource
            sys_ram = resource.getrusage(resource.RUSAGE_SELF).ru_maxrss / 1024
        except Exception:
            pass
        ga = gr_ = "N/A"
        try:
            ga = f"{torch.cuda.memory_allocated()/(1024**3):.2f} GB"
            gr_ = f"{torch.cuda.memory_reserved()/(1024**3):.2f} GB"
        except Exception:
            pass
        details = [{"name": n, "id": g["id"], "image_count": len(g["images"]),
                    "total_mb": round(sum(i.get("bytes",0) for i in g["images"])/(1024*1024),2),
                    "created_at": g["created_at"], "last_modified": g["last_modified"]}
                   for n, g in RAM_GALLERIES.items()]
        return {
            "total_galleries": len(RAM_GALLERIES), "max_galleries": RAM_GALLERY_MAX_GALLERIES,
            "total_images": _get_total_image_count(),
            "max_images_per_gallery": RAM_GALLERY_MAX_IMAGES_PER_GALLERY,
            "max_total_images": RAM_GALLERY_MAX_TOTAL_IMAGES,
            "total_ram_used_mb": round(tb/(1024*1024),2),
            "total_ram_used_gb": round(tb/(1024**3),4),
            "process_rss_mb": round(sys_ram,1),
            "gpu_allocated": ga, "gpu_reserved": gr_,
            "uptime_hours": round(uptime/3600,2), "galleries": details,
        }

print("🏞️ RAM Gallery System ready")

# =====================================================================
# 🎨 SECTION 14: CORE GENERATION — 15 PARAMS (v1.0.4 OFFICIAL RECIPE)
# =====================================================================
_retry_depth = {"count": 0}

def generate_image(
    prompt,                    # 1
    negative_prompt,           # 2
    sampler,                   # 3
    schedule,                  # 4
    steps,                     # 5
    cfg_scale,                 # 6
    seed,                      # 7
    batch_size,                # 8
    width,                     # 9
    height,                    # 10
    enable_cfg_normalization,  # 11
    cfg_trunc_ratio,           # 12  (1.0 = CFG semua step, official)
    prompt_style,              # 13  ← BARU (prefix variant)
    token_budget,              # 14
    hf_token,                  # 15
):
    try:
        if hf_token and str(hf_token).strip():
            os.environ["HF_TOKEN"] = str(hf_token).strip()
            safe_token = str(hf_token)[:8] + "..."
        else:
            safe_token = "none"

        param_errors = validate_params(int(steps), int(batch_size), int(width), int(height))
        if param_errors:
            return json.dumps({"success": False,
                "error": f"Parameter Error: {' | '.join(param_errors)}",
                "images": [], "retryable": False})

        steps = min(int(steps), MAX_STEPS)
        batch_size = min(int(batch_size), MAX_BATCH_SIZE)
        width = max(MIN_WIDTH, min(int(width), MAX_WIDTH))
        height = max(MIN_HEIGHT, min(int(height), MAX_HEIGHT))

        req_sampler = str(sampler).strip() if sampler else "Flow Euler"
        req_schedule = str(schedule).strip() if schedule else "Dynamic Shift"
        if req_sampler not in SAMPLER_REGISTRY:
            req_sampler = "Flow Euler"
        if req_schedule not in SCHEDULE_REGISTRY:
            req_schedule = "Dynamic Shift"
        scheduler, _ = build_scheduler(req_sampler, req_schedule)
        if scheduler is not None:
            pipe.scheduler = scheduler
            sampler_used, schedule_used = req_sampler, req_schedule
        else:
            fb, _ = build_scheduler("Flow Euler", "Dynamic Shift")
            if fb:
                pipe.scheduler = fb
            sampler_used, schedule_used = "Flow Euler (Fallback)", "Dynamic Shift"

        final_seed = random.randint(0, 2**32 - 1) if int(seed) == -1 else int(seed)

        # Wildcard {a|b|c}
        while "{" in prompt:
            prompt = re.sub(r"\{([^{}]*)\}",
                lambda m: random.choice(m.group(1).split("|")).strip(), prompt)

        # =============================================================
        # 📝 PROMPT PREP — OFFICIAL RECIPE v1.0.4
        # Positive: prefix resmi via param system_prompt (pipeline yang
        #   prepend: prefix + "  " + prompt). TANPA <Prompt Start>.
        # Negative: prepend MANUAL prefix "low-quality" (pipeline tidak
        #   pernah menyentuh negative).
        # =============================================================
        style = str(prompt_style).strip() if prompt_style else DEFAULT_PROMPT_STYLE
        if style not in POSITIVE_PREFIXES:
            style = DEFAULT_PROMPT_STYLE
        positive_prefix = POSITIVE_PREFIXES[style]

        clean_prompt = prompt.strip()
        clean_negative = NEGATIVE_PREFIX + (negative_prompt.strip() if negative_prompt else "")

        base_cfg = max(MIN_CFG, min(float(cfg_scale), MAX_CFG))

        guidance_summary = [
            f"CFG={base_cfg}",
            f"norm={bool(enable_cfg_normalization)}",
            f"trunc={float(cfg_trunc_ratio)}",
            f"style={style}",
        ]
        print(f"🎨 {sampler_used}+{schedule_used} | {' | '.join(guidance_summary)}")

        st = time.time()
        generator = torch.Generator("cuda").manual_seed(final_seed)

        with torch.inference_mode():
            gen_kwargs = dict(
                prompt=clean_prompt,
                negative_prompt=clean_negative,
                num_inference_steps=int(steps),
                guidance_scale=base_cfg,
                width=int(width),
                height=int(height),
                num_images_per_prompt=int(batch_size),
                generator=generator,
                system_prompt=positive_prefix,   # ← mekanisme resmi pipeline
                cfg_normalization=bool(enable_cfg_normalization),
                cfg_trunc_ratio=float(cfg_trunc_ratio),
                max_sequence_length=512,         # ← prompt tag+NL panjang aman
            )

            try:
                result = pipe(**gen_kwargs)
                images = result.images
            except TypeError as te:
                err_str = str(te)
                removed = []
                for param in ["cfg_normalization", "cfg_trunc_ratio",
                              "system_prompt", "max_sequence_length"]:
                    if param in err_str and param in gen_kwargs:
                        gen_kwargs.pop(param); removed.append(param)
                if removed:
                    print(f"⚠️ Fallback: removed {removed}")
                    result = pipe(**gen_kwargs)
                    images = result.images
                else:
                    raise

        duration = time.time() - st
        print(f"✅ Generated in {duration:.2f}s | Seed: {final_seed}")
        _retry_depth["count"] = 0

        base64_data = encode_images_to_base64(images, final_seed)

        gpu_mem = "N/A"
        try:
            gpu_mem = f"{torch.cuda.memory_allocated()/(1024**3):.1f}GB"
        except Exception:
            pass

        flow_shift = "dynamic(mu)" if schedule_used == "Dynamic Shift" else 6.0
        try:
            if schedule_used != "Dynamic Shift":
                flow_shift = pipe.scheduler.config.get("shift", 6.0)
        except Exception:
            pass

        return json.dumps({
            "success": True,
            "images": base64_data,
            "metadata": {
                "seed": final_seed, "sampler": sampler_used, "schedule": schedule_used,
                "steps": steps, "cfg_scale": base_cfg, "width": width, "height": height,
                "batch_size": batch_size, "duration": round(duration, 2),
                "pipeline": "Lumina2 (Flow Matching)", "guidance": guidance_summary,
                "text_encoder": TEXT_ENCODER_NAME, "vae": VAE_NAME,
                "flow_shift": flow_shift,
                "cfg_normalization": bool(enable_cfg_normalization),
                "cfg_trunc_ratio": float(cfg_trunc_ratio),
                "prompt_style": style,
                "system_prompt": positive_prefix.strip()[:90],
                "token": safe_token, "gpu_memory": gpu_mem,
                "prompt": prompt, "negative_prompt": negative_prompt,
            },
        })

    except Exception as e:
        print(f"❌ ERROR: {str(e)[:200]}")
        traceback.print_exc()
        err_msg = str(e)
        if is_gpu_cold_start_error(err_msg):
            print("❄️ GPU Cold Start - waiting 10s...")
            time.sleep(10)
            if _retry_depth["count"] < 2:
                _retry_depth["count"] += 1
                return generate_image(
                    prompt, negative_prompt, sampler, schedule, steps, cfg_scale,
                    seed, batch_size, width, height, enable_cfg_normalization,
                    cfg_trunc_ratio, prompt_style, token_budget, hf_token)
            _retry_depth["count"] = 0
        lower = err_msg.lower()
        retryable = any(kw in lower for kw in [
            "gpu", "cuda", "oom", "memory", "timeout", "cold start",
            "not ready", "unavailable", "quota", "rate limit"])
        return json.dumps({"success": False, "error": err_msg[:500],
                           "images": [], "retryable": retryable})
    finally:
        if "HF_TOKEN" in os.environ:
            del os.environ["HF_TOKEN"]

# =====================================================================
# ⏱️ SECTION 15: SMART GENERATE WRAPPER (15 PARAMS)
# =====================================================================
def smart_generate(
    prompt, negative_prompt, sampler, schedule, steps, cfg_scale, seed,
    batch_size, width, height, enable_cfg_normalization, cfg_trunc_ratio,
    prompt_style, token_budget, hf_token,
):
    try:
        param_errors = validate_params(int(steps), int(batch_size), int(width), int(height))
        if param_errors:
            return json.dumps({"success": False,
                "error": f"Parameter Error: {' | '.join(param_errors)}",
                "images": [], "retryable": False})

        if token_budget == "Auto":
            actual_duration, est_text = estimate_gpu_duration(
                int(steps), int(batch_size), int(width), int(height), sampler)
            print(f"⏱️ Auto: {est_text}")
        else:
            actual_duration = int(str(token_budget).replace("s", ""))
        if actual_duration not in GPU_DURATION_TIERS:
            actual_duration = min(GPU_DURATION_TIERS, key=lambda x: abs(x - actual_duration))

        if ZERO_GPU and actual_duration in GPU_WRAPPED_FNS:
            print(f"🚀 GPU wrapper: {actual_duration}s")
            try:
                result = GPU_WRAPPED_FNS[actual_duration](
                    prompt, negative_prompt, sampler, schedule, steps, cfg_scale,
                    seed, batch_size, width, height, enable_cfg_normalization,
                    cfg_trunc_ratio, prompt_style, actual_duration, hf_token)
                if result is None:
                    return json.dumps({"success": False,
                        "error": "GPU wrapper returned None", "images": [], "retryable": True})
                if not isinstance(result, str):
                    return json.dumps({"success": False,
                        "error": f"GPU wrapper returned {type(result).__name__}",
                        "images": [], "retryable": True})
                try:
                    parsed = json.loads(result)
                    if not isinstance(parsed, dict):
                        return json.dumps({"success": False,
                            "error": f"Invalid format: {type(parsed).__name__}",
                            "images": [], "retryable": True})
                except json.JSONDecodeError as je:
                    return json.dumps({"success": False,
                        "error": f"Invalid JSON: {str(je)}", "images": [], "retryable": True})
                return result
            except Exception as gpu_err:
                err_msg = str(gpu_err)
                print(f"❌ GPU EXCEPTION: {err_msg}")
                traceback.print_exc()
                lower = err_msg.lower()
                is_cold = any(kw in lower for kw in [
                    "no gpu", "gpu was available", "available after", "not ready",
                    "cold start", "gpu timeout", "quota", "rate limit", "exceeded",
                    "too many", "sleeping", "waking up"])
                return json.dumps({"success": False,
                    "error": f"GPU {'Cold Start' if is_cold else 'Error'}: {err_msg[:300]}",
                    "images": [], "retryable": True})
        else:
            return generate_image(
                prompt, negative_prompt, sampler, schedule, steps, cfg_scale,
                seed, batch_size, width, height, enable_cfg_normalization,
                cfg_trunc_ratio, prompt_style, actual_duration, hf_token)

    except Exception as outer_err:
        print(f"❌ OUTER EXCEPTION: {outer_err}")
        traceback.print_exc()
        return json.dumps({"success": False,
            "error": f"Unexpected: {str(outer_err)[:300]}", "images": [], "retryable": True})

# =====================================================================
# 🔧 SECTION 16: GPU WRAPPERS
# =====================================================================
if ZERO_GPU:
    for dur in GPU_DURATION_TIERS:
        GPU_WRAPPED_FNS[dur] = spaces.GPU(duration=dur)(generate_image)
    print(f"✅ GPU wrappers: {list(GPU_WRAPPED_FNS.keys())}")

# =====================================================================
# 🎨 SECTION 17: GRADIO UI (15 PARAMS)
# =====================================================================
RESOLUTION_CHOICES = [
    768, 832, 896, 960, 1024, 1088, 1152, 1216, 1280, 1344,
    1408, 1472, 1536, 1600, 1664, 1728, 1792, 1856, 1920, 1984, 2048,
]

with gr.Blocks(title="Neta-Lumina Backend API") as demo:
    gr.Markdown("# 🌟 Neta-Lumina v1.0.4 Backend API — Official Recipe\nLumina2 DiT • Flow Matching • Gemma-2-2B • FLUX VAE")
    with gr.Row():
        with gr.Column(scale=2):
            prompt_input = gr.Textbox(label="Prompt", lines=4,
                value="1girl, solo, long hair, beautiful detailed eyes, gentle smile, looking at viewer, upper body, soft lighting, cherry blossom background, warm lighting, best quality")
            negative_prompt_input = gr.Textbox(label="Negative Prompt", lines=3,
                value="blurry, worst quality, low quality, jpeg artifacts, signature, watermark, username, error, deformed hands, bad anatomy, extra limbs, poorly drawn hands, poorly drawn face, mutation, deformed, extra eyes, extra arms, extra legs, malformed limbs, fused fingers, too many fingers, long neck, cross-eyed, bad proportions, missing arms, missing legs, extra digit, fewer digits, cropped, normal quality")
            prompt_style_dropdown = gr.Dropdown(label="Prompt Style (official prefixes)",
                choices=list(POSITIVE_PREFIXES.keys()), value=DEFAULT_PROMPT_STYLE)
            with gr.Accordion("Settings", open=True):
                with gr.Row():
                    sampler_dropdown = gr.Dropdown(label="Sampler",
                        choices=list(SAMPLER_REGISTRY.keys()), value="Flow Euler", scale=2)
                    schedule_dropdown = gr.Dropdown(label="Schedule",
                        choices=["Dynamic Shift", "Linear"], value="Dynamic Shift", scale=1)
                with gr.Row():
                    steps_slider = gr.Slider(label="Steps", minimum=1, maximum=MAX_STEPS, value=30, step=1)
                    cfg_slider = gr.Slider(label="CFG Scale", minimum=MIN_CFG, maximum=MAX_CFG, value=5.5, step=0.1)
                with gr.Row():
                    seed_input = gr.Number(label="Seed (-1=random)", value=-1, precision=0)
                    batch_slider = gr.Slider(label="Batch", minimum=1, maximum=MAX_BATCH_SIZE, value=1, step=1)
                with gr.Row():
                    width_dropdown = gr.Dropdown(label="Width", choices=RESOLUTION_CHOICES, value=1024)
                    height_dropdown = gr.Dropdown(label="Height", choices=RESOLUTION_CHOICES, value=1024)
                token_budget_dropdown = gr.Dropdown(label="Token Budget",
                    choices=["Auto", "15s", "30s", "60s", "120s", "180s"], value="Auto")
            with gr.Accordion("Advanced", open=False):
                enable_cfg_norm = gr.Checkbox(label="CFG Normalization", value=True)
                cfg_trunc_slider = gr.Slider(label="CFG Trunc Ratio (1.0 = CFG semua step, official)",
                    minimum=0.0, maximum=1.0, value=1.0, step=0.05)
                gr.Markdown("_Official workflow: CFG di semua step (trunc=1.0). Flow Heun ≈ res_multistep resmi — wajib A/B test._")
            hf_token_input = gr.Textbox(label="HF Token", visible=False, value="")
            generate_btn = gr.Button("🎨 Generate", variant="primary")
        with gr.Column(scale=3):
            json_output = gr.Textbox(label="Result JSON", lines=20, value="")

    def update_schedule_ui(s):
        if s in SAMPLER_REGISTRY:
            c = SAMPLER_REGISTRY[s]["schedules"]
            return gr.update(choices=c, value=c[0])
        return gr.update()
    sampler_dropdown.change(fn=update_schedule_ui, inputs=[sampler_dropdown], outputs=[schedule_dropdown])

    # 15 inputs — URUTAN MATCH frontend data[] array
    generate_btn.click(
        fn=smart_generate,
        inputs=[
            prompt_input,              # 1
            negative_prompt_input,     # 2
            sampler_dropdown,          # 3
            schedule_dropdown,         # 4
            steps_slider,              # 5
            cfg_slider,                # 6
            seed_input,                # 7
            batch_slider,              # 8
            width_dropdown,            # 9
            height_dropdown,           # 10
            enable_cfg_norm,           # 11
            cfg_trunc_slider,          # 12
            prompt_style_dropdown,     # 13  ← BARU
            token_budget_dropdown,     # 14
            hf_token_input,            # 15
        ],
        outputs=[json_output],
        api_name="generate",
    )

# =====================================================================
# 🔧 SECTION 18: QUEUE + CORS
# =====================================================================
demo.queue(max_size=20, default_concurrency_limit=1)
print("✅ Queue enabled (max_size=20, concurrency=1)")
demo.app.add_middleware(CORSMiddleware, allow_origins=["*"], allow_credentials=False,
    allow_methods=["*"], allow_headers=["*"], expose_headers=["*"], max_age=86400)
print("✅ CORS: ALLOW ALL")

# =====================================================================
# 🧯 SECTION 19: EXCEPTION HANDLER
# =====================================================================
@demo.app.exception_handler(StarletteHTTPException)
async def api_http_exception_handler(request: Request, exc: StarletteHTTPException):
    sc = getattr(exc, "status_code", 500)
    detail = getattr(exc, "detail", "HTTP error")
    return JSONResponse(content={"success": False, "error": str(detail),
        "message": str(detail), "path": str(request.url.path)}, status_code=sc)

# =====================================================================
# 📡 SECTION 20: HEALTH & CONFIG
# =====================================================================
@demo.app.get("/health")
def health_check():
    gpu_mem = "N/A"
    try:
        gpu_mem = f"{torch.cuda.memory_allocated()/(1024**3):.1f}GB / {torch.cuda.memory_reserved()/(1024**3):.1f}GB"
    except Exception:
        pass
    with RAM_GALLERY_LOCK:
        gc_c = len(RAM_GALLERIES)
        gc_i = sum(len(g["images"]) for g in RAM_GALLERIES.values())
        gc_mb = round(sum(sum(i.get("bytes",0) for i in g["images"]) for g in RAM_GALLERIES.values())/(1024*1024),2)
    return {"status": "online", "model": DISPLAY_NAME, "architecture": MODEL_ARCHITECTURE,
        "text_encoder": TEXT_ENCODER_NAME, "vae": VAE_NAME,
        "gpu": "RTX 6000 Pro Blackwell (ZeroGPU)", "gpu_memory": gpu_mem, "zero_gpu": ZERO_GPU,
        "output_format": "Single JSON string",
        "ram_gallery": {"enabled": True, "galleries": gc_c, "total_images": gc_i, "used_mb": gc_mb},
        "timestamp": time.strftime("%Y-%m-%d %H:%M:%S")}

@demo.app.get("/api/config")
def get_config_api():
    return {"model": DISPLAY_NAME, "architecture": MODEL_ARCHITECTURE,
        "text_encoder": TEXT_ENCODER_NAME, "vae": VAE_NAME,
        "prediction_type": "flow_matching", "default_batch": DEFAULT_BATCH,
        "samplers": {n: {"schedules": i["schedules"], "tag": i["tag"]} for n, i in SAMPLER_REGISTRY.items()},
        "schedules": {n: {"description": i["description"]} for n, i in SCHEDULE_REGISTRY.items()},
        "prompt_styles": list(POSITIVE_PREFIXES.keys()),
        "gpu_duration_tiers": GPU_DURATION_TIERS,
        "negative_prefix": NEGATIVE_PREFIX.strip(),
        "official_recipe": {"sampler": "res_multistep/euler_ancestral (ComfyUI) → Flow Heun (diffusers)",
                            "scheduler": "linear_quadratic (ComfyUI) → Dynamic Shift (diffusers)",
                            "steps": ">=30", "guidance_scale": "4.0-5.5",
                            "cfg_trunc_ratio": 1.0, "cfg_normalization": True,
                            "resolution": "1024x1024, 768x1532, 968x1322, or >=1024"},
        "limits": {"max_steps": MAX_STEPS, "max_batch_size": MAX_BATCH_SIZE,
                   "max_width": MAX_WIDTH, "max_height": MAX_HEIGHT,
                   "min_width": MIN_WIDTH, "min_height": MIN_HEIGHT,
                   "max_cfg": MAX_CFG, "min_cfg": MIN_CFG},
        "ram_gallery": {"enabled": True, "max_galleries": RAM_GALLERY_MAX_GALLERIES,
                        "max_images_per_gallery": RAM_GALLERY_MAX_IMAGES_PER_GALLERY,
                        "max_total_images": RAM_GALLERY_MAX_TOTAL_IMAGES},
        "output_format": "Single JSON string"}

# =====================================================================
# 🏞️ SECTION 21: GALLERY REST API
# =====================================================================
async def _read_json(request: Request):
    try:
        return await request.json()
    except Exception:
        return {}

def _ok(payload, status_code=200):
    base = {"success": True, "error": None, "message": "OK"}
    base.update(payload)
    return JSONResponse(content=base, status_code=status_code)

def _err(message, status_code=400, extra=None):
    payload = {"success": False, "error": message, "message": message}
    if extra:
        payload.update(extra)
    return JSONResponse(content=payload, status_code=status_code)

@demo.app.get("/api/gallery/list")
async def api_gallery_list():
    try:
        g = ram_gallery_list()
        return _ok({"galleries": g, "total": len(g)})
    except Exception as e:
        return _err(str(e), 500)

@demo.app.get("/api/gallery/stats")
async def api_gallery_stats():
    try:
        return _ok(ram_gallery_stats())
    except Exception as e:
        return _err(str(e), 500)

@demo.app.post("/api/gallery/create")
async def api_gallery_create(request: Request):
    try:
        body = await _read_json(request)
        name = _sanitize_gallery_name(body.get("name", ""))
        desc = str(body.get("description", "")).strip()
        if not name:
            return _err("Gallery name is required", 400)
        ok, msg, gid = ram_gallery_create(name, desc)
        if not ok:
            return _err(msg, 400, {"name": name})
        return _ok({"message": msg, "gallery_id": gid, "name": name})
    except Exception as e:
        return _err(str(e), 500)

@demo.app.post("/api/gallery/clear-all")
async def api_gallery_clear_all():
    try:
        with RAM_GALLERY_LOCK:
            c = len(RAM_GALLERIES); t = sum(len(g["images"]) for g in RAM_GALLERIES.values())
            RAM_GALLERIES.clear()
        return _ok({"message": f"Cleared {c} galleries, {t} images"})
    except Exception as e:
        return _err(str(e), 500)

@demo.app.post("/api/gallery/{gallery_name}/add")
async def api_gallery_add(request: Request, gallery_name: str):
    try:
        gallery_name = _sanitize_gallery_name(unquote(gallery_name))
        body = await _read_json(request)
        images = body.get("images", [])
        if not images or not isinstance(images, list):
            return _err("No images provided", 400)
        ok, msg, added = ram_gallery_add_images(gallery_name, images)
        if not ok:
            return _err(msg, 400, {"gallery_name": gallery_name})
        summary, _ = ram_gallery_get_images(gallery_name)
        return _ok({"message": msg, "added": added, "gallery_name": gallery_name,
            "gallery_image_count": summary.get("image_count", 0) if summary else 0,
            "gallery_total_mb": summary.get("total_mb", 0) if summary else 0})
    except Exception as e:
        return _err(str(e), 500)

@demo.app.get("/api/gallery/{gallery_name}/images")
async def api_gallery_images(gallery_name: str):
    try:
        gallery_name = _sanitize_gallery_name(unquote(gallery_name))
        result, error = ram_gallery_get_images(gallery_name)
        if error:
            return _err(error, 404, {"gallery_name": gallery_name})
        images_meta = []
        for img in result.get("images", []):
            meta = {k: v for k, v in img.items() if k != "data"}
            meta["has_data"] = True
            images_meta.append(meta)
        return _ok({"name": result.get("name"), "description": result.get("description"),
            "image_count": result.get("image_count"), "total_mb": result.get("total_mb"),
            "created_at": result.get("created_at"), "last_modified": result.get("last_modified"),
            "images": images_meta})
    except Exception as e:
        return _err(str(e), 500)

@demo.app.get("/api/gallery/{gallery_name}/image/{image_id}")
async def api_gallery_get_single_image(gallery_name: str, image_id: str):
    try:
        gallery_name = _sanitize_gallery_name(unquote(gallery_name))
        image_id = unquote(image_id).strip()
        with RAM_GALLERY_LOCK:
            if gallery_name not in RAM_GALLERIES:
                return _err(f"Gallery '{gallery_name}' not found", 404)
            for img in RAM_GALLERIES[gallery_name]["images"]:
                if img["id"] == image_id:
                    return _ok({"image": img})
        return _err(f"Image '{image_id}' not found", 404)
    except Exception as e:
        return _err(str(e), 500)

@demo.app.get("/api/gallery/{gallery_name}/download")
async def api_gallery_download_all(gallery_name: str):
    try:
        gallery_name = _sanitize_gallery_name(unquote(gallery_name))
        result, error = ram_gallery_get_images(gallery_name)
        if error:
            return _err(error, 404, {"gallery_name": gallery_name})
        return _ok({"name": result.get("name"), "image_count": result.get("image_count"),
                    "images": result.get("images", [])})
    except Exception as e:
        return _err(str(e), 500)

@demo.app.delete("/api/gallery/{gallery_name}")
async def api_gallery_delete(gallery_name: str):
    try:
        gallery_name = _sanitize_gallery_name(unquote(gallery_name))
        ok, msg = ram_gallery_delete(gallery_name)
        if not ok:
            return _err(msg, 404, {"gallery_name": gallery_name})
        return _ok({"message": msg, "gallery_name": gallery_name})
    except Exception as e:
        return _err(str(e), 500)

@demo.app.delete("/api/gallery/{gallery_name}/image/{image_id}")
async def api_gallery_delete_image(gallery_name: str, image_id: str):
    try:
        gallery_name = _sanitize_gallery_name(unquote(gallery_name))
        image_id = unquote(image_id).strip()
        ok, msg = ram_gallery_delete_image(gallery_name, image_id)
        if not ok:
            return _err(msg, 404, {"gallery_name": gallery_name, "image_id": image_id})
        return _ok({"message": msg, "gallery_name": gallery_name, "image_id": image_id})
    except Exception as e:
        return _err(str(e), 500)

@demo.app.post("/api/gallery/{gallery_name}/rename")
async def api_gallery_rename(request: Request, gallery_name: str):
    try:
        gallery_name = _sanitize_gallery_name(unquote(gallery_name))
        body = await _read_json(request)
        new_name = _sanitize_gallery_name(body.get("new_name", ""))
        if not new_name:
            return _err("New name is required", 400)
        ok, msg = ram_gallery_rename(gallery_name, new_name)
        if not ok:
            return _err(msg, 400, {"gallery_name": gallery_name})
        return _ok({"message": msg, "old_name": gallery_name, "new_name": new_name})
    except Exception as e:
        return _err(str(e), 500)

# =====================================================================
# 🚀 SECTION 22: LAUNCH
# =====================================================================
print(" ")
print("=" * 70)
print("🚀 NETA-LUMINA v1.0.4 — OFFICIAL RECIPE")
print(f"   📦 {DISPLAY_NAME} | bfloat16 | SDPA")
print(f"   🏗️ {MODEL_ARCHITECTURE}")
print(f"   📝 Text Encoder: {TEXT_ENCODER_NAME} | VAE: {VAE_NAME}")
print(f"   📋 {len(SAMPLER_REGISTRY)} samplers × {len(SCHEDULE_REGISTRY)} schedules = {TOTAL_COMBOS}")
print(f"   🎯 ZeroGPU: {'✅' if ZERO_GPU else '❌'}")
print(f"   📝 Prefixes: official Neta (4 styles) + low-quality negative")
print(f"   🎯 Defaults: CFG 5.5 | trunc 1.0 | Dynamic Shift | steps 30")
print(f"   🔢 Params: 15 (prompt_style @ pos 13)")
print(f"   🔧 low_cpu_mem_usage: ✅ (OOM fix)")
print("=" * 70)

start_keep_alive()
demo.launch(server_name="0.0.0.0", server_port=7860, share=False, show_error=True, ssr_mode=False)