import gradio as gr import threading import os import shutil import tempfile import time import json import util from util import (process_image_edit, process_text_to_image, process_image_upscale, process_face_swap, process_multi_image_edit, process_watermark_removal, download_and_check_result_nsfw, GoodWebsiteUrl, RateLimitConfig, create_mask_from_layers, TASK_LOCK_MINUTES, PRINT_STATS_INTERVAL) from nfsw import NSFWDetector # 创建全局配置实例 rate_limit_config = RateLimitConfig() # ============================================================================= # i18n 翻译系统 - Load from encrypted modules # ============================================================================= import sys from pathlib import Path # Add i18n module to path _i18n_module_path = Path(__file__).parent / "__lib__" / "i18n" if str(_i18n_module_path) not in sys.path: sys.path.insert(0, str(_i18n_module_path)) # Import encrypted i18n loader from i18n import translations as _translations translations = _translations def load_translations(): """Compatibility function - translations are already loaded""" return translations def t(key, lang="en"): default_en = translations.get("en", {}) lang_dict = translations.get(lang) or default_en if key in lang_dict: return lang_dict[key] return default_en.get(key, key) def has_active_task(client_ip): """ 检查IP是否有正在进行的任务（3分钟内） Returns: tuple: (has_active, remaining_seconds) """ if client_ip not in util.IP_Active_Tasks: return False, 0 start_time = util.IP_Active_Tasks[client_ip] elapsed = time.time() - start_time lock_seconds = TASK_LOCK_MINUTES * 60 if elapsed < lock_seconds: remaining = lock_seconds - elapsed return True, remaining else: # 任务锁定已过期，清除 del util.IP_Active_Tasks[client_ip] return False, 0 def set_active_task(client_ip, active=True): """ 设置或清除IP的活动任务状态 """ if active: util.IP_Active_Tasks[client_ip] = time.time() print(f"🔒 Task locked for IP: {client_ip}") else: if client_ip in util.IP_Active_Tasks: del util.IP_Active_Tasks[client_ip] # 记录任务完成时间 util.IP_Last_Task_Time[client_ip] = time.time() print(f"🔓 Task unlocked for IP: {client_ip}") country_dict = { "zh": ["中国"], "hi": ["印度"], "fi": ["芬兰"], "en": ["美国", "澳大利亚", "英国", "加拿大", "新西兰", "爱尔兰"], "es": ["西班牙", "墨西哥", "阿根廷", "哥伦比亚", "智利", "秘鲁"], "pt": ["葡萄牙", "巴西"], "fr": ["法国", "摩纳哥"], "de": ["德国", "奥地利"], "it": ["意大利", "圣马力诺", "梵蒂冈"], "ja": ["日本"], "ru": ["俄罗斯"], "uk": ["乌克兰"], "ar": ["沙特阿拉伯", "埃及", "阿拉伯联合酋长国", "摩洛哥"], "nl": ["荷兰"], "no": ["挪威"], "sv": ["瑞典"], "id": ["印度尼西亚"], "vi": ["越南"], "he": ["以色列"], "tr": ["土耳其"], "da": ["丹麦"], } def query_ip_country(client_ip): """查询IP地址地理信息""" if client_ip in util.IP_Country_Cache: return util.IP_Country_Cache[client_ip] if not client_ip or client_ip in ["127.0.0.1", "localhost", "::1"]: default_geo = {"country": "Unknown", "region": "Unknown", "city": "Unknown"} util.IP_Country_Cache[client_ip] = default_geo return default_geo try: import requests from requests.exceptions import Timeout, ConnectionError, RequestException api_url = f"https://api.vore.top/api/IPdata?ip={client_ip}" response = requests.get(api_url, timeout=3) if response.status_code == 200: data = response.json() if data.get("code") == 200 and "ipdata" in data: ipdata = data["ipdata"] geo_info = { "country": ipdata.get("info1", "Unknown"), "region": ipdata.get("info2", "Unknown"), "city": ipdata.get("info3", "Unknown") } util.IP_Country_Cache[client_ip] = geo_info return geo_info except Exception as e: print(f"Error querying IP {client_ip}: {e}") default_geo = {"country": "Unknown", "region": "Unknown", "city": "Unknown"} util.IP_Country_Cache[client_ip] = default_geo return default_geo def get_lang_from_country(country): """根据国家获取语言代码""" if not country or country == "Unknown": return "en" for lang, countries in country_dict.items(): if country in countries: return lang return "en" def get_lang_from_ip(client_ip): """根据IP获取语言""" geo_info = query_ip_country(client_ip) return get_lang_from_country(geo_info.get("country", "Unknown")) def get_ip_generation_count(client_ip): """获取IP生成次数""" return util.IP_Generation_Count.get(client_ip, 0) def increment_ip_generation_count(client_ip): """增加IP生成次数""" if client_ip not in util.IP_Generation_Count: util.IP_Generation_Count[client_ip] = 0 util.IP_Generation_Count[client_ip] += 1 return util.IP_Generation_Count[client_ip] def get_ip_phase(client_ip): """获取IP当前阶段""" count = get_ip_generation_count(client_ip) geo_info = util.IP_Country_Cache.get(client_ip, {"country": "Unknown"}) country = geo_info.get("country", "Unknown") config = rate_limit_config.get_country_config(country) max_limit = config.get("max_limit", rate_limit_config.BLOCKED_LIMIT) if config.get("is_restricted"): if count >= max_limit: return 'blocked' elif count >= max(0, max_limit - 2): return 'rate_limit_3' elif count >= max(0, max_limit - 3): return 'rate_limit_2' elif count >= max(0, max_limit - 4): return 'rate_limit_1' else: return 'free' free_limit = config.get("free_phase_limit", rate_limit_config.FREE_PHASE_LIMIT) if count < free_limit: return 'free' elif count < rate_limit_config.SLOW_PHASE_1_LIMIT: return 'rate_limit_1' elif count < rate_limit_config.SLOW_PHASE_2_LIMIT: return 'rate_limit_2' elif count < max_limit: return 'rate_limit_3' else: return 'blocked' def check_rate_limit_for_phase(client_ip, phase): """检查阶段的速率限制""" if phase not in ['rate_limit_1', 'rate_limit_2', 'rate_limit_3']: return False, 0, 0 if phase == 'rate_limit_1': window_minutes = rate_limit_config.PHASE_1_WINDOW_MINUTES elif phase == 'rate_limit_2': window_minutes = rate_limit_config.PHASE_2_WINDOW_MINUTES else: window_minutes = rate_limit_config.PHASE_3_WINDOW_MINUTES current_time = time.time() window_key = f"{client_ip}_{phase}" if window_key in util.IP_Rate_Limit_Track: track_data = util.IP_Rate_Limit_Track[window_key] if current_time - track_data['start_time'] > window_minutes * 60: util.IP_Rate_Limit_Track[window_key] = { 'count': 0, 'start_time': current_time, 'last_generation': current_time } else: util.IP_Rate_Limit_Track[window_key] = { 'count': 0, 'start_time': current_time, 'last_generation': current_time } track_data = util.IP_Rate_Limit_Track[window_key] if track_data['count'] >= rate_limit_config.MAX_IMAGES_PER_WINDOW: elapsed = current_time - track_data['start_time'] wait_time = (window_minutes * 60) - elapsed wait_minutes = max(0, wait_time / 60) return True, wait_minutes, track_data['count'] return False, 0, track_data['count'] def update_country_stats(client_ip): """更新国家使用统计""" geo_info = util.IP_Country_Cache.get(client_ip, {"country": "Unknown"}) country = geo_info["country"] if country not in util.Country_Usage_Stats: util.Country_Usage_Stats[country] = 0 util.Country_Usage_Stats[country] += 1 util.Total_Request_Count += 1 if util.Total_Request_Count % PRINT_STATS_INTERVAL == 0: print("\n" + "="*60) print(f"📊 国家使用统计 (总请求数: {util.Total_Request_Count})") print("="*60) sorted_stats = sorted(util.Country_Usage_Stats.items(), key=lambda x: x[1], reverse=True) for country_name, count in sorted_stats: percentage = (count / util.Total_Request_Count) * 100 print(f" {country_name}: {count} 次 ({percentage:.1f}%)") print("="*60 + "\n") def record_generation_attempt(client_ip, phase): """记录生成尝试""" increment_ip_generation_count(client_ip) update_country_stats(client_ip) if phase in ['rate_limit_1', 'rate_limit_2', 'rate_limit_3']: window_key = f"{client_ip}_{phase}" current_time = time.time() if window_key in util.IP_Rate_Limit_Track: util.IP_Rate_Limit_Track[window_key]['count'] += 1 util.IP_Rate_Limit_Track[window_key]['last_generation'] = current_time else: util.IP_Rate_Limit_Track[window_key] = { 'count': 1, 'start_time': current_time, 'last_generation': current_time } def apply_gaussian_blur_to_image_url(image_url, blur_strength=50): """对图片URL应用高斯模糊""" try: import requests from PIL import Image, ImageFilter import io response = requests.get(image_url, timeout=30) if response.status_code != 200: return None image_data = io.BytesIO(response.content) image = Image.open(image_data) blurred_image = image.filter(ImageFilter.GaussianBlur(radius=blur_strength)) return blurred_image except Exception as e: print(f"⚠️ Failed to apply Gaussian blur: {e}") return None # Initialize NSFW detector try: nsfw_detector = NSFWDetector() print("✅ NSFW detector initialized successfully") except Exception as e: print(f"❌ NSFW detector initialization failed: {e}") nsfw_detector = None # ============================================================================= # 通用NSFW检测函数 # ============================================================================= def check_nsfw_for_input(input_image, country, current_count, client_ip): """ 检测输入图片是否为NSFW内容 返回: (is_nsfw, should_check) - 是否为NSFW，是否需要检测 """ if input_image is None: return False, False if not rate_limit_config.should_enable_nsfw(country, current_count): return False, False if nsfw_detector is None: return False, False try: nsfw_result = nsfw_detector.predict_pil_label_only(input_image) if nsfw_result.lower() == "nsfw": print(f"🔍 Input NSFW detected: ❌❌❌ - IP: {client_ip}") return True, True return False, True except Exception as e: print(f"⚠️ NSFW detection failed: {e}") return False, True def check_nsfw_for_result(result_url, country, current_count): """ 检测结果图片是否为NSFW内容 返回: is_nsfw """ if nsfw_detector is None: return False if not rate_limit_config.should_enable_nsfw(country, current_count): return False try: is_nsfw, _ = download_and_check_result_nsfw(result_url, nsfw_detector) return is_nsfw except Exception: return False def create_nsfw_blurred_response(result_url, redirect_url, lang): """ 创建NSFW模糊处理后的响应HTML 返回: (result_html, action_html) 或 None 如果模糊失败 """ blurred_image = apply_gaussian_blur_to_image_url(result_url) if blurred_image is None: return None blurred_html = pil_image_to_base64_html(blurred_image) nsfw_button_html = f"""
🔒 Unlock Full Image - Visit Website
""" return blurred_html, nsfw_button_html # ============================================================================= # 通用处理函数 # ============================================================================= def create_blocked_button_html(url): """创建封锁状态的按钮HTML""" return f"""
🚀 Unlimited Generation
""" def create_rate_limit_button_html(url): """创建限速状态的按钮HTML""" return f"""
⏰ Skip Wait - Unlimited Generation
""" def create_task_locked_button_html(url, remaining_seconds): """创建任务锁定状态的按钮HTML""" remaining_minutes = int(remaining_seconds / 60) + 1 return f"""
⏰ Process Multiple Tasks
""" def create_like_tip_html(): """创建点击红心提示HTML""" return """
👉 Click the ❤️ Like button to unlock more free trial attempts!
""" def create_result_image_html(image_url, max_height=500): """创建结果图片HTML（直接从源站加载，不经过HF Space服务器）""" if not image_url: return "" return f"""
Result Image
""" def pil_image_to_base64_html(pil_image, max_height=500): """将PIL Image转为base64嵌入HTML（不经过服务器中转）""" import io import base64 try: if pil_image is None: return "" # 转为JPEG格式 img_buffer = io.BytesIO() if pil_image.mode != 'RGB': pil_image = pil_image.convert('RGB') pil_image.save(img_buffer, format='JPEG', quality=85) img_data = img_buffer.getvalue() # 编码为base64 img_base64 = base64.b64encode(img_data).decode('utf-8') return f"""
Blurred Result
""" except Exception as e: print(f"⚠️ Failed to convert PIL to base64 HTML: {e}") return "" def create_action_buttons_html(task_uuid, input_image_url, result_url, prompt, lang, show_like_tip): """创建操作按钮HTML""" action_buttons_html = "" if task_uuid and lang not in ["zh", "hi", "ru"]: task_detail_url = f"https://omnicreator.net/my-creations/task/{task_uuid}" from urllib.parse import quote encoded_prompt = quote(prompt.strip()) if prompt else "" encoded_result_url = quote(result_url) if result_url else "" i2v_url = f"https://omnicreator.net/image-to-video?input_image={input_image_url}&end_image={result_url}&prompt={encoded_prompt}" face_swap_url = f"https://omnicreator.net/face-swap?user_image={encoded_result_url}" action_buttons_html = f"""
🎥 Convert to Video 🧍 Swap Face 💾 Download HD Image
""" if show_like_tip or lang in ["zh", "hi", "ru"]: action_buttons_html += create_like_tip_html() return action_buttons_html # ============================================================================= # 单图编辑接口 # ============================================================================= def edit_image_interface(input_image, prompt, lang, request: gr.Request, progress=gr.Progress()): """单图编辑接口""" try: client_ip = request.client.host x_forwarded_for = dict(request.headers).get('x-forwarded-for') if x_forwarded_for: client_ip = x_forwarded_for if client_ip not in util.IP_Dict: util.IP_Dict[client_ip] = 0 util.IP_Dict[client_ip] += 1 if input_image is None: return "", None, t("error_upload_first", lang), gr.update(visible=False) if not prompt or prompt.strip() == "": return "", None, t("error_enter_prompt", lang), gr.update(visible=False) if len(prompt.strip()) <= 3: return "", None, t("error_prompt_too_short", lang), gr.update(visible=False) except Exception as e: return "", None, t("error_request_processing", lang), gr.update(visible=False) geo_info = util.IP_Country_Cache.get(client_ip, {"country": "Unknown"}) country = geo_info.get("country", "Unknown") current_phase = get_ip_phase(client_ip) current_count = get_ip_generation_count(client_ip) is_restricted = rate_limit_config.is_restricted_country(country) show_like_tip = rate_limit_config.should_show_like_tip(country, current_count) redirect_url = rate_limit_config.get_redirect_url(country, lang) # 检查是否有活动任务（3分钟内） is_active, remaining_seconds = has_active_task(client_ip) if is_active: return "", None, f"⏰ You have a task in progress. Please wait until it is finished before submitting a new task.", gr.update(value=create_task_locked_button_html(redirect_url, remaining_seconds), visible=True) # 检查是否被封锁 if current_phase == 'blocked': return "", None, t("error_free_limit_reached", lang), gr.update(value=create_blocked_button_html(redirect_url), visible=True) # 检查速率限制 if current_phase in ['rate_limit_1', 'rate_limit_2', 'rate_limit_3']: is_limited, wait_minutes, window_count = check_rate_limit_for_phase(client_ip, current_phase) if is_limited: wait_minutes_int = int(wait_minutes) + 1 return "", None, t("error_free_limit_wait", lang).format(wait_minutes_int=wait_minutes_int), gr.update(value=create_rate_limit_button_html(redirect_url), visible=True) # NSFW检测 is_nsfw_task, _ = check_nsfw_for_input(input_image, country, current_count, client_ip) def progress_callback(message): try: if progress is not None: if "Queue:" in message or "tasks ahead" in message: progress(0.1, desc=message) elif "Processing" in message or "AI is processing" in message: progress(0.7, desc=message) elif "Generating" in message or "Almost done" in message: progress(0.9, desc=message) else: progress(0.5, desc=message) except Exception: pass try: # 设置任务锁定 set_active_task(client_ip, True) task_priority = 1 if current_count < rate_limit_config.HIGH_PRIORITY_COUNT else 0 record_generation_attempt(client_ip, current_phase) input_image_url, result_url, message, task_uuid = process_image_edit(input_image, prompt.strip(), None, progress_callback, priority=task_priority, client_ip=client_ip) if message and message.startswith("HF_LIMIT_EXCEEDED:"): return "", None, t("error_free_limit_reached", lang), gr.update(value=create_blocked_button_html(redirect_url), visible=True) if result_url: # 检查结果图片NSFW if check_nsfw_for_result(result_url, country, current_count): is_nsfw_task = True # 应用模糊处理 if is_nsfw_task: nsfw_response = create_nsfw_blurred_response(result_url, redirect_url, lang) if nsfw_response: blurred_html, nsfw_button_html = nsfw_response set_active_task(client_ip, False) # 解锁任务 # 返回None作为URL，防止用户通过"Use as Input"绕过NSFW检测 return blurred_html, None, t("warning_content_filter", lang), gr.update(value=nsfw_button_html, visible=True) # 正常情况：返回HTML显示图片 result_html = create_result_image_html(result_url) action_html = create_action_buttons_html(task_uuid, input_image_url, result_url, prompt, lang, show_like_tip) set_active_task(client_ip, False) # 解锁任务 return result_html, result_url, t("status_completed_message", lang).format(message=message), gr.update(value=action_html, visible=True) else: set_active_task(client_ip, False) # 解锁任务 return "", None, t("error_processing_failed", lang).format(message=message), gr.update(visible=False) except Exception as e: set_active_task(client_ip, False) # 解锁任务 return "", None, t("error_processing_exception", lang).format(error=str(e)), gr.update(visible=False) # ============================================================================= # 多图编辑接口 # ============================================================================= def multi_image_edit_interface(input_image1, input_image2, input_image3, prompt, aspect_ratio, lang, request: gr.Request, progress=gr.Progress()): """多图编辑接口""" try: client_ip = request.client.host x_forwarded_for = dict(request.headers).get('x-forwarded-for') if x_forwarded_for: client_ip = x_forwarded_for images = [img for img in [input_image1, input_image2, input_image3] if img is not None] if len(images) < 2: return "", t("error_multi_image_required", lang), gr.update(visible=False) if not prompt or prompt.strip() == "": return "", t("error_enter_prompt", lang), gr.update(visible=False) if len(prompt.strip()) <= 3: return "", t("error_prompt_too_short", lang), gr.update(visible=False) except Exception as e: return "", t("error_request_processing", lang), gr.update(visible=False) geo_info = util.IP_Country_Cache.get(client_ip, {"country": "Unknown"}) country = geo_info.get("country", "Unknown") country_config = rate_limit_config.get_country_config(country) current_phase = get_ip_phase(client_ip) current_count = get_ip_generation_count(client_ip) show_like_tip = rate_limit_config.should_show_like_tip(country, current_count) redirect_url = rate_limit_config.get_redirect_url(country, lang) # 检查是否有活动任务（3分钟内） is_active, remaining_seconds = has_active_task(client_ip) if is_active: return "", f"⏰ You have a task in progress. Please wait until it is finished before submitting a new task.", gr.update(value=create_task_locked_button_html(redirect_url, remaining_seconds), visible=True) # 根据用户次数和国家配置决定是否使用高速高清模式 high_speed_hd_count = country_config.get("high_speed_hd_count", rate_limit_config.HIGH_SPEED_HD_COUNT) use_high_speed_mode = current_count < high_speed_hd_count # 检查是否被封锁 if current_phase == 'blocked': return "", t("error_free_limit_reached", lang), gr.update(value=create_blocked_button_html(redirect_url), visible=True) if current_phase in ['rate_limit_1', 'rate_limit_2', 'rate_limit_3']: is_limited, wait_minutes, _ = check_rate_limit_for_phase(client_ip, current_phase) if is_limited: wait_minutes_int = int(wait_minutes) + 1 return "", t("error_free_limit_wait", lang).format(wait_minutes_int=wait_minutes_int), gr.update(value=create_rate_limit_button_html(redirect_url), visible=True) # NSFW检测 - 检测所有输入图片 is_nsfw_task = False for img in images: if img is not None: nsfw_detected, _ = check_nsfw_for_input(img, country, current_count, client_ip) if nsfw_detected: is_nsfw_task = True break # 决定任务参数（在progress_callback之前定义） if use_high_speed_mode: task_priority = 1 is_sr = 1 # 启用超分 mode_info = "🚀 High-Speed + HD Mode" else: task_priority = 0 is_sr = 0 # 关闭超分 mode_info = "⏱️ Low Resolution Mode | Visit https://omnicreator.net for High-Speed HD generation" def progress_callback(message): try: if progress is not None: progress(0.5, desc=f"{mode_info} | {message}") except Exception: pass try: # 设置任务锁定 set_active_task(client_ip, True) record_generation_attempt(client_ip, current_phase) input_url, result_url, message, task_uuid = process_multi_image_edit(images, prompt.strip(), progress_callback, priority=task_priority, client_ip=client_ip, is_sr=is_sr) if message and message.startswith("HF_LIMIT_EXCEEDED:"): set_active_task(client_ip, False) # 解锁任务 return "", t("error_free_limit_reached", lang), gr.update(value=create_blocked_button_html(redirect_url), visible=True) if result_url: # 检查结果图片NSFW if check_nsfw_for_result(result_url, country, current_count): is_nsfw_task = True # 应用模糊处理 if is_nsfw_task: nsfw_response = create_nsfw_blurred_response(result_url, redirect_url, lang) if nsfw_response: blurred_html, nsfw_button_html = nsfw_response set_active_task(client_ip, False) return blurred_html, t("warning_content_filter", lang), gr.update(value=nsfw_button_html, visible=True) result_html = create_result_image_html(result_url) action_html = create_action_buttons_html(task_uuid, input_url, result_url, prompt, lang, show_like_tip) # 添加模式提示信息 if use_high_speed_mode: mode_tip = "🚀 Generated with High-Speed + HD Mode" else: mode_tip = "⏱️ Generated with Low Resolution Mode | Visit https://omnicreator.net for High-Speed HD generation" status_message = f"{mode_tip}\n{t('status_completed_message', lang).format(message=message)}" set_active_task(client_ip, False) # 解锁任务 return result_html, status_message, gr.update(value=action_html, visible=True) else: set_active_task(client_ip, False) # 解锁任务 return "", t("error_processing_failed", lang).format(message=message), gr.update(visible=False) except Exception as e: set_active_task(client_ip, False) # 解锁任务 return "", t("error_processing_exception", lang).format(error=str(e)), gr.update(visible=False) # ============================================================================= # 文生图接口 # ============================================================================= def text_to_image_interface(prompt, aspect_ratio, lang, request: gr.Request, progress=gr.Progress()): """文生图接口""" try: client_ip = request.client.host x_forwarded_for = dict(request.headers).get('x-forwarded-for') if x_forwarded_for: client_ip = x_forwarded_for if not prompt or prompt.strip() == "": return "", t("error_enter_prompt", lang), gr.update(visible=False) if len(prompt.strip()) <= 3: return "", t("error_prompt_too_short", lang), gr.update(visible=False) except Exception as e: return "", t("error_request_processing", lang), gr.update(visible=False) geo_info = util.IP_Country_Cache.get(client_ip, {"country": "Unknown"}) country = geo_info.get("country", "Unknown") country_config = rate_limit_config.get_country_config(country) current_phase = get_ip_phase(client_ip) current_count = get_ip_generation_count(client_ip) show_like_tip = rate_limit_config.should_show_like_tip(country, current_count) redirect_url = rate_limit_config.get_redirect_url(country, lang) # 检查是否有活动任务（3分钟内） is_active, remaining_seconds = has_active_task(client_ip) if is_active: return "", f"⏰ You have a task in progress. Please wait until it is finished before submitting a new task.", gr.update(value=create_task_locked_button_html(redirect_url, remaining_seconds), visible=True) # 根据用户次数和国家配置决定是否使用高速高清模式 high_speed_hd_count = country_config.get("high_speed_hd_count", rate_limit_config.HIGH_SPEED_HD_COUNT) use_high_speed_mode = current_count < high_speed_hd_count # 检查是否被封锁 if current_phase == 'blocked': return "", t("error_free_limit_reached", lang), gr.update(value=create_blocked_button_html(redirect_url), visible=True) if current_phase in ['rate_limit_1', 'rate_limit_2', 'rate_limit_3']: is_limited, wait_minutes, _ = check_rate_limit_for_phase(client_ip, current_phase) if is_limited: wait_minutes_int = int(wait_minutes) + 1 return "", t("error_free_limit_wait", lang).format(wait_minutes_int=wait_minutes_int), gr.update(value=create_rate_limit_button_html(redirect_url), visible=True) # 决定任务参数（在progress_callback之前定义，以便在回调中使用） if use_high_speed_mode: task_priority = 1 is_sr = 1 # 启用超分 target_area = 800 * 800 # 高清分辨率 mode_info = "🚀 High-Speed + HD Mode (800×800)" else: task_priority = 0 is_sr = 0 # 关闭超分 target_area = 576 * 576 # 标准分辨率 mode_info = "⏱️ Low Resolution Mode (576×576) | Visit https://omnicreator.net for High-Speed HD generation" def progress_callback(message): try: if progress is not None: # 添加模式提示 progress(0.5, desc=f"{mode_info} | {message}") except Exception: pass try: # 设置任务锁定 set_active_task(client_ip, True) record_generation_attempt(client_ip, current_phase) result_url, message, task_uuid = process_text_to_image(prompt.strip(), aspect_ratio, progress_callback, priority=task_priority, client_ip=client_ip, is_sr=is_sr, target_area=target_area) if message and message.startswith("HF_LIMIT_EXCEEDED:"): set_active_task(client_ip, False) return "", t("error_free_limit_reached", lang), gr.update(value=create_blocked_button_html(redirect_url), visible=True) if result_url: # 检查结果图片NSFW is_nsfw_task = check_nsfw_for_result(result_url, country, current_count) # 应用模糊处理 if is_nsfw_task: nsfw_response = create_nsfw_blurred_response(result_url, redirect_url, lang) if nsfw_response: blurred_html, nsfw_button_html = nsfw_response set_active_task(client_ip, False) return blurred_html, t("warning_content_filter", lang), gr.update(value=nsfw_button_html, visible=True) result_html = create_result_image_html(result_url) # 为T2I生成3个操作按钮 action_html = "" if task_uuid and lang not in ["zh", "hi", "ru"]: task_detail_url = f"https://omnicreator.net/my-creations/task/{task_uuid}" from urllib.parse import quote encoded_prompt = quote(prompt.strip()) if prompt else "" encoded_result_url = quote(result_url) if result_url else "" i2v_url = f"https://omnicreator.net/image-to-video?input_image={encoded_result_url}&prompt={encoded_prompt}" face_swap_url = f"https://omnicreator.net/face-swap?user_image={encoded_result_url}" action_html = f"""
🎥 Convert to Video 🧍 Swap Face 💾 Download HD Image
""" if show_like_tip or lang in ["zh", "hi", "ru"]: action_html += create_like_tip_html() # 添加模式提示信息 if use_high_speed_mode: mode_tip = "🚀 Generated with High-Speed + HD Mode (800×800)" else: mode_tip = "⏱️ Generated with Low Resolution Mode (576×576) | Visit https://omnicreator.net for High-Speed HD generation" status_message = f"{mode_tip}\n{t('status_completed_message', lang).format(message=message)}" set_active_task(client_ip, False) # 解锁任务 return result_html, status_message, gr.update(value=action_html, visible=True) else: set_active_task(client_ip, False) # 解锁任务 return "", t("error_processing_failed", lang).format(message=message), gr.update(visible=False) except Exception as e: set_active_task(client_ip, False) # 解锁任务 return "", t("error_processing_exception", lang).format(error=str(e)), gr.update(visible=False) # ============================================================================= # 图片放大接口 # ============================================================================= def image_upscale_interface(input_image, lang, request: gr.Request, progress=gr.Progress()): """图片放大接口""" try: client_ip = request.client.host x_forwarded_for = dict(request.headers).get('x-forwarded-for') if x_forwarded_for: client_ip = x_forwarded_for if input_image is None: return "", t("error_upload_first", lang), gr.update(visible=False) except Exception as e: return "", t("error_request_processing", lang), gr.update(visible=False) geo_info = util.IP_Country_Cache.get(client_ip, {"country": "Unknown"}) country = geo_info.get("country", "Unknown") current_phase = get_ip_phase(client_ip) current_count = get_ip_generation_count(client_ip) show_like_tip = rate_limit_config.should_show_like_tip(country, current_count) redirect_url = rate_limit_config.get_redirect_url(country, lang) # 检查是否有活动任务（3分钟内） is_active, remaining_seconds = has_active_task(client_ip) if is_active: return "", f"⏰ You have a task in progress. Please wait until it is finished before submitting a new task.", gr.update(value=create_task_locked_button_html(redirect_url, remaining_seconds), visible=True) # 检查是否被封锁 if current_phase == 'blocked': return "", t("error_free_limit_reached", lang), gr.update(value=create_blocked_button_html(redirect_url), visible=True) if current_phase in ['rate_limit_1', 'rate_limit_2', 'rate_limit_3']: is_limited, wait_minutes, _ = check_rate_limit_for_phase(client_ip, current_phase) if is_limited: wait_minutes_int = int(wait_minutes) + 1 return "", t("error_free_limit_wait", lang).format(wait_minutes_int=wait_minutes_int), gr.update(value=create_rate_limit_button_html(redirect_url), visible=True) # NSFW检测 is_nsfw_task, _ = check_nsfw_for_input(input_image, country, current_count, client_ip) def progress_callback(message): try: if progress is not None: progress(0.5, desc=message) except Exception: pass try: # 设置任务锁定 set_active_task(client_ip, True) task_priority = 1 if current_count < rate_limit_config.HIGH_PRIORITY_COUNT else 0 record_generation_attempt(client_ip, current_phase) input_url, result_url, message, task_uuid = process_image_upscale(input_image, progress_callback, priority=task_priority, client_ip=client_ip) if message and message.startswith("HF_LIMIT_EXCEEDED:"): set_active_task(client_ip, False) return "", t("error_free_limit_reached", lang), gr.update(value=create_blocked_button_html(redirect_url), visible=True) if result_url: # 检查结果图片NSFW if check_nsfw_for_result(result_url, country, current_count): is_nsfw_task = True # 应用模糊处理 if is_nsfw_task: nsfw_response = create_nsfw_blurred_response(result_url, redirect_url, lang) if nsfw_response: blurred_html, nsfw_button_html = nsfw_response set_active_task(client_ip, False) return blurred_html, t("warning_content_filter", lang), gr.update(value=nsfw_button_html, visible=True) result_html = create_result_image_html(result_url) action_html = "" if task_uuid and lang not in ["zh", "hi", "ru"]: task_detail_url = f"https://omnicreator.net/my-creations/task/{task_uuid}" action_html = f"""
💾 Download HD Image
""" if show_like_tip or lang in ["zh", "hi", "ru"]: action_html += create_like_tip_html() set_active_task(client_ip, False) # 解锁任务 return result_html, t("status_completed_message", lang).format(message=message), gr.update(value=action_html, visible=bool(action_html)) else: set_active_task(client_ip, False) # 解锁任务 return "", t("error_processing_failed", lang).format(message=message), gr.update(visible=False) except Exception as e: set_active_task(client_ip, False) # 解锁任务 return "", t("error_processing_exception", lang).format(error=str(e)), gr.update(visible=False) # ============================================================================= # 换脸接口 # ============================================================================= def face_swap_interface(target_editor, source_face_image, lang, request: gr.Request, progress=gr.Progress()): """换脸接口 - 支持从ImageEditor提取mask""" try: client_ip = request.client.host x_forwarded_for = dict(request.headers).get('x-forwarded-for') if x_forwarded_for: client_ip = x_forwarded_for # 从ImageEditor提取图片和mask target_image = None mask_image = None if target_editor is None: return "", t("error_upload_target_image", lang), gr.update(visible=False) # ImageEditor返回字典格式: {"background": PIL, "layers": [PIL], "composite": PIL} if isinstance(target_editor, dict): target_image = target_editor.get("background") layers = target_editor.get("layers", []) # 从layers提取mask if layers and len(layers) > 0: from util import create_mask_from_layers mask_image = create_mask_from_layers(target_image, layers) else: # 兼容直接传入PIL Image的情况 target_image = target_editor if target_image is None: return "", t("error_upload_target_image", lang), gr.update(visible=False) if source_face_image is None: return "", t("error_upload_source_face", lang), gr.update(visible=False) except Exception as e: return "", t("error_request_processing", lang), gr.update(visible=False) geo_info = util.IP_Country_Cache.get(client_ip, {"country": "Unknown"}) country = geo_info.get("country", "Unknown") current_phase = get_ip_phase(client_ip) current_count = get_ip_generation_count(client_ip) show_like_tip = rate_limit_config.should_show_like_tip(country, current_count) redirect_url = rate_limit_config.get_redirect_url(country, lang) # 检查活动任务 is_active, remaining_seconds = has_active_task(client_ip) if is_active: return "", f"⏰ Task in progress. Wait {int(remaining_seconds/60)+1} min.", gr.update(value=create_task_locked_button_html(redirect_url, remaining_seconds), visible=True) # 检查是否被封锁 if current_phase == 'blocked': return "", t("error_free_limit_reached", lang), gr.update(value=create_blocked_button_html(redirect_url), visible=True) if current_phase in ['rate_limit_1', 'rate_limit_2', 'rate_limit_3']: is_limited, wait_minutes, _ = check_rate_limit_for_phase(client_ip, current_phase) if is_limited: return "", t("error_free_limit_wait", lang).format(wait_minutes_int=int(wait_minutes)+1), gr.update(value=create_rate_limit_button_html(redirect_url), visible=True) # NSFW检测 - 检测target和source图片 is_nsfw_task, _ = check_nsfw_for_input(target_image, country, current_count, client_ip) if not is_nsfw_task: is_nsfw_task, _ = check_nsfw_for_input(source_face_image, country, current_count, client_ip) def progress_callback(message): try: if progress is not None: progress(0.5, desc=message) except Exception: pass try: set_active_task(client_ip, True) task_priority = 1 if current_count < rate_limit_config.HIGH_PRIORITY_COUNT else 0 record_generation_attempt(client_ip, current_phase) input_url, result_url, message, task_uuid = process_face_swap(target_image, source_face_image, mask_image, progress_callback, priority=task_priority, client_ip=client_ip) if message and message.startswith("HF_LIMIT_EXCEEDED:"): set_active_task(client_ip, False) return "", t("error_free_limit_reached", lang), gr.update(value=create_blocked_button_html(redirect_url), visible=True) if result_url: # 检查结果图片NSFW if check_nsfw_for_result(result_url, country, current_count): is_nsfw_task = True # 应用模糊处理 if is_nsfw_task: nsfw_response = create_nsfw_blurred_response(result_url, redirect_url, lang) if nsfw_response: blurred_html, nsfw_button_html = nsfw_response set_active_task(client_ip, False) return blurred_html, t("warning_content_filter", lang), gr.update(value=nsfw_button_html, visible=True) result_html = create_result_image_html(result_url) action_html = "" if task_uuid and lang not in ["zh", "hi", "ru"]: task_detail_url = f"https://omnicreator.net/my-creations/task/{task_uuid}" action_html = f"""
💾 Download HD Image
""" if show_like_tip or lang in ["zh", "hi", "ru"]: action_html += create_like_tip_html() set_active_task(client_ip, False) return result_html, t("status_completed_message", lang).format(message=message), gr.update(value=action_html, visible=bool(action_html)) else: set_active_task(client_ip, False) return "", t("error_processing_failed", lang).format(message=message), gr.update(visible=False) except Exception as e: set_active_task(client_ip, False) return "", t("error_processing_exception", lang).format(error=str(e)), gr.update(visible=False) # ============================================================================= # 去水印接口 # ============================================================================= def watermark_removal_interface(input_image, force_removal, lang, request: gr.Request, progress=gr.Progress()): """去水印接口 - 支持强力模式""" try: client_ip = request.client.host x_forwarded_for = dict(request.headers).get('x-forwarded-for') if x_forwarded_for: client_ip = x_forwarded_for if input_image is None: return "", t("error_upload_image", lang), gr.update(visible=False) except Exception as e: return "", t("error_request_processing", lang), gr.update(visible=False) geo_info = util.IP_Country_Cache.get(client_ip, {"country": "Unknown"}) country = geo_info.get("country", "Unknown") current_phase = get_ip_phase(client_ip) current_count = get_ip_generation_count(client_ip) show_like_tip = rate_limit_config.should_show_like_tip(country, current_count) redirect_url = rate_limit_config.get_redirect_url(country, lang) # 检查活动任务 is_active, remaining_seconds = has_active_task(client_ip) if is_active: return "", f"⏰ Task in progress. Please wait until it is finished before submitting a new task.", gr.update(value=create_task_locked_button_html(redirect_url, remaining_seconds), visible=True) # 检查是否被封锁 if current_phase == 'blocked': return "", t("error_free_limit_reached", lang), gr.update(value=create_blocked_button_html(redirect_url), visible=True) if current_phase in ['rate_limit_1', 'rate_limit_2', 'rate_limit_3']: is_limited, wait_minutes, _ = check_rate_limit_for_phase(client_ip, current_phase) if is_limited: return "", t("error_free_limit_wait", lang).format(wait_minutes_int=int(wait_minutes)+1), gr.update(value=create_rate_limit_button_html(redirect_url), visible=True) # NSFW检测 is_nsfw_task, _ = check_nsfw_for_input(input_image, country, current_count, client_ip) def progress_callback(message): try: if progress is not None: mode_text = "💪 AI Force Mode | " if force_removal else "" progress(0.5, desc=f"{mode_text}🧽 Removing watermark... | {message}") except Exception: pass try: set_active_task(client_ip, True) task_priority = 1 if current_count < rate_limit_config.HIGH_PRIORITY_COUNT else 0 record_generation_attempt(client_ip, current_phase) input_url, result_url, message, task_uuid = process_watermark_removal(input_image, progress_callback, priority=task_priority, client_ip=client_ip, force_removal=force_removal) if message and message.startswith("HF_LIMIT_EXCEEDED:"): set_active_task(client_ip, False) return "", t("error_free_limit_reached", lang), gr.update(value=create_blocked_button_html(redirect_url), visible=True) if result_url: # 检查结果图片NSFW if check_nsfw_for_result(result_url, country, current_count): is_nsfw_task = True # 应用模糊处理 if is_nsfw_task: nsfw_response = create_nsfw_blurred_response(result_url, redirect_url, lang) if nsfw_response: blurred_html, nsfw_button_html = nsfw_response set_active_task(client_ip, False) return blurred_html, t("warning_content_filter", lang), gr.update(value=nsfw_button_html, visible=True) result_html = create_result_image_html(result_url) action_html = "" if task_uuid and lang not in ["zh", "hi", "ru"]: task_detail_url = f"https://omnicreator.net/my-creations/task/{task_uuid}" action_html = f"""
💾 Download HD Image 🚀 Unlimited Removal
""" if show_like_tip or lang in ["zh", "hi", "ru"]: action_html += create_like_tip_html() set_active_task(client_ip, False) return result_html, t("status_completed_message", lang).format(message=message), gr.update(value=action_html, visible=bool(action_html)) else: set_active_task(client_ip, False) return "", t("error_processing_failed", lang).format(message=message), gr.update(visible=False) except Exception as e: set_active_task(client_ip, False) return "", t("error_processing_exception", lang).format(error=str(e)), gr.update(visible=False) # ============================================================================= # 创建Gradio应用 # ============================================================================= def create_app(): with gr.Blocks( title="AI Image Editor", theme=gr.themes.Soft(), css=""" .main-container { max-width: 1200px; margin: 0 auto; } .news-banner-row { margin: 10px auto 15px auto; padding: 0 10px; max-width: 1200px; width: 100% !important; } .banner-lang-selector { margin-left: auto !important; display: flex !important; justify-content: flex-end !important; } .upload-area { border: 2px dashed #ccc; border-radius: 10px; padding: 20px; text-align: center; } .result-area { margin-top: 20px; padding: 20px; border-radius: 10px; background-color: #f8f9fa; } """ ) as app: lang_state = gr.State("en") # 主标题 header_title = gr.HTML(f"""

{t('header_title', 'en')}

""") with gr.Row(elem_classes=["news-banner-row"]): with gr.Column(scale=1, min_width=400): news_banner = gr.HTML(f"""
🎉 Please give us a ❤️ if you think it's helpful
""", visible=True) with gr.Column(scale=0, min_width=160, elem_classes=["banner-lang-selector"]): lang_dropdown = gr.Dropdown( choices=[("English", "en"), ("中文", "zh")], value="en", label="🌐", show_label=True, interactive=True, container=False ) # ============================================================================= # 多Tab界面 # ============================================================================= with gr.Tabs() as tabs: # ============================================================================= # Tab 1: 单图编辑 # ============================================================================= with gr.Tab(t("tab_single_image_edit", "en")) as single_edit_tab: with gr.Row(): with gr.Column(scale=1): gr.Markdown(t("upload_image_header", "en")) single_input_image = gr.Image(label=t("upload_image_label", "en"), type="pil", height=512, elem_classes=["upload-area"]) gr.Markdown(t("editing_instructions_header", "en")) single_prompt = gr.Textbox(label=t("prompt_input_label", "en"), placeholder=t("prompt_input_placeholder", "en"), lines=3) single_edit_btn = gr.Button(t("start_editing_button", "en"), variant="primary", size="lg") with gr.Column(scale=1): gr.Markdown(t("editing_result_header", "en")) single_output_image = gr.HTML(label=t("output_image_label", "en"), elem_classes=["result-area"]) single_output_url = gr.State() # 隐藏state保存URL用于Use as Input single_use_as_input_btn = gr.Button(t("use_as_input_button", "en"), variant="secondary", size="sm") single_status = gr.Textbox(label=t("status_output_label", "en"), lines=2, interactive=False) single_action_buttons = gr.HTML(visible=False) gr.Markdown(t("prompt_examples_header", "en")) with gr.Row(): for prompt in ["Change background to beach sunset", "Add rainbow in the sky", "Make it look like oil painting", "Remove the background", "Change outfit to red dress"]: gr.Button(prompt, size="sm").click(lambda p=prompt: p, outputs=single_prompt) single_edit_btn.click( fn=edit_image_interface, inputs=[single_input_image, single_prompt, lang_state], outputs=[single_output_image, single_output_url, single_status, single_action_buttons], show_progress=True, concurrency_limit=20 ) def use_result_as_input(url): if url: try: import requests from PIL import Image import io response = requests.get(url, timeout=10) if response.status_code == 200: return Image.open(io.BytesIO(response.content)) except: pass return None single_use_as_input_btn.click( fn=use_result_as_input, inputs=[single_output_url], outputs=[single_input_image] ) # ============================================================================= # Tab 2: 多图编辑 # ============================================================================= with gr.Tab(t("tab_multi_image_edit", "en")) as multi_edit_tab: with gr.Row(): with gr.Column(scale=1): gr.Markdown(t("multi_image_upload_header", "en")) # 3个图片组件放在同一排 with gr.Row(): multi_input_image1 = gr.Image(label=t("multi_image1_label", "en"), type="pil", height=200, elem_classes=["upload-area"]) multi_input_image2 = gr.Image(label=t("multi_image2_label", "en"), type="pil", height=200, elem_classes=["upload-area"]) multi_input_image3 = gr.Image(label=t("multi_image3_label", "en"), type="pil", height=200, elem_classes=["upload-area"]) gr.Markdown("📐 Output Size") multi_aspect_ratio = gr.Radio( choices=["Auto", "16:9", "4:3", "1:1", "3:4", "9:16"], value="Auto", label="Select aspect ratio", info="Choose output dimensions or Auto for original sizing" ) gr.Markdown(t("multi_prompt_header", "en")) multi_prompt = gr.Textbox(label=t("multi_prompt_label", "en"), placeholder=t("multi_prompt_placeholder", "en"), lines=3) multi_edit_btn = gr.Button(t("multi_edit_button", "en"), variant="primary", size="lg") with gr.Column(scale=1): gr.Markdown(t("editing_result_header", "en")) multi_output_image = gr.HTML(label=t("output_image_label", "en"), elem_classes=["result-area"]) multi_status = gr.Textbox(label=t("status_output_label", "en"), lines=2, interactive=False) multi_action_buttons = gr.HTML(visible=False) gr.Markdown(t("multi_examples_header", "en")) with gr.Row(): for prompt in ["Merge these two animals into one image", "Combine the elements from both images", "Create a scene with objects from all images"]: gr.Button(prompt, size="sm").click(lambda p=prompt: p, outputs=multi_prompt) multi_edit_btn.click( fn=multi_image_edit_interface, inputs=[multi_input_image1, multi_input_image2, multi_input_image3, multi_prompt, multi_aspect_ratio, lang_state], outputs=[multi_output_image, multi_status, multi_action_buttons], show_progress=True, concurrency_limit=20 ) # ============================================================================= # Tab 3: 文生图 # ============================================================================= with gr.Tab(t("tab_text_to_image", "en")) as t2i_tab: with gr.Row(): with gr.Column(scale=1): gr.Markdown(t("t2i_prompt_header", "en")) t2i_prompt = gr.Textbox(label=t("t2i_prompt_label", "en"), placeholder=t("t2i_prompt_placeholder", "en"), lines=5) gr.Markdown("📐 Output Size") t2i_aspect_ratio = gr.Radio( choices=["16:9", "4:3", "1:1", "3:4", "9:16"], value="1:1", label="Select aspect ratio", info="Choose output dimensions for generated image" ) t2i_btn = gr.Button(t("t2i_generate_button", "en"), variant="primary", size="lg") with gr.Column(scale=1): gr.Markdown(t("t2i_result_header", "en")) t2i_output_image = gr.HTML(label=t("t2i_output_label", "en"), elem_classes=["result-area"]) t2i_output_url = gr.State() # 保存URL用于生成按钮链接 t2i_status = gr.Textbox(label=t("status_output_label", "en"), lines=2, interactive=False) t2i_action_buttons = gr.HTML(visible=False) gr.Markdown(t("t2i_examples_header", "en")) with gr.Row(): for prompt in ["A beautiful young woman standing by the seaside at sunset", "A majestic dragon flying over a medieval castle", "A cute cat wearing a wizard hat, digital art", "Futuristic city skyline at night with neon lights"]: gr.Button(prompt, size="sm").click(lambda p=prompt: p, outputs=t2i_prompt) t2i_btn.click( fn=text_to_image_interface, inputs=[t2i_prompt, t2i_aspect_ratio, lang_state], outputs=[t2i_output_image, t2i_status, t2i_action_buttons], show_progress=True, concurrency_limit=20 ) # ============================================================================= # Tab 4: 图片放大 # ============================================================================= with gr.Tab(t("tab_image_upscale", "en")) as upscale_tab: with gr.Row(): with gr.Column(scale=1): gr.Markdown(t("upscale_upload_header", "en")) upscale_input_image = gr.Image(label=t("upscale_input_label", "en"), type="pil", height=400, elem_classes=["upload-area"]) upscale_btn = gr.Button(t("upscale_button", "en"), variant="primary", size="lg") with gr.Column(scale=1): gr.Markdown(t("upscale_result_header", "en")) upscale_output_image = gr.HTML(label=t("upscale_output_label", "en"), elem_classes=["result-area"]) upscale_status = gr.Textbox(label=t("status_output_label", "en"), lines=2, interactive=False) upscale_action_buttons = gr.HTML(visible=False) upscale_btn.click( fn=image_upscale_interface, inputs=[upscale_input_image, lang_state], outputs=[upscale_output_image, upscale_status, upscale_action_buttons], show_progress=True, concurrency_limit=20 ) # ============================================================================= # Tab 5: 换脸 # ============================================================================= with gr.Tab(t("tab_face_swap", "en")) as faceswap_tab: with gr.Row(): with gr.Column(scale=1): gr.Markdown(t("faceswap_target_header", "en")) gr.Markdown("*If multiple faces exist, draw on the face you want to replace*", elem_classes=["hint-text"]) faceswap_target_editor = gr.ImageEditor( label=t("faceswap_target_label", "en"), type="pil", height=350, brush=gr.Brush(colors=["#FFFFFF"], default_size=80, color_mode="fixed"), eraser=gr.Eraser(default_size=40), layers=False, sources=["upload", "clipboard"] ) gr.Markdown(t("faceswap_source_header", "en")) faceswap_source = gr.Image(label=t("faceswap_source_label", "en"), type="pil", height=256) faceswap_btn = gr.Button(t("faceswap_button", "en"), variant="primary", size="lg") with gr.Column(scale=1): gr.Markdown(t("faceswap_result_header", "en")) faceswap_output = gr.HTML(label=t("faceswap_output_label", "en"), elem_classes=["result-area"]) faceswap_status = gr.Textbox(label=t("status_output_label", "en"), lines=2, interactive=False) faceswap_action_buttons = gr.HTML(visible=False) faceswap_btn.click( fn=face_swap_interface, inputs=[faceswap_target_editor, faceswap_source, lang_state], outputs=[faceswap_output, faceswap_status, faceswap_action_buttons], show_progress=True, concurrency_limit=20 ) # ============================================================================= # Tab 6: 去水印 # ============================================================================= with gr.Tab("🧽 Remove Watermark") as watermark_tab: with gr.Row(): with gr.Column(scale=1): gr.Markdown("🖼️ Upload Image with Watermark") watermark_input = gr.Image(label="Select image to remove watermark", type="pil", height=400, elem_classes=["upload-area"]) watermark_force_removal = gr.Checkbox( label="💪 Force Removal", info="Ultimate AI-powered watermark removal. AI analyzes the watermark for better results.", value=False ) watermark_btn = gr.Button("🧽 Remove Watermark", variant="primary", size="lg") with gr.Column(scale=1): gr.Markdown("🎯 Result") watermark_output = gr.HTML(label="Image without watermark", elem_classes=["result-area"]) watermark_use_as_input_btn = gr.Button("🔄 Use as Input (Try Again)", variant="secondary", size="sm") watermark_status = gr.Textbox(label="Processing status", lines=2, interactive=False) watermark_action_buttons = gr.HTML(visible=False) watermark_btn.click( fn=watermark_removal_interface, inputs=[watermark_input, watermark_force_removal, lang_state], outputs=[watermark_output, watermark_status, watermark_action_buttons], show_progress=True, concurrency_limit=20 ) # 定义一个函数来实现"使用结果作为输入"功能 def watermark_use_result(output_html): # 从HTML中提取图片URL并下载 if output_html and 'src=' in str(output_html): import re match = re.search(r"src='([^']+)'", str(output_html)) if match: url = match.group(1) try: import requests from PIL import Image import io response = requests.get(url, timeout=10) if response.status_code == 200: return Image.open(io.BytesIO(response.content)) except: pass return None watermark_use_as_input_btn.click( fn=watermark_use_result, inputs=[watermark_output], outputs=[watermark_input] ) # ============================================================================= # SEO内容区域 # ============================================================================= seo_html = gr.HTML() def get_seo_html(lang): if lang in ["zh", "hi", "ru"]: return "" return f"""

🎨 {t('seo_unlimited_title', lang)}

{t('seo_unlimited_desc', lang)}

🚀 {t('seo_unlimited_button', lang)}

🤖 About Omni Creator 2.0 - 8B Multi-Modal Generation Model

OmniAI is an 8-billion parameter multi-modal generation model developed by the OmniCreator research team. Through extensive training on massive video datasets, we discovered that the network can simultaneously acquire text, image, and video editing and generation capabilities.

That's why we call it the Omni Creator - it handles everything from text-to-image, image editing, to video generation with a single unified architecture.

⚡ Performance Highlights

🏁 Image Generation

896×896 in under 10 seconds on RTX 5090 (32GB VRAM)

🎥 Video Generation

3-second 320p video in 30 seconds

💡 Knowledge Distillation

Real-time inference performance achieved

🎨 What Omni Model Can Do

This Space demonstrates just a small fraction of what the Omni model can do. Beyond what you see here, it can also perform:

👗 Virtual Try-On 🌅 Background Replacement 💇 Hairstyle Changer 📄 Poster Editing 🎨 Style Transfer + dozens more...

No fine-tuning required - just modify the prompt and input parameters!

""" # 收集所有需要更新的UI组件 all_ui_components = [ header_title, news_banner, single_edit_tab, multi_edit_tab, t2i_tab, upscale_tab, faceswap_tab, watermark_tab, seo_html, ] def update_ui_lang(lang): show_banner = lang not in ["zh", "hi", "ru"] return { header_title: gr.update(value=f"""

{t('header_title', lang)}

"""), news_banner: gr.update(visible=show_banner), single_edit_tab: gr.update(label=t("tab_single_image_edit", lang)), multi_edit_tab: gr.update(label=t("tab_multi_image_edit", lang)), t2i_tab: gr.update(label=t("tab_text_to_image", lang)), upscale_tab: gr.update(label=t("tab_image_upscale", lang)), faceswap_tab: gr.update(label=t("tab_face_swap", lang)), watermark_tab: gr.update(label="🧽 Remove Watermark"), seo_html: gr.update(value=get_seo_html(lang)), } def on_lang_change(lang): return lang, *update_ui_lang(lang).values() lang_dropdown.change( on_lang_change, inputs=[lang_dropdown], outputs=[lang_state] + all_ui_components ) ip_query_state = gr.State({"status": "pending", "ip": None, "lang": "en"}) def on_load_immediate(request: gr.Request): client_ip = None try: client_ip = request.client.host headers = dict(request.headers) if hasattr(request, 'headers') else {} x_forwarded_for = headers.get('x-forwarded-for') or headers.get('X-Forwarded-For') if x_forwarded_for: client_ip = x_forwarded_for.split(',')[0].strip() except Exception: client_ip = "unknown" try: if client_ip in util.IP_Country_Cache: cached_lang = get_lang_from_ip(client_ip) query_state = {"ip": client_ip, "cached": True} return cached_lang, cached_lang, query_state, *update_ui_lang(cached_lang).values() detected_lang = get_lang_from_ip(client_ip) query_state = {"ip": client_ip, "cached": False} return detected_lang, detected_lang, query_state, *update_ui_lang(detected_lang).values() except Exception: query_state = {"ip": client_ip or "unknown", "cached": False} return "en", "en", query_state, *update_ui_lang("en").values() app.load( on_load_immediate, inputs=None, outputs=[lang_state, lang_dropdown, ip_query_state] + all_ui_components, ) return app if __name__ == "__main__": app = create_app() app.queue( default_concurrency_limit=20, max_size=50, api_open=False ) app.launch( server_name="0.0.0.0", show_error=True, quiet=False, max_threads=40, height=800, favicon_path=None )