import gradio as gr import threading import os import shutil import tempfile import time from util import process_image_edit, OPENROUTER_API_KEY # Configuration parameters for watermark removal FREE_TRY_N = 15 # Free phase: first 8 tries without restrictions SLOW_TRY_N = 20 # Slow phase start: 20 tries RATE_LIMIT_20 = 25 # Full restriction: blocked after 20 tries # Time window configuration (minutes) PHASE_1_WINDOW = 5 # 8-20 tries: 5 minutes MAX_IMAGES_PER_WINDOW = 2 # Max images per time window IP_Dict = {} # IP generation statistics and time window tracking IP_Generation_Count = {} # Record total generation count for each IP IP_Rate_Limit_Track = {} # Record generation count and timestamp in current time window for each IP def get_ip_generation_count(client_ip): """ Get IP generation count """ if client_ip not in IP_Generation_Count: IP_Generation_Count[client_ip] = 0 return IP_Generation_Count[client_ip] def increment_ip_generation_count(client_ip): """ Increment IP generation count """ if client_ip not in IP_Generation_Count: IP_Generation_Count[client_ip] = 0 IP_Generation_Count[client_ip] += 1 return IP_Generation_Count[client_ip] def get_ip_phase(client_ip): """ Get current phase for IP Returns: str: 'free', 'rate_limit_1', 'blocked' """ count = get_ip_generation_count(client_ip) if count < FREE_TRY_N: # 0-7 tries return 'free' elif count < SLOW_TRY_N: # 8-19 tries return 'rate_limit_1' # 5 minutes 2 images else: # 20+ tries return 'blocked' # Generation blocked def check_rate_limit_for_phase(client_ip, phase): """ Check rate limit for specific phase Returns: tuple: (is_limited, wait_time_minutes, current_count) """ if phase not in ['rate_limit_1']: return False, 0, 0 # Determine time window window_minutes = PHASE_1_WINDOW # 5 minutes current_time = time.time() window_key = f"{client_ip}_{phase}" # Clean expired records if window_key in IP_Rate_Limit_Track: track_data = IP_Rate_Limit_Track[window_key] # Check if within current time window if current_time - track_data['start_time'] > window_minutes * 60: # Time window expired, reset IP_Rate_Limit_Track[window_key] = { 'count': 0, 'start_time': current_time, 'last_generation': current_time } else: # Initialize IP_Rate_Limit_Track[window_key] = { 'count': 0, 'start_time': current_time, 'last_generation': current_time } track_data = IP_Rate_Limit_Track[window_key] # Check if exceeded limit if track_data['count'] >= MAX_IMAGES_PER_WINDOW: # Calculate remaining wait time 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 record_generation_attempt(client_ip, phase): """ Record generation attempt """ # Increment total count increment_ip_generation_count(client_ip) # Record time window count if phase in ['rate_limit_1']: window_key = f"{client_ip}_{phase}" current_time = time.time() if window_key in IP_Rate_Limit_Track: IP_Rate_Limit_Track[window_key]['count'] += 1 IP_Rate_Limit_Track[window_key]['last_generation'] = current_time else: IP_Rate_Limit_Track[window_key] = { 'count': 1, 'start_time': current_time, 'last_generation': current_time } # Load sample images from samples directory def load_sample_images(): """Load sample images from samples directory""" samples_dir = "samples" sample_images = [] if os.path.exists(samples_dir): for filename in os.listdir(samples_dir): if filename.lower().endswith(('.png', '.jpg', '.jpeg', '.gif', '.bmp')): if "rm" in filename: continue file_path = os.path.join(samples_dir, filename) sample_images.append((file_path, filename)) return sample_images # Pre-processed results for sample images (local files) SAMPLE_RESULTS = { "sample01.jpeg": "samples/sample01_rm.jpeg", "sample02.jpeg": "samples/sample02_rm.jpeg", "1.jpg": "samples/1_rm.jpg", "4.jpg": "samples/4_rm.jpg", "10.jpg": "samples/10_rm.jpg" } def generate_smart_watermark_prompt(image_url): """Generate smart watermark removal prompt using OpenRouter API""" import requests import json try: api_key = OPENROUTER_API_KEY if 'OPENROUTER_API_KEY' in globals() else "" if not api_key: print("Warning: OPENROUTER_API_KEY not found") return "remove watermark in the image" headers = { 'Content-Type': 'application/json', 'Authorization': f'Bearer {api_key}' } data = { "model": "google/gemini-2.5-flash-lite", "messages": [ {"role": "user", "content": "现在用户输入了一张带水印的图片，需要生成修复prompt来引导图像生成模型去除水印。你需要找到水印，然后输出去除水印的英文prompt，指出水印的位置和特点，让模型知道即可，尽量简短。直接输出prompt即可。举例1: 'Remove the blue watermark in the lower right corner of the document'. 举例2: 'Remove the white watermark over the whole picture'"}, {"role": "assistant", "content": "好的"}, { "role": "user", "content": [ {"type": "text", "text": "Generate an English prompt to remove the watermark in this image."}, {"type": "image_url", "image_url": {"url": image_url}} ] } ] } response = requests.post( 'https://openrouter.ai/api/v1/chat/completions', headers=headers, json=data, timeout=30 ) if response.status_code == 200: result = response.json() if result.get("choices") and len(result["choices"]) > 0: prompt = result["choices"][0]["message"]["content"].strip() print(f"✅ Generated smart prompt: {prompt}") return prompt print(f"⚠️ OpenRouter API failed with status {response.status_code}") return "remove watermark in the image" except Exception as e: print(f"⚠️ Error generating smart prompt: {e}") return "remove watermark in the image" def remove_watermark_interface(input_image, force_removal, request: gr.Request, progress=gr.Progress()): """ Interface function for watermark removal with phase-based limitations """ try: # Extract user IP 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 IP_Dict: IP_Dict[client_ip] = 0 IP_Dict[client_ip] += 1 if input_image is None: return None, "Please upload an image first", gr.update(visible=False) # Set initial prompt for watermark removal prompt = "remove watermark in the image" except Exception as e: print(f"⚠️ Request preprocessing error: {e}") return None, "❌ Request processing error", gr.update(visible=False) # Get user current phase current_phase = get_ip_phase(client_ip) current_count = get_ip_generation_count(client_ip) print(f"📊 User phase info - IP: {client_ip}, current phase: {current_phase}, generation count: {current_count}") # Check if completely blocked if current_phase == 'blocked': # Generate blocked limit button blocked_button_html = f"""
🚀 Remove More Watermarks
""" return None, f"❌ You have reached Hugging Face's free watermark removal limit. Please visit https://omnicreator.net/remove-watermark#generator for unlimited removal", gr.update(value=blocked_button_html, visible=True) # Check rate limit (applies to rate_limit phases) if current_phase in ['rate_limit_1']: 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 # Generate rate limit button rate_limit_button_html = f"""
⏰ Skip Wait - Unlimited Watermark Removal
""" return None, f"❌ You have reached Hugging Face's free watermark removal limit. Please visit https://omnicreator.net/remove-watermark#generator for unlimited removal, or wait {wait_minutes_int} minutes before generating again", gr.update(value=rate_limit_button_html, visible=True) # No NSFW detection needed for watermark removal result_url = None status_message = "" def progress_callback(message): try: nonlocal status_message status_message = message # Add error handling to prevent progress update failure if progress is not None: progress(0.5, desc=message) except Exception as e: print(f"⚠️ Progress update failed: {e}") try: # Record generation attempt (before actual generation to ensure correct count) record_generation_attempt(client_ip, current_phase) updated_count = get_ip_generation_count(client_ip) print(f"✅ Processing started - IP: {client_ip}, phase: {current_phase}, total count: {updated_count}, prompt: {prompt.strip()}", flush=True) # Force removal mode: Generate smart prompt using OpenRouter API if force_removal: if progress is not None: progress(0.3, desc="💪 AI Force Mode: Analyzing watermark pattern...") try: # Upload image to get URL for OpenRouter API from util import upload_user_img_r2 time_id = int(time.time()) image_url = upload_user_img_r2(client_ip, time_id, input_image) if image_url: # Generate smart prompt using OpenRouter API smart_prompt = generate_smart_watermark_prompt(image_url) if smart_prompt and smart_prompt.strip() != "remove watermark in the image": prompt = smart_prompt print(f"🚀 Using AI Force Mode prompt: {prompt}") else: print("⚠️ Using fallback prompt") else: print("⚠️ Failed to upload image for AI analysis, using default prompt") except Exception as e: print(f"⚠️ Force Mode prompt generation failed: {e}, using default prompt") if progress is not None: progress(0.5, desc="Starting force watermark removal...") # Call image editing processing function result_url, message, task_uuid = process_image_edit(input_image, prompt.strip(), progress_callback) if result_url: print(f"✅ Watermark removal completed successfully - IP: {client_ip}, result_url: {result_url}, task_uuid: {task_uuid}", flush=True) # No NSFW detection needed for watermark removal final_result = result_url final_message = "✅ " + message try: if progress is not None: progress(1.0, desc="Processing completed") except Exception as e: print(f"⚠️ Final progress update failed: {e}") # Generate action buttons HTML like Trump AI Voice action_buttons_html = "" if task_uuid: task_detail_url = f"https://omnicreator.net/my-creations/task/{task_uuid}" action_buttons_html = f"""
🖼️ Download HD Image 🚀 Remove More Watermarks

Not satisfied with the removal result?

💪 Try Force Mode on Website
""" return final_result, final_message, gr.update(value=action_buttons_html, visible=True) else: print(f"❌ Processing failed - IP: {client_ip}, error: {message}", flush=True) return None, "❌ " + message, gr.update(visible=False) except Exception as e: print(f"❌ Processing exception - IP: {client_ip}, error: {str(e)}") return None, f"❌ Error occurred during processing: {str(e)}", gr.update(visible=False) # Create Gradio interface def create_app(): with gr.Blocks( title="AI Watermark Remover", theme=gr.themes.Soft(), css=""" .main-container { max-width: 1200px; margin: 0 auto; } .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; } .use-as-input-btn { margin-top: 10px; width: 100%; } """, # Improve concurrency performance configuration head=""" """ ) as app: # Main title - styled for watermark removal gr.HTML("""

🧽AI Watermark Remover

""", padding=False) # Powered by line below title - styled like Trump AI Voice gr.HTML("""

powered by omnicreator.net

""", padding=False) # Load sample images sample_images = load_sample_images() # Watermark removal interface (New 2-column layout) with gr.Row(): # --- LEFT COLUMN (INPUT & CONTROLS) --- with gr.Column(scale=1): gr.Markdown("### 📸 Upload Image with Watermark") input_image = gr.Image( label="Select image to remove watermark", type="pil", height=450, elem_classes=["upload-area"] ) force_removal = gr.Checkbox( label="💪 Force Removal", info="Ultimate AI-powered watermark removal. AI analyzes the watermark for better results.", value=False ) remove_button = gr.Button( "🧽Remove Watermark", variant="primary", size="lg" ) # --- RIGHT COLUMN (OUTPUT) --- with gr.Column(scale=1): gr.Markdown("### 🎯 Watermark Removal Result") output_image = gr.Image( label="Image without watermark", height=450, elem_classes=["result-area"] ) use_as_input_btn = gr.Button( "🔄 Use as Input", variant="secondary", size="sm", elem_classes=["use-as-input-btn"] ) status_output = gr.Textbox( label="Processing status", lines=2, max_lines=3, interactive=False ) action_buttons = gr.HTML(visible=False) # Sample images gallery (moved below main interface) sample_buttons = [] if sample_images: gr.Markdown("### 💡 Try with Sample Images") with gr.Row(): for idx, (img_path, img_name) in enumerate(sample_images[:4]): # Show first 4 samples with gr.Column(scale=1): sample_img = gr.Image( value=img_path, label=img_name, height=150, interactive=False ) use_sample_btn = gr.Button( f"Use {img_name}", size="sm", variant="secondary" ) sample_buttons.append((use_sample_btn, img_path)) # Add sample image button functionality with demo processing def create_sample_handler(img_path, img_name): def handler(): import time import os from PIL import Image # Wait 3 seconds as requested time.sleep(3) # Get the pre-processed result file path result_file_path = SAMPLE_RESULTS.get(img_name) if result_file_path and os.path.exists(result_file_path): # Load the result image from local file try: result_image = Image.open(result_file_path) return img_path, result_image, "✅ sample watermark removal completed! This is a demo result." except Exception as e: print(f"Failed to load sample result from {result_file_path}: {e}") # Fallback: just return the input image return img_path, None, "✅ sample image loaded. Click 'Remove Watermark' to process." return handler for sample_btn, img_path in sample_buttons: img_name = os.path.basename(img_path) sample_btn.click( fn=create_sample_handler(img_path, img_name), outputs=[input_image, output_image, status_output], show_progress=True ) # Bind button click events remove_button.click( fn=remove_watermark_interface, inputs=[input_image, force_removal], outputs=[output_image, status_output, action_buttons], show_progress=True, concurrency_limit=10, api_name="remove_watermark" ) # "Use as Input" button functionality def simple_use_as_input(output_img): if output_img is not None: return output_img return None use_as_input_btn.click( fn=simple_use_as_input, inputs=[output_image], outputs=[input_image] ) # Watermark Removal SEO Content Section gr.HTML("""

🧽Unlimited AI Watermark Removal

Remove watermarks from any image instantly with advanced AI technology! Clean your photos and documents without compromising image quality using our professional watermark removal tool.

🧽Get Unlimited Watermark Removal

Join thousands of users who trust Omni Creator for professional watermark removal!

🌟 Professional AI Watermark Remover - Fast & Accurate

Clean your images effortlessly with our advanced AI watermark removal technology. Whether you're dealing with text watermarks, logo stamps, or transparent overlays - our intelligent algorithm removes them while preserving the original image quality and details.

🎯 Unlimited Removal

Premium users enjoy unlimited watermark removal without daily limits or processing restrictions. Clean as many images as you need, whenever you need them, with no quality loss.

🎯 Smart Detection

AI automatically identifies and removes various types of watermarks including text, logos, stamps, and transparent overlays without affecting the main image content.

⚡ Lightning Fast Processing

Advanced AI infrastructure delivers watermark-free results in seconds. No waiting in queues, no processing delays - just instant, professional-grade watermark removal.

🎨 Quality Preservation

Intelligent inpainting technology reconstructs watermark areas seamlessly while maintaining original image quality, colors, textures, and details perfectly.

💎 Premium Quality

State-of-the-art AI models trained specifically for watermark removal deliver exceptional results. Professional quality output suitable for commercial use and high-end projects.

🌍 Universal Compatibility

Works with all image formats (JPG, PNG, GIF, BMP) and removes any type of watermark. From photos to documents, product images to artwork - we clean them all perfectly.

💎 Why Choose Premium Watermark Removal?

🚫 No Processing Limits

Remove watermarks from unlimited images without waiting periods or daily restrictions

🧽Advanced Removal

Remove any type of watermark including complex, transparent, and multi-layered marks

⚡ Priority Processing

Skip queues and get instant watermark removal with dedicated processing power

🎨 Perfect Quality

Access to latest AI models for highest quality watermark removal results

🌟 Start Creating Now

💡 Pro Tips for Best Watermark Removal

📝 Image Quality:

Higher resolution input images (up to 10MB) generally produce better watermark removal results and finer details.

🎯 Clean Images:

Images with simple backgrounds work best. Complex patterns behind watermarks may require multiple processing attempts.

⚡ Iterative Process:

Use "Use as Input" feature to refine results. Multiple iterations can remove stubborn or complex watermarks completely.

🖼️ File Formats:

JPG and PNG formats work best. Avoid heavily compressed images as they may affect watermark detection accuracy.

🚀 Perfect For Every Watermark Removal Need

📄 Documents

  • PDFs with watermarks
  • Scanned documents
  • Legal papers
  • Certificates

📸 Photography

  • Stock photos
  • Portrait photos
  • Event photography
  • Nature images

🛍️ E-commerce

  • Product images
  • Catalog photos
  • Supplier images
  • Brand cleanup

📱 Social Media

  • Content creation
  • Profile pictures
  • Marketing visuals
  • Clean presentations
🧽Start Removing Watermarks

Powered by Omni Creator

The ultimate AI watermark removal platform • Professional results, unlimited processing

""", padding=False) return app if __name__ == "__main__": app = create_app() # Improve queue configuration to handle high concurrency and prevent SSE connection issues app.queue( default_concurrency_limit=20, # Default concurrency limit max_size=50, # Maximum queue size api_open=False # Close API access to reduce resource consumption ) app.launch( server_name="0.0.0.0", show_error=True, # Show detailed error information quiet=False, # Keep log output max_threads=40, # Increase thread pool size height=800, favicon_path=None # Reduce resource loading )