From c030eca1780d1b222ba17a7d2e71fa0ed05fddb1 Mon Sep 17 00:00:00 2001 From: yeuimu <2197651308@qq.com> Date: Mon, 28 Sep 2026 03:02:05 +0000 Subject: [PATCH] =?UTF-8?q?fix:=20RIIN=20=E5=95=86=E5=93=81=E7=8A=B6?= =?UTF-8?q?=E6=80=81=E8=A1=A5=20COMPLETE/PAUSE=20=E6=98=A0=E5=B0=84,?= =?UTF-8?q?=E4=BF=AE=E5=A4=8D=20Goods.status=20=E5=A4=A7=E9=9D=A2=E7=A7=AF?= =?UTF-8?q?=20NULL?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 问题:RIIN 商品级 goodsStatus 复用平台状态枚举,COMPLETE/PAUSE 无映射, .get() 返回 None 落库为 NULL,累计 206 万行(占 RIIN 商品 33.75%), 终态订单不再被 refresh_pending 刷新而长期滞留。该缺口继承自 inkreach-bi-auth-mng TS 版,且 TS 的 warnUnknown 告警在移植时被省略, 导致从未暴露。 修复: - RIIN_GOODS_STATUS_MAP 补 COMPLETE=已完成、PAUSE=暂停 - 对齐 TS 版 warnUnknown:四个状态映射函数对未知值告警(同值只报一次) - docs/sql:Goods.status 约束新增允许"暂停"(原约束拒绝,已在库执行) - scripts/backfill_riin_goods_status.py:存量 NULL 按"已发货"订单终态 近似回填(幂等、分批);"已关闭"与非终态订单走 by-orders 重拉真值 - tests/test_mappers.py:9 个单测全过;实单 GSU1S555400MFLL e2e 验证 4 行 NULL → 已完成/暂停 --- .../2026-09-28-goods-status-allow-pause.sql | 13 ++ flows/shared/mappers.py | 57 ++++++-- scripts/backfill_riin_goods_status.py | 72 ++++++++++ tests/test_mappers.py | 125 ++++++++++++++++++ 4 files changed, 259 insertions(+), 8 deletions(-) create mode 100644 docs/sql/2026-09-28-goods-status-allow-pause.sql create mode 100644 scripts/backfill_riin_goods_status.py create mode 100644 tests/test_mappers.py diff --git a/docs/sql/2026-09-28-goods-status-allow-pause.sql b/docs/sql/2026-09-28-goods-status-allow-pause.sql new file mode 100644 index 0000000..ffc1834 --- /dev/null +++ b/docs/sql/2026-09-28-goods-status-allow-pause.sql @@ -0,0 +1,13 @@ +-- 2026-09-28 Goods.status 约束新增允许 '暂停' +-- 背景:RIIN 商品级 goodsStatus 复用平台状态枚举,存在 PAUSE(暂停), +-- 原约束未包含该值,导致映射层只能落 NULL(chk_goods_status 拒绝其余新值)。 +-- 配套:mappers.py 的 RIIN_GOODS_STATUS_MAP 补 COMPLETE/PAUSE。 +-- 执行方式(两步,避免阻塞 ETL 写入;VALIDATE 只锁 SHARE UPDATE EXCLUSIVE): +ALTER TABLE "Goods" DROP CONSTRAINT chk_goods_status; +ALTER TABLE "Goods" ADD CONSTRAINT chk_goods_status CHECK ( + status IS NULL OR status = ANY (ARRAY[ + '备货中', '已删除', '已取消', '已完成', '待付款', '待确认', + '搁置中', '待发货', '已发货', '退款中', '暂停' + ]) +) NOT VALID; +ALTER TABLE "Goods" VALIDATE CONSTRAINT chk_goods_status; diff --git a/flows/shared/mappers.py b/flows/shared/mappers.py index ec700d4..ac14f61 100644 --- a/flows/shared/mappers.py +++ b/flows/shared/mappers.py @@ -33,6 +33,8 @@ RIIN_EXPRESS_SOURCE_MAP: dict[int, str] = { RIIN_GOODS_STATUS_MAP: dict[str, str] = { "NOT_SHIPPED": "待发货", "SHIPPED": "已发货", "CANCEL": "已取消", "CANCELLED": "已取消", "REFUNDING": "退款中", + # 商品级状态复用平台状态枚举,COMPLETE/PAUSE 缺失曾导致 1/3 商品行 status 为 NULL + "COMPLETE": "已完成", "PAUSE": "暂停", } RIIN_PLATFORM_STATUS_MAP: dict[str, str] = { @@ -55,6 +57,17 @@ SDS_CHANNEL_TYPE_LABEL_MAP: dict[str, str] = { # ---------- 日期解析 ---------- +# 未知枚举值告警:同值只报一次,避免批量同步刷屏(对齐 TS 版 warnUnknown) +_warned_values: set[str] = set() + + +def _warn_unknown(map_name: str, value: object) -> None: + key = f"{map_name}:{value!r}" + if key not in _warned_values: + _warned_values.add(key) + print(f"[mapper] unknown {map_name} value: {value!r}") + + def safe_date(value: Any, context: str = "") -> datetime: """解析日期字符串(无时区视为 +08:00),返回 tz-aware datetime。""" if not value or not isinstance(value, str): @@ -188,15 +201,20 @@ class GoodsUpsertInput: # ---------- RIIN 映射 ---------- -def map_riin_system_status(order_status: int) -> str | None: - return RIIN_SYSTEM_STATUS_MAP.get(order_status) +def map_riin_system_status(order_status: int | None) -> str | None: + if order_status is None: + return None + v = RIIN_SYSTEM_STATUS_MAP.get(order_status) + if v is None: + _warn_unknown("RIIN systemStatus", order_status) + return v def map_riin_order(raw: Any) -> OrderUpsertInput: platform_oid = raw.platform_oid ordered_at = _parse(raw.push_time, f"RIIN order {platform_oid} pushTime") confirmed_at = _parse(raw.distributor_push_time, f"RIIN order {platform_oid} distributorPushTime") - system_status = map_riin_system_status(raw.order_status) if raw.order_status is not None else None + system_status = map_riin_system_status(raw.order_status) is_terminal = system_status in ("已发货", "已关闭", "已取消") completed_at: datetime | None = None if is_terminal: @@ -236,6 +254,15 @@ def map_riin_order(raw: Any) -> OrderUpsertInput: ) +def map_riin_goods_status(goods_status: str | None) -> str | None: + if not goods_status: + return None + v = RIIN_GOODS_STATUS_MAP.get(goods_status) + if v is None: + _warn_unknown("RIIN goodsStatus", goods_status) + return v + + def map_riin_goods(raw: Any, parent_order_no: str, parent_created_at: datetime) -> GoodsUpsertInput: product_code = raw.product_code item_name = product_code.split("-")[0] if product_code else None @@ -250,7 +277,7 @@ def map_riin_goods(raw: Any, parent_order_no: str, parent_created_at: datetime) itemSize=raw.size_code, itemStyleNo=style_no, quantity=raw.num if raw.num is not None else 1, - status=RIIN_GOODS_STATUS_MAP.get(raw.goods_status) if raw.goods_status else None, + status=map_riin_goods_status(raw.goods_status), ) @@ -262,13 +289,27 @@ def map_sds_channel_type(channel_type: Any) -> str | None: return SDS_CHANNEL_TYPE_LABEL_MAP.get(channel_type) -def map_sds_system_status(status: int) -> str | None: - return SDS_SYSTEM_STATUS_MAP.get(status) +def map_sds_system_status(status: int | None) -> str | None: + if status is None: + return None + v = SDS_SYSTEM_STATUS_MAP.get(status) + if v is None: + _warn_unknown("SDS systemStatus", status) + return v + + +def map_sds_goods_status(status: int | None) -> str | None: + if status is None: + return None + v = SDS_GOODS_STATUS_MAP.get(status) + if v is None: + _warn_unknown("SDS goodsStatus", status) + return v def map_sds_order(raw: Any, channel_type_map: dict[int, str | None], pre_courier: str | None = None) -> OrderUpsertInput: ordered_at = _parse(raw.gmt_pay, f"SDS order {raw.no} gmtPay") - system_status = map_sds_system_status(raw.status) if raw.status is not None else None + system_status = map_sds_system_status(raw.status) is_terminal = system_status in ("已完成", "已取消") completed_at: datetime | None = None if is_terminal and raw.finish_time and raw.finish_time > 0: @@ -369,7 +410,7 @@ def map_sds_goods(raw: Any, parent_order_no: str, parent_created_at: datetime) - itemSize=raw.product_size, itemStyleNo=_parse_sds_style_no(product_name), quantity=raw.num if raw.num is not None else 1, - status=SDS_GOODS_STATUS_MAP.get(raw.status) if raw.status is not None else None, + status=map_sds_goods_status(raw.status), ) diff --git a/scripts/backfill_riin_goods_status.py b/scripts/backfill_riin_goods_status.py new file mode 100644 index 0000000..db44a6e --- /dev/null +++ b/scripts/backfill_riin_goods_status.py @@ -0,0 +1,72 @@ +"""一次性回填:RIIN Goods.status 为 NULL 的历史行(终态"已发货"订单)。 + +背景:RIIN 商品级 goods_status 复用平台状态枚举,COMPLETE/PAUSE 此前无映射, +约 1/3 商品行 status 被写为 NULL(mappers.py 已补映射)。本脚本按订单终态 +将近似值回填;"已关闭"与非终态订单不在此列,走 riin-by-orders 重拉取真值。 +幂等:只更新 status IS NULL 的行。分批提交,避免长事务。 + +运行:uv run python scripts/backfill_riin_goods_status.py [--dry-run] [--batch N] +""" + +from __future__ import annotations + +import argparse +import os + +import psycopg2 +from dotenv import load_dotenv + +load_dotenv(os.path.join(os.path.dirname(__file__), "..", ".env")) + +DEFAULT_BATCH = 50_000 + +SQL_COUNT = """ +SELECT count(*) FROM "Goods" g +JOIN "Orders" o ON o."orderNo" = g."orderNo" AND o."createdAt" = g."createdAt" +WHERE o."systemName" = 'RIIN' AND o."systemStatus" = '已发货' AND g."status" IS NULL +""" + +# ctid 分批:PostgreSQL 的 UPDATE 不支持 LIMIT,用子查询圈定每批行 +SQL_BACKFILL_BATCH = """ +UPDATE "Goods" g SET "status" = '已发货' +WHERE g.ctid IN ( + SELECT g2.ctid FROM "Goods" g2 + JOIN "Orders" o ON o."orderNo" = g2."orderNo" AND o."createdAt" = g2."createdAt" + WHERE o."systemName" = 'RIIN' AND o."systemStatus" = '已发货' AND g2."status" IS NULL + LIMIT %s +) +""" + + +def main() -> None: + parser = argparse.ArgumentParser(description="回填 RIIN Goods.status NULL(已发货订单)") + parser.add_argument("--dry-run", action="store_true", help="只统计待回填行数,不写入") + parser.add_argument("--batch", type=int, default=DEFAULT_BATCH, help="每事务回填行数") + args = parser.parse_args() + + conn = psycopg2.connect(os.environ["DB_DSN"]) + try: + with conn.cursor() as cur: + cur.execute(SQL_COUNT) + (pending,) = cur.fetchone() + print(f"[backfill] 待回填(status IS NULL 且订单已发货): {pending} 行") + if args.dry_run or pending == 0: + return + + total = 0 + while True: + with conn.cursor() as cur: + cur.execute(SQL_BACKFILL_BATCH, (args.batch,)) + updated = cur.rowcount + conn.commit() + total += updated + print(f"[backfill] 本批 {updated} 行,累计 {total}") + if updated < args.batch: + break + print(f"[backfill] 完成,共回填 {total} 行") + finally: + conn.close() + + +if __name__ == "__main__": + main() diff --git a/tests/test_mappers.py b/tests/test_mappers.py new file mode 100644 index 0000000..e74d248 --- /dev/null +++ b/tests/test_mappers.py @@ -0,0 +1,125 @@ +"""mappers 状态映射单测。 + +运行:uv run python -m unittest discover -s tests -v +""" + +from __future__ import annotations + +import io +import unittest +from contextlib import redirect_stdout +from datetime import datetime, timedelta, timezone +from types import SimpleNamespace + +from flows.shared import mappers +from flows.shared.mappers import ( + map_riin_goods, + map_riin_goods_status, + map_riin_system_status, + map_sds_goods_status, + map_sds_system_status, +) + +TZ_CN = timezone(timedelta(hours=8)) + + +class WarnUnknownTest(unittest.TestCase): + def setUp(self): + mappers._warned_values.clear() + + def _capture(self): + return io.StringIO() + + def test_未知值告警且同值只报一次(self): + buf = self._capture() + with redirect_stdout(buf): + self.assertIsNone(map_riin_goods_status("SOMETHING_NEW")) + self.assertIsNone(map_riin_goods_status("SOMETHING_NEW")) # 第二次不再报 + lines = [ln for ln in buf.getvalue().splitlines() if "SOMETHING_NEW" in ln] + self.assertEqual(len(lines), 1) + self.assertIn("[mapper] unknown RIIN goodsStatus", lines[0]) + + +class RiinGoodsStatusTest(unittest.TestCase): + def setUp(self): + mappers._warned_values.clear() + + def test_全部已知枚举可映射(self): + cases = { + "NOT_SHIPPED": "待发货", + "SHIPPED": "已发货", + "CANCEL": "已取消", + "CANCELLED": "已取消", + "REFUNDING": "退款中", + "COMPLETE": "已完成", # 本次补齐(曾导致 NULL) + "PAUSE": "暂停", # 本次补齐(曾导致 NULL) + } + for raw, want in cases.items(): + with self.subTest(raw=raw): + self.assertEqual(map_riin_goods_status(raw), want) + + def test_空值返回None(self): + self.assertIsNone(map_riin_goods_status(None)) + self.assertIsNone(map_riin_goods_status("")) + + def test_map_riin_goods_接线_PAUSE与COMPLETE不再落NULL(self): + # 复现线上订单 GSU1S555400MFLL:4 行商品 goods_status=PAUSE/COMPLETE + created = datetime(2026, 9, 14, 9, 39, 24, tzinfo=TZ_CN) + base = dict( + platform_oid="GSU1S555400MFLL", product_code="DG001-BK", + color_code="BK", size_code="L", style_code=None, num=1, + ) + g1 = map_riin_goods(SimpleNamespace(id="P1", goods_status="PAUSE", **base), + "GSU1S555400MFLL", created) + g2 = map_riin_goods(SimpleNamespace(id="P2", goods_status="COMPLETE", **base), + "GSU1S555400MFLL", created) + self.assertEqual(g1.status, "暂停") + self.assertEqual(g2.status, "已完成") + self.assertEqual(g1.productionNo, "P1") + self.assertEqual(g1.quantity, 1) + self.assertEqual(g1.itemName, "DG001") + + +class RiinSystemStatusTest(unittest.TestCase): + def setUp(self): + mappers._warned_values.clear() + + def test_已知与空值(self): + self.assertEqual(map_riin_system_status(12), "已发货") + self.assertEqual(map_riin_system_status(-4), "待推送") + self.assertIsNone(map_riin_system_status(None)) + + def test_未知值返回None并告警(self): + buf = io.StringIO() + with redirect_stdout(buf): + self.assertIsNone(map_riin_system_status(999)) + self.assertIn("unknown RIIN systemStatus", buf.getvalue()) + + +class SdsStatusTest(unittest.TestCase): + def setUp(self): + mappers._warned_values.clear() + + def test_商品状态含已删除(self): + self.assertEqual(map_sds_goods_status(99), "已删除") + self.assertEqual(map_sds_goods_status(98), "已取消") + self.assertEqual(map_sds_goods_status(4), "已完成") + self.assertIsNone(map_sds_goods_status(None)) + + def test_未知商品状态返回None并告警(self): + buf = io.StringIO() + with redirect_stdout(buf): + self.assertIsNone(map_sds_goods_status(6)) + self.assertIn("unknown SDS goodsStatus", buf.getvalue()) + + def test_订单状态与未知告警(self): + self.assertEqual(map_sds_system_status(3), "备货中") + self.assertIsNone(map_sds_system_status(None)) + buf = io.StringIO() + with redirect_stdout(buf): + self.assertIsNone(map_sds_system_status(42)) + self.assertIn("unknown SDS systemStatus", buf.getvalue()) + + +if __name__ == "__main__": + unittest.main() -- 2.54.0