641 lines
20 KiB
Python
641 lines
20 KiB
Python
"""虎牙短信登录流程。"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import base64
|
|
import json
|
|
import random
|
|
import time
|
|
import uuid
|
|
from collections.abc import Mapping
|
|
from dataclasses import dataclass
|
|
from urllib.parse import quote, urlparse, urlunparse
|
|
|
|
import requests
|
|
from loguru import logger
|
|
|
|
from .login import (
|
|
APP_ID,
|
|
APP_SIGN,
|
|
APP_VERSION,
|
|
DEFAULT_UA,
|
|
HUYA_PAGE,
|
|
HUYA_PAGE_URL,
|
|
LCID,
|
|
HuyaLoginError,
|
|
cookie_mapping,
|
|
cookie_string,
|
|
generate_context,
|
|
generate_request_id,
|
|
)
|
|
|
|
|
|
SMS_CODE_URI = "60027"
|
|
SMS_LOGIN_URI = "60025"
|
|
SMS_CODE_URL = "https://udblgn.huya.com/web/v2/smsCode"
|
|
SMS_LOGIN_URL = "https://udblgn.huya.com/web/v2/smsLogin"
|
|
DF_TOKEN_URL = "https://df.huya.com/web/df/token"
|
|
DF_COLLECT_URL = "https://df.huya.com/web/df/collect"
|
|
HUYA_COUNTRY_CALLING_CODES = frozenset({
|
|
"1",
|
|
"7",
|
|
"20",
|
|
"27",
|
|
"30",
|
|
"31",
|
|
"32",
|
|
"33",
|
|
"34",
|
|
"36",
|
|
"39",
|
|
"40",
|
|
"41",
|
|
"43",
|
|
"44",
|
|
"45",
|
|
"46",
|
|
"47",
|
|
"48",
|
|
"49",
|
|
"52",
|
|
"55",
|
|
"60",
|
|
"61",
|
|
"62",
|
|
"63",
|
|
"64",
|
|
"65",
|
|
"66",
|
|
"81",
|
|
"82",
|
|
"84",
|
|
"86",
|
|
"90",
|
|
"91",
|
|
"92",
|
|
"93",
|
|
"94",
|
|
"95",
|
|
"98",
|
|
"212",
|
|
"213",
|
|
"216",
|
|
"218",
|
|
"234",
|
|
"254",
|
|
"351",
|
|
"352",
|
|
"353",
|
|
"354",
|
|
"355",
|
|
"356",
|
|
"357",
|
|
"358",
|
|
"359",
|
|
"370",
|
|
"371",
|
|
"372",
|
|
"373",
|
|
"374",
|
|
"375",
|
|
"376",
|
|
"377",
|
|
"378",
|
|
"380",
|
|
"381",
|
|
"382",
|
|
"385",
|
|
"386",
|
|
"387",
|
|
"389",
|
|
"420",
|
|
"421",
|
|
"852",
|
|
"853",
|
|
"855",
|
|
"856",
|
|
"886",
|
|
"960",
|
|
"961",
|
|
"962",
|
|
"963",
|
|
"964",
|
|
"965",
|
|
"966",
|
|
"967",
|
|
"968",
|
|
"971",
|
|
"972",
|
|
"973",
|
|
"974",
|
|
"975",
|
|
"976",
|
|
"977",
|
|
"992",
|
|
"993",
|
|
"994",
|
|
"995",
|
|
"996",
|
|
"998",
|
|
})
|
|
|
|
|
|
@dataclass
|
|
class HuyaSmsCodeResult:
|
|
"""虎牙短信发码结果。"""
|
|
|
|
success: bool
|
|
state: str = ""
|
|
message: str = ""
|
|
session_data: str = ""
|
|
raw: dict | None = None
|
|
sdid: str = ""
|
|
context: str = ""
|
|
request_id: str = ""
|
|
|
|
|
|
def _split_country_calling_code(digits: str, *, allow_single_digit: bool = True) -> tuple[str, str]:
|
|
"""从国际号码中拆出国家/地区区号。
|
|
|
|
allow_single_digit=False 时不匹配 1/7 等单位区号，避免把国内 1 开头手机号误判成美国号。
|
|
"""
|
|
sizes = (3, 2, 1) if allow_single_digit else (3, 2)
|
|
for size in sizes:
|
|
code = digits[:size]
|
|
if code in HUYA_COUNTRY_CALLING_CODES and len(digits) > size:
|
|
return code, digits[size:]
|
|
return "", digits
|
|
|
|
|
|
def _format_huya_international_phone(code: str, national_number: str) -> str:
|
|
"""转换成虎牙短信接口使用的国际号格式。
|
|
|
|
抓包（香港登录 HAR）确认：
|
|
+85291283650 → data.phone = \"0085291283650\"
|
|
即 00 + 区号 + 本地号。
|
|
|
|
中国大陆沿用历史格式：086 + 11 位手机号。
|
|
"""
|
|
code = str(code or "").strip().lstrip("0") or str(code or "").strip()
|
|
national_number = str(national_number or "").strip()
|
|
if not code or not national_number:
|
|
return f"{code}{national_number}"
|
|
# 中国大陆
|
|
if code == "86":
|
|
return f"086{national_number}"
|
|
# 国际号：00 + 国家/地区区号 + 本地号（美加 1 → 001…，香港 852 → 00852…）
|
|
return f"00{code}{national_number}"
|
|
|
|
|
|
def normalize_huya_phone(phone: str) -> str:
|
|
"""归一化虎牙短信接口手机号格式，兼容国内号和国际区号（含香港 +852）。"""
|
|
raw = str(phone or "").strip()
|
|
digits = "".join(ch for ch in raw if ch.isdigit())
|
|
if not digits:
|
|
return ""
|
|
|
|
# 显式国际号：+852… / 00852…
|
|
if raw.startswith("+"):
|
|
code, national_number = _split_country_calling_code(digits)
|
|
if code:
|
|
return _format_huya_international_phone(code, national_number)
|
|
if digits.startswith("00") and len(digits) > 4:
|
|
code, national_number = _split_country_calling_code(digits[2:])
|
|
if code:
|
|
return _format_huya_international_phone(code, national_number)
|
|
|
|
# 已是正确虎牙格式
|
|
# - 国内 086 + 11 位
|
|
if digits.startswith("086") and len(digits) == 14:
|
|
return digits
|
|
# - 国际 00 + 区号…（再走一遍规范化，兼容已是 00852…）
|
|
# 上面 00 分支已处理
|
|
|
|
# 错误的单 0 国际号（历史错误归一 0852…）→ 按 0 后内容重解析
|
|
if digits.startswith("0") and not digits.startswith("00") and len(digits) >= 12:
|
|
rest = digits[1:]
|
|
code, national_number = _split_country_calling_code(rest)
|
|
if code and code != "86" and len(national_number) >= 6:
|
|
return _format_huya_international_phone(code, national_number)
|
|
if digits.startswith("086"):
|
|
return digits
|
|
return digits
|
|
|
|
# 无 + 的裸国际号：
|
|
# - 3 位区号：852/886…
|
|
# - 2 位区号：仅当总长 > 10（印尼 62、英国 44），避免 10 位美国号被 20 误吃
|
|
if len(digits) >= 9:
|
|
code3 = digits[:3]
|
|
national3 = digits[3:]
|
|
if code3 in HUYA_COUNTRY_CALLING_CODES and len(national3) >= 6:
|
|
return _format_huya_international_phone(code3, national3)
|
|
if len(digits) > 10:
|
|
code2 = digits[:2]
|
|
national2 = digits[2:]
|
|
if code2 in HUYA_COUNTRY_CALLING_CODES and len(national2) >= 6:
|
|
return _format_huya_international_phone(code2, national2)
|
|
|
|
# 国内号
|
|
if digits.startswith("86") and len(digits) == 13:
|
|
return f"0{digits}"
|
|
if len(digits) == 11 and digits.startswith(("13", "14", "15", "16", "17", "18", "19")):
|
|
return f"086{digits}"
|
|
if len(digits) == 11:
|
|
return f"086{digits}"
|
|
if len(digits) == 10:
|
|
return f"001{digits}"
|
|
return digits
|
|
|
|
|
|
def encode_sms_behavior(page: str = HUYA_PAGE_URL) -> str:
|
|
"""生成短信登录抓包同款行为轨迹。"""
|
|
now = int(time.time() * 1000) - random.randint(8000, 18000)
|
|
elapsed = random.randint(1200, 2400)
|
|
actions = [{
|
|
"id": "12",
|
|
"x": random.randint(520, 620),
|
|
"y": random.randint(45, 75),
|
|
"d": elapsed,
|
|
"time": now,
|
|
}]
|
|
for action_id in ("17", "17"):
|
|
now += random.randint(1800, 4200)
|
|
elapsed += random.randint(1800, 4200)
|
|
actions.append({"id": action_id, "d": elapsed, "time": now})
|
|
now += random.randint(80, 240)
|
|
elapsed += random.randint(80, 240)
|
|
actions.append({
|
|
"id": "18",
|
|
"x": random.randint(610, 660),
|
|
"y": random.randint(165, 195),
|
|
"d": elapsed,
|
|
"time": now,
|
|
})
|
|
value = {"furl": page, "curl": page, "user_action": actions}
|
|
return quote(json.dumps(value, separators=(",", ":"), ensure_ascii=False), safe="~()*!.'")
|
|
|
|
|
|
class HuyaSmsLogin:
|
|
"""虎牙手机号短信登录器。"""
|
|
|
|
timeout = (8, 20)
|
|
|
|
def __init__(
|
|
self,
|
|
phone: str,
|
|
cookie: Mapping[str, str] | str | None = None,
|
|
ua: str = DEFAULT_UA,
|
|
session: requests.Session | None = None,
|
|
proxies: Mapping[str, str] | None = None,
|
|
timeout: tuple[float, float] | None = None,
|
|
):
|
|
self.phone = normalize_huya_phone(phone)
|
|
self.ua = ua or DEFAULT_UA
|
|
self.timeout = timeout or self.timeout
|
|
self.request_id = generate_request_id()
|
|
self.exchange = self.request_id
|
|
self.device_id = uuid.uuid4().hex
|
|
self.context = generate_context(self.device_id)
|
|
self.middle_url = f"https://udblgn.huya.com/web/middle/{APP_VERSION}/{self.exchange}/https/{self.device_id}"
|
|
self.sdid = ""
|
|
|
|
self.session = session or requests.Session()
|
|
self.session.trust_env = False
|
|
if cookie:
|
|
self.session.cookies.update(cookie_mapping(cookie))
|
|
if proxies:
|
|
self.session.proxies = dict(proxies)
|
|
self._setup_headers()
|
|
|
|
def _setup_headers(self) -> None:
|
|
self.session.headers.update({
|
|
"Accept": "*/*",
|
|
"Accept-Language": "zh-CN,zh;q=0.9",
|
|
"Cache-Control": "no-cache",
|
|
"Connection": "keep-alive",
|
|
"Content-Type": "application/json;charset=UTF-8",
|
|
"Origin": "https://aq.huya.com",
|
|
"Pragma": "no-cache",
|
|
"Referer": "https://aq.huya.com/",
|
|
"Sec-Fetch-Dest": "empty",
|
|
"Sec-Fetch-Mode": "cors",
|
|
"Sec-Fetch-Site": "same-site",
|
|
"User-Agent": self.ua,
|
|
"sec-ch-ua": '"Not;A=Brand";v="8", "Chromium";v="150", "Google Chrome";v="150"',
|
|
"sec-ch-ua-mobile": "?0",
|
|
"sec-ch-ua-platform": '"macOS"',
|
|
})
|
|
|
|
@staticmethod
|
|
def _safe_url(url: str) -> str:
|
|
parsed = urlparse(url)
|
|
return urlunparse((parsed.scheme, parsed.netloc, parsed.path, "", "", ""))
|
|
|
|
@staticmethod
|
|
def _payload_data(payload: dict) -> dict:
|
|
"""兼容虎牙失败响应里的 data=null。"""
|
|
data = payload.get("data")
|
|
return data if isinstance(data, dict) else {}
|
|
|
|
def _request_json(self, method: str, url: str, source: str, **kwargs) -> dict:
|
|
response = self.session.request(method, url, timeout=self.timeout, **kwargs)
|
|
logger.debug(f"{method.upper()} {self._safe_url(url)} -> {response.status_code}")
|
|
response.raise_for_status()
|
|
try:
|
|
return response.json()
|
|
except json.JSONDecodeError as exc:
|
|
preview = response.text[:200].replace("\n", "\\n")
|
|
raise HuyaLoginError(f"{source} 返回不是 JSON: {preview}") from exc
|
|
|
|
def init_udb_middle(self) -> None:
|
|
"""初始化 UDB middle，补齐抓包中的登录 Referer。"""
|
|
for host in ("udblgn.huya.com", "udb3lgn.huya.com", "udbreg.huya.com"):
|
|
url = f"https://{host}/web/middle/{APP_VERSION}/{self.exchange}/https/{self.device_id}"
|
|
try:
|
|
response = self.session.get(url, timeout=self.timeout)
|
|
logger.debug(f"GET {self._safe_url(url)} -> {response.status_code}")
|
|
except requests.RequestException as exc:
|
|
logger.debug(f"虎牙UDB middle初始化失败: {host}: {exc}")
|
|
self.session.headers.update({
|
|
"Origin": "https://udblgn.huya.com",
|
|
"Referer": self.middle_url,
|
|
})
|
|
|
|
def prepare_device(self) -> str:
|
|
"""获取虎牙风控 sdid。"""
|
|
token_payload = {"encryptVersion": "1.0.1", "fingerprintVersion": "1.2.41"}
|
|
token_res = self._request_json("post", DF_TOKEN_URL, "获取虎牙 df token", json=token_payload)
|
|
token = token_res.get("data", {}).get("token")
|
|
if not token:
|
|
raise HuyaLoginError(f"获取虎牙 df token 失败: {token_res}")
|
|
|
|
collect_res = self._request_json("post", DF_COLLECT_URL, "获取虎牙 sdid", json={"token": token})
|
|
self.sdid = collect_res.get("data", {}).get("sdid", "")
|
|
if not self.sdid:
|
|
raise HuyaLoginError(f"获取虎牙 sdid 失败: {collect_res}")
|
|
return self.sdid
|
|
|
|
def _solve_verification(self, payload: dict) -> str:
|
|
"""处理 10030/10039 风控并返回 authId。"""
|
|
from .verification import HuyaVerificationSolver
|
|
|
|
solver = HuyaVerificationSolver(
|
|
cookie=self.session.cookies.get_dict(),
|
|
ua=self.ua,
|
|
sdid=self.sdid,
|
|
session=self.session,
|
|
proxies=self.session.proxies,
|
|
timeout=self.timeout,
|
|
)
|
|
verify_data = solver.solve(payload)
|
|
auth_id = verify_data.get("authId") if isinstance(verify_data, dict) else ""
|
|
if not auth_id:
|
|
raise HuyaLoginError(f"虎牙风控验证未返回 authId: {verify_data}")
|
|
return str(auth_id)
|
|
|
|
def _common_payload(self, uri: str, data: dict) -> dict:
|
|
return {
|
|
"uri": uri,
|
|
"version": APP_VERSION,
|
|
"context": self.context,
|
|
"appId": APP_ID,
|
|
"appSign": APP_SIGN,
|
|
"authId": "",
|
|
"sdid": self.sdid,
|
|
"lcid": LCID,
|
|
"byPass": "3",
|
|
"requestId": self.request_id,
|
|
"data": data,
|
|
}
|
|
|
|
def _headers_for_uri(self, uri: str) -> dict:
|
|
return {
|
|
"Origin": "https://udblgn.huya.com",
|
|
"Referer": self.middle_url,
|
|
"context": self.context,
|
|
"lcid": LCID,
|
|
"reqid": self.request_id,
|
|
"uri": uri,
|
|
}
|
|
|
|
def _sms_code_once(self) -> dict:
|
|
data = {
|
|
"phone": self.phone,
|
|
"behavior": encode_sms_behavior(),
|
|
"page": HUYA_PAGE,
|
|
}
|
|
return self._request_json(
|
|
"post",
|
|
SMS_CODE_URL,
|
|
"虎牙短信发码",
|
|
json=self._common_payload(SMS_CODE_URI, data),
|
|
headers=self._headers_for_uri(SMS_CODE_URI),
|
|
)
|
|
|
|
def send_code(self) -> HuyaSmsCodeResult:
|
|
"""发送短信验证码，必要时自动处理滑块。"""
|
|
logger.info(f"开始虎牙短信发码: {self.phone}")
|
|
self.prepare_device()
|
|
self.init_udb_middle()
|
|
|
|
payload = self._sms_code_once()
|
|
for index in range(3):
|
|
return_code = int(payload.get("returnCode") or 0)
|
|
session_data = str(self._payload_data(payload).get("sessionData") or "")
|
|
if return_code == 0 and session_data:
|
|
state = self._dump_state(session_data)
|
|
logger.info("虎牙短信发码成功")
|
|
return HuyaSmsCodeResult(
|
|
success=True,
|
|
state=state,
|
|
message="短信已发送",
|
|
session_data=session_data,
|
|
raw=payload,
|
|
sdid=self.sdid,
|
|
context=self.context,
|
|
request_id=self.request_id,
|
|
)
|
|
if return_code not in (10030, 10039):
|
|
return HuyaSmsCodeResult(
|
|
success=False,
|
|
message=f"虎牙短信发码失败: {payload}",
|
|
raw=payload,
|
|
sdid=self.sdid,
|
|
context=self.context,
|
|
request_id=self.request_id,
|
|
)
|
|
|
|
logger.info(f"虎牙短信发码触发风控: {return_code} ({index + 1}/3)")
|
|
self._solve_verification(payload)
|
|
time.sleep(0.5)
|
|
payload = self._sms_code_once()
|
|
|
|
return HuyaSmsCodeResult(
|
|
success=False,
|
|
message=f"虎牙短信发码风控后仍失败: {payload}",
|
|
raw=payload,
|
|
sdid=self.sdid,
|
|
context=self.context,
|
|
request_id=self.request_id,
|
|
)
|
|
|
|
def _sms_login_once(self, authcode: str, session_data: str) -> dict:
|
|
data = {
|
|
"phone": self.phone,
|
|
"authId": "",
|
|
"authcode": authcode,
|
|
"sessionData": session_data,
|
|
"domainList": "",
|
|
"remember": "1",
|
|
"behavior": encode_sms_behavior(),
|
|
"page": HUYA_PAGE,
|
|
}
|
|
return self._request_json(
|
|
"post",
|
|
SMS_LOGIN_URL,
|
|
"虎牙短信登录",
|
|
json=self._common_payload(SMS_LOGIN_URI, data),
|
|
headers=self._headers_for_uri(SMS_LOGIN_URI),
|
|
)
|
|
|
|
def submit_code(self, authcode: str, session_data: str = ""):
|
|
"""提交短信验证码并返回登录结果。"""
|
|
from .login import HuyaLoginResult
|
|
|
|
session_data = session_data or getattr(self, "session_data", "")
|
|
if not session_data:
|
|
raise HuyaLoginError("缺少短信登录 sessionData，请先发送短信")
|
|
|
|
payload = self._sms_login_once(authcode, session_data)
|
|
for index in range(3):
|
|
return_code = int(payload.get("returnCode") or 0)
|
|
if return_code == 0:
|
|
cookie = cookie_string(self.session.cookies)
|
|
logger.success(f"虎牙短信登录成功: {self.phone}, Cookie长度: {len(cookie)}")
|
|
return HuyaLoginResult(
|
|
success=True,
|
|
cookie=cookie,
|
|
message="登录成功",
|
|
code=0,
|
|
raw=payload,
|
|
sdid=self.sdid,
|
|
context=self.context,
|
|
request_id=self.request_id,
|
|
)
|
|
if return_code not in (10030, 10039):
|
|
return HuyaLoginResult(
|
|
success=False,
|
|
message=f"虎牙短信登录失败: {payload}",
|
|
code=return_code,
|
|
raw=payload,
|
|
sdid=self.sdid,
|
|
context=self.context,
|
|
request_id=self.request_id,
|
|
)
|
|
|
|
logger.info(f"虎牙短信登录触发风控: {return_code} ({index + 1}/3)")
|
|
self._solve_verification(payload)
|
|
time.sleep(0.5)
|
|
payload = self._sms_login_once(authcode, session_data)
|
|
|
|
return HuyaLoginResult(
|
|
success=False,
|
|
message=f"虎牙短信登录风控后仍失败: {payload}",
|
|
code=int(payload.get("returnCode") or 0),
|
|
raw=payload,
|
|
sdid=self.sdid,
|
|
context=self.context,
|
|
request_id=self.request_id,
|
|
)
|
|
|
|
def _dump_state(self, session_data: str) -> str:
|
|
"""把两步短信登录需要的状态编码给调用方保存。"""
|
|
data = {
|
|
"phone": self.phone,
|
|
"sessionData": session_data,
|
|
"cookies": self.session.cookies.get_dict(),
|
|
"ua": self.ua,
|
|
"sdid": self.sdid,
|
|
"context": self.context,
|
|
"requestId": self.request_id,
|
|
"exchange": self.exchange,
|
|
"deviceId": self.device_id,
|
|
"middleUrl": self.middle_url,
|
|
"createdAt": int(time.time()),
|
|
}
|
|
raw = json.dumps(data, separators=(",", ":"), ensure_ascii=False).encode("utf-8")
|
|
return base64.urlsafe_b64encode(raw).decode("ascii")
|
|
|
|
@classmethod
|
|
def from_state(
|
|
cls,
|
|
state: str,
|
|
phone: str = "",
|
|
proxies: Mapping[str, str] | None = None,
|
|
timeout: tuple[float, float] | None = None,
|
|
) -> "HuyaSmsLogin":
|
|
"""从发码阶段返回的 state 恢复短信登录会话。"""
|
|
try:
|
|
raw = base64.urlsafe_b64decode(state.encode("ascii"))
|
|
data = json.loads(raw.decode("utf-8"))
|
|
except (ValueError, json.JSONDecodeError) as exc:
|
|
raise HuyaLoginError("短信登录 state 无效") from exc
|
|
|
|
state_phone = str(data.get("phone") or "")
|
|
expected_phone = normalize_huya_phone(phone or state_phone)
|
|
if state_phone and expected_phone != state_phone:
|
|
raise HuyaLoginError("短信登录手机号与发码阶段不一致")
|
|
|
|
login = cls(
|
|
phone=expected_phone,
|
|
cookie=data.get("cookies") or {},
|
|
ua=str(data.get("ua") or DEFAULT_UA),
|
|
proxies=proxies,
|
|
timeout=timeout,
|
|
)
|
|
login.sdid = str(data.get("sdid") or "")
|
|
login.context = str(data.get("context") or login.context)
|
|
login.request_id = str(data.get("requestId") or login.request_id)
|
|
login.exchange = str(data.get("exchange") or login.exchange)
|
|
login.device_id = str(data.get("deviceId") or login.device_id)
|
|
login.middle_url = str(data.get("middleUrl") or login.middle_url)
|
|
login.session_data = str(data.get("sessionData") or "")
|
|
login.session.headers.update({
|
|
"Origin": "https://udblgn.huya.com",
|
|
"Referer": login.middle_url,
|
|
})
|
|
return login
|
|
|
|
|
|
def send_huya_sms_code(
|
|
phone: str,
|
|
cookie: Mapping[str, str] | str | None = None,
|
|
ua: str = DEFAULT_UA,
|
|
proxies: Mapping[str, str] | None = None,
|
|
timeout: tuple[float, float] | None = None,
|
|
) -> HuyaSmsCodeResult:
|
|
"""函数式入口：发送虎牙短信验证码。"""
|
|
return HuyaSmsLogin(
|
|
phone=phone,
|
|
cookie=cookie,
|
|
ua=ua,
|
|
proxies=proxies,
|
|
timeout=timeout,
|
|
).send_code()
|
|
|
|
|
|
def login_huya_sms(
|
|
authcode: str,
|
|
state: str,
|
|
phone: str = "",
|
|
proxies: Mapping[str, str] | None = None,
|
|
timeout: tuple[float, float] | None = None,
|
|
):
|
|
"""函数式入口：提交短信验证码登录虎牙。"""
|
|
return HuyaSmsLogin.from_state(
|
|
state=state,
|
|
phone=phone,
|
|
proxies=proxies,
|
|
timeout=timeout,
|
|
).submit_code(authcode=authcode)
|