Compare commits

..
3 Commits
Author SHA1 Message Date
yml2213 09ab80e062 完成 Ruff 全量清理 2026-08-31 10:55:44 +08:00
yml2213 840af3108e 清理未使用导入和局部变量 2026-08-31 10:31:32 +08:00
yml2213 b58c6b4357 执行 Ruff 安全自动修复 2026-08-31 10:28:14 +08:00
154 changed files with 1255 additions and 1418 deletions
+8 -13
View File
@@ -2,21 +2,16 @@
## Ruff lint 清理
记录日期:2026-08-30
状态:已完成
`uv run ruff check .` 当前发现 1263 条 lint 诊断。该事项暂不在本次处理,后续按批次清理并逐批运行 pytest、Pyright 和 Ruff 检查。
完成日期:2026-08-31
建议顺序:
本轮完成了 Ruff lint 全量清理,包括未使用导入与变量、导入排序、类型注解现代化、FastAPI 依赖声明、异常处理标注、时区处理及其他规则。FastAPI 路由参数中的依赖调用通过 `web/backend/**/*.py` 的 B008 配置例外保留框架惯用写法;非 FastAPI 的可变默认对象已改为函数内惰性初始化。
1. `F401``F841`:未使用导入和变量。
2. `BLE001``S110``S112`:异常处理质量。
3. `B008`:FastAPI 依赖声明模式,区分真实问题与框架惯用写法。
4. `I001`:导入排序。
5. `UP045``UP007``UP017``UP035``UP006`:类型注解和 Python 版本语法现代化。
最终基线:
当前基线:
- Ruff lint1263 条
- Ruff format207 个文件全部通过
- pytest104 passed
- Ruff lint0 条
- Ruff format208 个文件全部通过
- pytest105 passed
- Pyright0 errors / 0 warnings
- Python compileall:通过
+10 -10
View File
@@ -1,11 +1,11 @@
"""斗鱼登录模块"""
from .login import DouyuLogin, CredentialError
from .login_api import LoginAPIStrategy
from .login_api_wgapi import WgapiLoginAPI
from .login_api_iframe import IframeLoginAPI
from .email_verifier import EmailVerifier
from .activity_client import DouyuActivityClient, DouyuActivityError
from .email_verifier import EmailVerifier
from .login import CredentialError, DouyuLogin
from .login_api import LoginAPIStrategy
from .login_api_iframe import IframeLoginAPI
from .login_api_wgapi import WgapiLoginAPI
from .recharge_api import (
FishFinRechargeClient,
FishFinRechargeConfig,
@@ -14,16 +14,16 @@ from .recharge_api import (
)
__all__ = [
"DouyuLogin",
"CredentialError",
"LoginAPIStrategy",
"WgapiLoginAPI",
"IframeLoginAPI",
"EmailVerifier",
"DouyuActivityClient",
"DouyuActivityError",
"DouyuLogin",
"EmailVerifier",
"FishFinRechargeClient",
"FishFinRechargeConfig",
"FishFinRechargeConfigError",
"FishFinRechargeError",
"IframeLoginAPI",
"LoginAPIStrategy",
"WgapiLoginAPI",
]
+1 -3
View File
@@ -18,7 +18,6 @@ from .cookie_utils import (
normalize_douyu_cookie,
)
PC_UA = (
"Mozilla/5.0 (Windows NT 10.0; Win64; x64) "
"AppleWebKit/537.36 (KHTML, like Gecko) "
@@ -1446,8 +1445,7 @@ class DouyuActivityClient:
if not match:
raise DouyuActivityError(f"响应中找不到 var {varname}=: {text[:200]}")
chunk = text[match.end() :].strip()
if chunk.endswith(";"):
chunk = chunk[:-1]
chunk = chunk.removesuffix(";")
try:
return json.loads(chunk)
except json.JSONDecodeError as exc:
+1 -1
View File
@@ -44,7 +44,7 @@ class CookieEnricher:
self.generate_acf_ccn_cookie()
logger.info("补CK完成")
return
except Exception as e:
except Exception as e: # noqa: BLE001 外部接口与任务边界需要保留宽泛异常兜底
last_error = e
logger.warning(f"补CK失败 {attempt}/{max_attempts}: {e}")
if attempt < max_attempts:
+2 -4
View File
@@ -2,11 +2,9 @@
import base64
import hashlib
from Crypto.PublicKey import RSA
from Crypto.Cipher import PKCS1_v1_5
from Crypto.Cipher import AES
from Crypto.Cipher import ARC4
from Crypto.Cipher import AES, ARC4, PKCS1_v1_5
from Crypto.PublicKey import RSA
# 斗鱼RSA公钥(从JS中提取)
DOUYU_RSA_PUBLIC_KEY = """-----BEGIN PUBLIC KEY-----
+28 -27
View File
@@ -5,12 +5,10 @@ import os
import re
import threading
import time
from datetime import datetime, timedelta
from typing import Optional
from datetime import UTC, datetime, timedelta
from loguru import logger
import requests
from loguru import logger
# Roundcube Webmail 地址。可通过环境变量覆盖。空值时回退到默认。
ROUNDCUBE_URL = os.getenv("MAIL_ROUNDCUBE_URL") or "http://111.229.206.54:8000/"
@@ -40,7 +38,7 @@ class EmailVerifier:
max_messages: int = 20,
use_ssl: bool = False,
roundcube_url: str = "",
backup_passwords: Optional[list[str]] = None,
backup_passwords: list[str] | None = None,
):
# 兼容旧参数名
self.imap_server = imap_server
@@ -54,14 +52,13 @@ class EmailVerifier:
)
# Roundcube 会话(懒初始化)
self._rc_session: Optional[requests.Session] = None
self._rc_token: Optional[str] = None
self._rc_session: requests.Session | None = None
self._rc_token: str | None = None
self._rc_logged_in = False
self._rc_login_error = ""
def connect(self) -> None:
"""兼容接口"""
pass
def disconnect(self) -> None:
"""关闭 Roundcube 会话"""
@@ -71,8 +68,8 @@ class EmailVerifier:
f"{self.roundcube_url}?_task=logout",
timeout=5,
)
except Exception:
pass
except Exception as exc: # noqa: BLE001
logger.debug(f"Roundcube logout failed: {exc}")
self._rc_session = None
self._rc_logged_in = False
@@ -179,7 +176,7 @@ class EmailVerifier:
logger.debug(f"Roundcube: {password_name}登录成功 ({self.username})")
return True, False
except Exception as e:
except Exception as e: # noqa: BLE001 外部接口与任务边界需要保留宽泛异常兜底
logger.warning(f"Roundcube: 登录异常: {e}")
return False, False
@@ -233,13 +230,13 @@ class EmailVerifier:
return messages
except Exception as e:
except Exception as e: # noqa: BLE001 外部接口与任务边界需要保留宽泛异常兜底
logger.warning(f"Roundcube: 获取邮件列表失败: {e}")
return []
# ── Roundcube 读取邮件内容 ─────────────────────────────
def _rc_fetch_email_body(self, uid: int) -> Optional[str]:
def _rc_fetch_email_body(self, uid: int) -> str | None:
"""通过 Roundcube 读取指定 UID 邮件的正文"""
if not self._ensure_roundcube_session():
if self._rc_login_error:
@@ -255,14 +252,14 @@ class EmailVerifier:
)
return resp.text
except Exception as e:
except Exception as e: # noqa: BLE001 外部接口与任务边界需要保留宽泛异常兜底
logger.warning(f"Roundcube: 读取邮件 {uid} 失败: {e}")
return None
# ── Roundcube 日期解析 ─────────────────────────────────
@staticmethod
def _parse_rc_date(date_str: str) -> Optional[datetime]:
def _parse_rc_date(date_str: str) -> datetime | None:
"""
解析 Roundcube 返回的日期字符串为 datetime。
@@ -273,7 +270,7 @@ class EmailVerifier:
- "2026-06-20" → 直接解析
- "06-20" → 今年的该日期
"""
now = datetime.now()
now = datetime.now(UTC)
# 今天
if date_str.startswith("今天"):
@@ -325,32 +322,36 @@ class EmailVerifier:
# 日期格式 2026-06-20
try:
return datetime.strptime(date_str.strip(), "%Y-%m-%d")
return datetime.strptime(date_str.strip(), "%Y-%m-%d").replace(tzinfo=UTC)
except ValueError:
pass
# 日期格式 2026-06-20 14:30
try:
return datetime.strptime(date_str.strip(), "%Y-%m-%d %H:%M")
return datetime.strptime(date_str.strip(), "%Y-%m-%d %H:%M").replace(
tzinfo=UTC
)
except ValueError:
pass
# 日期格式 2026-06-20 14:30:00
try:
return datetime.strptime(date_str.strip(), "%Y-%m-%d %H:%M:%S")
return datetime.strptime(date_str.strip(), "%Y-%m-%d %H:%M:%S").replace(
tzinfo=UTC
)
except ValueError:
pass
# 日期格式 06-20
try:
dt = datetime.strptime(date_str.strip(), "%m-%d")
dt = datetime.strptime(date_str.strip(), "%m-%d").replace(tzinfo=UTC)
return dt.replace(year=now.year)
except ValueError:
pass
# 日期格式 06-20 14:30
try:
dt = datetime.strptime(date_str.strip(), "%m-%d %H:%M")
dt = datetime.strptime(date_str.strip(), "%m-%d %H:%M").replace(tzinfo=UTC)
return dt.replace(year=now.year)
except ValueError:
pass
@@ -363,9 +364,9 @@ class EmailVerifier:
self,
max_wait: int = 60,
interval: int = 3,
after_timestamp: Optional[float] = None,
after_timestamp: float | None = None,
allow_old_seconds: int = 15,
stop_event: Optional[threading.Event] = None,
stop_event: threading.Event | None = None,
) -> str:
"""
轮询获取斗鱼验证码。
@@ -398,7 +399,7 @@ class EmailVerifier:
return code
except EmailLoginError:
raise
except Exception as e:
except Exception as e: # noqa: BLE001 外部接口与任务边界需要保留宽泛异常兜底
last_error = str(e)
logger.warning(f"Roundcube 读邮件异常: {e}")
@@ -417,9 +418,9 @@ class EmailVerifier:
def _fetch_code_via_roundcube(
self,
after_timestamp: Optional[float] = None,
after_timestamp: float | None = None,
allow_old_seconds: int = 15,
) -> Optional[str]:
) -> str | None:
"""通过 Roundcube API 获取最新斗鱼验证码"""
messages = self._rc_fetch_mail_list()
@@ -465,7 +466,7 @@ class EmailVerifier:
# ── 通用工具方法 ───────────────────────────────────────
def _extract_verification_code(self, text: str) -> Optional[str]:
def _extract_verification_code(self, text: str) -> str | None:
"""从文本中提取6位验证码"""
# 清理HTML标签和实体
text = html.unescape(text)
+30 -28
View File
@@ -4,28 +4,30 @@ import json
import re
import threading
import time
import requests
from typing import Mapping, Optional, Protocol, Tuple
from collections.abc import Mapping
from typing import Protocol
from urllib.parse import urlsplit, urlunsplit
import requests
from loguru import logger
from core.geetest.common.network import (
get_c_s,
get_js_address,
req_fullpage_validate,
)
from core.geetest.v3_slide.solver import (
_generate_seed,
get_w1,
get_w2,
)
from .cookie_enricher import CookieEnricher
from .email_verifier import EmailLoginError, EmailVerifier
from .login_api import LoginAPIStrategy
from .login_api_wgapi import WgapiLoginAPI
from .proxy_fetcher import ProxyFetcher
from core.geetest.v3_slide.solver import (
_generate_seed,
get_w1,
get_w2,
)
from core.geetest.common.network import (
get_js_address,
get_c_s,
req_fullpage_validate,
)
# ── 全局极验并发限制:同一时刻最多2个线程做极验验证 ──
_geetest_semaphore = threading.Semaphore(2)
@@ -95,14 +97,14 @@ class DouyuLogin:
def __init__(
self,
account: AccountLike,
proxy: Optional[str | Mapping[str, str]] = None,
proxy_api_url: Optional[str] = None,
proxy: str | Mapping[str, str] | None = None,
proxy_api_url: str | None = None,
timeout: tuple[float, float] = REQUEST_TIMEOUT,
max_login_retries: int = 0,
max_total_time: float = 0,
proxy_fetcher: Optional[ProxyFetcher] = None,
stop_event: Optional[threading.Event] = None,
api_strategy: Optional[LoginAPIStrategy] = None,
proxy_fetcher: ProxyFetcher | None = None,
stop_event: threading.Event | None = None,
api_strategy: LoginAPIStrategy | None = None,
):
self.account = account
self.proxy = proxy
@@ -117,7 +119,7 @@ class DouyuLogin:
ProxyFetcher(api_url=proxy_api_url) if proxy_api_url else None
)
self._current_proxy_url: Optional[str] = None
self._current_proxy_url: str | None = None
self._cookie_enrich_error = ""
self._static_retry_count = 0
self._setup_session()
@@ -396,7 +398,7 @@ class DouyuLogin:
return LoginResult(
success=False, message=str(e), code="email_login_failed"
)
except Exception as e:
except Exception as e: # noqa: BLE001 外部接口与任务边界需要保留宽泛异常兜底
elapsed = time.monotonic() - start_time
if self.max_login_retries > 0:
logger.error(
@@ -473,7 +475,7 @@ class DouyuLogin:
return LoginResult(
success=False, message=str(e), code="email_login_failed"
)
except Exception as e:
except Exception as e: # noqa: BLE001 外部接口与任务边界需要保留宽泛异常兜底
elapsed = time.monotonic() - start_time
if self.max_login_retries > 0:
logger.error(
@@ -565,7 +567,7 @@ class DouyuLogin:
)
return True
def _first_login(self) -> Tuple[str, str, str, dict]:
def _first_login(self) -> tuple[str, str, str, dict]:
"""
第一次登录,获取极验参数
@@ -608,7 +610,7 @@ class DouyuLogin:
def _solve_geetest(
self, gt: str, challenge: str, deadline: float = 0
) -> Tuple[str, str]:
) -> tuple[str, str]:
"""
解决极验 fullpage 验证(最多3次尝试,失败直接抛异常回到login换新代理)
@@ -635,7 +637,7 @@ class DouyuLogin:
def _solve_geetest_inner(
self, gt: str, challenge: str, deadline: float = 0
) -> Tuple[str, str]:
) -> tuple[str, str]:
"""极验验证内部实现:最多3次尝试,失败直接抛异常。"""
_MAX_ATTEMPTS = 3
@@ -652,7 +654,7 @@ class DouyuLogin:
# 超时兜底
if deadline and time.monotonic() > deadline:
raise ValueError(f"极验验证超时(登录整体时间耗尽)")
raise ValueError("极验验证超时(登录整体时间耗尽)")
try:
logger.info(f"极验验证尝试 {attempt}/{_MAX_ATTEMPTS}")
@@ -825,7 +827,7 @@ class DouyuLogin:
logger.info("验证邮件已发送")
def _get_email_code(self, after_timestamp: Optional[float] = None) -> str:
def _get_email_code(self, after_timestamp: float | None = None) -> str:
"""获取邮箱验证码"""
verifier = EmailVerifier(
imap_server=self.account.email_imap_server,
@@ -899,7 +901,7 @@ class DouyuLogin:
if response2.status_code == 200:
logger.info("WebLogin成功")
except Exception as e:
except Exception as e: # noqa: BLE001 外部接口与任务边界需要保留宽泛异常兜底
logger.warning(f"WebLogin请求失败(不影响登录): {e}")
# 登录成功后补齐 Web 侧 Cookie。补 CK 失败只影响完整度,不回滚已成功的登录态。
@@ -913,7 +915,7 @@ class DouyuLogin:
).enrich_with_retry()
except InterruptedError:
raise
except Exception as e:
except Exception as e: # noqa: BLE001 外部接口与任务边界需要保留宽泛异常兜底
self._cookie_enrich_error = self._truncate_error(str(e), 160)
logger.warning(
f"补CK最终失败,本次仍按登录成功保存基础CK: {self._cookie_enrich_error}"
+1 -2
View File
@@ -1,7 +1,6 @@
"""登录接口策略基类"""
from abc import ABC, abstractmethod
from typing import Tuple
class LoginAPIStrategy(ABC):
@@ -78,7 +77,7 @@ class LoginAPIStrategy(ABC):
"""构建提交验证码参数"""
...
def extract_geetest_params(self, payload: dict) -> Tuple[str, str, str]:
def extract_geetest_params(self, payload: dict) -> tuple[str, str, str]:
"""从第一次登录响应提取极验参数
Args:
+1
View File
@@ -10,6 +10,7 @@
import threading
import time
from loguru import logger
from .proxy_resolver import ProxyResolver
+1 -2
View File
@@ -2,10 +2,9 @@
import json
import re
from typing import Optional
def parse_proxy_response(text: str) -> tuple[list[str], Optional[str]]:
def parse_proxy_response(text: str) -> tuple[list[str], str | None]:
"""
解析代理 API 响应,支持 JSON 格式与旧版纯文本格式。
+4 -4
View File
@@ -6,14 +6,14 @@
3. 在 _PLATFORM_CREDENTIAL_FIELDS 中定义凭据字段
"""
from typing import Any, Optional, Type, cast
from typing import Any, cast
from .base import BaseWhitelistAdapter, get_exit_ip_via_proxy
from .base import BaseWhitelistAdapter
from .xiequ import XiequAdapter
from .xkdaili import XkdailiAdapter
# ── 平台注册表 ──
_PLATFORM_REGISTRY: dict[str, Type[BaseWhitelistAdapter]] = {
_PLATFORM_REGISTRY: dict[str, type[BaseWhitelistAdapter]] = {
"xiequ": XiequAdapter,
"xkdaili": XkdailiAdapter,
}
@@ -49,7 +49,7 @@ _PLATFORM_LABELS: dict[str, str] = {
}
def create_adapter(platform: str, credentials: dict) -> Optional[BaseWhitelistAdapter]:
def create_adapter(platform: str, credentials: dict) -> BaseWhitelistAdapter | None:
"""根据平台标识符和凭据创建适配器实例。"""
cls = _PLATFORM_REGISTRY.get(platform)
if not cls:
+8 -7
View File
@@ -8,7 +8,6 @@ import re
import threading
import time
from abc import ABC, abstractmethod
from typing import Optional
import requests
from loguru import logger
@@ -138,18 +137,18 @@ class BaseWhitelistAdapter(ABC):
return False, f"白名单添加失败,API响应: {resp}"
except Exception as e:
except Exception as e: # noqa: BLE001 外部接口与任务边界需要保留宽泛异常兜底
msg = f"白名单同步失败: {e}"
logger.error(msg)
return False, msg
@staticmethod
def get_local_exit_ip() -> Optional[str]:
def get_local_exit_ip() -> str | None:
"""获取本机当前公网出口IP(不走代理)。"""
return _get_local_exit_ip()
def _get_local_exit_ip() -> Optional[str]:
def _get_local_exit_ip() -> str | None:
"""获取本机当前公网出口IP(不走代理)。"""
targets = [
"https://myip.ipip.net",
@@ -166,12 +165,13 @@ def _get_local_exit_ip() -> Optional[str]:
match = re.search(r"(\d{1,3}(?:\.\d{1,3}){3})", text)
if match:
return match.group(1)
except Exception:
except Exception as exc: # noqa: BLE001
logger.debug(f"本机出口 IP 查询失败: {url}: {exc}")
continue
return None
def get_exit_ip_via_proxy(proxy: str) -> Optional[str]:
def get_exit_ip_via_proxy(proxy: str) -> str | None:
"""通过代理获取出口IP。"""
targets = [
"https://qifu-api.baidubce.com/ip/local/geo/v1/district",
@@ -198,6 +198,7 @@ def get_exit_ip_via_proxy(proxy: str) -> Optional[str]:
match = re.search(r"(\d{1,3}(?:\.\d{1,3}){3})", text)
if match:
return match.group(1)
except Exception:
except Exception as exc: # noqa: BLE001
logger.debug(f"代理出口 IP 查询失败: {url}: {exc}")
continue
return None
+4 -4
View File
@@ -75,7 +75,7 @@ class XiequAdapter(BaseWhitelistAdapter):
if r.get("IP") or r.get("ip")
]
except Exception as e:
except Exception as e: # noqa: BLE001 外部接口与任务边界需要保留宽泛异常兜底
logger.error(f"获取白名单失败: {e}")
return []
@@ -120,7 +120,7 @@ class XiequAdapter(BaseWhitelistAdapter):
logger.warning(f"白名单添加结果: {text}")
return False, text
except Exception as e:
except Exception as e: # noqa: BLE001 外部接口与任务边界需要保留宽泛异常兜底
logger.error(f"添加白名单失败: {e}")
return False, str(e)
@@ -141,7 +141,7 @@ class XiequAdapter(BaseWhitelistAdapter):
logger.warning(f"白名单删除结果: {text}")
return False, text
except Exception as e:
except Exception as e: # noqa: BLE001 外部接口与任务边界需要保留宽泛异常兜底
logger.error(f"删除白名单失败: {e}")
return False, str(e)
@@ -160,7 +160,7 @@ class XiequAdapter(BaseWhitelistAdapter):
logger.info(msg)
return True, msg
except Exception as e:
except Exception as e: # noqa: BLE001 外部接口与任务边界需要保留宽泛异常兜底
msg = f"连接失败: {e}"
logger.error(msg)
return False, msg
+3 -3
View File
@@ -115,7 +115,7 @@ class XkdailiAdapter(BaseWhitelistAdapter):
logger.warning(f"星空白名单添加失败: {msg}")
return ok, msg
except Exception as e:
except Exception as e: # noqa: BLE001 外部接口与任务边界需要保留宽泛异常兜底
logger.error(f"星空添加白名单失败: {e}")
return False, str(e)
@@ -136,7 +136,7 @@ class XkdailiAdapter(BaseWhitelistAdapter):
logger.warning(f"星空白名单删除失败: {msg}")
return ok, msg
except Exception as e:
except Exception as e: # noqa: BLE001 外部接口与任务边界需要保留宽泛异常兜底
logger.error(f"星空删除白名单失败: {e}")
return False, str(e)
@@ -159,5 +159,5 @@ class XkdailiAdapter(BaseWhitelistAdapter):
# 其他错误(如无效IP),说明认证通过了
return True, f"连接成功,API可访问: {msg}"
except Exception as e:
except Exception as e: # noqa: BLE001 外部接口与任务边界需要保留宽泛异常兜底
return False, f"连接失败: {e}"
+11 -10
View File
@@ -2,7 +2,8 @@
import threading
import time
from typing import Callable, Optional, Protocol
from collections.abc import Callable
from typing import Protocol
import requests
from loguru import logger
@@ -19,7 +20,7 @@ class WhitelistSyncer(Protocol):
def sync_ip(self, ip: str) -> tuple[bool, str]: ...
def get_local_exit_ip(self) -> Optional[str]: ...
def get_local_exit_ip(self) -> str | None: ...
class ProxyResolver:
@@ -28,11 +29,11 @@ class ProxyResolver:
def __init__(
self,
api_url: str,
whitelist_syncer: Optional[WhitelistSyncer] = None,
log_func: Optional[LogFunc] = None,
whitelist_syncer: WhitelistSyncer | None = None,
log_func: LogFunc | None = None,
sync_local_exit_ip: bool = False,
sync_whitelist_once: bool = True,
stop_event: Optional[threading.Event] = None,
stop_event: threading.Event | None = None,
):
self.api_url = api_url
self.whitelist_syncer = whitelist_syncer
@@ -40,7 +41,7 @@ class ProxyResolver:
self.sync_local_exit_ip = sync_local_exit_ip
self.sync_whitelist_once = sync_whitelist_once
self.stop_event = stop_event
self._last_synced_ip: Optional[str] = None
self._last_synced_ip: str | None = None
self._has_synced_whitelist = False
def _is_stopped(self) -> bool:
@@ -95,7 +96,7 @@ class ProxyResolver:
self,
max_attempts: int = 4,
return_all: bool = False,
) -> tuple[Optional[str | list[str]], str]:
) -> tuple[str | list[str] | None, str]:
"""
获取并验证代理。
@@ -166,7 +167,7 @@ class ProxyResolver:
f"代理预检 {attempt}/{max_attempts}: {last_error}: {text[:80]}",
)
except Exception as exc:
except Exception as exc: # noqa: BLE001 外部接口与任务边界需要保留宽泛异常兜底
last_error = f"代理API请求失败: {exc}"
self._log("warning", f"代理预检 {attempt}/{max_attempts}: {last_error}")
@@ -178,8 +179,8 @@ def resolve_working_proxy(
whitelist_platform: str = "xiequ",
whitelist_credentials: dict | None = None,
max_attempts: int = 4,
log_func: Optional[LogFunc] = None,
) -> tuple[Optional[str], str]:
log_func: LogFunc | None = None,
) -> tuple[str | None, str]:
"""从代理 API 获取可用代理,自动处理白名单同步。"""
syncer = (
DouyuWhitelistSyncer(
+6 -5
View File
@@ -1,7 +1,6 @@
"""代理可用性验证。"""
from concurrent.futures import ThreadPoolExecutor, as_completed
from typing import Optional
import requests
from loguru import logger
@@ -28,7 +27,7 @@ def verify_proxy_url(proxy_url: str, timeout: tuple = (3, 5)) -> tuple[bool, str
)
response.raise_for_status()
return True, "代理可用 → 斗鱼可达"
except Exception as exc:
except Exception as exc: # noqa: BLE001 外部接口与任务边界需要保留宽泛异常兜底
err_msg = str(exc)
if "Tunnel connection failed" in err_msg or "503" in err_msg:
detail = "代理拒绝连接(白名单可能未生效)"
@@ -45,7 +44,7 @@ def verify_proxies_concurrent(
timeout: tuple = (5, 8),
max_workers: int = 5,
return_all: bool = False,
) -> tuple[Optional[str | list[str]], str]:
) -> tuple[str | list[str] | None, str]:
"""
并发验证多个代理 URL。
@@ -76,7 +75,8 @@ def verify_proxies_concurrent(
ok, _ = future.result()
if ok:
available.append(future_map[future])
except Exception:
except Exception as exc: # noqa: BLE001
logger.debug(f"代理并发验证失败: {future_map[future]}: {exc}")
continue
if available:
logger.success(
@@ -100,7 +100,8 @@ def verify_proxies_concurrent(
if item != future:
item.cancel()
return proxy_url, msg
except Exception:
except Exception as exc: # noqa: BLE001
logger.debug(f"代理并发验证失败: {proxy_url}: {exc}")
continue
return None, f"{len(proxy_urls)} 个代理均不可用"
+3 -5
View File
@@ -1,7 +1,5 @@
"""代理模块使用的白名单适配器。"""
from typing import Optional
from .proxy_platforms import create_adapter
from .proxy_platforms.base import BaseWhitelistAdapter, _get_local_exit_ip
@@ -17,7 +15,7 @@ class DouyuWhitelistSyncer:
def __init__(
self,
platform: str = "xiequ",
credentials: Optional[dict] = None,
credentials: dict | None = None,
uid: str = "",
ukey: str = "",
):
@@ -28,7 +26,7 @@ class DouyuWhitelistSyncer:
self.platform = platform
self.credentials = credentials or {}
self._adapter: Optional[BaseWhitelistAdapter] = None
self._adapter: BaseWhitelistAdapter | None = None
if self.credentials:
self._adapter = create_adapter(platform, self.credentials)
@@ -38,5 +36,5 @@ class DouyuWhitelistSyncer:
return False, "未配置白名单凭据"
return self._adapter.sync_ip(ip)
def get_local_exit_ip(self) -> Optional[str]:
def get_local_exit_ip(self) -> str | None:
return _get_local_exit_ip()
+7 -6
View File
@@ -6,9 +6,10 @@ import hashlib
import json
import os
import time
from collections.abc import Callable, Mapping
from dataclasses import dataclass
from decimal import Decimal, InvalidOperation
from typing import Any, Callable, Mapping
from typing import Any
import requests
@@ -34,7 +35,7 @@ class FishFinRechargeConfig:
debug: bool = False
@classmethod
def from_env(cls) -> "FishFinRechargeConfig":
def from_env(cls) -> FishFinRechargeConfig:
"""从环境变量读取配置,不在代码或数据库中保存商户密钥。"""
timeout = float(os.getenv("FISH_FIN_RECHARGE_TIMEOUT", "20"))
return cls(
@@ -189,7 +190,7 @@ class FishFinRechargeClient:
"sign_params": sign_params,
# 不把可重放签名原文或 AppSecret 写入日志,只记录待签名串摘要。
"sign_source_digest": hashlib.sha256(
f"{sign_query}{method.upper()}".encode("utf-8")
f"{sign_query}{method.upper()}".encode()
).hexdigest()[:12],
}
)
@@ -240,7 +241,7 @@ class FishFinRechargeClient:
return payload
@staticmethod
def _amount(value: Decimal | int | float | str) -> str:
def _amount(value: Decimal | float | str) -> str:
"""规范化金额,避免浮点数表达式进入签名或订单请求。"""
try:
price = Decimal(str(value))
@@ -251,7 +252,7 @@ class FishFinRechargeClient:
return format(price.normalize(), "f")
@classmethod
def _json_amount(cls, value: Decimal | int | float | str) -> int | float:
def _json_amount(cls, value: Decimal | float | str) -> int | float:
"""按文档以 JSON 数字发送金额,整数不附带无意义的小数位。"""
amount_text = cls._amount(value)
return int(amount_text) if "." not in amount_text else float(amount_text)
@@ -267,7 +268,7 @@ class FishFinRechargeClient:
self,
*,
buy_num: int,
pay_amount: Decimal | int | float | str,
pay_amount: Decimal | float | str,
out_order_id: str,
product_id: str,
recharge_arg: list[dict[str, Any]],
+3 -138
View File
@@ -1,8 +1,9 @@
import random
import hashlib
import random
from typing import Any
from Crypto.Cipher import AES, PKCS1_v1_5
from Crypto.PublicKey import RSA
from Crypto.Cipher import PKCS1_v1_5, AES
# 随机产生4个字符组成的字符串
@@ -247,141 +248,5 @@ def simple_md5(message: str) -> str:
def verify_result(msg):
return hashlib.md5(msg.encode()).hexdigest()
# 轮移位常量
shifts = [
7,
12,
17,
22,
7,
12,
17,
22,
7,
12,
17,
22,
7,
12,
17,
22,
5,
9,
14,
20,
5,
9,
14,
20,
5,
9,
14,
20,
5,
9,
14,
20,
4,
11,
16,
23,
4,
11,
16,
23,
4,
11,
16,
23,
4,
11,
16,
23,
6,
10,
15,
21,
6,
10,
15,
21,
6,
10,
15,
21,
6,
10,
15,
21,
]
# K常数(与JavaScript版本中的常数对应)
K = [
0xD76AA478,
0xE8C7B756,
0x242070DB,
0xC1BDCEEE,
0xF57C0FAF,
0x4787C62A,
0xA8304613,
0xFD469501,
0x698098D8,
0x8B44F7AF,
0xFFFF5BB1,
0x895CD7BE,
0x6B901122,
0xFD987193,
0xA679438E,
0x49B40821,
0xF61E2562,
0xC040B340,
0x265E5A51,
0xE9B6C7AA,
0xD62F105D,
0x02441453,
0xD8A1E681,
0xE7D3FBC8,
0x21E1CDE6,
0xC33707D6,
0xF4D50D87,
0x455A14ED,
0xA9E3E905,
0xFCEFA3F8,
0x676F02D9,
0x8D2A4C8A,
0xFFFA3942,
0x8771F681,
0x6D9D6122,
0xFDE5380C,
0xA4BEEA44,
0x4BDECFA9,
0xF6BB4B60,
0xBEBFBC70,
0x289B7EC6,
0xEAA127FA,
0xD4EF3085,
0x04881D05,
0xD9D4D039,
0xE6DB99E5,
0x1FA27CF8,
0xC4AC5665,
0xF4292244,
0x432AFF97,
0xAB9423A7,
0xFC93A039,
0x655B59C3,
0x8F0CCC92,
0xFFEFF47D,
0x85845DD1,
0x6FA87E4F,
0xFE2CE6E0,
0xA3014314,
0x4E0811A1,
0xF7537E82,
0xBD3AF235,
0x2AD7D2BB,
0xEB86D391,
]
# 实际实现...
return verify_result(message)
+2 -4
View File
@@ -1,8 +1,8 @@
import requests
import cv2
import numpy as np
from PIL import Image
import requests
from loguru import logger
from PIL import Image
REQUEST_TIMEOUT = (3.05, 12)
@@ -103,9 +103,7 @@ def restore_geetest_image(input_path: str, output_path: str) -> None:
# 打开混淆图像
img = Image.open(input_path)
new_img = Image.new("RGB", (260, 160))
r = 160
for _ in range(len(Ut)):
a = r / 2
c = Ut[_] % 26 * 12 + 1
u = 80 if Ut[_] > 25 else 0
l = img.crop(box=(c, u, c + 10, u + 80))
+20 -19
View File
@@ -1,8 +1,9 @@
import time
import requests
import json
import re
from typing import Mapping, Optional, Tuple
import time
from collections.abc import Mapping
import requests
from loguru import logger
REQUEST_TIMEOUT = (10, 30)
@@ -12,9 +13,9 @@ PASSPORT_REFERER = "https://passport.douyu.com/"
def _get(
url: str,
*,
params: Optional[dict] = None,
headers: Optional[dict] = None,
proxies: Optional[Mapping[str, str]] = None,
params: dict | None = None,
headers: dict | None = None,
proxies: Mapping[str, str] | None = None,
) -> requests.Response:
"""发送极验 GET 请求,确保使用同一个代理出口。"""
return requests.get(
@@ -55,7 +56,7 @@ def _parse_json_response(response: requests.Response, source: str) -> dict:
raise ValueError(f"{source} 返回的不是有效 JSON: {preview}") from exc
def get_challenge_gt_bak() -> Tuple[str, str]:
def get_challenge_gt_bak() -> tuple[str, str]:
headers = {
"accept": "application/json, text/javascript, */*; q=0.01",
"accept-language": "zh-CN,zh;q=0.9",
@@ -72,7 +73,7 @@ def get_challenge_gt_bak() -> Tuple[str, str]:
}
params = {
"t": str(int(round(time.time() * 1000))),
"t": str(round(time.time() * 1000)),
}
response = requests.get(
@@ -85,7 +86,7 @@ def get_challenge_gt_bak() -> Tuple[str, str]:
return data["gt"], data["challenge"]
def get_challenge_gt() -> Tuple[str, str]:
def get_challenge_gt() -> tuple[str, str]:
headers = {
"User-Agent": "Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/146.0.0.0 Safari/537.36",
"Accept": "application/json, text/javascript, */*; q=0.01",
@@ -116,7 +117,7 @@ def get_challenge_gt() -> Tuple[str, str]:
raise ValueError(f"斗鱼登录接口返回中缺少极验参数: {preview}") from exc
def get_js_address(gt: str, proxies: Optional[Mapping[str, str]] = None) -> dict:
def get_js_address(gt: str, proxies: Mapping[str, str] | None = None) -> dict:
headers = {
"accept": "*/*",
"accept-language": "zh-CN,zh;q=0.9",
@@ -132,7 +133,7 @@ def get_js_address(gt: str, proxies: Optional[Mapping[str, str]] = None) -> dict
params = {
"gt": gt,
"callback": "geetest_" + str(int(round(time.time() * 1000))),
"callback": "geetest_" + str(round(time.time() * 1000)),
}
response = _get(
@@ -149,8 +150,8 @@ def get_c_s(
gt: str,
challenge: str,
w: str,
proxies: Optional[Mapping[str, str]] = None,
) -> Tuple[list[int], str]:
proxies: Mapping[str, str] | None = None,
) -> tuple[list[int], str]:
headers = {
"accept": "*/*",
"accept-language": "zh-CN,zh;q=0.9",
@@ -172,7 +173,7 @@ def get_c_s(
+ "&lang=zh-cn&pt=0&client_type=web&w="
+ w
+ "&callback=geetest_"
+ str(int(round(time.time() * 1000))),
+ str(round(time.time() * 1000)),
headers=headers,
proxies=proxies,
)
@@ -184,7 +185,7 @@ def req_fullpage_validate(
gt: str,
challenge: str,
w: str,
proxies: Optional[Mapping[str, str]] = None,
proxies: Mapping[str, str] | None = None,
) -> dict:
"""HAR 中的 fullpage 最终校验,成功后直接返回 validate。"""
headers = {
@@ -209,7 +210,7 @@ def req_fullpage_validate(
+ "&lang=zh-cn&pt=0&client_type=web&w="
+ w
+ "&callback=geetest_"
+ str(int(round(time.time() * 1000))),
+ str(round(time.time() * 1000)),
headers=headers,
proxies=proxies,
)
@@ -240,7 +241,7 @@ def req_slide(gt: str, challenge: str, w2: str) -> None:
+ "&lang=zh-cn&pt=0&client_type=web&w="
+ w2
+ "&callback=geetest_"
+ str(int(round(time.time() * 1000))),
+ str(round(time.time() * 1000)),
headers=headers,
timeout=REQUEST_TIMEOUT,
)
@@ -276,7 +277,7 @@ def get_picture(gt: str, challenge: str) -> tuple[str, str, list[int], str, str,
"isPC": "true",
"autoReset": "true",
"width": "100%",
"callback": "geetest_" + str(int(round(time.time() * 1000))),
"callback": "geetest_" + str(round(time.time() * 1000)),
}
response = requests.get(
@@ -342,7 +343,7 @@ def req_end(gt: str, challenge: str, w: str) -> dict:
+ "&lang=zh-cn&%24_BCm=0&client_type=web&w="
+ w
+ "&callback=geetest_"
+ str(int(round(time.time() * 1000))),
+ str(round(time.time() * 1000)),
headers=headers,
timeout=REQUEST_TIMEOUT,
)
+2 -3
View File
@@ -1,9 +1,8 @@
import time
import random
from typing import Optional
import time
def generate_fake_performance_timing(base_time: Optional[int] = None) -> dict[str, int]:
def generate_fake_performance_timing(base_time: int | None = None) -> dict[str, int]:
"""
生成伪造的浏览器性能时间戳数据
+4 -4
View File
@@ -1,6 +1,6 @@
import random
import math
from typing import Any, Optional
import random
from typing import Any
# 生成类人的鼠标轨迹
@@ -91,7 +91,7 @@ def generate_realistic_trajectory(
# 处理原始轨迹数组
def process_mouse_trajectory(
events: list[Any], max_records: Optional[int] = None
events: list[Any], max_records: int | None = None
) -> dict[str, Any]:
"""
处理鼠标/触摸轨迹数据,将绝对坐标转换为相对坐标和时间差
@@ -592,7 +592,7 @@ def H(t: int, e: str) -> str:
n = 36 * r[0] + r[1]
# 计算目标值
a = round(t) + n
a = (t) + n
# 构建字符池
_ = [[], [], [], [], []]
+19 -17
View File
@@ -1,35 +1,37 @@
import time
import random
import json
import random
import time
from loguru import logger
from core.geetest.common.trajectory import (
generate_realistic_trajectory,
process_mouse_trajectory,
compress_trajectory,
TrajectoryEncoder,
H,
)
from core.geetest.common.crypto import (
four_random_chart,
RSA_jiami_r,
AES_O,
geetest_base64_encode,
RSA_jiami_r,
encrypt_string,
four_random_chart,
geetest_base64_encode,
simple_md5,
)
from core.geetest.common.imaging import download_picture
from core.geetest.common.network import (
get_c_s,
get_challenge_gt,
get_js_address,
get_c_s,
req_slide,
get_picture,
req_end,
req_slide,
)
from core.geetest.common.performance import (
generate_fake_performance_timing,
get_slide_track,
)
from core.geetest.common.trajectory import (
H,
TrajectoryEncoder,
compress_trajectory,
generate_realistic_trajectory,
process_mouse_trajectory,
)
def _generate_seed() -> str:
@@ -83,7 +85,7 @@ def get_w2(gt: str, challenge: str, c: list[int], s: str, str_16: str) -> str:
"u": fake_timing["loadEventEnd"],
}
first_time = int(round(time.time() * 1000)) # 伪造脚本开始运行时间
first_time = round(time.time() * 1000) # 伪造脚本开始运行时间
guiji_yuanshu_shuzu = generate_realistic_trajectory(
start_x=random.randint(400, 600), # 起始位置随机
@@ -97,7 +99,7 @@ def get_w2(gt: str, challenge: str, c: list[int], s: str, str_16: str) -> str:
compressed = compress_trajectory(trajectory)
tt = encrypt_string(compressed, c, s)
passtime = str(int(round(time.time() * 1000)) - first_time)
passtime = str(round(time.time() * 1000) - first_time)
rp = simple_md5(gt + challenge + passtime)
@@ -217,4 +219,4 @@ def run_solver() -> None:
w3 = get_w3(str_16, challenge, hkjl, c, s, gt)
# 最后的验证
message = req_end(gt, challenge, w3)
req_end(gt, challenge, w3)
+7 -7
View File
@@ -33,23 +33,23 @@ if TYPE_CHECKING:
)
__all__ = [
"HuyaHttpClient",
"HuyaWssClient",
"GetUserScoreReq",
"GetUserScoreResp",
"HuyaAppLoginError",
"HuyaAppPasswordLogin",
"HuyaAppQrAuthRequiredError",
"HuyaCredentialError",
"HuyaHttpClient",
"HuyaLoginError",
"HuyaLoginResult",
"HuyaPasswordLogin",
"HuyaAppLoginError",
"HuyaAppQrAuthRequiredError",
"HuyaAppPasswordLogin",
"HuyaSmsCodeResult",
"HuyaSmsLogin",
"HuyaVerificationError",
"HuyaVerificationSolver",
"login_huya_password",
"HuyaWssClient",
"login_huya_app_password",
"login_huya_password",
"login_huya_sms",
"send_huya_sms_code",
"solve_huya_verification",
@@ -98,8 +98,8 @@ def __getattr__(name: str):
}:
from .app_login import (
HuyaAppLoginError,
HuyaAppQrAuthRequiredError,
HuyaAppPasswordLogin,
HuyaAppQrAuthRequiredError,
login_huya_app_password,
)
+1 -2
View File
@@ -35,13 +35,12 @@ import os
import sys
import time
from .app_login import HuyaAppPasswordLogin
from .device_profile import (
HDID32, # 登录帧 t1.t0 32hex 常量 (R15 公式结果, k1 来源见 R39; device_profile.py)
_load_db,
_save_db,
get_profile, # 幂等画像: 同账号永远复用同一套 (data/huya_device_profiles.json)
)
from .app_login import HuyaAppPasswordLogin
# ---------------------------------------------------------------------------
+3 -3
View File
@@ -306,7 +306,7 @@ class UserPrizeRecordItem(TafStruct):
self.exchangeDate = ins.read_int64(21, default=self.exchangeDate)
while True:
pos = ins.buf.tell()
tag, dtype = ins.read_head()
_tag, dtype = ins.read_head()
if dtype == TafType.STRUCT_END:
ins.buf.seek(pos)
return
@@ -406,7 +406,7 @@ class ActTaskPrizeInfo(TafStruct):
self.extra = ins.read_map(12)
while True:
pos = ins.buf.tell()
tag, dtype = ins.read_head()
_tag, dtype = ins.read_head()
if dtype == TafType.STRUCT_END:
ins.buf.seek(pos)
return
@@ -494,7 +494,7 @@ class ActTaskDetailItem(TafStruct):
self.endTime = ins.read_string(26, default=self.endTime)
while True:
pos = ins.buf.tell()
tag, dtype = ins.read_head()
_tag, dtype = ins.read_head()
if dtype == TafType.STRUCT_END:
ins.buf.seek(pos)
return
+8 -8
View File
@@ -130,7 +130,7 @@ def wup_password_login_raw(
固定 action/device_id。风控重试调用方应显式复用同一注册结果。
注册链失败抛 ``HuyaAppLoginError``,绝不静默回退旧固定值。
"""
uid_str = account[3:] if account.startswith("hy_") else account
uid_str = account.removeprefix("hy_")
dev = dict(device_info) if device_info is not None else get_profile(account)
mj, ua, _old_sd = _golden_session_assets()
if not safedeviceid:
@@ -232,7 +232,7 @@ def solve_safe_auth(
result = solver.solve(risk_url)
except HuyaQrAuthRequiredError:
raise
except Exception as exc:
except Exception as exc: # noqa: BLE001 外部接口与任务边界需要保留宽泛异常兜底
last_err = exc
logger.warning(f"[safe_auth] 第 {attempt + 1} 次过验异常: {exc}")
time.sleep(1.0)
@@ -304,7 +304,7 @@ def login_cred_with_flow(
logger.info(f"[huya-app] 第 {rnd + 1} 轮触发安全验证: {kind}")
if "qr_auth" in risk_url:
raise HuyaAppQrAuthRequiredError(
f"该账号 App 渠道要求扫码验证(qr_auth),请先在手机虎牙 App 上正常登录一次建立设备信任。"
"该账号 App 渠道要求扫码验证(qr_auth),请先在手机虎牙 App 上正常登录一次建立设备信任。"
)
solve_safe_auth(risk_url, proxies=proxies, device_info=dev)
logger.info("[huya-app] safe_auth 滑块过验成功,重发 WUP 登录...")
@@ -425,8 +425,8 @@ class HuyaAppPasswordLogin:
from .device_profile import record_login
record_login(self.username, result.success, result.message)
except Exception: # 元数据记录失败不影响登录结果
pass
except Exception as exc: # noqa: BLE001 # 元数据记录失败不影响登录结果
logger.debug(f"登录元数据记录失败: {exc}")
return result
def _login_impl(self) -> HuyaLoginResult:
@@ -450,7 +450,7 @@ class HuyaAppPasswordLogin:
message=str(exc),
code="QR_AUTH_REQUIRED",
)
except Exception as exc:
except Exception as exc: # noqa: BLE001 外部接口与任务边界需要保留宽泛异常兜底
return HuyaLoginResult(
success=False,
message=f"App 登录凭证获取失败: {exc}",
@@ -478,7 +478,7 @@ class HuyaAppPasswordLogin:
if env.uid != uid:
struct.pack_into(">Q", raw, env.uid_off, uid)
wup = base64.b64encode(bytes(raw)).decode("ascii")
except Exception as exc:
except Exception as exc: # noqa: BLE001 外部接口与任务边界需要保留宽泛异常兜底
return HuyaLoginResult(
success=False,
message=f"证书铸造/信封补丁失败: {exc}",
@@ -603,7 +603,7 @@ class HuyaAppPasswordLogin:
sdid=sdid,
context=pc.context,
)
except Exception as exc:
except Exception as exc: # noqa: BLE001 外部接口与任务边界需要保留宽泛异常兜底
return HuyaLoginResult(
success=False,
message=f"扫码绑定兑换 Cookie 失败: {exc}",
+4 -4
View File
@@ -6,7 +6,7 @@ import threading
import time
from collections.abc import Mapping
from dataclasses import dataclass
from datetime import datetime
from datetime import UTC, datetime
from core.sms_provider import SmsLine, SmsProviderClient
@@ -76,7 +76,7 @@ def register_huya_with_sms_line(
sms_url=item.url,
)
sent_at = datetime.now()
sent_at = datetime.now(UTC)
try:
code_result = send_huya_sms_code(phone=phone, proxies=proxies)
except HuyaLoginError as exc:
@@ -89,7 +89,7 @@ def register_huya_with_sms_line(
normalized_phone=normalized_phone,
sms_url=item.url,
)
except Exception as exc:
except Exception as exc: # noqa: BLE001 外部接口与任务边界需要保留宽泛异常兜底
return HuyaAutoRegisterResult(
phone=phone,
provider=item.provider,
@@ -155,7 +155,7 @@ def register_huya_with_sms_line(
sms_url=item.url,
attempts=attempts,
)
except Exception as exc:
except Exception as exc: # noqa: BLE001 外部接口与任务边界需要保留宽泛异常兜底
return HuyaAutoRegisterResult(
phone=phone,
provider=item.provider,
-1
View File
@@ -5,7 +5,6 @@
from __future__ import annotations
import base64
import os
import struct
+2 -3
View File
@@ -10,7 +10,7 @@ import time
import uuid
from collections.abc import Mapping
from dataclasses import dataclass
from datetime import datetime
from datetime import UTC, datetime
from urllib.parse import quote, urlsplit, urlunsplit
import requests
@@ -30,7 +30,6 @@ from .login import (
generate_request_id,
)
CHANGE_PASSWORD_VERSION = "2.5"
CHANGE_PASSWORD_CHECK_URI = "60011"
CHANGE_PASSWORD_SEND_SMS_URI = "60003"
@@ -497,7 +496,7 @@ def change_huya_password_with_sms_line(
) -> HuyaChangePasswordResult:
"""使用同一手机号接码链接完成改密短信验证。"""
changer = HuyaPasswordChanger(uid=uid, cookie=cookie, proxies=proxies)
sent_at = datetime.now()
sent_at = datetime.now(UTC)
code_result = changer.send_code()
if not code_result.success or not code_result.session_data:
return HuyaChangePasswordResult(
-1
View File
@@ -4,7 +4,6 @@ from __future__ import annotations
from collections.abc import Iterable, Mapping
import requests
from requests.cookies import RequestsCookieJar
+8 -6
View File
@@ -22,6 +22,8 @@ import random
from collections.abc import Mapping
from pathlib import Path
from loguru import logger
ROOT = Path(__file__).resolve().parent.parent.parent
DATA_DIR = ROOT / "data"
PRIMARY_PROFILE_DB = DATA_DIR / "huya_device_profiles.json"
@@ -119,13 +121,13 @@ def _load_db() -> dict:
if PRIMARY_PROFILE_DB.exists():
try:
return json.loads(PRIMARY_PROFILE_DB.read_text("utf-8"))
except Exception:
pass
except Exception as exc: # noqa: BLE001
logger.debug(f"读取主设备画像库失败: {exc}")
if FALLBACK_PROFILE_DB.exists():
try:
return json.loads(FALLBACK_PROFILE_DB.read_text("utf-8"))
except Exception:
pass
except Exception as exc: # noqa: BLE001
logger.debug(f"读取备用设备画像库失败: {exc}")
return {}
@@ -135,8 +137,8 @@ def _save_db(db: dict) -> None:
PRIMARY_PROFILE_DB.write_text(
json.dumps(db, indent=2, ensure_ascii=False), encoding="utf-8"
)
except Exception:
pass
except Exception as exc: # noqa: BLE001
logger.debug(f"保存设备画像库失败: {exc}")
def _enrich_profile(profile: dict) -> tuple[dict, bool]:
+9 -8
View File
@@ -10,6 +10,8 @@ import json
import struct
from pathlib import Path
from loguru import logger
INT8, INT16, INT32, INT64 = 0x00, 0x01, 0x02, 0x03
STRING1, STRING4 = 0x06, 0x07
MAP, LIST = 0x08, 0x09
@@ -116,7 +118,7 @@ class Envelope:
self._parse()
@classmethod
def load(cls, path: str | Path | None = None) -> "Envelope":
def load(cls, path: str | Path | None = None) -> Envelope:
"""加载信封模板,支持从文件加载或使用内嵌金样本。"""
if path:
p = Path(path)
@@ -131,12 +133,12 @@ class Envelope:
if candidate.exists():
try:
return cls._load_from_path(candidate)
except Exception:
pass
except Exception as exc: # noqa: BLE001
logger.debug(f"加载证书信封候选文件失败: {candidate}: {exc}")
return cls(base64.b64decode(DEFAULT_QURL_B64))
@classmethod
def _load_from_path(cls, p: Path) -> "Envelope":
def _load_from_path(cls, p: Path) -> Envelope:
if p.suffix == ".json":
j = json.loads(p.read_text("utf-8"))
q = next(
@@ -172,7 +174,6 @@ class Envelope:
assert (h & 0x0F) == MAP
cnt, p = _read_len_int(d, p)
for _ in range(cnt):
kh = d[p]
p += 1
kln = d[p]
p += 1
@@ -236,7 +237,7 @@ class Envelope:
assert self.uid_off is not None
return struct.unpack_from(">Q", self.raw, self.uid_off)[0]
def patch_uid(self, uid: int) -> "Envelope":
def patch_uid(self, uid: int) -> Envelope:
assert self.uid_off is not None
struct.pack_into(">Q", self.raw, self.uid_off, uid)
return self
@@ -246,7 +247,7 @@ class Envelope:
assert self.cert_off is not None
return bytes(self.raw[self.cert_off : self.cert_off + self.cert_len])
def patch_cert(self, cert: bytes) -> "Envelope":
def patch_cert(self, cert: bytes) -> Envelope:
assert self.cert_off is not None
b64 = base64.b64encode(cert)
if len(b64) != self.cert_len:
@@ -256,7 +257,7 @@ class Envelope:
self.raw[self.cert_off : self.cert_off + self.cert_len] = b64
return self
def patch_session(self, session: int) -> "Envelope":
def patch_session(self, session: int) -> Envelope:
if self.tag4_span:
struct.pack_into(">I", self.raw, self.tag4_span[0], session & 0xFFFFFFFF)
d = self.raw
+15 -17
View File
@@ -3,6 +3,9 @@ TAF/WUP 帧解码器 — 将二进制帧转为可读摘要,用于日志输出
"""
from typing import Any, cast
from loguru import logger
from .taf_protocol import TafInputStream, TafType
from .wup_protocol import normalize_wup_payload
@@ -100,13 +103,13 @@ def _decode_taf_struct(ins: TafInputStream, depth: int = 0) -> dict:
if depth < 5:
try:
val = _decode_taf_value(ins, dtype, depth)
except Exception:
except Exception: # noqa: BLE001 外部接口与任务边界需要保留宽泛异常兜底
val = f"<decode_err:0x{dtype:02x}>"
else:
val = f"<...>"
val = "<...>"
try:
ins.skip_field(dtype)
except Exception:
except Exception: # noqa: BLE001 外部接口与任务边界需要保留宽泛异常兜底
break
key = f"tag{tag}"
if key in fields:
@@ -152,13 +155,7 @@ def _decode_wup_body(body: bytes) -> dict:
if tag > 10:
break
ins.read_head()
if tag == 1:
if dtype != TafType.ZERO:
ins._read_int_value(dtype)
elif tag in (2, 3):
if dtype != TafType.ZERO:
ins._read_int_value(dtype)
elif tag == 4:
if tag == 1 or tag in (2, 3) or tag == 4:
if dtype != TafType.ZERO:
ins._read_int_value(dtype)
elif tag == 5:
@@ -200,19 +197,20 @@ def _decode_wup_body(body: bytes) -> dict:
if v:
try:
tins = TafInputStream(v)
ttag, tdt = tins.peek_head()
_ttag, tdt = tins.peek_head()
if tdt == TafType.STRUCT_BEGIN:
tins.read_head()
result[k] = _decode_taf_struct(tins)
else:
result[k] = f"<{len(v)}B>"
except Exception:
except Exception as exc: # noqa: BLE001
logger.debug(f"TAF 嵌套结构解码失败: {exc}")
result[k] = f"<{len(v)}B>"
else:
result[k] = _decode_taf_value(sins, vt)
except Exception:
pass
except Exception as e:
except Exception as exc: # noqa: BLE001
logger.debug(f"TAF 字段解码失败: {exc}")
except Exception as e: # noqa: BLE001 外部接口与任务边界需要保留宽泛异常兜底
result["err"] = str(e)
return result
@@ -291,7 +289,7 @@ def format_wss_log(body: bytes, cmd: int, seq: int, direction: str) -> str:
try:
ins = TafInputStream(clean)
# 看第一个 head
tag, dtype = ins.peek_head()
_tag, dtype = ins.peek_head()
if dtype == TafType.STRUCT_BEGIN:
ins.read_head()
fields = _decode_taf_struct(ins)
@@ -319,6 +317,6 @@ def format_wss_log(body: bytes, cmd: int, seq: int, direction: str) -> str:
if fields:
return f"{prefix} {cmd_name} {_fmt_fields(cast(dict[str, Any], _truncate(fields)))}"
return f"{prefix} {cmd_name}"
except Exception:
except Exception: # noqa: BLE001 外部接口与任务边界需要保留宽泛异常兜底
cmd_name = CMD_NAMES.get(cmd, f"0x{cmd:02x}")
return f"{prefix} {cmd_name} ({len(body)}B)"
+14 -14
View File
@@ -11,13 +11,13 @@ import base64
import hashlib
import json
import random
import struct
import urllib.parse
import urllib.request
from typing import Any, Optional, Callable
from collections.abc import Callable
from typing import Any
from .cookie_utils import cookie_pairs, normalize_cookie_pairs, normalize_huya_cookie
from .taf_protocol import TafOutputStream, TafInputStream, TafType, TafStruct
from .taf_protocol import TafInputStream, TafOutputStream, TafStruct
from .wup_protocol import WupRequest, WupResponse
CDNWS_HOST = "cdnws.api.huya.com"
@@ -92,7 +92,7 @@ class WSConnectParaInfo(TafStruct):
def _gen_trace_id() -> str:
"""生成 sTraceId (格式 hex8:hex8:0:0HAR 实证)"""
h = "%016x" % random.getrandbits(64)
h = f"{random.getrandbits(64):016x}"
return f"{h}:{h}:0:0"
@@ -294,7 +294,7 @@ class HuyaHttpClient:
import gzip
resp_data = gzip.decompress(resp_data)
except Exception as e:
except Exception as e: # noqa: BLE001 外部接口与任务边界需要保留宽泛异常兜底
self.logger(f"[HTTP] ❌ 请求失败: {type(e).__name__}: {e}")
return None
@@ -611,13 +611,13 @@ class HuyaHttpClient:
or resp.headers.get("Content-Encoding") == "gzip"
):
resp_data = gzip.decompress(resp_data)
except Exception as e:
except Exception as e: # noqa: BLE001 外部接口与任务边界需要保留宽泛异常兜底
self.logger(f"[LIVELINK] ❌ 小程序码请求失败: {type(e).__name__}: {e}")
return None
try:
data = json.loads(resp_data.decode("utf-8", "replace"))
except Exception as e:
except Exception as e: # noqa: BLE001 外部接口与任务边界需要保留宽泛异常兜底
self.logger(f"[LIVELINK] ❌ 小程序码响应解析失败: {type(e).__name__}: {e}")
return None
@@ -679,13 +679,13 @@ class HuyaHttpClient:
or resp.headers.get("Content-Encoding") == "gzip"
):
resp_data = gzip.decompress(resp_data)
except Exception as e:
except Exception as e: # noqa: BLE001 外部接口与任务边界需要保留宽泛异常兜底
self.logger(f"[LIVELINK] ❌ 二维码状态请求失败: {type(e).__name__}: {e}")
return None
try:
data = json.loads(resp_data.decode("utf-8", "replace"))
except Exception as e:
except Exception as e: # noqa: BLE001 外部接口与任务边界需要保留宽泛异常兜底
self.logger(
f"[LIVELINK] ❌ 二维码状态响应解析失败: {type(e).__name__}: {e}"
)
@@ -849,11 +849,11 @@ class HuyaHttpClient:
):
"""order_type=None 时自动尝试从 1 到 10 找到有效值"""
from .shop_structs import (
CreateOrderReqV5,
CreateOrderRsp,
CreateOrderAccountParam,
CreateOrderExtraParam,
CreateOrderPromotionParam,
CreateOrderAccountParam,
CreateOrderReqV5,
CreateOrderRsp,
)
# 如果指定了具体值,直接试
@@ -951,7 +951,7 @@ class HuyaHttpClient:
baseinfo = self._generate_rpc_baseinfo(uid, guid, cookie)
url = f"https://{CDNWS_HOST}/?baseinfo={baseinfo}"
self.logger(f"[HTTP] POST shopMiddleUI.payOrderSubmitV5")
self.logger("[HTTP] POST shopMiddleUI.payOrderSubmitV5")
self.logger(f"[HTTP] 发送 hex前60: {wup_data[:60].hex()}")
req = urllib.request.Request(
@@ -972,7 +972,7 @@ class HuyaHttpClient:
import gzip
resp_data = gzip.decompress(resp_data)
except Exception as e:
except Exception as e: # noqa: BLE001 外部接口与任务边界需要保留宽泛异常兜底
self.logger(f"[HTTP] ❌ payOrderSubmitV5 失败: {e}")
return None
+3 -5
View File
@@ -14,12 +14,11 @@ from http.cookies import SimpleCookie
from urllib.parse import quote, urlsplit, urlunsplit
import requests
from requests.cookies import RequestsCookieJar
from loguru import logger
from requests.cookies import RequestsCookieJar
from .cookie_utils import normalize_huya_cookie
APP_ID = "5002"
APP_VERSION = "2.6"
APP_SIGN = "1ce3bf682483d03f146f58232ec10635"
@@ -101,7 +100,7 @@ def generate_context(device_id: str | None = None, mid: str | None = None) -> st
def generate_request_id() -> str:
"""生成 requestId,形态参考旧实现的日内毫秒数。"""
now = dt.datetime.now(dt.timezone.utc)
now = dt.datetime.now(dt.UTC)
midnight = now.replace(hour=0, minute=0, second=0, microsecond=0)
return str(int((now - midnight).total_seconds() * 1000))
@@ -381,8 +380,7 @@ class HuyaPasswordLogin:
last_exc: Exception | None = None
for attempt in range(max_ip_retries + 1):
if attempt > 0:
if not self._swap_proxy():
if attempt > 0 and not self._swap_proxy():
logger.warning("无可用代理可切换,停止换 IP 重试")
break
try:
+15 -16
View File
@@ -5,8 +5,7 @@
来源:m-shop.yaoguo.com 的 jce/ShopFacade.js、api/orderui.ts、api/mall/PlayMallNewHome.ts
"""
from typing import List, Dict, Optional
from .taf_protocol import TafOutputStream, TafInputStream, TafStruct, TafType
from .taf_protocol import TafInputStream, TafOutputStream, TafStruct, TafType
class OrderType:
@@ -68,7 +67,7 @@ def _skip_to_struct_end(ins: TafInputStream):
"""跳过当前结构里未解析的尾部字段,停在 STRUCT_END 前。"""
while True:
pos = ins.buf.tell()
tag, dtype = ins.read_head()
_tag, dtype = ins.read_head()
if dtype == TafType.STRUCT_END:
ins.buf.seek(pos)
return
@@ -387,7 +386,7 @@ class GoodsPriceInfo(TafStruct):
self.spuId: str = ""
self.minPrice: int = 0
self.maxPrice: int = 0
self.skuMap: Dict[int, GoodsSkuItem] = {}
self.skuMap: dict[int, GoodsSkuItem] = {}
self.stock: int = 0
self.buyLimit: int = 0
self.price: int = 0
@@ -427,7 +426,7 @@ class GoodsPriceInfo(TafStruct):
def first_sku_id(self) -> int:
if not self.skuMap:
return 0
return sorted(self.skuMap.keys())[0]
return min(self.skuMap.keys())
@property
def sku_list(self) -> list[dict]:
@@ -500,7 +499,7 @@ class OrderListGoodsDetail(TafStruct):
self.buyerUid: int = 0 # tag 18
self.virtualType: int = 0 # tag 19
self.quantity: int = 0 # tag 20
self.shopInfo: Optional[OrderListShopInfo] = None # tag 21
self.shopInfo: OrderListShopInfo | None = None # tag 21
self.points: int = 0 # tag 23
def read_from(self, ins: TafInputStream):
@@ -544,7 +543,7 @@ class OrderListItem(TafStruct):
self.totalPrice: int = 0 # tag 12 分
self.createTime: int = 0 # tag 14 毫秒时间戳
self.payTime: int = 0 # tag 15 毫秒时间戳
self.goodsDetail: Optional[OrderListGoodsDetail] = None # tag 16
self.goodsDetail: OrderListGoodsDetail | None = None # tag 16
def read_from(self, ins: TafInputStream):
self.bizOrderId = ins.read_string(0, default=self.bizOrderId)
@@ -611,7 +610,7 @@ class QueryUserOrderListRsp(TafStruct):
self.code: int = 0
self.message: str = ""
self.totalCount: int = 0
self.orders: List[OrderListItem] = []
self.orders: list[OrderListItem] = []
@staticmethod
def _read_order_item(ins: TafInputStream, _tag: int):
@@ -692,9 +691,9 @@ class CreateOrderPromotionParam(TafStruct):
class CreateOrderAccountParam(TafStruct):
def __init__(self):
self.payoutTypeList: List[int] = [] # tag 0 Vector<INT32>
self.payoutTypeList: list[int] = [] # tag 0 Vector<INT32>
self.payoutChargeAmount: int = 0 # tag 1
self.cancelPayoutTypeList: List[int] = [] # tag 2
self.cancelPayoutTypeList: list[int] = [] # tag 2
self.recycleSupplierId: int = 0 # tag 3
self.claimPrice: int = 0 # tag 4
@@ -739,20 +738,20 @@ class CreateOrderReqV5(TafStruct):
self.gameId: str = "" # tag 8
self.orderId: int = 0 # tag 9
self.src: int = 0 # tag 10
self.couponUserIds: List[int] = [] # tag 11 Vector<INT64>
self.couponUserIds: list[int] = [] # tag 11 Vector<INT64>
self.orderType: int = 0 # tag 12
self.extraParam: Optional[CreateOrderExtraParam] = None # tag 13
self.extraParam: CreateOrderExtraParam | None = None # tag 13
self.scene: int = 0 # tag 14
self.promotionItems: List = [] # tag 15 Vector<PromotionItem>
self.promotionItems: list = [] # tag 15 Vector<PromotionItem>
self.sourceId: str = "" # tag 16
self.env: Dict[str, str] = {} # tag 17 Map<STRING,STRING>
self.env: dict[str, str] = {} # tag 17 Map<STRING,STRING>
self.orderScene: int = 0 # tag 18
self.watchWord: str = "" # tag 19
self.marketingChannel: str = "" # tag 20
self.promotionParam: Optional[CreateOrderPromotionParam] = None # tag 21
self.promotionParam: CreateOrderPromotionParam | None = None # tag 21
self.externalTraceKey: str = "" # tag 22
self.kefuUid: int = 0 # tag 23
self.accountParam: Optional[CreateOrderAccountParam] = None # tag 24
self.accountParam: CreateOrderAccountParam | None = None # tag 24
self.parentOrderId: int = 0 # tag 25
self.vendorAccountType: str = "" # tag 26
self.vendorAccountVal: str = "" # tag 27
+1 -2
View File
@@ -29,7 +29,6 @@ from .login import (
generate_request_id,
)
SMS_CODE_URI = "60027"
SMS_LOGIN_URI = "60025"
SMS_CODE_URL = "https://udblgn.huya.com/web/v2/smsCode"
@@ -604,7 +603,7 @@ class HuyaSmsLogin:
phone: str = "",
proxies: Mapping[str, str] | None = None,
timeout: tuple[float, float] | None = None,
) -> "HuyaSmsLogin":
) -> HuyaSmsLogin:
"""从发码阶段返回的 state 恢复短信登录会话。"""
try:
raw = base64.urlsafe_b64decode(state.encode("ascii"))
+13 -13
View File
@@ -9,9 +9,9 @@
0x0c ZERO 0x0d SIMPLE_LIST
"""
import struct
import io
from typing import Any, Dict, List, Optional, Tuple
import struct
from typing import Any
class TafType:
@@ -136,7 +136,7 @@ class TafOutputStream:
# ---- Map ----
def write_map(
self, tag: int, value: Dict[Any, Any], key_writer=None, val_writer=None
self, tag: int, value: dict[Any, Any], key_writer=None, val_writer=None
):
self.write_head(tag, TafType.MAP)
self.write_int32(0, len(value))
@@ -151,7 +151,7 @@ class TafOutputStream:
self._write_any(1, v)
# ---- List ----
def write_list(self, tag: int, value: List[Any], item_writer=None):
def write_list(self, tag: int, value: list[Any], item_writer=None):
self.write_head(tag, TafType.LIST)
self.write_int32(0, len(value))
for item in value:
@@ -192,7 +192,7 @@ class TafInputStream:
def __init__(self, data: bytes):
self.buf = io.BytesIO(data)
def peek_head(self) -> Tuple[int, int]:
def peek_head(self) -> tuple[int, int]:
"""读取 head 但不消费(用于探测)"""
pos = self.buf.tell()
try:
@@ -200,7 +200,7 @@ class TafInputStream:
finally:
self.buf.seek(pos)
def read_head(self) -> Tuple[int, int]:
def read_head(self) -> tuple[int, int]:
"""返回 (tag, type)"""
data = self.buf.read(1)
if not data:
@@ -255,7 +255,7 @@ class TafInputStream:
def _read_int_len(self) -> int:
"""读 map/list 长度(int32 带优化)"""
tag, dtype = self.read_head()
_tag, dtype = self.read_head()
return self._read_int_value(dtype)
def _read_int_value(self, dtype: int) -> int:
@@ -273,7 +273,7 @@ class TafInputStream:
def _skip_struct(self):
while True:
tag, dtype = self.read_head()
_tag, dtype = self.read_head()
if dtype == TafType.STRUCT_END:
break
self.skip_field(dtype)
@@ -293,7 +293,7 @@ class TafInputStream:
self.skip_field(it)
# ---- 带跳过策略的字段读取:找到 tag,否则返回默认 ----
def _find_tag(self, target_tag: int, required: bool) -> Optional[Tuple[int, int]]:
def _find_tag(self, target_tag: int, required: bool) -> tuple[int, int] | None:
"""逐个读 head,tag 相等则返回,tag 超过则回退并返回 None"""
while True:
pos = self.buf.tell()
@@ -402,7 +402,7 @@ class TafInputStream:
# ---- 复合类型 ----
def read_map(
self, tag: int, required: bool = False, key_reader=None, val_reader=None
) -> Dict:
) -> dict:
found = self._find_tag(tag, required)
if not found:
return {}
@@ -418,7 +418,7 @@ class TafInputStream:
result[k] = v
return result
def read_list(self, tag: int, required: bool = False, item_reader=None) -> List:
def read_list(self, tag: int, required: bool = False, item_reader=None) -> list:
found = self._find_tag(tag, required)
if not found:
return []
@@ -441,7 +441,7 @@ class TafInputStream:
obj = struct_class()
obj.read_from(self)
# 消费 STRUCT_END
t, dt = self.read_head()
_t, dt = self.read_head()
if dt != TafType.STRUCT_END:
raise ValueError(f"期望 STRUCT_END, 实际 0x{dt:02x}")
return obj
@@ -486,7 +486,7 @@ class TafStruct:
def read_from(self, ins: TafInputStream):
raise NotImplementedError
def to_dict(self) -> Dict[str, Any]:
def to_dict(self) -> dict[str, Any]:
"""调试用:转字典"""
return {k: v for k, v in self.__dict__.items() if not k.startswith("_")}
+1 -2
View File
@@ -13,7 +13,6 @@ import onnxruntime as ort
from PIL import Image
from scipy.optimize import linear_sum_assignment
MODEL_DIR = Path(__file__).resolve().parent / "models"
@@ -262,7 +261,7 @@ class HuyaCaptchaOcr:
target["cropped_image"],
char["cropped_image"],
)
except Exception:
except Exception: # noqa: BLE001 外部接口与任务边界需要保留宽泛异常兜底
score_matrix[target_index][char_index] = 1e6
row_ind, col_ind = linear_sum_assignment(score_matrix)
+4 -5
View File
@@ -17,13 +17,12 @@ import cv2
import execjs
import numpy as np
import requests
from PIL import Image
from loguru import logger
from PIL import Image
from .ocr import HuyaCaptchaOcr, default_ocr
from .track import format_track, generate_slide_track
DEFAULT_UA = (
"Mozilla/5.0 (Windows NT 10.0; Win64; x64) "
"AppleWebKit/537.36 (KHTML, like Gecko) "
@@ -138,7 +137,7 @@ class HuyaVerificationSolver:
max_width = max(bg.shape[1], tip.shape[1])
def pad_right(img, target_width):
height, width = img.shape[:2]
_height, width = img.shape[:2]
if width >= target_width:
return img
return cv2.copyMakeBorder(
@@ -300,8 +299,8 @@ class HuyaVerificationSolver:
"虎牙登录风控strategys完整结构: {}",
json.dumps(strategies, ensure_ascii=False)[:1200],
)
except Exception: # noqa: BLE001
pass
except Exception as exc: # noqa: BLE001
logger.debug(f"风控策略日志序列化失败: {exc}")
strategy_url_lower = strategy_url.lower()
# 判定依据是 URL 路径,不是 strategy 数值。
# 实测(2026-08-25): strategy=64 时 pt_auth.html 是滑块、qr_auth.html 才是扫码,
+26 -24
View File
@@ -21,14 +21,14 @@ import re
import struct
import time
from collections import deque
from typing import Any, Optional, Callable, cast
from collections.abc import Callable
from typing import Any, cast
import websockets
from .frame_decoder import _decode_taf_struct, _truncate, format_wss_log
from .taf_protocol import TafInputStream, TafOutputStream, TafStruct, TafType
from .wup_protocol import WupRequest, WupResponse
from .taf_protocol import TafOutputStream, TafInputStream, TafType, TafStruct
from .frame_decoder import format_wss_log, _decode_taf_struct, _truncate
SHOP_WS_HOST = "77bc035c-ws.va.huya.com"
# 商城端点 baseinfo (conn4, h5_/index.html) — 商城业务 shopMiddleUI 走此通道
@@ -76,7 +76,7 @@ class WssMessage:
def decode(cls, data: bytes) -> "WssMessage":
if len(data) < 6:
raise ValueError(f"消息太短: {len(data)} bytes")
version, command = struct.unpack(">BB", data[0:2])
_version, command = struct.unpack(">BB", data[0:2])
sequence = struct.unpack(">I", data[2:6])[0]
body = data[6:]
return cls(command=command, sequence=sequence, body=body)
@@ -225,7 +225,7 @@ class HuyaWssClient:
),
timeout=timeout,
)
except asyncio.TimeoutError:
except TimeoutError:
self.logger(f"[WSS] 连接超时({timeout}s")
raise
except Exception as e:
@@ -256,7 +256,7 @@ class HuyaWssClient:
format_wss_log(msg.body, msg.command, msg.sequence, "")
)
await self._handle_message(msg)
except Exception as e:
except Exception as e: # noqa: BLE001 外部接口与任务边界需要保留宽泛异常兜底
self.logger(
f"[WSS] 解析消息失败: {e} raw_hex={raw_bytes[:50].hex()}"
)
@@ -264,7 +264,7 @@ class HuyaWssClient:
pass
except websockets.exceptions.ConnectionClosed as e:
self.logger(f"[WSS] 连接关闭: code={e.code} reason={e.reason}")
except Exception as e:
except Exception as e: # noqa: BLE001 外部接口与任务边界需要保留宽泛异常兜底
self.logger(f"[WSS] 接收循环异常: {type(e).__name__}: {e}")
async def _handle_message(self, msg: WssMessage):
@@ -289,7 +289,7 @@ class HuyaWssClient:
await self.send_heartbeat()
except asyncio.CancelledError:
pass
except Exception as e:
except Exception as e: # noqa: BLE001 外部接口与任务边界需要保留宽泛异常兜底
self.logger(f"[WSS] 心跳循环异常: {e}")
async def send_heartbeat(self):
@@ -376,7 +376,7 @@ class HuyaWssClient:
try:
body = await asyncio.wait_for(future, timeout)
except asyncio.TimeoutError:
except TimeoutError:
if future in self.rpc_queue:
self.rpc_queue.remove(future)
self.logger("[✗] wsLaunch 超时")
@@ -399,7 +399,7 @@ class HuyaWssClient:
)
return
ins = TafInputStream(treq)
tag, dtype = ins.peek_head()
_tag, dtype = ins.peek_head()
if dtype != 0x0A: # STRUCT_BEGIN
self.logger(f"[RPC] wsLaunch tRsp 非结构体 dtype=0x{dtype:02x}")
return
@@ -432,7 +432,7 @@ class HuyaWssClient:
self.logger(
f"[RPC] wsLaunch 解析: guid={self._launch_guid} ip={self._launch_ip}"
)
except Exception as e:
except Exception as e: # noqa: BLE001 外部接口与任务边界需要保留宽泛异常兜底
self.logger(f"[RPC] wsLaunch 响应解析失败: {e}")
@staticmethod
@@ -526,7 +526,7 @@ class HuyaWssClient:
try:
body = await asyncio.wait_for(future, timeout)
except asyncio.TimeoutError:
except TimeoutError:
if future in self.rpc_queue:
self.rpc_queue.remove(future)
self.logger("[✗] getConfig 超时")
@@ -596,7 +596,7 @@ class HuyaWssClient:
try:
body = await asyncio.wait_for(future, timeout)
except asyncio.TimeoutError:
except TimeoutError:
if future in self.rpc_queue:
self.rpc_queue.remove(future)
self.logger(f"[✗] {service}.{method} 超时")
@@ -610,15 +610,17 @@ class HuyaWssClient:
if data and isinstance(data, bytes) and len(data) > 0:
try:
ins = TafInputStream(data)
tag, dtype = ins.peek_head()
_tag, dtype = ins.peek_head()
if dtype == TafType.STRUCT_BEGIN:
ins.read_head()
decoded = _decode_taf_struct(ins)
self.logger(
f"[←] {service}.{method} {key}: {_truncate(decoded)}"
)
except Exception:
pass
except Exception as exc: # noqa: BLE001
self.logger(
f"[debug] WUP 响应字段解码失败: {service}.{method}.{key}: {exc}"
)
if rsp_class is None:
return body
@@ -696,11 +698,11 @@ class HuyaWssClient:
order_type: int = 6,
):
from .shop_structs import (
CreateOrderReqV5,
CreateOrderRsp,
CreateOrderAccountParam,
CreateOrderExtraParam,
CreateOrderPromotionParam,
CreateOrderAccountParam,
CreateOrderReqV5,
CreateOrderRsp,
)
req = CreateOrderReqV5()
@@ -781,7 +783,7 @@ class HuyaWssClient:
try:
body = await asyncio.wait_for(future, 15.0)
except asyncio.TimeoutError:
except TimeoutError:
if future in self.rpc_queue:
self.rpc_queue.remove(future)
self.logger("[✗] payOrderSubmitV5 超时")
@@ -794,13 +796,13 @@ class HuyaWssClient:
if data and isinstance(data, bytes) and len(data) > 0:
try:
ins = TafInputStream(data)
tag, dtype = ins.peek_head()
_tag, dtype = ins.peek_head()
if dtype == TafType.STRUCT_BEGIN:
ins.read_head()
decoded = _decode_taf_struct(ins)
self.logger(f"[←] payOrderSubmitV5 {key}: {_truncate(decoded)}")
except Exception:
pass
except Exception as exc: # noqa: BLE001
self.logger(f"[debug] WUP 支付响应字段解码失败: {key}: {exc}")
result = wup_resp.readStruct("tRsp", PayOrderRes)
if result is None:
result = wup_resp.readStruct("tResp", PayOrderRes)
+4 -4
View File
@@ -9,7 +9,7 @@ import json
import random
import struct
import time as _time
from typing import Any, Dict
from typing import Any
# TAF 类型标签
INT8, INT16, INT32, INT64 = 0x00, 0x01, 0x02, 0x03
@@ -97,7 +97,7 @@ class _Writer:
def _build_meta_json(session: int, trace_id: str) -> str:
"""构造 _wup_data.t0.t2 元数据 JSON。"""
meta: Dict[str, Any] = {
meta: dict[str, Any] = {
"associationId": 8193,
"funcName": "hypasswordLogin",
"group": 1,
@@ -167,7 +167,7 @@ def _build_wup_data(
session: int,
trace_id: str,
user_action_json: str,
device_info: Dict[str, str],
device_info: dict[str, str],
) -> None:
"""编码 _wup_data struct。"""
meta_json = _build_meta_json(session, trace_id)
@@ -235,7 +235,7 @@ def build_password_login_wup(
session: int,
trace_id: str,
user_action_json: str,
device_info: Dict[str, str],
device_info: dict[str, str],
) -> bytes:
"""构造密码登录的 WUP TAF 请求体。"""
wd = _Writer()
+13 -12
View File
@@ -11,8 +11,9 @@ Wup 包结构:
"""
import struct
from typing import Any, Dict, Optional
from .taf_protocol import TafOutputStream, TafInputStream, TafType, TafStruct
from typing import Any
from .taf_protocol import TafInputStream, TafOutputStream, TafStruct, TafType
class WupRequest:
@@ -27,9 +28,9 @@ class WupRequest:
self.sFuncName: str = "" # tag 6
self.sBuffer: bytes = b"" # tag 7
self.iTimeout: int = 3000 # tag 8
self.context: Dict[str, str] = {} # tag 9
self.status: Dict[str, str] = {} # tag 10
self.newdata: Dict[str, bytes] = {}
self.context: dict[str, str] = {} # tag 9
self.status: dict[str, str] = {} # tag 10
self.newdata: dict[str, bytes] = {}
def setServant(self, name: str):
self.sServantName = name
@@ -141,9 +142,9 @@ class WupResponse:
self.sFuncName: str = ""
self.sBuffer: bytes = b""
self.iTimeout: int = 0
self.context: Dict[str, str] = {}
self.status: Dict[str, str] = {}
self.newdata: Dict[str, bytes] = {}
self.context: dict[str, str] = {}
self.status: dict[str, str] = {}
self.newdata: dict[str, bytes] = {}
def decode(self, data: bytes):
"""解码响应(不包含长度前缀;若含前缀会自动跳过)"""
@@ -210,7 +211,7 @@ class WupResponse:
_, vt = ins.read_head()
val = _read_bytes_value(ins, vt)
self.newdata[key] = val
except Exception as e:
except Exception as e: # noqa: BLE001 外部接口与任务边界需要保留宽泛异常兜底
print(f"[WupResponse] 解析 newdata 失败: {e}")
def readStruct(self, key: str, struct_class=None):
@@ -235,13 +236,13 @@ class WupResponse:
ins = TafInputStream(data)
# newdata 里的结构体以 STRUCT_BEGIN 开头
try:
tag, dtype = ins.peek_head()
_tag, dtype = ins.peek_head()
if dtype == TafType.STRUCT_BEGIN:
ins.read_head() # 消费 STRUCT_BEGIN
obj = struct_class()
obj.read_from(ins)
return obj
except Exception as e:
except Exception as e: # noqa: BLE001 外部接口与任务边界需要保留宽泛异常兜底
print(f"[WupResponse] 解析 {struct_class.__name__} 失败: {e}")
return None
@@ -269,7 +270,7 @@ def _read_bytes_value(ins: TafInputStream, dtype: int) -> bytes:
return ins.buf.read(length)
def _read_map_value(ins: TafInputStream, dtype: int) -> Dict:
def _read_map_value(ins: TafInputStream, dtype: int) -> dict:
if dtype != TafType.MAP:
return {}
count = ins._read_int_len()
+2 -3
View File
@@ -5,12 +5,11 @@ from __future__ import annotations
import json
import re
from dataclasses import dataclass
from datetime import datetime
from datetime import UTC, datetime
from urllib.parse import urlparse
import requests
CODE_PATTERN = re.compile(
r"(?:verification\s+code|验证码|校验码|动态码|安全码)\D{0,20}(\d{4,8})",
re.IGNORECASE,
@@ -99,7 +98,7 @@ def _parse_sms8_time(value: str) -> datetime | None:
return None
for fmt in ("%Y-%m-%d %H:%M:%S", "%Y/%m/%d %H:%M:%S"):
try:
return datetime.strptime(text, fmt)
return datetime.strptime(text, fmt).replace(tzinfo=UTC)
except ValueError:
continue
return None
+4
View File
@@ -38,6 +38,10 @@ packages = ["core", "utils", "web"]
[tool.pytest.ini_options]
testpaths = ["tests"]
[tool.ruff.lint.per-file-ignores]
# FastAPI evaluates dependency metadata in route parameter defaults by design.
"web/backend/**/*.py" = ["B008"]
[dependency-groups]
dev = [
"pyright>=1.1.411",
+2 -3
View File
@@ -14,7 +14,6 @@ from sqlalchemy import JSON, MetaData, create_engine, func, inspect, select
from sqlalchemy.engine import Connection, Engine
from sqlalchemy.schema import Table
PROJECT_ROOT = Path(__file__).resolve().parents[1]
DEFAULT_SOURCE = PROJECT_ROOT / "data" / "web.db"
IGNORED_SOURCE_TABLES = {"alembic_version"}
@@ -147,7 +146,7 @@ def main() -> int:
if target_url:
os.environ["DATABASE_URL"] = target_url
from web.backend import models # noqa: F401
from web.backend.database import Base, DATABASE_URL, run_migrations
from web.backend.database import DATABASE_URL, Base, run_migrations
target_url = target_url or DATABASE_URL
if not target_url.startswith("mysql+"):
@@ -226,6 +225,6 @@ def main() -> int:
if __name__ == "__main__":
try:
sys.exit(main())
except Exception as exc:
except Exception as exc: # noqa: BLE001 外部接口与任务边界需要保留宽泛异常兜底
print(f"迁移失败:{exc}", file=sys.stderr)
sys.exit(1)
+20 -16
View File
@@ -22,6 +22,7 @@ from __future__ import annotations
import argparse
import json
import logging
import os
import re
import secrets
@@ -34,19 +35,22 @@ from pathlib import Path
ROOT = Path(__file__).resolve().parent
sys.path.insert(0, str(ROOT))
from pyvm.algorithm import build_plaintext, generate_encrypt_msg_offline # noqa: E402
from pyvm.mall import MallSession, generate_encrypt_msg as mall_generate # noqa: E402
from pyvm.session import SessionState, load_session # noqa: E402
from pyvm.login_profile import midas_login_params # noqa: E402
from pyvm.payment_errors import describe_payment_failure # noqa: E402
from pyvm.protocol import ( # noqa: E402
from pyvm.algorithm import generate_encrypt_msg_offline
from pyvm.login_profile import midas_login_params
from pyvm.mall import MallSession
from pyvm.mall import generate_encrypt_msg as mall_generate
from pyvm.payment_errors import describe_payment_failure
from pyvm.protocol import (
GOODS_USER_AGENT,
MALL_API_URL,
MALL_USER_AGENT,
GOODS_USER_AGENT,
PAY_APPID,
validate_goods_materials,
validate_mall_materials,
)
from pyvm.session import SessionState, load_session
logger = logging.getLogger(__name__)
REPLAY = ROOT / "replay"
DEFAULT_APPID = PAY_APPID
@@ -102,7 +106,8 @@ def _load_cap(path: Path) -> dict:
d = attempt()
if isinstance(d, dict) and "C" in d:
return d
except Exception:
except Exception as exc: # noqa: BLE001
logger.debug("deepCap 候选格式解析失败: %s", exc)
continue
raise ValueError(f"无法解析 deepCap 文件: {path}(前 80 字符: {raw[:80]!r})")
@@ -121,8 +126,8 @@ def parse_plaintext(path: str | Path) -> dict:
d = json.loads(raw)
if isinstance(d, dict):
return {str(k): str(v) for k, v in d.items()}
except Exception:
pass
except Exception as exc: # noqa: BLE001
logger.debug("订单响应 JSON 解析失败: %s", exc)
fields: dict[str, str] = {}
for kv in raw.split("&"):
k, _, v = kv.partition("=")
@@ -179,7 +184,6 @@ def _gen_with_session(st: SessionState, order: dict) -> str:
def cmd_gen(args) -> int:
st = load_session(args.session)
order = load_order(args.order)
params = {k: order.get(k, "") for k in ORDER_FIELDS}
hex_msg = _gen_with_session(st, order)
out = Path(args.output) if args.output else ROOT / "config" / "encrypt_msg.txt"
out.parent.mkdir(parents=True, exist_ok=True)
@@ -249,7 +253,7 @@ def cmd_submit(args) -> int:
return 0
print(f"❌ web_save ret:{ret}({js.get('err_code', '')})—— 见 case 踩坑记录")
return 1
except Exception:
except Exception: # noqa: BLE001 外部接口与任务边界需要保留宽泛异常兜底
return 1
@@ -399,8 +403,8 @@ def cmd_mall_submit(args) -> int:
js = json.loads(raw)
if isinstance(js, dict):
ret = js.get("ret", js.get("result_code", js.get("code")))
except Exception:
pass
except Exception as exc: # noqa: BLE001
logger.debug("支付响应 JSON 解析失败: %s", exc)
ok = ret in (0, "0")
if ok:
_write_private_text(out, raw)
@@ -999,8 +1003,8 @@ def cmd_mall_pay(args) -> int:
print("\n ══ 微信扫码支付(终端二维码)══")
try:
qr.terminal(compact=False)
except Exception: # noqa: BLE001
pass
except Exception as exc: # noqa: BLE001
logger.debug("终端二维码输出失败: %s", exc)
return 0
+10 -8
View File
@@ -13,22 +13,24 @@ from .algorithm import (
generate_encrypt_msg,
generate_encrypt_msg_offline,
)
from .goods import GoodsSession, generate_encrypt_msg as goods_generate
from .mall import MallSession, generate_encrypt_msg as mall_generate
from .goods import GoodsSession
from .goods import generate_encrypt_msg as goods_generate
from .mall import MallSession
from .mall import generate_encrypt_msg as mall_generate
from .pagedoo_vm import PagedooVM, run_frame
from .session import SessionState, load_session
__all__ = [
"GoodsSession",
"MallSession",
"PagedooVM",
"SessionState",
"build_plaintext",
"decode_d",
"generate_encrypt_msg",
"generate_encrypt_msg_offline",
"GoodsSession",
"goods_generate",
"MallSession",
"mall_generate",
"PagedooVM",
"run_frame",
"SessionState",
"load_session",
"mall_generate",
"run_frame",
]
+20 -25
View File
@@ -13,9 +13,11 @@
"""
from __future__ import annotations
import itertools
import json
import math
import random
import urllib.parse
from pathlib import Path
@@ -39,7 +41,7 @@ def _ic(x):
try:
f = float(x)
return 0 if math.isnan(f) else int(f)
except Exception:
except Exception: # noqa: BLE001 外部接口与任务边界需要保留宽泛异常兜底
return 0
@@ -78,7 +80,7 @@ def js_typeof(v):
return "number"
if isinstance(v, str):
return "string"
if isinstance(v, JSFunction) or isinstance(v, HostFunction):
if isinstance(v, (JSFunction, HostFunction)):
return "function"
return "object"
@@ -98,7 +100,7 @@ def js_truthy(v):
def _nan_ok(x):
try:
return not (isinstance(x, float) and math.isnan(x))
except Exception:
except Exception: # noqa: BLE001 外部接口与任务边界需要保留宽泛异常兜底
return True
@@ -206,7 +208,7 @@ def js_num(v):
return v
try:
return float(v)
except Exception:
except Exception: # noqa: BLE001 外部接口与任务边界需要保留宽泛异常兜底
return float("nan")
@@ -225,12 +227,12 @@ def js_eq(a, b):
if isinstance(a, (int, float)) and isinstance(b, str):
try:
return a == float(b)
except Exception:
except Exception: # noqa: BLE001 外部接口与任务边界需要保留宽泛异常兜底
return False
if isinstance(b, (int, float)) and isinstance(a, str):
try:
return float(a) == b
except Exception:
except Exception: # noqa: BLE001 外部接口与任务边界需要保留宽泛异常兜底
return False
return a == b
@@ -354,9 +356,7 @@ def js_set(obj, key, val):
pass
if isinstance(obj, str):
raise TypeError("Cannot assign to read only property")
cur = getattr(
getattr(__import__("pyvm.algorithm", fromlist=["x"]), "VM"), "_last_u", None
)
cur = getattr(__import__("pyvm.algorithm", fromlist=["x"]).VM, "_last_u", None)
raise TypeError(
f"invalid set target obj={type(obj).__name__} key={key!r} val={type(val).__name__} u={cur}"
)
@@ -418,7 +418,7 @@ class Window:
class JSFunction:
"""VM 函数(解释器实例工厂返回的 h)。"""
__slots__ = ("entry", "args", "s", "n", "t", "vm", "name")
__slots__ = ("args", "entry", "n", "name", "s", "t", "vm")
def __init__(self, vm, entry, args, s, n, t):
self.vm = vm
@@ -455,7 +455,7 @@ def js_call(fn, this, args):
return fn.fn(this, *args)
if callable(fn):
return fn(this, *args)
raise TypeError(f"{str(fn)} is not a function u={getattr(VM, '_last_u', None)}")
raise TypeError(f"{fn!s} is not a function u={getattr(VM, '_last_u', None)}")
def js_apply(fn, this, args):
@@ -735,7 +735,7 @@ def h_parsefloat(this, s):
def h_isnan(this, x):
try:
return math.isnan(float(x))
except Exception:
except Exception: # noqa: BLE001 外部接口与任务边界需要保留宽泛异常兜底
return True
@@ -841,7 +841,6 @@ class VM:
def run(self, fn: JSFunction, call_args):
"""执行一个 VM 函数实例(h)。fn 来自 make() 或 op87/58 创建的闭包。"""
self.inst_id += 1
iid = self.inst_id
if self.use_init_c and self.init_c is not None:
C = self.init_c
self.use_init_c = False
@@ -854,7 +853,6 @@ class VM:
d = [] # 异常续延栈
l = UNDEF # 最近异常(op12 读取)
o = self.o
cap = None # 外部捕获回调(验证用)
if not hasattr(self, "_all_trace"):
self._all_trace = []
while True:
@@ -897,7 +895,7 @@ class VM:
for x in v:
try:
parts.append("%02x" % (int(x) & 255))
except Exception:
except Exception: # noqa: BLE001 外部接口与任务边界需要保留宽泛异常兜底
parts.append("??")
self.out_hex = "".join(parts)
# 仅供离线排查 webSave 输出缓冲何时变化,生产路径默认不启用。
@@ -1528,9 +1526,7 @@ class VM:
if not (
isinstance(fn, (JSFunction, HostFunction)) or callable(fn)
):
raise RuntimeError(
"op97 non-function at u=%s fn=%r" % (u, fn)
)
raise RuntimeError(f"op97 non-function at u={u} fn={fn!r}")
js_set(C, dest, js_apply(fn, thisv, f))
elif op == 98:
a = o[u + 1]
@@ -1648,10 +1644,10 @@ class VM:
u += 1
js_set(C, a, None)
else:
raise RuntimeError("unknown opcode %s at %s" % (op, u))
raise RuntimeError(f"unknown opcode {op} at {u}")
except _VMThrow as e:
if not d:
raise RuntimeError("VM uncaught throw: %s" % (e.value,))
raise RuntimeError(f"VM uncaught throw: {e.value}")
l = e.value
u = d.pop()
continue
@@ -1659,9 +1655,8 @@ class VM:
if not d:
raise
if not isinstance(d, list):
raise RuntimeError(
"d corrupted: %r (type %s) at trace %s"
% (d, type(d).__name__, self._trace[-3:])
raise TypeError(
f"d corrupted: {d!r} (type {type(d).__name__}) at trace {self._trace[-3:]}"
)
l = e
u = d.pop()
@@ -1747,14 +1742,14 @@ def decode_d(v):
vs = p[ci + 1 :]
try:
obj.set(key, decode_d(json.loads(vs)))
except Exception:
except Exception: # noqa: BLE001 外部接口与任务边界需要保留宽泛异常兜底
obj.set(key, decode_d(vs))
return obj
if v.lstrip("-").isdigit():
return int(v)
try:
return float(v)
except Exception:
except Exception: # noqa: BLE001 外部接口与任务边界需要保留宽泛异常兜底
return v
+2 -3
View File
@@ -12,7 +12,6 @@ from __future__ import annotations
import json
from pathlib import Path
from typing import Any
from .algorithm import generate_encrypt_msg_offline
@@ -64,7 +63,7 @@ class GoodsSession:
raise ValueError("key16/key1 应为 16 字节")
@classmethod
def from_session_state(cls, path: str | Path) -> "GoodsSession":
def from_session_state(cls, path: str | Path) -> GoodsSession:
"""从 scripts/capture-session.mjs 生成的 session-state.json 加载。"""
d = json.loads(Path(path).read_text(encoding="utf-8"))
return cls(
@@ -85,7 +84,7 @@ class GoodsSession:
}
@classmethod
def from_json(cls, d: dict) -> "GoodsSession":
def from_json(cls, d: dict) -> GoodsSession:
return cls(
d["xmidas_ops"],
d["key16"],
+8 -5
View File
@@ -19,7 +19,6 @@ import copy
import json
import re
from pathlib import Path
from typing import Any
from .algorithm import UNDEF, JSObject, Window
from .pagedoo_vm import PagedooVM
@@ -73,12 +72,16 @@ class MallSession:
f"mall xMidasOps 应为 59620(非 goods 59640),实际 {len(self.xmidas_ops)}"
)
mid = self.transform_input[10]
if isinstance(mid, list) and mid and isinstance(mid[0], list):
if len(mid[0]) != 624:
if (
isinstance(mid, list)
and mid
and isinstance(mid[0], list)
and len(mid[0]) != 624
):
raise ValueError(f"624B 中间态长度 != 624: {len(mid[0])}")
@classmethod
def from_capture_file(cls, frames_jsonl: str | Path) -> "MallSession":
def from_capture_file(cls, frames_jsonl: str | Path) -> MallSession:
"""从捕获的 frames.jsonl 提取同会话 transform_input + xMidasOps。
frames.jsonl scripts/capture-mall-session.mjs 实时落盘:
@@ -127,7 +130,7 @@ class MallSession:
}
@classmethod
def from_json(cls, d: dict) -> "MallSession":
def from_json(cls, d: dict) -> MallSession:
return cls(d["transform_input"], d["xmidas_ops"])
@@ -38,13 +38,13 @@ def get_official_orders(cookies: dict[str, str], count: int = 20) -> dict[str, A
except ValueError as exc:
raise RuntimeError("订单状态查询返回非 JSON") from exc
if not isinstance(document, dict):
raise RuntimeError("订单状态查询响应格式异常")
raise TypeError("订单状态查询响应格式异常")
if document.get("ret_code") not in (None, 0, "0"):
raise RuntimeError(
f"订单状态查询失败: {document.get('ret_code')} {document.get('ret_msg', '')}"
)
if not isinstance(document.get("list", []), list):
raise RuntimeError("订单状态查询响应缺少 list")
raise TypeError("订单状态查询响应缺少 list")
return document
+46 -55
View File
@@ -14,55 +14,55 @@ from __future__ import annotations
import json
import math
import urllib.parse
from pathlib import Path
from .algorithm import (
JSDate,
UNDEF,
HostFunction,
JSDate,
JSFunction,
JSObject,
HostFunction,
Window,
_ic,
i32,
u32,
ushr,
shl,
shr,
js_typeof,
js_truthy,
js_str,
js_add,
js_num,
js_eq,
js_streq,
js_index,
js_set,
js_del,
js_keys,
js_call,
js_apply,
js_new,
h_math_floor,
h_math_round,
h_math_ceil,
h_math_min,
h_math_max,
h_math_abs,
h_math_pow,
h_math_sqrt,
h_parseint,
h_parsefloat,
h_isnan,
h_encodeuri,
h_encodeuricomponent,
h_decodeuri,
h_decodeuricomponent,
h_string_fromcharcode,
h_encodeuri,
h_encodeuricomponent,
h_isnan,
h_math_abs,
h_math_ceil,
h_math_floor,
h_math_max,
h_math_min,
h_math_pow,
h_math_round,
h_math_sqrt,
h_new_date,
h_parsefloat,
h_parseint,
h_string_fromcharcode,
i32,
js_add,
js_apply,
js_call,
js_del,
js_eq,
js_index,
js_keys,
js_new,
js_num,
js_set,
js_str,
js_truthy,
js_typeof,
shl,
shr,
ushr,
)
__all__ = ["PagedooVM", "run_frame", "REPLAY"]
REPLAY = Path(__file__).resolve().parent.parent / "replay"
__all__ = ["REPLAY", "PagedooVM", "run_frame"]
# ---------------------------------------------------------------- 辅助
@@ -71,28 +71,28 @@ __all__ = ["PagedooVM", "run_frame", "REPLAY"]
def _cmp_lt(a, b):
try:
return a < b
except Exception:
except Exception: # noqa: BLE001 外部接口与任务边界需要保留宽泛异常兜底
return False
def _cmp_le(a, b):
try:
return a <= b
except Exception:
except Exception: # noqa: BLE001 外部接口与任务边界需要保留宽泛异常兜底
return False
def _cmp_gt(a, b):
try:
return a > b
except Exception:
except Exception: # noqa: BLE001 外部接口与任务边界需要保留宽泛异常兜底
return False
def _cmp_ge(a, b):
try:
return a >= b
except Exception:
except Exception: # noqa: BLE001 外部接口与任务边界需要保留宽泛异常兜底
return False
@@ -127,8 +127,8 @@ def h_arr_slice(this, a=None, b=None):
a = 0
if b is None or b is UNDEF:
b = n
a = int(a) if a == a else 0
b = int(b) if b == b else n
a = int(a) if not isinstance(a, float) or not math.isnan(a) else 0
b = int(b) if not isinstance(b, float) or not math.isnan(b) else n
if a < 0:
a = max(0, n + a)
if b < 0:
@@ -308,8 +308,7 @@ def _pg_index(obj, key):
return HostFunction(
lambda this: this.t if isinstance(this, JSDate) else obj.t, "valueOf"
)
if isinstance(obj, str):
if isinstance(key, str):
if isinstance(obj, str) and isinstance(key, str):
if key == "length":
return len(obj)
if key == "charCodeAt":
@@ -491,15 +490,7 @@ class PagedooVM:
and len(self._host_log) < 2000
):
# 记录调用目标(简化)
_tgt = (
o[u + 2]
if op in (4, 11, 44, 50)
else (
o[u + 2]
if op in (0, 18, 23, 26, 43, 48, 84, 107)
else o[u + 2]
)
)
_tgt = o[u + 2]
try:
_tv = C[_tgt] if 0 <= _tgt < len(C) else UNDEF
_tr = (
@@ -510,7 +501,7 @@ class PagedooVM:
else repr(_tv)[:30]
)
self._host_log.append((op, u, _tr))
except Exception:
except Exception: # noqa: BLE001 外部接口与任务边界需要保留宽泛异常兜底
self._host_log.append((op, u, "?"))
if (
getattr(self, "_trace", None) is not None
+2 -3
View File
@@ -10,7 +10,6 @@ import json
from pathlib import Path
from typing import Any
PAY_APPID = "1450243039"
PAY_GOODS_URL = "https://pay.qq.com/midas/minipay_v2/views/cpay/goods.shtml"
PAY_SAVE_URL = f"https://api.unipay.qq.com/v1/r/{PAY_APPID}/web_save"
@@ -87,7 +86,7 @@ def validate_goods_materials(
def validate_mall_materials(transform_fixed: dict[str, Any]) -> None:
"""校验 mall 固定槽模板;动态槽仍由当前会话的 GetPayToken 填充。"""
if not isinstance(transform_fixed, dict):
raise ValueError("mall transform-fixed 必须是对象")
raise TypeError("mall transform-fixed 必须是对象")
required = {
str(index) for index in (1, 2, 3, 4, 5, 7, 8, 9, 11, 12, 13, 14, 15, 16, 17)
}
@@ -96,7 +95,7 @@ def validate_mall_materials(transform_fixed: dict[str, Any]) -> None:
raise ValueError(f"mall transform-fixed 缺少槽: {', '.join(missing)}")
for index in required:
if not isinstance(transform_fixed[index], list):
raise ValueError(f"mall transform-fixed 槽 {index} 不是数组")
raise TypeError(f"mall transform-fixed 槽 {index} 不是数组")
def goods_material_diagnostics(
+1 -1
View File
@@ -68,7 +68,7 @@ class SessionState:
}
@classmethod
def from_json(cls, d: dict) -> "SessionState":
def from_json(cls, d: dict) -> SessionState:
return cls(
xmidas_ops=list(d.get("xmidas_ops", [])),
key16=list(d.get("key16", [])),
+5 -6
View File
@@ -31,10 +31,10 @@ from pyvm.algorithm import (
build_plaintext,
derive_key1_from_key16,
generate_encrypt_msg_offline,
) # noqa: E402
from pyvm.login_profile import midas_login_params # noqa: E402
from pyvm.payment_errors import describe_payment_failure # noqa: E402
from pyvm.protocol import ( # noqa: E402
)
from pyvm.login_profile import midas_login_params
from pyvm.payment_errors import describe_payment_failure
from pyvm.protocol import (
DEFAULT_ORDER_PF,
GOODS_USER_AGENT,
PAY_APPID,
@@ -450,7 +450,6 @@ def cmd_check_only(session_path: Path, out_dir: Path) -> int:
from pyvm.order_status import (
completion_summary,
get_official_orders,
is_finished,
)
session = load_json(session_path)
@@ -697,7 +696,7 @@ def main() -> int:
out_dir, fields, int(fp.get("device_fp_length", 0)), plaintext_length
)
body = urllib.parse.urlencode(fields)
from pyvm.order_status import order_completion_states, get_official_orders
from pyvm.order_status import get_official_orders, order_completion_states
# 尽量在发起支付前建立基线,缩小极快支付造成的检测窗口;失败不阻塞支付,付款码生成后重试。
baseline_document = None
+3 -3
View File
@@ -17,8 +17,8 @@ from pathlib import Path
from urllib.error import HTTPError
from urllib.parse import urlencode
from urllib.request import (
HTTPRedirectHandler,
HTTPCookieProcessor,
HTTPRedirectHandler,
Request,
build_opener,
)
@@ -43,7 +43,7 @@ USER_AGENT = (
class NoRedirect(HTTPRedirectHandler):
"""Keep OAuth redirects visible while CookieJar receives Set-Cookie headers."""
def redirect_request(self, req, fp, code, msg, headers, newurl): # noqa: D401
def redirect_request(self, req, fp, code, msg, headers, newurl):
return None
@@ -118,7 +118,7 @@ def g_tk(p_skey: str) -> int:
def parse_poll(body: str) -> tuple[int, str]:
match = re.search(r"ptuiCB\((.*)\)", body, re.S)
match = re.search(r"ptuiCB\((.*)\)", body, re.DOTALL)
if not match:
raise ValueError("QQ 二维码轮询响应缺少 ptuiCB")
values = re.findall(r"'([^']*)'", match.group(1))
+3 -3
View File
@@ -20,8 +20,8 @@ from pathlib import Path
from urllib.error import HTTPError
from urllib.parse import urlencode
from urllib.request import (
HTTPRedirectHandler,
HTTPCookieProcessor,
HTTPRedirectHandler,
Request,
build_opener,
)
@@ -42,7 +42,7 @@ USER_AGENT = (
class NoRedirect(HTTPRedirectHandler):
"""保留 OAuth 回调的 302 和 Set-Cookie。"""
def redirect_request(self, req, fp, code, msg, headers, newurl): # noqa: D401
def redirect_request(self, req, fp, code, msg, headers, newurl):
return None
@@ -88,7 +88,7 @@ def extract_uuid(page: str) -> str:
r"connect/qrcode/([A-Za-z0-9_-]{8,64})",
r"uuid=([A-Za-z0-9_-]{8,64})",
):
match = re.search(pattern, page, re.I)
match = re.search(pattern, page, re.IGNORECASE)
if match:
return match.group(1)
raise ValueError("微信授权页未找到二维码 UUID")
+1 -1
View File
@@ -18,7 +18,7 @@ from curl_cffi import requests
ROOT = Path(__file__).resolve().parent.parent
sys.path.insert(0, str(ROOT))
from pyvm.login_profile import midas_login_params # noqa: E402
from pyvm.login_profile import midas_login_params
YYB_APP_ID = 52575843
SOURCE_ID = "24292013"
+9 -5
View File
@@ -32,7 +32,7 @@ WORKER_KEY = os.environ.get("YYB_WORKER_KEY", "")
_jobs: dict[str, dict] = {}
_lock = threading.Lock()
from pyvm.payment_errors import describe_payment_failure # noqa: E402
from pyvm.payment_errors import describe_payment_failure
def _payment_failure_message(job: dict, phase: str, fallback: str) -> str:
@@ -81,13 +81,16 @@ def _safe_log(job: dict, line: str) -> None:
clean = re.sub(r"HOLD_[A-Za-z0-9_-]+", "[订单已隐藏]", clean)
clean = re.sub(r"(订单\s*[:]\s*)\S+", r"\1[订单已隐藏]", clean)
clean = re.sub(
r"(?:token|openid|openkey|cookie)=\S+", "[敏感字段已隐藏]", clean, flags=re.I
r"(?:token|openid|openkey|cookie)=\S+",
"[敏感字段已隐藏]",
clean,
flags=re.IGNORECASE,
)
clean = re.sub(
r"(?:pay_token|web_token|anti_token|session_id|sessionid)\s*[:= ]\s*\S+",
"[敏感字段已隐藏]",
clean,
flags=re.I,
flags=re.IGNORECASE,
)
timestamp = time.strftime("%H:%M:%S")
with _lock:
@@ -282,6 +285,7 @@ def _check_payment_once(job_id: str) -> int:
text=True,
env=environment,
timeout=90,
check=False,
)
except subprocess.TimeoutExpired:
_safe_log(
@@ -523,7 +527,7 @@ class Handler(BaseHTTPRequestHandler):
value = self.headers.get("Authorization", "")
return value == f"Bearer {WORKER_KEY}"
def do_POST(self) -> None: # noqa: N802
def do_POST(self) -> None:
if not self._authorized():
return self._json(401, {"detail": "未授权"})
path = urlparse(self.path).path.strip("/").split("/")
@@ -687,7 +691,7 @@ class Handler(BaseHTTPRequestHandler):
except Exception as exc: # noqa: BLE001
return self._json(500, {"detail": str(exc)})
def do_GET(self) -> None: # noqa: N802
def do_GET(self) -> None:
if urlparse(self.path).path == "/health":
return self._json(200, {"status": "ok"})
if not self._authorized():
@@ -8,7 +8,6 @@ import unittest
from pathlib import Path
from urllib.parse import parse_qs
_SCRIPT = Path(__file__).resolve().parents[1] / "scripts" / "jsdom-pay.py"
_SPEC = importlib.util.spec_from_file_location("yyb_jsdom_pay", _SCRIPT)
assert _SPEC and _SPEC.loader
@@ -16,8 +15,8 @@ _MODULE = importlib.util.module_from_spec(_SPEC)
sys.modules[_SPEC.name] = _MODULE
_SPEC.loader.exec_module(_MODULE)
from pyvm.payment_errors import describe_payment_failure
from pyvm.algorithm import derive_key1_from_key16, generate_encrypt_msg_offline
from pyvm.payment_errors import describe_payment_failure
from pyvm.protocol import goods_material_diagnostics, validate_goods_materials
+1 -2
View File
@@ -1,8 +1,7 @@
import pytest
from fastapi import HTTPException
from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker
from fastapi import HTTPException
from web.backend.database import Base
from web.backend.models import Account, User
+1 -2
View File
@@ -1,9 +1,8 @@
"""充值审计日志的权限、查询和脱敏测试。"""
import pytest
import json
import pytest
from fastapi import HTTPException
from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker
+2 -2
View File
@@ -1,4 +1,4 @@
from datetime import datetime, timedelta, timezone
from datetime import UTC, datetime, timedelta
from sqlalchemy import create_engine
from sqlalchemy.orm import Session
@@ -17,7 +17,7 @@ class TestCustomCookieOrder:
self.db.add(self.user)
self.db.flush()
now = datetime.now(timezone.utc)
now = datetime.now(UTC)
accounts = {
name: Account(username=name, password="p", email="e", email_password="ep")
for name in ("account-a", "account-b", "account-c")
+7 -7
View File
@@ -1,18 +1,13 @@
import pytest
from types import SimpleNamespace
from unittest.mock import patch
from typing import cast
from unittest.mock import patch
import pytest
from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker
from web.backend.database import Base
from web.backend.models import Account, LoginTask, ProxyConfig, User
from web.backend.services.login_service import (
LoginBatchRunner,
cleanup_orphan_relogin_tasks,
)
from web.backend.services.cookie_check_service import check_douyu_cookie
from web.backend.routers.cookies import (
check_cookie_operations,
get_cookie,
@@ -21,6 +16,11 @@ from web.backend.routers.cookies import (
list_cookies,
relogin_invalid_cookie_operations,
)
from web.backend.services.cookie_check_service import check_douyu_cookie
from web.backend.services.login_service import (
LoginBatchRunner,
cleanup_orphan_relogin_tasks,
)
class TestCookieOperation:
+2 -2
View File
@@ -1,9 +1,9 @@
"""精英手册兑换:csrf 复用、新链路接口与浏览器状态机路由的测试。"""
import pytest
from unittest.mock import Mock, call
import pytest
from core.douyu.activity_client import DouyuActivityClient, DouyuActivityError
+1 -1
View File
@@ -1,6 +1,6 @@
import pytest
from unittest.mock import Mock
import pytest
import requests
from core.douyu.activity_client import DouyuActivityClient, DouyuActivityError
+2 -2
View File
@@ -7,13 +7,13 @@ from unittest.mock import ANY, AsyncMock, Mock, patch
from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker
from core.douyu import FishFinRechargeClient, FishFinRechargeConfig
from web.backend.database import Base
from web.backend.models import Account, AuditLog, DouyuTask, LoginTask, User
from web.backend.routers.douyu import get_recharge_channel, supplier_recharge_callback
from web.backend.services.audit_service import record_audit
from web.backend.services.douyu_runner import DouyuBatchRunner
from web.backend.services.douyu_service import ensure_douyu_config
from web.backend.services.audit_service import record_audit
from core.douyu import FishFinRechargeClient, FishFinRechargeConfig
class TestDouyuGoldRechargeChannel:
+1 -1
View File
@@ -1,6 +1,6 @@
from types import SimpleNamespace
from unittest.mock import Mock
from typing import Any, cast
from unittest.mock import Mock
from core.douyu.login import DouyuLogin
from core.douyu.login_api_wgapi import WgapiLoginAPI
+2 -2
View File
@@ -3,12 +3,12 @@
from unittest.mock import Mock
from core.douyu.activity_client import DouyuActivityClient
from web.backend.models import ProxyConfig
from web.backend.services.douyu_runner import DouyuBatchRunner
from web.backend.services.douyu_runner_core import (
DouyuBatchRunnerCore,
DOUYU_PROXY_TASK_TYPES,
DouyuBatchRunnerCore,
)
from web.backend.models import ProxyConfig
def _cfg(**overrides) -> ProxyConfig:
-1
View File
@@ -1,5 +1,4 @@
import pytest
from sqlalchemy import create_engine, event
from sqlalchemy.orm import sessionmaker
+2 -1
View File
@@ -1,9 +1,10 @@
import pytest
import json
import re
from types import SimpleNamespace
from unittest.mock import Mock
import pytest
from core.douyu.activity_client import DouyuActivityClient, DouyuActivityError
+2 -2
View File
@@ -1,9 +1,9 @@
"""鱼翅直充供应商 API 客户端测试。"""
import pytest
from unittest.mock import Mock
import pytest
from core.douyu.recharge_api import (
FishFinRechargeClient,
FishFinRechargeConfig,
+12 -17
View File
@@ -1,21 +1,15 @@
"""虎牙 App 密码登录及相关组件测试。"""
import pytest
import base64
import os
import struct
from unittest.mock import patch, MagicMock
from unittest.mock import MagicMock, patch
import pytest
from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker
from core.huya import (
HuyaAppLoginError,
HuyaAppPasswordLogin,
HuyaAppQrAuthRequiredError,
login_huya_app_password,
login_huya_password,
)
from core.huya.app_login import (
DEFAULT_GOLDEN_DEV,
@@ -30,17 +24,16 @@ from core.huya.login import HuyaLoginResult
from core.huya.nonce_forge import K1_DEFAULT, gen_nonce
from core.huya.udb_aes import udb_decrypt, udb_encrypt
from core.huya.wup_encoder import build_password_login_wup
from web.backend.database import Base
from web.backend.models import User, HuyaAccount
from web.backend.schemas import (
HuyaAppPasswordLoginRequest,
HuyaPasswordLoginSelectedRequest,
)
from web.backend.models import HuyaAccount, User
from web.backend.routers.huya import (
app_password_login_account,
app_password_login_selected_accounts,
)
from web.backend.schemas import (
HuyaAppPasswordLoginRequest,
HuyaPasswordLoginSelectedRequest,
)
class TestHuyaAppLogin:
@@ -182,19 +175,21 @@ class TestHuyaAppLogin:
side_effect=DfpRegistrationError("注册链超时"),
) as m_reg,
patch("core.huya.app_login.requests.post") as m_post,
pytest.raises(HuyaAppLoginError),
):
with pytest.raises(HuyaAppLoginError):
wup_password_login_raw("300023887", "pw")
m_reg.assert_called_once()
m_post.assert_not_called()
def test_login_cred_flow_registration_failure_is_explicit(self):
"""login_cred_with_flow 注册失败同样包装为 HuyaAppLoginError 显式失败。"""
with patch(
with (
patch(
"core.huya.app_login.register_device",
side_effect=DfpRegistrationError("注册链 HTTP 500"),
),
pytest.raises(HuyaAppLoginError, match="注册失败"),
):
with pytest.raises(HuyaAppLoginError, match="注册失败"):
login_cred_with_flow("300023887", "pw")
def test_router_functions(self):
+5 -4
View File
@@ -3,13 +3,13 @@
import gzip
import logging
import tempfile
from datetime import datetime, timedelta
from datetime import UTC, datetime, timedelta
from pathlib import Path
from utils.logger import (
_parse_size,
_SensitiveDataFilter,
_SizeAndDayRotatingFileHandler,
_parse_size,
)
@@ -40,7 +40,8 @@ class TestLogger:
archives = list(Path(tmpdir).glob("app-2026-08-28.log.*.gz"))
assert len(archives) == 1
archived_content = gzip.open(archives[0], "rt", encoding="utf-8").read()
with gzip.open(archives[0], "rt", encoding="utf-8") as archive:
archived_content = archive.read()
current_content = log_path.read_text(encoding="utf-8")
combined = archived_content + current_content
assert "super-secret" not in combined
@@ -50,7 +51,7 @@ class TestLogger:
def test_daily_log_switches_to_a_new_dated_file(self):
with tempfile.TemporaryDirectory() as tmpdir:
today = datetime.now().date()
today = datetime.now(UTC).date()
old_path = (
Path(tmpdir) / f"app-{(today - timedelta(days=1)).isoformat()}.log"
)
+1 -1
View File
@@ -68,8 +68,8 @@ class TestMigrationSmoke:
def test_full_chain_applies_on_fresh_sqlite_and_matches_models(self):
"""空库执行整条迁移链,校验表/列/索引与模型元数据对齐。"""
from web.backend.database import Base
import web.backend.database as database_module
from web.backend.database import Base
original_url = database_module.DATABASE_URL
with tempfile.TemporaryDirectory() as tmp:
+13 -12
View File
@@ -8,13 +8,12 @@ import os
import re
import shutil
import sys
from datetime import datetime, timedelta
from datetime import UTC, datetime, timedelta
from logging.handlers import BaseRotatingHandler
from pathlib import Path
from loguru import logger
_LOG_LEVELS = {"DEBUG", "INFO", "WARNING", "ERROR", "CRITICAL"}
_DEFAULT_RETENTION_DAYS = 14
_DEFAULT_ROTATION_SIZE = 50 * 1024 * 1024
@@ -36,7 +35,9 @@ def _parse_positive_int(value: str | None, default: int) -> int:
def _parse_size(value: str | None, default: int = _DEFAULT_ROTATION_SIZE) -> int:
"""解析 50M、1GiB 等易读大小;非法值保持安全默认值。"""
matched = re.fullmatch(r"\s*(\d+)\s*([kmgt]?i?b?)?\s*", str(value or ""), re.I)
matched = re.fullmatch(
r"\s*(\d+)\s*([kmgt]?i?b?)?\s*", str(value or ""), re.IGNORECASE
)
if not matched:
return default
amount = int(matched.group(1))
@@ -102,7 +103,7 @@ class _SizeAndDayRotatingFileHandler(BaseRotatingHandler):
super().__init__(str(filename), "a", encoding="utf-8", delay=True)
self.max_bytes = max_bytes
self.retention_days = retention_days
self._active_day = datetime.now().date()
self._active_day = datetime.now(UTC).date()
path = Path(filename)
self._log_dir = path.parent
self._suffix = path.suffix
@@ -113,8 +114,8 @@ class _SizeAndDayRotatingFileHandler(BaseRotatingHandler):
self._log_dir / f"{self._filename_prefix}-{day.isoformat()}{self._suffix}"
)
def shouldRollover(self, record: logging.LogRecord) -> bool: # noqa: N802
if datetime.now().date() != self._active_day:
def shouldRollover(self, record: logging.LogRecord) -> bool:
if datetime.now(UTC).date() != self._active_day:
return True
if self.stream is None:
self.stream = self._open()
@@ -122,12 +123,12 @@ class _SizeAndDayRotatingFileHandler(BaseRotatingHandler):
message = f"{self.format(record)}\n"
return self.stream.tell() + len(message.encode("utf-8")) >= self.max_bytes
def doRollover(self) -> None: # noqa: N802
def doRollover(self) -> None:
if self.stream is not None:
self.stream.close()
self.stream = None
source = Path(self.baseFilename)
current_day = datetime.now().date()
current_day = datetime.now(UTC).date()
if current_day != self._active_day:
# 每日文件本身已带日期,跨日时直接切换到新文件,无需再移动旧文件。
self.baseFilename = os.fspath(self._path_for_day(current_day).resolve())
@@ -135,7 +136,7 @@ class _SizeAndDayRotatingFileHandler(BaseRotatingHandler):
self._delete_expired_archives()
return
if source.exists() and source.stat().st_size:
stamp = datetime.now().strftime("%Y-%m-%d_%H-%M-%S")
stamp = datetime.now(UTC).strftime("%Y-%m-%d_%H-%M-%S")
archive = source.with_name(f"{source.name}.{stamp}.{os.getpid()}.gz")
sequence = 1
while archive.exists():
@@ -149,10 +150,10 @@ class _SizeAndDayRotatingFileHandler(BaseRotatingHandler):
self._delete_expired_archives()
def _delete_expired_archives(self) -> None:
cutoff = datetime.now() - timedelta(days=self.retention_days)
cutoff = datetime.now(UTC) - timedelta(days=self.retention_days)
for archive in self._log_dir.glob(f"{self._filename_prefix}-*.log*.gz"):
try:
if datetime.fromtimestamp(archive.stat().st_mtime) < cutoff:
if datetime.fromtimestamp(archive.stat().st_mtime, UTC) < cutoff:
archive.unlink()
except OSError:
continue
@@ -227,7 +228,7 @@ def setup_logger(
if log_file:
file_path = Path(log_file).expanduser()
elif log_dir:
file_path = Path(log_dir).expanduser() / f"app-{datetime.now():%Y-%m-%d}.log"
file_path = Path(log_dir).expanduser() / f"app-{datetime.now(UTC):%Y-%m-%d}.log"
else:
return
file_path.parent.mkdir(parents=True, exist_ok=True)
+3 -4
View File
@@ -10,11 +10,10 @@ from dataclasses import dataclass
from functools import lru_cache
from Crypto.Cipher import AES
from loguru import logger
from sqlalchemy import text
from sqlalchemy.engine import Engine
from sqlalchemy.types import Text, TypeDecorator
from loguru import logger
_PREFIX = "enc:v1:"
_FALLBACK_SECRET = "douyu-login-py-dev-storage-key-change-me"
@@ -110,7 +109,7 @@ def decrypt_value(value: str | None) -> str | None:
for candidate in _key_candidates():
try:
return _decrypt_with_candidate(value, candidate)
except Exception as exc:
except Exception as exc: # noqa: BLE001 外部接口与任务边界需要保留宽泛异常兜底
last_error = exc
raise ValueError(
"敏感字段解密失败,请确认 APP_ENCRYPTION_KEY 是否正确"
@@ -125,7 +124,7 @@ def decrypt_value_with_key_name(value: str) -> tuple[str, str]:
for candidate in _key_candidates():
try:
return _decrypt_with_candidate(value, candidate), candidate.name
except Exception as exc:
except Exception as exc: # noqa: BLE001 外部接口与任务边界需要保留宽泛异常兜底
last_error = exc
raise ValueError(
"敏感字段解密失败,请确认 APP_ENCRYPTION_KEY 是否正确"
+2 -1
View File
@@ -5,7 +5,7 @@ from pathlib import Path
from urllib.parse import quote_plus
from sqlalchemy import create_engine, event
from sqlalchemy.orm import sessionmaker, declarative_base
from sqlalchemy.orm import declarative_base, sessionmaker
PROJECT_ROOT = Path(__file__).resolve().parents[2]
@@ -147,6 +147,7 @@ def _seed():
def _encrypt_existing_sensitive_data():
"""启动时把历史明文敏感数据迁移为密文。"""
from loguru import logger
from .crypto_storage import encrypt_existing_sensitive_data
changed = encrypt_existing_sensitive_data(engine)
+6 -7
View File
@@ -1,15 +1,14 @@
"""FastAPI 依赖注入"""
from typing import Optional
from fastapi import Depends, HTTPException, Request, status, WebSocket
from fastapi import Depends, HTTPException, Request, WebSocket, status
from fastapi.security import OAuth2PasswordBearer
from sqlalchemy.orm import Session
from jose import JWTError
from sqlalchemy.orm import Session
from .database import get_db, SessionLocal
from .security import decode_access_token
from .database import SessionLocal, get_db
from .models import User
from .permissions import get_user_permissions
from .security import decode_access_token
# auto_error=False: 允许 token 为空(后续从 cookie 读取)
oauth2_scheme = OAuth2PasswordBearer(tokenUrl="/api/auth/login", auto_error=False)
@@ -17,7 +16,7 @@ oauth2_scheme = OAuth2PasswordBearer(tokenUrl="/api/auth/login", auto_error=Fals
def get_current_user(
request: Request,
token: Optional[str] = Depends(oauth2_scheme),
token: str | None = Depends(oauth2_scheme),
db: Session = Depends(get_db),
) -> User:
credentials_exc = HTTPException(
@@ -61,7 +60,7 @@ def require_permission(permission: str):
return checker
def authenticate_websocket(websocket: WebSocket) -> Optional[User]:
def authenticate_websocket(websocket: WebSocket) -> User | None:
"""WebSocket 认证:从 cookie 或 query param token 中验证用户身份。
Returns:
+15 -13
View File
@@ -1,34 +1,36 @@
"""FastAPI 入口"""
import os
from pathlib import Path
import uvicorn
from contextlib import asynccontextmanager
from pathlib import Path
import uvicorn
from fastapi import FastAPI, Request
from fastapi.middleware.cors import CORSMiddleware
from fastapi.middleware.gzip import GZipMiddleware
from fastapi.staticfiles import StaticFiles
from fastapi.responses import FileResponse
from fastapi.staticfiles import StaticFiles
from starlette.middleware.base import BaseHTTPMiddleware
from utils import setup_logger
from .database import init_db
from .routers import (
auth,
users,
accounts,
account_check,
accounts,
audit,
auth,
cookies,
dashboard,
douyu,
huya,
login,
proxy,
cookies,
huya,
douyu,
users,
yyb,
audit,
)
from .schemas import AppInfo
from .version import get_app_version
from utils import setup_logger
@asynccontextmanager
@@ -45,9 +47,9 @@ async def lifespan(app: FastAPI):
from loguru import logger
from .database import SessionLocal
from .services.login_service import cleanup_orphan_relogin_tasks
from .services.huya_service import cleanup_orphan_huya_tasks
from .services.douyu_service import cleanup_orphan_douyu_tasks
from .services.huya_service import cleanup_orphan_huya_tasks
from .services.login_service import cleanup_orphan_relogin_tasks
db = SessionLocal()
try:
+4 -4
View File
@@ -1,9 +1,9 @@
"""Alembic 迁移环境。"""
from logging.config import fileConfig
from pathlib import Path
import os
import sys
from logging.config import fileConfig
from pathlib import Path
from alembic import context
from sqlalchemy import engine_from_config, pool
@@ -12,8 +12,8 @@ ROOT_DIR = Path(__file__).resolve().parents[3]
if str(ROOT_DIR) not in sys.path:
sys.path.insert(0, str(ROOT_DIR))
from web.backend.database import Base, DATABASE_URL # noqa: E402
from web.backend import models # noqa: F401,E402
from web.backend import models # noqa: F401
from web.backend.database import DATABASE_URL, Base
config = context.config
config.set_main_option("sqlalchemy.url", DATABASE_URL)
@@ -5,15 +5,15 @@ Revises:
Create Date: 2026-06-23
"""
from typing import Sequence, Union
from collections.abc import Sequence
from alembic import op
import sqlalchemy as sa
from alembic import op
revision: str = "20260623_0001"
down_revision: Union[str, None] = None
branch_labels: Union[str, Sequence[str], None] = None
depends_on: Union[str, Sequence[str], None] = None
down_revision: str | None = None
branch_labels: str | Sequence[str] | None = None
depends_on: str | Sequence[str] | None = None
def _has_table(bind, table_name: str) -> bool:
@@ -5,16 +5,15 @@ Revises: 20260623_0001
Create Date: 2026-06-24
"""
from typing import Sequence, Union
from collections.abc import Sequence
from alembic import op
import sqlalchemy as sa
from alembic import op
revision: str = "20260624_0002"
down_revision: Union[str, None] = "20260623_0001"
branch_labels: Union[str, Sequence[str], None] = None
depends_on: Union[str, Sequence[str], None] = None
down_revision: str | None = "20260623_0001"
branch_labels: str | Sequence[str] | None = None
depends_on: str | Sequence[str] | None = None
def upgrade() -> None:
@@ -5,16 +5,15 @@ Revises: 20260624_0002
Create Date: 2026-06-24
"""
from typing import Sequence, Union
from collections.abc import Sequence
from alembic import op
import sqlalchemy as sa
from alembic import op
revision: str = "20260624_0003"
down_revision: Union[str, None] = "20260624_0002"
branch_labels: Union[str, Sequence[str], None] = None
depends_on: Union[str, Sequence[str], None] = None
down_revision: str | None = "20260624_0002"
branch_labels: str | Sequence[str] | None = None
depends_on: str | Sequence[str] | None = None
def upgrade() -> None:
@@ -5,16 +5,15 @@ Revises: 20260624_0003
Create Date: 2026-07-04
"""
from typing import Sequence, Union
from collections.abc import Sequence
from alembic import op
import sqlalchemy as sa
from alembic import op
revision: str = "20260704_0004"
down_revision: Union[str, None] = "20260624_0003"
branch_labels: Union[str, Sequence[str], None] = None
depends_on: Union[str, Sequence[str], None] = None
down_revision: str | None = "20260624_0003"
branch_labels: str | Sequence[str] | None = None
depends_on: str | Sequence[str] | None = None
def _has_table(bind, table_name: str) -> bool:
@@ -5,16 +5,15 @@ Revises: 20260704_0004
Create Date: 2026-07-04
"""
from typing import Sequence, Union
from collections.abc import Sequence
from alembic import op
import sqlalchemy as sa
from alembic import op
revision: str = "20260704_0005"
down_revision: Union[str, None] = "20260704_0004"
branch_labels: Union[str, Sequence[str], None] = None
depends_on: Union[str, Sequence[str], None] = None
down_revision: str | None = "20260704_0004"
branch_labels: str | Sequence[str] | None = None
depends_on: str | Sequence[str] | None = None
def _has_table(bind, table_name: str) -> bool:
@@ -5,16 +5,15 @@ Revises: 20260704_0005
Create Date: 2026-07-05
"""
from typing import Sequence, Union
from collections.abc import Sequence
from alembic import op
import sqlalchemy as sa
from alembic import op
revision: str = "20260705_0006"
down_revision: Union[str, None] = "20260704_0005"
branch_labels: Union[str, Sequence[str], None] = None
depends_on: Union[str, Sequence[str], None] = None
down_revision: str | None = "20260704_0005"
branch_labels: str | Sequence[str] | None = None
depends_on: str | Sequence[str] | None = None
def _has_table(bind, table_name: str) -> bool:
@@ -5,16 +5,15 @@ Revises: 20260705_0006
Create Date: 2026-07-12
"""
from typing import Sequence, Union
from collections.abc import Sequence
from alembic import op
import sqlalchemy as sa
from alembic import op
revision: str = "20260712_0007"
down_revision: Union[str, None] = "20260705_0006"
branch_labels: Union[str, Sequence[str], None] = None
depends_on: Union[str, Sequence[str], None] = None
down_revision: str | None = "20260705_0006"
branch_labels: str | Sequence[str] | None = None
depends_on: str | Sequence[str] | None = None
def _has_table(bind, table_name: str) -> bool:
@@ -5,16 +5,15 @@ Revises: 20260712_0007
Create Date: 2026-07-24
"""
from typing import Sequence, Union
from collections.abc import Sequence
from alembic import op
import sqlalchemy as sa
from alembic import op
revision: str = "20260724_0008"
down_revision: Union[str, None] = "20260712_0007"
branch_labels: Union[str, Sequence[str], None] = None
depends_on: Union[str, Sequence[str], None] = None
down_revision: str | None = "20260712_0007"
branch_labels: str | Sequence[str] | None = None
depends_on: str | Sequence[str] | None = None
def _has_table(bind, table_name: str) -> bool:
@@ -5,16 +5,15 @@ Revises: 20260724_0008
Create Date: 2026-07-25
"""
from typing import Sequence, Union
from collections.abc import Sequence
from alembic import op
import sqlalchemy as sa
from alembic import op
revision: str = "20260725_0009"
down_revision: Union[str, None] = "20260724_0008"
branch_labels: Union[str, Sequence[str], None] = None
depends_on: Union[str, Sequence[str], None] = None
down_revision: str | None = "20260724_0008"
branch_labels: str | Sequence[str] | None = None
depends_on: str | Sequence[str] | None = None
def _has_table(bind, table_name: str) -> bool:
@@ -5,16 +5,15 @@ Revises: 20260725_0009
Create Date: 2026-07-28
"""
from typing import Sequence, Union
from collections.abc import Sequence
from alembic import op
import sqlalchemy as sa
from alembic import op
revision: str = "20260728_0010"
down_revision: Union[str, None] = "20260725_0009"
branch_labels: Union[str, Sequence[str], None] = None
depends_on: Union[str, Sequence[str], None] = None
down_revision: str | None = "20260725_0009"
branch_labels: str | Sequence[str] | None = None
depends_on: str | Sequence[str] | None = None
def _columns(bind, table_name: str) -> set[str]:
@@ -5,16 +5,15 @@ Revises: 20260728_0010
Create Date: 2026-07-28
"""
from typing import Sequence, Union
from collections.abc import Sequence
from alembic import op
import sqlalchemy as sa
from alembic import op
revision: str = "20260728_0011"
down_revision: Union[str, None] = "20260728_0010"
branch_labels: Union[str, Sequence[str], None] = None
depends_on: Union[str, Sequence[str], None] = None
down_revision: str | None = "20260728_0010"
branch_labels: str | Sequence[str] | None = None
depends_on: str | Sequence[str] | None = None
def _columns(bind, table_name: str) -> set[str]:
@@ -5,16 +5,15 @@ Revises: 20260728_0011
Create Date: 2026-07-28
"""
from typing import Sequence, Union
from collections.abc import Sequence
from alembic import op
import sqlalchemy as sa
from alembic import op
revision: str = "20260728_0012"
down_revision: Union[str, None] = "20260728_0011"
branch_labels: Union[str, Sequence[str], None] = None
depends_on: Union[str, Sequence[str], None] = None
down_revision: str | None = "20260728_0011"
branch_labels: str | Sequence[str] | None = None
depends_on: str | Sequence[str] | None = None
def upgrade() -> None:
@@ -5,16 +5,15 @@ Revises: 20260728_0012
Create Date: 2026-07-28
"""
from typing import Sequence, Union
from collections.abc import Sequence
from alembic import op
import sqlalchemy as sa
from alembic import op
revision: str = "20260728_0013"
down_revision: Union[str, None] = "20260728_0012"
branch_labels: Union[str, Sequence[str], None] = None
depends_on: Union[str, Sequence[str], None] = None
down_revision: str | None = "20260728_0012"
branch_labels: str | Sequence[str] | None = None
depends_on: str | Sequence[str] | None = None
def _columns(bind, table_name: str) -> set[str]:
@@ -5,16 +5,15 @@ Revises: 20260728_0013
Create Date: 2026-07-28
"""
from typing import Sequence, Union
from collections.abc import Sequence
from alembic import op
import sqlalchemy as sa
from alembic import op
revision: str = "20260728_0014"
down_revision: Union[str, None] = "20260728_0013"
branch_labels: Union[str, Sequence[str], None] = None
depends_on: Union[str, Sequence[str], None] = None
down_revision: str | None = "20260728_0013"
branch_labels: str | Sequence[str] | None = None
depends_on: str | Sequence[str] | None = None
def upgrade() -> None:
@@ -5,16 +5,15 @@ Revises: 20260728_0014
Create Date: 2026-08-05
"""
from typing import Sequence, Union
from collections.abc import Sequence
from alembic import op
import sqlalchemy as sa
from alembic import op
revision: str = "20260805_0015"
down_revision: Union[str, None] = "20260728_0014"
branch_labels: Union[str, Sequence[str], None] = None
depends_on: Union[str, Sequence[str], None] = None
down_revision: str | None = "20260728_0014"
branch_labels: str | Sequence[str] | None = None
depends_on: str | Sequence[str] | None = None
INDEXES = [
@@ -5,16 +5,15 @@ Revises: 20260805_0015
Create Date: 2026-08-06
"""
from typing import Sequence, Union
from collections.abc import Sequence
from alembic import op
import sqlalchemy as sa
from alembic import op
revision: str = "20260806_0016"
down_revision: Union[str, None] = "20260805_0015"
branch_labels: Union[str, Sequence[str], None] = None
depends_on: Union[str, Sequence[str], None] = None
down_revision: str | None = "20260805_0015"
branch_labels: str | Sequence[str] | None = None
depends_on: str | Sequence[str] | None = None
def _columns(bind, table_name: str) -> set[str]:

Some files were not shown because too many files have changed in this diff Show More