q792602257andClaude Sonnet 5
5f7921b788
- 新卡代填改用真实 DOM 结构：卡号/有效期分别托管在 Rakuten PCI 代付 vault 的 跨域 iframe 里，此前按 autocomplete/name 猜的 selector 在主文档里根本找不到 元素；持卡人姓名字段改用「名義人」标签相对定位，不用 placeholder 示例文案 （TARO RAKUTEN 只是示例用户名，不是稳定标识） - 订单号正则修正 实体导致的分隔符匹配失败（2026-08-13 真实下单验证） - login_runner._is_logged_in 改用与 AuthSession 一致的 __INITIAL_STATE__ 判据， 修掉此前误判「已登录」导致保存无效 storage_state 的问题 - 新增 CheckoutBlockedError：站点风控拦截（session upgrade/3DS）转 needs_human， 不当普通失败重试 - account.yaml 支持 phone / payment.credit-card 字段 - data/evidence/（真实 PII）加入 .gitignore Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
508 lines
18 KiB
Python
508 lines
18 KiB
Python
"""账号密码自动登录的运行时核心，供 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 CreditCard:
|
|
"""account.yaml 中 payment.credit-card 的结构化形式"""
|
|
|
|
number: str
|
|
month: str
|
|
year: str
|
|
name: str
|
|
|
|
|
|
@dataclass(slots=True)
|
|
class Account:
|
|
"""account.yaml 中的一条账号记录"""
|
|
|
|
site: str
|
|
id: str
|
|
username: str
|
|
password: str
|
|
user_data_dir: Path
|
|
state_filename: str
|
|
default: bool
|
|
# 电话号码：可选。仅用于 checkout 流程里「账号缺电话号码」时的补录步骤
|
|
# （site_interact.py::_complete_phone_registration），登录本身不需要它。
|
|
phone: str | None = None
|
|
# 信用卡：可选。仅用于 checkout 流程里支付方式页「账号没有已保存卡」时的
|
|
# 新卡代填步骤（site_interact.py::_fill_new_card_form）。该表单选择器从未
|
|
# 见过真实 HTML，代填后不会自动提交，须人工核对再点提交，详见该方法文档字符串。
|
|
credit_card: CreditCard | None = None
|
|
|
|
|
|
# ---- 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"])
|
|
payment = rec.get("payment") or {}
|
|
cc_raw = payment.get("credit-card") if isinstance(payment, dict) else None
|
|
credit_card = None
|
|
if cc_raw:
|
|
if not isinstance(cc_raw, dict):
|
|
raise ValueError(
|
|
f"account.yaml: {site}[{idx}].payment.credit-card 应为字典"
|
|
)
|
|
for required in ("card-no", "month", "year", "name"):
|
|
if not cc_raw.get(required):
|
|
raise ValueError(
|
|
f"account.yaml: {site}[{idx}].payment.credit-card.{required} 必填"
|
|
)
|
|
credit_card = CreditCard(
|
|
number=str(cc_raw["card-no"]),
|
|
month=str(cc_raw["month"]),
|
|
year=str(cc_raw["year"]),
|
|
name=str(cc_raw["name"]),
|
|
)
|
|
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)),
|
|
phone=str(rec["phone"]) if rec.get("phone") else None,
|
|
credit_card=credit_card,
|
|
)
|
|
)
|
|
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.probe_url（购物车页）读 `__INITIAL_STATE__.user.isLoggedIn`
|
|
|
|
`site` 目前恒为 `"rakuten"`。此前用「落地 URL 是否被踢到 SSO」判断
|
|
profile.login_url（myrakuten）：实测发现未登录时该页会落到 my.rakuten.co.jp
|
|
而不重定向到 login.account.rakuten.com，导致误判为「已登录」，
|
|
login_one() 因此跳过填表、保存一份只有 Akamai/负载均衡 cookie（无真实会话）
|
|
的 storage_state——看似登录成功，实际购物车页仍判未登录。
|
|
改用与 AuthSession._check_rakuten / site_interact.py 一致的判据（购物车页，
|
|
Playwright 渲染后取 `__INITIAL_STATE__.user.isLoggedIn`），三处判据不再漂移。
|
|
"""
|
|
if site != "rakuten":
|
|
raise ValueError(f"未知站点：{site}")
|
|
profile = auth_site.profile(site)
|
|
await page.goto(profile.probe_url, wait_until="domcontentloaded", timeout=60_000)
|
|
try:
|
|
await page.wait_for_function(
|
|
"() => window.__INITIAL_STATE__ !== undefined", timeout=8_000
|
|
)
|
|
except Exception:
|
|
pass # 未渲染出来时按 isLoggedIn=false 处理（下面 evaluate 拿到 None）
|
|
state = await page.evaluate(
|
|
"() => (window.__INITIAL_STATE__ && window.__INITIAL_STATE__.user) "
|
|
"? window.__INITIAL_STATE__.user.isLoggedIn : null"
|
|
)
|
|
return bool(state)
|
|
|
|
|
|
# ---- 自动填表（启发式选择器）----
|
|
|
|
|
|
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()
|
|
|
|
# 已经登录则直接保存（_is_logged_in 自己打购物车页判定，不依赖 login_url 落地）
|
|
if await _is_logged_in(page, account.site):
|
|
log(f"已处于登录态，跳过填表：{account.id}")
|
|
else:
|
|
await page.goto(profile.login_url, wait_until="domcontentloaded", timeout=60_000)
|
|
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
|