add: 支持远程滑块服务

api_captcha_remote.py

@@ -39,11 +39,80 @@ class MouseEvent(BaseModel):
     y: int
 
 
+class SliderSolveRequest(BaseModel):
+    """远程过滑块请求。"""
+    secret_key: str
+    url: str
+    account_id: str = "external"
+    browser_timeout: int = 60
+    show_browser: bool = False
+
+
 class SessionCheckRequest(BaseModel):
     """会话检查请求"""
     session_id: str
 
 
+# =============================================================================
+# 远程过滑块接口
+# =============================================================================
+
+@router.post("/slider-solve")
+async def slider_solve(request: SliderSolveRequest):
+    """远程过滑块服务端接口。
+
+    需要配置环境变量 XY_SLIDER_REMOTE_SECRET；调用方传入相同 secret_key 后，
+    服务端使用本机 Playwright + DrissionPage 兜底处理 punish 链接。
+    """
+    configured_secret = os.environ.get("XY_SLIDER_REMOTE_SECRET", "").strip()
+    if not configured_secret:
+        return {"success": False, "message": "远程过滑块服务未启用：缺少 XY_SLIDER_REMOTE_SECRET", "data": None}
+    if not request.secret_key or request.secret_key != configured_secret:
+        return {"success": False, "message": "无效的秘钥", "data": None}
+
+    target_url = str(request.url or "").strip()
+    if not target_url:
+        return {"success": False, "message": "punish 链接不能为空", "data": None}
+    if not target_url.lower().startswith(("http://", "https://")):
+        return {"success": False, "message": "punish 链接必须以 http:// 或 https:// 开头", "data": None}
+
+    try:
+        from utils.xianyu_slider_stealth import XianyuSliderStealth
+        from utils.slider_orchestrator import run_slider_async_with_fallback
+
+        account_id = str(request.account_id or "external").strip() or "external"
+        slider = XianyuSliderStealth(
+            user_id=account_id,
+            enable_learning=True,
+            headless=not bool(request.show_browser),
+        )
+        result = await run_slider_async_with_fallback(
+            slider,
+            target_url,
+            engine="playwright",
+            remote_enabled=False,  # 防止远程接口指回本机时递归调用
+            remote_timeout=max(20, min(int(request.browser_timeout or 60), 180)),
+        )
+        if result.success:
+            return {
+                "success": True,
+                "message": result.message,
+                "data": {
+                    "engine": result.engine,
+                    "cookies": result.cookies,
+                    "x5_cookies": result.x5_cookies,
+                },
+            }
+        return {
+            "success": False,
+            "message": result.message,
+            "data": {"engine": result.engine, "x5_cookies": result.x5_cookies},
+        }
+    except Exception as exc:
+        logger.error(f"远程过滑块失败: {exc}")
+        return {"success": False, "message": f"过滑块失败: {exc}", "data": None}
+
+
 # =============================================================================
 # WebSocket 端点 - 实时通信
 # =============================================================================

tests/test_xianyu_token_refresh_request.py

@@ -80,6 +80,38 @@ class SliderOrchestratorTest(unittest.TestCase):
         self.assertTrue(result.success)
         self.assertEqual(result.cookies["x5sec"], "ticket")
         self.assertEqual(result.x5_cookies, {"x5sec": "ticket"})
+    def test_remote_solver_runs_before_local_slider_when_configured(self):
+        class _PrimarySlider:
+            user_id = "remote_user"
+            initial_cookies = "unb=remote_user; cookie2=old"
+            headless = True
+
+            def run(self, *_args, **_kwargs):
+                raise AssertionError("remote success should short-circuit local slider")
+
+        class _FakeResponse:
+            def raise_for_status(self):
+                return None
+
+            def json(self):
+                return {
+                    "success": True,
+                    "data": {"cookies": {"unb": "remote_user", "x5sec": "remote_ticket"}},
+                }
+
+        with mock.patch("utils.slider_orchestrator.requests.post", return_value=_FakeResponse()) as post_mock:
+            result = run_slider_with_fallback(
+                _PrimarySlider(),
+                "https://example.com/punish?action=captcha",
+                remote_enabled=True,
+                remote_config=("https://remote.example/api/captcha/slider-solve", "secret"),
+            )
+
+        self.assertTrue(result.success)
+        self.assertEqual(result.engine, "remote")
+        self.assertEqual(result.x5_cookies, {"x5sec": "remote_ticket"})
+        self.assertEqual(post_mock.call_args.kwargs["json"]["secret_key"], "secret")
+
     def test_drissionpage_fallback_can_recover_primary_failure(self):
         class _PrimarySlider:
             user_id = "fallback_user"

utils/slider_orchestrator.py

@@ -2,17 +2,19 @@
 
 调用现有 XianyuSliderStealth 后，必须拿到 x5/x5sec 相关 Cookie 才认为
 平台真正放行，避免“视觉通过但未下发 x5sec”被误当成功而导致 token
-刷新死循环；同时提供可选 DrissionPage 兜底入口。
+刷新死循环；同时提供可选远程服务与 DrissionPage 兜底入口。
 """
 from __future__ import annotations
 
 from dataclasses import dataclass
 import os
+import requests
 from typing import Any, Callable, Dict, Mapping, Optional, Tuple, Union
 
 
 DEFAULT_SLIDER_ENGINE = "playwright"
 DRISSIONPAGE_ENGINE = "drissionpage"
+REMOTE_ENGINE = "remote"
 _COOKIE_ATTR_NAMES = {"path", "domain", "expires", "max-age", "secure", "httponly", "samesite"}
 
 
@@ -38,6 +40,13 @@ def _env_bool(name: str, default: bool = False) -> bool:
     return str(raw).strip().lower() in {"1", "true", "yes", "on"}
 
 
+def _remote_config_from_env() -> Tuple[str, str]:
+    return (
+        os.environ.get("XY_SLIDER_REMOTE_URL", "").strip(),
+        os.environ.get("XY_SLIDER_REMOTE_SECRET", "").strip(),
+    )
+
+
 def parse_cookie_string(cookie_text: Optional[str]) -> Dict[str, str]:
     """解析 Cookie / Set-Cookie 字符串为字典。"""
     result: Dict[str, str] = {}
@@ -133,6 +142,52 @@ def validate_slider_result(
     )
 
 
+def _call_remote_solve(
+    url: str,
+    *,
+    user_id: str,
+    remote_url: str,
+    remote_secret: str,
+    timeout: int = 60,
+) -> SliderVerificationResult:
+    """调用远程过滑块服务，返回严格判定结果。"""
+    try:
+        response = requests.post(
+            remote_url,
+            json={
+                "secret_key": remote_secret,
+                "account_id": user_id,
+                "url": url,
+                "browser_timeout": timeout,
+            },
+            timeout=max(10, int(timeout or 60)),
+        )
+        response.raise_for_status()
+        payload = response.json()
+        data = payload.get("data") if isinstance(payload, dict) else None
+        data = data if isinstance(data, dict) else {}
+        cookies = data.get("cookies") or data.get("x5_cookies") or data.get("cookie")
+        success = bool(payload.get("success") if isinstance(payload, dict) else False)
+        result = validate_slider_result(success, cookies, engine=REMOTE_ENGINE)
+        if not result.success and isinstance(payload, dict) and payload.get("message"):
+            return SliderVerificationResult(
+                success=False,
+                cookies=result.cookies,
+                engine=REMOTE_ENGINE,
+                x5_cookies=result.x5_cookies,
+                message=str(payload.get("message")),
+            )
+        return result
+    except Exception as exc:
+        return SliderVerificationResult(
+            success=False,
+            cookies=None,
+            engine=REMOTE_ENGINE,
+            x5_cookies={},
+            message=f"远程滑块服务不可用: {exc}",
+        )
+
+
 def _run_drissionpage_fallback(
     url: str,
     *,
@@ -183,12 +238,30 @@ def run_slider_with_fallback(
     *,
     engine: Optional[str] = DEFAULT_SLIDER_ENGINE,
     fallback_enabled: Optional[bool] = None,
+    remote_enabled: Optional[bool] = None,
+    remote_config: Optional[Tuple[str, str]] = None,
+    remote_timeout: int = 60,
     fallback_headless: Optional[bool] = None,
     fallback_max_retries: int = 3,
     handler_factory: Optional[Callable[..., Any]] = None,
     **kwargs: Any,
 ) -> SliderVerificationResult:
-    """先运行现有 Playwright 滑块，失败时可用 DrissionPage 兜底。"""
+    """先运行远程/现有 Playwright 滑块，失败时可用 DrissionPage 兜底。"""
+    user_id = str(getattr(slider, "user_id", None) or getattr(slider, "pure_user_id", None) or "unknown")
+
+    use_remote = _env_bool("XY_SLIDER_REMOTE_ENABLED", False) if remote_enabled is None else bool(remote_enabled)
+    remote_url, remote_secret = remote_config or _remote_config_from_env()
+    if use_remote and remote_url and remote_secret:
+        remote_result = _call_remote_solve(
+            url,
+            user_id=user_id,
+            remote_url=remote_url,
+            remote_secret=remote_secret,
+            timeout=remote_timeout,
+        )
+        if remote_result.success:
+            return remote_result
+
     primary_result = run_slider_strict(slider, url, engine=engine, **kwargs)
     if primary_result.success:
         return primary_result
@@ -197,7 +270,6 @@ def run_slider_with_fallback(
     if not enabled:
         return primary_result
 
-    user_id = str(getattr(slider, "user_id", None) or getattr(slider, "pure_user_id", None) or "unknown")
     existing_cookies_str = str(getattr(slider, "initial_cookies", "") or "")
     headless = bool(getattr(slider, "headless", True)) if fallback_headless is None else bool(fallback_headless)
     fallback_result = _run_drissionpage_fallback(
@@ -229,12 +301,33 @@ async def run_slider_async_with_fallback(
     *,
     engine: Optional[str] = DEFAULT_SLIDER_ENGINE,
     fallback_enabled: Optional[bool] = None,
+    remote_enabled: Optional[bool] = None,
+    remote_config: Optional[Tuple[str, str]] = None,
+    remote_timeout: int = 60,
     fallback_headless: Optional[bool] = None,
     fallback_max_retries: int = 3,
     handler_factory: Optional[Callable[..., Any]] = None,
     **kwargs: Any,
 ) -> SliderVerificationResult:
-    """异步版本：Playwright 严格判定失败后可运行 DrissionPage 兜底。"""
+    """异步版本：远程/Playwright 严格判定失败后可运行 DrissionPage 兜底。"""
+    import asyncio
+
+    user_id = str(getattr(slider, "user_id", None) or getattr(slider, "pure_user_id", None) or "unknown")
+
+    use_remote = _env_bool("XY_SLIDER_REMOTE_ENABLED", False) if remote_enabled is None else bool(remote_enabled)
+    remote_url, remote_secret = remote_config or _remote_config_from_env()
+    if use_remote and remote_url and remote_secret:
+        remote_result = await asyncio.to_thread(
+            _call_remote_solve,
+            url,
+            user_id=user_id,
+            remote_url=remote_url,
+            remote_secret=remote_secret,
+            timeout=remote_timeout,
+        )
+        if remote_result.success:
+            return remote_result
+
     primary_result = await run_slider_async_strict(slider, url, engine=engine, **kwargs)
     if primary_result.success:
         return primary_result
@@ -243,9 +336,6 @@ async def run_slider_async_with_fallback(
     if not enabled:
         return primary_result
 
-    import asyncio
-
-    user_id = str(getattr(slider, "user_id", None) or getattr(slider, "pure_user_id", None) or "unknown")
     existing_cookies_str = str(getattr(slider, "initial_cookies", "") or "")
     headless = bool(getattr(slider, "headless", True)) if fallback_headless is None else bool(fallback_headless)
     fallback_result = await asyncio.to_thread(