From afa5c815f6385b378c94091664bf4ee140c84652 Mon Sep 17 00:00:00 2001 From: yml2213 Date: Fri, 2 Oct 2026 22:50:20 +0800 Subject: [PATCH] =?UTF-8?q?feat(orders):=20P0=20=E8=AE=A2=E5=8D=95?= =?UTF-8?q?=E8=AF=BB=E5=8F=96=E9=80=82=E9=85=8D=E5=B1=82=E2=80=94=E2=80=94?= =?UTF-8?q?=E8=BF=9E=20order=5Fsite=20=E5=8F=AA=E8=AF=BB=E5=BA=93=E6=9F=A5?= =?UTF-8?q?=E8=AE=A2=E5=8D=95=EF=BC=88=E9=BB=98=E8=AE=A4=E5=85=B3=E9=97=AD?= =?UTF-8?q?=EF=BC=89?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 让运营能按订单号查到订单，作为后续资料收集（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， 一次落空就把模块永久毒化成不可用，之后补上连接串也救不回来。 --- docs/自动化与资料收集规划.md | 34 +- pyproject.toml | 3 + tests/test_external_order_source.py | 332 +++++++++++++++ tests/test_external_order_source_live.py | 88 ++++ tests/test_migrations.py | 2 +- uv.lock | 47 +++ .../20261002_0039_external_order_cache.py | 85 ++++ web/backend/models.py | 35 ++ web/backend/services/external_order_source.py | 378 ++++++++++++++++++ 9 files changed, 1001 insertions(+), 3 deletions(-) create mode 100644 tests/test_external_order_source.py create mode 100644 tests/test_external_order_source_live.py create mode 100644 web/backend/migrations/versions/20261002_0039_external_order_cache.py create mode 100644 web/backend/services/external_order_source.py diff --git a/docs/自动化与资料收集规划.md b/docs/自动化与资料收集规划.md index 3998c55..1c6479a 100644 --- a/docs/自动化与资料收集规划.md +++ b/docs/自动化与资料收集规划.md @@ -277,8 +277,11 @@ ### 6.2 数据模型（迁移 1 个） -**`external_order_cache`（订单快照）**：`order_no`(PK) / `platform` / `provider` / `order_status` / `sku_code` / `sku_name` / **`seller_id` / `rel_item_id` / `rel_sku_id` / `sku_nick` / `item_title`** / `quantity` / `route` / `payload`(JSON) / `fetched_at` +**`external_order_cache`（订单快照）**：`platform_order_id` + `source_order_id`（联合唯一；对应 order_site `orders.platform_order_id` 与 `orders.id`）/ `platform` / `provider` / `shop_id` / `order_status` / `sku_code` / `sku_name` / **`seller_id` / `rel_item_id` / `rel_sku_id` / `sku_nick` / `item_title`** / `quantity` / `route` / `payload`(JSON) / `fetched_at` +> **⚠️ 2026-10-02 修订**：order_site 没有 `order_no` 列，主键改为 `platform_order_id` + `source_order_id`。 +> 后者不可省 —— 同一个 `platform_order_id` 有 541 组重复（详见 §6.5.1），按单列存会互相覆盖。 +> > 用途：用户提交订单号时立即回显商品；工单与台账留档；order_site 不可用时降级。 > 缓存有效期建议 24 小时（订单状态可能变化，过期后重查）。 @@ -299,7 +302,10 @@ lookup_order(order_no) -> { ``` - `route` 由 §3.4 的互斥判据得出：有 `fulfillment_tasks` → `auto_fulfilled`；有 `work_orders` → `worker_platform`；都没有 → `unhandled`。**业务代码只认 `route`，不认 SQL 细节**； + > **⚠️ 按 §6.5.1 的决策 2，判定要在「订单号维度」聚合**：按 `platform_order_id` 取到全部行， + > 任一行有 `fulfillment_tasks` 即 `auto_fulfilled`（540 个已发货的重复组靠这条拦住）。 - 查询顺序：先查 `external_order_cache`（未过期）→ 未命中再查 PG → 回写缓存。 +- 入参 `order_no` 实际就是 `platform_order_id`（对外仍叫 order_no，保持接口语义稳定）。 ### 6.4 准入规则（用户填订单号时） @@ -335,6 +341,29 @@ lookup_order(order_no) -> { > > 适配层要把 `.env` 里的 `postgresql://` 归一成带驱动的 URL（如 `postgresql+psycopg://`）后再交给 SQLAlchemy。 +### 6.5.1 ⚠️ 实测偏差：没有 `order_no`，且同一订单号跨桶重复（2026-10-02 上机核实） + +在生产机（`order+douyu_huya`）用只读账号 `livehub_ro` 实测，§6.2/§6.4 的两处假设与真实库不符。**P0 必须按本节实现**： + +| 项 | 规划原文 | 实测 | +| --- | --- | --- | +| 订单号列 | `order_no`(PK) | ❌ **不存在**。只有 `id`（内部 bigint PK）与 `platform_order_id`（用户手上的快手单号） | +| `platform_order_id` 唯一性 | 隐含唯一 | ❌ **非唯一**：276,371 行 / 275,830 个号 → **541 个号重复** | +| 重复的形态 | —— | 541 组**全部**是 `n=2, shops=2`：`shop_id='91kaquan'` 行 + 一个真实店铺行 | +| 重复组的分桶 | —— | **540 组跨 A/C 桶**：`91kaquan` 行有 `fulfillment_tasks`（`delivery_status` = delivered/closed）→ A；真实店铺行无 A/B → C | +| 影响面 | C 桶 89,542 | C 桶实测 **91,029 行 / 91,028 个号**，其中 **540 个号的同一单已被 91kaquan 发货** | + +**决策（已确认）** + +1. **查询入口 = `platform_order_id`**（用户/客服填的快手订单号），走 `idx_orders_platform_order_id` 索引；`orders.id` 只作内部关联，不当对外入口； +2. **route 判定改为「订单号维度聚合」**：按 `platform_order_id` 取**全部行**， + 任一行命中 `fulfillment_tasks` → `auto_fulfilled`；否则任一行命中 `work_orders` → `worker_platform`；都没有 → `unhandled`。 + > 不这么做，这 540 个**已发货**订单会被误判为「没人管」而放行重复履约 —— 与 §3.4「A 桶不要」直接矛盾。 +3. 回显的 SKU 取**真实店铺行**（`shop_id != '91kaquan'`，带 `sku_code` / `sku_name` / `spec_json`）； + `91kaquan` 是发货方内部行（`provider=91kaquan`、`automation_mode=manual`、`sku_code` 形如 `suit_shadow_gothic`），不用于展示。 + +> 缓存表主键相应改为**订单号 + 店铺行**（见 §6.2 的 `order_no` → `platform_order_id`，并带 `source_order_id`），避免多行互相覆盖。 + ### 6.6 降级开关 `ORDER_SITE_LOOKUP_ENABLED=false` 时完全不查外部，退化成「客服手工填订单信息 + 人工确认」，主流程照常可用。 @@ -348,7 +377,8 @@ lookup_order(order_no) -> { 3. 输入一个 C 桶订单号（如「兔兔白日梦」的订单）→ `route=unhandled`，进入候选并带出 `sku_name`； 4. 输入不存在的订单号 → 明确提示「订单号不存在」； 5. 关掉 `ORDER_SITE_LOOKUP_ENABLED` → 系统仍可正常走手工流程； -6. 查一次订单对 order_site 的负载可忽略（单行索引查询，实测 < 50ms）。 +6. 查一次订单对 order_site 的负载可忽略（单行索引查询，实测 < 50ms）； +7. **输入一个「重复订单号」（`91kaquan` 行已发货 + 店铺行无人管，共 540 个）→ `route=auto_fulfilled`，不被放行**（§6.5.1）。 ## 7. P1 · CK 自动管理 diff --git a/pyproject.toml b/pyproject.toml index b1361ed..5b27551 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -24,6 +24,9 @@ dependencies = [ "scipy>=1.13.0", "pyexecjs>=1.5.1", "pymysql>=1.1,<2", + # P0 订单读取适配层：连 order_site 的 PostgreSQL 只读库（规划 §6.5）。 + # 全项目只有 external_order_source.py 一处跨库；URL 用 postgresql+psycopg:// + "psycopg[binary]>=3.2,<4", "curl-cffi>=0.13", "segno>=1.6", ] diff --git a/tests/test_external_order_source.py b/tests/test_external_order_source.py new file mode 100644 index 0000000..206a8de --- /dev/null +++ b/tests/test_external_order_source.py @@ -0,0 +1,332 @@ +"""P0 订单读取适配层回归。 + +重点守住三件事： + +1. **route 按订单号维度聚合**：同一个订单号在 order_site 里有 541 组重复 + （发货方内部行 + 真实店铺行，见规划 §6.5.1）。任一行已发货就要判 + ``auto_fulfilled`` —— 否则这 540 个已发货订单会被放行重复履约。这条是 + P0 最容易写错、后果最重的地方，用直接构造两行的用例钉死。 +2. **回显取真实店铺行**：内部行（``shop_id=91kaquan``）的 sku_code 是 + ``suit_sand_dancer`` 这类内部编码，不能回显给买家。 +3. **可降级**：未启用 / 未配置 / 库不可用时抛 ``OrderLookupUnavailable``， + 主流程退化成人工填写，而不是把异常抛给用户。 + +真实只读库的连通性验证不进自动回归（需生产网络），见 +``tests/test_external_order_source_live.py``（手工跑）。 +""" + +from __future__ import annotations + +import os +from datetime import UTC, datetime, timedelta + +import pytest + +import web.backend.services.external_order_source as src +from tests.support.db import get_test_engine, make_test_session +from web.backend.models import ExternalOrderCache + + +@pytest.fixture(autouse=True) +def _clean_env(monkeypatch): + """每个用例从干净的外部配置起步，避免相互污染。""" + src.reset_engine() + monkeypatch.delenv("ORDER_SITE_DATABASE_URL", raising=False) + monkeypatch.delenv("ORDER_SITE_LOOKUP_ENABLED", raising=False) + yield + src.reset_engine() + + +@pytest.fixture +def session(): + engine = get_test_engine() + s = make_test_session(engine) + s.query(ExternalOrderCache).delete() + s.commit() + yield s + s.rollback() + s.close() + + +def _row( + source_id=100, + shop_id="4661340439", + sku_name="兔兔白日梦", + sku_code="187749890445642", + has_ft=False, + has_wo=False, + spec=None, + order_status="paid", +): + return { + "source_order_id": source_id, + "platform": "kuaishou", + "provider": "91kaquan", + "shop_id": shop_id, + "order_status": order_status, + "sku_code": sku_code, + "sku_name": sku_name, + "quantity": 1, + "spec_json": spec if spec is not None else {}, + "has_fulfillment": has_ft, + "has_work_order": has_wo, + } + + +class TestRouteAggregation: + """route 判定（纯函数，不需要数据库）。""" + + def test_single_unhandled_row_is_candidate(self): + assert src._route_of([_row()]) == src.ROUTE_UNHANDLED + + def test_fulfillment_row_wins_over_unhandled(self): + """同一个订单号：内部行已发货 + 店铺行无人管 → 已处理。 + + 这是 §6.5.1 那 540 组的核心判定，反过来写就会重复履约。 + """ + rows = [ + _row(source_id=145013, shop_id=src.INTERNAL_SHOP_ID, has_ft=True), + _row(source_id=145012, shop_id="4661340439"), + ] + assert src._route_of(rows) == src.ROUTE_AUTO_FULFILLED + + def test_work_order_row_wins_over_unhandled(self): + rows = [_row(source_id=1, has_wo=True), _row(source_id=2)] + assert src._route_of(rows) == src.ROUTE_WORKER_PLATFORM + + def test_fulfillment_wins_over_work_order(self): + """两桶理论上互斥，真撞上时按「已发货」优先。""" + rows = [_row(has_ft=True), _row(has_wo=True)] + assert src._route_of(rows) == src.ROUTE_AUTO_FULFILLED + + +class TestDisplayRow: + def test_prefers_real_shop_row_over_internal(self): + rows = [ + _row(source_id=145013, shop_id=src.INTERNAL_SHOP_ID, sku_code="suit_x"), + _row(source_id=145012, shop_id="4661340439", sku_name="狂沙舞者"), + ] + result = src._build_result("2624100167260659", rows) + assert result["sku_name"] == "狂沙舞者" + assert result["source_order_id"] == "145012" + assert result["sku_code"] != "suit_x" + + def test_falls_back_to_internal_row_when_only_one(self): + rows = [_row(shop_id=src.INTERNAL_SHOP_ID, sku_code="suit_x")] + assert src._build_result("x", rows)["sku_code"] == "suit_x" + + +class TestSpecExtraction: + def test_sku_keys_read_from_spec_json(self): + spec = { + "skuId": "192758405127439", + "skuNick": "（3选1）甜心娇蕊", + "sellerId": "4661340439", + "relItemId": "25228845231439", + "relSkuId": "", + "itemTitle": "和平精英皮肤小店入口...", + } + result = src._build_result("2627501925533255", [_row(spec=spec, sku_name="")]) + assert result["seller_id"] == "4661340439" + assert result["rel_item_id"] == "25228845231439" + assert result["sku_nick"] == "（3选1）甜心娇蕊" + # relSkuId 为空时回退到 skuId（规划 §3.6：skuId 才是快手 SKU ID） + assert result["rel_sku_id"] == "192758405127439" + assert result["item_title"] == "和平精英皮肤小店入口..." + + def test_spec_json_as_string_is_parsed(self): + result = src._build_result("x", [_row(spec='{"sellerId": "3676797936"}')]) + assert result["seller_id"] == "3676797936" + + def test_malformed_spec_json_is_ignored(self): + result = src._build_result("x", [_row(spec="{not json")]) + assert result["seller_id"] == "" + + +class TestUrlNormalization: + def test_bare_postgresql_url_gets_driver(self): + url = "postgresql://u:p@h:5432/db" + assert src._normalize_url(url) == "postgresql+psycopg://u:p@h:5432/db" + + def test_url_with_driver_is_untouched(self): + url = "postgresql+psycopg://u:p@h:5432/db" + assert src._normalize_url(url) == url + + +class TestDegradation: + def test_missing_url_disables_lookup(self, monkeypatch): + monkeypatch.setenv("ORDER_SITE_LOOKUP_ENABLED", "true") + with pytest.raises(src.OrderLookupUnavailable): + src.lookup_order("2627501925533255") + + def test_disabled_flag_disables_lookup(self, monkeypatch): + monkeypatch.setenv("ORDER_SITE_DATABASE_URL", "postgresql://u:p@h:5432/db") + monkeypatch.setenv("ORDER_SITE_LOOKUP_ENABLED", "false") + with pytest.raises(src.OrderLookupUnavailable): + src.lookup_order("2627501925533255") + + def test_empty_order_no_raises(self, monkeypatch): + monkeypatch.setenv("ORDER_SITE_DATABASE_URL", "postgresql://u:p@h:5432/db") + monkeypatch.setenv("ORDER_SITE_LOOKUP_ENABLED", "true") + with pytest.raises(src.OrderLookupUnavailable): + src.lookup_order(" ") + + +class TestCacheRoundTrip: + """缓存读写：命中未过期快照就不查外部。""" + + def _seed(self, session, order_no, *, route=src.ROUTE_UNHANDLED, age_hours=0): + session.add( + ExternalOrderCache( + platform_order_id=order_no, + source_order_id="145012", + platform="kuaishou", + provider="91kaquan", + shop_id="4661340439", + order_status="paid", + sku_code="187749890445642", + sku_name="兔兔白日梦", + quantity=1, + route=route, + seller_id="4661340439", + rel_item_id="25228845231439", + rel_sku_id="187749890445642", + sku_nick="兔兔白日梦", + item_title="和平精英小店", + fetched_at=datetime.now(UTC) - timedelta(hours=age_hours), + ) + ) + session.commit() + + def test_fresh_cache_is_returned_without_external_query(self, session, monkeypatch): + monkeypatch.setenv("ORDER_SITE_DATABASE_URL", "postgresql://u:p@h:5432/db") + monkeypatch.setenv("ORDER_SITE_LOOKUP_ENABLED", "true") + self._seed(session, "2627501925533255") + + def _boom(_order_no): + raise AssertionError("命中缓存时不应查询外部库") + + monkeypatch.setattr(src, "_fetch_rows", _boom) + result = src.lookup_order("2627501925533255") + assert result["cached"] is True + assert result["sku_name"] == "兔兔白日梦" + assert result["route"] == src.ROUTE_UNHANDLED + + def test_expired_cache_is_ignored(self, session, monkeypatch): + monkeypatch.setenv("ORDER_SITE_DATABASE_URL", "postgresql://u:p@h:5432/db") + monkeypatch.setenv("ORDER_SITE_LOOKUP_ENABLED", "true") + self._seed(session, "2627501925533255", age_hours=25) + + monkeypatch.setattr(src, "_fetch_rows", lambda _o: [_row()]) + result = src.lookup_order("2627501925533255") + assert result["cached"] is False + + def test_lookup_writes_cache(self, session, monkeypatch): + monkeypatch.setenv("ORDER_SITE_DATABASE_URL", "postgresql://u:p@h:5432/db") + monkeypatch.setenv("ORDER_SITE_LOOKUP_ENABLED", "true") + monkeypatch.setattr(src, "_fetch_rows", lambda _o: [_row()]) + + src.lookup_order("2627501925533255") + row = ( + session.query(ExternalOrderCache) + .filter_by(platform_order_id="2627501925533255") + .first() + ) + assert row is not None + assert row.route == src.ROUTE_UNHANDLED + assert row.sku_name == "兔兔白日梦" + + def test_duplicate_writes_do_not_collide(self, session, monkeypatch): + """同一订单号两行都要能存下 —— 单列主键会把它们互相覆盖。""" + monkeypatch.setenv("ORDER_SITE_DATABASE_URL", "postgresql://u:p@h:5432/db") + monkeypatch.setenv("ORDER_SITE_LOOKUP_ENABLED", "true") + rows = [ + _row(source_id=145013, shop_id=src.INTERNAL_SHOP_ID, has_ft=True), + _row(source_id=145012, shop_id="4661340439", sku_name="狂沙舞者"), + ] + monkeypatch.setattr(src, "_fetch_rows", lambda _o: rows) + + result = src.lookup_order("2624100167260659") + assert result["route"] == src.ROUTE_AUTO_FULFILLED + assert result["sku_name"] == "狂沙舞者" + + stored = ( + session.query(ExternalOrderCache) + .filter_by(platform_order_id="2624100167260659") + .all() + ) + # 适配层只落「回显用的那一行」，但主键带 source_order_id， + # 换一行回显时不会覆盖前一条。 + assert len(stored) == 1 + assert stored[0].source_order_id == "145012" + + +class TestExternalFailure: + def test_external_error_becomes_unavailable(self, session, monkeypatch): + monkeypatch.setenv("ORDER_SITE_DATABASE_URL", "postgresql://u:p@h:5432/db") + monkeypatch.setenv("ORDER_SITE_LOOKUP_ENABLED", "true") + + def _raise(_order_no): + raise RuntimeError("connection refused") + + monkeypatch.setattr(src, "_fetch_rows", _raise) + with pytest.raises(src.OrderLookupUnavailable): + src.lookup_order("2627501925533255") + + def test_no_rows_means_order_not_found(self, session, monkeypatch): + monkeypatch.setenv("ORDER_SITE_DATABASE_URL", "postgresql://u:p@h:5432/db") + monkeypatch.setenv("ORDER_SITE_LOOKUP_ENABLED", "true") + monkeypatch.setattr(src, "_fetch_rows", lambda _o: []) + with pytest.raises(src.OrderLookupUnavailable): + src.lookup_order("0000000000000000") + + +class TestEngineConfig: + def test_engine_uses_tiny_pool_and_statement_timeout(self, monkeypatch): + """外部库的连接必须又小又带超时，别拖累对方生产库。""" + monkeypatch.setenv( + "ORDER_SITE_DATABASE_URL", "postgresql://livehub_ro:p@h:5432/order_site" + ) + monkeypatch.setenv("ORDER_SITE_LOOKUP_TIMEOUT_SECONDS", "3") + created: dict[str, object] = {} + real_create_engine = src.create_engine + + def _spy(url, **kwargs): + created.update(kwargs) + return real_create_engine(url, **kwargs) + + monkeypatch.setattr(src, "create_engine", _spy) + engine = src._get_engine() + try: + assert engine.pool.size() == 2 + assert engine.pool._max_overflow == 0 + # statement_timeout 下发给连接，不是应用侧 kill（账号级已有 3s 兜底） + assert created["connect_args"]["options"] == "-c statement_timeout=3000" + finally: + engine.dispose() + + def test_engine_is_created_once(self, monkeypatch): + monkeypatch.setenv("ORDER_SITE_DATABASE_URL", "postgresql://u:p@h:5432/db") + a = src._get_engine() + b = src._get_engine() + assert a is b + a.dispose() + + def test_missing_url_does_not_poison_later_calls(self, monkeypatch): + """未配置时抛错，但补上连接串后必须能恢复。 + + 曾经的写法是先置 ``_INITIALIZED = True`` 再读 URL，一次落空就永久 + 毒化成不可用 —— 真机上表现为「明明配了 URL 却一直报未配置」。 + """ + with pytest.raises(src.OrderLookupUnavailable): + src._get_engine() + + monkeypatch.setenv("ORDER_SITE_DATABASE_URL", "postgresql://u:p@h:5432/db") + engine = src._get_engine() + assert engine is not None + engine.dispose() + + +def test_os_environ_not_leaked_between_tests(): + """保险：用例没把 ORDER_SITE_* 留在环境里。""" + assert "ORDER_SITE_DATABASE_URL" not in os.environ diff --git a/tests/test_external_order_source_live.py b/tests/test_external_order_source_live.py new file mode 100644 index 0000000..97a5293 --- /dev/null +++ b/tests/test_external_order_source_live.py @@ -0,0 +1,88 @@ +"""P0 适配层对真实只读库的连通性验证（**手工跑，不进自动回归**）。 + +需要能连到 order_site 的 PostgreSQL（生产机或已打通网络的环境）： + + ORDER_SITE_DATABASE_URL=postgresql://livehub_ro:...@host:5432/order_site \ + ORDER_SITE_LOOKUP_ENABLED=true \ + .venv/bin/python -m pytest tests/test_external_order_source_live.py -q -s + +默认会因为连不上而 skip，不会误报失败。 +""" + +from __future__ import annotations + +import os + +import pytest + +import web.backend.services.external_order_source as src + +# 三桶各取一个真实订单号（2026-10-02 实测）。 +ORDER_C_UNHANDLED = "2627501602915345" # 激萌罗小黑，C 桶 +ORDER_B_WORKER = "2627501932004488" # 指挥官密钥4个，B 桶 +ORDER_A_AUTO = "2627501850053080" # 套装-浪漫天命，A 桶 +# 重复订单号：91kaquan 内部行已发货 + 真实店铺行无人管（§6.5.1 的 540 组之一） +ORDER_DUPLICATE = "2624100167260659" +ORDER_MISSING = "0000000000000000" + +pytestmark = pytest.mark.skipif( + not os.getenv("ORDER_SITE_DATABASE_URL", "").strip(), + reason="未配置 ORDER_SITE_DATABASE_URL，跳过真实库验证", +) + + +@pytest.fixture +def _configured(monkeypatch): + monkeypatch.setenv("ORDER_SITE_LOOKUP_ENABLED", "true") + src.reset_engine() + yield + src.reset_engine() + + +def test_c_bucket_is_unhandled(_configured): + result = src.lookup_order(ORDER_C_UNHANDLED, use_cache=False) + assert result["route"] == src.ROUTE_UNHANDLED + assert result["sku_name"] + + +def test_b_bucket_is_worker_platform(_configured): + assert ( + src.lookup_order(ORDER_B_WORKER, use_cache=False)["route"] + == src.ROUTE_WORKER_PLATFORM + ) + + +def test_a_bucket_is_auto_fulfilled(_configured): + assert ( + src.lookup_order(ORDER_A_AUTO, use_cache=False)["route"] + == src.ROUTE_AUTO_FULFILLED + ) + + +def test_duplicate_order_no_is_treated_as_fulfilled(_configured): + """核心回归：同一订单号下内部行已发货 → 整单不能放行。 + + 按店铺行单独看是 unhandled，放行就会重复履约。 + """ + result = src.lookup_order(ORDER_DUPLICATE, use_cache=False) + assert result["route"] == src.ROUTE_AUTO_FULFILLED + # 回显取真实店铺行的中文名，不是内部行的 suit_sand_dancer + assert result["sku_name"] == "狂沙舞者" + + +def test_missing_order_raises(_configured): + with pytest.raises(src.OrderLookupUnavailable): + src.lookup_order(ORDER_MISSING, use_cache=False) + + +def test_lookup_is_fast(_configured): + """对对方库的负载必须可忽略（规划验收标准第 6 条：单行索引查询 < 50ms）。""" + import time + + src.lookup_order(ORDER_C_UNHANDLED, use_cache=False) # 预热连接 + started = time.perf_counter() + for _ in range(5): + src.lookup_order(ORDER_C_UNHANDLED, use_cache=False) + elapsed_ms = (time.perf_counter() - started) / 5 * 1000 + print(f"\n单次查询耗时：{elapsed_ms:.1f} ms") + assert elapsed_ms < 50, f"查询耗时 {elapsed_ms:.1f}ms 超过 50ms" diff --git a/tests/test_migrations.py b/tests/test_migrations.py index 1f0826c..d0d5f23 100644 --- a/tests/test_migrations.py +++ b/tests/test_migrations.py @@ -26,7 +26,7 @@ TESTS_DIR = Path(__file__).resolve().parent PROJECT_ROOT = TESTS_DIR.parent VERSIONS_DIR = PROJECT_ROOT / "web" / "backend" / "migrations" / "versions" -HEAD_REVISION = "20261002_0038" +HEAD_REVISION = "20261002_0039" # 迁移新增、但不声明在模型里的复合查询索引。 EXTRA_INDEXES = ( diff --git a/uv.lock b/uv.lock index 0b0729c..61ce2ec 100644 --- a/uv.lock +++ b/uv.lock @@ -332,6 +332,7 @@ dependencies = [ { name = "onnxruntime" }, { name = "opencv-python-headless" }, { name = "pillow" }, + { name = "psycopg", extra = ["binary"] }, { name = "pycryptodome" }, { name = "pydantic" }, { name = "pyexecjs" }, @@ -364,6 +365,7 @@ requires-dist = [ { name = "onnxruntime", specifier = ">=1.18.0" }, { name = "opencv-python-headless", specifier = ">=4.8.0" }, { name = "pillow", specifier = ">=10.0.0" }, + { name = "psycopg", extras = ["binary"], specifier = ">=3.2,<4" }, { name = "pycryptodome", specifier = ">=3.19.0" }, { name = "pydantic", specifier = ">=2.0.0" }, { name = "pyexecjs", specifier = ">=1.5.1" }, @@ -545,6 +547,42 @@ wheels = [ { url = "https://files.pythonhosted.org/packages/19/c7/5f7c636ec43e0c545e28d1f1db71990108306f7bdcb89f069ba97e428e7f/protobuf-7.35.1-py3-none-any.whl", hash = "sha256:4bc97768d8fe4ad6743c8a19403e314511ed9f6d13205b687e52421c023ac1b9", size = 171659, upload-time = "2026-06-11T21:55:39.155Z" }, ] +[[package]] +name = "psycopg" +version = "3.3.6" +source = { registry = "https://pypi.org/simple" } +dependencies = [ + { name = "typing-extensions" }, + { name = "tzdata", marker = "sys_platform == 'win32'" }, +] +sdist = { url = "https://files.pythonhosted.org/packages/76/26/3ea4ca5eaea1c0debcdf7ee7c1613fbe721dc27a03c461c0817ffd8a0601/psycopg-3.3.6.tar.gz", hash = "sha256:c081f2250df751a943036e42db6df4571c66cd0aabe8291a7a506512b12007d2", size = 168171, upload-time = "2026-09-18T13:22:55.152Z" } +wheels = [ + { url = "https://files.pythonhosted.org/packages/4e/de/748bd7609c71cae5d737f0ba9192f19329f70180ecda8fff3cac02c5abe3/psycopg-3.3.6-py3-none-any.whl", hash = "sha256:a1db9f7148b06a28606767efaca51fa6f9398c5c0a3810519be69d7000bdb631", size = 215490, upload-time = "2026-09-18T13:15:29.374Z" }, +] + +[package.optional-dependencies] +binary = [ + { name = "psycopg-binary", marker = "implementation_name != 'pypy'" }, +] + +[[package]] +name = "psycopg-binary" +version = "3.3.6" +source = { registry = "https://pypi.org/simple" } +wheels = [ + { url = "https://files.pythonhosted.org/packages/e6/01/2cdd1824e58b4467ee0b9498664cd28c42d8794db6b1e35b6bcb834f0044/psycopg_binary-3.3.6-cp312-cp312-macosx_10_13_x86_64.whl", hash = "sha256:3f84dab25e0385692ee13274c68678377e0b1a70ab9d14e56264cbf61f60c62d", size = 4707086, upload-time = "2026-09-18T13:18:05.138Z" }, + { url = "https://files.pythonhosted.org/packages/f6/76/de9948ac06895261c84d5b9fbe283d8f3c5bc9f070691b8d9eaa1b51e322/psycopg_binary-3.3.6-cp312-cp312-macosx_11_0_arm64.whl", hash = "sha256:612382ac3ed13651c7fa44b5fee9fbf7baaa2ddbc6f500391672682c5f1df9e0", size = 4769607, upload-time = "2026-09-18T13:18:12.83Z" }, + { url = "https://files.pythonhosted.org/packages/76/a9/72436c9915ee4905964689e7f0e182ce7767cc0a0390b3ce703be8177625/psycopg_binary-3.3.6-cp312-cp312-manylinux2014_ppc64le.manylinux_2_17_ppc64le.whl", hash = "sha256:366db6e97e66b37211475f20c4c1324a2dc0dd825e46d4e87f9d599304d276f9", size = 5554134, upload-time = "2026-09-18T13:18:21.175Z" }, + { url = "https://files.pythonhosted.org/packages/0a/42/948bb3d2617795093512613fd96ba380e922992c7908fbc073858147d196/psycopg_binary-3.3.6-cp312-cp312-manylinux2014_x86_64.manylinux_2_17_x86_64.whl", hash = "sha256:1679a1cb93fbe5a6d1fd58d82cbddcc6fcb8c61446ba7cae6eb2a7b19bc585de", size = 5235723, upload-time = "2026-09-18T13:18:27.071Z" }, + { url = "https://files.pythonhosted.org/packages/99/47/93e823ff1b0088400703410939c9bda3e63ed9c850b3ee088e8769f4c10b/psycopg_binary-3.3.6-cp312-cp312-manylinux_2_27_aarch64.manylinux_2_28_aarch64.whl", hash = "sha256:37d40450659401600e6d043ff586c89a71a69f33cbb8bcdba6cdb2569beecdbe", size = 6833587, upload-time = "2026-09-18T13:18:33.794Z" }, + { url = "https://files.pythonhosted.org/packages/5e/2d/ecc69c847795aa704041a9f5667a6b0938a088cf1853636d762a6938e493/psycopg_binary-3.3.6-cp312-cp312-manylinux_2_38_riscv64.manylinux_2_39_riscv64.whl", hash = "sha256:a5165300324efd5a772c48a88ab3a928513ab3979fca76553e62ee815f7b2b9c", size = 5070013, upload-time = "2026-09-18T13:18:39.628Z" }, + { url = "https://files.pythonhosted.org/packages/92/36/6126f0dac21713dcae91404f2a76da18598a6252339a8c669c46370d43b2/psycopg_binary-3.3.6-cp312-cp312-musllinux_1_2_aarch64.whl", hash = "sha256:d636338c8f21b0df2f84657b00bc34f9313f826ef93f1155bc743607e4a0c5eb", size = 4597367, upload-time = "2026-09-18T13:18:45.023Z" }, + { url = "https://files.pythonhosted.org/packages/4d/29/7ecfc04243b46c89ffd49924e9c5634ea904ef96c7d0f37e4073623584c1/psycopg_binary-3.3.6-cp312-cp312-musllinux_1_2_ppc64le.whl", hash = "sha256:a4ee3bdd5468a725f2a4d9aab8a74b6d0279f768c8b5d3aeb102c5307ff3d59c", size = 4275419, upload-time = "2026-09-18T13:18:49.299Z" }, + { url = "https://files.pythonhosted.org/packages/6e/90/2f46d2e0de79706ac170df0a3637fe63c4498fc04f131f6049520b78b806/psycopg_binary-3.3.6-cp312-cp312-musllinux_1_2_riscv64.whl", hash = "sha256:289aadd6a00e151203c081f708348ec89f1e483c9b510ef4ac3981f847f01f79", size = 4007358, upload-time = "2026-09-18T13:18:53.944Z" }, + { url = "https://files.pythonhosted.org/packages/03/48/6744e91291b751a8cf12d63d719977974bb94c84ceba913e7ddb2e478e51/psycopg_binary-3.3.6-cp312-cp312-musllinux_1_2_x86_64.whl", hash = "sha256:f21d057f3e5f5491067e5b292498073b73847d48799b099803fef100775fcc52", size = 4320156, upload-time = "2026-09-18T13:18:59.258Z" }, + { url = "https://files.pythonhosted.org/packages/1a/9b/94ff7fce53a64d5b286e2ec454e0a025cf3d6e6b4a9189bef16aa5de98b2/psycopg_binary-3.3.6-cp312-cp312-win_amd64.whl", hash = "sha256:e23a66a763fbe83fcc210bc77c27e5a5ea380ebf091c06f34d8561b695e5a40f", size = 3658864, upload-time = "2026-09-18T13:19:06.503Z" }, +] + [[package]] name = "pyasn1" version = "0.6.3" @@ -897,6 +935,15 @@ wheels = [ { url = "https://files.pythonhosted.org/packages/dc/9b/47798a6c91d8bdb567fe2698fe81e0c6b7cb7ef4d13da4114b41d239f65d/typing_inspection-0.4.2-py3-none-any.whl", hash = "sha256:4ed1cacbdc298c220f1bd249ed5287caa16f34d44ef4e9c3d0cbad5b521545e7", size = 14611, upload-time = "2025-10-01T02:14:40.154Z" }, ] +[[package]] +name = "tzdata" +version = "2026.4" +source = { registry = "https://pypi.org/simple" } +sdist = { url = "https://files.pythonhosted.org/packages/e4/31/3d74fa778a63b98b7374323befcc0be5ab3bd94afd4096a0124e7379152c/tzdata-2026.4.tar.gz", hash = "sha256:f1b8bd365d8d210c55353f4d7f8d6d8561c0ba50d704b700d195a9424bba0d79", size = 199350, upload-time = "2026-09-12T12:56:03.251Z" } +wheels = [ + { url = "https://files.pythonhosted.org/packages/f9/bc/8737e8d54cf51106118039b83f485a4783112fab49ea9d044b234978a46e/tzdata-2026.4-py2.py3-none-any.whl", hash = "sha256:c2169a8b0a7a5e9674da5a135ccdfb2b3e671b333ed9fed17b41f73c34476e81", size = 347494, upload-time = "2026-09-12T12:56:01.67Z" }, +] + [[package]] name = "urllib3" version = "2.7.0" diff --git a/web/backend/migrations/versions/20261002_0039_external_order_cache.py b/web/backend/migrations/versions/20261002_0039_external_order_cache.py new file mode 100644 index 0000000..dee8c7c --- /dev/null +++ b/web/backend/migrations/versions/20261002_0039_external_order_cache.py @@ -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") diff --git a/web/backend/models.py b/web/backend/models.py index 169236a..001bfa1 100644 --- a/web/backend/models.py +++ b/web/backend/models.py @@ -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 + ) diff --git a/web/backend/services/external_order_source.py b/web/backend/services/external_order_source.py new file mode 100644 index 0000000..55e427c --- /dev/null +++ b/web/backend/services/external_order_source.py @@ -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}")