"""斗鱼登录核心模块""" import json import re import threading import time import requests from typing import Mapping, Optional, Protocol, Tuple from urllib.parse import urlsplit, urlunsplit from loguru import logger from .cookie_enricher import CookieEnricher from .crypto import encrypt_password, encrypt_nickname_or_phone from .email_verifier import EmailLoginError, EmailVerifier from .proxy_fetcher import ProxyFetcher from core.geetest.v3_slide.solver import ( _generate_seed, get_w1, get_w2, ) from core.geetest.common.network import ( get_js_address, get_c_s, req_fullpage_validate, ) # ── 全局极验并发限制:同一时刻最多2个线程做极验验证 ── _geetest_semaphore = threading.Semaphore(2) class AccountLike(Protocol): """DouyuLogin 所需的最小账号接口,ORM 对象或 SimpleNamespace 均可满足。""" username: str password: str email: str email_password: str email_imap_server: str email_imap_port: int email_imap_ssl: bool class LoginResult: """登录结果""" def __init__(self, success: bool, cookie: str = "", message: str = "", code: str = ""): self.success = success self.cookie = cookie self.message = message self.code = code # 需要验证时的code class DouyuLogin: """斗鱼登录器""" # 斗鱼API地址 LOGIN_API = "https://passport.douyu.com/wgapi/member/passport/login" SEND_EMAIL_API = "https://passport.douyu.com/wgapi/member/passport/remotelogin/sendemail" VERIFY_API = "https://passport.douyu.com/wgapi/member/passport/remotelogin/verify" WEBLOGIN_API = "https://msg.douyu.com/webLogin" LOGIN_REFERER = ( "https://passport.douyu.com/index/login?" "passport_reg_callback=PASSPORT_REG_SUCCESS_CALLBACK&" "passport_login_callback=PASSPORT_LOGIN_SUCCESS_CALLBACK&" "passport_close_callback=PASSPORT_CLOSE_CALLBACK&" "passport_dp_callback=PASSPORT_DP_CALLBACK&" "type=login&client_id=1&" "state=https%3A%2F%2Fwww.douyu.com%2Fdirectory" ) REQUEST_TIMEOUT = (10, 30) MAX_STATIC_RETRY = 3 # 静态代理失败上限(无法切换代理时的最大重试次数) def __init__( self, account: AccountLike, proxy: Optional[str | Mapping[str, str]] = None, proxy_api_url: Optional[str] = None, timeout: tuple[float, float] = REQUEST_TIMEOUT, max_login_retries: int = 0, max_total_time: float = 0, proxy_fetcher: Optional[ProxyFetcher] = None, stop_event: Optional[threading.Event] = None, ): self.account = account self.proxy = proxy self.timeout = timeout self.max_login_retries = max_login_retries # 0=无限重试直到成功 self.max_total_time = max_total_time # 0=不限制单账号登录总时长 self.stop_event = stop_event self.session = requests.Session() self.proxy_fetcher = proxy_fetcher or ( ProxyFetcher(api_url=proxy_api_url) if proxy_api_url else None ) self._current_proxy_url: Optional[str] = None self._cookie_enrich_error = "" self._static_retry_count = 0 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""" # 只使用配置文件里显式传入的代理,避免系统环境变量悄悄影响请求。 self.session.trust_env = False self.session.headers.update({ 'User-Agent': 'Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/149.0.0.0 Safari/537.36', 'Accept': 'application/json, text/javascript, */*; q=0.01', 'Accept-Language': 'zh-CN,zh;q=0.9', 'Content-Type': 'application/x-www-form-urlencoded; charset=UTF-8', 'Referer': self.LOGIN_REFERER, 'Origin': 'https://passport.douyu.com', 'X-Requested-With': 'XMLHttpRequest', }) # 设置代理 self._apply_proxy() def _apply_proxy(self, proxy: str = None) -> None: """应用代理到Session""" if proxy: # 使用指定的代理 self.session.proxies = { 'http': proxy, 'https': proxy, } self._current_proxy_url = proxy elif self.proxy: # 使用配置的静态代理 if isinstance(self.proxy, str): self.session.proxies = { 'http': self.proxy, 'https': self.proxy, } self._current_proxy_url = self.proxy else: self.session.proxies = { scheme: url for scheme, url in self.proxy.items() if url } self._current_proxy_url = self.proxy.get('http') or self.proxy.get('https') elif self.proxy_fetcher: # 从代理获取器取新代理 new_proxy = self.proxy_fetcher.fetch_new_proxy() if new_proxy: self.session.proxies = { 'http': new_proxy, 'https': new_proxy, } self._current_proxy_url = new_proxy else: logger.warning("获取代理失败,将尝试直连") @staticmethod def _truncate_error(err_str: str, max_len: int = 80) -> str: """截断错误信息,避免日志刷屏。""" if len(err_str) <= max_len: return err_str return err_str[:max_len] + "..." def _safe_url(self, url: str) -> str: """隐藏查询参数,避免日志泄露登录回调 code。""" parsed = urlsplit(url) return urlunsplit((parsed.scheme, parsed.netloc, parsed.path, "", "")) def _request(self, method: str, url: str, **kwargs) -> requests.Response: """ 统一发送请求,附带分段超时和更明确的错误信息。 代理失败/超时直接抛异常,由 login 整体重试换新代理(短效代理失效后重试同一个无意义)。 """ timeout = kwargs.pop('timeout', self.timeout) safe_url = self._safe_url(url) self._ensure_not_stopped() started = time.monotonic() try: response = self.session.request(method, url, timeout=timeout, **kwargs) elapsed = time.monotonic() - started logger.debug( f"{method.upper()} {safe_url} -> {response.status_code} " f"({elapsed:.2f}s)" ) return response except requests.Timeout as exc: elapsed = time.monotonic() - started raise TimeoutError( f"{method.upper()} {safe_url} 超时,耗时 {elapsed:.1f}s," f"timeout={timeout}" ) from exc except requests.ConnectionError as exc: err_str = str(exc) raise ConnectionError( f"{method.upper()} {safe_url} 连接失败: {self._truncate_error(err_str)}" ) from exc except requests.RequestException as exc: elapsed = time.monotonic() - started raise ConnectionError( f"{method.upper()} {safe_url} 请求失败,耗时 {elapsed:.1f}s: {exc}" ) from exc def _request_json(self, method: str, url: str, source: str, **kwargs) -> dict: """请求 JSON 接口,并在响应异常时输出可定位的信息。""" response = self._request(method, url, **kwargs) response.raise_for_status() body = response.text.strip() if not body: raise ValueError(f"{source} 返回为空,无法解析 JSON") try: return response.json() except json.JSONDecodeError as exc: preview = body[:200].replace("\n", "\\n") raise ValueError(f"{source} 返回的不是有效 JSON: {preview}") from exc def login(self) -> LoginResult: """ 完整登录流程(带整体重试)。 每一轮用一个代理走完所有步骤,任何步骤失败就取新代理从头重来。 max_login_retries=0 表示无限重试,max_total_time=0 表示不限制总时长。 Returns: LoginResult: 登录结果,包含cookie """ logger.info(f"开始登录账号: {self.account.username}") start_time = time.monotonic() deadline = start_time + self.max_total_time if self.max_total_time > 0 else 0 attempt = 0 while True: attempt += 1 if self._is_stopped(): logger.warning("登录任务已停止") return LoginResult(success=False, message="任务已停止") elapsed = time.monotonic() - start_time if self.max_total_time > 0 and elapsed > self.max_total_time: logger.warning(f"登录总耗时 {elapsed:.0f}s 超过上限 {self.max_total_time}s,放弃") return LoginResult(success=False, message=f"登录超时({elapsed:.0f}s > {self.max_total_time}s)") if attempt > 1: if self.max_login_retries > 0: logger.info(f"登录整体重试 {attempt}/{self.max_login_retries},换新代理从头开始") else: logger.info(f"登录整体重试 {attempt} (无限重试),换新代理从头开始") # 重试前:取新代理 + 重置 session if not self._prepare_retry(): return LoginResult( success=False, message=f"静态代理连续失败 {self.MAX_STATIC_RETRY} 次,无法切换代理", ) try: # 1️⃣ 第一次登录(获取极验参数) logger.info("步骤1: 第一次登录,获取极验参数...") gt, challenge, code_token, _ = self._first_login() # 2️⃣ 极验 fullpage 验证 logger.info("步骤2: 极验 fullpage 验证...") validate, seccode = self._solve_geetest(gt, challenge, deadline=deadline) # 3️⃣ 第二次登录(带极验) logger.info("步骤3: 第二次登录(带极验验证)...") remote_code = self._second_login(gt, challenge, validate, seccode, code_token) # 4️⃣ 发送邮箱验证 logger.info("步骤4: 发送邮箱验证...") email_sent_at = time.time() self._send_email_verify(remote_code) # 5️⃣ IMAP获取验证码 logger.info("步骤5: 获取邮箱验证码...") verify_code = self._get_email_code(after_timestamp=email_sent_at) # 6️⃣ 提交验证码 logger.info("步骤6: 提交验证码...") login_url = self._submit_verify_code(remote_code, verify_code) # 7️⃣ 完成登录获取Cookie logger.info("步骤7: 完成登录,获取Cookie...") cookie = self._complete_login(login_url) message = "登录成功" if self._cookie_enrich_error: message = f"登录成功,补CK失败: {self._cookie_enrich_error}" logger.success(f"登录成功! Cookie长度: {len(cookie)}") return LoginResult(success=True, cookie=cookie, message=message) except InterruptedError as e: logger.warning(f"登录任务已停止: {e}") return LoginResult(success=False, message=str(e)) except EmailLoginError as e: logger.error(f"登录失败(邮箱登录失败,不再重试): {e}") return LoginResult(success=False, message=str(e), code="email_login_failed") except Exception as e: elapsed = time.monotonic() - start_time if self.max_login_retries > 0: logger.error(f"登录失败(尝试 {attempt}/{self.max_login_retries},已耗时 {elapsed:.0f}s): {e}") else: logger.error(f"登录失败(尝试 {attempt},已耗时 {elapsed:.0f}s): {e}") has_retry = self.max_login_retries <= 0 or attempt < self.max_login_retries has_time = self.max_total_time <= 0 or elapsed < self.max_total_time if has_retry and has_time: self._sleep_interruptible(1) continue # 所有重试耗尽或超时 return LoginResult(success=False, message=str(e)) def _prepare_retry(self) -> bool: """重试前准备:取新代理、重置 session cookies。 Returns: True - 可以继续重试 False - 静态代理已连续失败 MAX_STATIC_RETRY 次,无法切换代理,应放弃 """ self.session.cookies.clear() if self.proxy_fetcher: new_proxy = self.proxy_fetcher.fetch_new_proxy() if new_proxy: self._apply_proxy(new_proxy) logger.info(f"重试换新代理: {new_proxy}") return True if self.proxy: self._static_retry_count += 1 if self._static_retry_count > self.MAX_STATIC_RETRY: logger.error( f"静态代理连续失败 {self.MAX_STATIC_RETRY} 次,无法切换代理" ) return False logger.warning( f"静态代理失败重试 {self._static_retry_count}/{self.MAX_STATIC_RETRY}" ) return True def _first_login(self) -> Tuple[str, str, str, dict]: """ 第一次登录,获取极验参数 Returns: (gt, challenge, code_token, cookies) """ # 加密用户名和密码 encrypted_username = encrypt_nickname_or_phone(self.account.username) encrypted_password = encrypt_password(self.account.password) data = { 'type': '1', 'nicknameOrPhoneEncrypt': encrypted_username, 'password': encrypted_password, 'biz_type': '1', 'room_id': '0', 'redirect_url': self.LOGIN_REFERER, 't': str(int(time.time() * 1000)), 'client_id': '1', 'did': '', 'lang': '', 'isMultiAccount': '0', } payload = self._request_json( 'post', self.LOGIN_API, '第一次登录接口', data=data, ) logger.debug(f"第一次登录响应: {json.dumps(payload, ensure_ascii=False)[:200]}") if payload.get('error') != 81: error_msg = payload.get('msg', '未知错误') raise ValueError(f"第一次登录失败: {error_msg}") # 提取极验参数 geetest_data = payload.get('data', {}).get('geetest', {}) code_data = geetest_data.get('code_data', {}) code_token = geetest_data.get('code_token', '') gt = code_data.get('gt', '') challenge = code_data.get('challenge', '') if not gt or not challenge: raise ValueError("获取极验参数失败") logger.info(f"获取极验参数成功: gt={gt[:10]}..., challenge={challenge[:10]}...") return gt, challenge, code_token, self.session.cookies.get_dict() def _solve_geetest(self, gt: str, challenge: str, deadline: float = 0) -> Tuple[str, str]: """ 解决极验 fullpage 验证(最多3次尝试,失败直接抛异常回到login换新代理) 不在内部做代理切换——短效代理寿命宝贵,换代理由 login 整体重试负责。 Args: gt: 极验gt参数 challenge: 极验challenge参数(第一次登录返回的) deadline: 登录整体超时截止时间(monotonic),0=不限 Returns: (validate, seccode) """ logger.info("开始极验 fullpage 验证...") # ── 全局并发限制:同一时刻最多2个线程做极验,避免被极验限流 ── acquired = _geetest_semaphore.acquire(timeout=120) if not acquired: raise ValueError("极验验证等待超时(并发排队120秒未获得信号量)") try: return self._solve_geetest_inner(gt, challenge, deadline) finally: _geetest_semaphore.release() def _solve_geetest_inner(self, gt: str, challenge: str, deadline: float = 0) -> Tuple[str, str]: """极验验证内部实现:最多3次尝试,失败直接抛异常。""" _MAX_ATTEMPTS = 3 attempt = 0 while True: attempt += 1 self._ensure_not_stopped() if attempt > _MAX_ATTEMPTS: raise ValueError( f"极验验证已尝试 {attempt - 1} 次,超过上限 {_MAX_ATTEMPTS}," f"回到 login 整体重试换新代理" ) # 超时兜底 if deadline and time.monotonic() > deadline: raise ValueError(f"极验验证超时(登录整体时间耗尽)") try: logger.info(f"极验验证尝试 {attempt}/{_MAX_ATTEMPTS}") str_16 = _generate_seed() proxies = dict(self.session.proxies) # 获取JS地址 get_js_address(gt, proxies=proxies) # 获取第一个w值 w1 = get_w1(gt, challenge, str_16) # 获取c和s c, s = get_c_s(gt, challenge, w1, proxies=proxies) # 获取第二个w值 w2 = get_w2(gt, challenge, c, s, str_16) # HAR 中 ajax.php 直接返回 validate。 result = req_fullpage_validate(gt, challenge, w2, proxies=proxies) # 从 fullpage ajax.php 响应中提取 validate if isinstance(result, dict): data = result.get('data', {}) validate = data.get('validate', '') or result.get('validate', '') success = data.get('result') == 'success' or result.get('success') == 1 message = data.get('result') or result.get('message', '') if success and validate: seccode = f"{validate}|jordan" logger.success(f"极验 fullpage 验证成功! validate={validate[:20]}...") return validate, seccode else: logger.warning(f"极验验证失败: {message},回到 login 换新代理") raise ValueError(f"极验验证失败: {message}") else: validate = str(result) if validate: seccode = f"{validate}|jordan" logger.success(f"极验 fullpage 验证成功! validate={validate[:20]}...") return validate, seccode except ValueError: # 极验逻辑失败(slide等)或超过上限,直接上抛 raise except Exception as e: err_str = str(e) # 所有网络/代理异常都直接上抛,由 login 换新代理 logger.warning(f"极验验证异常: {self._truncate_error(err_str)},回到 login 换新代理") raise ValueError(f"极验验证异常: {self._truncate_error(err_str)}") from e def _second_login(self, gt: str, challenge: str, validate: str, seccode: str, code_token: str) -> str: """ 第二次登录(带极验验证) Args: gt: 极验gt参数 challenge: 第一次登录返回的challenge validate: 极验验证返回的validate seccode: 极验验证返回的seccode code_token: 第一次登录返回的code_token Returns: remote_code: 用于邮箱验证的code """ encrypted_username = encrypt_nickname_or_phone(self.account.username) encrypted_password = encrypt_password(self.account.password) data = { 'type': '1', 'nicknameOrPhoneEncrypt': encrypted_username, 'password': encrypted_password, 'room_id': '0', 'code_type': '1', 'code_token': code_token, 'gt_version': 'v3', 'geetest_challenge': challenge, 'geetest_validate': validate, 'geetest_seccode': seccode, 'code_data[geetest_challenge]': challenge, 'code_data[geetest_validate]': validate, 'code_data[geetest_seccode]': seccode, 'code_data[gt_version]': 'v3', 'code_data[code]': '', 'redirect_url': self.LOGIN_REFERER, 't': str(int(time.time() * 1000)), 'client_id': '1', 'did': '', 'lang': '', 'isMultiAccount': '0', 'biz_type': '1', } logger.debug(f"第二次登录参数: challenge={challenge[:20]}..., validate={validate[:20]}...") payload = self._request_json( 'post', self.LOGIN_API, '第二次登录接口', data=data, ) logger.debug(f"第二次登录响应: {json.dumps(payload, ensure_ascii=False)[:200]}") if payload.get('error') != 130014: error_msg = payload.get('msg', '未知错误') raise ValueError(f"第二次登录失败: {error_msg}") # 提取remote_code remote_code = payload.get('data', {}).get('remoteLogin', {}).get('code', '') if not remote_code: raise ValueError("获取remote_code失败") logger.info(f"获取remote_code成功: {remote_code[:20]}...") return remote_code def _send_email_verify(self, remote_code: str) -> None: """发送邮箱验证""" data = { 'code': remote_code, 'client_id': '1', } payload = self._request_json( 'post', self.SEND_EMAIL_API, '发送验证邮件接口', data=data, ) if payload.get('error') != 0: raise ValueError(f"发送验证邮件失败: {payload.get('msg')}") logger.info("验证邮件已发送") def _get_email_code(self, after_timestamp: Optional[float] = None) -> str: """获取邮箱验证码""" verifier = EmailVerifier( imap_server=self.account.email_imap_server, imap_port=self.account.email_imap_port, username=self.account.email, password=self.account.email_password, use_ssl=self.account.email_imap_ssl, ) 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: """ 提交验证码 Returns: login_url: 登录回调URL """ data = { 'verify_type': '2', 'captcha': verify_code, 'isMultiAccount': '0', 'code': remote_code, 'client_id': '1', 'redirect_url': '//www.douyu.com/api/passport/login', } payload = self._request_json( 'post', self.VERIFY_API, '提交验证码接口', data=data, ) if payload.get('error') != 0: raise ValueError(f"提交验证码失败: {payload.get('msg')}") login_url = payload.get('data', {}).get('url', '') if not login_url: raise ValueError("获取登录URL失败") # 补全URL if login_url.startswith('//'): login_url = 'https:' + login_url logger.info(f"获取登录URL成功: {login_url[:50]}...") return login_url def _complete_login(self, login_url: str) -> str: """ 完成登录,获取Cookie Returns: cookie: 完整的Cookie字符串 """ # 访问登录URL response = self._request('get', login_url) response.raise_for_status() # 尝试访问webLogin获取用户信息 try: code_match = re.search(r'code=([^&]+)', login_url) if code_match: code = code_match.group(1) weblogin_url = f"{self.WEBLOGIN_API}?code={code}" response2 = self._request('get', weblogin_url) if response2.status_code == 200: logger.info("WebLogin成功") except Exception as e: logger.warning(f"WebLogin请求失败(不影响登录): {e}") # 登录成功后补齐 Web 侧 Cookie。补 CK 失败只影响完整度,不回滚已成功的登录态。 self._cookie_enrich_error = "" try: CookieEnricher( session=self.session, request_func=self._request, ensure_not_stopped=self._ensure_not_stopped, sleep_interruptible=self._sleep_interruptible, ).enrich_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 _format_cookie_string(self) -> str: """把当前 session 中的 Cookie 格式化为可导出的 CK 字符串。""" cookies = self.session.cookies.get_dict() # 格式化Cookie字符串 cookie_str = '; '.join([f"{k}={v}" for k, v in cookies.items()]) return cookie_str