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 11:12:36 +08:00
parent 5534293ed1
commit 2824c5a54c
9 changed files with 657 additions and 39 deletions
Download Patch File Download Diff File Expand all files Collapse all files
core/huya/__init__.py
+29
View File
@@ -8,8 +8,13 @@ __all__ = [
"HuyaWssClient",
"GetUserScoreReq",
"GetUserScoreResp",
"HuyaCredentialError",
"HuyaLoginError",
"HuyaLoginResult",
"HuyaPasswordLogin",
"HuyaVerificationError",
"HuyaVerificationSolver",
"login_huya_password",
"solve_huya_verification",
]
@@ -24,6 +29,30 @@ def __getattr__(name: str):
}
globals().update(values)
return values[name]
if name in {
"HuyaCredentialError",
"HuyaLoginError",
"HuyaLoginResult",
"HuyaPasswordLogin",
"login_huya_password",
}:
from .login import (
HuyaCredentialError,
HuyaLoginError,
HuyaLoginResult,
HuyaPasswordLogin,
login_huya_password,
)
values = {
"HuyaCredentialError": HuyaCredentialError,
"HuyaLoginError": HuyaLoginError,
"HuyaLoginResult": HuyaLoginResult,
"HuyaPasswordLogin": HuyaPasswordLogin,
"login_huya_password": login_huya_password,
}
globals().update(values)
return values[name]
if name in {"HuyaVerificationError", "HuyaVerificationSolver", "solve_huya_verification"}:
from .verification import HuyaVerificationError, HuyaVerificationSolver, solve_huya_verification
core/huya/login.py
+370
View File
@@ -0,0 +1,370 @@
"""虎牙密码登录流程。"""
from __future__ import annotations
import datetime as dt
import hashlib
import json
import random
import time
import uuid
from collections.abc import Mapping
from dataclasses import dataclass
from http.cookies import SimpleCookie
from urllib.parse import quote, urlsplit, urlunsplit
import requests
from loguru import logger
APP_ID = "5002"
APP_VERSION = "2.6"
APP_SIGN = "1ce3bf682483d03f146f58232ec10635"
LCID = "2052"
PASSWORD_LOGIN_URI = "30001"
DF_TOKEN_URL = "https://df.huya.com/web/df/token"
DF_COLLECT_URL = "https://df.huya.com/web/df/collect"
PASSWORD_LOGIN_URL = "https://udblgn.huya.com/web/v2/passwordLogin"
HUYA_PAGE_URL = "https://www.huya.com/g"
HUYA_PAGE = "https%3A%2F%2Fwww.huya.com%2Fg"
DEFAULT_UA = (
"Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) "
"AppleWebKit/537.36 (KHTML, like Gecko) "
"Chrome/150.0.0.0 Safari/537.36"
)
class HuyaLoginError(RuntimeError):
"""虎牙登录失败。"""
class HuyaCredentialError(HuyaLoginError):
"""账号或密码错误，不应继续重试。"""
@dataclass
class HuyaLoginResult:
"""虎牙登录结果。"""
success: bool
cookie: str = ""
message: str = ""
code: int | str = ""
raw: dict | None = None
sdid: str = ""
context: str = ""
request_id: str = ""
def password_sha1(password: str) -> str:
"""虎牙 passwordLogin 使用的密码摘要。"""
return hashlib.sha1(password.encode("utf-8")).hexdigest()
def cookie_string(cookies: requests.cookies.RequestsCookieJar | Mapping[str, str]) -> str:
"""把 CookieJar/dict 转成浏览器 Cookie 字符串。"""
items = cookies.items() if isinstance(cookies, Mapping) else cookies.items()
return "; ".join(f"{key}={value}" for key, value in items if value is not None)
def cookie_mapping(cookie: Mapping[str, str] | str | None) -> dict[str, str]:
"""兼容 dict 和 Cookie 字符串。"""
if not cookie:
return {}
if isinstance(cookie, str):
parsed = SimpleCookie()
parsed.load(cookie)
return {key: morsel.value for key, morsel in parsed.items()}
return {str(key): str(value) for key, value in cookie.items()}
def generate_context(device_id: str | None = None, mid: str | None = None) -> str:
"""生成抓包同款虎牙 web 登录 context。"""
device_id = device_id or uuid.uuid4().hex
mid = mid or uuid.uuid4().hex.upper()
return f"WB-{device_id}-{mid}-"
def generate_request_id() -> str:
"""生成 requestId，形态参考旧实现的日内毫秒数。"""
now = dt.datetime.now(dt.timezone.utc)
midnight = now.replace(hour=0, minute=0, second=0, microsecond=0)
return str(int((now - midnight).total_seconds() * 1000))
def encode_behavior(page: str = HUYA_PAGE_URL) -> str:
"""生成基础行为轨迹。"""
now = int(time.time() * 1000) - random.randint(3000, 12000)
actions = []
elapsed = random.randint(900, 2200)
for action_id in ("7", "7", "8", "8"):
now += random.randint(700, 2500)
elapsed += random.randint(700, 2500)
actions.append({"id": action_id, "d": elapsed, "time": now})
now += random.randint(120, 400)
elapsed += random.randint(120, 400)
actions.append({
"id": "11",
"x": random.randint(430, 560),
"y": random.randint(250, 330),
"d": elapsed,
"time": now,
})
value = {"furl": page, "curl": page, "user_action": actions}
return quote(json.dumps(value, separators=(",", ":"), ensure_ascii=False), safe="~()*!.'")
class HuyaPasswordLogin:
"""虎牙账号密码登录器。"""
timeout = (8, 20)
def __init__(
self,
username: str,
password: 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.username = username
self.password = password
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.mid = uuid.uuid4().hex.upper()
self.context = generate_context(self.device_id, self.mid)
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 = urlsplit(url)
return urlunsplit((parsed.scheme, parsed.netloc, parsed.path, "", ""))
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 _password_payload(self, auth_id: str = "", session_data: str = "") -> dict:
data = {
"userName": self.username,
"password": password_sha1(self.password),
"domainList": "",
"remember": "1",
"behavior": encode_behavior(),
"randomStr": "",
"page": HUYA_PAGE,
}
if auth_id or session_data:
data["authcode"] = ""
data["sessionData"] = session_data
return {
"uri": PASSWORD_LOGIN_URI,
"version": APP_VERSION,
"context": self.context,
"appId": APP_ID,
"appSign": APP_SIGN,
"authId": auth_id,
"sdid": self.sdid,
"lcid": LCID,
"byPass": "3",
"requestId": self.request_id,
"data": data,
}
def _password_login_once(self, auth_id: str = "", session_data: str = "") -> dict:
headers = {
"Origin": "https://udblgn.huya.com",
"Referer": self.middle_url,
"context": self.context,
"lcid": LCID,
"reqid": self.request_id,
"uri": PASSWORD_LOGIN_URI,
}
return self._request_json(
"post",
PASSWORD_LOGIN_URL,
"虎牙密码登录",
json=self._password_payload(auth_id=auth_id, session_data=session_data),
headers=headers,
)
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)
@staticmethod
def _is_credential_error(payload: dict) -> bool:
text = json.dumps(payload, ensure_ascii=False)
return any(word in text for word in ("密码错误", "账号或密码", "账号不存在", "用户不存在"))
@staticmethod
def _payload_data(payload: dict) -> dict:
"""兼容虎牙失败响应里的 data=null。"""
data = payload.get("data")
return data if isinstance(data, dict) else {}
def login(self) -> HuyaLoginResult:
"""执行密码登录，成功返回 Cookie。"""
logger.info(f"开始虎牙密码登录: {self.username}")
self.prepare_device()
self.init_udb_middle()
payload = self._password_login_once()
return_code = int(payload.get("returnCode") or 0)
session_data = str(self._payload_data(payload).get("sessionData") or "")
if return_code == 0:
return self._success(payload)
if self._is_credential_error(payload):
raise HuyaCredentialError(f"虎牙账号或密码错误: {payload}")
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}")
auth_id = self._solve_verification(payload)
for index in range(3):
payload = self._password_login_once(auth_id=auth_id, session_data=session_data)
return_code = int(payload.get("returnCode") or 0)
if return_code == 0:
return self._success(payload)
if self._is_credential_error(payload):
raise HuyaCredentialError(f"虎牙账号或密码错误: {payload}")
session_data = str(self._payload_data(payload).get("sessionData") or session_data)
if return_code in (10030, 10039):
logger.info(f"虎牙密码登录二次风控: {return_code} ({index + 1}/3)")
auth_id = self._solve_verification(payload)
continue
break
return HuyaLoginResult(
success=False,
message=f"虎牙密码登录失败: {payload}",
code=return_code,
raw=payload,
sdid=self.sdid,
context=self.context,
request_id=self.request_id,
)
def _success(self, payload: dict) -> HuyaLoginResult:
cookie = cookie_string(self.session.cookies)
logger.success(f"虎牙密码登录成功: {self.username}, 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,
)
def login_huya_password(
username: str,
password: str,
cookie: Mapping[str, str] | str | None = None,
ua: str = DEFAULT_UA,
proxies: Mapping[str, str] | None = None,
timeout: tuple[float, float] | None = None,
) -> HuyaLoginResult:
"""函数式入口，便于服务层直接调用。"""
return HuyaPasswordLogin(
username=username,
password=password,
cookie=cookie,
ua=ua,
proxies=proxies,
timeout=timeout,
).login()
core/huya/verification/solver.py
+55 -9
View File
@@ -222,6 +222,12 @@ class HuyaVerificationSolver:
self.strategy_url = strategy_url
if not self.url_param_map.get("param"):
raise HuyaVerificationError("虎牙验证链接缺少 param")
logger.debug(
"虎牙验证参数: sceneType={}, mobile={}, param长度={}",
self.scene_type,
self.url_param_map.get("mobile", ""),
len(self.url_param_map.get("param", "")),
)
def _restore_slider_background(self, bg_bytes: bytes, segment: str, huyapk: str) -> bytes:
recover_list = self.js_ctx.call("get_regoin_arr", segment, huyapk)
@@ -236,6 +242,7 @@ class HuyaVerificationSolver:
return output.getvalue()
def _get_track_point(self, url: str, payload: dict):
logger.debug("虎牙验证获取图片: {}", url)
result = self._request_json("post", url, json=payload)
if result.get("returnCode") != 0 or not result.get("data", {}).get("code"):
raise HuyaVerificationError(f"获取验证码图片失败: {result}")
@@ -254,7 +261,9 @@ class HuyaVerificationSolver:
yolo_x = int(slider_box[0] - 13)
opencv_x = self._opencv_slider_x(tip_bytes, restored_bg)
if abs(opencv_x - yolo_x) < 10:
logger.debug("虎牙滑块识别距离: yolo={}, opencv={}", yolo_x, opencv_x)
return int((opencv_x + yolo_x) / 2)
logger.debug("虎牙滑块识别距离: yolo={}, opencv={}, 使用yolo", yolo_x, opencv_x)
return yolo_x
logger.info("虎牙验证类型: 点选")
@@ -334,23 +343,60 @@ class HuyaVerificationSolver:
return self._submit_verify(url, payload)
def _submit_verify(self, url: str, payload: dict) -> dict:
logger.debug("虎牙验证提交: {}", url)
body = json.dumps(payload, separators=(",", ":"))
result = self._request_json("post", url, params={"lock": "true"}, data=body)
return result.get("data") or result
data = result.get("data") or result
if isinstance(data, dict):
logger.debug(
"虎牙验证提交结果: authId={}, nextAuthUrl={}",
bool(data.get("authId")),
bool(data.get("nextAuthUrl")),
)
return data
def _solve_current_strategy(self) -> dict:
"""提交当前 strategy_url 对应的验证。"""
config_payload = self._build_common_payload({"cTuIndex": 0})
config = self._request_json("post", "https://udbrtt.huya.com/auth/client/config3", json=config_payload)
try:
gurl = config["data"]["gurl"]
vurl = config["data"]["vurl"]
dx_js = config["data"]["js"]
except KeyError as exc:
raise HuyaVerificationError(f"config3 取参失败: {config}") from exc
config = {}
gurl = ""
vurl = ""
dx_js = ""
for index in range(2):
logger.debug("虎牙验证获取config3 ({}/2)", index + 1)
config = self._request_json("post", "https://udbrtt.huya.com/auth/client/config3", json=config_payload)
data = config.get("data") if isinstance(config.get("data"), dict) else {}
gurl = str(data.get("gurl") or "")
vurl = str(data.get("vurl") or "")
dx_js = str(data.get("js") or "")
logger.debug(
"虎牙config3返回: returnCode={}, desc={}, gurl={}, vurl={}, js长度={}",
config.get("returnCode"),
config.get("description") or config.get("message") or "",
bool(gurl),
bool(vurl),
len(dx_js),
)
if gurl and vurl and dx_js:
break
if index == 0 and self.url_param_map.get("sceneType") != "4":
self.url_param_map["sceneType"] = "4"
self.scene_type = "4"
config_payload = self._build_common_payload({"cTuIndex": 0})
logger.debug("虎牙config3取参不完整，改用旧流程sceneType=4重试")
time.sleep(0.3)
if not gurl or not vurl or not dx_js:
raise HuyaVerificationError(
"config3取参失败: "
f"returnCode={config.get('returnCode')}, "
f"description={config.get('description') or config.get('message') or ''}, "
f"gurl={bool(gurl)}, vurl={bool(vurl)}, js={bool(dx_js)}"
)
self._compile_js(dx_js)
point = self._get_track_point(gurl, config_payload)
point_payload = self._build_common_payload({"ver": 5})
point = self._get_track_point(gurl, point_payload)
logger.debug(f"虎牙验证识别结果: {point}")
if isinstance(point, list):
return self._send_click_verify(vurl, point)