wangyunlong/gyxx-flow
1
0 Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

fix(product): 天猫百补报名登录链路三类阻断兜底

Browse Source 
vault 假登录态（SPA 缓存店铺名误判已登录）与 SSO 中间态停留时改走显式
账密登录兜底；登录页识别服务协议/隐私弹窗并先处理，二次验证标记补充
滑块；报名导入增加账号级锁防并发挤兑。
This commit is contained in:
wangyunlong committed 2026-10-14 18:33:32 +08:00
1 parent 0cc4fafc7c
commit 6cd563eab0
2 files changed
+666 -28
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

src/gyxx_flow/modules/product_commerce/tmall_baibu_apply_common.py
+252 -28
View File
@@ -29,15 +29,19 @@ from runtime_paths import TMP_ROOT
from gyxx_flow.adapters.browser import BrowserCookieStore
from gyxx_flow.adapters.scrapling import ScraplingBrowser, is_browser_timeout_error
from gyxx_flow.core.config import Settings
from gyxx_flow.core.locks import LockManager, ResourceBusyError
_CONFIG_NAME = "tmall-baibu.json"
_LOGIN_PAGE_URL = "https://login.taobao.com/member/login.jhtml"
_EXPORT_TIMEOUT_SECONDS = 240
_EXPORT_MAX_ATTEMPTS = 3
_EXPORT_RETRY_DELAYS_SECONDS = (5, 15)
_PAGE_TIMEOUT_MS = 60_000
_IMPORT_TIMEOUT_SECONDS = 300
_IMPORT_ACCEPTED_AFTER_PROCESSING_SECONDS = 120
_STATUS_SETTLE_TIMEOUT_SECONDS = 5
_ACCOUNT_LOCK_TIMEOUT_SECONDS = 120
_RESULT_REFRESH_INTERVAL_SECONDS = 8
_DRAFT_ACTION_TIMEOUT_SECONDS = 30
_DRAFT_NAVIGATION_TIMEOUT_SECONDS = 15
@@ -103,6 +107,7 @@ _LOGIN_SECOND_FACTOR_MARKERS = (
"安全验证",
"二次验证",
"验证码",
"滑块",
)
@@ -805,6 +810,41 @@ def _find_password_login_form(
return locate()
def _is_login_agreement_visible(page: Any) -> bool:
"""Return whether Tmall's 服务协议及隐私保护 popup is blocking the login."""
for context in _page_contexts(page):
get_by_role = getattr(context, "get_by_role", None)
if not callable(get_by_role):
continue
try:
dialogs = get_by_role("dialog")
for index in range(min(dialogs.count(), 8)):
dialog = dialogs.nth(index)
try:
if not dialog.is_visible():
continue
text = _normalized_page_text(dialog.inner_text(timeout=2_000))
except Exception:
continue
if "服务协议" in text or "隐私" in text:
return True
except Exception:
continue
return False
def _click_login_agreement_accept(page: Any) -> bool:
for context in _page_contexts(page):
for candidate in _visible_text_candidates(context, "同意"):
try:
candidate.click(timeout=8_000)
return True
except Exception:
continue
return False
def _login_if_needed(
entry: BaibuEntry,
page: Any,
@@ -817,6 +857,11 @@ def _login_if_needed(
return False
login_url = _is_login_url(str(getattr(page, "url", "")))
form = _find_password_login_form(page, use_password_form=False)
if form is None and not login_url:
# A redirect can land on a scanner-tab seller login page (qianniu)
# whose URL carries no login marker; probe the password tab before
# deciding there is nothing to log in to.
form = _find_password_login_form(page, use_password_form=True)
if not login_url and form is None:
return False
@@ -852,10 +897,92 @@ def _login_if_needed(
raise BaibuApplyError(
f"{entry.name} 自动填写账号密码并提交登录失败（{type(exc).__name__}）"
) from exc
# The first password submit can pop the 服务协议及隐私保护 confirmation
# which swallows the submit; accept it and submit once more so the login
# actually runs. Both popups and form re-renders are tolerated.
for _attempt in range(2):
wait_for_timeout = getattr(page, "wait_for_timeout", None)
if callable(wait_for_timeout):
wait_for_timeout(2_000)
else:
time.sleep(2)
if not _is_login_agreement_visible(page):
break
if not _click_login_agreement_accept(page):
break
print(f"[百补] {entry.name} 已确认登录服务协议弹窗，重新提交登录")
refreshed = _find_password_login_form(page, use_password_form=True)
if refreshed is None:
break
_context, username, secret_input, submit = refreshed
try:
username.fill(credentials.username)
secret_input.fill(credentials.password)
submit.click(timeout=15_000)
except Exception as exc:
raise BaibuApplyError(
f"{entry.name} 协议确认后重新提交登录失败（{type(exc).__name__}）"
) from exc
print(f"[百补] {entry.name} 已提交密码登录，等待店铺身份确认")
return True
def _goto_tolerant(page: Any, url: str) -> None:
"""Navigate, tolerating an aborted load when a redirect takes over.
An unauthenticated apply URL is answered with a redirect chain that can
abort the original navigation from inside the browser; treat that as the
navigation landing wherever the chain pointed and let it settle.
"""
try:
page.goto(url, wait_until="domcontentloaded", timeout=_PAGE_TIMEOUT_MS)
except Exception as exc:
if "ERR_ABORTED" not in str(exc):
raise
wait_for_timeout = getattr(page, "wait_for_timeout", None)
if callable(wait_for_timeout):
wait_for_timeout(_LOGIN_FORM_SETTLE_MILLISECONDS)
else:
time.sleep(_LOGIN_FORM_SETTLE_MILLISECONDS / 1000)
def _explicit_password_login(
entry: BaibuEntry,
page: Any,
*,
manual_login: bool,
) -> None:
"""Drive the Tmall password form when the apply page hid its login state.
The myseller/qn apply SPA can sit on an SSO interstitial that never shows
a login URL, or render the cached shop name on an expired session. In
both cases ``_login_if_needed`` sees nothing to do. Re-entering the apply
URL redirects to the seller login page (qianniu) which carries the right
merchant context; only when that shows no form fall back to the plain
login page. Login there, return to the apply URL, and confirm identity.
"""
print(f"[百补] {entry.name} 显式重新进入报名页并走密码登录")
_goto_tolerant(page, entry.apply_url)
wait_for_timeout = getattr(page, "wait_for_timeout", None)
if callable(wait_for_timeout):
wait_for_timeout(_LOGIN_FORM_SETTLE_MILLISECONDS * 3)
else:
time.sleep(_LOGIN_FORM_SETTLE_MILLISECONDS / 1000 * 3)
login_attempted = _login_if_needed(entry, page, manual_login=manual_login)
if not login_attempted:
_goto_tolerant(page, _LOGIN_PAGE_URL)
login_attempted = _login_if_needed(entry, page, manual_login=manual_login)
_goto_tolerant(page, entry.apply_url)
_wait_for_authenticated(
entry,
page,
login_attempted=login_attempted,
manual_login=manual_login,
)
def _click_text(page: Any, labels: Iterable[str]) -> bool:
contexts = _page_contexts(page)
for label in labels:
@@ -1040,7 +1167,10 @@ def _open_batch_import(
f"{entry.name} 页面已有批量导入操作正在处理，已停止，避免重复提交"
)
time.sleep(0.5)
raise BaibuApplyError(f"{entry.name} 页面没有找到可用的「商品批量导入」入口")
raise BaibuApplyError(
f"{entry.name} 页面没有找到可用的「商品批量导入」入口"
"（活动可能已报名/已结束或会话无效，请人工打开报名页确认）"
)
def _wait_import_result(
@@ -1062,6 +1192,7 @@ def _wait_import_result(
last_processing_log = 0.0
last_refresh = 0.0
observed_new_operation = False
first_new_operation_at: float | None = None
while time.monotonic() < deadline:
text = _page_text(page)
operation = _latest_batch_import_status(text)
@@ -1070,6 +1201,8 @@ def _wait_import_result(
or operation.signature != previous_operation.signature
):
observed_new_operation = True
if first_new_operation_at is None:
first_new_operation_at = time.monotonic()
if operation.state == "success":
print(
f"[百补] {entry.name} 页面已完成导入处理，"
@@ -1082,6 +1215,21 @@ def _wait_import_result(
f"忽略行结果并按导入提交成功：{operation.detail}"
)
return
if (
observed_new_operation
and first_new_operation_at is not None
and time.monotonic() - first_new_operation_at
>= _IMPORT_ACCEPTED_AFTER_PROCESSING_SECONDS
):
# Tmall has already acknowledged the batch ("执行了商品批量导入
# 操作，正在处理中"); the submission itself is the business
# effect, so a slow per-row summary must not fail the run.
print(
f"[百补] {entry.name} 天猫已受理批量导入，"
f"{_IMPORT_ACCEPTED_AFTER_PROCESSING_SECONDS} 秒内未返回行结果，"
"按导入提交成功，行结果以报名页面为准"
)
return
if observed_new_operation and any(marker in text for marker in _SUCCESS_MARKERS):
print(f"[百补] {entry.name} 页面已确认导入提交，忽略行结果")
return
@@ -1284,10 +1432,21 @@ def _complete_draft_products(page: Any, entry: BaibuEntry) -> None:
if not entry.complete_drafts_after_import:
return
if not _click_draft_tab(page):
count = _draft_count(_page_text(page))
if count is None and not _has_visible_text(page, ("完善商品",)):
raise BaibuApplyError(f"{entry.name} 未找到草稿页入口")
# Right after the import acknowledgement the SPA may still be hydrating;
# the draft tab, the draft count and the per-row action rarely all appear
# on the first probe, so retry briefly before declaring the page broken.
deadline = time.monotonic() + _DRAFT_NAVIGATION_TIMEOUT_SECONDS
while time.monotonic() < deadline:
if _click_draft_tab(page):
break
text = _page_text(page)
if _draft_count(text) is not None or _has_visible_text(page, ("完善商品",)):
break
if any(marker in text for marker in _DRAFT_EMPTY_MARKERS):
break
time.sleep(0.5)
else:
raise BaibuApplyError(f"{entry.name} 未找到草稿页入口")
draft_count, action_count = _wait_for_draft_state(page, entry)
if draft_count == 0:
@@ -1439,6 +1598,8 @@ def _has_account_cookie(
def _persist_account_state(
browser: ScraplingBrowser,
entry: BaibuEntry,
*,
session_lock_held: bool = False,
) -> bool:
"""Publish a verified login to the account vault without leaking cookies."""
@@ -1463,8 +1624,12 @@ def _persist_account_state(
)
cookies = _account_cookies(cookies, required_domains)
if not _has_account_cookie(cookies, required_names):
names = sorted({str(item.get("name", "")) for item in cookies})
preview = "、".join(names[:12]) if names else "无"
raise BaibuApplyError(
f"{entry.name} 登录后未获得可用的账号 cookie，未更新账号登录态"
f"{entry.name} 登录后未获得可用的账号 cookie"
f"（域内 cookie：{preview}；要求含：{'、'.join(required_names)}），"
"未更新账号登录态"
)
raw_state = context.storage_state()
@@ -1472,23 +1637,33 @@ def _persist_account_state(
filtered_state = _account_storage_state(state, cookies, required_domains)
cookie_file, storage_state_file = paths
store = BrowserCookieStore(cookie_file, storage_state_file)
lock_root = cookie_file.parent.parent / "locks"
try:
with LockManager(lock_root).acquire(
f"browser-account:{entry.account_id}",
owner=f"tmall-baibu:{entry.entry_id}",
timeout_seconds=0,
):
def _save() -> None:
try:
store.save_cookies(cookies)
store.save_storage_state(filtered_state)
except ResourceBusyError as exc:
raise BaibuApplyError(
f"{entry.name} 账号登录态正在被其他进程更新，未覆盖账号登录态"
) from exc
except (OSError, ValueError, TypeError) as exc:
raise BaibuApplyError(
f"{entry.name} 保存账号登录态失败（{type(exc).__name__}）"
) from exc
except (OSError, ValueError, TypeError) as exc:
raise BaibuApplyError(
f"{entry.name} 保存账号登录态失败（{type(exc).__name__}）"
) from exc
if session_lock_held:
# The whole browser session already holds the scheduler-side
# browser-account lock; re-acquiring here would self-deadlock.
_save()
else:
lock_root = Settings.from_env().data_root / "state" / "locks"
try:
with LockManager(lock_root).acquire(
f"browser-account:{entry.account_id}",
owner=f"tmall-baibu:{entry.entry_id}",
timeout_seconds=0,
):
_save()
except ResourceBusyError as exc:
raise BaibuApplyError(
f"{entry.name} 账号登录态正在被其他进程更新，未覆盖账号登录态"
) from exc
print(f"[百补] 已将 {entry.account_id} 的已验证登录态同步到账号 vault")
return True
@@ -1498,6 +1673,33 @@ def _apply_entry(
template: Path,
*,
manual_login: bool = False,
) -> None:
# Hold the scheduler-side account browser lock for the whole session.
# The resident keepalive heartbeats the same profile on an interval and
# its timeout-0 acquire skips busy accounts, so this lock is what stops
# the heartbeat from killing the import mid-run (TargetClosedError /
# silent page death). Members must use the same lock directory as the
# scheduler (Settings.data_root/state/locks), not one derived from the
# vault paths.
locks = LockManager(Settings.from_env().data_root / "state" / "locks")
try:
with locks.acquire(
f"browser-account:{entry.account_id}",
owner=f"tmall-baibu:{entry.entry_id}",
timeout_seconds=_ACCOUNT_LOCK_TIMEOUT_SECONDS,
):
_apply_entry_locked(entry, template, manual_login=manual_login)
except ResourceBusyError as exc:
raise BaibuApplyError(
f"{entry.name} 账号浏览器正被保活/登录占用，请稍后重试"
) from exc
def _apply_entry_locked(
entry: BaibuEntry,
template: Path,
*,
manual_login: bool = False,
) -> None:
browser = _start_browser()
try:
@@ -1516,14 +1718,36 @@ def _apply_entry(
page,
manual_login=manual_login,
)
_wait_for_authenticated(
entry,
page,
login_attempted=login_attempted,
manual_login=manual_login,
)
try:
_wait_for_authenticated(
entry,
page,
login_attempted=login_attempted,
manual_login=manual_login,
)
except BaibuApplyError:
if login_attempted or manual_login:
raise
# The apply page may sit on an SSO interstitial without a login
# URL, so the passive check above never triggers the credentials;
# fall back to the explicit password login once.
_explicit_password_login(entry, page, manual_login=manual_login)
login_attempted = True
browser.persist_bound_state()
_persist_account_state(browser, entry)
try:
_persist_account_state(browser, entry, session_lock_held=True)
except BaibuApplyError as exc:
# A failed vault refresh can also mean the shop name was rendered
# from cache on top of a guest session; retry once through the
# explicit login before giving up on the vault refresh. The
# refresh only caches this session for later runs and must never
# block the import submission itself.
print(f"[WARN] {exc}；尝试显式重新登录后重试", file=sys.stderr)
try:
_explicit_password_login(entry, page, manual_login=manual_login)
_persist_account_state(browser, entry, session_lock_held=True)
except BaibuApplyError as retry_exc:
print(f"[WARN] {retry_exc}；本次继续执行报名导入", file=sys.stderr)
print(f"[百补] 已确认账号 {entry.account_id}，打开 {entry.name} 报名入口")
previous_operation = _open_batch_import(page, entry)
time.sleep(1)
tests/modules/product_commerce/test_tmall_baibu_apply.py
+414
View File
@@ -237,6 +237,62 @@ def test_login_if_needed_fills_the_password_form_inside_an_iframe(monkeypatch) -
assert frame.submit.click_count == 1
def test_login_if_needed_accepts_agreement_popup_and_resubmits(monkeypatch) -> None:
events: list[str] = []
class Locator:
def __init__(self, name: str) -> None:
self.name = name
def fill(self, _value: str) -> None:
events.append(f"fill:{self.name}")
def click(self, timeout: int = 0) -> None:
events.append(f"click:{self.name}")
username = Locator("username")
password = Locator("password")
submit = Locator("submit")
page = SimpleNamespace(wait_for_timeout=lambda ms: events.append(f"wait:{ms}"))
monkeypatch.setattr(baibu, "_authenticated", lambda *_args: False)
monkeypatch.setattr(baibu, "_is_login_url", lambda _url: True)
def fake_form(_page, *, use_password_form):
return (_page, username, password, submit)
monkeypatch.setattr(baibu, "_find_password_login_form", fake_form)
monkeypatch.setattr(
baibu,
"_credentials_from_environment",
lambda: (baibu._LoginCredentials(username="acct", secret_value="secret"), ()),
)
agreement = {"visible": True}
monkeypatch.setattr(
baibu, "_is_login_agreement_visible", lambda _page: agreement["visible"]
)
monkeypatch.setattr(
baibu,
"_click_login_agreement_accept",
lambda _page: agreement.update(visible=False) or True,
)
assert baibu._login_if_needed(_entry(), page) is True
# First submit is swallowed by the 服务协议 popup; after accepting it the
# credentials are re-filled and submitted a second time.
assert events == [
"fill:username",
"fill:password",
"click:submit",
"wait:2000",
"fill:username",
"fill:password",
"click:submit",
"wait:2000",
]
def test_manual_login_waits_without_filling_the_password_form(monkeypatch) -> None:
monkeypatch.setattr(baibu, "_authenticated", lambda _entry, _page: False)
frame = _LoginFrame()
@@ -301,6 +357,14 @@ def test_login_second_factor_marker_is_detected(monkeypatch) -> None:
assert baibu._login_second_factor_marker(SimpleNamespace()) == "短信验证"
def test_login_slider_captcha_marker_is_detected(monkeypatch) -> None:
monkeypatch.setattr(
baibu, "_page_text", lambda _page: "向右拖动滑块填充拼图完成验证"
)
assert baibu._login_second_factor_marker(SimpleNamespace()) == "滑块"
def test_persist_account_state_filters_cookies_to_the_account_domains(
monkeypatch,
tmp_path: Path,
@@ -351,6 +415,289 @@ def test_persist_account_state_filters_cookies_to_the_account_domains(
]
def _browser_spy(calls: list[str]):
class Page:
def goto(self, *_args, **_kwargs) -> None:
calls.append("goto")
class Context:
pages = [Page()]
def new_page(self):
return Page()
class Browser:
context = Context()
def start(self) -> None:
calls.append("start")
def persist_bound_state(self) -> None:
calls.append("persist-bound")
def close(self) -> None:
calls.append("close")
return Browser()
def test_apply_entry_continues_when_account_vault_refresh_fails(monkeypatch) -> None:
calls: list[str] = []
monkeypatch.setattr(baibu, "_start_browser", lambda: _browser_spy(calls))
monkeypatch.setattr(baibu, "_login_if_needed", lambda *_args, **_kwargs: False)
monkeypatch.setattr(baibu, "_wait_for_authenticated", lambda *_args, **_kwargs: None)
def fail_vault_refresh(_browser, _entry, **_kwargs):
calls.append("persist-fail")
raise baibu.BaibuApplyError(
"测试入口 登录后未获得可用的账号 cookie（域内 cookie：无；"
"要求含：unb、cookie2、_tb_token_），未更新账号登录态"
)
monkeypatch.setattr(baibu, "_persist_account_state", fail_vault_refresh)
def record_explicit_login(_entry, _page, *, manual_login):
calls.append("explicit-login")
monkeypatch.setattr(baibu, "_explicit_password_login", record_explicit_login)
def record_open(_page, _entry):
calls.append("open")
return None
monkeypatch.setattr(baibu, "_open_batch_import", record_open)
monkeypatch.setattr(baibu, "_upload_template", lambda *_args: calls.append("upload"))
monkeypatch.setattr(baibu, "_wait_import_result", lambda *_args, **_kwargs: calls.append("result"))
monkeypatch.setattr(baibu, "_complete_draft_products", lambda *_args: calls.append("drafts"))
monkeypatch.setattr(baibu.time, "sleep", lambda _seconds: None)
baibu._apply_entry(_entry(), Path("template.xlsx"))
assert calls == [
"start",
"goto",
"persist-bound",
"persist-fail",
"explicit-login",
"persist-fail",
"open",
"upload",
"result",
"drafts",
"close",
]
def test_apply_entry_retries_through_explicit_login_when_page_never_shows_login(
monkeypatch,
) -> None:
calls: list[str] = []
monkeypatch.setattr(baibu, "_start_browser", lambda: _browser_spy(calls))
monkeypatch.setattr(baibu, "_login_if_needed", lambda *_args, **_kwargs: False)
attempts = {"count": 0}
def fail_first_authentication(*_args, **_kwargs):
attempts["count"] += 1
if attempts["count"] == 1:
raise baibu.BaibuApplyError("测试入口 当前未登录对应店铺，已停止")
calls.append("authenticated")
monkeypatch.setattr(baibu, "_wait_for_authenticated", fail_first_authentication)
def record_explicit_login(_entry, _page, *, manual_login):
calls.append("explicit-login")
monkeypatch.setattr(baibu, "_explicit_password_login", record_explicit_login)
monkeypatch.setattr(
baibu,
"_persist_account_state",
lambda *_args, **_kwargs: calls.append("persist-ok") or True,
)
def record_open(_page, _entry):
calls.append("open")
return None
monkeypatch.setattr(baibu, "_open_batch_import", record_open)
monkeypatch.setattr(baibu, "_upload_template", lambda *_args: calls.append("upload"))
monkeypatch.setattr(baibu, "_wait_import_result", lambda *_args, **_kwargs: calls.append("result"))
monkeypatch.setattr(baibu, "_complete_draft_products", lambda *_args: calls.append("drafts"))
monkeypatch.setattr(baibu.time, "sleep", lambda _seconds: None)
baibu._apply_entry(_entry(), Path("template.xlsx"))
assert attempts["count"] == 1
assert calls == [
"start",
"goto",
"explicit-login",
"persist-bound",
"persist-ok",
"open",
"upload",
"result",
"drafts",
"close",
]
def test_explicit_password_login_drives_login_page_then_returns_to_apply(
monkeypatch,
) -> None:
calls: list[str] = []
class Page:
def goto(self, url, **_kwargs) -> None:
calls.append(f"goto:{url}")
def wait_for_timeout(self, _ms: int) -> None:
return None
monkeypatch.setattr(
baibu, "_login_if_needed", lambda *_args, **_kwargs: calls.append("fill-login") or True
)
monkeypatch.setattr(
baibu,
"_wait_for_authenticated",
lambda *_args, **_kwargs: calls.append("authenticated"),
)
entry = _entry()
baibu._explicit_password_login(entry, Page(), manual_login=False)
# The apply URL is re-entered first so its redirect chain can surface the
# seller login page; only a missing form there falls back to login.taobao.
assert calls == [
f"goto:{entry.apply_url}",
"fill-login",
f"goto:{entry.apply_url}",
"authenticated",
]
class _FakeLock:
def __init__(self, manager: "_FakeLockManager", resource: str) -> None:
self._manager = manager
self._resource = resource
def __enter__(self):
self._manager.held.add(self._resource)
self._manager.entered.append(self._resource)
return self
def __exit__(self, *_args) -> None:
self._manager.held.discard(self._resource)
class _FakeLockManager:
def __init__(self, lock_root) -> None:
self.lock_root = lock_root
self.held: set[str] = set()
self.entered: list[str] = []
self.requested: list[tuple[str, int]] = []
def acquire(self, resource: str, *, owner: str, timeout_seconds: int):
self.requested.append((resource, timeout_seconds))
if getattr(self, "busy", False):
raise baibu.ResourceBusyError(resource)
return _FakeLock(self, resource)
def test_apply_entry_holds_account_lock_for_whole_session(monkeypatch) -> None:
calls: list[str] = []
lock_manager = _FakeLockManager(lock_root="locks-dir")
class FakeSettings:
data_root = Path("data-root")
@classmethod
def from_env(cls):
return cls()
monkeypatch.setattr(baibu, "Settings", FakeSettings)
monkeypatch.setattr(baibu, "LockManager", lambda lock_root: lock_manager)
monkeypatch.setattr(baibu, "_start_browser", lambda: _browser_spy(calls))
monkeypatch.setattr(baibu, "_login_if_needed", lambda *_args, **_kwargs: False)
monkeypatch.setattr(baibu, "_wait_for_authenticated", lambda *_args, **_kwargs: None)
monkeypatch.setattr(
baibu,
"_persist_account_state",
lambda *_args, **_kwargs: True,
)
monkeypatch.setattr(baibu, "_open_batch_import", lambda _page, _entry: None)
def assert_lock_held_during_upload(_page, _template):
assert lock_manager.held == {"browser-account:tmall-shop"}
calls.append("upload")
monkeypatch.setattr(baibu, "_upload_template", assert_lock_held_during_upload)
monkeypatch.setattr(baibu, "_wait_import_result", lambda *_args, **_kwargs: None)
monkeypatch.setattr(baibu, "_complete_draft_products", lambda *_args: None)
monkeypatch.setattr(baibu.time, "sleep", lambda _seconds: None)
baibu._apply_entry(_entry(), Path("template.xlsx"))
assert lock_manager.requested == [("browser-account:tmall-shop", 120)]
assert lock_manager.entered == ["browser-account:tmall-shop"]
assert lock_manager.held == set()
def test_apply_entry_reports_busy_account_lock(monkeypatch) -> None:
lock_manager = _FakeLockManager(lock_root="locks-dir")
lock_manager.busy = True
class FakeSettings:
data_root = Path("data-root")
@classmethod
def from_env(cls):
return cls()
monkeypatch.setattr(baibu, "Settings", FakeSettings)
monkeypatch.setattr(baibu, "LockManager", lambda lock_root: lock_manager)
started: list[str] = []
monkeypatch.setattr(
baibu, "_start_browser", lambda: started.append("start") or object()
)
with pytest.raises(baibu.BaibuApplyError, match="正被保活/登录占用"):
baibu._apply_entry(_entry(), Path("template.xlsx"))
assert started == []
def test_explicit_password_login_falls_back_to_plain_login_page(
monkeypatch,
) -> None:
calls: list[str] = []
class Page:
def goto(self, url, **_kwargs) -> None:
calls.append(f"goto:{url}")
def wait_for_timeout(self, _ms: int) -> None:
return None
monkeypatch.setattr(baibu, "_login_if_needed", lambda *_args, **_kwargs: False)
monkeypatch.setattr(
baibu,
"_wait_for_authenticated",
lambda *_args, **_kwargs: calls.append("authenticated"),
)
entry = _entry()
baibu._explicit_password_login(entry, Page(), manual_login=False)
assert calls == [
f"goto:{entry.apply_url}",
f"goto:{baibu._LOGIN_PAGE_URL}",
f"goto:{entry.apply_url}",
"authenticated",
]
def test_upload_template_supports_tmall_dialog_import_button(
monkeypatch,
tmp_path: Path,
@@ -558,6 +905,32 @@ def test_wait_import_result_accepts_new_failure_summary_as_submitted(
baibu._wait_import_result(SimpleNamespace(), _entry())
def test_wait_import_result_accepts_slow_processing_after_acknowledgement(
monkeypatch,
capsys,
) -> None:
# Tmall acknowledged the batch but never returned a row-level summary
# within the acceptance window; the acknowledged submission still counts.
processing_text = (
"您于2026-10-14 14:51:00执行了商品批量导入操作，正在处理中。"
)
clock = {"now": 0.0}
def advance_monotonic() -> float:
clock["now"] += 15.0
return clock["now"]
monkeypatch.setattr(baibu, "_page_text", lambda _page: processing_text)
monkeypatch.setattr(baibu, "_refresh_batch_results", lambda _page: True)
monkeypatch.setattr(baibu.time, "sleep", lambda _seconds: None)
monkeypatch.setattr(baibu.time, "monotonic", advance_monotonic)
monkeypatch.setattr(baibu, "_RESULT_REFRESH_INTERVAL_SECONDS", 0)
baibu._wait_import_result(SimpleNamespace(), _entry())
assert "按导入提交成功" in capsys.readouterr().out
def test_wait_import_result_ignores_stale_summary_until_signature_changes(
monkeypatch,
) -> None:
@@ -663,6 +1036,47 @@ def test_complete_draft_products_accepts_empty_data_page(monkeypatch) -> None:
baibu._complete_draft_products(page, entry)
def test_complete_draft_products_retries_draft_tab_during_hydration(
monkeypatch,
) -> None:
entry = replace(_entry(), complete_drafts_after_import=True)
page = SimpleNamespace()
attempts = {"n": 0}
def flaky_tab(_page):
attempts["n"] += 1
return attempts["n"] >= 3
monkeypatch.setattr(baibu, "_click_draft_tab", flaky_tab)
monkeypatch.setattr(baibu, "_page_text", lambda _page: "加载中")
monkeypatch.setattr(baibu, "_has_visible_text", lambda _page, _labels: False)
monkeypatch.setattr(baibu, "_wait_for_draft_state", lambda _page, _entry: (0, 0))
monkeypatch.setattr(baibu.time, "sleep", lambda _seconds: None)
ticks = iter(range(10_000))
monkeypatch.setattr(baibu.time, "monotonic", lambda: next(ticks) * 0.4)
baibu._complete_draft_products(page, entry)
assert attempts["n"] == 3
def test_complete_draft_products_reports_missing_draft_entry_after_retries(
monkeypatch,
) -> None:
entry = replace(_entry(), complete_drafts_after_import=True)
page = SimpleNamespace()
monkeypatch.setattr(baibu, "_click_draft_tab", lambda _page: False)
monkeypatch.setattr(baibu, "_page_text", lambda _page: "加载中")
monkeypatch.setattr(baibu, "_has_visible_text", lambda _page, _labels: False)
monkeypatch.setattr(baibu.time, "sleep", lambda _seconds: None)
ticks = iter(range(10_000))
monkeypatch.setattr(baibu.time, "monotonic", lambda: next(ticks) * 0.4)
with pytest.raises(baibu.BaibuApplyError, match="未找到草稿页入口"):
baibu._complete_draft_products(page, entry)
def test_return_to_draft_list_falls_back_to_browser_history(monkeypatch) -> None:
events: list[str] = []