This commit is contained in:
2026-07-27 21:29:36 +08:00
parent 18cf7ae079
commit 6247d68fb5
10 changed files with 849 additions and 128 deletions
+5
View File
@@ -12,4 +12,9 @@ openapi.json
# 登录态目录:含可冒充账号的 cookie,绝不可提交 # 登录态目录:含可冒充账号的 cookie,绝不可提交
.auth/ .auth/
# 自动登录凭据与浏览器持久化目录
# account.yaml 含明文账号密码;.browser-data/ 含浏览器历史/localStorage/cookie
account.yaml
.browser-data/
.claude .claude
+40
View File
@@ -0,0 +1,40 @@
# 自动登录账号配置
#
# 复制为 account.yaml(已在 .gitignore 内)后填入真实凭据。
# account.yaml 不入库,含明文密码,绝不可提交。
#
# 字段说明:
# id (必填) 本地标识,命令行 --account 用它选择账号
# username (必填) 登录用户名(邮箱或会员 ID)
# password (必填) 登录密码明文。仅在本文件与站点之间流转,不入代码/日志/记忆
# user_data_dir (可选) Playwright 持久化浏览器目录。相对项目根或绝对路径。
# 不填则默认 .browser-data/<site>-<id>
# state_filename(可选) storage_state 落盘文件名,放在 settings.auth_state_dir。
# 不填则按 id 派生;数组首项若也没填,回退到 <site>_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
+10
View File
@@ -109,6 +109,16 @@ class Settings(BaseSettings):
# 页面改版导致买到远超预期的订单。设为 0 表示不设上限(不建议)。 # 页面改版导致买到远超预期的订单。设为 0 表示不设上限(不建议)。
order_max_total_yen: int = 30000 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 使用)---- # ---- 下单任务网关(仅网关进程 app.gateway.main 使用)----
# 网关的 SQLite 文件路径(相对项目根目录)。任务队列与状态镜像都在这里, # 网关的 SQLite 文件路径(相对项目根目录)。任务队列与状态镜像都在这里,
# 部署时务必放在持久化卷上,丢了等于丢了一批下单任务。 # 部署时务必放在持久化卷上,丢了等于丢了一批下单任务。
+97 -9
View File
@@ -4,14 +4,18 @@
而是分进程运行,原因: 而是分进程运行,原因:
- 抓取是**匿名**的,cookie 只用来过 Akamai 限速,丢了重新预热即可,无状态可言。 - 抓取是**匿名**的,cookie 只用来过 Akamai 限速,丢了重新预热即可,无状态可言。
- 加购下单必须**带账号**,cookie 一旦失效不能自动恢复——乐天与 ラクマ 登录都有 - 加购下单必须**带账号**,cookie 一旦失效需要重登——乐天与 ラクマ 登录都有
reCAPTCHA 与设备验证,只能由人重新登录一次。因此这里的失效处理是「明确报错让 reCAPTCHA 与设备验证,全自动不可靠。本服务支持「失效时自动重登 + 撞验证码弹
上游停下」,而不是像抓取那样静默重试 有头浏览器让人工接管」的降级路径(依赖 account.yaml 与 relogin_enabled)
- 这份登录态在整个系统里只能有一份。跟抓取同进程的话,抓取一扩容就会复制出 N 份 - 这份登录态在整个系统里只能有一份。跟抓取同进程的话,抓取一扩容就会复制出 N 份
登录态与 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 会被同时喂给两处: cookie 会被同时喂给两处:
- httpx 客户端 —— 加购这类纯表单提交走 HTTP 更快 - httpx 客户端 —— 加购这类纯表单提交走 HTTP 更快
@@ -80,6 +84,8 @@ class AuthSession:
def __init__(self, settings: Settings): def __init__(self, settings: Settings):
self._settings = settings self._settings = settings
self._sites: dict[str, _SiteAuth] = {} 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: 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) status = await self.check(site)
if not status.logged_in: if status.logged_in:
raise NotLoggedInError(site=site, detail=status.detail) 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
# ---- 对外访问 ---- # ---- 对外访问 ----
+455
View File
@@ -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 解析:显式 > 首项默认 <site>_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
+46
View File
@@ -91,6 +91,9 @@ class SiteInteractor:
self._browser = None self._browser = None
self._context = None # playwright BrowserContext self._context = None # playwright BrowserContext
self._per_task_state = {} 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 状态启动," "登录态文件不存在:site_interactor 以无 cookie 状态启动,"
"加购请求会被站点拒认" "加购请求会被站点拒认"
) )
else:
self._state_mtime = state_path.stat().st_mtime
self._playwright = await async_playwright().start() self._playwright = await async_playwright().start()
self._browser = await self._playwright.chromium.launch( self._browser = await self._playwright.chromium.launch(
@@ -127,6 +132,41 @@ class SiteInteractor:
) )
logger.info("SiteInteractor 已就绪:storage_state=%s", storage_state or "(none)") 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: async def close(self) -> None:
"""关闭 context、browser、playwright,吞掉单个 close 异常""" """关闭 context、browser、playwright,吞掉单个 close 异常"""
for resource, name in ( for resource, name in (
@@ -171,6 +211,9 @@ class SiteInteractor:
await self._auth_session.require_logged_in("rakuten") await self._auth_session.require_logged_in("rakuten")
# 任务开始前刷新 context:AuthSession 可能刚自动重登过,storage_state 变了
await self._refresh_context_if_stale()
page = await self._context.new_page() page = await self._context.new_page()
try: try:
try: try:
@@ -261,6 +304,9 @@ class SiteInteractor:
""" """
await self._auth_session.require_logged_in("rakuten") 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, {}) per_task = self._per_task_state.get(task.task_id, {})
item_id = (task.intent or {}).get("item_id") or per_task.get("item_id") item_id = (task.intent or {}).get("item_id") or per_task.get("item_id")
if not item_id: if not item_id:
+1
View File
@@ -16,6 +16,7 @@ dependencies = [
"playwright>=1.61.0", "playwright>=1.61.0",
"pydantic>=2.0.0,<3.0.0", "pydantic>=2.0.0,<3.0.0",
"pydantic-settings>=2.4.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", "selectolax>=0.3.21,<1.0.0",
"uvicorn[standard]>=0.30.0,<1.0.0", "uvicorn[standard]>=0.30.0,<1.0.0",
] ]
+9 -116
View File
@@ -1,135 +1,28 @@
"""人工登录:起一个有头浏览器,由人完成登录,落盘 cookie 供服务复用 """自动登录 CLI 入口:按 account.yaml 填账号密码,撞验证码则人工接管
为什么必须人工:乐天与 ラクマ 的登录都带 reCAPTCHA 与设备验证(短信/邮箱 OTP) 实现已移到 app/trading/services/login_runner.py,本文件保留为命令行入口
自动填表的成功率既低又不稳定,还要在本地存明文密码。这里的取舍是——**密码只经过 让 AuthSession(运行时重登)与 CLI 共用同一份登录逻辑,避免漂移。
你和站点,不经过本服务**:脚本只负责把浏览器打开、等你登录完,然后把 cookie 存下来。
用法: 用法:
.venv/Scripts/python.exe scripts/login.py --site rakuten .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 rakuten --account trdian022
.venv/Scripts/python.exe scripts/login.py --site all .venv/Scripts/python.exe scripts/login.py --site all --all-accounts
浏览器窗口打开后手动完成登录,脚本会自动轮询登录态;检测到已登录即保存并退出。 凭据安全:account.yaml 含明文密码,已在 .gitignore 排除;账号密码只经过
也可以登录完成后回到终端按回车立即保存 account.yaml → 浏览器 → 站点,不入代码/日志/记忆/commit
产物落在 settings.auth_state_dir(默认 .auth/),内含可直接冒充账号的 cookie,
已在 .gitignore 里排除,不要提交、不要外传。
""" """
from __future__ import annotations from __future__ import annotations
import argparse
import asyncio import asyncio
import json
import sys import sys
from pathlib import Path from pathlib import Path
# 允许以 `python scripts/login.py` 直接运行 # 允许以 `python scripts/login.py` 直接运行
sys.path.insert(0, str(Path(__file__).resolve().parent.parent)) sys.path.insert(0, str(Path(__file__).resolve().parent.parent))
from app.shared.config import get_settings # noqa: E402 from app.trading.services.login_runner import cli_main # 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
if __name__ == "__main__": if __name__ == "__main__":
raise SystemExit(asyncio.run(main())) raise SystemExit(asyncio.run(cli_main()))
+184 -3
View File
@@ -1,10 +1,11 @@
"""登录态会话测试:cookie 加载、失效探测、重新加载 """登录态会话测试:cookie 加载、失效探测、重新加载、自动重登
全部用 httpx.MockTransport 拦截,不触达真实站点、不需要真实账号。 全部用 httpx.MockTransport 拦截,不触达真实站点、不需要真实账号。
登录态文件写在 tmp_path 下,不碰仓库里的 .auth/。 登录态文件写在 tmp_path 下,不碰仓库里的 .auth/。
""" """
from __future__ import annotations from __future__ import annotations
import asyncio
import json import json
from pathlib import Path from pathlib import Path
@@ -14,6 +15,7 @@ import pytest
from app.shared.config import Settings from app.shared.config import Settings
from app.shared.errors import NotLoggedInError, UpstreamRequestError from app.shared.errors import NotLoggedInError, UpstreamRequestError
from app.trading.core import auth_site 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 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): 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)) session = await build_session(settings, lambda r: httpx.Response(200, text=CART_LOGGED_OUT))
try: try:
with pytest.raises(NotLoggedInError) as excinfo: 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() 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()
# ---- 重新加载 ---- # ---- 重新加载 ----
Generated
+2
View File
@@ -813,6 +813,7 @@ dependencies = [
{ name = "playwright" }, { name = "playwright" },
{ name = "pydantic" }, { name = "pydantic" },
{ name = "pydantic-settings" }, { name = "pydantic-settings" },
{ name = "pyyaml" },
{ name = "selectolax" }, { name = "selectolax" },
{ name = "uvicorn", extra = ["standard"] }, { name = "uvicorn", extra = ["standard"] },
] ]
@@ -843,6 +844,7 @@ requires-dist = [
{ name = "pydantic-settings", specifier = ">=2.4.0,<3.0.0" }, { name = "pydantic-settings", specifier = ">=2.4.0,<3.0.0" },
{ name = "pytest", marker = "extra == 'dev'", specifier = ">=8.3.0,<9.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 = "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 = "selectolax", specifier = ">=0.3.21,<1.0.0" },
{ name = "uvicorn", extras = ["standard"], specifier = ">=0.30.0,<1.0.0" }, { name = "uvicorn", extras = ["standard"], specifier = ">=0.30.0,<1.0.0" },
] ]