Files
live-hub-py/core/douyu/login.py
T
yml2213 e778acbccd fix: 极验重试区分代理死亡和临时波动,减少无效换代理
- 新增 _is_proxy_connection_error 判断异常类型
- 代理连接类错误(ProxyError/超时/502):立即换代理+标记坏
- 临时波动(slide/网络不给力/KeyError):先用原代理重试2次,才换代理
- _refresh_proxy 新增 mark_bad 参数,临时换代理不标记坏
- 解决日志中'网络不给力'反复换代理导致池枯竭的问题
2026-06-23 09:39:43 +08:00

709 lines
27 KiB
Python
Raw Blame History

This file contains ambiguous Unicode characters
This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.
"""斗鱼登录核心模块"""
import re
import json
import time
import requests
from typing import Mapping, Optional, Protocol, Tuple
from urllib.parse import urlsplit, urlunsplit
from loguru import logger
from .crypto import encrypt_password, encrypt_nickname_or_phone
from .email_verifier import EmailVerifier
from .proxy import ProxyManager, get_proxy_manager
from core.geetest import run_solver
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,
)
from utils.http_logger import log_http
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"
LOGIN_CALLBACK_API = "https://www.douyu.com/api/passport/login"
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)
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_geetest_retries: int = 5,
max_proxy_retries: int = 10,
whitelist_uid: str = "",
whitelist_ukey: str = "",
proxy_manager: Optional[ProxyManager] = None,
):
self.account = account
self.proxy = proxy
self.timeout = timeout
self.max_geetest_retries = max_geetest_retries
self.max_proxy_retries = max_proxy_retries # 0=无限切换直到成功
self.session = requests.Session()
# 初始化代理管理器(优先使用外部传入的共享实例,避免并发刷新冲突)
if proxy_manager:
self.proxy_manager = proxy_manager
elif proxy_api_url:
self.proxy_manager = get_proxy_manager(
proxy_api_url,
whitelist_uid=whitelist_uid,
whitelist_ukey=whitelist_ukey,
)
else:
self.proxy_manager = None
self._current_proxy_url: Optional[str] = None
self._setup_session()
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,并记录当前代理URL供刷新时mark_bad"""
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')
else:
# 从代理管理器获取代理
if self.proxy_manager:
new_proxy = self.proxy_manager.get_proxy()
if new_proxy:
self.session.proxies = {
'http': new_proxy,
'https': new_proxy,
}
self._current_proxy_url = new_proxy
def _refresh_proxy(self, mark_bad: bool = True) -> Optional[str]:
"""
刷新代理IP。
Args:
mark_bad: 是否标记当前代理为坏。默认True(代理确实不可用时)。
设为False时仅换代理,不标记坏(临时网络波动,代理本身可能没问题)。
"""
if not self.proxy_manager:
return None
if mark_bad and self._current_proxy_url:
self.proxy_manager.mark_bad(self._current_proxy_url)
new_proxy = self.proxy_manager.get_proxy()
if new_proxy:
self._apply_proxy(new_proxy)
logger.info(f"已切换代理: {new_proxy}{' (旧代理已标记坏)' if mark_bad else ' (旧代理保留)'}")
return new_proxy
@staticmethod
def _is_proxy_connection_error(err_str: str) -> bool:
"""判断异常是否为代理连接类错误(代理已死),而非临时网络波动。"""
err_lower = err_str.lower()
# 代理连接/超时类:代理本身不可达
proxy_dead_keywords = [
'proxyerror',
'tunnel connection failed',
'connecttimeouterror',
'connection refused',
'unable to connect to proxy',
'proxy connection',
'502 bad gateway',
'503 service unavailable',
]
if any(kw in err_lower for kw in proxy_dead_keywords):
return True
# SSL/证书错误也视为代理问题
if 'ssl' in err_lower and 'proxy' in err_lower:
return True
return False
@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, max_retries: int = 3, **kwargs) -> requests.Response:
"""
统一发送请求,附带分段超时和更明确的错误信息。
代理连接失败时自动重试获取新的代理IP。
所有请求详情会记录到 HTTP 详情日志。
"""
timeout = kwargs.pop('timeout', self.timeout)
safe_url = self._safe_url(url)
req_data = kwargs.get('data')
req_body = req_data if req_data else kwargs.get('json')
current_proxy = self._current_proxy_url
tag = self.account.username if self.account else ""
for attempt in range(max_retries):
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)"
)
# 记录请求/响应详情
resp_body = response.text[:500] if response.text else ""
log_http(
category="douyu_login",
method=method,
url=safe_url,
request_headers=dict(self.session.headers),
request_body=req_body,
status_code=response.status_code,
response_headers=dict(response.headers),
response_body=resp_body,
duration=elapsed,
proxy=current_proxy,
tag=tag,
)
return response
except requests.Timeout as exc:
elapsed = time.monotonic() - started
log_http(
category="douyu_login", method=method, url=safe_url,
request_headers=dict(self.session.headers), request_body=req_body,
duration=elapsed, error=f"超时 timeout={timeout}: {exc}",
proxy=current_proxy, tag=tag,
)
raise TimeoutError(
f"{method.upper()} {safe_url} 超时,耗时 {elapsed:.1f}s"
f"timeout={timeout}"
) from exc
except requests.ConnectionError as exc:
elapsed = time.monotonic() - started
err_str = str(exc)
is_proxy_err = "proxy" in err_str.lower() or "Proxy" in type(exc).__name__
log_http(
category="douyu_login", method=method, url=safe_url,
request_headers=dict(self.session.headers), request_body=req_body,
duration=elapsed,
error=f"{'代理' if is_proxy_err else ''}连接失败: {err_str[:300]}",
proxy=current_proxy, tag=tag,
)
if is_proxy_err:
logger.warning(f"代理连接失败,尝试 {attempt + 1}/{max_retries}: {exc}")
if attempt < max_retries - 1:
self._refresh_proxy()
current_proxy = self._current_proxy_url
time.sleep(1)
continue
raise ConnectionError(
f"{method.upper()} {safe_url} 代理连接失败,已重试 {max_retries} 次"
) from exc
# 非代理的 ConnectionError 也重试
logger.warning(f"连接失败,尝试 {attempt + 1}/{max_retries}: {exc}")
if attempt < max_retries - 1:
time.sleep(1)
continue
raise ConnectionError(
f"{method.upper()} {safe_url} 连接失败,已重试 {max_retries} 次"
) from exc
except requests.RequestException as exc:
elapsed = time.monotonic() - started
log_http(
category="douyu_login", method=method, url=safe_url,
request_headers=dict(self.session.headers), request_body=req_body,
duration=elapsed, error=f"请求异常: {exc}",
proxy=current_proxy, tag=tag,
)
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:
"""
完整登录流程
Returns:
LoginResult: 登录结果,包含cookie
"""
logger.info(f"开始登录账号: {self.account.username}")
try:
# 1️⃣ 第一次登录(获取极验参数)
logger.info("步骤1: 第一次登录,获取极验参数...")
gt, challenge, code_token, initial_cookies = self._first_login()
# 2️⃣ 极验 fullpage 验证
logger.info("步骤2: 极验 fullpage 验证...")
validate, seccode = self._solve_geetest(gt, challenge)
# 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)
logger.success(f"登录成功! Cookie长度: {len(cookie)}")
return LoginResult(success=True, cookie=cookie, message="登录成功")
except Exception as e:
logger.error(f"登录失败: {e}")
return LoginResult(success=False, message=str(e))
finally:
# 归还代理到池,让其他账号可以复用
if self.proxy_manager and self._current_proxy_url:
self.proxy_manager.release_proxy(self._current_proxy_url)
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) -> Tuple[str, str]:
"""
解决极验 fullpage 验证(带重试机制)
区分两类错误:
- 临时波动(slide、网络不给力、KeyError):先用原代理重试,连续失败才换
- 代理死亡(ProxyError、连接超时):立即换代理
Args:
gt: 极验gt参数
challenge: 极验challenge参数(第一次登录返回的)
Returns:
(validate, seccode)
"""
logger.info("开始极验 fullpage 验证...")
# max_proxy_retries=0 表示无限重试直到成功
max_attempts = self.max_proxy_retries if self.max_proxy_retries > 0 else 999999
# 连续临时失败计数(同一代理下),超过阈值才换代理
_soft_fail_streak = 0
_SOFT_FAIL_THRESHOLD = 2 # 同一代理连续临时失败2次才换
for attempt in range(max_attempts):
try:
if self.max_proxy_retries > 0:
logger.info(f"极验验证尝试 {attempt + 1}/{max_attempts}")
else:
logger.info(f"极验验证尝试 {attempt + 1} (无限重试)")
# 按斗鱼登录页 HAR:fullpage 智能检测流程,不进入图片滑块。
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:
# 极验返回失败(slide 等)—— 通常是临时问题,先原代理重试
_soft_fail_streak += 1
if _soft_fail_streak >= _SOFT_FAIL_THRESHOLD:
logger.warning(f"极验验证失败: {message},同一代理连续 {_soft_fail_streak} 次,换代理")
self._refresh_proxy(mark_bad=False)
_soft_fail_streak = 0
else:
logger.warning(f"极验验证失败: {message},原代理重试 ({_soft_fail_streak}/{_SOFT_FAIL_THRESHOLD})")
time.sleep(1)
continue
else:
validate = str(result)
if validate:
seccode = f"{validate}|jordan"
logger.success(f"极验 fullpage 验证成功! validate={validate[:20]}...")
return validate, seccode
except Exception as e:
err_str = str(e)
is_proxy_dead = self._is_proxy_connection_error(err_str)
if is_proxy_dead:
# 代理确实不可用:立即换,标记坏
logger.warning(f"极验验证代理连接失败: {_truncate_error(err_str)},换代理")
self._refresh_proxy(mark_bad=True)
_soft_fail_streak = 0
else:
# 临时异常(KeyError、网络不给力等):先原代理重试
_soft_fail_streak += 1
if _soft_fail_streak >= _SOFT_FAIL_THRESHOLD:
logger.warning(f"极验验证临时异常: {_truncate_error(err_str)},连续 {_soft_fail_streak} 次,换代理")
self._refresh_proxy(mark_bad=False)
_soft_fail_streak = 0
else:
logger.warning(f"极验验证临时异常: {_truncate_error(err_str)},原代理重试 ({_soft_fail_streak}/{_SOFT_FAIL_THRESHOLD})")
time.sleep(2)
continue
if self.max_proxy_retries > 0:
raise ValueError(f"极验验证失败,已重试 {max_attempts} 次")
raise ValueError("极验验证失败(无限重试模式仍未能通过)")
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)
# 参考HAR文件中的完整参数
# 注意:geetest_challenge应该使用第一次登录返回的challenge
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,
)
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}")
# 收集所有Cookie
cookies = self.session.cookies.get_dict()
# 格式化Cookie字符串
cookie_str = '; '.join([f"{k}={v}" for k, v in cookies.items()])
return cookie_str
def save_cookie(self, cookie: str, filepath: str) -> None:
"""保存Cookie到文件"""
Path(filepath).parent.mkdir(parents=True, exist_ok=True)
with open(filepath, 'w', encoding='utf-8') as f:
json.dump({
'username': self.account.username,
'cookie': cookie,
'timestamp': int(time.time()),
}, f, ensure_ascii=False, indent=2)
logger.info(f"Cookie已保存到: {filepath}")