feat(orders): P0 订单读取适配层——连 order_site 只读库查订单(默认关闭)

让运营能按订单号查到订单,作为后续资料收集(P2)与皮肤准入(P3/P4)的输入。
order_site 零改动:只读直连、独立极小连接池、可一键降级。

上机核实推翻了规划两处假设(详见 §6.5.1):
- orders 表没有 order_no 列,只有 id 与 platform_order_id → 查询入口用后者;
- platform_order_id 非唯一:541 组重复,形态固定为「91kaquan 内部行(已发货)
  + 真实店铺行(无人管)」。按规划原文的单行判据,这 540 个已发货订单会被误判
  成「没人管」而放行重复履约。改为**订单号维度聚合**:任一行已发货即 auto_fulfilled。

真机验证(只读账号 livehub_ro):三桶判定全对,重复组正确拦下并回显真实
店铺行的中文名(狂沙舞者,而非内部码 suit_sand_dancer),单次查询 3.3ms。

顺带修一个只在真机才暴露的 bug:_get_engine 先置 _INITIALIZED 再读 URL,
一次落空就把模块永久毒化成不可用,之后补上连接串也救不回来。
This commit is contained in:
yml2213
2026-10-02 22:50:20 +08:00
parent d6c6113112
commit afa5c815f6
9 changed files with 1001 additions and 3 deletions
@@ -0,0 +1,85 @@
"""订单快照缓存表(P0 订单读取适配层)。
用户提交订单号时,适配层会去 order_site 的 PostgreSQL 只读库查一次,把结果
落到这里:一来立即回显商品,二来在对方库不可用时还能降级用旧快照。
主键是 **platform_order_id + source_order_id** 两列,不是单列 —— 上机实测
(见规划 §6.5.1)同一个 platform_order_id 在 orders 表里有 541 组重复,形态
固定为「shop_id=91kaquan 的发货方内部行 + 一个真实店铺行」。按订单号单列存
会让这两行互相覆盖,且回显时可能取到内部行的 sku_code(如 suit_shadow_gothic)
而不是买家真正买的中文名。
缓存只是快照,不是业务数据:order_site 不可用时由此降级,24 小时后重查。
"""
from collections.abc import Sequence
import sqlalchemy as sa
from alembic import op
revision: str = "20261002_0039"
down_revision: str | None = "20261002_0038"
branch_labels: str | Sequence[str] | None = None
depends_on: str | Sequence[str] | None = None
def _tables(bind) -> set[str]:
return set(sa.inspect(bind).get_table_names())
def upgrade() -> None:
bind = op.get_bind()
if "external_order_cache" in _tables(bind):
return
op.create_table(
"external_order_cache",
# 用户/客服填入的快手订单号(order_site orders.platform_order_id)。
sa.Column("platform_order_id", sa.String(length=64), primary_key=True),
# 对应的 order_site orders.id —— 同一个订单号可能有两行,靠它区分。
sa.Column("source_order_id", sa.String(length=64), primary_key=True),
sa.Column("platform", sa.String(length=32), nullable=False, server_default=""),
sa.Column("provider", sa.String(length=32), nullable=False, server_default=""),
sa.Column("shop_id", sa.String(length=64), nullable=False, server_default=""),
sa.Column(
"order_status", sa.String(length=32), nullable=False, server_default=""
),
sa.Column("sku_code", sa.String(length=128), nullable=False, server_default=""),
sa.Column("sku_name", sa.String(length=512), nullable=False, server_default=""),
# SKU 定位键,取自 order_items.spec_json(规划 §3.6)。
sa.Column("seller_id", sa.String(length=64), nullable=False, server_default=""),
sa.Column(
"rel_item_id", sa.String(length=64), nullable=False, server_default=""
),
sa.Column(
"rel_sku_id", sa.String(length=64), nullable=False, server_default=""
),
sa.Column("sku_nick", sa.String(length=512), nullable=False, server_default=""),
sa.Column(
"item_title", sa.String(length=512), nullable=False, server_default=""
),
sa.Column("quantity", sa.Integer(), nullable=True),
# route ∈ {unhandled, auto_fulfilled, worker_platform};按订单号维度聚合得出。
sa.Column("route", sa.String(length=32), nullable=False, server_default=""),
sa.Column("payload", sa.JSON(), nullable=True),
sa.Column("fetched_at", sa.DateTime(), nullable=False),
)
# 按订单号查快照的入口(适配层唯一的查询形态)。
op.create_index(
"ix_external_order_cache_platform_order_id",
"external_order_cache",
["platform_order_id"],
)
op.create_index(
"ix_external_order_cache_fetched_at",
"external_order_cache",
["fetched_at"],
)
def downgrade() -> None:
bind = op.get_bind()
if "external_order_cache" not in _tables(bind):
return
# MySQL 删表会连同索引一起删,不必逐个 drop_index。
op.drop_table("external_order_cache")
+35
View File
@@ -741,3 +741,38 @@ class AutomationRun(Base):
skipped: Mapped[int] = mapped_column(Integer, default=0, server_default="0")
message: Mapped[str] = mapped_column(String(512), default="", server_default="")
detail: Mapped[dict | None] = mapped_column(JSON, nullable=True)
class ExternalOrderCache(Base):
"""order_site 订单快照(P0 适配层的本地缓存)。
主键是 (platform_order_id, source_order_id) 两列:同一个订单号在 order_site
里可能有两行(发货方内部行 + 真实店铺行,实测 541 组,见规划 §6.5.1),
按订单号单列存会互相覆盖。
只作快照与降级用,24 小时后重查;不参与任何写业务。
"""
__tablename__ = "external_order_cache"
platform_order_id: Mapped[str] = mapped_column(
String(64), primary_key=True, index=True
)
source_order_id: Mapped[str] = mapped_column(String(64), primary_key=True)
platform: Mapped[str] = mapped_column(String(32), default="", server_default="")
provider: Mapped[str] = mapped_column(String(32), default="", server_default="")
shop_id: Mapped[str] = mapped_column(String(64), default="", server_default="")
order_status: Mapped[str] = mapped_column(String(32), default="", server_default="")
sku_code: Mapped[str] = mapped_column(String(128), default="", server_default="")
sku_name: Mapped[str] = mapped_column(String(512), default="", server_default="")
seller_id: Mapped[str] = mapped_column(String(64), default="", server_default="")
rel_item_id: Mapped[str] = mapped_column(String(64), default="", server_default="")
rel_sku_id: Mapped[str] = mapped_column(String(64), default="", server_default="")
sku_nick: Mapped[str] = mapped_column(String(512), default="", server_default="")
item_title: Mapped[str] = mapped_column(String(512), default="", server_default="")
quantity: Mapped[int | None] = mapped_column(Integer, nullable=True)
route: Mapped[str] = mapped_column(String(32), default="", server_default="")
payload: Mapped[dict | None] = mapped_column(JSON, nullable=True)
fetched_at: Mapped[datetime] = mapped_column(
UTCDateTime, default=_utcnow, index=True
)
@@ -0,0 +1,378 @@
"""order_site 订单读取适配层(P0)。
**全项目唯一允许出现跨库 SQL 的地方。** order_site 是别人的生产库,表结构
随时可能变;把所有 PG 查询收在这一个文件,将来换数据源只改这里。
设计约束(规划 §6.5 / §6.5.1):
- **只读**:账号 `livehub_ro` 本身是 SELECT-only + `default_transaction_read_only`,
这里再叠一层 `statement_timeout`,避免慢查询拖住对方库;
- **极小连接池**:独立引擎、2 条连接,绝不影响本系统主库连接池;
- **可降级**:`ORDER_SITE_LOOKUP_ENABLED=false` 或未配 URL 时整体不查外部,
主流程退化成客服手工填写;
- **route 按订单号维度聚合**:同一个 `platform_order_id` 在 order_site 里有
541 组重复(发货方内部行 + 真实店铺行),任一行已发货即算 `auto_fulfilled`,
否则这 540 个已发货订单会被误判成「没人管」而重复履约。
对外只暴露 :func:`lookup_order`,业务代码只认语义化的 `route`,不认 SQL。
"""
from __future__ import annotations
import json
import os
from datetime import UTC, datetime, timedelta
from typing import Any
from loguru import logger
from sqlalchemy import Engine, create_engine, text
CACHE_TTL = timedelta(hours=24)
ROUTE_UNHANDLED = "unhandled"
ROUTE_AUTO_FULFILLED = "auto_fulfilled"
ROUTE_WORKER_PLATFORM = "worker_platform"
# 发货方内部行:同一订单号下与真实店铺行成对出现,其 sku_code 是内部编码
# (如 suit_shadow_gothic),不能用来回显给买家。
INTERNAL_SHOP_ID = "91kaquan"
_ENGINE: Engine | None = None
_INITIALIZED = False
class OrderLookupUnavailable(Exception):
"""外部订单库不可用(未配置 / 关闭 / 连不上 / 查询失败)。
调用方应据此走降级:允许提交但标记 `order_unverified`,交人工确认。
"""
def _env_flag(name: str, default: bool = False) -> bool:
return os.getenv(name, "").strip().lower() in {"1", "true", "yes", "on"}
def _enabled() -> bool:
"""是否启用外部查询。未配置连接串时一律关闭。"""
if not os.getenv("ORDER_SITE_DATABASE_URL", "").strip():
return False
return _env_flag("ORDER_SITE_LOOKUP_ENABLED", False)
def _normalize_url(url: str) -> str:
"""把 `postgresql://` 归一成带驱动的 `postgresql+psycopg://`。
`.env` 里写的是不带驱动的裸 URL,SQLAlchemy 2 会去猜默认驱动;显式指定
能避免容器里驱动缺失时报出含糊的错误。
"""
if url.startswith("postgresql://"):
return "postgresql+psycopg://" + url[len("postgresql://") :]
return url
def _get_engine() -> Engine:
"""惰性创建(且只创建一次)外部只读引擎。
``_INITIALIZED`` 只在引擎真正建出来之后才置位 —— 否则一次「未配置」就会
把这个模块永久毒化成不可用,之后补上连接串也救不回来。
"""
global _ENGINE, _INITIALIZED
if _ENGINE is not None:
return _ENGINE
raw = os.getenv("ORDER_SITE_DATABASE_URL", "").strip()
if not raw:
raise OrderLookupUnavailable("ORDER_SITE_DATABASE_URL 未配置")
timeout_ms = int(float(os.getenv("ORDER_SITE_LOOKUP_TIMEOUT_SECONDS", "3")) * 1000)
engine = create_engine(
_normalize_url(raw),
pool_size=2,
max_overflow=0,
pool_timeout=timeout_ms / 1000,
pool_recycle=1800,
pool_pre_ping=True,
connect_args={
# 双保险:账号级已设 statement_timeout,这里再按连接设一次。
"options": f"-c statement_timeout={timeout_ms}",
},
)
_ENGINE = engine
_INITIALIZED = True
return _ENGINE
def reset_engine() -> None:
"""丢弃已缓存的引擎(测试或配置变更后用)。"""
global _ENGINE, _INITIALIZED
if _ENGINE is not None:
_ENGINE.dispose()
_ENGINE = None
_INITIALIZED = False
def _fetch_rows(order_no: str) -> list[dict[str, Any]]:
"""按订单号取全部行(通常 1 行,重复订单号 2 行)。
只走 `idx_orders_platform_order_id` 单行索引查询,禁止全表扫描与聚合。
"""
engine = _get_engine()
# 一个订单号最多几行,限 10 防止异常数据把结果撑爆。
sql = text(
"""
SELECT o.id AS source_order_id,
o.platform AS platform,
o.provider AS provider,
o.shop_id AS shop_id,
o.order_status AS order_status,
oi.sku_code AS sku_code,
oi.sku_name AS sku_name,
oi.quantity AS quantity,
oi.spec_json AS spec_json,
EXISTS (SELECT 1 FROM fulfillment_tasks ft WHERE ft.order_id = o.id)
AS has_fulfillment,
EXISTS (SELECT 1 FROM work_orders wo WHERE wo.order_id = o.id)
AS has_work_order
FROM orders o
LEFT JOIN order_items oi ON oi.order_id = o.id
WHERE o.platform_order_id = :order_no
ORDER BY o.id DESC
LIMIT 10
"""
)
try:
with engine.connect() as conn:
return [
dict(row)
for row in conn.execute(sql, {"order_no": order_no}).mappings()
]
except Exception as exc:
logger.warning(f"查询 order_site 失败(订单号 {order_no}):{exc}")
raise
def _route_of(rows: list[dict[str, Any]]) -> str:
"""订单号维度的聚合判定:任一行已处理即算已处理。"""
if any(row["has_fulfillment"] for row in rows):
return ROUTE_AUTO_FULFILLED
if any(row["has_work_order"] for row in rows):
return ROUTE_WORKER_PLATFORM
return ROUTE_UNHANDLED
def _spec_of(spec_json: Any) -> dict[str, Any]:
"""spec_json 可能是 dict(jsonb 已解析)或字符串,统一成 dict。"""
if isinstance(spec_json, dict):
return spec_json
if isinstance(spec_json, str) and spec_json.strip():
import json
try:
loaded = json.loads(spec_json)
except ValueError:
return {}
return loaded if isinstance(loaded, dict) else {}
return {}
def _pick_display_row(rows: list[dict[str, Any]]) -> dict[str, Any]:
"""回显优先取真实店铺行,跳过发货方内部行。"""
for row in rows:
if row.get("shop_id") != INTERNAL_SHOP_ID:
return row
return rows[0]
def _build_result(order_no: str, rows: list[dict[str, Any]]) -> dict[str, Any]:
row = _pick_display_row(rows)
spec = _spec_of(row.get("spec_json"))
return {
"order_no": order_no,
"source_order_id": str(row.get("source_order_id") or ""),
"platform": row.get("platform") or "",
"provider": row.get("provider") or "",
"shop_id": row.get("shop_id") or "",
"order_status": row.get("order_status") or "",
"sku_code": row.get("sku_code") or "",
"sku_name": row.get("sku_name") or "",
"quantity": row.get("quantity"),
"route": _route_of(rows),
# SKU 定位键(规划 §3.6):优先 spec_json,sku_code 兜底。
"seller_id": str(spec.get("sellerId") or ""),
"rel_item_id": str(spec.get("relItemId") or ""),
"rel_sku_id": str(spec.get("relSkuId") or spec.get("skuId") or ""),
"sku_nick": str(spec.get("skuNick") or ""),
"item_title": str(spec.get("itemTitle") or ""),
"cached": False,
"error": None,
}
def lookup_order(order_no: str, use_cache: bool = True) -> dict[str, Any]:
"""按订单号查订单,返回语义化结果。
- 命中缓存且未过期 → 直接返回(`cached=True`);
- 未命中 → 查 PG → 回写缓存;
- 外部不可用 → 抛 :class:`OrderLookupUnavailable`(调用方走降级)。
订单号不存在时返回 `route=unhandled` 且各字段为空,由调用方按
「订单号不存在」提示(见 §6.4)。
"""
order_no = (order_no or "").strip()
if not order_no:
raise OrderLookupUnavailable("订单号为空")
if not _enabled():
raise OrderLookupUnavailable("外部订单查询未启用")
if use_cache:
cached = _read_cache(order_no)
if cached is not None:
return cached
try:
rows = _fetch_rows(order_no)
except Exception as exc:
# 降级保证落在公开入口这一层:外部库任何异常(连不上、超时、权限、
# 表结构变了)都统一成「不可用」,调用方据此走人工确认,而不是把
# 原始异常抛给用户。
logger.warning(f"订单查询不可用({order_no}):{exc}")
raise OrderLookupUnavailable(str(exc)) from exc
if not rows:
raise OrderLookupUnavailable("订单号不存在")
result = _build_result(order_no, rows)
_write_cache(result)
return result
def _read_cache(order_no: str) -> dict[str, Any] | None:
"""读本地快照,未过期才用。缓存读写失败不影响主流程。"""
from web.backend.database import SessionLocal
try:
with SessionLocal() as session:
row = (
session.execute(
text(
"""
SELECT source_order_id, platform, provider, shop_id,
order_status, sku_code, sku_name, quantity, route,
seller_id, rel_item_id, rel_sku_id, sku_nick,
item_title, fetched_at
FROM external_order_cache
WHERE platform_order_id = :order_no
ORDER BY fetched_at DESC
LIMIT 1
"""
),
{"order_no": order_no},
)
.mappings()
.first()
)
except Exception as exc: # noqa: BLE001 - 缓存不可用就当没命中
logger.warning(f"读取订单快照失败:{exc}")
return None
if row is None:
return None
fetched_at = row["fetched_at"]
if fetched_at is not None:
if fetched_at.tzinfo is None:
fetched_at = fetched_at.replace(tzinfo=UTC)
if datetime.now(UTC) - fetched_at > CACHE_TTL:
return None
return {
"order_no": order_no,
"source_order_id": str(row["source_order_id"] or ""),
"platform": row["platform"] or "",
"provider": row["provider"] or "",
"shop_id": row["shop_id"] or "",
"order_status": row["order_status"] or "",
"sku_code": row["sku_code"] or "",
"sku_name": row["sku_name"] or "",
"quantity": row["quantity"],
"route": row["route"] or "",
"seller_id": row["seller_id"] or "",
"rel_item_id": row["rel_item_id"] or "",
"rel_sku_id": row["rel_sku_id"] or "",
"sku_nick": row["sku_nick"] or "",
"item_title": row["item_title"] or "",
"cached": True,
"error": None,
}
def _write_cache(result: dict[str, Any]) -> None:
"""回写快照。写失败只记日志 —— 缓存不是业务数据。"""
from web.backend.database import SessionLocal
sql = text(
"""
INSERT INTO external_order_cache (
platform_order_id, source_order_id, platform, provider, shop_id,
order_status, sku_code, sku_name, quantity, route,
seller_id, rel_item_id, rel_sku_id, sku_nick, item_title,
payload, fetched_at
) VALUES (
:platform_order_id, :source_order_id, :platform, :provider, :shop_id,
:order_status, :sku_code, :sku_name, :quantity, :route,
:seller_id, :rel_item_id, :rel_sku_id, :sku_nick, :item_title,
:payload, :fetched_at
)
ON DUPLICATE KEY UPDATE
platform = VALUES(platform),
provider = VALUES(provider),
shop_id = VALUES(shop_id),
order_status = VALUES(order_status),
sku_code = VALUES(sku_code),
sku_name = VALUES(sku_name),
quantity = VALUES(quantity),
route = VALUES(route),
seller_id = VALUES(seller_id),
rel_item_id = VALUES(rel_item_id),
rel_sku_id = VALUES(rel_sku_id),
sku_nick = VALUES(sku_nick),
item_title = VALUES(item_title),
payload = VALUES(payload),
fetched_at = VALUES(fetched_at)
"""
)
payload = json.dumps(
{
"platform": result["platform"],
"provider": result["provider"],
"shop_id": result["shop_id"],
"order_status": result["order_status"],
},
ensure_ascii=False,
)
params = {
"platform_order_id": result["order_no"],
"source_order_id": result["source_order_id"],
"platform": result["platform"],
"provider": result["provider"],
"shop_id": result["shop_id"],
"order_status": result["order_status"],
"sku_code": result["sku_code"],
"sku_name": result["sku_name"],
"quantity": result["quantity"],
"route": result["route"],
"seller_id": result["seller_id"],
"rel_item_id": result["rel_item_id"],
"rel_sku_id": result["rel_sku_id"],
"sku_nick": result["sku_nick"],
"item_title": result["item_title"],
"payload": payload,
"fetched_at": datetime.now(UTC).replace(tzinfo=None),
}
try:
with SessionLocal() as session:
session.execute(sql, params)
session.commit()
except Exception as exc: # noqa: BLE001
logger.warning(f"写入订单快照失败:{exc}")