Merge pull request 'Develop' (#4) from develop into main

Reviewed-on: #4

.gitea/workflows/ci.yml

@@ -12,10 +12,7 @@ jobs:
       - name: 检出代码
         run: |
           REF="${{ github.event.pull_request.head.ref || github.ref_name }}"
-          git clone --depth 1 "${{ github.server_url }}/${{ github.repository }}.git" .
-          # PR head 可能是完整 ref(AGit 建的 PR 为 refs/pull/N/head),clone --branch 只认分支名,改 fetch 后 checkout
-          git fetch --depth 1 origin "$REF"
-          git checkout --detach FETCH_HEAD
+          git clone --depth 1 --branch "$REF" "${{ github.server_url }}/${{ github.repository }}.git" .
       - name: 安装 uv
         run: curl -LsSf https://astral.sh/uv/install.sh | sh
       - name: 安装依赖

api/ocr/models.py

@@ -37,16 +37,14 @@ class OcrResult(BaseModel):
 
 
 # OCR 服务已有分类器模型的国家(版面见 api.md §4)
-SUPPORTED_COUNTRIES = {"US", "UK", "DE", "MX", "SA", "CA", "ES", "IT"}
+SUPPORTED_COUNTRIES = {"US", "UK", "DE", "MX", "SA"}
 
 # ISO 国家代码 / 中文国名 → OCR 国家代码
 _COUNTRY_MAP: dict[str, str] = {
     # 代码
     "GB": "UK", "UK": "UK", "US": "US", "DE": "DE", "MX": "MX", "SA": "SA",
-    "CA": "CA", "ES": "ES", "IT": "IT",
     # 中文名
     "英国": "UK", "美国": "US", "德国": "DE", "墨西哥": "MX", "沙特": "SA",
-    "加拿大": "CA", "西班牙": "ES", "意大利": "IT",
 }
 
 

docs/sql/2026-09-28-goods-status-allow-pause.sql

@@ -1,13 +0,0 @@
--- 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;

flows/label_ocr.py

@@ -137,12 +137,12 @@ def label_ocr(
     courier_source: str | None = None,
     days: int = 7,
 ) -> dict:
-    """面单 OCR enrichment。默认处理全部已支持国家(见 api.ocr.models.SUPPORTED_COUNTRIES),近 7 天下单的订单。
+    """面单 OCR enrichment。默认处理英国+美国,近 7 天下单的订单。
 
-    手动跑指定国家: label_ocr(country_regions=['德国'])
+    手动跑其他国家: label_ocr(country_regions=['德国'])
     """
     if country_regions is None:
-        country_regions = ['英国', '美国', '墨西哥', '德国', '沙特', '加拿大', '西班牙', '意大利']
+        country_regions = ['英国', '美国', '墨西哥', '德国', '沙特']
     if _has_running_peer():
         print("已有 label-ocr 在运行,跳过本次")
         return {"skipped": True}

flows/platform_order/platform_order_upper_sync.py

@@ -1,65 +0,0 @@
-# -*- coding: utf-8 -*-
-"""平台订单 platform 字段大写同步:把 platform_order_data.platform 统一转为大写。
-
-优化自 platform_order_upper_sync.py,对应 platform-order-upper-sync。
-"""
-
-import time
-
-import psycopg2
-from prefect import flow
-
-# 数据库连接参数
-conn_params = {
-    "host": "192.168.124.92",
-    "port": 5432,
-    "user": "postgres",
-    "password": "Inkreach123456.",
-    "database": "smt",
-}
-
-
-def log_msg(msg, level="info"):
-    ts = time.strftime('%H:%M:%S')
-    tag = {"success": "[OK]", "error": "[ERR]"}.get(level, "")
-    print(f"[{ts}] {tag} {msg}")
-
-
-def update_platform_upper() -> int:
-    conn = None
-    cur = None
-    try:
-        conn = psycopg2.connect(**conn_params)
-        cur = conn.cursor()
-
-        # 仅更新需要变更的行,避免每次都全表写入
-        update_sql = """
-            UPDATE platform_order_data
-            SET platform = UPPER(platform)
-            WHERE platform IS NOT NULL
-              AND platform <> UPPER(platform);
-        """
-        cur.execute(update_sql)
-        affected_rows = cur.rowcount
-        conn.commit()
-        log_msg(f"platform 大写转换完成，共 {affected_rows} 行受影响", "success")
-        return affected_rows
-    except Exception as e:
-        log_msg(f"platform 大写转换失败：{e}", "error")
-        if conn:
-            conn.rollback()
-        raise
-    finally:
-        if cur:
-            cur.close()
-        if conn:
-            conn.close()
-
-
-@flow(name="platform-order-upper-sync", log_prints=True)
-def platform_order_upper_sync() -> int:
-    return update_platform_upper()
-
-
-if __name__ == "__main__":
-    platform_order_upper_sync.serve(name="platform-order-upper-sync", cron="*/2 * * * *")

flows/platform_order/sdspod.py

@@ -359,18 +359,10 @@ def sd_format_item(item):
     create_time = item.get("gmtCreated", "")
     parcel_list = item.get("parcelList", [])
     if parcel_list and isinstance(parcel_list, list):
-        def safe_int(val):
-            try:
-                return int(val)
-            except (ValueError, TypeError):
-                return 0
-
-        qty = sum(safe_int(p.get("totalQty", 0)) for p in parcel_list)
+        qty = sum(p.get("totalQty", 0) for p in parcel_list)
     else:
         qty = 0
 
-
-
     return [
         to_chinese_country(country),
         to_chinese_country(country),

flows/shared/mappers.py

@@ -33,8 +33,6 @@ 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] = {
@@ -57,17 +55,6 @@ 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):
@@ -201,20 +188,15 @@ class GoodsUpsertInput:
 
 # ---------- RIIN 映射 ----------
 
-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_system_status(order_status: int) -> str | None:
+    return RIIN_SYSTEM_STATUS_MAP.get(order_status)
 
 
 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)
+    system_status = map_riin_system_status(raw.order_status) if raw.order_status is not None else None
     is_terminal = system_status in ("已发货", "已关闭", "已取消")
     completed_at: datetime | None = None
     if is_terminal:
@@ -254,15 +236,6 @@ 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
@@ -277,7 +250,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=map_riin_goods_status(raw.goods_status),
+        status=RIIN_GOODS_STATUS_MAP.get(raw.goods_status) if raw.goods_status else None,
     )
 
 
@@ -289,27 +262,13 @@ 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 | 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_system_status(status: int) -> str | None:
+    return SDS_SYSTEM_STATUS_MAP.get(status)
 
 
 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)
+    system_status = map_sds_system_status(raw.status) if raw.status is not None else None
     is_terminal = system_status in ("已完成", "已取消")
     completed_at: datetime | None = None
     if is_terminal and raw.finish_time and raw.finish_time > 0:
@@ -410,7 +369,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=map_sds_goods_status(raw.status),
+        status=SDS_GOODS_STATUS_MAP.get(raw.status) if raw.status is not None else None,
     )
 
 

prefect.yaml

@@ -97,14 +97,7 @@ deployments:
 - name: platform-order-sdspod
   entrypoint: flows/platform_order/sdspod.py:sdspod_platform_sync
   schedules:
-  - cron: 0 * * * *
-    timezone: Asia/Shanghai
-  work_pool:
-    name: inkreach-pool
-- name: platform-order-upper-sync
-  entrypoint: flows/platform_order/platform_order_upper_sync.py:platform_order_upper_sync
-  schedules:
-  - cron: '*/2 * * * *'
+  - cron: '*/30 * * * *'
     timezone: Asia/Shanghai
   work_pool:
     name: inkreach-pool

scripts/backfill_riin_goods_status.py

@@ -1,72 +0,0 @@
-"""一次性回填: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()

tests/test_mappers.py

@@ -1,125 +0,0 @@
-"""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()