q792602257
48c6f2a28f
- POST /api/orders 新增可选 callback_url（仅 http/https，其余 422） - 仅终结类事件各通知一次：terminal report 推入终态（succeeded/failed/ needs_human）、租约过期被 sweep 置 stale；中间态与终结后的监控上报不通知 - 幂等重发不更新既有任务的回调地址；投递 best-effort 单次尝试，失败只记日志 - CallbackNotifier 发后不管（create_task + 在途任务强引用），关停等在途发完； 生产路径按回调地址逐次构造客户端，满足统一出站代理策略（test_proxy.py） - tasks 表加 callback_url 列，GatewayDB.start() 内置迁移兼容既有库 - 新增 RAKUTEN_CALLBACK_TIMEOUT_SECONDS（默认 10）；文档补 §4.8； openapi.json 重新导出（gitignore 未跟踪）；新增 17 条测试，全量 492 通过
322 lines
16 KiB
Python
322 lines
16 KiB
Python
"""应用配置：通过环境变量和 .env 文件加载所有配置项
|
|
|
|
配置项统一使用 RAKUTEN_ 前缀，例如 RAKUTEN_APP_PORT=31107。
|
|
支持 .env 文件自动加载。
|
|
|
|
抓取、交易与下单任务网关是三个进程，但共用这一个 Settings 类：三方都要日志、
|
|
代理、超时与同一个 Bearer token，拆成三份配置只会让部署时多维护一套。下面按
|
|
「通用 / 仅抓取 / 仅交易 / 仅网关 / 仅交易 worker」分区标注，各进程只读自己那部分。
|
|
"""
|
|
import socket
|
|
from fnmatch import fnmatchcase
|
|
from functools import lru_cache
|
|
from pathlib import Path
|
|
from typing import Literal
|
|
from urllib.parse import quote, urlsplit
|
|
|
|
from pydantic import Field
|
|
from pydantic_settings import BaseSettings, SettingsConfigDict
|
|
|
|
BASE_DIR = Path(__file__).resolve().parent.parent.parent
|
|
|
|
# 乐天市场首页。这里不 import app.scraping —— shared 不能反向依赖两侧任何一方，
|
|
# 否则交易服务也会被迫加载整套抓取模块。
|
|
DEFAULT_HOME_URL = "https://www.rakuten.co.jp/"
|
|
|
|
|
|
class Settings(BaseSettings):
|
|
"""应用全局配置
|
|
|
|
所有配置项均可通过环境变量覆盖，前缀为 RAKUTEN_。
|
|
例如：RAKUTEN_APP_PORT=31107 对应 app_port 配置项。
|
|
"""
|
|
|
|
model_config = SettingsConfigDict(
|
|
env_file=str(BASE_DIR / ".env"),
|
|
env_file_encoding="utf-8",
|
|
env_prefix="RAKUTEN_",
|
|
extra="ignore",
|
|
)
|
|
|
|
# ---- 服务基本配置（通用）----
|
|
app_name: str = "Rakuten Scraper Service"
|
|
app_env: Literal["dev", "prod", "test"] = "dev"
|
|
|
|
# 抓取服务监听地址；交易服务用下面的 trading_host / trading_port
|
|
app_host: str = "0.0.0.0"
|
|
app_port: int = 31107
|
|
|
|
# 交易服务监听地址。两个服务同机部署时端口必须错开；交易服务只能单实例，
|
|
# 不要在它前面挂多副本负载均衡。
|
|
trading_host: str = "0.0.0.0"
|
|
trading_port: int = 31108
|
|
|
|
# 下单任务网关监听地址（第三个部署单元）。网关部署在服务器侧，本地 worker
|
|
# 通过出站长轮询从这里领任务。详见 docs/order-gateway.md。
|
|
gateway_host: str = "0.0.0.0"
|
|
gateway_port: int = 31109
|
|
|
|
# ---- 日志配置 ----
|
|
log_level: str = "INFO"
|
|
log_to_file: bool | None = None # None 表示根据环境自动决定
|
|
log_dir: str = "logs"
|
|
log_rotation: str = "100 MB"
|
|
log_retention: str = "14 days"
|
|
log_compression: str = "zip"
|
|
log_format: str = "{time:YYYY-MM-DD HH:mm:ss} {level} {message}"
|
|
log_enqueue: bool = True
|
|
|
|
# ---- 鉴权配置（通用：两个服务共用同一个对外 token）----
|
|
bearer_token: str = "REPLACE_WITH_TOKEN_32CHARS"
|
|
|
|
# ---- HTTP 抓取配置（仅抓取服务）----
|
|
request_timeout_seconds: float = 30.0
|
|
max_site_concurrency: int = 8 # 对站点的最大并发请求数
|
|
http_max_attempts: int = 3 # 单次抓取的最大尝试次数（含首次）
|
|
session_ttl_seconds: float = 1800.0 # Akamai cookie 会话最长复用时长，超时后重新预热
|
|
|
|
# ---- 浏览器兜底配置（仅抓取服务）----
|
|
# 纯 HTTP 被 Akamai 拦截时，用 Playwright 打开页面取回 cookie 再回灌给
|
|
# HTTP 客户端重试。日常流量不会触发；未安装 playwright 时自动降级为不兜底。
|
|
browser_fallback_enabled: bool = True
|
|
browser_headless: bool | None = None # None 表示根据环境自动决定
|
|
browser_channel: str | None = None # 例如 chrome；留空使用 bundled chromium
|
|
browser_launch_timeout_seconds: float = 60.0
|
|
browser_nav_timeout_seconds: float = 60.0
|
|
|
|
# ---- 代理配置（通用，可选，用于日本 IP）----
|
|
# 两个服务同机部署时通常各配各的：抓取高频匿名，出口 IP 被限速换掉即可；
|
|
# 交易带账号，出口 IP 频繁漂移反而会触发风控。
|
|
proxy_server: str | None = None
|
|
proxy_username: str | None = None
|
|
proxy_password: str | None = None
|
|
# 逗号分隔的直连主机名或通配符。服务间请求不应绕到公网代理。
|
|
proxy_bypass: str = "localhost,127.0.0.1,::1,rakuten-api,rakuten-trading,rakuten-gateway"
|
|
|
|
# ---- OpenTelemetry traces（通用，可选；默认关闭）----
|
|
# 启用后把抓取-解析链路以 span 导出到 OTLP/HTTP endpoint，用于排查"抓到
|
|
# 内容能否被解析"。两侧 main.py 在 lifespan 启动时按服务名分别初始化。
|
|
otel_enabled: bool = False
|
|
otel_endpoint: str | None = None # OTLP/HTTP traces endpoint，例如 https://oltp.jerryyan.top/v1/traces
|
|
otel_service_name: str = "rakuten" # 仅作兜底；实际值由两侧 main.py 显式覆盖
|
|
otel_headers: str | None = None # OTLP 鉴权头，形如 "k=v,k=v"；当前 endpoint 裸跑，留空
|
|
otel_export_interval_ms: int = 5000
|
|
# 解析失败时把页面 HTML 作为 span event 上报的上限字节；超出截断并标注。
|
|
# 单个搜索页 HTML 可达 200KB-2MB，调高时同步关注 OTLP 单次请求大小限制。
|
|
otel_snapshot_max_bytes: int = 2_000_000
|
|
|
|
# ---- 登录态与下单配置（仅交易服务）----
|
|
# 人工登录一次后落盘的 Playwright storage_state 目录（相对项目根目录）。
|
|
# 目录里是可直接冒充账号的 cookie，务必不要提交到版本库。
|
|
auth_state_dir: str = ".auth"
|
|
# 下单金额上限（日元）。实际应付金额超过该值时拒绝提交，防止解析出错或
|
|
# 页面改版导致买到远超预期的订单。设为 0 表示不设上限（不建议）。
|
|
order_max_total_yen: int = 30000
|
|
# 自动下单时在支付方式页选择的选项文案（按可见文案匹配，非 value/id）。
|
|
# 2026-08-11 未经真实确认页验证：见 site_interact.py::_select_payment_method。
|
|
order_payment_method: str = "クレジットカード"
|
|
# 付款后监控（订单列表页轮询）：间隔与最大轮询次数。2026-08-13 用真实订单号
|
|
# 实测过 order.my.rakuten.co.jp 的结构，见 site_interact.py::check_order_status。
|
|
# 间隔不宜太短——同一账号频繁访问订单页有被风控盯上的风险；默认 3 小时一次，
|
|
# 最多轮询 80 次（约 10 天，覆盖绝大多数国内配送时长），到期仍未到「配達完了」
|
|
# 就停止（不是失败，只是不再继续追踪，详见 runner.py::_monitor_order）。
|
|
order_monitor_poll_interval_seconds: int = 3 * 3600
|
|
order_monitor_max_checks: int = 80
|
|
|
|
# ---- 自动重登（仅交易服务，需要 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
|
|
# 启动时是否自动登录一次。容器部署用：镜像里没有落盘的 storage_state，
|
|
# 开启后启动即按 account.yaml 自己登一次，不必先在宿主机跑 scripts/login.py。
|
|
# 后台执行（不阻塞 HTTP 端口），失败只记日志——服务照常提供 /health 与
|
|
# /api/auth/*，撞验证码可人工接管后调 /api/auth/login 重试。
|
|
auto_login_on_start: bool = False
|
|
|
|
# ---- 下单任务网关（仅网关进程 app.gateway.main 使用）----
|
|
# 网关的 SQLite 文件路径（相对项目根目录）。任务队列与状态镜像都在这里，
|
|
# 部署时务必放在持久化卷上，丢了等于丢了一批下单任务。
|
|
gateway_db_path: str = "data/gateway.db"
|
|
# 任务租约 TTL（秒）。worker 领取后必须在此时间内首次 report 或 renew，
|
|
# 否则网关把任务标记为 stale。**绝不自动重投**（见 docs/order-gateway.md §5）。
|
|
lease_ttl_seconds: int = 300
|
|
# 长轮询单次最长挂起秒数。worker 端的 wait 参数会被夹到这个上限。
|
|
lease_max_wait_seconds: int = Field(default=60, ge=1, le=300)
|
|
# worker 心跳超时阈值（秒）。网关 /health 据此判断 worker 是否失联：
|
|
# 正常 worker 每 30 秒来一次 lease，超过该阈值未来 lease 即视为异常。
|
|
worker_offline_alert_seconds: int = 300
|
|
# 终结类事件回调（callback_url）的单次 HTTP 超时（秒）。投递是 best-effort：
|
|
# 超时/失败只记日志不重试，权威状态以 GET /api/orders/{task_id} 为准。
|
|
callback_timeout_seconds: float = 10.0
|
|
|
|
# ---- 账号只读查询通道（网关 + worker 两侧共用，见 docs/order-gateway.md §11）----
|
|
# 查询单租约 TTL（秒）。worker 领走后必须在此时间内回结果，否则网关把它
|
|
# **重投回 queued**——只读操作重复执行没有副作用，与下单任务的 stale 语义相反。
|
|
query_lease_ttl_seconds: int = 180
|
|
# 查询单整体存活上限（秒）。从创建算起，超过仍未拿到结果即置 expired
|
|
# （多半是本地 worker 不在线）。上游据此判断「这张单不用再等了」。
|
|
query_ttl_seconds: int = 900
|
|
# 同一张查询单最多被领取几次。租约过期重投累计到这个次数仍无结果即置 failed，
|
|
# 避免某条查询让 worker 反复卡死在同一个页面上。
|
|
query_max_attempts: int = 3
|
|
# 终态查询单的保留时长（秒）。结果里带站点原始 JSON，体积不小，超期由 sweep
|
|
# 清掉；默认 7 天，够上游对账与排查。
|
|
query_retention_seconds: int = 7 * 24 * 3600
|
|
# worker 侧单次站点读取的耗时上限（秒）。查询与下单共用同一把账号锁，正在
|
|
# 跑的下单会让查询排队等待，这个上限保证查询不会一直挂着——超时按失败回报，
|
|
# 上游重发即可（只读，重发无副作用）。
|
|
account_query_timeout_seconds: int = 120
|
|
# 订单列表查询缺省翻页上限。上游可在 params.max_pages 覆盖（1..20）。
|
|
account_query_default_max_pages: int = 3
|
|
# 回报给网关的结果 JSON 体积上限（字节）。超过时丢掉站点原始 JSON、只回
|
|
# 规范化字段并标记 raw_omitted——避免把几 MB 的页面状态塞进网关 SQLite。
|
|
query_result_max_bytes: int = 1024 * 1024
|
|
|
|
# ---- 定时下派通道（仅网关进程 app.gateway.main 使用，见 docs/order-gateway.md §12）----
|
|
# 网关周期性派 order_list / order_detail 查询、把账号真实订单沉淀进编目
|
|
# account_orders 表（回写网关）。关闭则 collector 完全不启动。
|
|
account_discovery_enabled: bool = True
|
|
# 两轮 order_list 扫描的间隔（秒）。间隔也是「该订单详情何时到期需刷新」的
|
|
# 参照之一，默认 6 小时一次列表、24 小时刷新一次单笔详情，符合「周期性盘点
|
|
# 账号」的节奏，也不至于频繁戳订单页触发风控。
|
|
account_discovery_interval_seconds: int = 6 * 3600
|
|
account_discovery_max_pages: int = 3
|
|
# 已编目订单的详情刷新间隔（秒）。detail_fetched_at 距今超过这个值就再派一张
|
|
# order_detail 查询更新配送阶段。
|
|
account_detail_refresh_seconds: int = 24 * 3600
|
|
|
|
# ---- 本地下单 worker（仅交易服务内的 worker 子模块使用）----
|
|
# 网关 URL。**留空则不启动 worker**，交易服务只跑登录态接口。
|
|
# 部署形态：本地机（NAT 后无公网入口）通过出站长轮询领任务，详见
|
|
# docs/order-gateway.md。
|
|
order_gateway_url: str = ""
|
|
# worker 标识。同一时间只能有一个 worker 持有 lease，这个值用来区分不同
|
|
# 本地机；留空时取主机名。
|
|
worker_id: str | None = None
|
|
# 本地订单 SQLite 文件路径。订单主表 + 状态事件 + 证据索引都在这里，
|
|
# 是执行事实的权威记录；网关上只是镜像。
|
|
trading_db_path: str = "data/trading.db"
|
|
# 页面证据落盘目录（HTML 快照 + 截图 + meta.json），按 task_id 分子目录。
|
|
evidence_dir: str = "data/evidence"
|
|
# 抓取服务基地址。worker 需要商品数据（加购要用 purchase 块）时出站请求这里，
|
|
# 不直接 import 解析器——抓取与交易是两个进程，详见 README「两个部署单元」。
|
|
scraper_base_url: str = ""
|
|
|
|
# ---- 目标站点 ----
|
|
home_url: str = DEFAULT_HOME_URL
|
|
|
|
@property
|
|
def browser_headless_effective(self) -> bool:
|
|
"""浏览器无头模式：显式配置优先，否则开发环境使用有头模式方便调试"""
|
|
if self.browser_headless is not None:
|
|
return self.browser_headless
|
|
return self.app_env != "dev"
|
|
|
|
@property
|
|
def log_to_file_effective(self) -> bool:
|
|
"""是否写日志文件：显式配置优先，否则生产环境默认写文件"""
|
|
if self.log_to_file is not None:
|
|
return self.log_to_file
|
|
return self.app_env == "prod"
|
|
|
|
@property
|
|
def playwright_proxy(self) -> dict[str, str] | None:
|
|
"""构建 Playwright 代理配置字典"""
|
|
if not self.proxy_server:
|
|
return None
|
|
|
|
proxy: dict[str, str] = {"server": self.proxy_server}
|
|
if self.proxy_username:
|
|
proxy["username"] = self.proxy_username
|
|
if self.proxy_password:
|
|
proxy["password"] = self.proxy_password
|
|
if self.proxy_bypass.strip():
|
|
proxy["bypass"] = self.proxy_bypass
|
|
return proxy
|
|
|
|
@property
|
|
def httpx_proxy(self) -> str | None:
|
|
"""构建 httpx 代理 URL（含认证信息）"""
|
|
if not self.proxy_server:
|
|
return None
|
|
if not self.proxy_username:
|
|
return self.proxy_server
|
|
|
|
scheme, _, rest = self.proxy_server.partition("://")
|
|
if not rest:
|
|
return self.proxy_server
|
|
credentials = f"{quote(self.proxy_username, safe='')}:{quote(self.proxy_password or '', safe='')}"
|
|
return f"{scheme}://{credentials}@{rest}"
|
|
|
|
def proxy_bypasses(self, url: str) -> bool:
|
|
"""目标 URL 的主机是否应绕过外部代理。"""
|
|
host = urlsplit(url).hostname
|
|
if not host:
|
|
return False
|
|
|
|
normalized_host = host.rstrip(".").lower()
|
|
for raw_pattern in self.proxy_bypass.split(","):
|
|
pattern = raw_pattern.strip().rstrip(".").lower()
|
|
if not pattern:
|
|
continue
|
|
if pattern.startswith("."):
|
|
suffix = pattern[1:]
|
|
if normalized_host == suffix or normalized_host.endswith(f".{suffix}"):
|
|
return True
|
|
continue
|
|
if fnmatchcase(normalized_host, pattern):
|
|
return True
|
|
return False
|
|
|
|
@property
|
|
def auth_state_path(self) -> Path:
|
|
"""登录态目录的绝对路径，不存在时创建"""
|
|
path = Path(self.auth_state_dir)
|
|
if not path.is_absolute():
|
|
path = BASE_DIR / path
|
|
path.mkdir(parents=True, exist_ok=True)
|
|
return path
|
|
|
|
@property
|
|
def gateway_db_path_resolved(self) -> Path:
|
|
"""网关 SQLite 文件的绝对路径，父目录不存在时创建"""
|
|
path = Path(self.gateway_db_path)
|
|
if not path.is_absolute():
|
|
path = BASE_DIR / path
|
|
path.parent.mkdir(parents=True, exist_ok=True)
|
|
return path
|
|
|
|
@property
|
|
def trading_db_path_resolved(self) -> Path:
|
|
"""本地订单 SQLite 文件的绝对路径，父目录不存在时创建"""
|
|
path = Path(self.trading_db_path)
|
|
if not path.is_absolute():
|
|
path = BASE_DIR / path
|
|
path.parent.mkdir(parents=True, exist_ok=True)
|
|
return path
|
|
|
|
@property
|
|
def evidence_path(self) -> Path:
|
|
"""证据目录的绝对路径，不存在时创建"""
|
|
path = Path(self.evidence_dir)
|
|
if not path.is_absolute():
|
|
path = BASE_DIR / path
|
|
path.mkdir(parents=True, exist_ok=True)
|
|
return path
|
|
|
|
@property
|
|
def worker_id_effective(self) -> str:
|
|
"""worker 标识：显式配置优先，否则取主机名"""
|
|
return self.worker_id or socket.gethostname()
|
|
|
|
|
|
@lru_cache(maxsize=1)
|
|
def get_settings() -> Settings:
|
|
"""获取全局配置单例"""
|
|
return Settings()
|