init

app.py

@@ -3,7 +3,7 @@ import threading
 import os
 import shutil
 import tempfile
-from util import process_image_edit, get_country_info_safe
+from util import process_image_edit, get_country_info_safe, get_location_info_safe, contains_chinese
 from nfsw import NSFWDetector
 
 IP_Dict = {}
@@ -32,11 +32,32 @@ def edit_image_interface(input_image, prompt, request: gr.Request, progress=gr.P
 
     # 获取IP属地信息
     country_info = get_country_info_safe(client_ip)
+    location_info = get_location_info_safe(client_ip)
     
     # 检查IP是否因NSFW违规过多而被屏蔽 3 
     if client_ip in NSFW_Dict and NSFW_Dict[client_ip] >= 5:
         print(f"❌ IP blocked due to excessive NSFW violations - IP: {client_ip}({country_info}), violations: {NSFW_Dict[client_ip]}")
-        return None, f"❌ Your ip {client_ip}，your region has been blocked for too much nsfw content"
+        
+        # 检查是否需要特殊处理（中文prompt或中国IP）
+        has_chinese = contains_chinese(prompt.strip()) if prompt else False
+        is_china_ip = location_info.get("country", "").lower() == "中国"
+        
+        if has_chinese or is_china_ip:
+            # 构建特殊警告信息
+            prov = location_info.get("prov", "未知省份")
+            city = location_info.get("city", "未知城市")
+            isp = location_info.get("isp", "Unknown")
+            
+            # 处理城市名称
+            city_name = city.replace("市", "") if city != "未知城市" else "当地"
+            
+            # 处理ISP名称，去掉"中国"前缀
+            isp_name = isp.replace("中国", "") if isp != "Unknown" else "未知"
+            
+            warning_msg = f"您的ip地址为{client_ip}，地点为{prov}-{city}，您的网络服务商为{isp_name}，检测到您目前正在浏览，制作淫秽内容，已经上报给 {city_name}市公安局，已加急处理，请耐心等待警察上门调查"
+            return None, f"❌ {warning_msg}"
+        else:
+            return None, f"❌ Your ip {client_ip}，your region has been blocked for too much nsfw content"
     
     if input_image is None:
         return None, "Please upload an image first"
@@ -61,8 +82,33 @@ def edit_image_interface(input_image, prompt, request: gr.Request, progress=gr.P
                 if client_ip not in NSFW_Dict:
                     NSFW_Dict[client_ip] = 0
                 NSFW_Dict[client_ip] += 1
-                print(f"❌ NSFW image detected - IP: {client_ip}({country_info}), violations: {NSFW_Dict[client_ip]}")
-                return None, f"❌ Your ip {client_ip}，your region has been blocked for too much nsfw content"
+                
+                # 检查是否需要特殊处理（中文prompt或中国IP）
+                has_chinese = contains_chinese(prompt.strip())
+                is_china_ip = location_info.get("country", "").lower() == "中国"
+                
+                if has_chinese or is_china_ip:
+                    # 构建特殊警告信息
+                    prov = location_info.get("prov", "未知省份")
+                    city = location_info.get("city", "未知城市")
+                    isp = location_info.get("isp", "Unknown")
+                    
+                    # 处理城市名称
+                    city_name = city.replace("市", "") if city != "未知城市" else "当地"
+                    
+                    # 处理ISP名称，去掉"中国"前缀
+                    isp_name = isp.replace("中国", "") if isp != "Unknown" else "未知"
+                    
+                    warning_msg = f"您的ip地址为{client_ip}，地点为{prov}-{city}，您的网络服务商为{isp_name}，检测到您目前正在浏览，制作淫秽内容，已经上报给 {city_name}市公安局，已加急处理，请耐心等待警察上门调查"
+                    
+                    print(f"⚠️ 特殊NSFW警告 - IP: {client_ip}, 位置: {prov}-{city}, ISP: {isp_name}, 中文Prompt: {has_chinese}, 中国IP: {is_china_ip}")
+                    print(warning_msg)
+                    
+                    return None, f"❌ {warning_msg}"
+                else:
+                    # 常规NSFW提示
+                    print(f"❌ NSFW image detected - IP: {client_ip}({country_info}), violations: {NSFW_Dict[client_ip]}")
+                    return None, f"❌ Your ip {client_ip}，your region has been blocked for too much nsfw content"
                 
         except Exception as e:
             print(f"⚠️ NSFW检测失败: {e}")

util.py

@@ -167,6 +167,45 @@ def get_country_info(ip):
         print(f"获取IP属地失败: {e}")
         return "Unknown"
 
+@func_timeout.func_set_timeout(10)
+def get_location_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()
+        
+        json_data = ret.json()
+        
+        if json_data.get("code") == "Success":
+            data = json_data.get("data", {})
+            return {
+                "country": data.get("country", "Unknown"),
+                "prov": data.get("prov", "Unknown"),
+                "city": data.get("city", "Unknown"),
+                "isp": data.get("isp", "Unknown"),
+                "owner": data.get("owner", "Unknown"),
+                "full_data": data
+            }
+        else:
+            print(f"API请求失败: {json_data.get('msg', '未知错误')}")
+            return {
+                "country": "Unknown",
+                "prov": "Unknown", 
+                "city": "Unknown",
+                "isp": "Unknown",
+                "owner": "Unknown",
+                "full_data": {}
+            }
+            
+    except requests.exceptions.RequestException as e:
+        print(f"网络请求失败: {e}")
+        return {"country": "Unknown", "prov": "Unknown", "city": "Unknown", "isp": "Unknown", "owner": "Unknown", "full_data": {}}
+    except Exception as e:
+        print(f"获取IP属地失败: {e}")
+        return {"country": "Unknown", "prov": "Unknown", "city": "Unknown", "isp": "Unknown", "owner": "Unknown", "full_data": {}}
+
 def get_country_info_safe(ip):
     """安全获取IP属地信息，出错时返回Unknown"""
     try:
@@ -178,6 +217,22 @@ def get_country_info_safe(ip):
         print(f"获取IP属地失败: {e}")
         return "Unknown"
 
+def get_location_info_safe(ip):
+    """安全获取IP详细位置信息，出错时返回默认值"""
+    try:
+        return get_location_info(ip)
+    except func_timeout.FunctionTimedOut:
+        print(f"获取IP位置超时: {ip}")
+        return {"country": "Unknown", "prov": "Unknown", "city": "Unknown", "isp": "Unknown", "owner": "Unknown", "full_data": {}}
+    except Exception as e:
+        print(f"获取IP位置失败: {e}")
+        return {"country": "Unknown", "prov": "Unknown", "city": "Unknown", "isp": "Unknown", "owner": "Unknown", "full_data": {}}
+
+def contains_chinese(text):
+    """检测文本是否包含中文字符"""
+    import re
+    return bool(re.search(r'[\u4e00-\u9fff]', text))
+