init

.gitignore

@@ -0,0 +1,2 @@
+*.jpg
+*.png

README.md

@@ -8,7 +8,7 @@ sdk_version: 5.44.1
 app_file: app.py
 pinned: false
 license: mit
-short_description: image edit
+short_description: Advanced AI-powered image editing tool with multiple intelligent features
 ---
 
 Check out the configuration reference at https://huggingface.co/docs/hub/spaces-config-reference

app.py

@@ -0,0 +1,190 @@
+import gradio as gr
+import threading
+from util import process_image_edit, check_nsfw, get_country_info_safe
+
+IP_Dict = {}
+NSFW_Dict = {}  # 记录每个IP的NSFW违规次数
+
+def edit_image_interface(input_image, prompt, request: gr.Request, progress=gr.Progress()):
+    """
+    Interface function for processing image editing
+    """
+    # 提取用户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
+
+    # 获取IP属地信息
+    country_info = get_country_info_safe(client_ip)
+    
+    # 检查IP是否因NSFW违规过多而被屏蔽 3 
+    if client_ip in NSFW_Dict and NSFW_Dict[client_ip] >= 3:
+        print(f"❌ IP blocked due to excessive NSFW violations - IP: {client_ip}({country_info}), violations: {NSFW_Dict[client_ip]}")
+        return None, "❌ NSFW content too much. Access denied due to policy violations"
+    
+    if input_image is None:
+        return None, "Please upload an image first"
+    
+    if not prompt or prompt.strip() == "":
+        return None, "Please enter editing prompt"
+    
+    # 检查prompt长度是否大于3个字符
+    if len(prompt.strip()) <= 3:
+        return None, "❌ Editing prompt must be more than 3 characters"
+    
+    # 检查是否包含NSFW内容
+    if check_nsfw(prompt.strip()) == 1:
+        # 记录NSFW违规次数
+        if client_ip not in NSFW_Dict:
+            NSFW_Dict[client_ip] = 0
+        NSFW_Dict[client_ip] += 1
+        print(f"❌ NSFW content detected - IP: {client_ip}({country_info}), violations: {NSFW_Dict[client_ip]}, prompt: {prompt.strip()}")
+        return None, "❌ NSFW content detected. Please modify your prompt. NSFW detected {NSFW_Dict[client_ip]}/10 times"
+    
+    if IP_Dict[client_ip]>8 and country_info.lower() in ["印度", "巴基斯坦"]:
+        print(f"❌ Content not allowed - IP: {client_ip}({country_info}), count: {IP_Dict[client_ip]}, prompt: {prompt.strip()}")
+        return None, "❌ Content not allowed. Please modify your prompt"
+    if IP_Dict[client_ip]>18 and country_info.lower() in ["中国"]:
+        print(f"❌ Content not allowed - IP: {client_ip}({country_info}), count: {IP_Dict[client_ip]}, prompt: {prompt.strip()}")
+        return None, "❌ Content not allowed. Please modify your prompt"
+        
+    if client_ip.lower() in ["221.194.171.230", "101.126.56.37", "101.126.56.44"]:
+        print(f"❌ Content not allowed - IP: {client_ip}({country_info}), count: {IP_Dict[client_ip]}, prompt: {prompt.strip()}")
+        return None, "❌ Content not allowed. Please modify your prompt"
+        
+
+    result_url = None
+    status_message = ""
+    
+    def progress_callback(message):
+        nonlocal status_message
+        status_message = message
+        progress(0.5, desc=message)
+    
+    try:
+        # 打印成功访问的信息
+        print(f"✅ Processing started - IP: {client_ip}({country_info}), count: {IP_Dict[client_ip]}, prompt: {prompt.strip()}")
+        
+        # Call image editing processing function
+        result_url, message = process_image_edit(input_image, prompt.strip(), progress_callback)
+        
+        if result_url:
+            print(f"✅ Processing completed successfully - IP: {client_ip}({country_info}), result_url: {result_url}")
+            progress(1.0, desc="Processing completed")
+            return result_url, "✅ " + message
+        else:
+            print(f"❌ Processing failed - IP: {client_ip}({country_info}), error: {message}")
+            return None, "❌ " + message
+            
+    except Exception as e:
+        return None, f"❌ Error occurred during processing: {str(e)}"
+
+# Create Gradio interface
+def create_app():
+    with gr.Blocks(
+        title="AI Image Editor",
+        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;
+        }
+        """
+    ) as app:
+        
+        gr.Markdown(
+            """
+            # 🎨 AI Image Editor
+            """,
+            elem_classes=["main-container"]
+        )
+        
+        with gr.Row():
+            with gr.Column(scale=1):
+                gr.Markdown("### 📸 Upload Image")
+                input_image = gr.Image(
+                    label="Select image to edit",
+                    type="filepath",
+                    height=400,
+                    elem_classes=["upload-area"]
+                )
+                
+                gr.Markdown("### ✍️ Editing Instructions")
+                prompt_input = gr.Textbox(
+                    label="Enter editing prompt",
+                    placeholder="For example: change background to beach, add rainbow, remove background, etc...",
+                    lines=3,
+                    max_lines=5
+                )
+                
+                edit_button = gr.Button(
+                    "🚀 Start Editing",
+                    variant="primary",
+                    size="lg"
+                )
+            
+            with gr.Column(scale=1):
+                gr.Markdown("### 🎯 Editing Result")
+                output_image = gr.Image(
+                    label="Edited image",
+                    height=400,
+                    elem_classes=["result-area"]
+                )
+                
+                status_output = gr.Textbox(
+                    label="Processing status",
+                    lines=2,
+                    max_lines=3,
+                    interactive=False
+                )
+        
+        # Example area
+        gr.Markdown("### 💡 Prompt Examples")
+        with gr.Row():
+            example_prompts = [
+                "Change the character's background to a sunny seaside with blue waves.",
+                "Change the character's background to New York at night with neon lights.",
+                "Change the character's background to a fairytale castle with bright colors.",
+                "Change background to forest",
+                "Change background to snow mountain"
+            ]
+            
+            for prompt in example_prompts:
+                gr.Button(
+                    prompt,
+                    size="sm"
+                ).click(
+                    lambda p=prompt: p,
+                    outputs=prompt_input
+                )
+        
+        # Bind button click event
+        edit_button.click(
+            fn=edit_image_interface,
+            inputs=[input_image, prompt_input],
+            outputs=[output_image, status_output],
+            show_progress=True
+        )
+    
+    return app
+
+if __name__ == "__main__":
+    app = create_app()
+    app.queue()  # Enable queue to handle concurrent requests
+    app.launch()

push.sh

@@ -0,0 +1,13 @@
+
+# 设置仓库级别用户名
+git config user.name "selfitcamera"
+git config user.email "ethan.blake@heybeauty.ai"
+
+# 验证
+git config user.name
+git config user.email
+
+
+git add .
+git commit -m "init"
+git push

requirements.txt

@@ -0,0 +1,7 @@
+gradio>=4.0.0
+opencv-python>=4.8.0
+requests>=2.28.0
+func-timeout>=4.3.5
+numpy>=1.24.0
+boto3
+botocore

util.py

@@ -0,0 +1,350 @@
+
+import os
+import sys
+import cv2
+import json
+import random
+import time
+import datetime
+import requests
+import func_timeout
+import numpy as np
+import gradio as gr
+import boto3
+from botocore.client import Config
+
+
+# TOKEN = os.environ['TOKEN']
+# APIKEY = os.environ['APIKEY']
+# UKAPIURL = os.environ['UKAPIURL']
+
+OneKey = os.environ['OneKey'].strip()
+OneKey = OneKey.split("#")
+TOKEN = OneKey[0]
+APIKEY = OneKey[1]
+UKAPIURL = OneKey[2]
+LLMKEY = OneKey[3]
+R2_ACCESS_KEY = OneKey[4]
+R2_SECRET_KEY = OneKey[5]
+R2_ENDPOINT = OneKey[6]
+
+
+tmpFolder = "tmp"
+os.makedirs(tmpFolder, exist_ok=True)
+
+
+def upload_user_img(clientIp, timeId, img):
+    fileName = clientIp.replace(".", "")+str(timeId)+".jpg"
+    local_path = os.path.join(tmpFolder, fileName)
+    img = cv2.imread(img)
+    cv2.imwrite(os.path.join(tmpFolder, fileName), img)
+
+    json_data = {
+        "token": TOKEN,  
+        "input1": fileName,
+        "input2": "",
+        "protocol": "",
+        "cloud": "ali"
+    }
+    
+    session = requests.session()
+    ret = requests.post(
+        f"{UKAPIURL}/upload",
+        headers={'Content-Type': 'application/json'},
+        json=json_data
+    )
+    
+    res = ""
+    if ret.status_code==200:
+        if 'upload1' in ret.json():
+            upload_url = ret.json()['upload1']
+            headers = {'Content-Type': 'image/jpeg'}
+            response = session.put(upload_url, data=open(local_path, 'rb').read(), headers=headers)
+            # print(response.status_code)
+            if response.status_code == 200:
+                res = upload_url
+    if os.path.exists(local_path):
+        os.remove(local_path)
+    return res
+
+
+class R2Api:
+
+    def __init__(self, session=None):
+        super().__init__()
+        self.R2_BUCKET = "trump-ai-voice"
+        self.domain = "https://www.trumpaivoice.net/"
+        self.R2_ACCESS_KEY = R2_ACCESS_KEY
+        self.R2_SECRET_KEY = R2_SECRET_KEY
+        self.R2_ENDPOINT = R2_ENDPOINT
+
+        self.client = boto3.client(
+            "s3",
+            endpoint_url=self.R2_ENDPOINT,
+            aws_access_key_id=self.R2_ACCESS_KEY,
+            aws_secret_access_key=self.R2_SECRET_KEY,
+            config=Config(signature_version="s3v4")
+        )
+
+        self.session = requests.Session() if session is None else session
+
+    def upload_file(self, local_path, cloud_path):
+        t1 = time.time()
+        head_dict = {
+            'jpg': 'image/jpeg',
+            'jpeg': 'image/jpeg',
+            'png': 'image/png',
+            'gif': 'image/gif',
+            'bmp': 'image/bmp',
+            'webp': 'image/webp',
+            'ico': 'image/x-icon'
+        }
+        ftype = os.path.basename(local_path).split(".")[-1].lower()
+        ctype = head_dict.get(ftype, 'application/octet-stream')
+        headers = {"Content-Type": ctype}
+
+
+        cloud_path = f"QwenImageEdit/Uploads/{str(datetime.date.today())}/{os.path.basename(local_path)}"
+        url = self.client.generate_presigned_url(
+            "put_object",
+            Params={"Bucket": self.R2_BUCKET, "Key": cloud_path, "ContentType": ctype},
+            ExpiresIn=604800
+        )
+
+        retry_count = 0
+        while retry_count < 3:
+            try:
+                with open(local_path, 'rb') as f:
+                    self.session.put(url, data=f.read(), headers=headers, timeout=8)
+                break
+            except (requests.exceptions.Timeout, requests.exceptions.RequestException):
+                retry_count += 1
+                if retry_count == 3:
+                    raise Exception('Failed to upload file to R2 after 3 retries!')
+                continue
+        print("upload_file time is ====>", time.time() - t1)
+        return f"{self.domain}{cloud_path}"
+    
+def upload_user_img_r2(clientIp, timeId, img):
+    fileName = clientIp.replace(".", "")+str(timeId)+".jpg"
+    local_path = os.path.join(tmpFolder, fileName)
+    img = cv2.imread(img)
+    cv2.imwrite(os.path.join(tmpFolder, fileName), img)
+
+    res = R2Api().upload_file(local_path, fileName)
+
+    if os.path.exists(local_path):
+        os.remove(local_path)
+    return res
+
+
+@func_timeout.func_set_timeout(10)
+def get_country_info(ip):
+    """获取IP对应的国家信息"""
+    try:
+        # 使用您指定的新接口 URL
+        url = f"https://qifu-api.baidubce.com/ip/geo/v1/district?ip={ip}"
+        ret = requests.get(url)
+        ret.raise_for_status()  # 如果请求失败 (例如 404, 500), 会抛出异常
+        
+        json_data = ret.json()
+        
+        # 根据新的JSON结构，国家信息在 'data' -> 'country' 路径下
+        if json_data.get("code") == "Success":
+            country = json_data.get("data", {}).get("country")
+            return country if country else "Unknown"
+        else:
+            # 处理API返回错误码的情况
+            print(f"API请求失败: {json_data.get('msg', '未知错误')}")
+            return "Unknown"
+            
+    except requests.exceptions.RequestException as e:
+        print(f"网络请求失败: {e}")
+        return "Unknown"
+    except Exception as e:
+        print(f"获取IP属地失败: {e}")
+        return "Unknown"
+
+def get_country_info_safe(ip):
+    """安全获取IP属地信息，出错时返回Unknown"""
+    try:
+        return get_country_info(ip)
+    except func_timeout.FunctionTimedOut:
+        print(f"获取IP属地超时: {ip}")
+        return "Unknown"
+    except Exception as e:
+        print(f"获取IP属地失败: {e}")
+        return "Unknown"
+
+
+def check_nsfw(prompt):
+    """
+    检查prompt是否包含NSFW内容，包含返回1，否则返回0
+    """
+    try:
+        response = requests.post(
+            url="https://openrouter.ai/api/v1/chat/completions",
+            headers={
+                "Authorization": f"Bearer {LLMKEY}",
+                "Content-Type": "application/json",
+            },
+            data=json.dumps({
+                "model": "google/gemini-2.5-flash",
+                "messages": [
+                    {
+                        "role": "system",
+                        "content": "你是一个nsfw指令判断助手，请判断用户输入的prompt指令是否会导致nsfw内容? 你只需要回答 是 或者 否"
+                    },
+                    {
+                        "role": "user",
+                        "content": prompt
+                    }
+                ],
+            })
+        )
+        res_json = response.json()
+        # 兼容不同模型返回格式
+        if "choices" in res_json and len(res_json["choices"]) > 0:
+            content = res_json["choices"][0].get("message", {}).get("content", "")
+            if "是" in content:
+                return 1
+            else:
+                return 0
+        else:
+            return 0
+    except Exception as e:
+        # 出错时默认返回0
+        return 0
+
+
+def submit_image_edit_task(user_image_url, prompt):
+    """
+    提交图片编辑任务
+    """
+    headers = {
+        'Content-Type': 'application/json',
+        'Authorization': f'Bearer {APIKEY}'
+    }
+    
+    data = {
+        "user_image": user_image_url,
+        "task_type": "80",
+        "prompt": prompt,
+        "secret_key": "219ngu",
+        "is_private": "0"
+    }
+    
+    try:
+        response = requests.post(
+            f'{UKAPIURL}/public_image_edit',
+            headers=headers,
+            json=data
+        )
+        
+        if response.status_code == 200:
+            result = response.json()
+            if result.get('code') == 0:
+                return result['data']['task_id'], None
+            else:
+                return None, f"API 错误: {result.get('message', '未知错误')}"
+        else:
+            return None, f"HTTP 错误: {response.status_code}"
+    except Exception as e:
+        return None, f"请求异常: {str(e)}"
+
+
+def check_task_status(task_id):
+    """
+    查询任务状态
+    """
+    headers = {
+        'Content-Type': 'application/json',
+        'Authorization': f'Bearer {APIKEY}'
+    }
+    
+    data = {
+        "task_id": task_id
+    }
+    
+    try:
+        response = requests.post(
+            f'{UKAPIURL}/status_image_edit',
+            headers=headers,
+            json=data
+        )
+        
+        if response.status_code == 200:
+            result = response.json()
+            if result.get('code') == 0:
+                task_data = result['data']
+                return task_data['status'], task_data.get('output1'), task_data
+            else:
+                return 'error', None, result.get('message', '未知错误')
+        else:
+            return 'error', None, f"HTTP 错误: {response.status_code}"
+    except Exception as e:
+        return 'error', None, f"请求异常: {str(e)}"
+
+
+def process_image_edit(img_path, prompt, progress_callback=None):
+    """
+    处理图片编辑的完整流程
+    """
+    try:
+        # 生成客户端 IP 和时间戳
+        client_ip = "127.0.0.1"  # 默认IP
+        time_id = int(time.time())
+        
+        if progress_callback:
+            progress_callback("uploading image...")
+        
+        # 上传用户图片
+        uploaded_url = upload_user_img_r2(client_ip, time_id, img_path)
+        if not uploaded_url:
+            return None, "image upload failed"
+        
+        # 从上传 URL 中提取实际的图片 URL
+        if "?" in uploaded_url:
+            uploaded_url = uploaded_url.split("?")[0]
+        
+        if progress_callback:
+            progress_callback("submitting edit task...")
+        
+        # 提交图片编辑任务
+        task_id, error = submit_image_edit_task(uploaded_url, prompt)
+        if error:
+            return None, error
+        
+        if progress_callback:
+            progress_callback(f"task submitted, ID: {task_id}, processing...")
+        
+        # 等待任务完成
+        max_attempts = 60  # 最多等待10分钟
+        for attempt in range(max_attempts):
+            status, output_url, task_data = check_task_status(task_id)
+            
+            if status == 'completed':
+                if output_url:
+                    return output_url, "image edit completed"
+                else:
+                    return None, "任务完成但未返回结果图片"
+            elif status == 'error' or status == 'failed':
+                return None, f"task processing failed: {task_data}"
+            elif status in ['queued', 'processing', 'running', 'created', 'working']:
+                if progress_callback:
+                    progress_callback(f"task processing... (status: {status})")
+                time.sleep(1)  # 等待10秒后重试
+            else:
+                if progress_callback:
+                    progress_callback(f"unknown status: {status}")
+                time.sleep(1)
+        
+        return None, "task processing timeout"
+        
+    except Exception as e:
+        return None, f"error occurred during processing: {str(e)}"
+
+
+if __name__ == "__main__":
+    
+    pass