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 通过
725 lines
26 KiB
Python
725 lines
26 KiB
Python
"""网关 SQLite 访问层：四张表 + 原生 SQL，无业务逻辑
|
|
|
|
业务规则（状态机、并发度、幂等）放在 `task_queue.py`（下单任务）与
|
|
`query_queue.py`（账号只读查询），本模块只负责持久化与查询，
|
|
返回 dataclass 行。所有时间戳以 ISO8601 UTC 字符串存储（以 `Z` 结尾），便于
|
|
跨进程对账；时间戳运算在上层完成。
|
|
|
|
写操作由上层的 asyncio.Lock 串行化（详见对应模块），本层不重复加锁，
|
|
因此**调用方必须确保写操作在外层锁的保护下进行**。
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
import logging
|
|
from dataclasses import dataclass
|
|
from pathlib import Path
|
|
from typing import Any
|
|
|
|
import aiosqlite
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
SCHEMA = """
|
|
CREATE TABLE IF NOT EXISTS tasks (
|
|
task_id TEXT PRIMARY KEY,
|
|
site TEXT NOT NULL,
|
|
intent_json TEXT NOT NULL,
|
|
callback_url TEXT, -- 终结类事件通知地址，见 docs/order-gateway.md §4.8
|
|
status TEXT NOT NULL,
|
|
lease_owner TEXT,
|
|
lease_expires_at TEXT,
|
|
lease_count INTEGER NOT NULL DEFAULT 0,
|
|
created_at TEXT NOT NULL,
|
|
updated_at TEXT NOT NULL
|
|
);
|
|
CREATE INDEX IF NOT EXISTS idx_tasks_pending ON tasks(status, created_at);
|
|
|
|
CREATE TABLE IF NOT EXISTS task_reports (
|
|
task_id TEXT NOT NULL,
|
|
state TEXT NOT NULL,
|
|
payable_yen INTEGER,
|
|
pay_deadline TEXT,
|
|
site_order_id TEXT,
|
|
evidence_ref TEXT,
|
|
detail TEXT,
|
|
reported_at TEXT NOT NULL,
|
|
PRIMARY KEY (task_id, state)
|
|
);
|
|
|
|
CREATE TABLE IF NOT EXISTS workers (
|
|
worker_id TEXT PRIMARY KEY,
|
|
last_seen_at TEXT NOT NULL
|
|
);
|
|
|
|
-- 账号只读查询单（见 docs/order-gateway.md §11）。与 tasks 分表，不是洁癖：
|
|
-- 两者的安全约束相反（只读可重投 vs 写操作绝不重投），共表会让「租约过期怎么办」
|
|
-- 这个最关键的分支变成一个 if，迟早被改错。
|
|
CREATE TABLE IF NOT EXISTS account_queries (
|
|
query_id TEXT PRIMARY KEY, -- 幂等键，语义同 task_id
|
|
site TEXT NOT NULL,
|
|
kind TEXT NOT NULL, -- order_list / order_detail
|
|
params_json TEXT NOT NULL, -- 查询参数原文，网关不解释内容
|
|
status TEXT NOT NULL, -- 见 shared.task_state.QueryStatus
|
|
lease_owner TEXT,
|
|
lease_expires_at TEXT,
|
|
attempts INTEGER NOT NULL DEFAULT 0, -- 被领取过几次
|
|
result_json TEXT, -- worker 回的结果原文
|
|
error_code INTEGER,
|
|
error_message TEXT,
|
|
created_at TEXT NOT NULL,
|
|
updated_at TEXT NOT NULL,
|
|
completed_at TEXT
|
|
);
|
|
CREATE INDEX IF NOT EXISTS idx_queries_pending ON account_queries(status, created_at);
|
|
|
|
-- 定时下派通道（docs/order-gateway.md §12）：账号里真实订单的「已发现」目录。
|
|
-- collector 周期性派 order_list / order_detail 查询，把 worker 回结果里的
|
|
-- **规范化字段**（订单号/店铺/日期/配送状态）沉淀到这里——这是网关自己的编目，
|
|
-- 不是上游查询单的原样透传。只消费规范化契约，绝不碰结果里的站点原始 JSON。
|
|
CREATE TABLE IF NOT EXISTS account_orders (
|
|
order_number TEXT PRIMARY KEY,
|
|
shop_id TEXT,
|
|
shop_name TEXT,
|
|
order_date TEXT, -- 站点 ISO8601 字符串
|
|
delivery_status TEXT, -- 最近一次 detail 查询带回的站点码（如 CHECKING_ORDER）
|
|
order_state TEXT, -- 映射后的 OrderState（如 delivered）
|
|
discovered_at TEXT NOT NULL, -- 首次经 account 采集到该订单的时间
|
|
detail_fetched_at TEXT, -- 最近一次成功取到详情页的时间
|
|
last_seen_at TEXT NOT NULL -- 最近一次出现在订单列表里的时间
|
|
);
|
|
CREATE INDEX IF NOT EXISTS idx_account_orders_last_seen ON account_orders(last_seen_at);
|
|
"""
|
|
|
|
|
|
@dataclass(slots=True)
|
|
class TaskRow:
|
|
"""tasks 表一行的强类型视图"""
|
|
|
|
task_id: str
|
|
site: str
|
|
intent_json: str
|
|
callback_url: str | None
|
|
status: str
|
|
lease_owner: str | None
|
|
lease_expires_at: str | None
|
|
lease_count: int
|
|
created_at: str
|
|
updated_at: str
|
|
|
|
@property
|
|
def intent(self) -> dict[str, Any]:
|
|
return json.loads(self.intent_json)
|
|
|
|
|
|
@dataclass(slots=True)
|
|
class ReportRow:
|
|
"""task_reports 表一行的强类型视图"""
|
|
|
|
task_id: str
|
|
state: str
|
|
payable_yen: int | None
|
|
pay_deadline: str | None
|
|
site_order_id: str | None
|
|
evidence_ref: str | None
|
|
detail: str
|
|
reported_at: str
|
|
|
|
|
|
@dataclass(slots=True)
|
|
class WorkerRow:
|
|
"""workers 表一行的强类型视图"""
|
|
|
|
worker_id: str
|
|
last_seen_at: str
|
|
|
|
|
|
@dataclass(slots=True)
|
|
class QueryRow:
|
|
"""account_queries 表一行的强类型视图"""
|
|
|
|
query_id: str
|
|
site: str
|
|
kind: str
|
|
params_json: str
|
|
status: str
|
|
lease_owner: str | None
|
|
lease_expires_at: str | None
|
|
attempts: int
|
|
result_json: str | None
|
|
error_code: int | None
|
|
error_message: str | None
|
|
created_at: str
|
|
updated_at: str
|
|
completed_at: str | None
|
|
|
|
@property
|
|
def params(self) -> dict[str, Any]:
|
|
return json.loads(self.params_json)
|
|
|
|
@property
|
|
def result(self) -> dict[str, Any] | None:
|
|
return json.loads(self.result_json) if self.result_json else None
|
|
|
|
|
|
@dataclass(slots=True)
|
|
class AccountOrderRow:
|
|
"""account_orders 表一行的强类型视图（定时下派通道的编目，见 §12）"""
|
|
|
|
order_number: str
|
|
shop_id: str | None
|
|
shop_name: str
|
|
order_date: str | None
|
|
delivery_status: str | None
|
|
order_state: str | None
|
|
discovered_at: str
|
|
detail_fetched_at: str | None
|
|
last_seen_at: str
|
|
|
|
|
|
def _row_to_task(row: aiosqlite.Row) -> TaskRow:
|
|
return TaskRow(
|
|
task_id=row["task_id"],
|
|
site=row["site"],
|
|
intent_json=row["intent_json"],
|
|
callback_url=row["callback_url"],
|
|
status=row["status"],
|
|
lease_owner=row["lease_owner"],
|
|
lease_expires_at=row["lease_expires_at"],
|
|
lease_count=row["lease_count"],
|
|
created_at=row["created_at"],
|
|
updated_at=row["updated_at"],
|
|
)
|
|
|
|
|
|
def _row_to_report(row: aiosqlite.Row) -> ReportRow:
|
|
return ReportRow(
|
|
task_id=row["task_id"],
|
|
state=row["state"],
|
|
payable_yen=row["payable_yen"],
|
|
pay_deadline=row["pay_deadline"],
|
|
site_order_id=row["site_order_id"],
|
|
evidence_ref=row["evidence_ref"],
|
|
detail=row["detail"],
|
|
reported_at=row["reported_at"],
|
|
)
|
|
|
|
|
|
def _row_to_worker(row: aiosqlite.Row) -> WorkerRow:
|
|
return WorkerRow(worker_id=row["worker_id"], last_seen_at=row["last_seen_at"])
|
|
|
|
|
|
def _row_to_query(row: aiosqlite.Row) -> QueryRow:
|
|
return QueryRow(
|
|
query_id=row["query_id"],
|
|
site=row["site"],
|
|
kind=row["kind"],
|
|
params_json=row["params_json"],
|
|
status=row["status"],
|
|
lease_owner=row["lease_owner"],
|
|
lease_expires_at=row["lease_expires_at"],
|
|
attempts=row["attempts"],
|
|
result_json=row["result_json"],
|
|
error_code=row["error_code"],
|
|
error_message=row["error_message"],
|
|
created_at=row["created_at"],
|
|
updated_at=row["updated_at"],
|
|
completed_at=row["completed_at"],
|
|
)
|
|
|
|
|
|
def _row_to_account_order(row: aiosqlite.Row) -> AccountOrderRow:
|
|
return AccountOrderRow(
|
|
order_number=row["order_number"],
|
|
shop_id=row["shop_id"],
|
|
shop_name=row["shop_name"] or "",
|
|
order_date=row["order_date"],
|
|
delivery_status=row["delivery_status"],
|
|
order_state=row["order_state"],
|
|
discovered_at=row["discovered_at"],
|
|
detail_fetched_at=row["detail_fetched_at"],
|
|
last_seen_at=row["last_seen_at"],
|
|
)
|
|
|
|
|
|
class GatewayDB:
|
|
"""网关 SQLite 访问对象
|
|
|
|
单连接 + aiosqlite 的内部线程。生命周期由 `GatewayContainer` 管理：
|
|
`start()` 在 lifespan 启动时调用，`close()` 在关闭时调用。
|
|
"""
|
|
|
|
def __init__(self, db_path: Path):
|
|
self._db_path = db_path
|
|
self._conn: aiosqlite.Connection | None = None
|
|
|
|
async def start(self) -> None:
|
|
"""打开连接并初始化 schema（IF NOT EXISTS，可重复执行）"""
|
|
self._db_path.parent.mkdir(parents=True, exist_ok=True)
|
|
self._conn = await aiosqlite.connect(str(self._db_path))
|
|
self._conn.row_factory = aiosqlite.Row
|
|
await self._conn.executescript(SCHEMA)
|
|
await self._migrate()
|
|
await self._conn.commit()
|
|
logger.info("网关 DB 已就绪：%s", self._db_path)
|
|
|
|
async def _migrate(self) -> None:
|
|
"""给既有库补新列（CREATE TABLE IF NOT EXISTS 不会改老表）"""
|
|
async with self.conn.execute("PRAGMA table_info(tasks)") as cur:
|
|
columns = {row["name"] for row in await cur.fetchall()}
|
|
if "callback_url" not in columns:
|
|
await self.conn.execute("ALTER TABLE tasks ADD COLUMN callback_url TEXT")
|
|
logger.info("迁移：tasks 表补充 callback_url 列")
|
|
|
|
async def close(self) -> None:
|
|
if self._conn is not None:
|
|
await self._conn.close()
|
|
self._conn = None
|
|
|
|
@property
|
|
def conn(self) -> aiosqlite.Connection:
|
|
if self._conn is None:
|
|
raise RuntimeError("GatewayDB 未启动：先调用 start()")
|
|
return self._conn
|
|
|
|
# ---- tasks ----
|
|
|
|
async def get_task(self, task_id: str) -> TaskRow | None:
|
|
async with self.conn.execute("SELECT * FROM tasks WHERE task_id = ?", (task_id,)) as cur:
|
|
row = await cur.fetchone()
|
|
return _row_to_task(row) if row else None
|
|
|
|
async def insert_task(self, row: TaskRow) -> bool:
|
|
"""插入新任务。返回 True=新建，False=task_id 已存在（幂等命中）"""
|
|
try:
|
|
await self.conn.execute(
|
|
"INSERT INTO tasks (task_id, site, intent_json, callback_url, status, "
|
|
"lease_owner, lease_expires_at, lease_count, created_at, updated_at) "
|
|
"VALUES (?, ?, ?, ?, ?, NULL, NULL, 0, ?, ?)",
|
|
(
|
|
row.task_id,
|
|
row.site,
|
|
row.intent_json,
|
|
row.callback_url,
|
|
row.status,
|
|
row.created_at,
|
|
row.updated_at,
|
|
),
|
|
)
|
|
await self.conn.commit()
|
|
return True
|
|
except aiosqlite.IntegrityError:
|
|
return False
|
|
|
|
async def update_task(
|
|
self,
|
|
task_id: str,
|
|
*,
|
|
status: str | None = None,
|
|
lease_owner: str | None = None,
|
|
lease_expires_at: str | None = None,
|
|
lease_count: int | None = None,
|
|
updated_at: str | None = None,
|
|
clear_lease: bool = False,
|
|
) -> None:
|
|
"""更新任务字段。clear_lease=True 时把 lease_owner/expires_at 置 NULL"""
|
|
sets: list[str] = []
|
|
params: list[Any] = []
|
|
if status is not None:
|
|
sets.append("status = ?")
|
|
params.append(status)
|
|
if lease_owner is not None:
|
|
sets.append("lease_owner = ?")
|
|
params.append(lease_owner)
|
|
if lease_expires_at is not None:
|
|
sets.append("lease_expires_at = ?")
|
|
params.append(lease_expires_at)
|
|
if lease_count is not None:
|
|
sets.append("lease_count = ?")
|
|
params.append(lease_count)
|
|
if updated_at is not None:
|
|
sets.append("updated_at = ?")
|
|
params.append(updated_at)
|
|
if clear_lease:
|
|
sets.append("lease_owner = NULL")
|
|
sets.append("lease_expires_at = NULL")
|
|
if not sets:
|
|
return
|
|
params.append(task_id)
|
|
await self.conn.execute(
|
|
f"UPDATE tasks SET {', '.join(sets)} WHERE task_id = ?",
|
|
params,
|
|
)
|
|
await self.conn.commit()
|
|
|
|
async def list_tasks(
|
|
self,
|
|
*,
|
|
status: str | None = None,
|
|
site: str | None = None,
|
|
limit: int = 50,
|
|
offset: int = 0,
|
|
) -> tuple[list[TaskRow], int]:
|
|
"""分页列出任务，按 created_at 升序"""
|
|
where = []
|
|
params: list[Any] = []
|
|
if status:
|
|
where.append("status = ?")
|
|
params.append(status)
|
|
if site:
|
|
where.append("site = ?")
|
|
params.append(site)
|
|
clause = f"WHERE {' AND '.join(where)}" if where else ""
|
|
|
|
async with self.conn.execute(
|
|
f"SELECT COUNT(*) FROM tasks {clause}",
|
|
params,
|
|
) as cur:
|
|
total = (await cur.fetchone())[0]
|
|
|
|
sql = f"SELECT * FROM tasks {clause} ORDER BY created_at ASC LIMIT ? OFFSET ?"
|
|
async with self.conn.execute(sql, [*params, limit, offset]) as cur:
|
|
rows = await cur.fetchall()
|
|
return [_row_to_task(r) for r in rows], total
|
|
|
|
async def list_tasks_in_statuses(self, statuses: tuple[str, ...]) -> list[TaskRow]:
|
|
"""取处于给定状态集合的全部任务（不分页，用于健康检查的 active 列表）"""
|
|
if not statuses:
|
|
return []
|
|
placeholders = ", ".join("?" for _ in statuses)
|
|
sql = (
|
|
f"SELECT * FROM tasks WHERE status IN ({placeholders}) "
|
|
"ORDER BY created_at ASC"
|
|
)
|
|
async with self.conn.execute(sql, statuses) as cur:
|
|
rows = await cur.fetchall()
|
|
return [_row_to_task(r) for r in rows]
|
|
|
|
async def pick_queued_task(self, site: str | None) -> TaskRow | None:
|
|
"""取最早的 queued 任务，可选按站点过滤"""
|
|
if site:
|
|
sql = "SELECT * FROM tasks WHERE status = ? AND site = ? ORDER BY created_at ASC LIMIT 1"
|
|
params: tuple[Any, ...] = ("queued", site)
|
|
else:
|
|
sql = "SELECT * FROM tasks WHERE status = ? ORDER BY created_at ASC LIMIT 1"
|
|
params = ("queued",)
|
|
async with self.conn.execute(sql, params) as cur:
|
|
row = await cur.fetchone()
|
|
return _row_to_task(row) if row else None
|
|
|
|
# ---- task_reports ----
|
|
|
|
async def upsert_report(self, row: ReportRow) -> bool:
|
|
"""写入一条 report。同一 (task_id, state) 已存在则覆盖（幂等）。
|
|
|
|
返回 True=本次新写入，False=覆盖了既有行。
|
|
|
|
先 SELECT 再 INSERT OR REPLACE，避免依赖 SQLite rowcount——后者在
|
|
INSERT OR REPLACE 上即使命中既有行也报 1，无法区分。
|
|
"""
|
|
async with self.conn.execute(
|
|
"SELECT 1 FROM task_reports WHERE task_id = ? AND state = ?",
|
|
(row.task_id, row.state),
|
|
) as cur:
|
|
existed = await cur.fetchone() is not None
|
|
|
|
await self.conn.execute(
|
|
"INSERT OR REPLACE INTO task_reports "
|
|
"(task_id, state, payable_yen, pay_deadline, site_order_id, "
|
|
"evidence_ref, detail, reported_at) VALUES (?, ?, ?, ?, ?, ?, ?, ?)",
|
|
(
|
|
row.task_id,
|
|
row.state,
|
|
row.payable_yen,
|
|
row.pay_deadline,
|
|
row.site_order_id,
|
|
row.evidence_ref,
|
|
row.detail,
|
|
row.reported_at,
|
|
),
|
|
)
|
|
await self.conn.commit()
|
|
return not existed
|
|
|
|
async def get_reports(self, task_id: str) -> list[ReportRow]:
|
|
async with self.conn.execute(
|
|
"SELECT * FROM task_reports WHERE task_id = ? ORDER BY reported_at ASC",
|
|
(task_id,),
|
|
) as cur:
|
|
rows = await cur.fetchall()
|
|
return [_row_to_report(r) for r in rows]
|
|
|
|
async def get_latest_state(self, task_id: str) -> str | None:
|
|
"""取该任务最近一次 report 的 state，无 report 返回 None"""
|
|
async with self.conn.execute(
|
|
"SELECT state FROM task_reports WHERE task_id = ? "
|
|
"ORDER BY reported_at DESC LIMIT 1",
|
|
(task_id,),
|
|
) as cur:
|
|
row = await cur.fetchone()
|
|
return row["state"] if row else None
|
|
|
|
# ---- workers ----
|
|
|
|
async def upsert_worker(self, worker_id: str, last_seen_at: str) -> None:
|
|
await self.conn.execute(
|
|
"INSERT INTO workers (worker_id, last_seen_at) VALUES (?, ?) "
|
|
"ON CONFLICT(worker_id) DO UPDATE SET last_seen_at = excluded.last_seen_at",
|
|
(worker_id, last_seen_at),
|
|
)
|
|
await self.conn.commit()
|
|
|
|
async def list_workers(self) -> list[WorkerRow]:
|
|
async with self.conn.execute("SELECT * FROM workers ORDER BY last_seen_at DESC") as cur:
|
|
rows = await cur.fetchall()
|
|
return [_row_to_worker(r) for r in rows]
|
|
|
|
# ---- 计数 ----
|
|
|
|
async def count_by_status(self, status: str) -> int:
|
|
async with self.conn.execute(
|
|
"SELECT COUNT(*) FROM tasks WHERE status = ?",
|
|
(status,),
|
|
) as cur:
|
|
return (await cur.fetchone())[0]
|
|
|
|
# ---- account_queries ----
|
|
|
|
async def get_query(self, query_id: str) -> QueryRow | None:
|
|
async with self.conn.execute(
|
|
"SELECT * FROM account_queries WHERE query_id = ?", (query_id,)
|
|
) as cur:
|
|
row = await cur.fetchone()
|
|
return _row_to_query(row) if row else None
|
|
|
|
async def insert_query(self, row: QueryRow) -> bool:
|
|
"""插入新查询单。返回 True=新建，False=query_id 已存在（幂等命中）"""
|
|
try:
|
|
await self.conn.execute(
|
|
"INSERT INTO account_queries (query_id, site, kind, params_json, status, "
|
|
"lease_owner, lease_expires_at, attempts, result_json, error_code, "
|
|
"error_message, created_at, updated_at, completed_at) "
|
|
"VALUES (?, ?, ?, ?, ?, NULL, NULL, 0, NULL, NULL, NULL, ?, ?, NULL)",
|
|
(
|
|
row.query_id,
|
|
row.site,
|
|
row.kind,
|
|
row.params_json,
|
|
row.status,
|
|
row.created_at,
|
|
row.updated_at,
|
|
),
|
|
)
|
|
await self.conn.commit()
|
|
return True
|
|
except aiosqlite.IntegrityError:
|
|
return False
|
|
|
|
async def update_query(
|
|
self,
|
|
query_id: str,
|
|
*,
|
|
status: str | None = None,
|
|
lease_owner: str | None = None,
|
|
lease_expires_at: str | None = None,
|
|
attempts: int | None = None,
|
|
result_json: str | None = None,
|
|
error_code: int | None = None,
|
|
error_message: str | None = None,
|
|
updated_at: str | None = None,
|
|
completed_at: str | None = None,
|
|
clear_lease: bool = False,
|
|
) -> None:
|
|
"""更新查询单字段。clear_lease=True 时把 lease_owner/expires_at 置 NULL"""
|
|
sets: list[str] = []
|
|
params: list[Any] = []
|
|
for column, value in (
|
|
("status", status),
|
|
("lease_owner", lease_owner),
|
|
("lease_expires_at", lease_expires_at),
|
|
("attempts", attempts),
|
|
("result_json", result_json),
|
|
("error_code", error_code),
|
|
("error_message", error_message),
|
|
("updated_at", updated_at),
|
|
("completed_at", completed_at),
|
|
):
|
|
if value is not None:
|
|
sets.append(f"{column} = ?")
|
|
params.append(value)
|
|
if clear_lease:
|
|
sets.append("lease_owner = NULL")
|
|
sets.append("lease_expires_at = NULL")
|
|
if not sets:
|
|
return
|
|
params.append(query_id)
|
|
await self.conn.execute(
|
|
f"UPDATE account_queries SET {', '.join(sets)} WHERE query_id = ?",
|
|
params,
|
|
)
|
|
await self.conn.commit()
|
|
|
|
async def pick_queued_query(self, site: str | None) -> QueryRow | None:
|
|
"""取最早的 queued 查询单，可选按站点过滤"""
|
|
if site:
|
|
sql = (
|
|
"SELECT * FROM account_queries WHERE status = ? AND site = ? "
|
|
"ORDER BY created_at ASC LIMIT 1"
|
|
)
|
|
params: tuple[Any, ...] = ("queued", site)
|
|
else:
|
|
sql = "SELECT * FROM account_queries WHERE status = ? ORDER BY created_at ASC LIMIT 1"
|
|
params = ("queued",)
|
|
async with self.conn.execute(sql, params) as cur:
|
|
row = await cur.fetchone()
|
|
return _row_to_query(row) if row else None
|
|
|
|
async def list_queries(
|
|
self,
|
|
*,
|
|
status: str | None = None,
|
|
kind: str | None = None,
|
|
limit: int = 50,
|
|
offset: int = 0,
|
|
) -> tuple[list[QueryRow], int]:
|
|
"""分页列出查询单，按 created_at 降序（查询是即时行为，最近的先看）"""
|
|
where = []
|
|
params: list[Any] = []
|
|
if status:
|
|
where.append("status = ?")
|
|
params.append(status)
|
|
if kind:
|
|
where.append("kind = ?")
|
|
params.append(kind)
|
|
clause = f"WHERE {' AND '.join(where)}" if where else ""
|
|
|
|
async with self.conn.execute(
|
|
f"SELECT COUNT(*) FROM account_queries {clause}", params
|
|
) as cur:
|
|
total = (await cur.fetchone())[0]
|
|
|
|
sql = (
|
|
f"SELECT * FROM account_queries {clause} "
|
|
"ORDER BY created_at DESC LIMIT ? OFFSET ?"
|
|
)
|
|
async with self.conn.execute(sql, [*params, limit, offset]) as cur:
|
|
rows = await cur.fetchall()
|
|
return [_row_to_query(r) for r in rows], total
|
|
|
|
async def list_queries_in_statuses(self, statuses: tuple[str, ...]) -> list[QueryRow]:
|
|
"""取处于给定状态集合的全部查询单（sweep 与健康检查用）"""
|
|
if not statuses:
|
|
return []
|
|
placeholders = ", ".join("?" for _ in statuses)
|
|
sql = (
|
|
f"SELECT * FROM account_queries WHERE status IN ({placeholders}) "
|
|
"ORDER BY created_at ASC"
|
|
)
|
|
async with self.conn.execute(sql, statuses) as cur:
|
|
rows = await cur.fetchall()
|
|
return [_row_to_query(r) for r in rows]
|
|
|
|
async def count_queries_by_status(self, status: str) -> int:
|
|
async with self.conn.execute(
|
|
"SELECT COUNT(*) FROM account_queries WHERE status = ?", (status,)
|
|
) as cur:
|
|
return (await cur.fetchone())[0]
|
|
|
|
async def delete_queries_completed_before(self, cutoff: str) -> int:
|
|
"""清理 completed_at 早于 cutoff 的终态查询单，返回删除条数
|
|
|
|
结果里带站点原始 JSON，不清会一直涨。只删已完成的，进行中的一律不动。
|
|
"""
|
|
cursor = await self.conn.execute(
|
|
"DELETE FROM account_queries WHERE completed_at IS NOT NULL AND completed_at < ?",
|
|
(cutoff,),
|
|
)
|
|
await self.conn.commit()
|
|
return cursor.rowcount or 0
|
|
|
|
# ---- account_orders（定时下派通道的编目，见 docs/order-gateway.md §12）----
|
|
|
|
async def upsert_account_order(self, row: AccountOrderRow) -> bool:
|
|
"""把一笔账号订单写进编目（回写）。存在则刷新，返回 True=新写入
|
|
|
|
这正是「采集到网关中不存在的订单就写回」的落点：主键是 order_number，
|
|
重复采集同一笔订单只是刷新 last_seen 等字段，不会产生重复行。
|
|
"""
|
|
async with self.conn.execute(
|
|
"SELECT 1 FROM account_orders WHERE order_number = ?", (row.order_number,)
|
|
) as cur:
|
|
existed = await cur.fetchone() is not None
|
|
await self.conn.execute(
|
|
"INSERT INTO account_orders (order_number, shop_id, shop_name, order_date, "
|
|
"delivery_status, order_state, discovered_at, detail_fetched_at, last_seen_at) "
|
|
"VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?) "
|
|
"ON CONFLICT(order_number) DO UPDATE SET "
|
|
"shop_id = excluded.shop_id, shop_name = excluded.shop_name, "
|
|
"order_date = excluded.order_date, delivery_status = excluded.delivery_status, "
|
|
"order_state = excluded.order_state, detail_fetched_at = excluded.detail_fetched_at, "
|
|
"last_seen_at = excluded.last_seen_at",
|
|
(
|
|
row.order_number,
|
|
row.shop_id,
|
|
row.shop_name,
|
|
row.order_date,
|
|
row.delivery_status,
|
|
row.order_state,
|
|
row.discovered_at,
|
|
row.detail_fetched_at,
|
|
row.last_seen_at,
|
|
),
|
|
)
|
|
await self.conn.commit()
|
|
return not existed
|
|
|
|
async def get_account_order(self, order_number: str) -> AccountOrderRow | None:
|
|
async with self.conn.execute(
|
|
"SELECT * FROM account_orders WHERE order_number = ?", (order_number,)
|
|
) as cur:
|
|
row = await cur.fetchone()
|
|
return _row_to_account_order(row) if row else None
|
|
|
|
async def list_account_orders(
|
|
self,
|
|
*,
|
|
state: str | None = None,
|
|
limit: int = 50,
|
|
offset: int = 0,
|
|
) -> tuple[list[AccountOrderRow], int]:
|
|
"""分页列出编目订单，按 last_seen（最近出现）倒序"""
|
|
where = []
|
|
params: list[Any] = []
|
|
if state:
|
|
where.append("order_state = ?")
|
|
params.append(state)
|
|
clause = f"WHERE {' AND '.join(where)}" if where else ""
|
|
|
|
async with self.conn.execute(
|
|
f"SELECT COUNT(*) FROM account_orders {clause}", params
|
|
) as cur:
|
|
total = (await cur.fetchone())[0]
|
|
|
|
sql = (
|
|
f"SELECT * FROM account_orders {clause} "
|
|
"ORDER BY last_seen_at DESC LIMIT ? OFFSET ?"
|
|
)
|
|
async with self.conn.execute(sql, [*params, limit, offset]) as cur:
|
|
rows = await cur.fetchall()
|
|
return [_row_to_account_order(r) for r in rows], total
|
|
|
|
async def list_account_orders_needing_detail(self, cutoff_iso: str) -> list[AccountOrderRow]:
|
|
"""取「还没成功取过详情」或「详情已过刷新期（detail_fetched_at < cutoff）」的编目订单
|
|
|
|
collector 据此派新一批 order_detail 查询。校验用 detail_fetched_at
|
|
IS NULL OR < cutoff，保证不会对同一笔订单反复派（详情查询结果回来后才
|
|
落 detail_fetched_at）。
|
|
"""
|
|
async with self.conn.execute(
|
|
"SELECT * FROM account_orders WHERE detail_fetched_at IS NULL "
|
|
"OR detail_fetched_at < ? ORDER BY last_seen_at ASC",
|
|
(cutoff_iso,),
|
|
) as cur:
|
|
rows = await cur.fetchall()
|
|
return [_row_to_account_order(r) for r in rows]
|