q792602257andClaude Opus 5
e2875f0c00
- 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) <noreply@anthropic.com>
376 lines
16 KiB
Python
376 lines
16 KiB
Python
"""登录态会话：持有已登录账号的 cookie，供加购与下单链路使用
|
|
|
|
与抓取链路（app/scraping 的 site_session / rakuma_session）不只是分模块，
|
|
而是分进程运行，原因：
|
|
|
|
- 抓取是**匿名**的，cookie 只用来过 Akamai 限速，丢了重新预热即可，无状态可言。
|
|
- 加购下单必须**带账号**，cookie 一旦失效需要重登——乐天登录有 reCAPTCHA 与设备
|
|
验证，全自动不可靠。本服务支持「失效时自动重登 + 撞验证码弹有头浏览器让人工接管」
|
|
的降级路径（依赖 account.yaml 与 relogin_enabled）。
|
|
- 这份登录态在整个系统里只能有一份。跟抓取同进程的话，抓取一扩容就会复制出 N 份
|
|
登录态与 N 个订单轮询，同一个账号被并发操作。
|
|
|
|
交易服务只管理乐天市场的购物车购买、支付与订单监控；ラクマ 抓取仍在抓取服务里
|
|
提供，但不进入交易链路。本模块只持有 rakuten 一个站点的登录态。
|
|
|
|
登录态来源是 Playwright 的 storage_state：
|
|
- 首次登录由 scripts/login.py 完成；
|
|
- 运行时失效且 relogin_enabled=true 时，本模块自动触发 login_runner.login_one()
|
|
重登（同账号串行，site 级 asyncio.Lock 防并发）。
|
|
账号密码不在本模块持有——try_relogin() 时从 account.yaml 临时读取，调用结束即
|
|
被 GC。日志只打 account.id，不打 username/password。
|
|
|
|
cookie 会被同时喂给两处：
|
|
- httpx 客户端 —— 加购这类纯表单提交走 HTTP 更快
|
|
- Playwright context —— 下单确认页有 JS 参与，必要时用浏览器走完
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import asyncio
|
|
import json
|
|
import logging
|
|
import time
|
|
from dataclasses import dataclass, field
|
|
from pathlib import Path
|
|
from typing import Any
|
|
|
|
import httpx
|
|
|
|
from app.shared.config import Settings
|
|
from app.shared.errors import NotLoggedInError, UpstreamRequestError
|
|
from app.trading.core import auth_site
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
|
|
@dataclass(slots=True)
|
|
class AuthStatus:
|
|
"""一个站点的登录态快照"""
|
|
|
|
site: str
|
|
state_file_exists: bool
|
|
logged_in: bool | None # None 表示尚未探测过
|
|
checked_at: float | None = None
|
|
detail: str = ""
|
|
|
|
def to_dict(self) -> dict[str, Any]:
|
|
return {
|
|
"site": self.site,
|
|
"state_file_exists": self.state_file_exists,
|
|
"logged_in": self.logged_in,
|
|
"checked_age_seconds": (
|
|
round(time.monotonic() - self.checked_at, 1) if self.checked_at else None
|
|
),
|
|
"detail": self.detail,
|
|
}
|
|
|
|
|
|
@dataclass(slots=True)
|
|
class _SiteAuth:
|
|
"""一个站点的登录通道：独立的 httpx 客户端与登录态缓存"""
|
|
|
|
name: str
|
|
client: httpx.AsyncClient
|
|
lock: asyncio.Lock = field(default_factory=asyncio.Lock)
|
|
logged_in: bool | None = None
|
|
checked_at: float | None = None
|
|
detail: str = ""
|
|
|
|
|
|
class AuthSession:
|
|
"""持有登录态的会话
|
|
|
|
职责边界：只负责「有没有登录态、cookie 是什么、还有效吗」，
|
|
具体加购/下单的业务请求由各自的 client 组装后借这里的 HTTP 客户端发出。
|
|
"""
|
|
|
|
def __init__(self, settings: Settings):
|
|
self._settings = settings
|
|
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] = {}
|
|
|
|
# ---- 生命周期 ----
|
|
|
|
async def start(self) -> None:
|
|
"""为站点创建带登录 cookie 的 HTTP 客户端
|
|
|
|
登录态文件不存在时同样创建客户端（只是没有 cookie），这样 /health 与
|
|
/api/auth/status 能如实回报「未登录」，而不是整个服务起不来。
|
|
"""
|
|
for name, profile in auth_site.PROFILES.items():
|
|
if name in self._sites:
|
|
continue
|
|
client = httpx.AsyncClient(
|
|
headers=profile.headers(),
|
|
timeout=self._settings.request_timeout_seconds,
|
|
follow_redirects=True,
|
|
proxy=self._settings.httpx_proxy,
|
|
http2=True,
|
|
)
|
|
self._sites[name] = _SiteAuth(name=name, client=client)
|
|
loaded = self._load_cookies(name)
|
|
logger.info("登录通道已就绪：site=%s cookies=%s", name, loaded)
|
|
|
|
async def close(self) -> None:
|
|
for auth in self._sites.values():
|
|
try:
|
|
await auth.client.aclose()
|
|
except Exception:
|
|
logger.debug("关闭登录 HTTP 客户端失败：site=%s", auth.name, exc_info=True)
|
|
self._sites.clear()
|
|
|
|
# ---- 登录态文件 ----
|
|
|
|
def state_path(self, site: str) -> Path:
|
|
"""某站点 storage_state 文件的落盘路径"""
|
|
return self._settings.auth_state_path / auth_site.profile(site).state_filename
|
|
|
|
def _load_cookies(self, site: str) -> int:
|
|
"""把 storage_state 里的 cookie 灌进该站的 httpx 客户端，返回条数"""
|
|
path = self.state_path(site)
|
|
if not path.exists():
|
|
return 0
|
|
|
|
try:
|
|
state = json.loads(path.read_text(encoding="utf-8"))
|
|
except (OSError, json.JSONDecodeError) as exc:
|
|
logger.warning("登录态文件读取失败：site=%s path=%s err=%s", site, path, exc)
|
|
return 0
|
|
|
|
auth = self._sites[site]
|
|
auth.client.cookies.clear()
|
|
count = 0
|
|
for cookie in state.get("cookies", []):
|
|
name = cookie.get("name")
|
|
value = cookie.get("value")
|
|
if not name or value is None:
|
|
continue
|
|
auth.client.cookies.set(
|
|
name,
|
|
value,
|
|
domain=cookie.get("domain") or "",
|
|
path=cookie.get("path") or "/",
|
|
)
|
|
count += 1
|
|
return count
|
|
|
|
def reload(self, site: str) -> int:
|
|
"""重新从磁盘加载登录态（人工登录完成后调用），返回 cookie 条数"""
|
|
auth = self._sites.get(site)
|
|
if auth is None:
|
|
raise ValueError(f"未知站点：{site}")
|
|
count = self._load_cookies(site)
|
|
auth.logged_in = None
|
|
auth.checked_at = None
|
|
auth.detail = "已重新加载登录态，尚未探测"
|
|
logger.info("登录态已重新加载：site=%s cookies=%s", site, count)
|
|
return count
|
|
|
|
# ---- 登录态探测 ----
|
|
|
|
async def check(self, site: str) -> AuthStatus:
|
|
"""探测某站登录态是否仍然有效
|
|
|
|
每次都真实打一次请求——登录态过期没有可靠的本地判据（cookie 的 expires
|
|
与服务端会话不是一回事），只能问站点。
|
|
"""
|
|
auth = self._require_site(site)
|
|
async with auth.lock:
|
|
try:
|
|
if site != "rakuten":
|
|
raise ValueError(f"未知站点：{site}")
|
|
logged_in, detail = await self._check_rakuten(auth)
|
|
except httpx.HTTPError as exc:
|
|
raise UpstreamRequestError(
|
|
f"登录态探测请求失败：site={site} err={type(exc).__name__}: {exc}"
|
|
) from exc
|
|
|
|
auth.logged_in = logged_in
|
|
auth.checked_at = time.monotonic()
|
|
auth.detail = detail
|
|
logger.info("登录态探测：site=%s logged_in=%s detail=%s", site, logged_in, detail)
|
|
return self.status(site)
|
|
|
|
async def _check_rakuten(self, auth: _SiteAuth) -> tuple[bool, str]:
|
|
"""购物车页含「現在ログインしていません」即未登录"""
|
|
response = await auth.client.get(auth_site.PROFILES["rakuten"].probe_url)
|
|
if response.status_code >= 400:
|
|
return False, f"购物车页返回 status {response.status_code}"
|
|
if not auth_site.is_logged_in(
|
|
"rakuten", final_url=str(response.url), body=response.text
|
|
):
|
|
return False, "购物车页显示未登录"
|
|
return True, "购物车页未出现未登录标记"
|
|
|
|
async def require_logged_in(self, site: str) -> None:
|
|
"""确保某站处于登录态，否则抛错
|
|
|
|
加购与下单前的统一入口。失效时的处理顺序：
|
|
1. 检测到未登录 → 若 relogin_enabled，尝试 try_relogin(site)
|
|
2. 重登成功 → 重新 check 一次，登录态转好即放行
|
|
3. 重登失败 / 未启用 / account.yaml 缺失 → 抛 NotLoggedInError 让 worker 转 needs_human
|
|
|
|
**写**操作（加购 / 提交订单 / 付款）只在这一层（动作前置检查）触发重登；
|
|
执行过程中失效不重试，避免脏状态（cart 已提交但响应后 cookie 失效等场景）。
|
|
**只读**操作（订单列表 / 订单详情查询）另有一层执行中重试，见
|
|
`site_interact.SiteInteractor._read_with_relogin_retry`——重跑一次查询没有
|
|
副作用，这条边界是刻意区分的，不要把它推广到写操作上。
|
|
"""
|
|
status = await self.check(site)
|
|
if status.logged_in:
|
|
return
|
|
|
|
# 尝试自动重登（relogin_enabled=False 或无 account.yaml 时返回 False）
|
|
if await self.try_relogin(site):
|
|
status = await self.check(site)
|
|
if status.logged_in:
|
|
return
|
|
detail = f"重登后仍判定未登录：{status.detail}"
|
|
else:
|
|
detail = status.detail
|
|
raise NotLoggedInError(site=site, detail=detail)
|
|
|
|
# ---- 自动重登 ----
|
|
|
|
async def try_relogin(self, site: str) -> bool:
|
|
"""触发自动重登：读 account.yaml 默认账号 → 跑 login_one → reload cookie
|
|
|
|
- relogin_enabled=False 直接返回 False（不抛错，让调用方走原降级路径）
|
|
- account.yaml 缺失或该站没账号 → 返回 False，仅记日志
|
|
- 同 site 串行（asyncio.Lock），并发请求只跑一次登录流程
|
|
- 账号密码不进日志：只打 account.id；login_runner 内部完成填表
|
|
|
|
成功返回前已完成 self.reload(site)，httpx 客户端拿到新 cookie；
|
|
Playwright 侧（SiteInteractor）通过 storage_state 文件 mtime 自检刷新。
|
|
"""
|
|
if not self._settings.relogin_enabled:
|
|
return False
|
|
|
|
# site 级锁：同账号同 user_data_dir，并发重登会撞锁
|
|
lock = self._relogin_locks.setdefault(site, asyncio.Lock())
|
|
epoch_before_wait = self._relogin_epochs.get(site, 0)
|
|
async with lock:
|
|
# 在锁上等过、且期间有人跑完了一轮重登 → 大概率不需要再登一次。
|
|
# 这里必须**真探测**而不是读 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:
|
|
from app.trading.services import login_runner
|
|
accounts_by_site = login_runner.load_accounts()
|
|
except FileNotFoundError as exc:
|
|
logger.warning("自动重登跳过：%s", exc)
|
|
return False
|
|
except ValueError as exc:
|
|
logger.warning("自动重登跳过：account.yaml 格式错误：%s", exc)
|
|
return False
|
|
|
|
account = login_runner.default_account_for(site, accounts_by_site)
|
|
if account is None:
|
|
logger.warning("自动重登跳过：account.yaml 没有 %s 的账号", site)
|
|
return False
|
|
|
|
logger.info(
|
|
"自动重登启动：site=%s account_id=%s timeout=%ss",
|
|
site, account.id, self._settings.relogin_timeout_seconds,
|
|
)
|
|
try:
|
|
ok = await login_runner.login_one(
|
|
account,
|
|
self._settings,
|
|
timeout_seconds=self._settings.relogin_timeout_seconds,
|
|
progress=None, # 用 logger，不打到 stdout
|
|
)
|
|
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)
|
|
return False
|
|
|
|
# 重登成功：reload cookie 到 httpx 客户端
|
|
count = self.reload(site)
|
|
logger.info(
|
|
"自动重登成功：site=%s account_id=%s cookies=%s",
|
|
site, account.id, count,
|
|
)
|
|
return True
|
|
|
|
# ---- 对外访问 ----
|
|
|
|
@property
|
|
def sites(self) -> tuple[str, ...]:
|
|
"""已初始化的站点名"""
|
|
return tuple(self._sites)
|
|
|
|
def client(self, site: str) -> httpx.AsyncClient:
|
|
"""取该站带登录 cookie 的 HTTP 客户端"""
|
|
return self._require_site(site).client
|
|
|
|
def cookies_for_browser(self, site: str) -> list[dict[str, Any]]:
|
|
"""导出 cookie 供 Playwright context 使用"""
|
|
path = self.state_path(site)
|
|
if not path.exists():
|
|
return []
|
|
try:
|
|
state = json.loads(path.read_text(encoding="utf-8"))
|
|
except (OSError, json.JSONDecodeError):
|
|
return []
|
|
return list(state.get("cookies", []))
|
|
|
|
def status(self, site: str) -> AuthStatus:
|
|
"""该站登录态快照（不触发探测，用缓存结果）"""
|
|
auth = self._require_site(site)
|
|
return AuthStatus(
|
|
site=site,
|
|
state_file_exists=self.state_path(site).exists(),
|
|
logged_in=auth.logged_in,
|
|
checked_at=auth.checked_at,
|
|
detail=auth.detail,
|
|
)
|
|
|
|
def status_all(self) -> dict[str, dict[str, Any]]:
|
|
"""登录态快照，供健康检查展示"""
|
|
return {name: self.status(name).to_dict() for name in self._sites}
|
|
|
|
def _require_site(self, site: str) -> _SiteAuth:
|
|
auth = self._sites.get(site)
|
|
if auth is None:
|
|
raise ValueError(f"未知站点或登录会话未初始化：{site}")
|
|
return auth
|