From e2875f0c00e818278f8037022f498d6286dede42 Mon Sep 17 00:00:00 2001 From: Jerry Yan <792602257@qq.com> Date: Fri, 14 Aug 2026 14:05:47 +0800 Subject: [PATCH] =?UTF-8?q?=E8=87=AA=E5=8A=A8=E9=87=8D=E7=99=BB=E4=BF=AE?= =?UTF-8?q?=E4=B8=A4=E5=A4=84=EF=BC=9A=E5=B9=B6=E5=8F=91=E5=8E=BB=E9=87=8D?= =?UTF-8?q?=E8=AF=BB=E9=94=99=E7=BC=93=E5=AD=98=E3=80=81=E5=8F=AA=E8=AF=BB?= =?UTF-8?q?=E8=AE=A2=E5=8D=95=E6=9F=A5=E8=AF=A2=E6=8E=89=E7=99=BB=E5=BD=95?= =?UTF-8?q?=E8=A2=AB=E9=9D=99=E9=BB=98=E5=90=9E=E6=8E=89?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - try_relogin 的并发去重原本读 status().logged_in，但重登成功后的 reload() 会把它重置成 None，排队在 site 锁上的调用方一律判「还没人登上」，N 个并发 调用会串行触发 N 次真实登录（各自最长 relogin_timeout）。改用 _relogin_epochs 计数：等锁期间 epoch 变过就真探测一次，已登录即跳过；仍未登录说明这轮站点侧 就是登不上（验证码/密码错/风控），直接失败，不在同一波并发里重复触发。 原并发测试的桩自相矛盾（login_one 返回成功、探针页始终回未登录），断言只能 松到 count >= 1；桩改为登录成功时翻转探针页，断言收紧到 count == 1。 - check_order_status / list_recent_orders 执行中掉登录此前会被静默吞掉：订单页 被踢到 SSO 后既不报错也没订单号，_parse_order_status 返回 found=False 被 _monitor_order 当成「订单还没反映出来」继续轮询（默认 3 小时一轮）， _parse_order_list 则退化成空列表让 verify_on_site 转 unknown 卡住等人工。 新增 SiteInteractor._read_with_relogin_retry 外壳：只读操作中途判定掉登录时 重登一次并整个重跑，第二次仍失败抛 NotLoggedInError。只给读操作用——写操作 中途掉登录不能重跑（上次动作可能已在站点侧生效），这条边界在两边文档里写明。 - 判据是新增的 auth_site.looks_logged_out：与探针页上权威的 is_logged_in 分开， 它是业务页上的单边启发式（返回 False 不代表登录着），只用于「判错最多多花一次 重登」的重试决策。刻意排除 session/upgrade——那是已登录时的站点风控复核密码， 不是 cookie 过期，误判会把风控当掉登录去重登。 判据里「掉登录会跳到 SSO 域」这一步没有真实探测证据（要复现得先让一份真实登录态 过期），是按站点通行行为的推断，已在常量注释标注；新增测试用替身页面，不是真实 站点 HTML。399 测试全绿（仓库未配 ruff/flake8/mypy，只跑了 pytest）。 Co-Authored-By: Claude Opus 5 (1M context) --- app/trading/core/auth_site.py | 48 ++++++ app/trading/services/auth_session.py | 48 +++++- app/trading/worker/site_interact.py | 114 +++++++++++-- tests/test_auth_session.py | 58 ++++++- tests/test_site_interact.py | 239 ++++++++++++++++++++++++++- 5 files changed, 480 insertions(+), 27 deletions(-) diff --git a/app/trading/core/auth_site.py b/app/trading/core/auth_site.py index 35282e3..a48a46c 100644 --- a/app/trading/core/auth_site.py +++ b/app/trading/core/auth_site.py @@ -37,6 +37,22 @@ RAKUTEN_LOGIN_URL: Final = "https://www.rakuten.co.jp/myrakuten/" # 购物车页上表示「当前会话未登录」的文案。出现即判定登录态失效。 RAKUTEN_LOGGED_OUT_MARKER: Final = "現在ログインしていません" +# 未登录时访问「需要登录才能看」的页面（订单列表/详情等）会被整页跳到 SSO 域。 +# 判据来源：login_runner 模块文档记录的实测——SSO 落地域名确实是这两个之一。 +# 但「订单页在登录态失效时一定跳到这里」这一步**没有**真实探测证据（要复现得 +# 先让一份真实登录态过期），是按站点通行行为的推断，用途也限定在 +# `looks_logged_out` 那种「判错只多花一次重登」的场景，不作为正向结论使用。 +RAKUTEN_SSO_URL_MARKERS: Final = ( + "login.account.rakuten.com", + "grp01.id.rakuten.co.jp", +) + +# SSO 域下的「已登录但要求复核密码」路径。这**不是**登录态失效：结算流程里即使 +# 会话有效也会被站点风控要求重输密码（见 site_interact.py 模块文档与 +# _SESSION_UPGRADE_URL_MARKERS）。落在这些路径上时不能判未登录，否则会把风控 +# 拦截误当成 cookie 过期去重登。 +RAKUTEN_SSO_UPGRADE_PATH_MARKERS: Final = ("session/upgrade",) + # ---- 登录态请求指纹 ---- # 这个 UA 必须与 scripts/login.py 起浏览器时用的一致：cookie 是在那个 UA 下拿到的， # 服务端再拿它发请求时换了 UA，可能触发站点的设备校验让登录态提前失效。 @@ -101,3 +117,35 @@ def is_logged_in(site: str, *, final_url: str, body: str) -> bool: if site != "rakuten": raise ValueError(f"未知站点：{site}") return RAKUTEN_LOGGED_OUT_MARKER not in body + + +def looks_logged_out(site: str, *, final_url: str, body: str) -> bool: + """在**非探针页**上判断「这次拿到的页面像是因为掉登录才长这样」 + + 与 `is_logged_in` 的区别，别混用： + + - `is_logged_in` 是探针页（购物车页）上的**正向**判据，用来回答「这套 cookie + 还有效吗」，是登录态的权威结论。 + - 本函数是订单列表页 / 订单详情页这类业务页上的**单边启发式**：返回 True 只 + 表示「值得回探针页复核一次登录态」，返回 False **不代表登录着**（业务页 + 正常渲染时本来就没有任何未登录信号）。所以它只该用在「判错了最多多花一次 + 探测/重登」的重试决策上，不能拿来当登录态结论对外报告。 + + 判据（前者是推断，后者是实测）： + 1. 落地 URL 进了 SSO 域（`RAKUTEN_SSO_URL_MARKERS`），且不是「已登录但要求 + 复核密码」的 session upgrade 路径——后者是站点风控，不是掉登录， + 误判会把风控拦截当 cookie 过期去重登。 + 注意 SSO 域下的 `sign_in/password`（真正的密码输入页）**不**在豁免里： + 它既可能是全新登录的密码步、也可能是 upgrade 的后续步，含义有歧义， + 按「像掉登录」处理——代价只是多跑一次 `login_one`，而它自己会先探测 + 登录态、已登录就直接跳过。 + 2. 正文出现旧版未登录 marker（SSR 落地 HTML 上实测有效，新 SPA 渲染后不再 + 输出，所以这条同样是单边的）。 + """ + if site != "rakuten": + raise ValueError(f"未知站点：{site}") + lowered = (final_url or "").lower() + if any(marker in lowered for marker in RAKUTEN_SSO_URL_MARKERS): + if not any(path in lowered for path in RAKUTEN_SSO_UPGRADE_PATH_MARKERS): + return True + return RAKUTEN_LOGGED_OUT_MARKER in body diff --git a/app/trading/services/auth_session.py b/app/trading/services/auth_session.py index 390e80e..cd55dd6 100644 --- a/app/trading/services/auth_session.py +++ b/app/trading/services/auth_session.py @@ -89,6 +89,11 @@ class AuthSession: self._sites: dict[str, _SiteAuth] = {} # 自动重登的 site 级互斥锁：同账号同时只能一个登录流程（user_data_dir 被锁） self._relogin_locks: dict[str, asyncio.Lock] = {} + # 每个 site 已完成的重登尝试计数。用途只有一个：让在锁上排队的调用方能分辨 + # 「我等的这段时间里已经有人替我登过了」。不能用 status().logged_in 代替—— + # 重登成功后的 reload() 会把 logged_in 重置成 None（未探测），排队者读到 + # None 会误判成「还没人登过」，于是 N 个并发调用串行触发 N 次真实登录。 + self._relogin_epochs: dict[str, int] = {} # ---- 生命周期 ---- @@ -211,8 +216,11 @@ class AuthSession: 2. 重登成功 → 重新 check 一次，登录态转好即放行 3. 重登失败 / 未启用 / account.yaml 缺失 → 抛 NotLoggedInError 让 worker 转 needs_human - 重登只在这一层（任务前置检查）触发；任务执行过程中失效不重试， - 避免脏状态（cart 已提交但响应后 cookie 失效等场景）。 + **写**操作（加购 / 提交订单 / 付款）只在这一层（动作前置检查）触发重登； + 执行过程中失效不重试，避免脏状态（cart 已提交但响应后 cookie 失效等场景）。 + **只读**操作（订单列表 / 订单详情查询）另有一层执行中重试，见 + `site_interact.SiteInteractor._read_with_relogin_retry`——重跑一次查询没有 + 副作用，这条边界是刻意区分的，不要把它推广到写操作上。 """ status = await self.check(site) if status.logged_in: @@ -246,11 +254,34 @@ class AuthSession: # site 级锁：同账号同 user_data_dir，并发重登会撞锁 lock = self._relogin_locks.setdefault(site, asyncio.Lock()) + epoch_before_wait = self._relogin_epochs.get(site, 0) async with lock: - # 拿锁后再 check 一次——可能别的协程刚重登过 - status = self.status(site) - if status.logged_in: - return True + # 在锁上等过、且期间有人跑完了一轮重登 → 大概率不需要再登一次。 + # 这里必须**真探测**而不是读 self.status()：重登成功后的 reload() 把 + # logged_in 重置成 None，读缓存会一律判「还没登上」，白跑一次 login_one + # （最坏 N 个并发调用串行触发 N 次真实登录，各自最长 relogin_timeout）。 + if self._relogin_epochs.get(site, 0) != epoch_before_wait: + # 探测失败（网络问题）时不在这里抛错——try_relogin 的契约是「降级 + # 返回 False / 继续走登录」，不是抛异常；这种情况直接落到下面正常 + # 跑一次 login_one（它自己也会先探测登录态，已登录就跳过填表）。 + try: + status = await self.check(site) + except UpstreamRequestError: + logger.warning( + "自动重登：site=%s 等锁后复核登录态失败，继续尝试登录", site, exc_info=True + ) + else: + if status.logged_in: + logger.info("自动重登跳过：site=%s 等锁期间已由其他调用方登录完成", site) + return True + # 刚有人登过还是没登上，说明这轮站点侧就是登不上（验证码/密码错/ + # 风控）。同一波并发里再串一遍只是把每个调用方各卡一个 + # relogin_timeout，站点侧结果不会变——直接失败，交人工。 + logger.warning( + "自动重登跳过：site=%s 等锁期间刚失败过一次，不在同一波并发里重复触发", + site, + ) + return False # 读 account.yaml（失败时降级，不抛错） try: @@ -282,6 +313,11 @@ class AuthSession: except Exception: logger.exception("自动重登异常：site=%s account_id=%s", site, account.id) return False + finally: + # 成败都记一次「本站跑过一轮登录流程」。放 finally 是因为排队者要 + # 分辨的是「有没有人替我尝试过」，失败的尝试同样算——否则失败后 + # 排队的调用方会一个接一个重跑同一个注定失败的登录。 + self._relogin_epochs[site] = self._relogin_epochs.get(site, 0) + 1 if not ok: logger.warning("自动重登失败：site=%s account_id=%s", site, account.id) diff --git a/app/trading/worker/site_interact.py b/app/trading/worker/site_interact.py index 4ccc68c..699937c 100644 --- a/app/trading/worker/site_interact.py +++ b/app/trading/worker/site_interact.py @@ -81,7 +81,7 @@ import logging import re from dataclasses import dataclass, field from datetime import datetime -from typing import TYPE_CHECKING +from typing import TYPE_CHECKING, TypeVar from app.shared.errors import ( CartOperationError, @@ -102,6 +102,8 @@ from app.trading.core import auth_site from app.trading.worker.models import LeaseTask if TYPE_CHECKING: + from collections.abc import Awaitable, Callable + from playwright.async_api import Page from app.shared.config import Settings @@ -110,6 +112,9 @@ if TYPE_CHECKING: logger = logging.getLogger(__name__) +# _read_with_relogin_retry 的返回值类型：只读操作的结果原样透传 +_T = TypeVar("_T") + # 普通购买事件标识、inventory_flag 映射、basketDomain 反转义、form_fields 基础构造 # 均在 app.shared.purchase_contract，与 scraping/parsers/item.py::_purchase_info 共用同一份契约 _BASKET_PATH = "/rms/mall/bss/cartadd/set" @@ -338,6 +343,16 @@ _ORDER_LIST_URL = "https://order.my.rakuten.co.jp/purchase-history/order-list" _ORDER_LIST_MAX_PAGES = 20 +class _LoggedOutMidRead(Exception): + """内部信号：只读操作执行到一半，发现页面像是因为掉登录才长这样 + + 不对外暴露（`_read_with_relogin_retry` 一定会把它转成 `NotLoggedInError` + 或吞掉后重试）。存在的理由是「掉登录」在页面解析层与业务失败长得完全不一样： + 订单页被踢到 SSO 后既不会报错也不会有订单号，静默走进解析逻辑就会被当成 + 「订单还没出现」，日志里查不出真正原因。用一个专门的信号把这条路径截出来。 + """ + + @dataclass(slots=True) class OrderListItem: """订单列表页单笔订单里的一个商品条目""" @@ -589,6 +604,51 @@ class SiteInteractor: ) self._state_mtime = mtime + async def _read_with_relogin_retry( + self, label: str, read: "Callable[[], Awaitable[_T]]" + ) -> "_T": + """**只读**站点操作的统一外壳：前置登录检查 → 执行 → 中途掉登录则重登重跑 + + `read` 在 `self._lock` 内、`require_logged_in` 与 context 刷新之后执行； + 它通过抛 `_LoggedOutMidRead` 表达「这页看着是掉登录了」，本方法负责重登 + 一次并把 `read` 整个重跑一遍。 + + 为什么只给只读操作用：写操作（加购 / 提交订单 / 付款）中途掉登录**不能** + 重跑——上一次动作可能已经在站点侧生效，只是响应没拿到，重跑等于重复提交。 + 这条边界在 `AuthSession.require_logged_in` 的文档里已经写明，本外壳不动它， + 只把「读」这一侧本来就安全的可重试性补上。 + + 只重试一次：重登一次还是没登上，就是需要人工（验证码 / 密码错 / 风控）， + 再试只是把调用方多卡一个 relogin_timeout，结果不会变。 + """ + async with self._lock: + for attempt in (1, 2): + await self._auth_session.require_logged_in("rakuten") + # 重登会重写 storage_state，本 context 的 cookie 是启动时的快照， + # 必须在每次 attempt 前都刷一遍，否则重登后仍拿旧 cookie 去读。 + await self._refresh_context_if_stale() + try: + return await read() + except _LoggedOutMidRead as exc: + if attempt == 2: + raise NotLoggedInError( + site="rakuten", + detail=f"{label}：重登后仍被判定未登录（{exc}）", + ) from exc + logger.warning( + "%s：执行中检测到登录态失效（%s），尝试重登后重跑一次", label, exc, + ) + # require_logged_in 只在「前置 check 判未登录」时才会重登，而这里 + # 是它放行之后才掉的登录——缓存里还是 logged_in=True，下一轮 + # require_logged_in 的 check 未必立刻翻转。所以主动触发一次重登， + # 让下一轮 attempt 有新 cookie 可用。 + if not await self._auth_session.try_relogin("rakuten"): + raise NotLoggedInError( + site="rakuten", + detail=f"{label}：执行中登录态失效且自动重登未成功（{exc}）", + ) from exc + raise AssertionError("unreachable") # pragma: no cover + async def close(self) -> None: """关闭 context、browser、playwright，吞掉单个 close 异常""" for task_id in list(self._checkout_pages): @@ -1719,21 +1779,22 @@ class SiteInteractor: submit_order/pay 已经处理完并关闭了 `_checkout_pages` 里留存的会话， 监控阶段没有需要跨调用复用的页面状态。 + 执行中掉登录会自动重登一次并重跑（详见 `_read_with_relogin_retry`）： + 本方法是纯读，重跑没有副作用；不加这层的话订单页被踢到 SSO 会静默走进 + 解析逻辑、被当成「订单号还没出现」，轮询白转几个小时也看不出原因。 + Returns: OrderStatusSnapshot；订单号暂时查不到、或进度条解析不出新阶段都 **不算错误**（详见该 dataclass 文档），由调用方决定是否继续轮询。 Raises: - NotLoggedInError: 登录态失效 + NotLoggedInError: 登录态失效且自动重登没能恢复 OrderOperationError: 订单详情页打开/渲染失败 """ shop_id = site_order_id.split("-", 1)[0] url = _ORDER_DETAIL_URL_TEMPLATE.format(order_number=site_order_id, shop_id=shop_id) - async with self._lock: - await self._auth_session.require_logged_in("rakuten") - await self._refresh_context_if_stale() - + async def read() -> OrderStatusSnapshot: page = await self._context.new_page() try: try: @@ -1745,10 +1806,23 @@ class SiteInteractor: f"{type(exc).__name__}: {exc}" ) from exc html = await page.content() + final_url = page.url finally: await page.close() - return _parse_order_status(html, site_order_id) + # 只在「没解析到订单」时才查掉登录：解析到了就说明页面是真的订单页， + # 没必要再判一次；反过来，found=False 有两种可能（订单还没反映出来 / + # 被踢去登录），这里把后者摘出来，不让它伪装成前者。 + snapshot = _parse_order_status(html, site_order_id) + if not snapshot.found and auth_site.looks_logged_out( + "rakuten", final_url=final_url, body=html + ): + raise _LoggedOutMidRead(f"订单详情页落地 {final_url}") + return snapshot + + return await self._read_with_relogin_retry( + f"check_order_status site_order_id={site_order_id}", read + ) async def list_recent_orders(self, *, since: datetime) -> OrderListWindow: """恢复核对用：拉取「任务创建时间之后」的订单列表（规格 §5 依赖它） @@ -1763,14 +1837,16 @@ class SiteInteractor: page=2 返回空列表这两点；多页翻页的排序假设（新订单在前）未经真实数据 验证，多页场景上线前应重新探测确认。 + 执行中掉登录会自动重登一次并从第一页重跑（详见 `_read_with_relogin_retry`）。 + 这条路径对恢复核对尤其要紧：掉登录时 `_parse_order_list` 拿不到 + `pageType="ph-list"`，会退化成「空列表 + 没覆盖完窗口」，调用方 + （verify_on_site）只能转 unknown 卡住等人工——本来自动重登一次就能查下去。 + Raises: - NotLoggedInError: 登录态失效 + NotLoggedInError: 登录态失效且自动重登没能恢复 OrderOperationError: 订单列表页打开/渲染失败 """ - async with self._lock: - await self._auth_session.require_logged_in("rakuten") - await self._refresh_context_if_stale() - + async def read() -> OrderListWindow: acc = _OrderListAccumulator() stop = False @@ -1786,14 +1862,26 @@ class SiteInteractor: f"订单列表页打开失败：page={page_num} {type(exc).__name__}: {exc}" ) from exc html = await page.content() + final_url = page.url page_result = _parse_order_list(html) acc, stop = _accumulate_order_list_page(acc, page_result, since=since) + # gave_up＝第一页就拿不到结构化列表数据。原因可能是页面改版，也 + # 可能是掉登录被踢走；后者可以自愈，摘出来重登重试，改版则继续 + # 按原路径返回「没覆盖完」交人工。 + if acc.gave_up and auth_site.looks_logged_out( + "rakuten", final_url=final_url, body=html + ): + raise _LoggedOutMidRead(f"订单列表页落地 {final_url}") if stop: break finally: await page.close() - return OrderListWindow(entries=acc.entries, window_fully_covered=stop and not acc.gave_up) + return OrderListWindow( + entries=acc.entries, window_fully_covered=stop and not acc.gave_up + ) + + return await self._read_with_relogin_retry("list_recent_orders", read) # ---- 模块级辅助函数（纯函数，便于单测）---- diff --git a/tests/test_auth_session.py b/tests/test_auth_session.py index b4435e6..3c22fe4 100644 --- a/tests/test_auth_session.py +++ b/tests/test_auth_session.py @@ -330,10 +330,16 @@ async def test_try_relogin_is_disabled_when_flag_off(tmp_path): async def test_try_relogin_serializes_concurrent_calls_same_site(tmp_path, monkeypatch): - """同 site 并发触发只跑一次 login_one：靠 site 级锁串行化""" + """同 site 并发触发**只跑一次** login_one：锁串行化 + 排队者复核登录态后跳过 + + 不能只断言「没并发执行」：重登成功后的 reload() 会把 logged_in 重置成 None， + 排队者如果读缓存去重就会一律判「还没人登上」，于是 N 个并发调用串行触发 N 次 + 真实登录（每次最长 relogin_timeout）。这里断言 count == 1 就是钉住这一点。 + """ _patch_accounts_file(monkeypatch, tmp_path) invocations = {"count": 0, "in_flight_max": 0, "current": 0} + state = {"logged_in": False} from app.trading.services import login_runner async def counting_login_one(account, s, *, timeout_seconds=300, progress=None): @@ -348,12 +354,19 @@ async def test_try_relogin_serializes_concurrent_calls_same_site(tmp_path, monke "rakuten", [{"name": "X", "value": "1", "domain": ".rakuten.co.jp", "path": "/"}], ) + # 登录成功后探针页也必须随之翻转，否则桩自相矛盾（login_one 说成功、站点说 + # 没登上），测不出真实行为 + state["logged_in"] = True return True monkeypatch.setattr(login_runner, "login_one", counting_login_one) + def handler(request: httpx.Request) -> httpx.Response: + text = CART_LOGGED_IN if state["logged_in"] else CART_LOGGED_OUT + return httpx.Response(200, text=text) + settings = make_settings(tmp_path, relogin_enabled=True, relogin_timeout_seconds=10) - session = await build_session(settings, lambda r: httpx.Response(200, text=CART_LOGGED_OUT)) + session = await build_session(settings, handler) try: # 第一次状态为 logged_in=None，三个并发都进入 try_relogin @@ -362,12 +375,43 @@ async def test_try_relogin_serializes_concurrent_calls_same_site(tmp_path, monke session.try_relogin("rakuten"), session.try_relogin("rakuten"), ) - # login_one 至少被调一次（串行下后续可能因 status 已 logged_in 跳过） - assert invocations["count"] >= 1 - # 关键：login_one 永远没并发执行 + assert invocations["count"] == 1 + # login_one 永远没并发执行（同账号同 user_data_dir，撞锁会失败） assert invocations["in_flight_max"] == 1 - # 结果都成功（要么真重登，要么拿到锁后发现已 logged_in） - assert all(results) + # 三个调用方都拿到「已登录」结论：一个真登录，两个复核后跳过 + assert results == [True, True, True] + finally: + await session.close() + + +async def test_try_relogin_does_not_retry_after_queued_failure(tmp_path, monkeypatch): + """一次重登失败后，排在锁上的并发调用不再重复触发同一个注定失败的登录 + + 站点侧登不上（验证码/密码错/风控）时，把每个调用方各卡一个 relogin_timeout + 不会改变结果，只会让整批任务慢几倍。 + """ + _patch_accounts_file(monkeypatch, tmp_path) + + calls = {"count": 0} + from app.trading.services import login_runner + + async def failing_login_one(account, s, *, timeout_seconds=300, progress=None): + calls["count"] += 1 + await asyncio.sleep(0.05) # 让其余并发调用叠到锁上 + return False + + monkeypatch.setattr(login_runner, "login_one", failing_login_one) + + settings = make_settings(tmp_path, relogin_enabled=True, relogin_timeout_seconds=10) + session = await build_session(settings, lambda r: httpx.Response(200, text=CART_LOGGED_OUT)) + try: + results = await asyncio.gather( + session.try_relogin("rakuten"), + session.try_relogin("rakuten"), + session.try_relogin("rakuten"), + ) + assert results == [False, False, False] + assert calls["count"] == 1 finally: await session.close() diff --git a/tests/test_site_interact.py b/tests/test_site_interact.py index 75f6832..92e7e3b 100644 --- a/tests/test_site_interact.py +++ b/tests/test_site_interact.py @@ -17,7 +17,13 @@ import pytest from datetime import datetime, timezone -from app.shared.errors import CartOperationError, InvalidRequestError, OrderOperationError +from app.shared.errors import ( + CartOperationError, + InvalidRequestError, + NotLoggedInError, + OrderOperationError, +) +from app.trading.core import auth_site from app.trading.worker.models import LeaseTask from app.shared.task_state import OrderState from app.trading.worker.site_interact import ( @@ -337,6 +343,237 @@ async def test_check_order_status_without_start_raises(): await site.check_order_status(_REAL_ORDER_ID) +# ---- auth_site.looks_logged_out：业务页上的单边「像是掉登录」判据 ---- + + +def test_looks_logged_out_detects_sso_redirect(): + """订单页落地到 SSO 域 → 判为像掉登录""" + assert auth_site.looks_logged_out( + "rakuten", + final_url="https://login.account.rakuten.com/sso/authorize?client_id=x", + body="ログイン", + ) + + +def test_looks_logged_out_ignores_session_upgrade(): + """session upgrade 是「已登录但要求复核密码」的站点风控，不是掉登录 + + 误判会把风控拦截当 cookie 过期，白跑一次重登还掩盖真实原因。 + """ + assert not auth_site.looks_logged_out( + "rakuten", + final_url="https://login.account.rakuten.com/sign_in/session/upgrade?client_id=x", + body="パスワードを入力", + ) + + +def test_looks_logged_out_detects_legacy_marker_on_business_page(): + """业务页正文出现旧版未登录 marker（SSR 落地 HTML）→ 判为像掉登录""" + assert auth_site.looks_logged_out( + "rakuten", + final_url="https://order.my.rakuten.co.jp/purchase-history/order-list", + body=f"{auth_site.RAKUTEN_LOGGED_OUT_MARKER}", + ) + + +def test_looks_logged_out_false_on_normal_order_page(): + """正常订单页没有任何未登录信号 → False（注意这**不**等于「确认登录着」）""" + assert not auth_site.looks_logged_out( + "rakuten", + final_url="https://order.my.rakuten.co.jp/purchase-history/order-list", + body=_stepper_html(active_stage="出荷"), + ) + + +def test_looks_logged_out_rejects_unknown_site(): + with pytest.raises(ValueError): + auth_site.looks_logged_out("rakuma", final_url="https://x/", body="") + + +# ---- 只读操作的「执行中掉登录 → 重登一次 → 重跑」---- +# +# 用最小替身顶掉 Playwright（_FakeContext/_FakeOrderPage）与 AuthSession +# （_FakeAuthSession），测的是 SiteInteractor 自己的重试编排：什么时候判掉登录、 +# 重登几次、重登失败怎么收场。页面解析由上面那批 _parse_* 纯函数测试覆盖。 + + +class _FakeOrderPage: + """只实现 goto/content/url/close 的 page 替身；每次 goto 按脚本换一份响应""" + + def __init__(self, responses: list[tuple[str, str]]): + # responses: [(final_url, html), ...]，按 goto 次序消费，用尽后重复最后一份 + self._responses = responses + self.goto_urls: list[str] = [] + self.url = "" + self._html = "" + self.closed = False + + async def goto(self, url: str, **kwargs: Any) -> None: + self.goto_urls.append(url) + idx = min(len(self.goto_urls) - 1, len(self._responses) - 1) + self.url, self._html = self._responses[idx] + + async def wait_for_timeout(self, ms: int) -> None: + return None + + async def content(self) -> str: + return self._html + + async def close(self) -> None: + self.closed = True + + +class _FakeContext: + """new_page() 按脚本发页面；页面用尽则复用最后一个""" + + def __init__(self, pages: list[_FakeOrderPage]): + self._pages = pages + self.handed_out: list[_FakeOrderPage] = [] + + async def new_page(self) -> _FakeOrderPage: + page = self._pages[min(len(self.handed_out), len(self._pages) - 1)] + self.handed_out.append(page) + return page + + +class _FakeAuthSession: + """记录 require_logged_in / try_relogin 调用次数的 AuthSession 替身""" + + def __init__(self, *, relogin_ok: bool = True): + self.relogin_ok = relogin_ok + self.require_calls = 0 + self.relogin_calls = 0 + + async def require_logged_in(self, site: str) -> None: + self.require_calls += 1 + + async def try_relogin(self, site: str) -> bool: + self.relogin_calls += 1 + return self.relogin_ok + + +_SSO_URL = "https://login.account.rakuten.com/sso/authorize?client_id=x" +_ORDER_DETAIL_LANDED_URL = "https://order.my.rakuten.co.jp/purchase-history/?order_number=x" +_ORDER_LIST_LANDED_URL = "https://order.my.rakuten.co.jp/purchase-history/order-list" + + +def _build_site(tmp_path: Path, context: _FakeContext, auth: _FakeAuthSession) -> SiteInteractor: + """装配一个不依赖 Playwright 的 SiteInteractor + + auth_state_dir 指向空目录：_refresh_context_if_stale 见不到 storage_state + 文件就直接返回，不会试图重建 context（重建需要真实 browser）。 + """ + from app.shared.config import Settings + + site = SiteInteractor( + auth_session=auth, # type: ignore[arg-type] + settings=Settings(auth_state_dir=str(tmp_path / "auth"), evidence_dir=str(tmp_path)), + ) + site._context = context # type: ignore[assignment] + return site + + +async def test_check_order_status_relogins_and_retries_when_kicked_to_sso(tmp_path): + """详情页被踢到 SSO → 重登一次 → 重跑拿到真实快照""" + pages = [ + _FakeOrderPage([(_SSO_URL, "ログイン")]), + _FakeOrderPage([(_ORDER_DETAIL_LANDED_URL, _stepper_html(active_stage="出荷"))]), + ] + auth = _FakeAuthSession(relogin_ok=True) + site = _build_site(tmp_path, _FakeContext(pages), auth) + + snapshot = await site.check_order_status(_REAL_ORDER_ID) + + assert snapshot.found is True + assert snapshot.order_state == OrderState.SHIPPED + assert auth.relogin_calls == 1 + assert auth.require_calls == 2 # 每次 attempt 前都过一遍前置检查 + assert all(p.closed for p in pages) # 两次 attempt 的 page 都关掉了 + + +async def test_check_order_status_raises_not_logged_in_when_relogin_fails(tmp_path): + """重登失败 → 抛 NotLoggedInError，而不是把「被踢走」伪装成 found=False""" + page = _FakeOrderPage([(_SSO_URL, "ログイン")]) + auth = _FakeAuthSession(relogin_ok=False) + site = _build_site(tmp_path, _FakeContext([page]), auth) + + with pytest.raises(NotLoggedInError): + await site.check_order_status(_REAL_ORDER_ID) + assert auth.relogin_calls == 1 + + +async def test_check_order_status_raises_not_logged_in_when_still_kicked_after_relogin(tmp_path): + """重登「成功」了但页面还是被踢走 → 只重试一次就抛错，不无限重登""" + page = _FakeOrderPage([(_SSO_URL, "ログイン")]) + auth = _FakeAuthSession(relogin_ok=True) + site = _build_site(tmp_path, _FakeContext([page]), auth) + + with pytest.raises(NotLoggedInError): + await site.check_order_status(_REAL_ORDER_ID) + assert auth.relogin_calls == 1 + assert len(page.goto_urls) == 2 # 只跑了两轮 + + +async def test_check_order_status_order_not_yet_visible_is_not_treated_as_logged_out(tmp_path): + """订单号还没反映到详情页（站点自己说要等 10 分钟）不是掉登录：不重登，返回 found=False""" + page = _FakeOrderPage([ + (_ORDER_DETAIL_LANDED_URL, "ご注文の反映に10分ほどかかります") + ]) + auth = _FakeAuthSession(relogin_ok=True) + site = _build_site(tmp_path, _FakeContext([page]), auth) + + snapshot = await site.check_order_status(_REAL_ORDER_ID) + + assert snapshot.found is False + assert auth.relogin_calls == 0 + + +async def test_list_recent_orders_relogins_and_retries_from_first_page(tmp_path): + """列表页第一页拿不到结构化数据且像掉登录 → 重登后从 page 1 重跑""" + logged_out_page = _FakeOrderPage([(_SSO_URL, "ログイン")]) + good_html = _wrap_state(_real_order_list_state()) + good_page = _FakeOrderPage([(_ORDER_LIST_LANDED_URL, good_html)]) + auth = _FakeAuthSession(relogin_ok=True) + site = _build_site(tmp_path, _FakeContext([logged_out_page, good_page]), auth) + + window = await site.list_recent_orders(since=datetime(2026, 8, 1, tzinfo=timezone.utc)) + + assert [e.order_number for e in window.entries] == [_REAL_ORDER_ID] + assert window.window_fully_covered is True + assert auth.relogin_calls == 1 + # 重跑确实是从第一页开始，不是接着上次的页码 + assert good_page.goto_urls[0].endswith("/order-list") + + +async def test_list_recent_orders_page_structure_change_is_not_treated_as_logged_out(tmp_path): + """第一页拿不到 ph-list 但没有任何未登录信号（页面改版）→ 不重登，按原路径转「没覆盖完」""" + page = _FakeOrderPage([(_ORDER_LIST_LANDED_URL, "改版了")]) + auth = _FakeAuthSession(relogin_ok=True) + site = _build_site(tmp_path, _FakeContext([page]), auth) + + window = await site.list_recent_orders(since=datetime(2026, 8, 1, tzinfo=timezone.utc)) + + assert window.entries == [] + assert window.window_fully_covered is False + assert auth.relogin_calls == 0 + + +async def test_read_retry_does_not_swallow_navigation_failure(tmp_path): + """导航失败仍是 OrderOperationError，不会被重登重试路径吃掉""" + + class _FailingPage(_FakeOrderPage): + async def goto(self, url: str, **kwargs: Any) -> None: + raise RuntimeError("net::ERR_TIMED_OUT") + + page = _FailingPage([(_ORDER_LIST_LANDED_URL, "")]) + auth = _FakeAuthSession(relogin_ok=True) + site = _build_site(tmp_path, _FakeContext([page]), auth) + + with pytest.raises(OrderOperationError): + await site.list_recent_orders(since=datetime(2026, 8, 1, tzinfo=timezone.utc)) + assert auth.relogin_calls == 0 + + # ---- _parse_order_list：verify_on_site 恢复核对用，2026-08-13 用真实账号跑 # order.my.rakuten.co.jp/purchase-history/order-list 验证过这套 __INITIAL_STATE__ # 结构（pageType="ph-list" 时 orderListData 直接是结构化 JSON，不需要正则抠 DOM）。