yml/live-hub-py
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

添加虎牙短信登录

Browse Source
This commit is contained in:
yml2213
2026-07-05 23:10:50 +08:00
parent 3928b43460
commit 3f50d962a0
7 changed files with 742 additions and 2 deletions
Download Patch File Download Diff File Expand all files Collapse all files
core/huya/__init__.py
+25
View File
@@ -12,9 +12,13 @@ __all__ = [
"HuyaLoginError",
"HuyaLoginResult",
"HuyaPasswordLogin",
"HuyaSmsCodeResult",
"HuyaSmsLogin",
"HuyaVerificationError",
"HuyaVerificationSolver",
"login_huya_password",
"login_huya_sms",
"send_huya_sms_code",
"solve_huya_verification",
]
@@ -53,6 +57,27 @@ def __getattr__(name: str):
}
globals().update(values)
return values[name]
if name in {
"HuyaSmsCodeResult",
"HuyaSmsLogin",
"login_huya_sms",
"send_huya_sms_code",
}:
from .sms_login import (
HuyaSmsCodeResult,
HuyaSmsLogin,
login_huya_sms,
send_huya_sms_code,
)
values = {
"HuyaSmsCodeResult": HuyaSmsCodeResult,
"HuyaSmsLogin": HuyaSmsLogin,
"login_huya_sms": login_huya_sms,
"send_huya_sms_code": send_huya_sms_code,
}
globals().update(values)
return values[name]
if name in {"HuyaVerificationError", "HuyaVerificationSolver", "solve_huya_verification"}:
from .verification import HuyaVerificationError, HuyaVerificationSolver, solve_huya_verification
core/huya/sms_login.py
+457
View File
@@ -0,0 +1,457 @@
"""虎牙短信登录流程。"""
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"
@dataclass
class HuyaSmsCodeResult:
"""虎牙短信发码结果。"""
success: bool
state: str = ""
message: str = ""
session_data: str = ""
raw: dict | None = None
sdid: str = ""
context: str = ""
request_id: str = ""
def normalize_huya_phone(phone: str) -> str:
"""归一化虎牙短信接口手机号格式。"""
digits = "".join(ch for ch in str(phone or "") if ch.isdigit())
if digits.startswith("086"):
return digits
if digits.startswith("86") and len(digits) == 13:
return f"0{digits}"
if len(digits) == 11:
return f"086{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)