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

feat(huya): 支持四段式账号导入接码及旋转/文本点选多级验证码闭环过验

Browse Source 
- 账号导入：支持'账号----密码----手机号----接码链接'四段式解析与持久化存储
- 旋转验证码 (rotate_auth / xz)：逆向分析官方协议，基于圆周边界采样互相关算法秒级计算旋转还原角度
- 文本点选验证码 (dx_auth)：内置成语与古诗名句语料，支持题干菜名提取与古诗/成语挖空正则求解及小图动态合成
- 短信二次验证 (sms_auth2)：结合 sms_url 自动轮询接码链接并 Hex 编码提交官方过验
- 前端交互：增加接码标签、在线查询短信验证码、一键自动短信登录与测试接码链接功能
- 回归测试：新增旋转验证、短信接码流与文本点选测试用例（173 passed）
This commit is contained in:
yml2213
2026-09-06 13:56:44 +08:00
parent 4f9a6864eb
commit d3005478a9
15 changed files with 31411 additions and 24 deletions
Download Patch File Download Diff File Expand all files Collapse all files
web/backend/routers/huya.py
+136 -1
View File
@@ -87,6 +87,7 @@ from ..schemas import (
HuyaSmsLoginRequest,
HuyaTaskBatchRequest,
HuyaTaskOut,
HuyaTestSmsUrlRequest,
HuyaWorkbenchAccountsUpdate,
)
from ..services.audit_service import record_audit
@@ -384,6 +385,7 @@ def _account_out(account: HuyaAccount, include_cookie: bool = True) -> HuyaAccou
game_name=account.game_name or "",
game_channel=account.game_channel or "",
game_phone=account.game_phone or "",
sms_url=account.sms_url or "",
assigned_to=account.assigned_to,
assigned_username=account.assigned_user.username
if account.assigned_user
@@ -415,6 +417,7 @@ def _account_out_light(
game_name=account.game_name or "",
game_channel=account.game_channel or "",
game_phone=account.game_phone or "",
sms_url=account.sms_url or "",
assigned_to=account.assigned_to,
assigned_username=account.assigned_user.username
if account.assigned_user
@@ -756,11 +759,17 @@ def app_password_login_account(
):
"""使用 App 协议密码登录虎牙（推荐），自动过滑块并保存 Cookie。"""
_require_huya_perm(current, "huya:import")
username_clean = req.username.strip()
account_db = (
db.query(HuyaAccount).filter(HuyaAccount.username == username_clean).first()
)
sms_url = account_db.sms_url if account_db and account_db.sms_url else ""
try:
result = login_huya_app_password(
username=req.username.strip(),
username=username_clean,
password=req.password,
force_new_device=req.force_new_device,
sms_url=sms_url,
)
except HuyaAppQrAuthRequiredError as exc:
raise HTTPException(status_code=400, detail=str(exc)) from exc
@@ -883,6 +892,130 @@ def sms_login_account(
}
@router.post("/accounts/test-sms-url")
def test_sms_url_endpoint(
req: HuyaTestSmsUrlRequest,
current: User = Depends(get_current_user),
):
"""测试从指定接码链接获取验证码。"""
_require_huya_perm(current, "huya:import")
url = req.url.strip()
if not url:
raise HTTPException(status_code=400, detail="接码链接不能为空")
from core.sms_provider import SmsLine, SmsProviderClient, detect_provider
client = SmsProviderClient(timeout=(8, 15))
item = SmsLine(
phone=req.phone.strip(),
url=url,
provider=detect_provider(url),
raw=f"{req.phone}----{url}",
)
poll_result = client.poll(item)
return {
"success": poll_result.has_code,
"code": poll_result.code,
"status": poll_result.status,
"message": poll_result.message
or ("成功识别验证码" if poll_result.has_code else "未识别到验证码"),
"raw": poll_result.raw[:500],
}
@router.post("/accounts/{account_id}/fetch-sms-code")
def fetch_account_sms_code(
account_id: int,
db: Session = Depends(get_db),
current: User = Depends(get_current_user),
):
"""通过该账号绑定的接码链接获取最新短信验证码。"""
_require_huya_perm(current, "huya:import")
account = (
_visible_huya_accounts_query(db, current)
.filter(HuyaAccount.id == account_id)
.first()
)
if account is None:
raise HTTPException(status_code=404, detail="虎牙账号不存在或无权操作")
if not account.sms_url:
raise HTTPException(status_code=400, detail="该账号未配置接码链接 (sms_url)")
from core.sms_provider import SmsLine, SmsProviderClient, detect_provider
client = SmsProviderClient(timeout=(8, 15))
item = SmsLine(
phone=account.game_phone or "",
url=account.sms_url,
provider=detect_provider(account.sms_url),
raw=f"{account.game_phone}----{account.sms_url}",
)
poll_result = client.poll(item)
return {
"account_id": account.id,
"username": account.username or "",
"phone": account.game_phone or "",
"sms_url": account.sms_url,
"success": poll_result.has_code,
"code": poll_result.code,
"status": poll_result.status,
"message": poll_result.message
or ("成功识别验证码" if poll_result.has_code else "未识别到验证码"),
"raw": poll_result.raw[:500],
}
@router.post("/accounts/{account_id}/auto-sms-login")
def auto_sms_login_account(
account_id: int,
db: Session = Depends(get_db),
current: User = Depends(get_current_user),
):
"""使用该账号绑定的手机号与接码链接一键执行虎牙短信登录。"""
_require_huya_perm(current, "huya:import")
account = (
_visible_huya_accounts_query(db, current)
.filter(HuyaAccount.id == account_id)
.first()
)
if account is None:
raise HTTPException(status_code=404, detail="虎牙账号不存在或无权操作")
phone = (account.game_phone or "").strip()
sms_url = (account.sms_url or "").strip()
if not phone or not sms_url:
raise HTTPException(status_code=400, detail="该账号缺少手机号或接码链接")
from core.huya.auto_register import auto_register_huya_phone
from core.sms_provider import SmsLine, detect_provider
item = SmsLine(
phone=phone,
url=sms_url,
provider=detect_provider(sms_url),
raw=f"{phone}----{sms_url}",
)
res = auto_register_huya_phone(item, wait_seconds=90, poll_interval=2)
if not res.success or not res.cookie:
account.status = "login_failed"
account.updated_at = datetime.now(UTC)
db.commit()
raise HTTPException(
status_code=502, detail=res.message or "虎牙短信接码登录失败"
)
saved = save_huya_login_cookie_to_account(
db,
account,
res.cookie,
tag=account.tag or "",
username_hint=account.username or "",
)
saved.status = "active"
saved.login_channel = "sms"
db.commit()
return {
"message": "短信接码登录成功，Cookie 已保存",
"success": True,
"account": _account_out(saved),
}
@router.post("/register/batches", response_model=HuyaAutoRegisterBatchOut)
def create_auto_register_batch(
req: HuyaAutoRegisterRequest,
@@ -1266,11 +1399,13 @@ def app_password_login_selected_accounts(
)
continue
sms_url = (account.sms_url or "").strip()
try:
result = login_huya_app_password(
username=username,
password=password,
force_new_device=req.force_new_device,
sms_url=sms_url,
)
if not result.success or not result.cookie:
account.status = "login_failed"
web/backend/schemas.py
+10
View File
@@ -305,6 +305,13 @@ class HuyaSmsLoginRequest(BaseModel):
tag: str = ""
class HuyaTestSmsUrlRequest(BaseModel):
"""在线测试从接码链接提取验证码。"""
url: str = Field(..., min_length=1)
phone: str = ""
class HuyaAutoRegisterRequest(BaseModel):
"""虎牙自动注册批次。每行格式：手机号----短信查询URL。"""
@@ -441,6 +448,7 @@ class HuyaAccountOut(BaseModel):
game_name: str = ""
game_channel: str = ""
game_phone: str = ""
sms_url: str = ""
assigned_to: int | None = None
assigned_username: str | None = None
created_at: datetime | None = None
@@ -463,10 +471,12 @@ class HuyaAccountOut(BaseModel):
"tag": self.tag,
"remark": self.remark,
"status": self.status,
"login_channel": self.login_channel,
"points": self.points,
"game_name": self.game_name,
"game_channel": self.game_channel,
"game_phone": self.game_phone,
"sms_url": self.sms_url,
"assigned_to": self.assigned_to,
"assigned_username": self.assigned_username,
"created_at": _isoformat(self.created_at),
web/backend/services/huya_service.py
+3
View File
@@ -210,6 +210,7 @@ def import_huya_password_accounts(
account_password=password,
cookie=normalize_huya_cookie(cookie) if cookie else "",
game_phone=parsed.phone,
sms_url=parsed.sms_url,
tag=tag,
status="password_imported",
)
@@ -220,6 +221,8 @@ def import_huya_password_accounts(
account.cookie = normalize_huya_cookie(cookie)
if parsed.phone:
account.game_phone = parsed.phone
if parsed.sms_url:
account.sms_url = parsed.sms_url
if tag:
account.tag = tag
if not account.cookie: