import asyncio import gc import httpx import logging import os import psutil import random import threading import time import traceback import weakref from dataclasses import dataclass, field from typing import Any import subprocess subprocess.run(["pip", "install", "gradio==5.34.0"]) import gradio as gr # 常量定义区域 # HTTP 状态码 HTTP_STATUS_OK = 200 HTTP_STATUS_CENSORED = 451 # 系统配置 MAX_SEED = 2147483647 # (2**31 - 1) MAX_IMAGE_SIZE = 2048 MIN_IMAGE_SIZE = 256 MEMORY_CHECK_INTERVAL = 300 # 5分钟检查一次内存 MEMORY_THRESHOLD_PERCENT = 80 # 内存使用百分比阈值 DEBUG_MODE = os.environ.get("DEBUG_MODE", "true").lower() == "true" # API配置 API_TOKEN = os.environ.get("API_TOKEN", "") if not API_TOKEN: raise ValueError("环境变量中未设置API_TOKEN") # 飞书通知配置 CHAT_WEBHOOK_URL = os.environ.get("CHAT_WEBHOOK_URL", "") # 外部链接 DISCORD_LINK = os.environ.get("DISCORD_LINK", "https://discord.com/invite/AtRtbe9W8w") APP_INDEX_LINK = os.environ.get("APP_INDEX_LINK", "https://huggingface.co/neta-art") APP_INDEX_ICON = "https://cdn-avatars.huggingface.co/v1/production/uploads/62be651a1e22ec8427aa7096/Pf1jPtZZT7PR5ettxM6_P.png" PROMPT_BOOK_LINK = "https://nieta-art.feishu.cn/wiki/RVBgwvzBqiCvQ7kOMm1cM6NdnNc?from=from_copylink" PROMPT_BOOK_LINK_CN = "https://nieta-art.feishu.cn/wiki/RZAawlH2ci74qckRLRPc9tOynrb?from=from_copylink" # 模型配置 MODEL_CONFIGS = { "beta-0624.pth": "beta-0624.pth", "beta-a5nA_e109.pth": "a5nA_e109.pth", "neta-lumina-1.0": "5nA3_e3.pth" # a5nA3_e3.pth } # HTTP 客户端配置 HTTP_LIMITS = httpx.Limits( max_keepalive_connections=5, max_connections=10, keepalive_expiry=30.0 ) HTTP_TIMEOUT = httpx.Timeout(connect=30.0, read=300.0, write=30.0, pool=10.0) # 日志配置 logging.basicConfig( level=logging.INFO, format="%(asctime)s - %(name)s - %(levelname)s - %(message)s", handlers=[ logging.StreamHandler(), logging.FileHandler("app.log", mode="a", encoding="utf-8"), ], ) logger = logging.getLogger(__name__) # 数据模型定义 @dataclass class LuminaConfig: """Lumina模型配置""" model_name: str | None = None cfg: float | None = None step: int | None = None @dataclass class ImageGenerationConfig: """图像生成配置""" prompts: list[dict[str, Any]] = field(default_factory=list) width: int = 1024 height: int = 1024 seed: int | None = None use_polish: bool = False is_lumina: bool = True lumina_config: LuminaConfig = field(default_factory=LuminaConfig) # 全局变量 image_client = None polish_client = None # 示例提示词 example_titles = [ "Empress in hanfu admires lotus pond at sunset.", "1girl, solo, red hair, heterochromia, blue eyes, green eyes", "剪纸风格神秘生物坐在树梢持长枪", "Monochrome cartoon figures battle in crazed, exquisitely detailed abstract style; spectacular black masterpiece.", "向日葵田中持花微笑的金发少女，温暖宁静", ] full_prompts = { example_titles[ 0 ]: "majestic young empress, elegant hanfu with flowing wide sleeves and intricate gold-thread embroidery, rich vermilion outer robe layered over mist-white inner silk, long raven hair in twin coiled buns secured by jade hairpins, phoenix-shaped hair ornament with dangling pearls and red tassels, serene yet commanding expression, almond eyes reflecting warm sunset light, subtle blush, standing atop an ancient marble terrace overlooking a lotus pond, ornate carved balustrade, blooming pink and white lotuses, gentle ripples, floating petals, distant misty mountains shrouded in soft clouds, traditional curved roof pavilions with upturned eaves and glazed tiles, crimson lanterns swaying, pair of white cranes mid-flight, delicate sakura petals drifting on breeze, paper umbrellas resting against stone pillar, atmospheric depth, cinematic composition obeying rule of thirds, low-angle three-quarter view emphasizing grandeur, dynamic flowing fabric caught in wind, high detail fabric folds, realistic subsurface scattering on skin, volumetric golden hour lighting with subtle rim-light accents, ambient occlusion for depth, ultra-sharp focus on character, background slightly defocused bokeh for depth of field, painterly brushstroke texture blended with crisp anime line art, ink-wash influence on distant scenery, subtle grain for traditional scroll feeling, vibrant yet harmonious color palette (vermillion, jade green, imperial gold, cloud white), 8k resolution, ultra-high detail, masterful shading, ray-traced reflections in water, global illumination, HDR, physically-based rendering, artwork by (籠夏:1.2), inspired by (Guweiz), (WLOP:0.9) but in classical Chinese style, professional illustration quality, no watermark, no signature, perfect anatomy, perfect hands, expressive eyes, clean line work, soft gradient background sky transitioning from peach to lavender, subtle haze, mystical atmosphere, gentle glow particles, cinematic ", example_titles[ 1 ]: "1girl, solo, red hair, heterochromia, blue eyes, green eyes, white blouse, bouquet, white lilies, close-up, floral, painterly, expressive, warm tones, soft lighting, heartfelt, anime", example_titles[ 2 ]: "(剪纸艺术:1.4), 白色背景, 蓝色色调, 坐在树梢上, 手拿长枪, (an otherworldly being Nixilith:0.5), (Multiple eyes varying in hues:0.5), (spikey unpredictable hair of intertwined colors:0.5), (Patchy skin of varied tones with slimy and mottled textures:0.5), (Surreal drapery with geometric patterns merging with scales and fur:0.5), (Outfit composed of twisted limbs and unnatural angles:0.5), (Interlaced with sharp fangs and oversized mouth:0.5), (Accessorized by a dark:0.5), (surreal landscape with warped elements:0.5), (and eerie lighting casting shadows and contrasts:0.5), (Artistic fusion of cubist lines and distorted forms:0.5), (Imparting a sense of high-definition intensity:0.5), (Incorporating aspects of cubism and geometric abstraction:0.5), (The overall design permeated with unsettling horror and macabre aesthetics:0.5), (Engaging the viewer with its vivid presence and alien landscape:0.5), (flower - filled dungeon:0.15), (painter style:0.15), (movie:0.1), (dark and damp:0.15), (color_noise_reduction:0.12), (horror movie:0.1), (corruption:0.15), (horror artist style:0.15), (wilted:0.15), (brightness_noise_reduction:0.12), (flowers covering the face:0.15), (artist style:0.15), (rough texture:0.15), (sharpness:0.12), (underwater:0.15), (yellow tint:0.15)", example_titles[ 3 ]: "you are a painter. You draw a picture. In the picture, the stragglers are fighting, and the cartoon line drawing style and the messy and abstract artistic style are in a crazy painting state, but the details are extremely exquisite, the painter's style is outstanding, the bold color matching, the ingenious composition, the strange landscape, the lofty artistic conception, the gorgeous style, the unparalleled creativity, the beautiful lines, the order and the regularity have constructed a spectacular and shocking masterpiece., (pure black:2.0), (black sky:2.0), (black earth:2.0), (black city:2.0), (black person:2.0), (black clouds:2.0), (black trees:2.0)", example_titles[ 4 ]: "平面风格，儿童画风格，生成一幅动漫平面风格女孩图像：主体为一位纤细匀称的少女，鹅蛋脸、尖下巴、大眼睛，金色波浪长发及肩，佩戴向日葵发夹。她身着白色棉质衬衫和黄色丝绸长裙，脚穿棕色凉鞋，站立于广阔向日葵花田中，手持一朵盛开的向日葵，面带天真微笑。构图采用三分法则，少女位于右侧黄金点，视线从脸部引导至向日葵。画面以温暖色调为主（黄、橙、绿），柔和左侧日光营造低对比光影，平视中景浅景深突出焦点。氛围温暖宁静，暗示鸟鸣、花香和阳光温暖感，背景有飞舞蝴蝶增强叙事。", } # 工具函数 def validate_dimensions(width: int, height: int) -> tuple[int, int]: """验证并调整图片尺寸""" width = max(MIN_IMAGE_SIZE, min(int(width), MAX_IMAGE_SIZE)) height = max(MIN_IMAGE_SIZE, min(int(height), MAX_IMAGE_SIZE)) width = (width // 32) * 32 height = (height // 32) * 32 return width, height def feishu_notify(message: str): """发送飞书通知（异步非阻塞）""" threading.Thread(target=_send_feishu_notify, args=(message,)).start() def _send_feishu_notify(message: str): """发送飞书通知的内部实现""" try: headers = { "Content-Type": "application/json", } content = { "msg_type": "text", "content": { "text": message, }, } with httpx.Client(timeout=10.0) as client: response = client.post(CHAT_WEBHOOK_URL, headers=headers, json=content) if response.status_code == 200: logger.info("飞书通知发送成功") else: logger.warning(f"飞书通知发送失败: {response.status_code}") except Exception as e: logger.error(f"发送飞书通知时出错: {str(e)}") def format_error_for_notification(error_type: str, error_message: str, traceback_str: str) -> str: """格式化错误信息用于飞书通知""" return f""" 🚨 应用错误警报 错误类型: {error_type} 错误信息: {error_message} 详细错误栈: {traceback_str} 时间: {time.strftime('%Y-%m-%d %H:%M:%S')} """ class PolishClient: """提示词润色客户端""" def __init__(self): self.x_token = API_TOKEN if not self.x_token: raise ValueError("环境变量中未设置API_TOKEN") self.url = "https://api.talesofai.cn/v3/gpt/dify/text-complete" self.headers = { "x-token": self.x_token, "Content-Type": "application/json", "x-nieta-app-version": "5.14.0", "x-platform": "nieta-app/web", } async def polish_text(self, input_text: str) -> str: """润色文本""" payload = { "query": "", "response_mode": "blocking", "preset_key": "latitude://60|live|lumina_polish", "inputs": {"query": input_text}, } async with httpx.AsyncClient(timeout=HTTP_TIMEOUT) as client: response = await client.post(self.url, headers=self.headers, json=payload) if response.status_code == HTTP_STATUS_OK: response_data = response.json() polished_text = response_data.get("answer", input_text) return polished_text.strip() else: logger.warning(f"润色API调用失败: {response.status_code}") return input_text async def polish_prompt(prompt: str) -> str: """提示词润色函数 - 使用外部API润色""" global polish_client if polish_client is None: polish_client = PolishClient() try: logger.info(f"润色提示词: {prompt[:50]}...") polished_prompt = await polish_client.polish_text(prompt) logger.info("提示词润色完成") return polished_prompt except Exception as e: error_message = f"提示词润色异常: {str(e)}" traceback_str = traceback.format_exc() logger.error(error_message) # 发送飞书通知 notification_message = format_error_for_notification("PolishPromptError", error_message, traceback_str) feishu_notify(notification_message) return prompt # 返回原始提示词作为fallback # 核心类定义 class ImageClient: """图像生成客户端""" def __init__(self) -> None: self.x_token = API_TOKEN if not self.x_token: raise ValueError("环境变量中未设置API_TOKEN") self.lumina_api_url = "https://ops.api.talesofai.cn/v3/make_image" self.lumina_task_status_url = ( "https://ops.api.talesofai.cn/v1/artifact/task/{task_uuid}" ) self.max_polling_attempts = 100 self.polling_interval = 3.0 self.default_headers = { "Content-Type": "application/json", "x-platform": "nieta-app/web", "X-Token": self.x_token, } self._client_config = { "limits": HTTP_LIMITS, "timeout": HTTP_TIMEOUT, "headers": self.default_headers, } self._active_tasks = weakref.WeakSet() def _prepare_prompt_data( self, prompt: str, negative_prompt: str = "" ) -> list[dict[str, Any]]: """准备提示词数据""" prompts_data = [{"type": "freetext", "value": prompt, "weight": 1.0}] if negative_prompt: prompts_data.append( {"type": "freetext", "value": negative_prompt, "weight": -1.0} ) # prompts_data.append( # { # "type": "elementum", # "value": "b5edccfe-46a2-4a14-a8ff-f4d430343805", # "uuid": "b5edccfe-46a2-4a14-a8ff-f4d430343805", # "weight": 1.0, # "name": "lumina1", # "img_url": "https://oss.talesofai.cn/picture_s/1y7f53e6itfn_0.jpeg", # "domain": "", # "parent": "", # "label": None, # "sort_index": 0, # "status": "IN_USE", # "polymorphi_values": {}, # "sub_type": None, # } # ) prompts_data.append( { "type": "elementum", "value": "dccc4881-c8da-4bb7-ae39-46ae3992e660", "uuid": "dccc4881-c8da-4bb7-ae39-46ae3992e660", "weight": 1.0, "name": "lumina2", "img_url": "https://oss.talesofai.cn/picture_s/x2an269l5qrz_0.jpeg", "domain": "", "parent": "", "label": None, "sort_index": 0, "status": "IN_USE", "polymorphi_values": {}, "sub_type": None, } ) return prompts_data def _build_payload(self, config: ImageGenerationConfig) -> dict[str, Any]: """构建请求载荷""" payload = { "storyId": "", "jobType": "universal", "width": config.width, "height": config.height, "rawPrompt": config.prompts, "seed": config.seed, "meta": {"entrance": "PICTURE,PURE"}, "context_model_series": None, "negative_freetext": "", "advanced_translator": config.use_polish, } if config.is_lumina: client_args = {} client_args["seed"] = config.seed if config.lumina_config.model_name: client_args["ckpt_name"] = config.lumina_config.model_name if config.lumina_config.cfg is not None: client_args["cfg"] = str(config.lumina_config.cfg) if config.lumina_config.step is not None: client_args["steps"] = str(config.lumina_config.step) if client_args: payload["client_args"] = client_args return payload async def _poll_task_status(self, task_uuid: str) -> dict[str, Any]: """轮询任务状态 - 优化内存使用和连接管理""" status_url = self.lumina_task_status_url.format(task_uuid=task_uuid) poll_timeout = httpx.Timeout(connect=10.0, read=30.0, write=10.0, pool=5.0) try: async with httpx.AsyncClient( limits=HTTP_LIMITS, timeout=poll_timeout, headers=self.default_headers ) as client: for attempt in range(self.max_polling_attempts): try: response = await client.get(status_url) if response.status_code != HTTP_STATUS_OK: logger.warning(f"轮询失败 - 状态码: {response.status_code}") return { "success": False, "error": f"获取任务状态失败: {response.status_code} - {response.text[:200]}", } try: result = response.json() except Exception as e: logger.warning(f"JSON解析失败: {str(e)}") return { "success": False, "error": f"任务状态响应解析失败: {response.text[:500]}", } task_status = result.get("task_status") if task_status == "SUCCESS": artifacts = result.get("artifacts", []) if artifacts and len(artifacts) > 0: image_url = artifacts[0].get("url") if image_url: return {"success": True, "image_url": image_url} return { "success": False, "error": "无法从结果中提取图像URL", } elif task_status == "FAILURE": return { "success": False, "error": result.get("error", "任务执行失败"), } elif task_status == "ILLEGAL_IMAGE": return {"success": False, "error": "图片不合规"} elif task_status == "TIMEOUT": return {"success": False, "error": "任务超时"} await asyncio.sleep(self.polling_interval) if attempt % 10 == 0 and attempt > 0: gc.collect() logger.debug(f"轮询第{attempt}次，执行内存清理") except asyncio.TimeoutError: logger.warning(f"轮询超时 - 尝试 {attempt}") if attempt >= 3: return { "success": False, "error": "连接超时，请检查网络状况", } await asyncio.sleep(self.polling_interval * 2) continue except Exception as e: logger.warning(f"轮询异常 - 尝试 {attempt}: {str(e)}") if attempt >= 3: return {"success": False, "error": f"网络异常: {str(e)}"} await asyncio.sleep(self.polling_interval) continue except Exception as e: logger.error(f"轮询过程异常: {str(e)}") return {"success": False, "error": f"轮询过程异常: {str(e)}"} finally: gc.collect() return { "success": False, "error": "⏳ 生图任务超时（5分钟），服务器可能正在处理大量请求，请稍后重试", } async def generate_image( self, prompt: str, negative_prompt: str, seed: int, width: int, height: int, cfg: float, steps: int, model_name: str = "neta-lumina-1.0", use_polish: bool = False, ) -> tuple[str | None, str | None]: """生成图片""" try: model_path = MODEL_CONFIGS.get(model_name, MODEL_CONFIGS["neta-lumina-1.0"]) config = ImageGenerationConfig( prompts=self._prepare_prompt_data(prompt, negative_prompt), width=width, height=height, seed=seed, use_polish=use_polish, is_lumina=True, lumina_config=LuminaConfig(model_name=model_path, cfg=cfg, step=steps), ) async with httpx.AsyncClient(**self._client_config) as client: payload = self._build_payload(config) if DEBUG_MODE: logger.debug(f"发送API请求到 {self.lumina_api_url}") logger.info(f"请求载荷: {payload}") print(payload) try: response = await client.post(self.lumina_api_url, json=payload) except asyncio.TimeoutError: return None, "请求超时，请稍后重试" except Exception as e: logger.error(f"API请求异常: {str(e)}") return None, f"网络请求失败: {str(e)}" if DEBUG_MODE: logger.debug(f"API响应状态码: {response.status_code}") logger.debug(f"API响应内容: {response.text[:1000]}") if response.status_code == HTTP_STATUS_CENSORED: return None, "内容不合规" if response.status_code == 433: return None, "⏳ 服务器正忙，同时生成的图片数量已达上限，请稍后重试" if response.status_code != HTTP_STATUS_OK: return ( None, f"API请求失败: {response.status_code} - {response.text[:200]}", ) content = response.text.strip() task_uuid = content.replace('"', "") if DEBUG_MODE: logger.debug(f"API返回UUID: {task_uuid}") if not task_uuid: return None, f"未获取到任务ID，API响应: {response.text[:200]}" result = await self._poll_task_status(task_uuid) if result["success"]: return result["image_url"], None else: return None, result["error"] except Exception as e: error_message = f"生成图片异常: {str(e)}" traceback_str = traceback.format_exc() logger.error(error_message) # 发送飞书通知 notification_message = format_error_for_notification("ImageGenerationError", error_message, traceback_str) feishu_notify(notification_message) return None, f"生成图片时发生错误: {str(e)}" finally: gc.collect() def cleanup(self): """清理资源""" try: self._active_tasks.clear() gc.collect() logger.info("ImageClient资源清理完成") except Exception as e: logger.error(f"资源清理异常: {str(e)}") class MemoryMonitor: """内存监控器""" def __init__(self): self.running = False self.monitor_thread = None def start_monitoring(self): """启动内存监控""" if not self.running: self.running = True self.monitor_thread = threading.Thread( target=self._monitor_loop, daemon=True ) self.monitor_thread.start() logger.info("内存监控已启动") def stop_monitoring(self): """停止内存监控""" self.running = False if self.monitor_thread: self.monitor_thread.join(timeout=5) logger.info("内存监控已停止") def _monitor_loop(self): """内存监控循环""" while self.running: try: process = psutil.Process() memory_info = process.memory_info() memory_mb = memory_info.rss / 1024 / 1024 memory_percent = process.memory_percent() if memory_percent > MEMORY_THRESHOLD_PERCENT: logger.warning( f"内存使用量过高: {memory_mb:.2f} MB ({memory_percent:.1f}%)" ) gc.collect() logger.info("执行垃圾回收") else: logger.debug( f"当前内存使用: {memory_mb:.2f} MB ({memory_percent:.1f}%)" ) time.sleep(MEMORY_CHECK_INTERVAL) except Exception as e: logger.error(f"内存监控异常: {str(e)}") time.sleep(MEMORY_CHECK_INTERVAL) # 辅助函数 def get_image_client() -> ImageClient: """获取图片生成客户端实例 - 单例模式""" global image_client if image_client is None: try: image_client = ImageClient() logger.info("ImageClient初始化成功") except Exception as e: logger.error(f"ImageClient初始化失败: {e}") raise return image_client async def cleanup_resources(): """清理应用资源""" global image_client, memory_monitor, polish_client try: memory_monitor.stop_monitoring() if image_client is not None: image_client.cleanup() image_client = None if polish_client is not None: polish_client = None gc.collect() try: process = psutil.Process() memory_mb = process.memory_info().rss / 1024 / 1024 logger.info(f"清理后内存使用: {memory_mb:.2f} MB") except Exception: pass logger.info("应用资源清理完成") except Exception as e: logger.error(f"资源清理异常: {str(e)}") async def infer( prompt_text, use_polish_val, seed_val, randomize_seed_val, width_val, height_val, cfg_val, steps_val, model_name_val, ): """推理函数 - 优化内存管理和错误处理""" try: logger.info(f"开始生成图像: {prompt_text[:50]}...") try: client = get_image_client() except Exception as e: logger.error(f"获取ImageClient失败: {str(e)}") raise gr.Error("ImageClient 未正确初始化。请检查应用日志和API_TOKEN配置。") if not prompt_text.strip(): raise gr.Error("提示词不能为空") final_prompt = prompt_text if use_polish_val: final_prompt = await polish_prompt(prompt_text) # 为生成添加系统前缀，但不在UI显示 # system_prefix = "You are an assistant designed to generate anime images with the highest degree of image-text alignment based on danbooru tags. " # generation_prompt = f"{system_prefix} {final_prompt}" generation_prompt = final_prompt current_seed = int(seed_val) if randomize_seed_val: current_seed = random.randint(0, MAX_SEED) width_val, height_val = validate_dimensions(width_val, height_val) if not (1.0 <= float(cfg_val) <= 20.0): raise gr.Error("CFG Scale 必须在 1.0 到 20.0 之间") if not (1 <= int(steps_val) <= 50): raise gr.Error("Steps 必须在 1 到 50 之间") image_url, error = await client.generate_image( prompt=generation_prompt, # 使用包含系统前缀的提示词 negative_prompt="", seed=current_seed, width=width_val, height=height_val, cfg=float(cfg_val), steps=int(steps_val), model_name=model_name_val, use_polish=False, ) if error: logger.error(f"图像生成失败: {error}") raise gr.Error(error) logger.info("图像生成成功") return image_url, final_prompt, current_seed # 返回不带系统前缀的用户提示词 except gr.Error: raise except Exception as e: error_message = f"推理过程异常: {str(e)}" traceback_str = traceback.format_exc() logger.error(error_message) # 发送飞书通知 notification_message = format_error_for_notification("InferenceError", error_message, traceback_str) feishu_notify(notification_message) raise gr.Error(f"生成图像时发生意外错误: {str(e)}") finally: gc.collect() def setup_signal_handlers(): """设置信号处理器以实现优雅关闭""" import signal import atexit def cleanup_handler(signum=None, frame=None): """清理处理器""" logger.info("接收到关闭信号，开始清理资源...") try: asyncio.run(cleanup_resources()) except Exception as e: logger.error(f"清理资源时出错: {str(e)}") logger.info("应用已安全关闭") signal.signal(signal.SIGINT, cleanup_handler) signal.signal(signal.SIGTERM, cleanup_handler) atexit.register(cleanup_handler) # UI 构建 def build_ui(): """构建Gradio UI""" with gr.Blocks(theme=gr.themes.Soft(), title="Lumina Image Playground") as demo: gr.Markdown("

🎨 NetaLumina_T2I_Playground | 捏Ta Lumina

") gr.Markdown( "Fine-tuned Lumina model specialized for anime/manga style generation! Supports Chinese, English, and Japanese prompts. " ) gr.Markdown( "🌸 专为二次元风格优化的Lumina模型！支持中文、英文、日文三语提示词，让您的创意无界限！" ) gr.HTML(f"""
Discord Join Discord | 加入Discord App Index Nieta Home | 捏Ta主页 📚 PromptBook | 📖 提示词宝典
""") with gr.Row(variant="panel"): with gr.Column(scale=2): # Controls Panel gr.Markdown("## ⚙️ Generation Controls | 生成控制") prompt = gr.Textbox( label="Prompt | 提示词", lines=5, placeholder="e.g., A majestic dragon soaring through a cyberpunk city skyline, neon lights reflecting off its scales, intricate details. | 例如：一条威武的巨龙翱翔在赛博朋克城市天际线，霓虹灯映照在它的鳞片上，细节精美。", info="System prompt for anime image generation is automatically included: 'You are an assistant designed to generate anime images based on textual prompts. '", ) use_polish = gr.Checkbox( label="✨ Auto Polish Prompt | 自动润色提示词", value=True, info="Automatically optimize and enhance your prompt for better results. | 自动优化和增强您的提示词以获得更好的效果。", ) run_button = gr.Button( "🚀 Generate Image | 生成图像", variant="primary", scale=0 ) with gr.Accordion("🔧 Advanced Settings | 高级设置", open=False): model_name = gr.Dropdown( label="Model Version | 模型版本", choices=list(MODEL_CONFIGS.keys()), value="neta-lumina-1.0", info="Select the generation model. | 选择生成模型。", ) with gr.Row(): cfg = gr.Slider( label="CFG Scale | CFG缩放", minimum=1.0, maximum=20.0, step=0.1, value=5.5, info="Guidance strength. Higher values adhere more to prompt. | 引导强度，更高的值更贴近提示词。", ) steps = gr.Slider( label="Sampling Steps | 采样步数", minimum=1, maximum=50, step=1, value=30, info="Number of steps. More steps can improve quality but take longer. | 步数，更多步数可提高质量但耗时更长。", ) with gr.Row(): width = gr.Slider( label="Width | 宽度", minimum=MIN_IMAGE_SIZE, maximum=MAX_IMAGE_SIZE, step=32, value=1024, ) height = gr.Slider( label="Height | 高度", minimum=MIN_IMAGE_SIZE, maximum=MAX_IMAGE_SIZE, step=32, value=1024, ) with gr.Row(): seed = gr.Slider( label="Seed | 种子", minimum=0, maximum=MAX_SEED, step=1, value=random.randint(0, MAX_SEED), ) randomize_seed = gr.Checkbox( label="Randomize Seed | 随机种子", value=True, info="Use a new random seed for each generation if checked. | 勾选后每次生成使用新的随机种子。", ) with gr.Group(): gr.Markdown("### ✨ Example Prompts | 示例提示词") for i, title in enumerate(example_titles): btn = gr.Button(title) btn.click(lambda t=title: (full_prompts[t], False), outputs=[prompt, use_polish]) with gr.Column(scale=3): # Output Panel gr.Markdown("## 🖼️ Generated Image | 生成图像") result_image = gr.Image( label="Output Image | 输出图像", show_label=False, type="filepath", width=1024, height=576, show_download_button=True, interactive=False, elem_id="result_image_display", container=True, show_fullscreen_button=True, ) used_prompt_info = gr.Textbox( label="Used Prompt | 使用的提示词", interactive=False, lines=3, placeholder="The actual prompt used for generation will appear here. | 生成时实际使用的提示词将显示在此处。", ) generated_seed_info = gr.Textbox( label="Seed Used | 使用的种子", interactive=False, placeholder="The seed for the generated image will appear here. | 生成图像所使用的种子值将显示在此处。", ) # 事件处理 inputs_list = [ prompt, use_polish, seed, randomize_seed, width, height, cfg, steps, model_name, ] outputs_list = [result_image, used_prompt_info, generated_seed_info] run_button.click( fn=infer, inputs=inputs_list, outputs=outputs_list, api_name="generate_image", ) prompt.submit( fn=infer, inputs=inputs_list, outputs=outputs_list, api_name="generate_image_submit", ) return demo # 主函数 def main(): """主函数""" global memory_monitor try: # 设置信号处理器 setup_signal_handlers() # 启动内存监控 memory_monitor = MemoryMonitor() memory_monitor.start_monitoring() if DEBUG_MODE: logger.info("调试模式已启用") if not os.environ.get("API_TOKEN"): logger.warning("API_TOKEN环境变量未设置") print( "**************************************************************************************" ) print("WARNING: API_TOKEN environment variable is not set locally.") print( "The application will run, but image generation will fail until API_TOKEN is provided." ) print( "You can set it by running: export API_TOKEN='your_actual_token_here'" ) print( "Or if using a .env file, ensure it's loaded or API_TOKEN is set in your run config." ) print( "**************************************************************************************" ) # 记录启动时内存使用情况 try: process = psutil.Process() startup_memory = process.memory_info().rss / 1024 / 1024 logger.info(f"启动时内存使用: {startup_memory:.2f} MB") except Exception: pass logger.info("启动Gradio应用...") demo = build_ui() # 启用队列支持多用户并发 demo.queue( max_size=300, # 最大队列长度 default_concurrency_limit=30, # 默认并发限制 ) demo.launch( debug=DEBUG_MODE, show_error=True, server_name="0.0.0.0", server_port=7860, share=False, max_threads=40, favicon_path=None, ) except KeyboardInterrupt: logger.info("接收到键盘中断，正在关闭...") except Exception as e: error_message = f"应用启动失败: {str(e)}" traceback_str = traceback.format_exc() logger.error(error_message) # 发送飞书通知 notification_message = format_error_for_notification("ApplicationStartupError", error_message, traceback_str) feishu_notify(notification_message) raise finally: # 确保资源清理 try: asyncio.run(cleanup_resources()) except Exception as e: error_message = f"最终清理时出错: {str(e)}" traceback_str = traceback.format_exc() logger.error(error_message) # 发送飞书通知 notification_message = format_error_for_notification("CleanupError", error_message, traceback_str) feishu_notify(notification_message) logger.info("应用已退出") if __name__ == "__main__": main()