fix(douyu): reject invalid activity cookies

core/douyu/activity_client.py

@@ -7,7 +7,7 @@ import re
 import time
 from collections.abc import Callable
 from typing import Any
-from urllib.parse import parse_qs, unquote, urlsplit
+from urllib.parse import parse_qs, unquote, urlsplit, urlunsplit
 
 import requests
 
@@ -118,6 +118,28 @@ class DouyuActivityClient:
             "Cookie": self.cookie,
         }
 
+    @staticmethod
+    def _safe_url(url: str) -> str:
+        """移除 query/fragment，避免将 Cookie、签名或一次性参数写入日志。"""
+        parsed = urlsplit(url)
+        return urlunsplit((parsed.scheme, parsed.netloc, parsed.path, "", ""))
+
+    def _redirect_error(
+        self,
+        method: str,
+        url: str,
+        source: str,
+        response: requests.Response,
+    ) -> DouyuActivityError:
+        location = response.headers.get("Location", "")
+        safe_location = self._safe_url(location) if location else "-"
+        trace = (
+            f"method={method.upper()} url={self._safe_url(url)} "
+            f"location={safe_location} history={response.status_code}->{safe_location}"
+        )
+        self.logger(f"{source}: 已拦截重定向 | {trace}")
+        return DouyuActivityError(f"{source} 请求被重定向，Cookie 已失效，请重新登录（{trace}）")
+
     def _request(
         self,
         method: str,
@@ -130,14 +152,24 @@ class DouyuActivityClient:
         req_headers = self._headers()
         if headers:
             req_headers.update(headers)
+        # 活动 JSON 接口不应自动跳到登录页；忽略调用方传入的覆盖值。
+        kwargs.pop("allow_redirects", None)
+        try:
             response = self.session.request(
                 method,
                 url,
                 headers=req_headers,
                 timeout=kwargs.pop("timeout", self.timeout),
+                allow_redirects=False,
                 **kwargs,
             )
+        except requests.RequestException as exc:
+            raise DouyuActivityError(
+                f"{source} 请求失败: {method.upper()} {self._safe_url(url)}: {exc}"
+            ) from exc
         self._merge_response_cookie(response)
+        if 300 <= response.status_code < 400:
+            raise self._redirect_error(method, url, source, response)
         self.logger(f"{source}: {response.status_code}")
         response.raise_for_status()
         return response

tests/test_douyu_elite_lock_exchange.py

@@ -1,6 +1,8 @@
 import unittest
 from unittest.mock import Mock
 
+import requests
+
 from core.douyu.activity_client import DouyuActivityClient, DouyuActivityError
 
 
@@ -100,6 +102,31 @@ class EliteLockExchangeTests(unittest.TestCase):
             "page": 2, "pageSize": 20, "manualID": "manual",
         })
 
+    def test_redirect_is_not_followed_and_is_reported_without_query_values(self):
+        response = requests.Response()
+        response.status_code = 302
+        response.url = "https://www.douyu.com/japi/example?request_secret=hidden"
+        response.headers["Location"] = "https://www.douyu.com/member/login?ticket=secret"
+        self.client.session.request = Mock(return_value=response)
+        self.client.logger = Mock()
+
+        with self.assertRaisesRegex(DouyuActivityError, "Cookie 已失效，请重新登录") as caught:
+            self.client._request(
+                "get",
+                "https://www.douyu.com/japi/example?request_secret=hidden",
+                source="测试活动接口",
+            )
+
+        self.client.session.request.assert_called_once()
+        self.assertFalse(self.client.session.request.call_args.kwargs["allow_redirects"])
+        message = str(caught.exception)
+        self.assertIn("method=GET", message)
+        self.assertIn("url=https://www.douyu.com/japi/example", message)
+        self.assertIn("location=https://www.douyu.com/member/login", message)
+        self.assertNotIn("request_secret", message)
+        self.assertNotIn("ticket=secret", message)
+        self.assertIn("已拦截重定向", self.client.logger.call_args.args[0])
+
 
 if __name__ == "__main__":
     unittest.main()

tests/test_douyu_gold_recharge_channel.py

@@ -12,7 +12,7 @@ from sqlalchemy import create_engine
 from sqlalchemy.orm import sessionmaker
 
 from web.backend.database import Base
-from web.backend.models import Account, AuditLog, DouyuTask, User
+from web.backend.models import Account, AuditLog, DouyuTask, LoginTask, User
 from web.backend.routers.douyu import get_recharge_channel, supplier_recharge_callback
 from web.backend.services.douyu_runner import DouyuBatchRunner
 from web.backend.services.douyu_service import ensure_douyu_config
@@ -176,6 +176,46 @@ class DouyuGoldRechargeChannelTests(unittest.TestCase):
         self.assertEqual(self.task.status, "failed")
         self.assertEqual(self.task.message, "商品已下架")
 
+    @patch("web.backend.services.douyu_runner.check_douyu_cookie")
+    @patch("web.backend.services.douyu_runner.SessionLocal")
+    def test_invalid_cookie_skips_activity_and_marks_relogin_required(self, session_local, check_cookie):
+        login_task = LoginTask(
+            batch_id="login-batch",
+            account_id=self.account.id,
+            created_by=self.task.created_by,
+            status="success",
+            cookie="acf_uid=10001",
+        )
+        self.task.status = "planned"
+        self.session.add(login_task)
+        self.session.commit()
+        task_id = self.task.id
+        login_task_id = login_task.id
+        session_local.return_value = self.session
+        check_cookie.return_value = {
+            "valid": False,
+            "message": "鱼丸接口: 未登录；等级接口: 未登录",
+            "fish_ball": None,
+            "nickname": None,
+            "level": None,
+            "checked_at": None,
+        }
+        self.runner._execute_create_gold_qr = Mock()
+
+        self.runner._execute_one(task_id, {}, 1)
+
+        self.runner._execute_create_gold_qr.assert_not_called()
+        verify_session = sessionmaker(bind=self.engine)()
+        try:
+            task = verify_session.get(DouyuTask, task_id)
+            checked_login = verify_session.get(LoginTask, login_task_id)
+            self.assertEqual(task.status, "failed")
+            self.assertIn("Cookie 已失效，请重新登录", task.message)
+            self.assertEqual(checked_login.ck_check_status, "invalid")
+            self.assertEqual(checked_login.ck_check_result["message"], "鱼丸接口: 未登录；等级接口: 未登录")
+        finally:
+            verify_session.close()
+
     def test_supplier_callback_verifies_signature_and_updates_terminal_task(self):
         self.task.supplier_out_order_id = f"DYGFBATCHT{self.task.id}"
         self.task.result = {"recharge_channel": "supplier_api", "out_order_id": self.task.supplier_out_order_id}

web/backend/services/douyu_runner.py

@@ -29,10 +29,11 @@ from .douyu_service import (
     account_uid,
     douyu_config_value,
     ensure_douyu_config,
-    latest_success_cookie,
+    latest_success_login_task,
     douyu_task_payload,
     update_account_profile_from_cookie,
 )
+from .cookie_check_service import check_douyu_cookie
 
 
 DOUYU_LEGACY_BIND_ACT_ALIAS = "20250213NQCYX"
@@ -3159,11 +3160,27 @@ class DouyuBatchRunner:
                 current = self._started
 
             self._push_log("info", f"[{current}/{total}] 开始: {self._account_name(account)}")
-            cookie = latest_success_cookie(worker_db, account.id)
+            login_task = latest_success_login_task(worker_db, account.id)
+            cookie = login_task.cookie if login_task else ""
             if not cookie:
                 self._mark_task(worker_db, task, "failed", "账号没有成功登录 Cookie")
                 self._push_log("warning", f"[{current}] {self._account_name(account)} 无 Cookie")
                 return
+
+            cookie_check = check_douyu_cookie(cookie)
+            login_task.ck_check_status = "valid" if cookie_check["valid"] else "invalid"
+            login_task.ck_check_result = {
+                key: cookie_check.get(key)
+                for key in ("fish_ball", "nickname", "level", "message")
+            }
+            login_task.ck_checked_at = cookie_check["checked_at"]
+            worker_db.commit()
+            if not cookie_check["valid"]:
+                message = f"Cookie 已失效，请重新登录：{cookie_check['message']}"
+                self._mark_task(worker_db, task, "failed", message)
+                self._push_log("warning", f"[{current}] {self._account_name(account)} {message}")
+                return
+
             update_account_profile_from_cookie(account, cookie)
 
             handler = {

web/backend/services/douyu_service.py

@@ -145,9 +145,9 @@ def ensure_douyu_config(db: Session) -> DouyuConfig:
     return config
 
 
-def latest_success_cookie(db: Session, account_id: int) -> str:
+def latest_success_login_task(db: Session, account_id: int) -> LoginTask | None:
-    """读取账号最近一次成功登录 Cookie。"""
+    """读取账号最近一次成功登录记录。"""
-    task = (
+    return (
         db.query(LoginTask)
         .filter(
             LoginTask.account_id == account_id,
@@ -157,6 +157,11 @@ def latest_success_cookie(db: Session, account_id: int) -> str:
         .order_by(LoginTask.id.desc())
         .first()
     )
+
+
+def latest_success_cookie(db: Session, account_id: int) -> str:
+    """读取账号最近一次成功登录 Cookie。"""
+    task = latest_success_login_task(db, account_id)
     return task.cookie if task else ""