From dd3508d61408c8586340821952d00c70f0dd681d Mon Sep 17 00:00:00 2001 From: yml2213 Date: Sun, 8 Nov 2026 10:29:33 +0800 Subject: [PATCH] =?UTF-8?q?fix(weibo):=20=E5=B0=86=E6=94=AF=E4=BB=98?= =?UTF-8?q?=E9=A3=8E=E6=8E=A7=E6=A3=80=E6=B5=8B=E5=B9=B6=E5=85=A5=E8=B4=A6?= =?UTF-8?q?=E5=8F=B7=20CK=20=E6=A0=A1=E9=AA=8C?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- core/weibo/client.py | 103 +++++++++++++++- tests/test_weibo_accounts.py | 39 ++++++ tests/test_weibo_client.py | 163 ++++++++++++++++++++++---- web/backend/services/weibo_service.py | 8 +- 4 files changed, 282 insertions(+), 31 deletions(-) diff --git a/core/weibo/client.py b/core/weibo/client.py index b6fafc1..890b439 100644 --- a/core/weibo/client.py +++ b/core/weibo/client.py @@ -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( ( diff --git a/tests/test_weibo_accounts.py b/tests/test_weibo_accounts.py index ebf4d57..15f244f 100644 --- a/tests/test_weibo_accounts.py +++ b/tests/test_weibo_accounts.py @@ -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( diff --git a/tests/test_weibo_client.py b/tests/test_weibo_client.py index 5d7a397..0436f1c 100644 --- a/tests/test_weibo_client.py +++ b/tests/test_weibo_client.py @@ -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, diff --git a/web/backend/services/weibo_service.py b/web/backend/services/weibo_service.py index 6e96def..88515ac 100644 --- a/web/backend/services/weibo_service.py +++ b/web/backend/services/weibo_service.py @@ -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)