diff --git a/.gitignore b/.gitignore index 2cbb5c6..34f249e 100644 --- a/.gitignore +++ b/.gitignore @@ -12,4 +12,9 @@ openapi.json # 登录态目录:含可冒充账号的 cookie,绝不可提交 .auth/ + +# 自动登录凭据与浏览器持久化目录 +# account.yaml 含明文账号密码;.browser-data/ 含浏览器历史/localStorage/cookie +account.yaml +.browser-data/ .claude diff --git a/account.yaml.example b/account.yaml.example new file mode 100644 index 0000000..937df8d --- /dev/null +++ b/account.yaml.example @@ -0,0 +1,40 @@ +# 自动登录账号配置 +# +# 复制为 account.yaml(已在 .gitignore 内)后填入真实凭据。 +# account.yaml 不入库,含明文密码,绝不可提交。 +# +# 字段说明: +# id (必填) 本地标识,命令行 --account 用它选择账号 +# username (必填) 登录用户名(邮箱或会员 ID) +# password (必填) 登录密码明文。仅在本文件与站点之间流转,不入代码/日志/记忆 +# user_data_dir (可选) Playwright 持久化浏览器目录。相对项目根或绝对路径。 +# 不填则默认 .browser-data/- +# state_filename(可选) storage_state 落盘文件名,放在 settings.auth_state_dir。 +# 不填则按 id 派生;数组首项若也没填,回退到 _state.json +# 以兼容现有 AuthSession/SiteInteractor +# default (可选) true 标记默认账号;--account 不指定时用此项;无标记用数组首项 +# +# 用法: +# .venv/Scripts/python.exe scripts/login.py --site rakuten +# # 用 default=true 或数组首项 +# .venv/Scripts/python.exe scripts/login.py --site rakuten --account trdian022 +# .venv/Scripts/python.exe scripts/login.py --site all --all-accounts +# # 依次登录每个站点所有账号 + +rakuten: + - id: primary + username: your-email@example.com + password: REPLACE_ME + # user_data_dir: .browser-data/rakuten-primary + # state_filename: rakuten_state.json + default: true + # 多账号示例: + # - id: secondary + # username: another@example.com + # password: REPLACE_ME + +rakuma: + - id: primary + username: your-email@example.com + password: REPLACE_ME + default: true diff --git a/app/shared/config.py b/app/shared/config.py index 2b51d75..c90f53c 100644 --- a/app/shared/config.py +++ b/app/shared/config.py @@ -109,6 +109,16 @@ class Settings(BaseSettings): # 页面改版导致买到远超预期的订单。设为 0 表示不设上限(不建议)。 order_max_total_yen: int = 30000 + # ---- 自动重登(仅交易服务,需要 account.yaml)---- + # 检测到登录态失效时,是否在 require_logged_in 内自动触发重登。需要项目根 + # 存在 account.yaml(含明文密码,已在 .gitignore 排除)。未配置或文件缺失 + # 时退化为原行为(抛 NotLoggedInError 让 worker 转 needs_human)。 + relogin_enabled: bool = True + # 自动重登(含人工 fallback 等待验证码)的最长总耗时。超时即判定失败。 + # worker 内触发时整个任务会被卡住这段时间,所以不宜过长;本地机无人值守时 + # 可以设小(如 60)尽快失败转 needs_human。 + relogin_timeout_seconds: int = 300 + # ---- 下单任务网关(仅网关进程 app.gateway.main 使用)---- # 网关的 SQLite 文件路径(相对项目根目录)。任务队列与状态镜像都在这里, # 部署时务必放在持久化卷上,丢了等于丢了一批下单任务。 diff --git a/app/trading/services/auth_session.py b/app/trading/services/auth_session.py index 33c4611..d699be5 100644 --- a/app/trading/services/auth_session.py +++ b/app/trading/services/auth_session.py @@ -4,14 +4,18 @@ 而是分进程运行,原因: - 抓取是**匿名**的,cookie 只用来过 Akamai 限速,丢了重新预热即可,无状态可言。 -- 加购下单必须**带账号**,cookie 一旦失效不能自动恢复——乐天与 ラクマ 登录都有 - reCAPTCHA 与设备验证,只能由人重新登录一次。因此这里的失效处理是「明确报错让 - 上游停下」,而不是像抓取那样静默重试。 +- 加购下单必须**带账号**,cookie 一旦失效需要重登——乐天与 ラクマ 登录都有 + reCAPTCHA 与设备验证,全自动不可靠。本服务支持「失效时自动重登 + 撞验证码弹 + 有头浏览器让人工接管」的降级路径(依赖 account.yaml 与 relogin_enabled)。 - 这份登录态在整个系统里只能有一份。跟抓取同进程的话,抓取一扩容就会复制出 N 份 登录态与 N 个订单轮询,同一个账号被并发操作。 -登录态来源是 Playwright 的 storage_state:由 `scripts/login.py` 起一个有头浏览器 -让人工登录一次后落盘,本服务只读取、不生产。账号密码不经过本服务,也不写日志。 +登录态来源是 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 更快 @@ -80,6 +84,8 @@ class AuthSession: def __init__(self, settings: Settings): self._settings = settings self._sites: dict[str, _SiteAuth] = {} + # 自动重登的 site 级互斥锁:同账号同时只能一个登录流程(user_data_dir 被锁) + self._relogin_locks: dict[str, asyncio.Lock] = {} # ---- 生命周期 ---- @@ -209,12 +215,94 @@ class AuthSession: async def require_logged_in(self, site: str) -> None: """确保某站处于登录态,否则抛错 - 加购与下单前的统一入口。登录态失效时不做任何自动恢复尝试——两站登录都需要 - 人工过验证码,只能让上游停下来重新跑一次 scripts/login.py。 + 加购与下单前的统一入口。失效时的处理顺序: + 1. 检测到未登录 → 若 relogin_enabled,尝试 try_relogin(site) + 2. 重登成功 → 重新 check 一次,登录态转好即放行 + 3. 重登失败 / 未启用 / account.yaml 缺失 → 抛 NotLoggedInError 让 worker 转 needs_human + + 重登只在这一层(任务前置检查)触发;任务执行过程中失效不重试, + 避免脏状态(cart 已提交但响应后 cookie 失效等场景)。 """ status = await self.check(site) - if not status.logged_in: - raise NotLoggedInError(site=site, detail=status.detail) + 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()) + async with lock: + # 拿锁后再 check 一次——可能别的协程刚重登过 + status = self.status(site) + if status.logged_in: + return True + + # 读 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 + + 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 # ---- 对外访问 ---- diff --git a/app/trading/services/login_runner.py b/app/trading/services/login_runner.py new file mode 100644 index 0000000..b913b8c --- /dev/null +++ b/app/trading/services/login_runner.py @@ -0,0 +1,455 @@ +"""账号密码自动登录的运行时核心,供 scripts/login.py(CLI)与 AuthSession(运行时重登)共用 + +设计要点: + +- **凭据安全**:Account 在内存中只存到 login_one() 返回;日志只打 account.id,绝不打 + username/password。account.yaml 由调用方负责(默认项目根,已在 .gitignore 排除)。 +- **持久化浏览器**:用 launch_persistent_context,user_data_dir 来自 account.yaml, + cookie/localStorage/缓存跨重启保留。 +- **人工 fallback**:自动填账号密码后撞 reCAPTCHA / OTP 时,浏览器保持有头模式, + 由人完成验证,脚本继续轮询登录态;超时则失败返回。 +- **登录态判据**:访问 profile.login_url 看落地 URL 是否被踢到 SSO。旧 marker + 「現在ログインしていません」在新版 SP cart SPA 上不可靠(详见 + project://jp-rakuten/checkout-flow-probe-findings §2),本模块不用。 +""" +from __future__ import annotations + +import asyncio +import logging +import sys +from collections.abc import Callable +from dataclasses import dataclass +from pathlib import Path +from typing import Any + +import yaml + +from app.shared.config import BASE_DIR, Settings +from app.trading.core import auth_site + +logger = logging.getLogger(__name__) + +# 自动填表后通常秒级;fallback 人工要给足过验证码、收短信的时间 +_DEFAULT_TIMEOUT_SECONDS = 600 +_POLL_INTERVAL_SECONDS = 5 + +_ACCOUNTS_FILE = BASE_DIR / "account.yaml" + + +@dataclass(slots=True) +class Account: + """account.yaml 中的一条账号记录""" + + site: str + id: str + username: str + password: str + user_data_dir: Path + state_filename: str + default: bool + + +# ---- YAML 读取与账号选择 ---- + + +def accounts_file_path() -> Path: + """account.yaml 默认在项目根""" + return _ACCOUNTS_FILE + + +def _resolve_user_data_dir(site: str, raw: dict[str, Any], account_id: str) -> Path: + """从 YAML 取 user_data_dir,未配置则回退默认""" + raw_dir = raw.get("user_data_dir") or f".browser-data/{site}-{account_id}" + p = Path(raw_dir) + if not p.is_absolute(): + p = BASE_DIR / p + return p + + +def _resolve_state_filename(site: str, raw: dict[str, Any], is_first: bool) -> str: + """state_filename 解析:显式 > 首项默认 _state.json(兼容 AuthSession)> 按 id 派生""" + explicit = raw.get("state_filename") + if explicit: + return str(explicit) + if is_first: + return f"{site}_state.json" + return f"{site}_state_{raw.get('id', 'account')}.json" + + +def load_accounts() -> dict[str, list[Account]]: + """读 account.yaml,按站点分组返回所有账号 + + 文件不存在时抛 FileNotFoundError,调用方决定如何降级。 + """ + path = accounts_file_path() + if not path.exists(): + raise FileNotFoundError( + f"未找到 {path}。请复制 account.yaml.example 为 account.yaml 并填入凭据。" + ) + + raw = yaml.safe_load(path.read_text(encoding="utf-8")) or {} + result: dict[str, list[Account]] = {} + for site in auth_site.SITES: + records = raw.get(site) or [] + if not isinstance(records, list): + raise ValueError(f"account.yaml: {site} 应为列表,收到 {type(records).__name__}") + accounts: list[Account] = [] + for idx, rec in enumerate(records): + if not isinstance(rec, dict): + raise ValueError(f"account.yaml: {site}[{idx}] 应为字典") + for required in ("id", "username", "password"): + if not rec.get(required): + raise ValueError( + f"account.yaml: {site}[{idx}].{required} 必填" + ) + account_id = str(rec["id"]) + accounts.append( + Account( + site=site, + id=account_id, + username=str(rec["username"]), + password=str(rec["password"]), + user_data_dir=_resolve_user_data_dir(site, rec, account_id), + state_filename=_resolve_state_filename(site, rec, is_first=(idx == 0)), + default=bool(rec.get("default", False)), + ) + ) + result[site] = accounts + return result + + +def select_accounts( + site: str, + accounts_by_site: dict[str, list[Account]], + *, + account_id: str | None, + all_accounts: bool, +) -> list[Account]: + """按 --account / --all-accounts 选择要登录的账号列表 + + `site="all"` 时跳过 account.yaml 未配置的站点;若所有站点都没配则报错。 + """ + if site == "all": + targets: list[Account] = [] + for s in auth_site.SITES: + pool = accounts_by_site.get(s, []) + if not pool: + continue + targets.extend( + select_accounts(s, accounts_by_site, account_id=account_id, all_accounts=all_accounts) + ) + if not targets: + raise ValueError("account.yaml 没有任何站点的账号记录") + return targets + + pool = accounts_by_site.get(site, []) + if not pool: + raise ValueError(f"account.yaml 没有 {site} 的账号记录") + if all_accounts: + return list(pool) + if account_id: + for acc in pool: + if acc.id == account_id: + return [acc] + raise ValueError(f"account.yaml 中 {site} 没有 id={account_id} 的账号") + # 默认:标 default 的;无标记用首项 + for acc in pool: + if acc.default: + return [acc] + return [pool[0]] + + +def default_account_for(site: str, accounts_by_site: dict[str, list[Account]]) -> Account | None: + """取某站的默认账号(标 default 或首项),用于自动重登。无配置返回 None""" + pool = accounts_by_site.get(site, []) + if not pool: + return None + for acc in pool: + if acc.default: + return acc + return pool[0] + + +# ---- 登录态判定(URL 判据,不依赖旧 marker)---- + + +async def _is_logged_in(page, site: str) -> bool: + """访问 profile.login_url 看落地是否被踢到 SSO + + - rakuten:未登录会被重定向到 login.account.rakuten.com 或 /login 路径 + - rakuma:未登录会被重定向到 /users/sign_in + """ + profile = auth_site.profile(site) + await page.goto(profile.login_url, wait_until="domcontentloaded", timeout=60_000) + final_url = page.url + if site == "rakuten": + return "login.account.rakuten.com" not in final_url and "/login" not in final_url + return auth_site.RAKUMA_SIGN_IN_PATH not in final_url + + +# ---- 自动填表(启发式选择器)---- + + +async def _try_autofill(page, account: Account) -> bool: + """启发式填账号密码并提交 + + Returns: + True 表示已尝试自动填表(无论是否撞验证码); + False 表示页面没有密码框(已经登录或落地异常),交给调用方判定。 + """ + try: + await page.wait_for_selector( + "input[type='password']", state="visible", timeout=10_000 + ) + except Exception: + return False + + # 用户名框:按常见命名启发式找 + username_selectors = [ + "input[type='email']", + "input[name='username']", + "input[name='user_id']", + "input[name*='email' i]", + "input[name*='userid' i]", + "input[name*='login' i]", + "input[autocomplete='username']", + "input[type='text']:visible", + "input[type='tel']:visible", + ] + username_input = None + for sel in username_selectors: + username_input = await page.query_selector(sel) + if username_input and await username_input.is_visible(): + break + username_input = None + if username_input: + await username_input.fill(account.username) + + pwd = await page.query_selector("input[type='password']:visible") + if pwd: + await pwd.fill(account.password) + + # 提交按钮:按文案与类型启发式 + submit_selectors = [ + "button[type='submit']:visible", + "input[type='submit']:visible", + "button:has-text('ログイン')", + "button:has-text('Sign in')", + "button:has-text('Login')", + "[data-role='submit']:visible", + ] + for sel in submit_selectors: + btn = await page.query_selector(sel) + if btn and await btn.is_visible(): + try: + await btn.click() + except Exception: + continue + break + return True + + +async def _detect_human_challenge(page) -> str | None: + """检测页面是否出现需要人工处理的验证(reCAPTCHA / OTP / 设备验证)""" + if await page.query_selector("iframe[title*='reCAPTCHA' i], iframe[src*='recaptcha']"): + return "reCAPTCHA" + if await page.query_selector("iframe[src*='hcaptcha']"): + return "hCaptcha" + if await page.query_selector( + "input[name*='otp' i], input[name*='code' i], input[autocomplete='one-time-code']" + ): + return "短信/邮箱 OTP" + if await page.query_selector( + "input[name*='verification' i], input[name*='captcha' i]" + ): + return "设备验证/图形验证码" + return None + + +# ---- 单账号登录主流程 ---- + + +async def login_one( + account: Account, + settings: Settings, + *, + timeout_seconds: int = _DEFAULT_TIMEOUT_SECONDS, + progress: Callable[[str], None] | None = None, +) -> bool: + """对一个账号执行自动登录,成功则保存 storage_state + + Args: + account: 账号记录(含明文凭据,调用方负责保密) + settings: 全局配置(取 auth_state_path / browser_channel / playwright_proxy) + timeout_seconds: 整个登录流程的最长等待(含人工 fallback) + progress: 可选的进度回调,默认用 logger.info;CLI 传 print + + Returns: + True 表示登录成功且 storage_state 已落盘;False 表示超时或失败 + """ + from playwright.async_api import async_playwright + + log = progress or (lambda msg: logger.info(msg)) + profile = auth_site.profile(account.site) + state_path = settings.auth_state_path / account.state_filename + account.user_data_dir.mkdir(parents=True, exist_ok=True) + + log(f"=== {profile.label}({account.site}/{account.id})登录 ===") + log(f"持久化目录:{account.user_data_dir}") + log(f"登录入口:{profile.login_url}") + + async with async_playwright() as playwright: + context = await playwright.chromium.launch_persistent_context( + user_data_dir=str(account.user_data_dir), + headless=False, # 自动登录仍需有头:撞验证码要人工接管 + channel=settings.browser_channel or None, + proxy=settings.playwright_proxy, + user_agent=profile.user_agent, + locale="ja-JP", + timezone_id="Asia/Tokyo", + viewport=( + {"width": 390, "height": 844} + if profile.mobile + else {"width": 1440, "height": 900} + ), + is_mobile=profile.mobile, + has_touch=profile.mobile, + args=["--no-first-run", "--disable-blink-features=AutomationControlled"], + ) + try: + page = context.pages[0] if context.pages else await context.new_page() + await page.goto(profile.login_url, wait_until="domcontentloaded", timeout=60_000) + + # 已经登录则直接保存 + if await _is_logged_in(page, account.site): + log(f"已处于登录态,跳过填表:{account.id}") + else: + filled = await _try_autofill(page, account) + if not filled: + log("未找到登录表单且未登录,请在浏览器里手动完成登录") + else: + # 不打账号/密码值,只标识已提交 + log(f"已自动填写凭据并提交:account_id={account.id}") + + await page.wait_for_load_state("domcontentloaded", timeout=30_000) + + challenge = await _detect_human_challenge(page) + if challenge: + log( + f"检测到 {challenge}:请在浏览器窗口里人工完成验证。" + f"最长等待 {timeout_seconds // 60} 分钟。" + ) + + # 轮询登录态 + waited = 0 + ok = False + while waited < timeout_seconds: + try: + if await _is_logged_in(page, account.site): + ok = True + break + except Exception as exc: + log(f"探测中({waited}s):{type(exc).__name__}") + await asyncio.sleep(_POLL_INTERVAL_SECONDS) + waited += _POLL_INTERVAL_SECONDS + if waited % 30 == 0: + log(f"等待登录完成…({waited}s)") + + if not ok: + log(f"等待超时,未检测到登录态:{account.id}") + return False + + import json + state = await context.storage_state() + state_path.write_text( + json.dumps(state, ensure_ascii=False, indent=2), encoding="utf-8" + ) + log(f"登录成功,已保存 {len(state.get('cookies', []))} 条 cookie → {state_path}") + return True + finally: + await context.close() + + +# ---- CLI 入口(供 scripts/login.py 调用)---- + + +async def cli_main(argv: list[str] | None = None) -> int: + """命令行入口:argparse + 调用 login_one + + 返回进程退出码。stdout 用 print(给人看),不打账号密码值。 + """ + import argparse + + parser = argparse.ArgumentParser(description="按 account.yaml 自动登录乐天/ラクマ") + parser.add_argument( + "--site", + choices=[*auth_site.SITES, "all"], + default="all", + help="要登录的站点,默认 all", + ) + parser.add_argument( + "--account", + help="account.yaml 中的账号 id;不传则用 default=true 或首项", + ) + parser.add_argument( + "--all-accounts", + action="store_true", + help="登录该站点 account.yaml 里的所有账号(按顺序串行)", + ) + parser.add_argument( + "--log-level", + default="INFO", + choices=["DEBUG", "INFO", "WARNING", "ERROR"], + ) + args = parser.parse_args(argv) + + logging.basicConfig( + level=args.log_level, + format="{asctime} {levelname} {message}", + style="{", + ) + + try: + accounts_by_site = load_accounts() + except (FileNotFoundError, ValueError) as exc: + print(f"配置错误:{exc}", file=sys.stderr) + return 2 + + try: + targets = select_accounts( + args.site, + accounts_by_site, + account_id=args.account, + all_accounts=args.all_accounts, + ) + except ValueError as exc: + print(f"账号选择失败:{exc}", file=sys.stderr) + return 2 + + if not targets: + print("没有匹配的账号可登录", file=sys.stderr) + return 2 + + print( + f"将登录 {len(targets)} 个账号:" + + ", ".join(f"{a.site}/{a.id}" for a in targets) + ) + + settings = Settings() + results: dict[str, bool] = {} + for acc in targets: + key = f"{acc.site}/{acc.id}" + try: + results[key] = await login_one(acc, settings, progress=print) + except Exception as exc: + print(f"✗ {key} 登录异常:{type(exc).__name__}: {exc}", file=sys.stderr) + results[key] = False + + print("\n=== 结果 ===") + for key, ok in results.items(): + print(f" {key}: {'已登录' if ok else '失败'}") + + if all(results.values()): + print("\n登录态已就绪,可启动服务并用 POST /api/auth/status 复核。") + return 0 + return 1 diff --git a/app/trading/worker/site_interact.py b/app/trading/worker/site_interact.py index e8612a7..d18038f 100644 --- a/app/trading/worker/site_interact.py +++ b/app/trading/worker/site_interact.py @@ -91,6 +91,9 @@ class SiteInteractor: self._browser = None self._context = None # playwright BrowserContext self._per_task_state = {} + # 上次 build context 时读到的 storage_state 文件 mtime。 + # 用于在任务间检测 AuthSession 重登后产生的新 storage_state,触发 context 重建。 + self._state_mtime: float | None = None # ---- 生命周期 ---- @@ -109,6 +112,8 @@ class SiteInteractor: "登录态文件不存在:site_interactor 以无 cookie 状态启动," "加购请求会被站点拒认" ) + else: + self._state_mtime = state_path.stat().st_mtime self._playwright = await async_playwright().start() self._browser = await self._playwright.chromium.launch( @@ -127,6 +132,41 @@ class SiteInteractor: ) logger.info("SiteInteractor 已就绪:storage_state=%s", storage_state or "(none)") + async def _refresh_context_if_stale(self) -> None: + """检查 storage_state 文件 mtime,变化则重建 context + + AuthSession.try_relogin 成功后会重写 storage_state 文件。本 context 启动时 + 用快照式 storage_state 创建,cookie 不会自动同步——必须关掉旧 context、 + 用新文件重建。在 add_to_cart / verify_cart 开头各调一次,开销可接受 + (只在 mtime 变了才重建)。 + """ + state_path = self._settings.auth_state_path / auth_site.profile("rakuten").state_filename + if not state_path.exists(): + return + mtime = state_path.stat().st_mtime + if mtime == self._state_mtime: + return + + logger.info( + "storage_state 文件变化(mtime %s → %s),重建 context", + self._state_mtime, mtime, + ) + if self._context is not None: + try: + await self._context.close() + except Exception: + logger.debug("关闭旧 context 失败", exc_info=True) + self._context = await self._browser.new_context( + storage_state=str(state_path), + user_agent=auth_site.RAKUTEN_USER_AGENT, + locale="ja-JP", + timezone_id="Asia/Tokyo", + viewport={"width": 390, "height": 844}, + is_mobile=True, + has_touch=True, + ) + self._state_mtime = mtime + async def close(self) -> None: """关闭 context、browser、playwright,吞掉单个 close 异常""" for resource, name in ( @@ -171,6 +211,9 @@ class SiteInteractor: await self._auth_session.require_logged_in("rakuten") + # 任务开始前刷新 context:AuthSession 可能刚自动重登过,storage_state 变了 + await self._refresh_context_if_stale() + page = await self._context.new_page() try: try: @@ -261,6 +304,9 @@ class SiteInteractor: """ await self._auth_session.require_logged_in("rakuten") + # 同 add_to_cart:刷新 context 以反映可能的自动重登结果 + await self._refresh_context_if_stale() + per_task = self._per_task_state.get(task.task_id, {}) item_id = (task.intent or {}).get("item_id") or per_task.get("item_id") if not item_id: diff --git a/pyproject.toml b/pyproject.toml index ae6411e..09eea1c 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -16,6 +16,7 @@ dependencies = [ "playwright>=1.61.0", "pydantic>=2.0.0,<3.0.0", "pydantic-settings>=2.4.0,<3.0.0", + "pyyaml>=6.0,<7.0.0", "selectolax>=0.3.21,<1.0.0", "uvicorn[standard]>=0.30.0,<1.0.0", ] diff --git a/scripts/login.py b/scripts/login.py index a81d168..ea1ca6d 100644 --- a/scripts/login.py +++ b/scripts/login.py @@ -1,135 +1,28 @@ -"""人工登录:起一个有头浏览器,由人完成登录,落盘 cookie 供服务复用 +"""自动登录 CLI 入口:按 account.yaml 填账号密码,撞验证码则人工接管 -为什么必须人工:乐天与 ラクマ 的登录都带 reCAPTCHA 与设备验证(短信/邮箱 OTP), -自动填表的成功率既低又不稳定,还要在本地存明文密码。这里的取舍是——**密码只经过 -你和站点,不经过本服务**:脚本只负责把浏览器打开、等你登录完,然后把 cookie 存下来。 +实现已移到 app/trading/services/login_runner.py,本文件保留为命令行入口, +让 AuthSession(运行时重登)与 CLI 共用同一份登录逻辑,避免漂移。 用法: .venv/Scripts/python.exe scripts/login.py --site rakuten - .venv/Scripts/python.exe scripts/login.py --site rakuma - .venv/Scripts/python.exe scripts/login.py --site all + .venv/Scripts/python.exe scripts/login.py --site rakuten --account trdian022 + .venv/Scripts/python.exe scripts/login.py --site all --all-accounts -浏览器窗口打开后手动完成登录,脚本会自动轮询登录态;检测到已登录即保存并退出。 -也可以登录完成后回到终端按回车立即保存。 - -产物落在 settings.auth_state_dir(默认 .auth/),内含可直接冒充账号的 cookie, -已在 .gitignore 里排除,不要提交、不要外传。 +凭据安全:account.yaml 含明文密码,已在 .gitignore 排除;账号密码只经过 +account.yaml → 浏览器 → 站点,不入代码/日志/记忆/commit。 """ from __future__ import annotations -import argparse import asyncio -import json import sys from pathlib import Path # 允许以 `python scripts/login.py` 直接运行 sys.path.insert(0, str(Path(__file__).resolve().parent.parent)) -from app.shared.config import get_settings # noqa: E402 -from app.trading.core import auth_site # noqa: E402 - -# 轮询间隔与总时长:给足人工过验证码、收短信的时间 -_POLL_INTERVAL_SECONDS = 5 -_POLL_TIMEOUT_SECONDS = 600 - - -async def _is_logged_in(page, site: str) -> bool: - """在浏览器里探测登录态 - - 判据与 AuthSession 共用 auth_site.is_logged_in,只是这里喂浏览器页面内容、 - 那里喂 httpx 响应,避免两条链路对「算不算登录」各判各的。 - """ - await page.goto( - auth_site.profile(site).probe_url, wait_until="domcontentloaded", timeout=60_000 - ) - return auth_site.is_logged_in(site, final_url=page.url, body=await page.content()) - - -async def login(site: str) -> bool: - """打开浏览器让用户登录指定站点,成功则保存 storage_state""" - from playwright.async_api import async_playwright - - settings = get_settings() - profile = auth_site.profile(site) - state_path = settings.auth_state_path / profile.state_filename - - print(f"\n=== {profile.label}({site})登录 ===") - print(f"即将打开浏览器:{profile.login_url}") - print("请在浏览器窗口里完成登录(账号密码只在浏览器与站点之间传递,本脚本不读取)。") - print(f"登录完成后脚本会自动检测,最长等待 {_POLL_TIMEOUT_SECONDS // 60} 分钟。\n") - - async with async_playwright() as playwright: - browser = await playwright.chromium.launch( - headless=False, # 人工登录必须有头 - channel=settings.browser_channel or None, - proxy=settings.playwright_proxy, - args=["--no-first-run", "--disable-blink-features=AutomationControlled"], - ) - try: - context = await browser.new_context( - # UA 取自 auth_site:服务端后续用同一个 UA 发请求,换了可能触发 - # 站点的设备校验,让这次辛苦登来的 cookie 提前失效。 - user_agent=profile.user_agent, - locale="ja-JP", - timezone_id="Asia/Tokyo", - viewport=( - {"width": 390, "height": 844} - if profile.mobile - else {"width": 1440, "height": 900} - ), - is_mobile=profile.mobile, - has_touch=profile.mobile, - ) - page = await context.new_page() - await page.goto(profile.login_url, wait_until="domcontentloaded", timeout=60_000) - - waited = 0 - while waited < _POLL_TIMEOUT_SECONDS: - await asyncio.sleep(_POLL_INTERVAL_SECONDS) - waited += _POLL_INTERVAL_SECONDS - try: - if await _is_logged_in(page, site): - break - except Exception as exc: # 页面正在跳转时探测可能失败,继续等 - print(f" 探测中({waited}s):{type(exc).__name__}") - continue - print(f" 等待登录中…({waited}s)") - else: - print(f"✗ {profile.label} 等待超时,未检测到登录态。未保存。") - return False - - state = await context.storage_state() - state_path.write_text( - json.dumps(state, ensure_ascii=False, indent=2), encoding="utf-8" - ) - print(f"✓ {profile.label} 登录成功,已保存 {len(state.get('cookies', []))} 条 cookie") - print(f" → {state_path}") - return True - finally: - await browser.close() - - -async def main() -> int: - parser = argparse.ArgumentParser(description="人工登录并保存乐天/ラクマ 登录态") - parser.add_argument( - "--site", - choices=[*auth_site.SITES, "all"], - default="all", - help="要登录的站点,默认两站都登录", - ) - args = parser.parse_args() - - targets = list(auth_site.SITES) if args.site == "all" else [args.site] - results = {site: await login(site) for site in targets} - - print("\n=== 结果 ===") - for site, ok in results.items(): - print(f" {site}: {'已登录' if ok else '失败'}") - print("\n登录态已就绪,可启动服务并用 POST /api/auth/status 复核。") - return 0 if all(results.values()) else 1 +from app.trading.services.login_runner import cli_main # noqa: E402 if __name__ == "__main__": - raise SystemExit(asyncio.run(main())) + raise SystemExit(asyncio.run(cli_main())) diff --git a/tests/test_auth_session.py b/tests/test_auth_session.py index 201bda4..141cd09 100644 --- a/tests/test_auth_session.py +++ b/tests/test_auth_session.py @@ -1,10 +1,11 @@ -"""登录态会话测试:cookie 加载、失效探测、重新加载 +"""登录态会话测试:cookie 加载、失效探测、重新加载、自动重登 全部用 httpx.MockTransport 拦截,不触达真实站点、不需要真实账号。 登录态文件写在 tmp_path 下,不碰仓库里的 .auth/。 """ from __future__ import annotations +import asyncio import json from pathlib import Path @@ -14,6 +15,7 @@ import pytest from app.shared.config import Settings from app.shared.errors import NotLoggedInError, UpstreamRequestError from app.trading.core import auth_site +from app.trading.services import auth_session as auth_session_mod from app.trading.services.auth_session import AuthSession # 购物车页的两种形态:含未登录标记 = 未登录,不含 = 已登录 @@ -200,11 +202,11 @@ async def test_network_error_raises_upstream(tmp_path): async def test_require_logged_in_raises_when_logged_out(tmp_path): - """未登录时 require_logged_in 抛 5001,且标记为不可重试 + """relogin_enabled=False 时,未登录直接抛 5001,且标记为不可重试 登录需要人工过验证码,自动重试没有意义,必须让上游停下来。 """ - settings = make_settings(tmp_path) + settings = make_settings(tmp_path, relogin_enabled=False) session = await build_session(settings, lambda r: httpx.Response(200, text=CART_LOGGED_OUT)) try: with pytest.raises(NotLoggedInError) as excinfo: @@ -227,6 +229,185 @@ async def test_require_logged_in_passes_when_logged_in(tmp_path): await session.close() +# ---- 自动重登 ---- + + +def _patch_relogin(monkeypatch, *, return_value: bool, sleep_quick: bool = True): + """把 login_runner.login_one 替换成可控桩 + + login_one 在 AuthSession.try_relogin 里通过 from import 触发,所以 patch + 的对象要是 login_runner 模块上的 login_one。 + """ + calls: list[dict] = [] + + async def fake_login_one(account, settings, *, timeout_seconds=..., progress=None): + calls.append({ + "site": account.site, + "account_id": account.id, + "username": account.username, # 仅测试中校验不泄密给日志 + "timeout_seconds": timeout_seconds, + }) + if sleep_quick: + await asyncio.sleep(0) # 让协程调度 + return return_value + + # AuthSession.try_relogin 内部用 `from ... import login_runner` 形式 + from app.trading.services import login_runner + monkeypatch.setattr(login_runner, "login_one", fake_login_one) + return calls + + +def _patch_accounts_file(monkeypatch, tmp_path, *, has_rakuten: bool = True): + """让 login_runner 读到一份假的 account.yaml""" + accounts = {"rakuten": []} if has_rakuten else {} + if has_rakuten: + accounts["rakuten"].append({ + "id": "testacct", + "username": "u@example.com", + "password": "pw", + "user_data_dir": str(tmp_path / "ud"), + "state_filename": "rakuten_state.json", + "default": True, + }) + yaml_text = "".join( + f"{site}:\n" + "".join( + f" - id: {r['id']}\n username: {r['username']}\n password: {r['password']}\n" + f" user_data_dir: {r['user_data_dir']}\n state_filename: {r['state_filename']}\n" + f" default: true\n" + for r in recs + ) + for site, recs in accounts.items() + ) + from app.trading.services import login_runner + fake_path = tmp_path / "account.yaml" + fake_path.write_text(yaml_text, encoding="utf-8") + monkeypatch.setattr(login_runner, "accounts_file_path", lambda: fake_path) + + +async def test_require_logged_in_tries_relogin_and_passes(tmp_path, monkeypatch): + """relogin_enabled=True:失效 → login_one 成功 → check 通过 → 放行 + + login_one 模拟重登成功:落盘一份新 storage_state,并把 MockTransport 切到 + 「已登录」分支,让 require_logged_in 第二次 check 通过。 + """ + _patch_accounts_file(monkeypatch, tmp_path) + + state = {"logged_in": False} + + def handler(request: httpx.Request) -> httpx.Response: + if state["logged_in"]: + return httpx.Response(200, text=CART_LOGGED_IN) + return httpx.Response(200, text=CART_LOGGED_OUT) + + settings = make_settings(tmp_path, relogin_enabled=True, relogin_timeout_seconds=5) + session = await build_session(settings, handler) + + from app.trading.services import login_runner + + async def fake_login_one(account, s, *, timeout_seconds=300, progress=None): + # 模拟重登成功:落盘新 storage_state(让 reload 能读到),翻转 check 行为 + write_state( + settings, + "rakuten", + [{"name": "FRESH", "value": "xyz", "domain": ".rakuten.co.jp", "path": "/"}], + ) + state["logged_in"] = True + return True + + monkeypatch.setattr(login_runner, "login_one", fake_login_one) + + try: + await session.require_logged_in("rakuten") # 不应抛出 + assert session.status("rakuten").logged_in is True + # 重登后的 cookie 已灌进 client + names = {c.name for c in session.client("rakuten").cookies.jar} + assert "FRESH" in names + finally: + await session.close() + + +async def test_require_logged_in_raises_when_relogin_fails(tmp_path, monkeypatch): + """relogin_enabled=True 但 login_one 失败 → 抛 NotLoggedInError""" + _patch_accounts_file(monkeypatch, tmp_path) + _patch_relogin(monkeypatch, return_value=False) + + settings = make_settings(tmp_path, relogin_enabled=True, relogin_timeout_seconds=5) + session = await build_session(settings, lambda r: httpx.Response(200, text=CART_LOGGED_OUT)) + try: + with pytest.raises(NotLoggedInError): + await session.require_logged_in("rakuten") + finally: + await session.close() + + +async def test_require_logged_in_falls_back_when_no_accounts_file(tmp_path, monkeypatch): + """relogin_enabled=True 但 account.yaml 不存在 → try_relogin 返回 False,抛 NotLoggedInError""" + from app.trading.services import login_runner + monkeypatch.setattr(login_runner, "accounts_file_path", lambda: tmp_path / "missing.yaml") + + settings = make_settings(tmp_path, relogin_enabled=True) + session = await build_session(settings, lambda r: httpx.Response(200, text=CART_LOGGED_OUT)) + try: + with pytest.raises(NotLoggedInError): + await session.require_logged_in("rakuten") + finally: + await session.close() + + +async def test_try_relogin_is_disabled_when_flag_off(tmp_path): + """relogin_enabled=False:try_relogin 直接返回 False,不读 account.yaml""" + settings = make_settings(tmp_path, relogin_enabled=False) + session = await build_session(settings, lambda r: httpx.Response(200, text=CART_LOGGED_OUT)) + try: + ok = await session.try_relogin("rakuten") + assert ok is False + finally: + await session.close() + + +async def test_try_relogin_serializes_concurrent_calls_same_site(tmp_path, monkeypatch): + """同 site 并发触发只跑一次 login_one:靠 site 级锁串行化""" + _patch_accounts_file(monkeypatch, tmp_path) + + invocations = {"count": 0, "in_flight_max": 0, "current": 0} + from app.trading.services import login_runner + + async def counting_login_one(account, s, *, timeout_seconds=300, progress=None): + invocations["current"] += 1 + invocations["in_flight_max"] = max(invocations["in_flight_max"], invocations["current"]) + await asyncio.sleep(0.05) # 故意拉长,让并发请求有机会叠上来 + invocations["current"] -= 1 + invocations["count"] += 1 + # 落盘新 storage_state,让 reload 有东西可读 + write_state( + make_settings(tmp_path), # 用同一路径 + "rakuten", + [{"name": "X", "value": "1", "domain": ".rakuten.co.jp", "path": "/"}], + ) + return True + + monkeypatch.setattr(login_runner, "login_one", counting_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: + # 第一次状态为 logged_in=None,三个并发都进入 try_relogin + results = await asyncio.gather( + session.try_relogin("rakuten"), + session.try_relogin("rakuten"), + session.try_relogin("rakuten"), + ) + # login_one 至少被调一次(串行下后续可能因 status 已 logged_in 跳过) + assert invocations["count"] >= 1 + # 关键:login_one 永远没并发执行 + assert invocations["in_flight_max"] == 1 + # 结果都成功(要么真重登,要么拿到锁后发现已 logged_in) + assert all(results) + finally: + await session.close() + + # ---- 重新加载 ---- diff --git a/uv.lock b/uv.lock index ea4cbf7..5179176 100644 --- a/uv.lock +++ b/uv.lock @@ -813,6 +813,7 @@ dependencies = [ { name = "playwright" }, { name = "pydantic" }, { name = "pydantic-settings" }, + { name = "pyyaml" }, { name = "selectolax" }, { name = "uvicorn", extra = ["standard"] }, ] @@ -843,6 +844,7 @@ requires-dist = [ { name = "pydantic-settings", specifier = ">=2.4.0,<3.0.0" }, { name = "pytest", marker = "extra == 'dev'", specifier = ">=8.3.0,<9.0.0" }, { name = "pytest-asyncio", marker = "extra == 'dev'", specifier = ">=0.24.0,<1.0.0" }, + { name = "pyyaml", specifier = ">=6.0,<7.0.0" }, { name = "selectolax", specifier = ">=0.3.21,<1.0.0" }, { name = "uvicorn", extras = ["standard"], specifier = ">=0.30.0,<1.0.0" }, ]