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-06 01:20:22 +08:00
parent af98ea8416
commit db4ca78280
8 changed files with 777 additions and 29 deletions
Download Patch File Download Diff File Expand all files Collapse all files
core/huya/auto_register.py
+100 -2
View File
@@ -9,6 +9,8 @@ from datetime import datetime
from core.sms_provider import SmsLine, SmsProviderClient
from .change_password import change_huya_password_with_sms_line, generate_huya_password
from .cookie_utils import cookie_value
from .login import HuyaLoginError
from .sms_login import login_huya_sms, normalize_huya_phone, send_huya_sms_code
@@ -24,9 +26,16 @@ class HuyaAutoRegisterResult:
message: str
cookie: str = ""
code: str = ""
change_code: str = ""
sdid: str = ""
normalized_phone: str = ""
username: str = ""
uid: str = ""
password: str = ""
sms_url: str = ""
attempts: int = 0
change_attempts: int = 0
password_changed: bool = False
def _sleep_or_stop(stop_event: threading.Event | None, seconds: float) -> bool:
@@ -43,6 +52,9 @@ def register_huya_with_sms_line(
item: SmsLine,
wait_seconds: float = 180,
poll_interval: float = 5,
change_password: bool = True,
password_prefix: str = "hy",
fixed_password: str = "",
stop_event: threading.Event | None = None,
) -> HuyaAutoRegisterResult:
"""使用固定手机号和接码地址完成虎牙短信注册/登录。"""
@@ -59,6 +71,7 @@ def register_huya_with_sms_line(
status="stopped",
message="已停止",
normalized_phone=normalized_phone,
sms_url=item.url,
)
sent_at = datetime.now()
@@ -72,6 +85,7 @@ def register_huya_with_sms_line(
status="error",
message=str(exc),
normalized_phone=normalized_phone,
sms_url=item.url,
)
except Exception as exc:
return HuyaAutoRegisterResult(
@@ -81,6 +95,7 @@ def register_huya_with_sms_line(
status="error",
message=f"发送虎牙短信失败: {exc}",
normalized_phone=normalized_phone,
sms_url=item.url,
)
if not code_result.success or not code_result.state:
@@ -92,6 +107,7 @@ def register_huya_with_sms_line(
message=code_result.message or "发送虎牙短信失败",
sdid=code_result.sdid,
normalized_phone=normalized_phone,
sms_url=item.url,
)
client = SmsProviderClient()
@@ -109,6 +125,7 @@ def register_huya_with_sms_line(
message="已停止",
sdid=code_result.sdid,
normalized_phone=normalized_phone,
sms_url=item.url,
attempts=attempts,
)
@@ -132,6 +149,7 @@ def register_huya_with_sms_line(
code=poll_result.code,
sdid=code_result.sdid,
normalized_phone=normalized_phone,
sms_url=item.url,
attempts=attempts,
)
except Exception as exc:
@@ -144,21 +162,98 @@ def register_huya_with_sms_line(
code=poll_result.code,
sdid=code_result.sdid,
normalized_phone=normalized_phone,
sms_url=item.url,
attempts=attempts,
)
if login_result.success and login_result.cookie:
cookie = login_result.cookie
uid = cookie_value(cookie, "udb_uid") or cookie_value(cookie, "yyuid")
username = cookie_value(cookie, "udb_passport") or cookie_value(cookie, "username") or uid
if not change_password:
return HuyaAutoRegisterResult(
phone=phone,
provider=item.provider,
success=True,
status="success",
message="注册/登录成功",
cookie=cookie,
code=poll_result.code,
sdid=login_result.sdid or code_result.sdid,
normalized_phone=normalized_phone,
username=username,
uid=uid,
sms_url=item.url,
attempts=attempts,
)
if not uid:
return HuyaAutoRegisterResult(
phone=phone,
provider=item.provider,
success=False,
status="error",
message="注册成功但 Cookie 中没有识别到虎牙 uid，无法改密",
cookie=cookie,
code=poll_result.code,
sdid=login_result.sdid or code_result.sdid,
normalized_phone=normalized_phone,
username=username,
uid=uid,
sms_url=item.url,
attempts=attempts,
)
password = fixed_password.strip() or generate_huya_password(password_prefix)
change_result = change_huya_password_with_sms_line(
uid=uid,
cookie=cookie,
password=password,
item=item,
wait_seconds=wait_seconds,
poll_interval=poll_interval,
ignore_codes={poll_result.code},
stop_event=stop_event,
)
if not change_result.success:
failed_status = "stopped" if change_result.message == "已停止" else "error"
return HuyaAutoRegisterResult(
phone=phone,
provider=item.provider,
success=False,
status=failed_status,
message=f"注册成功但改密失败: {change_result.message}",
cookie=cookie,
code=poll_result.code,
change_code=change_result.code,
sdid=change_result.sdid or login_result.sdid or code_result.sdid,
normalized_phone=normalized_phone,
username=username,
uid=uid,
password=password,
sms_url=item.url,
attempts=attempts,
change_attempts=change_result.attempts,
)
return HuyaAutoRegisterResult(
phone=phone,
provider=item.provider,
success=True,
status="success",
message="注册/登录成功",
cookie=login_result.cookie,
message="注册并改密成功",
cookie=cookie,
code=poll_result.code,
change_code=change_result.code,
sdid=login_result.sdid or code_result.sdid,
normalized_phone=normalized_phone,
username=username,
uid=uid,
password=password,
sms_url=item.url,
attempts=attempts,
change_attempts=change_result.attempts,
password_changed=True,
)
return HuyaAutoRegisterResult(
phone=phone,
@@ -169,6 +264,7 @@ def register_huya_with_sms_line(
code=poll_result.code,
sdid=login_result.sdid or code_result.sdid,
normalized_phone=normalized_phone,
sms_url=item.url,
attempts=attempts,
)
@@ -184,6 +280,7 @@ def register_huya_with_sms_line(
message="已停止",
sdid=code_result.sdid,
normalized_phone=normalized_phone,
sms_url=item.url,
attempts=attempts,
)
@@ -195,5 +292,6 @@ def register_huya_with_sms_line(
message=f"等待验证码超时: {last_message}",
sdid=code_result.sdid,
normalized_phone=normalized_phone,
sms_url=item.url,
attempts=attempts,
)
core/huya/change_password.py
+530
View File
@@ -0,0 +1,530 @@
"""虎牙短信改密流程。"""
from __future__ import annotations
import json
import random
import string
import threading
import time
import uuid
from collections.abc import Mapping
from dataclasses import dataclass
from datetime import datetime
from urllib.parse import quote, urlsplit, urlunsplit
import requests
from loguru import logger
from core.sms_provider import SmsLine, SmsProviderClient
from .login import (
APP_ID,
APP_SIGN,
DEFAULT_UA,
DF_COLLECT_URL,
DF_TOKEN_URL,
HuyaLoginError,
cookie_mapping,
generate_context,
generate_request_id,
)
CHANGE_PASSWORD_VERSION = "2.5"
CHANGE_PASSWORD_CHECK_URI = "60011"
CHANGE_PASSWORD_SEND_SMS_URI = "60003"
CHANGE_PASSWORD_SUBMIT_URI = "60007"
CHANGE_PASSWORD_CHECK_URL = "https://udbreg.huya.com/web/v2/modifyPswByCheck"
CHANGE_PASSWORD_SEND_SMS_URL = "https://udbreg.huya.com/sms/v2/send/changePsw"
CHANGE_PASSWORD_SUBMIT_URL = "https://udbreg.huya.com/web/v2/changePswBySms"
CHANGE_PASSWORD_PAGE_URL = "https://aq.huya.com/i/modify.html"
CHANGE_PASSWORD_FROM_URL = "https://i.huya.com/"
CHANGE_PASSWORD_PAGE = quote(CHANGE_PASSWORD_FROM_URL, safe="~()*!.'")
LCID = "2052"
@dataclass
class HuyaChangePasswordCodeResult:
"""虎牙改密短信发送结果。"""
success: bool
session_data: str = ""
message: str = ""
raw: dict | None = None
sdid: str = ""
context: str = ""
request_id: str = ""
@dataclass
class HuyaChangePasswordResult:
"""虎牙改密结果。"""
success: bool
message: str = ""
code: str = ""
raw: dict | None = None
sdid: str = ""
context: str = ""
request_id: str = ""
attempts: int = 0
def generate_huya_password(prefix: str = "hy", random_length: int = 8) -> str:
"""生成适合虎牙账号使用的随机密码。"""
clean_prefix = "".join(ch for ch in str(prefix or "hy") if ch.isalnum())[:8] or "hy"
alphabet = string.ascii_lowercase + string.digits
suffix = "".join(random.SystemRandom().choice(alphabet) for _ in range(max(6, random_length)))
return f"{clean_prefix}{suffix}"
def encode_change_password_behavior(stage: str) -> str:
"""生成改密页行为轨迹。"""
now = int(time.time() * 1000) - random.randint(2500, 9000)
elapsed = random.randint(900, 1800)
actions: list[dict] = []
if stage in {"send", "submit"}:
actions.append({
"id": "pmodify.btn.getsms",
"x": random.randint(470, 520),
"y": random.randint(120, 145),
"d": elapsed,
"time": now,
})
if stage == "submit":
for action_id in ("pmodify.input.sms", "pmodify.input.pw", "pmodify.input.sms"):
now += random.randint(600, 3200)
elapsed += random.randint(600, 3200)
actions.append({"id": action_id, "d": elapsed, "time": now})
now += random.randint(800, 2400)
elapsed += random.randint(800, 2400)
actions.append({
"id": "pmodify.btn.submit",
"x": random.randint(395, 445),
"y": random.randint(220, 245),
"d": elapsed,
"time": now,
})
value = {
"furl": CHANGE_PASSWORD_FROM_URL,
"curl": CHANGE_PASSWORD_PAGE_URL,
"user_action": actions,
}
return quote(json.dumps(value, separators=(",", ":"), ensure_ascii=False), safe="~()*!.'")
def _sleep_or_stop(stop_event: threading.Event | None, seconds: float) -> bool:
if seconds <= 0:
return not (stop_event and stop_event.is_set())
if stop_event:
return not stop_event.wait(seconds)
time.sleep(seconds)
return True
def _payload_data(payload: dict) -> dict:
data = payload.get("data")
return data if isinstance(data, dict) else {}
def _session_data(payload: dict, fallback: str = "") -> str:
data = _payload_data(payload)
return str(data.get("sessionData") or data.get("sessiondata") or payload.get("sessionData") or fallback or "")
class HuyaPasswordChanger:
"""虎牙已登录账号短信改密。"""
timeout = (8, 20)
def __init__(
self,
uid: 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.uid = str(uid or "").strip()
if not self.uid:
raise HuyaLoginError("缺少虎牙 uid，无法修改密码")
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://udbreg.huya.com/web/middle/{CHANGE_PASSWORD_VERSION}/"
f"{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": "application/json, text/plain, */*",
"Accept-Language": "zh-CN,zh;q=0.9",
"Cache-Control": "no-cache",
"Connection": "keep-alive",
"Content-Type": "application/json;charset=UTF-8",
"Origin": "https://udbreg.huya.com",
"Pragma": "no-cache",
"Referer": self.middle_url,
"Sec-Fetch-Dest": "empty",
"Sec-Fetch-Mode": "cors",
"Sec-Fetch-Site": "same-origin",
"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:
"""初始化改密 middle 页面。"""
for host in ("udblgn.huya.com", "udbreg.huya.com"):
url = f"https://{host}/web/middle/{CHANGE_PASSWORD_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"虎牙改密 middle 初始化失败: {host}: {exc}")
self.session.headers.update({
"Origin": "https://udbreg.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) -> None:
"""处理虎牙风控验证。"""
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,
)
solver.solve(payload)
def _common_payload(self, uri: str, data: dict) -> dict:
return {
"uri": uri,
"version": CHANGE_PASSWORD_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://udbreg.huya.com",
"Referer": self.middle_url,
"context": self.context,
"lcid": LCID,
"reqid": self.request_id,
"uri": uri,
}
def _check_once(self) -> dict:
data = {
"uid": self.uid,
"behavior": encode_change_password_behavior("check"),
"page": CHANGE_PASSWORD_PAGE,
}
return self._request_json(
"post",
CHANGE_PASSWORD_CHECK_URL,
"虎牙改密前置校验",
json=self._common_payload(CHANGE_PASSWORD_CHECK_URI, data),
headers=self._headers_for_uri(CHANGE_PASSWORD_CHECK_URI),
)
def _send_sms_once(self, session_data: str) -> dict:
data = {
"phone": "",
"sessionData": session_data,
"behavior": encode_change_password_behavior("send"),
"page": CHANGE_PASSWORD_PAGE,
}
return self._request_json(
"post",
CHANGE_PASSWORD_SEND_SMS_URL,
"虎牙改密短信发码",
json=self._common_payload(CHANGE_PASSWORD_SEND_SMS_URI, data),
headers=self._headers_for_uri(CHANGE_PASSWORD_SEND_SMS_URI),
)
def _submit_once(self, password: str, sms_code: str, session_data: str) -> dict:
data = {
"password": password,
"smsCode": sms_code,
"sessionData": session_data,
"behavior": encode_change_password_behavior("submit"),
"page": CHANGE_PASSWORD_PAGE,
}
return self._request_json(
"post",
CHANGE_PASSWORD_SUBMIT_URL,
"虎牙短信改密",
json=self._common_payload(CHANGE_PASSWORD_SUBMIT_URI, data),
headers=self._headers_for_uri(CHANGE_PASSWORD_SUBMIT_URI),
)
def send_code(self) -> HuyaChangePasswordCodeResult:
"""发送改密短信，返回提交改密需要的 sessionData。"""
self.prepare_device()
self.init_udb_middle()
payload = self._check_once()
for index in range(3):
return_code = int(payload.get("returnCode") or 0)
if return_code in (10030, 10039):
logger.info(f"虎牙改密前置校验触发风控: {return_code} ({index + 1}/3)")
self._solve_verification(payload)
time.sleep(0.5)
payload = self._check_once()
continue
if return_code != 0:
return HuyaChangePasswordCodeResult(
success=False,
message=f"虎牙改密前置校验失败: {payload}",
raw=payload,
sdid=self.sdid,
context=self.context,
request_id=self.request_id,
)
break
else:
return HuyaChangePasswordCodeResult(
success=False,
message=f"虎牙改密前置校验风控后仍失败: {payload}",
raw=payload,
sdid=self.sdid,
context=self.context,
request_id=self.request_id,
)
session_data = _session_data(payload)
if not session_data:
return HuyaChangePasswordCodeResult(
success=False,
message=f"虎牙改密前置校验未返回 sessionData: {payload}",
raw=payload,
sdid=self.sdid,
context=self.context,
request_id=self.request_id,
)
payload = self._send_sms_once(session_data)
for index in range(3):
return_code = int(payload.get("returnCode") or 0)
session_data = _session_data(payload, fallback=session_data)
if return_code == 0 and session_data:
return HuyaChangePasswordCodeResult(
success=True,
session_data=session_data,
message="改密短信已发送",
raw=payload,
sdid=self.sdid,
context=self.context,
request_id=self.request_id,
)
if return_code not in (10030, 10039):
return HuyaChangePasswordCodeResult(
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._send_sms_once(session_data)
return HuyaChangePasswordCodeResult(
success=False,
message=f"虎牙改密短信发码风控后仍失败: {payload}",
raw=payload,
sdid=self.sdid,
context=self.context,
request_id=self.request_id,
)
def submit_code(self, password: str, sms_code: str, session_data: str) -> HuyaChangePasswordResult:
"""提交改密短信验证码。"""
if not session_data:
raise HuyaLoginError("缺少改密 sessionData，请先发送改密短信")
payload = self._submit_once(password=password, sms_code=sms_code, session_data=session_data)
for index in range(3):
return_code = int(payload.get("returnCode") or 0)
if return_code == 0:
return HuyaChangePasswordResult(
success=True,
message="密码修改成功",
code=sms_code,
raw=payload,
sdid=self.sdid,
context=self.context,
request_id=self.request_id,
attempts=index + 1,
)
if return_code not in (10030, 10039):
return HuyaChangePasswordResult(
success=False,
message=f"虎牙短信改密失败: {payload}",
code=sms_code,
raw=payload,
sdid=self.sdid,
context=self.context,
request_id=self.request_id,
attempts=index + 1,
)
logger.info(f"虎牙短信改密触发风控: {return_code} ({index + 1}/3)")
self._solve_verification(payload)
time.sleep(0.5)
payload = self._submit_once(password=password, sms_code=sms_code, session_data=session_data)
return HuyaChangePasswordResult(
success=False,
message=f"虎牙短信改密风控后仍失败: {payload}",
code=sms_code,
raw=payload,
sdid=self.sdid,
context=self.context,
request_id=self.request_id,
attempts=3,
)
def change_huya_password_with_sms_line(
uid: str,
cookie: Mapping[str, str] | str,
password: str,
item: SmsLine,
wait_seconds: float = 180,
poll_interval: float = 5,
ignore_codes: set[str] | None = None,
stop_event: threading.Event | None = None,
) -> HuyaChangePasswordResult:
"""使用同一手机号接码链接完成改密短信验证。"""
changer = HuyaPasswordChanger(uid=uid, cookie=cookie)
sent_at = datetime.now()
code_result = changer.send_code()
if not code_result.success or not code_result.session_data:
return HuyaChangePasswordResult(
success=False,
message=code_result.message or "虎牙改密短信发码失败",
raw=code_result.raw,
sdid=code_result.sdid,
context=code_result.context,
request_id=code_result.request_id,
)
client = SmsProviderClient()
deadline = time.monotonic() + max(15.0, float(wait_seconds or 180))
interval = max(1.0, float(poll_interval or 5))
attempts = 0
ignored = {str(code) for code in (ignore_codes or set()) if code}
last_message = "等待改密验证码"
while time.monotonic() < deadline:
if stop_event and stop_event.is_set():
return HuyaChangePasswordResult(
success=False,
message="已停止",
sdid=code_result.sdid,
context=code_result.context,
request_id=code_result.request_id,
attempts=attempts,
)
attempts += 1
poll_result = client.poll(item, after_time=sent_at)
last_message = poll_result.message or last_message
if poll_result.has_code:
if poll_result.code in ignored:
last_message = "忽略注册阶段旧验证码"
else:
result = changer.submit_code(
password=password,
sms_code=poll_result.code,
session_data=code_result.session_data,
)
result.attempts = attempts
return result
remaining = deadline - time.monotonic()
if remaining <= 0:
break
if not _sleep_or_stop(stop_event, min(interval, remaining)):
return HuyaChangePasswordResult(
success=False,
message="已停止",
sdid=code_result.sdid,
context=code_result.context,
request_id=code_result.request_id,
attempts=attempts,
)
return HuyaChangePasswordResult(
success=False,
message=f"等待改密验证码超时: {last_message}",
sdid=code_result.sdid,
context=code_result.context,
request_id=code_result.request_id,
attempts=attempts,
)