feat: 新增代理连接测试功能、支持自定义测试域名及实时状态反馈

src/api/admin.py

@@ -122,6 +122,9 @@ class UpdateProxyConfigRequest(BaseModel):
     proxy_enabled: bool
     proxy_url: Optional[str] = None
 
+class TestProxyRequest(BaseModel):
+    test_url: Optional[str] = "https://sora.chatgpt.com"
+
 class UpdateAdminPasswordRequest(BaseModel):
     old_password: str
     new_password: str
@@ -794,6 +797,50 @@ async def update_proxy_config(
     except Exception as e:
         raise HTTPException(status_code=400, detail=str(e))
 
+@router.post("/api/proxy/test")
+async def test_proxy_config(
+    request: TestProxyRequest,
+    token: str = Depends(verify_admin_token)
+) -> dict:
+    """Test proxy connectivity with custom URL"""
+    from curl_cffi.requests import AsyncSession
+
+    config_obj = await proxy_manager.get_proxy_config()
+    if not config_obj.proxy_enabled or not config_obj.proxy_url:
+        return {"success": False, "message": "代理未启用或地址为空"}
+
+    # Use provided test URL or default
+    test_url = request.test_url or "https://sora.chatgpt.com"
+
+    try:
+        async with AsyncSession() as session:
+            response = await session.get(
+                test_url,
+                timeout=15,
+                impersonate="chrome",
+                proxy=config_obj.proxy_url
+            )
+        status_code = response.status_code
+        if 200 <= status_code < 400:
+            return {
+                "success": True,
+                "message": f"代理可用 (HTTP {status_code})",
+                "status_code": status_code,
+                "test_url": test_url
+            }
+        return {
+            "success": False,
+            "message": f"代理响应异常: HTTP {status_code}",
+            "status_code": status_code,
+            "test_url": test_url
+        }
+    except Exception as e:
+        return {
+            "success": False,
+            "message": f"代理连接失败: {str(e)}",
+            "test_url": test_url
+        }
+
 # Watermark-free config endpoints
 @router.get("/api/watermark-free/config")
 async def get_watermark_free_config(token: str = Depends(verify_admin_token)) -> dict: