q792602257andClaude Opus 5
f6c3976c0a
新增 §12 定时下派通道：网关自己定期派 order_list / order_detail 查询，把账号
真实订单沉淀进新表 account_orders（回写网关），上游可直接查 GET /api/account/orders
拿到账号里实际有哪些订单，不必自己记 order_number。
- collector.py: OrderDiscoveryCollector 常驻后台任务（与 sweep 并列）。每隔
account_discovery_interval_seconds 派 order_list，收割结果后把「编目里没有的
订单」upsert 进编目，再逐笔派 order_detail 沉淀 delivery_status/order_state。
状态无痕：不新增编排跟踪表，仅内存 _pending + _detail_dispatched_this_day；
detail query_id 按天分段（discover-detail-<order>-<日期>），同天幂等防重复派。
边界：collector 只消费 worker 结果的规范化字段，绝不解析 raw/raw_pages。
- db.py: account_orders 表 + AccountOrderRow + upsert/single/list/stale 访问；
upsert 以 order_number 为主键，重复采集只刷新，列表重扫不抹详情节点。
- 路由: GET /api/account/orders（列编目）、GET /api/account/orders/{n}(6005)、
POST /api/account/discovery/trigger（手动立即派一轮 list）。
- config: RAKUTEN_ACCOUNT_DISCOVERY_ENABLED / _INTERVAL_SECONDS / _MAX_PAGES /
RAKUTEN_ACCOUNT_DETAIL_REFRESH_SECONDS。
- 既有网关测试夹具统一关 discovery（会启动即派扫描污染查询队列语义），
新表测试在 tests/test_gateway_discovery.py（13 用例，真实 QueryQueue+DB 装配）。
472 tests passed.
Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
43 lines
1.7 KiB
Python
43 lines
1.7 KiB
Python
"""网关容器：集中管理网关侧服务实例，用于依赖注入"""
|
|
from __future__ import annotations
|
|
|
|
import asyncio
|
|
import logging
|
|
from dataclasses import dataclass, field
|
|
|
|
from app.gateway.db import GatewayDB
|
|
from app.gateway.query_queue import QueryQueue
|
|
from app.gateway.task_queue import TaskQueue
|
|
from app.shared.config import Settings
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
|
|
@dataclass(slots=True)
|
|
class GatewayContainer:
|
|
"""网关容器
|
|
|
|
与抓取/交易容器同样的依赖注入风格，但持有的是任务队列与 SQLite 连接，
|
|
生命周期由 main.lifespan 管理：start 时打开 DB，close 时关闭。
|
|
|
|
两条通道各有一个队列对象，共用同一个 GatewayDB：
|
|
- `task_queue` 下单任务（不可逆写，租约过期只置 stale）
|
|
- `query_queue` 账号只读查询（可安全重投，见 query_queue.py 顶部对照表）
|
|
|
|
`sweep_task` 是常驻后台扫描，把过期的 leased/running 推到 stale，并顺带扫
|
|
一轮查询单（重投 / 过期 / 清理过保留期的结果）。即使没有 lease 请求，过期的
|
|
任务也会被及时发现（规格 §5 的关键约束）。
|
|
|
|
`collector` 是定时下派通道（§12）：周期性派 order_list / order_detail 查询、
|
|
把账户真实订单沉淀进 `account_orders` 编目。`collector_task` 是它的后台任务，
|
|
在 lifespan 里与 `sweep_task` 一起启动/取消。
|
|
"""
|
|
|
|
settings: Settings
|
|
db: GatewayDB
|
|
task_queue: TaskQueue
|
|
query_queue: QueryQueue
|
|
collector: object | None = field(default=None) # app.gateway.collector.OrderDiscoveryCollector
|
|
sweep_task: asyncio.Task | None = None
|
|
collector_task: asyncio.Task | None = None
|