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

fix(weibo): 将支付风控检测并入账号 CK 校验

Browse Source
This commit is contained in:
yml2213
2026-11-08 10:29:33 +08:00
parent ffbe0f41f5
commit dd3508d614
4 changed files with 282 additions and 31 deletions
Download Patch File Download Diff File Expand all files Collapse all files
core/weibo/client.py
+99 -4
View File
@@ -606,8 +606,10 @@ class WeiboClient:
def _load_recharge_context(
self,
activity: str,
) -> tuple[dict[str, str], dict[str, Any], dict[str, Any]]:
home = self.fetch_home(activity)
*,
home: WeiboHomeSnapshot | None = None,
) -> tuple[str, dict[str, str], dict[str, Any], dict[str, Any]]:
home = home or self.fetch_home(activity)
if not home.recharge_url:
raise WeiboRechargeError("当前活动没有可用的助威充值入口")
subscribe = _subscribe_params(home.recharge_url)
@@ -679,7 +681,98 @@ class WeiboClient:
base=WEIBO_VPLUS_API_BASE,
headers=page_headers,
)
return subscribe, current_tab, cashier_payload.get("data") or {}
return home.uid, subscribe, current_tab, cashier_payload.get("data") or {}
def check_payment_access(
self,
activity: str = "",
*,
home: WeiboHomeSnapshot | None = None,
) -> None:
"""账号检测时确认 CK 也能进入微博支付域。"""
uid, _, _, cashier_data = self._load_recharge_context(
activity,
home=home,
)
tabs = cashier_data.get("tab_info") or []
tab = next(
(
item
for item in tabs
if isinstance(item, dict) and item.get("is_current")
),
tabs[0] if tabs and isinstance(tabs[0], dict) else {},
)
menu = next(
(
item
for item in tab.get("menu") or []
if isinstance(item, dict) and item.get("weibo_unified_params")
),
None,
)
if menu is None:
raise WeiboRechargeError("收银台没有返回支付检测参数")
self._preflight_recharge(uid, menu)
def _preflight_recharge(
self,
uid: str,
menu: dict[str, Any],
) -> None:
"""确认当前 CK 能通过微博支付域；此接口不会创建订单。"""
unified = menu.get("weibo_unified_params") or {}
response = self.session.post(
f"{WEIBO_PAY_BASE}/aj/pay/modular/prepare",
data={
"uid": uid,
"seller_id": str(unified.get("seller_id") or ""),
"modular_id": str(unified.get("modular_id") or ""),
"default_cashier": str(unified.get("default_cashier") or ""),
"total_amount": str(menu.get("price") or 6000),
"hidden_channel": "0",
"extra": json.dumps(
{"goods_type": 2},
separators=(",", ":"),
),
"gsid": "",
"lang": "zh_cn",
"is4g": "0",
"from": "",
"aid": "",
"wm": "",
},
headers={
"Content-Type": "application/x-www-form-urlencoded;charset=UTF-8",
"User-Agent": WEIBO_DESKTOP_UA,
"Origin": "https://vipclub.weibo.com",
"Referer": "https://vipclub.weibo.com/",
},
allow_redirects=False,
timeout=self.timeout,
)
location = response.headers.get("location", "")
location_url = urlsplit(location) if location else None
if response.status_code in {301, 302, 303, 307, 308}:
if location_url and location_url.path == "/sso/login.php":
raise WeiboRechargeError(
"微博支付未登录：当前 CK 能访问微豆页面，"
"但不能用于支付，请重新登录微博后重新导入 CK"
)
if location_url and location_url.path == "/sorry":
query = parse_qs(location_url.query, keep_blank_values=True)
if "usernotexists" in query:
raise WeiboRechargeError(
"微博支付账号不存在或未开通支付资格，"
"请先在微博 App/网页确认该账号可以正常发起支付"
)
raise WeiboRechargeError(
"微博支付预检被拒绝，请重新登录微博后重新导入 CK"
)
if response.status_code >= 400:
raise WeiboRechargeError(f"微博支付预检失败: HTTP {response.status_code}")
def _recharge_bid(
self,
@@ -718,7 +811,9 @@ class WeiboClient:
if amount_yuan < 1:
raise WeiboRechargeError("充值金额至少为 1 元")
subscribe, current_tab, cashier_data = self._load_recharge_context(activity)
_uid, subscribe, current_tab, cashier_data = self._load_recharge_context(
activity
)
tabs = cashier_data.get("tab_info") or []
tab = next(
(
tests/test_weibo_accounts.py
+39
View File
@@ -53,6 +53,7 @@ class StubWeiboClient:
bind_url=BIND_URL,
)
scheme_errors: ClassVar[dict[str, str]] = {}
payment_error: ClassVar[WeiboRechargeError | None] = None
calls: ClassVar[list[tuple[str, str]]] = []
def __init__(self, cookie: str = "", **_kwargs):
@@ -64,6 +65,12 @@ class StubWeiboClient:
raise WeiboAuthError("未登陆")
return self.snapshot
def check_payment_access(self, activity: str = "", *, home=None):
_ = home
self.calls.append(("payment", activity))
if self.payment_error is not None:
raise self.payment_error
def generate_bind_scheme(self, bind_url: str, account_type: str):
self.calls.append(("scheme", account_type))
error = self.scheme_errors.get(account_type)
@@ -89,6 +96,7 @@ def current_user(db):
def stub_client(monkeypatch):
StubWeiboClient.calls = []
StubWeiboClient.scheme_errors = {}
StubWeiboClient.payment_error = None
StubWeiboClient.snapshot = WeiboHomeSnapshot(
uid="4014000001",
nickname="测试微博",
@@ -147,6 +155,37 @@ def test_import_reports_invalid_ck_per_line_and_keeps_valid_account(db, current_
assert db.query(WeiboAccount).count() == 1
def test_import_reports_payment_access_failure(db, current_user):
StubWeiboClient.payment_error = WeiboRechargeError("触发微博账号风控")
result = import_cookies(
WeiboCookieImport(text="SCF=a; SUB=b", tag=""),
db=db,
current=current_user,
)
assert result.count == 0
assert len(result.errors) == 1
assert "触发微博账号风控" in result.errors[0].message
assert db.query(WeiboAccount).count() == 0
def test_refresh_marks_payment_access_failure(db, current_user):
account_out = import_cookies(
WeiboCookieImport(text="SCF=a; SUB=b", tag=""),
db=db,
current=current_user,
).accounts[0]
account = db.query(WeiboAccount).filter(WeiboAccount.id == account_out.id).one()
StubWeiboClient.payment_error = WeiboRechargeError("触发微博账号风控")
with pytest.raises(WeiboRechargeError, match="触发微博账号风控"):
weibo_service.refresh_weibo_account(db, account)
assert account.ck_check_status == "error"
assert "触发微博账号风控" in account.ck_check_message
def test_list_accounts_returns_parallel_page_shape(db, current_user):
assert (
import_cookies(
tests/test_weibo_client.py
+138 -25
View File
@@ -2,6 +2,8 @@
from __future__ import annotations
from typing import Any
import pytest
import requests
@@ -12,6 +14,7 @@ from core.weibo import (
WeiboClient,
WeiboExchangeError,
WeiboExchangeUncertainError,
WeiboHomeSnapshot,
WeiboRechargeError,
extract_bind_raw_url,
normalize_weibo_cookie,
@@ -27,13 +30,22 @@ BIND_URL = (
class FakeResponse:
def __init__(self, payload, url: str = ""):
def __init__(
self,
payload,
url: str = "",
status_code: int = 200,
headers: dict[str, str] | None = None,
):
self.payload = payload
self.text = payload if isinstance(payload, str) else ""
self.url = url
self.status_code = status_code
self.headers = {key.lower(): value for key, value in (headers or {}).items()}
def raise_for_status(self):
return None
if self.status_code >= 400:
raise requests.HTTPError(f"HTTP {self.status_code}")
def json(self):
if not isinstance(self.payload, dict):
@@ -336,33 +348,53 @@ def test_create_recharge_order_full_manual_chain():
assert create_call[2]["data"]["single_reward"] == "100"
assert '"bid":"1000294191"' in create_call[2]["data"]["ext"]
assert create_call[2]["params"] == create_call[2]["data"]
assert not any("/aj/pay/modular/prepare" in call[1] for call in session.calls)
def test_create_recharge_order_reports_payment_rejection():
client = WeiboClient("SCF=a; SUB=b", session=FakeSession([]))
client._load_recharge_context = lambda _activity: (
{"vuid": "vuid", "source_itemid": "20601", "F": "bowen_tjg"},
{"tab_type": "8", "cashier_id": "205"},
{
"tab_info": [
{
"is_current": 1,
"tab_type": "8",
"cashier_id": "205",
"menu": [
{
"subitemid": "2060101",
"itemid": 20601,
"appkey": 2000,
"commodity_id": "2060101",
"max_money": 1000,
}
],
}
]
},
)
client._recharge_bid = lambda *_args: "1000294191"
def fake_context(
activity: str,
*,
home: WeiboHomeSnapshot | None = None,
) -> tuple[str, dict[str, str], dict[str, Any], dict[str, Any]]:
_ = (activity, home)
return (
"4014000001",
{"vuid": "vuid", "source_itemid": "20601", "F": "bowen_tjg"},
{"tab_type": "8", "cashier_id": "205"},
{
"tab_info": [
{
"is_current": 1,
"tab_type": "8",
"cashier_id": "205",
"menu": [
{
"subitemid": "2060101",
"itemid": 20601,
"appkey": 2000,
"commodity_id": "2060101",
"max_money": 1000,
}
],
}
]
},
)
def fake_bid(
vuid: str,
source_itemid: str,
tab_type: str,
activity: str,
) -> str:
_ = (vuid, source_itemid, tab_type, activity)
return "1000294191"
client._load_recharge_context = fake_context
client._recharge_bid = fake_bid
client._request_json = lambda *_args, **_kwargs: {
"code": 100000,
"data": {
@@ -381,6 +413,87 @@ def test_create_recharge_order_reports_payment_rejection():
client.create_recharge_order(amount_yuan=1, activity="bowen_tjg")
@pytest.mark.parametrize(
("location", "message"),
[
(
"https://pay.sc.weibo.com/sso/login.php?url=test",
"微博支付未登录",
),
(
"https://weibo.com/sorry?usernotexists",
"支付账号不存在",
),
],
)
def test_recharge_preflight_detects_payment_auth_redirect(location, message):
home = WeiboHomeSnapshot(
uid="4014000001",
nickname="测试微博",
is_bound=False,
role_name="",
can_rebind=True,
rebind_remain_days=0,
weidou_balance=100,
bind_url=BIND_URL,
recharge_url=(
"https://vipclub.weibo.com/vmember/subscribe"
"?vuid=8455793145&source_itemid=20601&F=bowen_tjg"
),
)
session = FakeSession(
[
{"code": 100000, "data": {}},
{
"code": 100000,
"data": {
"cashier_id": 205,
"tab_info": [
{
"cashier_id": "205",
"tab_id": "536",
"tab_type": "8",
"is_current": 1,
}
],
},
},
{
"code": 100000,
"data": {
"tab_info": [
{
"cashier_id": "205",
"tab_id": "536",
"tab_type": "8",
"is_current": 1,
"menu": [
{
"price": 6000,
"weibo_unified_params": {
"seller_id": 6329048632,
"modular_id": "7d5d81dfb1",
"default_cashier": "ali_cashier",
},
}
],
}
]
},
},
FakeResponse(
"",
status_code=302,
headers={"Location": location},
),
]
)
client = WeiboClient("SCF=a; SUB=b", session=session)
with pytest.raises(WeiboRechargeError, match=message):
client.check_payment_access("bowen_tjg", home=home)
def test_create_recharge_order_rejects_missing_dynamic_menu():
home = {
"code": 100000,
web/backend/services/weibo_service.py
+6 -2
View File
@@ -80,7 +80,9 @@ def import_weibo_cookies(
continue
try:
cookie = normalize_weibo_cookie(raw)
snapshot = WeiboClient(cookie).fetch_home(activity)
client = WeiboClient(cookie)
snapshot = client.fetch_home(activity)
client.check_payment_access(activity, home=snapshot)
except (ValueError, WeiboError) as exc:
errors.append(WeiboImportIssue(line=line_no, message=str(exc)))
continue
@@ -117,7 +119,9 @@ def refresh_weibo_account(
"""刷新一个微博账号；失败时保留旧资料并记录本次检查状态。"""
try:
snapshot = WeiboClient(account.cookie or "").fetch_home(activity)
client = WeiboClient(account.cookie or "")
snapshot = client.fetch_home(activity)
client.check_payment_access(activity, home=snapshot)
except (WeiboError, ValueError) as exc:
account.ck_check_status = _error_status(exc)
account.ck_check_message = str(exc)