From 7bb50a9bbf573cabf20812110404afb5cd935118 Mon Sep 17 00:00:00 2001 From: yml2213 Date: Wed, 24 Jun 2026 09:17:15 +0800 Subject: [PATCH] =?UTF-8?q?=E5=A2=9E=E5=BC=BA=E7=99=BB=E5=BD=95=E7=A8=B3?= =?UTF-8?q?=E5=AE=9A=E6=80=A7?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- core/douyu/email_verifier.py | 12 ++++- core/douyu/login.py | 72 +++++++++++++++++++++++---- web/backend/database.py | 19 ++++++- web/backend/services/login_service.py | 17 ++++++- 4 files changed, 106 insertions(+), 14 deletions(-) diff --git a/core/douyu/email_verifier.py b/core/douyu/email_verifier.py index 454b7a3..9345e2e 100644 --- a/core/douyu/email_verifier.py +++ b/core/douyu/email_verifier.py @@ -2,6 +2,7 @@ import html import re +import threading import time from datetime import datetime, timedelta from typing import Optional @@ -261,6 +262,7 @@ class EmailVerifier: interval: int = 3, after_timestamp: Optional[float] = None, allow_old_seconds: int = 15, + stop_event: Optional[threading.Event] = None, ) -> str: """ 轮询获取斗鱼验证码。优先使用 Roundcube API，失败则回退到 read.php。 @@ -270,6 +272,7 @@ class EmailVerifier: interval: 轮询间隔（秒） after_timestamp: 发送验证码请求的时间戳，用于过滤旧邮件 allow_old_seconds: 允许的时间偏移（秒） + stop_event: 外部停止信号，触发后尽快中断等待 Returns: 6位验证码 @@ -281,6 +284,9 @@ class EmailVerifier: tried_roundcube = False while time.monotonic() < deadline: + if stop_event and stop_event.is_set(): + raise InterruptedError("任务已停止") + # 优先尝试 Roundcube if not tried_roundcube or self._rc_logged_in: try: @@ -304,7 +310,11 @@ class EmailVerifier: logger.warning(f"read.php 读邮件异常: {e}") logger.debug("未找到验证码，等待中...") - time.sleep(interval) + sleep_deadline = time.monotonic() + interval + while time.monotonic() < sleep_deadline: + if stop_event and stop_event.is_set(): + raise InterruptedError("任务已停止") + time.sleep(min(0.2, sleep_deadline - time.monotonic())) raise TimeoutError(f"等待验证码超时{'（' + last_error + '）' if last_error else ''}") diff --git a/core/douyu/login.py b/core/douyu/login.py index 601d388..a1e4805 100644 --- a/core/douyu/login.py +++ b/core/douyu/login.py @@ -1,7 +1,8 @@ """斗鱼登录核心模块""" -import re import json +import re +import threading import time import requests from typing import Mapping, Optional, Protocol, Tuple @@ -58,6 +59,8 @@ class DouyuLogin: CSRF_REFERER = "https://www.douyu.com/pages/live-peace-handbook/web/shop?ditchname=pass0&roomId=9263298" ACF_CCN_API = "https://www.douyu.com/curl/csrfNlApi/getCsrfCookie" ACF_CCN_REFERER = "https://www.douyu.com/pages/ord-task-center?clientType=web&panelSource=1&rid=0" + COOKIE_ENRICH_TIMEOUT = (5, 10) + COOKIE_ENRICH_RETRIES = 3 LOGIN_REFERER = ( "https://passport.douyu.com/index/login?" "passport_reg_callback=PASSPORT_REG_SUCCESS_CALLBACK&" @@ -82,6 +85,7 @@ class DouyuLogin: whitelist_uid: str = "", whitelist_ukey: str = "", proxy_manager: Optional[ProxyManager] = None, + stop_event: Optional[threading.Event] = None, ): self.account = account self.proxy = proxy @@ -90,6 +94,7 @@ class DouyuLogin: self.max_proxy_retries = max_proxy_retries # 0=无限切换直到成功 self.max_login_retries = max_login_retries # 登录整体重试次数（换代理从头重跑） self.max_total_time = max_total_time # 单账号登录总时长上限（秒），超时则放弃 + self.stop_event = stop_event self.session = requests.Session() # 初始化代理管理器（优先使用外部传入的共享实例，避免并发刷新冲突） @@ -108,6 +113,28 @@ class DouyuLogin: self._cookie_enrich_error = "" self._setup_session() + def _is_stopped(self) -> bool: + """判断外部批量任务是否请求停止。""" + return bool(self.stop_event and self.stop_event.is_set()) + + def _ensure_not_stopped(self) -> None: + """在长流程边界主动中断。""" + if self._is_stopped(): + raise InterruptedError("任务已停止") + + def _sleep_interruptible(self, seconds: float) -> None: + """可被 stop_event 打断的短睡眠。""" + if seconds <= 0: + self._ensure_not_stopped() + return + deadline = time.monotonic() + seconds + while True: + self._ensure_not_stopped() + remaining = deadline - time.monotonic() + if remaining <= 0: + return + time.sleep(min(0.2, remaining)) + def _setup_session(self) -> None: """配置Session""" # 只使用配置文件里显式传入的代理，避免系统环境变量悄悄影响请求。 @@ -175,6 +202,7 @@ class DouyuLogin: self.proxy_manager.mark_bad(self._current_proxy_url) new_proxy = self.proxy_manager.get_proxy() + self._ensure_not_stopped() if new_proxy: self._apply_proxy(new_proxy) logger.info(f"已切换代理: {new_proxy}{' (旧代理已标记坏)' if mark_bad else ' (旧代理保留)'}") @@ -228,6 +256,7 @@ class DouyuLogin: tag = self.account.username if self.account else "" for attempt in range(max_retries): + self._ensure_not_stopped() started = time.monotonic() try: @@ -281,7 +310,7 @@ class DouyuLogin: if attempt < max_retries - 1: self._refresh_proxy() current_proxy = self._current_proxy_url - time.sleep(1) + self._sleep_interruptible(1) continue raise ConnectionError( f"{method.upper()} {safe_url} 代理连接失败，已重试 {max_retries} 次" @@ -289,7 +318,7 @@ class DouyuLogin: # 非代理的 ConnectionError 也重试 logger.warning(f"连接失败，尝试 {attempt + 1}/{max_retries}: {exc}") if attempt < max_retries - 1: - time.sleep(1) + self._sleep_interruptible(1) continue raise ConnectionError( f"{method.upper()} {safe_url} 连接失败，已重试 {max_retries} 次" @@ -336,6 +365,9 @@ class DouyuLogin: deadline = start_time + self.max_total_time for attempt in range(1, self.max_login_retries + 1): + if self._is_stopped(): + logger.warning("登录任务已停止") + return LoginResult(success=False, message="任务已停止") elapsed = time.monotonic() - start_time if elapsed > self.max_total_time: logger.warning(f"登录总耗时 {elapsed:.0f}s 超过上限 {self.max_total_time}s，放弃") @@ -385,6 +417,9 @@ class DouyuLogin: self.proxy_manager.release_proxy(self._current_proxy_url) return LoginResult(success=True, cookie=cookie, message=message) + except InterruptedError as e: + logger.warning(f"登录任务已停止: {e}") + return LoginResult(success=False, message=str(e)) except Exception as e: elapsed = time.monotonic() - start_time logger.error(f"登录失败（尝试 {attempt}/{self.max_login_retries}，已耗时 {elapsed:.0f}s）: {e}") @@ -392,7 +427,7 @@ class DouyuLogin: # 还有重试机会且未超时：标记当前代理坏，下一轮自动换新代理 if self.proxy_manager and self._current_proxy_url: self.proxy_manager.mark_bad(self._current_proxy_url) - time.sleep(2) + self._sleep_interruptible(2) continue # 所有重试耗尽或超时 return LoginResult(success=False, message=str(e)) @@ -490,6 +525,7 @@ class DouyuLogin: _SOFT_FAIL_THRESHOLD = 2 # 同一代理连续临时失败2次才换 for attempt in range(max_attempts): + self._ensure_not_stopped() # 超时兜底：极验验证不应超过登录整体时间上限 if deadline and time.monotonic() > deadline: raise ValueError(f"极验验证超时（登录整体时间耗尽）") @@ -539,7 +575,7 @@ class DouyuLogin: _soft_fail_streak = 0 else: logger.warning(f"极验验证失败: {message}，原代理重试 ({_soft_fail_streak}/{_SOFT_FAIL_THRESHOLD})") - time.sleep(1) + self._sleep_interruptible(1) continue else: validate = str(result) @@ -566,7 +602,7 @@ class DouyuLogin: _soft_fail_streak = 0 else: logger.warning(f"极验验证临时异常: {self._truncate_error(err_str)}，原代理重试 ({_soft_fail_streak}/{_SOFT_FAIL_THRESHOLD})") - time.sleep(2) + self._sleep_interruptible(2) continue if self.max_proxy_retries > 0: @@ -675,6 +711,7 @@ class DouyuLogin: return verifier.get_verification_code( max_wait=60, after_timestamp=after_timestamp, + stop_event=self.stop_event, ) def _submit_verify_code(self, remote_code: str, verify_code: str) -> str: @@ -744,16 +781,19 @@ class DouyuLogin: self._cookie_enrich_error = "" try: self._enrich_web_cookies_with_retry() + except InterruptedError: + raise except Exception as e: self._cookie_enrich_error = self._truncate_error(str(e), 160) logger.warning(f"补CK最终失败，本次仍按登录成功保存基础CK: {self._cookie_enrich_error}") return self._format_cookie_string() - def _enrich_web_cookies_with_retry(self, max_attempts: int = 3) -> None: + def _enrich_web_cookies_with_retry(self, max_attempts: int = COOKIE_ENRICH_RETRIES) -> None: """独立重试补齐 cvl_csrf_token 和 acf_ccn，不触发整条登录链路重跑。""" last_error: Exception | None = None for attempt in range(1, max_attempts + 1): + self._ensure_not_stopped() try: logger.info(f"补CK尝试 {attempt}/{max_attempts}...") self._generate_csrf_cookie() @@ -764,7 +804,7 @@ class DouyuLogin: last_error = e logger.warning(f"补CK失败 {attempt}/{max_attempts}: {e}") if attempt < max_attempts: - time.sleep(1) + self._sleep_interruptible(1) raise ValueError(f"已重试 {max_attempts} 次仍未补齐CK: {last_error}") from last_error def _generate_csrf_cookie(self) -> str: @@ -786,7 +826,13 @@ class DouyuLogin: 'Content-Type': None, 'X-Requested-With': None, } - response = self._request('post', self.CSRF_API, headers=headers) + response = self._request( + 'post', + self.CSRF_API, + headers=headers, + timeout=self.COOKIE_ENRICH_TIMEOUT, + max_retries=1, + ) response.raise_for_status() body = response.text.strip() @@ -831,7 +877,13 @@ class DouyuLogin: 'Content-Type': None, 'X-Requested-With': None, } - response = self._request('get', self.ACF_CCN_API, headers=headers) + response = self._request( + 'get', + self.ACF_CCN_API, + headers=headers, + timeout=self.COOKIE_ENRICH_TIMEOUT, + max_retries=1, + ) response.raise_for_status() cookies = self.session.cookies.get_dict() diff --git a/web/backend/database.py b/web/backend/database.py index 8299d76..bcf44c1 100644 --- a/web/backend/database.py +++ b/web/backend/database.py @@ -2,7 +2,7 @@ import os from pathlib import Path -from sqlalchemy import create_engine +from sqlalchemy import create_engine, event from sqlalchemy.orm import sessionmaker, declarative_base PROJECT_ROOT = Path(__file__).resolve().parents[2] @@ -10,7 +10,11 @@ DB_PATH = PROJECT_ROOT / "data" / "web.db" DB_PATH.parent.mkdir(parents=True, exist_ok=True) DATABASE_URL = os.getenv("DATABASE_URL", f"sqlite:///{DB_PATH}") -connect_args = {"check_same_thread": False} if DATABASE_URL.startswith("sqlite") else {} +connect_args = ( + {"check_same_thread": False, "timeout": 30} + if DATABASE_URL.startswith("sqlite") + else {} +) engine = create_engine( DATABASE_URL, @@ -18,6 +22,17 @@ engine = create_engine( echo=False, ) + +if DATABASE_URL.startswith("sqlite"): + @event.listens_for(engine, "connect") + def _set_sqlite_pragmas(dbapi_connection, connection_record): + """提升 SQLite 并发写入稳定性。""" + cursor = dbapi_connection.cursor() + cursor.execute("PRAGMA journal_mode=WAL") + cursor.execute("PRAGMA busy_timeout=30000") + cursor.execute("PRAGMA foreign_keys=ON") + cursor.close() + SessionLocal = sessionmaker(bind=engine, autocommit=False, autoflush=False) Base = declarative_base() diff --git a/web/backend/services/login_service.py b/web/backend/services/login_service.py index 5209b0c..eccb0d7 100644 --- a/web/backend/services/login_service.py +++ b/web/backend/services/login_service.py @@ -68,6 +68,15 @@ class LoginBatchRunner: def stop(self): self._stop.set() + def _sleep_or_stop(self, seconds: float) -> bool: + """可中断等待；返回 True 表示收到停止信号。""" + deadline = time.monotonic() + seconds + while time.monotonic() < deadline: + if self._stop.is_set(): + return True + time.sleep(min(0.2, deadline - time.monotonic())) + return self._stop.is_set() + def _push_log(self, level: str, message: str): if self.log_queue and self.loop: asyncio.run_coroutine_threadsafe( @@ -145,7 +154,12 @@ class LoginBatchRunner: task.finished_at = datetime.now(timezone.utc) worker_db.commit() return - time.sleep(10) + if self._sleep_or_stop(10): + task.status = "error" + task.message = "任务已停止" + task.finished_at = datetime.now(timezone.utc) + worker_db.commit() + return test_proxy = self._shared_proxy_manager.get_proxy() if test_proxy: self._shared_proxy_manager.release_proxy(test_proxy) @@ -191,6 +205,7 @@ class LoginBatchRunner: max_login_retries=self.max_login_retries, max_total_time=self.max_total_time, proxy_manager=self._shared_proxy_manager, + stop_event=self._stop, ) result = loginer.login()