116 lines
3.0 KiB
Python
116 lines
3.0 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
|
|
"""
|
|
return payload.get('data', {}).get('url', '')
|