158 lines
4.6 KiB
Python
158 lines
4.6 KiB
Python
"""登录接口策略基类"""
|
|
|
|
from abc import ABC, abstractmethod
|
|
from typing import Tuple
|
|
|
|
|
|
class LoginAPIStrategy(ABC):
|
|
"""登录接口策略基类
|
|
|
|
定义斗鱼登录各步骤的接口地址和参数构建方式。
|
|
子类需实现 wgapi 和 iframe 两套接口的差异部分。
|
|
"""
|
|
|
|
@property
|
|
@abstractmethod
|
|
def name(self) -> str:
|
|
"""策略名称，如 'wgapi'、'iframe'"""
|
|
...
|
|
|
|
@property
|
|
@abstractmethod
|
|
def first_login_url(self) -> str:
|
|
"""第一次登录接口 URL（获取极验参数）"""
|
|
...
|
|
|
|
@property
|
|
@abstractmethod
|
|
def login_url(self) -> str:
|
|
"""第二次登录接口 URL（带极验验证）"""
|
|
...
|
|
|
|
@property
|
|
@abstractmethod
|
|
def send_email_url(self) -> str:
|
|
"""发送邮箱验证 URL"""
|
|
...
|
|
|
|
@property
|
|
@abstractmethod
|
|
def verify_url(self) -> str:
|
|
"""提交验证码 URL"""
|
|
...
|
|
|
|
@abstractmethod
|
|
def encrypt_username(self, username: str) -> str:
|
|
"""加密用户名（各接口加密方式不同）"""
|
|
...
|
|
|
|
@abstractmethod
|
|
def build_first_login_data(self, username: str, password: str, referer: str) -> dict:
|
|
"""构建第一次登录请求参数（获取极验参数）"""
|
|
...
|
|
|
|
@abstractmethod
|
|
def build_second_login_data(
|
|
self,
|
|
username: str,
|
|
password: str,
|
|
gt: str,
|
|
challenge: str,
|
|
validate: str,
|
|
seccode: str,
|
|
code_token: str,
|
|
referer: str,
|
|
) -> dict:
|
|
"""构建第二次登录请求参数（带极验验证）"""
|
|
...
|
|
|
|
@abstractmethod
|
|
def build_send_email_data(self, code: str) -> dict:
|
|
"""构建发送邮箱验证参数"""
|
|
...
|
|
|
|
@abstractmethod
|
|
def build_verify_data(self, code: str, captcha: str) -> dict:
|
|
"""构建提交验证码参数"""
|
|
...
|
|
|
|
def extract_geetest_params(self, payload: dict) -> Tuple[str, str, str]:
|
|
"""从第一次登录响应提取极验参数
|
|
|
|
Args:
|
|
payload: 登录接口响应 JSON
|
|
|
|
Returns:
|
|
(gt, challenge, code_token)
|
|
"""
|
|
geetest_data = payload.get('data', {}).get('geetest', {})
|
|
code_data = geetest_data.get('code_data', {})
|
|
gt = code_data.get('gt', '')
|
|
challenge = code_data.get('challenge', '')
|
|
code_token = geetest_data.get('code_token', '')
|
|
return gt, challenge, code_token
|
|
|
|
def extract_remote_code(self, payload: dict) -> str:
|
|
"""从第二次登录响应提取 remote_code
|
|
|
|
Args:
|
|
payload: 登录接口响应 JSON
|
|
|
|
Returns:
|
|
remote_code 字符串
|
|
"""
|
|
return payload.get('data', {}).get('remoteLogin', {}).get('code', '')
|
|
|
|
def extract_login_url(self, payload: dict) -> str:
|
|
"""从验证码响应提取登录 URL
|
|
|
|
Args:
|
|
payload: 提交验证码接口响应 JSON
|
|
|
|
Returns:
|
|
登录回调 URL
|
|
"""
|
|
data = payload.get('data', {})
|
|
if not isinstance(data, dict):
|
|
return ''
|
|
|
|
# `remoteLogin` 与“跳过绑定手机号”完成后都会返回同类回调地址，
|
|
# 但不同版本接口的字段命名不完全一致。
|
|
for key in ('url', 'loginUrl', 'login_url', 'redirectUrl', 'redirect_url'):
|
|
value = data.get(key)
|
|
if isinstance(value, str) and value:
|
|
return value
|
|
return ''
|
|
|
|
def extract_mobile_bind_unique_key(self, payload: dict) -> str:
|
|
"""提取服务端要求绑定手机号时返回的一次性继续登录标识。"""
|
|
data = payload.get('data', {})
|
|
if not isinstance(data, dict):
|
|
return ''
|
|
|
|
def find_unique_key(value: object) -> str:
|
|
if isinstance(value, dict):
|
|
unique_key = value.get('uniqueKey') or value.get('unique_key')
|
|
if isinstance(unique_key, str) and unique_key:
|
|
return unique_key
|
|
for child in value.values():
|
|
result = find_unique_key(child)
|
|
if result:
|
|
return result
|
|
elif isinstance(value, (list, tuple)):
|
|
for child in value:
|
|
result = find_unique_key(child)
|
|
if result:
|
|
return result
|
|
return ''
|
|
|
|
return find_unique_key(data)
|
|
|
|
def build_skip_mobile_bind_data(self, unique_key: str) -> dict:
|
|
"""构建跳过手机号绑定、继续完成网页登录的请求参数。"""
|
|
return {
|
|
'type': '3',
|
|
'uniqueKey': unique_key,
|
|
'biz_type': '1',
|
|
}
|