From 6cd563eab0043cbc172fa1fa7e330a9924ea33a7 Mon Sep 17 00:00:00 2001 From: wangyunlong <3080433063@qq.com> Date: Wed, 14 Oct 2026 18:33:32 +0800 Subject: [PATCH] =?UTF-8?q?fix(product):=20=E5=A4=A9=E7=8C=AB=E7=99=BE?= =?UTF-8?q?=E8=A1=A5=E6=8A=A5=E5=90=8D=E7=99=BB=E5=BD=95=E9=93=BE=E8=B7=AF?= =?UTF-8?q?=E4=B8=89=E7=B1=BB=E9=98=BB=E6=96=AD=E5=85=9C=E5=BA=95?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit vault 假登录态（SPA 缓存店铺名误判已登录）与 SSO 中间态停留时改走显式 账密登录兜底；登录页识别服务协议/隐私弹窗并先处理，二次验证标记补充 滑块；报名导入增加账号级锁防并发挤兑。 --- .../tmall_baibu_apply_common.py | 280 ++++++++++-- .../test_tmall_baibu_apply.py | 414 ++++++++++++++++++ 2 files changed, 666 insertions(+), 28 deletions(-) diff --git a/src/gyxx_flow/modules/product_commerce/tmall_baibu_apply_common.py b/src/gyxx_flow/modules/product_commerce/tmall_baibu_apply_common.py index cd38201..7174043 100644 --- a/src/gyxx_flow/modules/product_commerce/tmall_baibu_apply_common.py +++ b/src/gyxx_flow/modules/product_commerce/tmall_baibu_apply_common.py @@ -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) diff --git a/tests/modules/product_commerce/test_tmall_baibu_apply.py b/tests/modules/product_commerce/test_tmall_baibu_apply.py index a84f985..9205647 100644 --- a/tests/modules/product_commerce/test_tmall_baibu_apply.py +++ b/tests/modules/product_commerce/test_tmall_baibu_apply.py @@ -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] = []