Update app.py

app.py

@@ -1,44 +1,51 @@
-import spaces
-from dataclasses import dataclass
-import json
-import logging
 import os
-import random
-import re
 import sys
+import re
+import json
+import random
+import logging
 import warnings
+from dataclasses import dataclass
 
-from PIL import Image
-from diffusers import AutoencoderKL, FlowMatchEulerDiscreteScheduler
 import gradio as gr
 import torch
-from transformers import AutoModelForCausalLM, AutoTokenizer
+from PIL import Image, ImageDraw, ImageFont
 
-from prompt_check import is_unsafe_prompt
+import spaces
+from diffusers import AutoencoderKL, FlowMatchEulerDiscreteScheduler
+from transformers import AutoModelForCausalLM, AutoTokenizer
 
-sys.path.append(os.path.dirname(os.path.abspath(__file__)))
+# ------------------------- 可选依赖：Prompt Enhancer 模板 -------------------------
+# 你的原工程里如果有 pe.py，会自动使用；没有也不会报错（enhance 默认关闭）
+try:
+    sys.path.append(os.path.dirname(os.path.abspath(__file__)))
+    from pe import prompt_template  # type: ignore
+except Exception:
+    prompt_template = (
+        "You are a helpful prompt engineer. Expand the user prompt into a richer, detailed prompt. "
+        "Return JSON with key revised_prompt."
+    )
 
+# ------------------------- Z-Image 相关（依赖你环境中 diffusers 的实现） -------------------------
 from diffusers import ZImagePipeline
 from diffusers.models.transformers.transformer_z_image import ZImageTransformer2DModel
 
-from pe import prompt_template
-
 # ==================== Environment Variables ==================================
 MODEL_PATH = os.environ.get("MODEL_PATH", "Tongyi-MAI/Z-Image-Turbo")
 ENABLE_COMPILE = os.environ.get("ENABLE_COMPILE", "true").lower() == "true"
 ENABLE_WARMUP = os.environ.get("ENABLE_WARMUP", "true").lower() == "true"
 ATTENTION_BACKEND = os.environ.get("ATTENTION_BACKEND", "flash_3")
-UNSAFE_MAX_NEW_TOKEN = int(os.environ.get("UNSAFE_MAX_NEW_TOKEN", "10"))
 DASHSCOPE_API_KEY = os.environ.get("DASHSCOPE_API_KEY")
 HF_TOKEN = os.environ.get("HF_TOKEN")
-UNSAFE_PROMPT_CHECK = os.environ.get("UNSAFE_PROMPT_CHECK")
 # =============================================================================
 
-
 os.environ["TOKENIZERS_PARALLELISM"] = "false"
 warnings.filterwarnings("ignore")
 logging.getLogger("transformers").setLevel(logging.ERROR)
 
+DEVICE = "cuda" if torch.cuda.is_available() else "cpu"
+DTYPE = torch.bfloat16 if DEVICE == "cuda" else torch.float32
+
 RES_CHOICES = {
     "1024": [
         "1024x1024 ( 1:1 )",
@@ -82,79 +89,100 @@ RES_CHOICES = {
 }
 
 RESOLUTION_SET = []
-for resolutions in RES_CHOICES.values():
-    RESOLUTION_SET.extend(resolutions)
+for _k, v in RES_CHOICES.items():
+    RESOLUTION_SET.extend(v)
 
 EXAMPLE_PROMPTS = [
     ["一位男士和他的贵宾犬穿着配套的服装参加狗狗秀，室内灯光，背景中有观众。"],
-    [
-        "极具氛围感的暗调人像，一位优雅的中国美女在黑暗的房间里。一束强光通过遮光板，在她的脸上投射出一个清晰的闪电形状的光影，正好照亮一只眼睛。高对比度，明暗交界清晰，神秘感，莱卡相机色调。"
-    ],
-    [
-        "一张中景手机自拍照片拍摄了一位留着长黑发的年轻东亚女子在灯光明亮的电梯内对着镜子自拍。她穿着一件带有白色花朵图案的黑色露肩短上衣和深色牛仔裤。她的头微微倾斜，嘴唇嘟起做亲吻状，非常可爱俏皮。她右手拿着一部深灰色智能手机，遮住了部分脸，后置摄像头镜头对着镜子"
-    ],
-    [
-        "Young Chinese woman in red Hanfu, intricate embroidery. Impeccable makeup, red floral forehead pattern. Elaborate high bun, golden phoenix headdress, red flowers, beads. Holds round folding fan with lady, trees, bird. Neon lightning-bolt lamp (⚡️), bright yellow glow, above extended left palm. Soft-lit outdoor night background, silhouetted tiered pagoda (西安大雁塔), blurred colorful distant lights."
-    ],
-    [
-        '''A vertical digital illustration depicting a serene and majestic Chinese landscape, rendered in a style reminiscent of traditional Shanshui painting but with a modern, clean aesthetic. The scene is dominated by towering, steep cliffs in various shades of blue and teal, which frame a central valley. In the distance, layers of mountains fade into a light blue and white mist, creating a strong sense of atmospheric perspective and depth. A calm, turquoise river flows through the center of the composition, with a small, traditional Chinese boat, possibly a sampan, navigating its waters. The boat has a bright yellow canopy and a red hull, and it leaves a gentle wake behind it. It carries several indistinct figures of people. Sparse vegetation, including green trees and some bare-branched trees, clings to the rocky ledges and peaks. The overall lighting is soft and diffused, casting a tranquil glow over the entire scene. Centered in the image is overlaid text. At the top of the text block is a small, red, circular seal-like logo containing stylized characters. Below it, in a smaller, black, sans-serif font, are the words 'Zao-Xiang * East Beauty & West Fashion * Z-Image'. Directly beneath this, in a larger, elegant black serif font, is the word 'SHOW & SHARE CREATIVITY WITH THE WORLD'. Among them, there are "SHOW & SHARE", "CREATIVITY", and "WITH THE WORLD"'''
-    ],
-    [
-        """一张虚构的英语电影《回忆之味》（The Taste of Memory）的电影海报。场景设置在一个质朴的19世纪风格厨房里。画面中央，一位红棕色头发、留着小胡子的中年男子（演员阿瑟·彭哈利根饰）站在一张木桌后，他身穿白色衬衫、黑色马甲和米色围裙，正看着一位女士，手中拿着一大块生红肉，下方是一个木制切菜板。在他的右边，一位梳着高髻的黑发女子（演员埃莉诺·万斯饰）倚靠在桌子上，温柔地对他微笑。她穿着浅色衬衫和一条上白下蓝的长裙。桌上除了放有切碎的葱和卷心菜丝的切菜板外，还有一个白色陶瓷盘、新鲜香草，左侧一个木箱上放着一串深色葡萄。背景是一面粗糙的灰白色抹灰墙，墙上挂着一幅风景画。最右边的一个台面上放着一盏复古油灯。海报上有大量的文字信息。左上角是白色的无衬线字体"ARTISAN FILMS PRESENTS"，其下方是"ELEANOR VANCE"和"ACADEMY AWARD® WINNER"。右上角写着"ARTHUR PENHALIGON"和"GOLDEN GLOBE® AWARD WINNER"。顶部中央是圣丹斯电影节的桂冠标志，下方写着"SUNDANCE FILM FESTIVAL GRAND JURY PRIZE 2024"。主标题"THE TASTE OF MEMORY"以白色的大号衬线字体醒目地显示在下半部分。标题下方注明了"A FILM BY Tongyi Interaction Lab"。底部区域用白色小字列出了完整的演职员名单，包括"SCREENPLAY BY ANNA REID"、"CULINARY DIRECTION BY JAMES CARTER"以及Artisan Films、Riverstone Pictures和Heritage Media等众多出品公司标志。整体风格是写实主义，采用温暖柔和的灯光方案，营造出一种亲密的氛围。色调以棕色、米色和柔和的绿色等大地色系为主。两位演员的身体都在腰部被截断。"""
-    ],
-    [
-        """一张方形构图的特写照片，主体是一片巨大的、鲜绿色的植物叶片，并叠加了文字，使其具有海报或杂志封面的外观。主要拍摄对象是一片厚实、有蜡质感的叶子，从左下角到右上角呈对角线弯曲穿过画面。其表面反光性很强，捕捉到一个明亮的直射光源，形成了一道突出的高光，亮面下显露出平行的精细叶脉。背景由其他深绿色的叶子组成，这些叶子轻微失焦，营造出浅景深效果，突出了前景的主叶片。整体风格是写实摄影，明亮的叶片与黑暗的阴影背景之间形成高对比度。图像上有多处渲染文字。左上角是白色的衬线字体文字"PIXEL-PEEPERS GUILD Presents"。右上角同样是白色衬线字体的文字"[Instant Noodle] 泡面调料包"。左侧垂直排列着标题"Render Distance: Max"，为白色衬线字体。左下角是五个硕大的白色宋体汉字"显卡在...燃烧"。右下角是较小的白色衬线字体文字"Leica Glow™ Unobtanium X-1"，其正上方是用白色宋体字书写的名字"蔡几"。识别出的核心实体包括品牌像素偷窥者协会、其产品线泡面调料包、相机型号买不到™ X-1以及摄影师名字造相。"""
-    ],
+    ["极具氛围感的暗调人像，一位优雅的中国美女在黑暗的房间里。一束强光通过遮光板，在她的脸上投射出一个清晰的闪电形状的光影，正好照亮一只眼睛。高对比度，明暗交界清晰，神秘感，莱卡相机色调。"],
 ]
 
+# ------------------------- HF token 兼容参数 -------------------------
+def _hf_token_kwargs(token: str | None):
+    """
+    transformers / diffusers 的 from_pretrained 近年来从 use_auth_token 迁移到 token。
+    这里做一个兼容：优先传 token，不支持则回退 use_auth_token。
+    """
+    if not token:
+        return {}
+    return {"token": token, "use_auth_token": token}
 
-def get_resolution(resolution):
+def get_resolution(resolution: str):
     match = re.search(r"(\d+)\s*[×x]\s*(\d+)", resolution)
     if match:
         return int(match.group(1)), int(match.group(2))
     return 1024, 1024
 
+def _make_blocked_image(width=1024, height=1024, text="Blocked by Safety Checker"):
+    img = Image.new("RGB", (width, height), (20, 20, 20))
+    draw = ImageDraw.Draw(img)
+    try:
+        font = ImageFont.load_default()
+    except Exception:
+        font = None
+    draw.rectangle([0, 0, width, 90], fill=(160, 0, 0))
+    draw.text((20, 30), text, fill=(255, 255, 255), font=font)
+    return img
+
+def _load_nsfw_placeholder(width=1024, height=1024):
+    """
+    命中 NSFW 时优先加载工作目录的 nsfw.png；
+    不存在就生成一张占位图，避免文件缺失导致再次报错。
+    """
+    if os.path.exists("nsfw.png"):
+        try:
+            return Image.open("nsfw.png").convert("RGB")
+        except Exception:
+            pass
+    return _make_blocked_image(width, height, "NSFW blocked")
 
-def load_models(model_path, enable_compile=False, attention_backend="native"):
-    print(f"Loading models from {model_path}...")
+def load_models(model_path: str, enable_compile=False, attention_backend="native"):
+    print(f"[Init] Loading models from: {model_path}")
+    print(f"[Init] DEVICE={DEVICE}, DTYPE={DTYPE}, ENABLE_COMPILE={enable_compile}, ATTENTION_BACKEND={attention_backend}")
 
-    use_auth_token = HF_TOKEN if HF_TOKEN else True
+    # 远端 repo-id（不存在的本地路径） vs 本地目录
+    is_local_dir = os.path.exists(model_path)
+    token_kwargs = _hf_token_kwargs(HF_TOKEN) if not is_local_dir else {}
 
-    if not os.path.exists(model_path):
+    # 1) VAE
+    if not is_local_dir:
         vae = AutoencoderKL.from_pretrained(
-            f"{model_path}",
+            model_path,
             subfolder="vae",
-            torch_dtype=torch.bfloat16,
-            device_map="cuda",
-            use_auth_token=use_auth_token,
+            torch_dtype=DTYPE if DEVICE == "cuda" else torch.float32,
+            **token_kwargs,
+        )
+    else:
+        vae = AutoencoderKL.from_pretrained(
+            os.path.join(model_path, "vae"),
+            torch_dtype=DTYPE if DEVICE == "cuda" else torch.float32,
         )
 
+    # 2) Text Encoder + Tokenizer
+    if not is_local_dir:
         text_encoder = AutoModelForCausalLM.from_pretrained(
-            f"{model_path}",
+            model_path,
             subfolder="text_encoder",
-            torch_dtype=torch.bfloat16,
-            device_map="cuda",
-            use_auth_token=use_auth_token,
+            torch_dtype=DTYPE if DEVICE == "cuda" else torch.float32,
+            **token_kwargs,
         ).eval()
-
-        tokenizer = AutoTokenizer.from_pretrained(f"{model_path}", subfolder="tokenizer", use_auth_token=use_auth_token)
-    else:
-        vae = AutoencoderKL.from_pretrained(
-            os.path.join(model_path, "vae"), torch_dtype=torch.bfloat16, device_map="cuda"
+        tokenizer = AutoTokenizer.from_pretrained(
+            model_path,
+            subfolder="tokenizer",
+            **token_kwargs,
         )
-
+    else:
         text_encoder = AutoModelForCausalLM.from_pretrained(
             os.path.join(model_path, "text_encoder"),
-            torch_dtype=torch.bfloat16,
-            device_map="cuda",
+            torch_dtype=DTYPE if DEVICE == "cuda" else torch.float32,
         ).eval()
-
         tokenizer = AutoTokenizer.from_pretrained(os.path.join(model_path, "tokenizer"))
 
     tokenizer.padding_side = "left"
 
-    if enable_compile:
-        print("Enabling torch.compile optimizations...")
+    # compile 优化（仅 CUDA 才建议打开）
+    if enable_compile and DEVICE == "cuda":
+        print("[Init] Enabling torch.compile optimizations...")
         torch._inductor.config.conv_1x1_as_mm = True
         torch._inductor.config.coordinate_descent_tuning = True
         torch._inductor.config.epilogue_fusion = False
@@ -165,42 +193,68 @@ def load_models(model_path, enable_compile=False, attention_backend="native"):
 
     pipe = ZImagePipeline(scheduler=None, vae=vae, text_encoder=text_encoder, tokenizer=tokenizer, transformer=None)
 
-    if enable_compile:
-        pipe.vae.disable_tiling()
-
-    if not os.path.exists(model_path):
+    # 3) Transformer
+    if not is_local_dir:
         transformer = ZImageTransformer2DModel.from_pretrained(
-            f"{model_path}", subfolder="transformer", use_auth_token=use_auth_token
-        ).to("cuda", torch.bfloat16)
-    else:
-        transformer = ZImageTransformer2DModel.from_pretrained(os.path.join(model_path, "transformer")).to(
-            "cuda", torch.bfloat16
+            model_path,
+            subfolder="transformer",
+            **token_kwargs,
         )
+    else:
+        transformer = ZImageTransformer2DModel.from_pretrained(os.path.join(model_path, "transformer"))
 
+    transformer = transformer.to(DEVICE, DTYPE)
     pipe.transformer = transformer
-    pipe.transformer.set_attention_backend(attention_backend)
 
-    if enable_compile:
-        print("Compiling transformer...")
-        pipe.transformer = torch.compile(pipe.transformer, mode="max-autotune-no-cudagraphs", fullgraph=False)
+    # attention backend 可能在不同环境不支持，做容错
+    try:
+        pipe.transformer.set_attention_backend(attention_backend)
+    except Exception as e:
+        print(f"[Init] set_attention_backend('{attention_backend}') failed, fallback to 'native'. Error: {e}")
+        try:
+            pipe.transformer.set_attention_backend("native")
+        except Exception as e2:
+            print(f"[Init] fallback set_attention_backend('native') failed: {e2}")
 
-    pipe.to("cuda", torch.bfloat16)
+    if enable_compile and DEVICE == "cuda":
+        try:
+            print("[Init] Compiling transformer...")
+            pipe.transformer = torch.compile(pipe.transformer, mode="max-autotune-no-cudagraphs", fullgraph=False)
+        except Exception as e:
+            print(f"[Init] torch.compile failed, continue without compile. Error: {e}")
 
-    from diffusers.pipelines.stable_diffusion import StableDiffusionSafetyChecker
-    from transformers import CLIPImageProcessor
+    pipe = pipe.to(DEVICE, DTYPE)
 
-    safety_model_id = "CompVis/stable-diffusion-safety-checker"
-    safety_feature_extractor = CLIPImageProcessor.from_pretrained(safety_model_id)
-    safety_checker = StableDiffusionSafetyChecker.from_pretrained(safety_model_id, torch_dtype=torch.float16).to("cuda")
+    # 4) Safety Checker（用于生成后过滤）
+    try:
+        from diffusers.pipelines.stable_diffusion import StableDiffusionSafetyChecker
+        try:
+            from transformers import CLIPImageProcessor as _CLIPProcessor
+        except Exception:
+            # 老版本兼容
+            from transformers import CLIPFeatureExtractor as _CLIPProcessor  # type: ignore
+
+        safety_model_id = "CompVis/stable-diffusion-safety-checker"
+        safety_feature_extractor = _CLIPProcessor.from_pretrained(safety_model_id, **_hf_token_kwargs(HF_TOKEN))
+        safety_checker = StableDiffusionSafetyChecker.from_pretrained(
+            safety_model_id,
+            torch_dtype=torch.float16 if DEVICE == "cuda" else torch.float32,
+            **_hf_token_kwargs(HF_TOKEN),
+        ).to(DEVICE)
+
+        pipe.safety_feature_extractor = safety_feature_extractor
+        pipe.safety_checker = safety_checker
+        print("[Init] Safety checker loaded.")
+    except Exception as e:
+        print(f"[Init] Safety checker init failed. NSFW filtering will be skipped. Error: {e}")
+        pipe.safety_feature_extractor = None
+        pipe.safety_checker = None
 
-    pipe.safety_feature_extractor = safety_feature_extractor
-    pipe.safety_checker = safety_checker
     return pipe
 
-
 def generate_image(
     pipe,
-    prompt,
+    prompt: str,
     resolution="1024x1024",
     seed=42,
     guidance_scale=5.0,
@@ -211,48 +265,46 @@ def generate_image(
 ):
     width, height = get_resolution(resolution)
 
-    generator = torch.Generator("cuda").manual_seed(seed)
+    if DEVICE == "cuda":
+        generator = torch.Generator(device="cuda").manual_seed(int(seed))
+    else:
+        generator = torch.Generator().manual_seed(int(seed))
 
-    scheduler = FlowMatchEulerDiscreteScheduler(num_train_timesteps=1000, shift=shift)
+    scheduler = FlowMatchEulerDiscreteScheduler(num_train_timesteps=1000, shift=float(shift))
     pipe.scheduler = scheduler
 
-    image = pipe(
+    out = pipe(
         prompt=prompt,
-        height=height,
-        width=width,
-        guidance_scale=guidance_scale,
-        num_inference_steps=num_inference_steps,
+        height=int(height),
+        width=int(width),
+        guidance_scale=float(guidance_scale),
+        num_inference_steps=int(num_inference_steps),
         generator=generator,
-        max_sequence_length=max_sequence_length,
-    ).images[0]
-
+        max_sequence_length=int(max_sequence_length),
+    )
+    image = out.images[0]
     return image
 
-
 def warmup_model(pipe, resolutions):
-    print("Starting warmup phase...")
-
+    print("[Warmup] Starting warmup phase...")
     dummy_prompt = "warmup"
-
     for res_str in resolutions:
-        print(f"Warming up for resolution: {res_str}")
+        print(f"[Warmup] Resolution: {res_str}")
         try:
-            for i in range(3):
+            for i in range(2):
                 generate_image(
                     pipe,
                     prompt=dummy_prompt,
-                    resolution=res_str,
-                    num_inference_steps=9,
+                    resolution=res_str.split(" ")[0],
+                    num_inference_steps=6,
                     guidance_scale=0.0,
                     seed=42 + i,
                 )
         except Exception as e:
-            print(f"Warmup failed for {res_str}: {e}")
+            print(f"[Warmup] Failed for {res_str}: {e}")
+    print("[Warmup] Completed.")
 
-    print("Warmup completed.")
-
-
-# ==================== Prompt Expander ====================
+# ==================== Prompt Expander（保留但默认不启用） ====================
 @dataclass
 class PromptOutput:
     status: bool
@@ -261,7 +313,6 @@ class PromptOutput:
     system_prompt: str
     message: str
 
-
 class PromptExpander:
     def __init__(self, backend="api", **kwargs):
         self.backend = backend
@@ -269,7 +320,6 @@ class PromptExpander:
     def decide_system_prompt(self, template_name=None):
         return prompt_template
 
-
 class APIPromptExpander(PromptExpander):
     def __init__(self, api_config=None, **kwargs):
         super().__init__(backend="api", **kwargs)
@@ -284,15 +334,15 @@ class APIPromptExpander(PromptExpander):
             base_url = self.api_config.get("base_url", "https://dashscope.aliyuncs.com/compatible-mode/v1")
 
             if not api_key:
-                print("Warning: DASHSCOPE_API_KEY not found.")
+                print("[PE] Warning: DASHSCOPE_API_KEY not found. Prompt enhance unavailable.")
                 return None
 
             return OpenAI(api_key=api_key, base_url=base_url)
         except ImportError:
-            print("Please install openai: pip install openai")
+            print("[PE] Please install openai: pip install openai")
             return None
         except Exception as e:
-            print(f"Failed to initialize API client: {e}")
+            print(f"[PE] Failed to initialize API client: {e}")
             return None
 
     def __call__(self, prompt, system_prompt=None, seed=-1, **kwargs):
@@ -300,7 +350,7 @@ class APIPromptExpander(PromptExpander):
 
     def extend(self, prompt, system_prompt=None, seed=-1, **kwargs):
         if self.client is None:
-            return PromptOutput(False, "", seed, system_prompt, "API client not initialized")
+            return PromptOutput(False, "", seed, system_prompt or "", "API client not initialized")
 
         if system_prompt is None:
             system_prompt = self.decide_system_prompt()
@@ -317,65 +367,60 @@ class APIPromptExpander(PromptExpander):
                 temperature=0.7,
                 top_p=0.8,
             )
+            content = response.choices[0].message.content or ""
 
-            content = response.choices[0].message.content
+            # 尝试从 ```json 块中解析 revised_prompt
+            expanded_prompt = content
             json_start = content.find("```json")
             if json_start != -1:
                 json_end = content.find("```", json_start + 7)
-                try:
+                if json_end != -1:
                     json_str = content[json_start + 7 : json_end].strip()
-                    data = json.loads(json_str)
-                    expanded_prompt = data.get("revised_prompt", content)
-                except:
-                    expanded_prompt = content
-            else:
-                expanded_prompt = content
+                    try:
+                        data = json.loads(json_str)
+                        expanded_prompt = data.get("revised_prompt", content)
+                    except Exception:
+                        expanded_prompt = content
 
-            return PromptOutput(
-                status=True, prompt=expanded_prompt, seed=seed, system_prompt=system_prompt, message=content
-            )
+            return PromptOutput(True, expanded_prompt, seed, system_prompt, content)
         except Exception as e:
             return PromptOutput(False, "", seed, system_prompt, str(e))
 
-
 def create_prompt_expander(backend="api", **kwargs):
     if backend == "api":
         return APIPromptExpander(**kwargs)
     raise ValueError("Only 'api' backend is supported.")
 
-
 pipe = None
 prompt_expander = None
 
-
 def init_app():
     global pipe, prompt_expander
 
     try:
         pipe = load_models(MODEL_PATH, enable_compile=ENABLE_COMPILE, attention_backend=ATTENTION_BACKEND)
-        print(f"Model loaded. Compile: {ENABLE_COMPILE}, Backend: {ATTENTION_BACKEND}")
+        print("[Init] Model loaded.")
 
-        if ENABLE_WARMUP:
-            all_resolutions = []
+        if ENABLE_WARMUP and pipe is not None:
+            all_res = []
             for cat in RES_CHOICES.values():
-                all_resolutions.extend(cat)
-            warmup_model(pipe, all_resolutions)
+                all_res.extend(cat)
+            warmup_model(pipe, all_res)
 
     except Exception as e:
-        print(f"Error loading model: {e}")
+        print(f"[Init] Error loading model: {e}")
         pipe = None
 
     try:
         prompt_expander = create_prompt_expander(backend="api", api_config={"model": "qwen3-max-preview"})
-        print("Prompt expander initialized.")
+        print("[Init] Prompt expander ready (disabled by default).")
     except Exception as e:
-        print(f"Error initializing prompt expander: {e}")
+        print(f"[Init] Error initializing prompt expander: {e}")
         prompt_expander = None
 
-
-def prompt_enhance(prompt, enable_enhance):
+def prompt_enhance(prompt, enable_enhance: bool):
     if not enable_enhance or not prompt_expander:
-        return prompt, "Enhancement disabled or not available."
+        return prompt, "Enhancement disabled or unavailable."
 
     if not prompt.strip():
         return "", "Please enter a prompt."
@@ -384,11 +429,35 @@ def prompt_enhance(prompt, enable_enhance):
         result = prompt_expander(prompt)
         if result.status:
             return result.prompt, result.message
-        else:
-            return prompt, f"Enhancement failed: {result.message}"
+        return prompt, f"Enhancement failed: {result.message}"
     except Exception as e:
         return prompt, f"Error: {str(e)}"
 
+def try_enable_aoti(pipe):
+    """
+    AoTI（ZeroGPU 加速）可用则启用；不可用则跳过，不影响主流程。
+    """
+    if pipe is None:
+        return
+    try:
+        # 优先按你原代码的结构尝试：pipe.transformer.layers
+        if hasattr(pipe, "transformer") and pipe.transformer is not None:
+            target = None
+            if hasattr(pipe.transformer, "layers"):
+                target = pipe.transformer.layers
+                if hasattr(target, "_repeated_blocks"):
+                    target._repeated_blocks = ["ZImageTransformerBlock"]
+            else:
+                # 兜底：直接对 transformer 设置
+                target = pipe.transformer
+                if hasattr(target, "_repeated_blocks"):
+                    target._repeated_blocks = ["ZImageTransformerBlock"]
+
+            if target is not None:
+                spaces.aoti_blocks_load(target, "zerogpu-aoti/Z-Image", variant="fa3")
+                print("[Init] AoTI blocks loaded.")
+    except Exception as e:
+        print(f"[Init] AoTI not enabled (safe to ignore). Error: {e}")
 
 @spaces.GPU
 def generate(
@@ -399,175 +468,42 @@ def generate(
     shift=3.0,
     random_seed=True,
     gallery_images=None,
-    enhance=False,
+    enhance=False,  # 默认不启用
     progress=gr.Progress(track_tqdm=True),
 ):
-    """
-    Generate an image using the Z-Image model based on the provided prompt and settings.
-    This function is triggered when the user clicks the "Generate" button. It processes
-    the input prompt (optionally enhancing it), configures generation parameters, and
-    produces an image using the Z-Image diffusion transformer pipeline.
-    Args:
-        prompt (str): Text prompt describing the desired image content
-        resolution (str): Output resolution in format "WIDTHxHEIGHT ( RATIO )" (e.g., "1024x1024 ( 1:1 )")
-        seed (int): Seed for reproducible generation
-        steps (int): Number of inference steps for the diffusion process
-        shift (float): Time shift parameter for the flow matching scheduler
-        random_seed (bool): Whether to generate a new random seed, if True will ignore the seed input
-        gallery_images (list): List of previously generated images to append to (only needed for the Gradio UI)
-        enhance (bool): This was Whether to enhance the prompt (DISABLED! Do not use)
-        progress (gr.Progress): Gradio progress tracker for displaying generation progress (only needed for the Gradio UI)
-    Returns:
-        tuple: (gallery_images, seed_str, seed_int)
-            - gallery_images: Updated list of generated images including the new image
-            - seed_str: String representation of the seed used for generation
-            - seed_int: Integer representation of the seed used for generation
-    """
-
     if random_seed:
         new_seed = random.randint(1, 1000000)
     else:
-        new_seed = seed if seed != -1 else random.randint(1, 1000000)
-
-    class UnsafeContentError(Exception):
-        pass
-
-    try:
-        if pipe is None:
-            raise gr.Error("Model not loaded.")
-
-        has_unsafe_concept = is_unsafe_prompt(
-            pipe.text_encoder,
-            pipe.tokenizer,
-            system_prompt=UNSAFE_PROMPT_CHECK,
-            user_prompt=prompt,
-            max_new_token=UNSAFE_MAX_NEW_TOKEN,
-        )
-        if has_unsafe_concept:
-            raise UnsafeContentError("Input unsafe")
-
-        final_prompt = prompt
-
-        if enhance:
-            final_prompt, _ = prompt_enhance(prompt, True)
-            print(f"Enhanced prompt: {final_prompt}")
-
-        try:
-            resolution_str = resolution.split(" ")[0]
-        except:
-            resolution_str = "1024x1024"
-
-        image = generate_image(
-            pipe=pipe,
-            prompt=final_prompt,
-            resolution=resolution_str,
-            seed=new_seed,
-            guidance_scale=0.0,
-            num_inference_steps=int(steps + 1),
-            shift=shift,
-        )
-
-        safety_checker_input = pipe.safety_feature_extractor([image], return_tensors="pt").pixel_values.cuda()
-        _, has_nsfw_concept = pipe.safety_checker(images=[torch.zeros(1)], clip_input=safety_checker_input)
-        has_nsfw_concept = has_nsfw_concept[0]
-        if has_nsfw_concept:
-            raise UnsafeContentError("input unsafe")
+        new_seed = int(seed) if int(seed) != -1 else random.randint(1, 1000000)
 
-    except UnsafeContentError:
-        image = Image.open("nsfw.png")
+    if pipe is None:
+        raise gr.Error("Model not loaded. Please check logs.")
 
-    if gallery_images is None:
-        gallery_images = []
-    # gallery_images.append(image)
-    gallery_images = [image] + gallery_images # latest output to be at the top of the list
+    final_prompt = prompt or ""
+    if enhance:
+        # 你原注释说 DISABLED，这里仍保留能力但默认关闭
+        final_prompt, _msg = prompt_enhance(final_prompt, True)
+        print(f"[PE] Enhanced prompt: {final_prompt}")
 
-    return gallery_images, str(new_seed), int(new_seed)
-
-
-init_app()
-
-# ==================== AoTI (Ahead of Time Inductor compilation) ====================
-
-pipe.transformer.layers._repeated_blocks = ["ZImageTransformerBlock"]
-spaces.aoti_blocks_load(pipe.transformer.layers, "zerogpu-aoti/Z-Image", variant="fa3")
-
-with gr.Blocks(title="Z-Image Demo") as demo:
-    gr.Markdown(
-        """<div align="center">
-# Z-Image Generation Demo
-[![GitHub](https://img.shields.io/badge/GitHub-Z--Image-181717?logo=github&logoColor=white)](https://github.com/Tongyi-MAI/Z-Image)
-*An Efficient Image Generation Foundation Model with Single-Stream Diffusion Transformer*
-</div>"""
-    )
-
-    with gr.Row():
-        with gr.Column(scale=1):
-            prompt_input = gr.Textbox(label="Prompt", lines=3, placeholder="Enter your prompt here...")
-            # PE components (Temporarily disabled)
-            # with gr.Row():
-            #     enable_enhance = gr.Checkbox(label="Enhance Prompt (DashScope)", value=False)
-            #     enhance_btn = gr.Button("Enhance Only")
-
-            with gr.Row():
-                choices = [int(k) for k in RES_CHOICES.keys()]
-                res_cat = gr.Dropdown(value=1024, choices=choices, label="Resolution Category")
-
-                initial_res_choices = RES_CHOICES["1024"]
-                resolution = gr.Dropdown(
-                    value=initial_res_choices[0], choices=RESOLUTION_SET, label="Width x Height (Ratio)"
-                )
-
-            with gr.Row():
-                seed = gr.Number(label="Seed", value=42, precision=0)
-                random_seed = gr.Checkbox(label="Random Seed", value=True)
-
-            with gr.Row():
-                steps = gr.Slider(label="Steps", minimum=1, maximum=100, value=8, step=1, interactive=False)
-                shift = gr.Slider(label="Time Shift", minimum=1.0, maximum=10.0, value=3.0, step=0.1)
-
-            generate_btn = gr.Button("Generate", variant="primary")
-
-            # Example prompts
-            gr.Markdown("### 📝 Example Prompts")
-            gr.Examples(examples=EXAMPLE_PROMPTS, inputs=prompt_input, label=None)
-
-        with gr.Column(scale=1):
-            output_gallery = gr.Gallery(
-                label="Generated Images",
-                columns=2,
-                rows=2,
-                height=600,
-                object_fit="contain",
-                format="png",
-                interactive=False,
-            )
-            used_seed = gr.Textbox(label="Seed Used", interactive=False)
-
-    def update_res_choices(_res_cat):
-        if str(_res_cat) in RES_CHOICES:
-            res_choices = RES_CHOICES[str(_res_cat)]
-        else:
-            res_choices = RES_CHOICES["1024"]
-        return gr.update(value=res_choices[0], choices=res_choices)
-
-    res_cat.change(update_res_choices, inputs=res_cat, outputs=resolution, api_visibility="private")
-
-    # PE enhancement button (Temporarily disabled)
-    # enhance_btn.click(
-    #     prompt_enhance,
-    #     inputs=[prompt_input, enable_enhance],
-    #     outputs=[prompt_input, final_prompt_output]
-    # )
-
-    generate_btn.click(
-        generate,
-        inputs=[prompt_input, resolution, seed, steps, shift, random_seed, output_gallery],
-        outputs=[output_gallery, used_seed, seed],
-        api_visibility="public",
+    # 解析 "1024x1024 ( 1:1 )" -> "1024x1024"
+    try:
+        resolution_str = str(resolution).split(" ")[0]
+    except Exception:
+        resolution_str = "1024x1024"
+
+    width, height = get_resolution(resolution_str)
+
+    # 生成
+    image = generate_image(
+        pipe=pipe,
+        prompt=final_prompt,
+        resolution=resolution_str,
+        seed=new_seed,
+        guidance_scale=0.0,
+        num_inference_steps=int(steps) + 1,
+        shift=float(shift),
     )
 
-css = """
-.fillable{max-width: 1230px !important}
-"""
-if __name__ == "__main__":
-    demo.launch(css=css, mcp_server=True)
+    # 生成后 NSFW 安全检查（已去掉 prompt_check）
+    try:
+        if getattr(pipe, "safety_feature_extractor", None) is not None and getattr(pipe, "safety_checker", None) is not_