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

fix(tooling): 修复 pyright 解释器配置与类型守卫，补齐 ruff 门禁与两处历史 bug

Browse Source 
- pyrightconfig.json 钉死 Python 3.12 虚拟环境，消除由于查找系统 Python 3.9 带来的假性报错
- 修复 huya/login.py 引用不存在函数的问题并补充单元测试
- 修复 browser_harvest.py 对 set 下标访问的错误并提升报错信息质量
- 虎牙 6 个子路由显式导入依赖，消除 import * 带来的未定义变量报警
- 规范化 Pydantic Field(default=...) 默认值声明
- 完善 solver.py / app_login.py 等核心路径类型守卫与 @overload 重载
- 在 pyproject.toml 中显式锁定 ruff lint 规则集 (E4, E7, E9, F, B, I, BLE)
- 门禁脚本同步增加 ruff check 检查并在 AGENTS.md 补齐实践经验总结
This commit is contained in:
yml2213
2026-10-08 12:33:39 +08:00
parent ec07e4fc06
commit a3c3a46b9c
31 changed files with 599 additions and 96 deletions
Download Patch File Download Diff File Expand all files Collapse all files
core/huya/app_login.py
+47 -7
View File
@@ -31,6 +31,7 @@ import struct
import time
import uuid
from collections.abc import Mapping
from typing import Literal, overload
from urllib.parse import parse_qs, quote, urlparse
import requests
@@ -245,7 +246,7 @@ def solve_safe_auth(
k: v[0]
for k, v in parse_qs(urlparse(risk_url).query, keep_blank_values=True).items()
}
app_id = str(q.get("appId") or "5002")
app_id = q.get("appId") or "5002"
last_err: Exception | None = None
for attempt in range(max_retry):
solver = HuyaVerificationSolver(
@@ -293,6 +294,42 @@ def parse_real_uid(resp: bytes) -> int:
raise HuyaAppLoginError("无法从登录响应解析真实 UID")
@overload
def login_cred_with_flow(
account: str,
password: str,
max_rounds: int = ...,
device_info: dict | None = ...,
proxies: dict | None = ...,
include_device_token: Literal[False] = ...,
sms_url: str = ...,
) -> tuple[bytes, int]: ...
@overload
def login_cred_with_flow(
account: str,
password: str,
max_rounds: int = ...,
device_info: dict | None = ...,
proxies: dict | None = ...,
include_device_token: Literal[True] = ...,
sms_url: str = ...,
) -> tuple[bytes, int, str]: ...
@overload
def login_cred_with_flow(
account: str,
password: str,
max_rounds: int = ...,
device_info: dict | None = ...,
proxies: dict | None = ...,
include_device_token: bool = ...,
sms_url: str = ...,
) -> tuple[bytes, int] | tuple[bytes, int, str]: ...
def login_cred_with_flow(
account: str,
password: str,
@@ -405,7 +442,11 @@ class QrRole:
self.req_counter = random.randint(40_000_000, 41_000_000)
self.s.headers.update(
{
"User-Agent": UA_PC if pc else mobile_user_agent(device_info),
"User-Agent": (
UA_PC
if pc or device_info is None
else mobile_user_agent(device_info)
),
"Origin": UDB_BASE,
"content-type": "application/json;charset=UTF-8",
"Accept": "*/*",
@@ -550,8 +591,7 @@ class HuyaAppPasswordLogin:
jar = session.cookies
if hasattr(jar, "items"):
pairs = [
(str(key), "" if value is None else str(value))
for key, value in jar.items()
(key, "" if value is None else value) for key, value in jar.items()
]
else:
pairs = [
@@ -576,13 +616,13 @@ class HuyaAppPasswordLogin:
raw_cred = (
base64.urlsafe_b64encode(cred).decode("ascii").rstrip("=")
if isinstance(cred, bytes)
else str(cred or "")
else (cred or "")
)
# 本次登录/指纹响应明确返回的字段，必须覆盖旧 Session 值。
set_value("udb_cred", raw_cred, overwrite=True)
if uid:
set_value("udb_uid", int(uid), overwrite=True)
set_value("yyuid", int(uid), overwrite=True)
set_value("udb_uid", uid, overwrite=True)
set_value("yyuid", uid, overwrite=True)
set_value("udb_biztoken", biztoken, overwrite=True)
set_value("udb_version", "1.0", overwrite=True)
set_value("udb_origin", "0", overwrite=True)
core/huya/browser_harvest.py
+23 -6
View File
@@ -136,11 +136,18 @@ def launch(
proc = subprocess.Popen(
args, stdout=subprocess.DEVNULL, stderr=subprocess.PIPE, text=True
)
# 上面用的是 stderr=PIPE，所以 proc.stderr 一定不是 None（静态类型是
# IO[str] | None）。收一次窄，顺便在真拿不到 stderr 时给出明确报错。
stderr = proc.stderr
if stderr is None:
proc.kill()
raise RuntimeError("Chrome stderr 未接通，无法读取调试端口")
ws_url = None
deadline = time.time() + 20
last_probe_error: Exception | None = None
while time.time() < deadline:
# 1) 从 DevTools 输出的 ws url（browser 级）
line = proc.stderr.readline()
line = stderr.readline()
if "DevTools listening on" in line:
browser_ws = line.strip().split(" ")[-1]
else:
@@ -151,23 +158,33 @@ def launch(
f"http://127.0.0.1:{port}/json/list", timeout=2
) as resp:
targets = json.load(resp)
pages = {
# 必须是 list 而不是 set：下面要按下标取「第一个」page target，
# set 不可下标。这里曾写成集合推导式，pages[0] 抛出的 TypeError
# 被下面的宽 except 吞掉，于是每次都静默退回 browser 级 endpoint，
# 页面级探测长期失效却无人察觉。回归用例：
# tests/test_huya_browser_harvest.py::test_launch_prefers_page_target_over_browser_endpoint
pages = [
t
for t in targets
if t.get("type") == "page" and t.get("webSocketDebuggerUrl")
}
]
if pages:
ws_url = pages[0]["webSocketDebuggerUrl"]
break
except Exception: # noqa: BLE001, S110 - 端口未就绪时轮询重试
pass
except Exception as exc: # noqa: BLE001 - 端口未就绪时轮询重试
# 不静默：记下最后一次探测异常，端口最终没起来时一并报出来，
# 避免「代码 bug」被读成「端口没起来」。
last_probe_error = exc
# 3) 兜底：browser 级 endpoint
if browser_ws:
ws_url = browser_ws
break
if not ws_url:
proc.kill()
raise RuntimeError("Chrome 调试端口未就绪")
detail = (
f"；最后一次 CDP 探测异常: {last_probe_error!r}" if last_probe_error else ""
)
raise RuntimeError(f"Chrome 调试端口未就绪{detail}")
return proc, ws_url
core/huya/elite_session.py
+7 -1
View File
@@ -35,7 +35,13 @@ class HuyaEliteWssSession:
self.logger = logger or (lambda _message: None)
self.http = HuyaHttpClient(logger=self.logger)
self.loop = asyncio.new_event_loop()
self.client: HuyaWssClient | None = None
# 会话内**始终**持有 client：下面（本函数的 try 之前）无条件创建，
# close() 只置 self._closed，从不把它改回 None。原先标成 `| None = None`
# 是陈旧声明，代价是 20 处 `self.client.xxx(...)` 全被报「可能为 None」。
# 顺带说明：_run() 里 `self.client is None` 这条兜底因此是死代码 ——
# 而且它本来就被求值顺序绕过（`self.client.xxx()` 作为实参先于 _run 执行），
# 保留它只为防御未来有人真的把 client 置空。
self.client: HuyaWssClient
self._closed = False
if kind == "activity":
core/huya/verification/solver.py
+23 -13
View File
@@ -65,7 +65,7 @@ def _cookie_mapping(cookie: Mapping[str, str] | str | None) -> dict[str, 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()}
return {key: value for key, value in cookie.items()}
class HuyaVerificationSolver:
@@ -214,7 +214,7 @@ class HuyaVerificationSolver:
background = cv2.cvtColor(cv2.Canny(background, 100, 200), cv2.COLOR_GRAY2RGB)
result = cv2.matchTemplate(background, target, cv2.TM_CCOEFF_NORMED)
_, _, _, max_loc = cv2.minMaxLoc(result)
return int(max_loc[0])
return max_loc[0]
@staticmethod
def _opencv_rotate_angle(
@@ -331,7 +331,10 @@ class HuyaVerificationSolver:
if Path(font_path).is_file():
try:
return ImageFont.truetype(font_path, size)
except Exception:
except OSError:
# 字体文件在、但 FreeType 打不开或格式不对（实测「cannot open
# resource」和「unknown file format」都是 OSError）—— 换下一个候选。
# 收窄异常类型：宽 except 会把参数写错之类的真 bug 一起吞掉。
continue
return ImageFont.load_default()
@@ -627,7 +630,7 @@ class HuyaVerificationSolver:
restored_bg, tip_bytes, int(data.get("location_y") or 0)
)
slider_box = self.ocr.ocr_slider_box(composite)
yolo_x = int(slider_box[0] - 13)
yolo_x = 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)
@@ -666,7 +669,7 @@ class HuyaVerificationSolver:
if tip_bytes:
merged = self._vertical_concat_align_left(bg_bytes, tip_bytes)
points = self.ocr.ocr_click_points(merged)
final_points = [[int(x), int(y) - 44] for x, y in points]
final_points = [[x, y - 44] for x, y in points]
else:
question = str(data.get("question") or "")
target_chars = self._extract_target_chars_from_question(question)
@@ -675,7 +678,7 @@ class HuyaVerificationSolver:
synth_tip = self._synthesize_text_tip(target_chars)
merged = self._vertical_concat_align_left(bg_bytes, synth_tip)
points = self.ocr.ocr_click_points(merged)
final_points = [[int(x), int(y) - 44] for x, y in points]
final_points = [[x, y - 44] for x, y in points]
tip_bytes = synth_tip
else:
logger.info(f"虎牙点选题目未能提取明确目标字: {question}")
@@ -699,11 +702,11 @@ class HuyaVerificationSolver:
drag_distance = max(5, int((angle / 360.0) * 260))
track_points = generate_slide_track(drag_distance)
act_data = {
"point": int(round(angle)),
"point": round(angle),
"travel": format_track(track_points),
"code": self.code,
"sceneType": self.scene_type,
"endTime": int(track_points[-1][-1]) - int(track_points[0][-1]),
"endTime": track_points[-1][-1] - track_points[0][-1],
}
act = self.js_ctx.call("get_act", act_data, self.huyapk)
self.behavior_events.append(
@@ -726,11 +729,11 @@ class HuyaVerificationSolver:
def _send_slider_verify(self, url: str, point: int) -> dict:
track_points = generate_slide_track(point)
act_data = {
"point": int(point),
"point": point,
"travel": format_track(track_points),
"code": self.code,
"sceneType": self.scene_type,
"endTime": int(track_points[-1][-1]) - int(track_points[0][-1]),
"endTime": track_points[-1][-1] - track_points[0][-1],
}
act = self.js_ctx.call("get_act", act_data, self.huyapk)
self.behavior_events.append(
@@ -759,7 +762,7 @@ class HuyaVerificationSolver:
current_time = start_time
for point in points:
current_time += random.randint(1000, 2000)
click_data.append([int(point[0]), int(point[1]), current_time])
click_data.append([point[0], point[1], current_time])
interval_time_list = [item[-1] for item in click_data]
act_data = {
@@ -958,11 +961,18 @@ class HuyaVerificationSolver:
or "/auth/xz/" in vurl
or "rotate" in self.strategy_url
):
last_result = self._send_rotate_verify(vurl, float(point))
if isinstance(point, (int, float)):
last_result = self._send_rotate_verify(vurl, float(point))
else:
logger.warning(f"旋转验证识别返回非数值结果: {point}")
continue
elif isinstance(point, list):
last_result = self._send_click_verify(vurl, point)
elif point is not None:
last_result = self._send_slider_verify(vurl, point)
else:
last_result = self._send_slider_verify(vurl, int(point))
logger.warning("验证码识别未返回有效结果")
continue
if isinstance(last_result, dict) and (
last_result.get("authId") or last_result.get("nextAuthUrl")
core/huya/wup_protocol.py
+1 -1
View File
@@ -56,7 +56,7 @@ class WupRequest:
os.write_struct(0, struct_data)
elif isinstance(struct_data, dict):
os.write_struct_begin(0)
for tag_id, (field_name, field_value) in enumerate(struct_data.items()):
for tag_id, (_, field_value) in enumerate(struct_data.items()):
self._write_field(os, tag_id, field_value)
os.write_struct_end()
elif isinstance(struct_data, (list, tuple)):