yeuimu
c030eca178
CI / test (pull_request) Successful in 1m35s
问题: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 → 已完成/暂停
426 lines
16 KiB
Python
426 lines
16 KiB
Python
"""数据映射:RIIN / SDS 原始订单 → 统一 Order/Goods 写入格式。
|
|
|
|
忠实移植自 inkreach-bi-auth-mng 的 sync/mappers/{riin,sds,normalize-country}.ts。
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import re
|
|
from datetime import datetime, timedelta, timezone
|
|
from typing import Any
|
|
|
|
TZ_CN = timezone(timedelta(hours=8))
|
|
|
|
# ---------- RIIN 状态 / 平台 / 快递来源映射 ----------
|
|
|
|
RIIN_SYSTEM_STATUS_MAP: dict[int, str] = {
|
|
-4: "待推送", -3: "工厂反审", -5: "待审核", -6: "分配失败",
|
|
1: "待推送", 3: "已反审", 4: "生产中", 5: "生产中",
|
|
12: "已发货", 13: "已关闭", 14: "已取消", 15: "已关闭",
|
|
}
|
|
|
|
RIIN_PLATFORM_TYPE_MAP: dict[int, str] = {
|
|
2: "淘宝", 3: "拼多多", 6: "Shopee", 7: "LAZADA", 8: "1688",
|
|
11: "TIKTOK", 13: "TEMU", 15: "自建站(海外)", 16: "SHOPIFY",
|
|
17: "SHEIN", 18: "自建站(国内)", 19: "WALMART", 21: "AMAZON",
|
|
26: "OZON", 27: "Wildberries", 30: "ETSY", 31: "MERCADO",
|
|
}
|
|
|
|
RIIN_EXPRESS_SOURCE_MAP: dict[int, str] = {
|
|
1: "工厂发货", 2: "平台面单", 3: "自有面单", 4: "分销面单", 5: "电商面单",
|
|
}
|
|
|
|
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] = {
|
|
"SHIPPED": "已发货", "NOT_SHIPPED": "待发货", "CANCEL": "已取消",
|
|
"PAUSE": "暂停", "COMPLETE": "已完成",
|
|
}
|
|
|
|
# ---------- SDS 状态 / 物流渠道映射 ----------
|
|
|
|
SDS_SYSTEM_STATUS_MAP: dict[int, str] = {
|
|
1: "待付款", 2: "待确认", 3: "备货中", 4: "已完成", 5: "搁置中",
|
|
7: "待设计", 9: "部分发货", 98: "已取消",
|
|
}
|
|
SDS_GOODS_STATUS_MAP: dict[int, str] = {**SDS_SYSTEM_STATUS_MAP, 99: "已删除"}
|
|
|
|
SDS_CHANNEL_TYPE_LABEL_MAP: dict[str, str] = {
|
|
"online": "线上物流", "offline": "普通物流", "manual": "手动物流", "self_pickup": "自提",
|
|
}
|
|
|
|
|
|
# ---------- 日期解析 ----------
|
|
|
|
# 未知枚举值告警:同值只报一次,避免批量同步刷屏(对齐 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):
|
|
raise ValueError(f"Invalid date for {context}: {value!r}")
|
|
s = value.strip()
|
|
if "+" not in s and "T" not in s and not s.endswith("Z"):
|
|
s = f"{s}T00:00:00+08:00" if len(s) == 10 else f"{s}+08:00"
|
|
try:
|
|
dt = datetime.fromisoformat(s)
|
|
except ValueError:
|
|
raise ValueError(f"Invalid date for {context}: {value!r}")
|
|
return dt if dt.tzinfo else dt.replace(tzinfo=TZ_CN)
|
|
|
|
|
|
def _from_epoch_ms(value: Any) -> datetime | None:
|
|
try:
|
|
n = int(value)
|
|
except (TypeError, ValueError):
|
|
return None
|
|
return datetime.fromtimestamp(n / 1000, tz=TZ_CN) if n > 0 else None
|
|
|
|
|
|
# ---------- 国家/地区归一 ----------
|
|
|
|
_US_ALIASES = {"US", "UNITED STATES", "USA", "ESTADOS UNIDOS", "美国加州发货仓"}
|
|
_MX_ALIASES = {"MX", "MEXICO", "MÉXICO", "MEX", "ME"}
|
|
_CA_ALIASES = {"加拿大工厂", "CA"}
|
|
_PL_ALIASES = {"欧州波兰发货仓", "欧洲波兰法兰克福发货仓"}
|
|
_WAREHOUSE_MAP = {
|
|
"印美达澳洲发货仓": "澳大利亚", "欧洲德国发货仓": "德国", "欧洲匈牙利发货仓": "匈牙利",
|
|
"欧洲西班牙发货仓": "西班牙", "欧洲意大利发货仓": "意大利",
|
|
"印美达美西发货仓": "美国", "印美达美西二发货仓": "美国",
|
|
"印美达美东新泽西发货仓": "美国", "印美达美西毛毯发货仓": "美国",
|
|
"印美达亚特兰大发货仓": "美国",
|
|
}
|
|
_DIRECT_MAP = {"Saudi Arabia": "沙特", "Japan": "日本"}
|
|
|
|
_ISO2_CN: dict[str, str] = {
|
|
"US": "美国", "GB": "英国", "UK": "英国", "JP": "日本", "MX": "墨西哥",
|
|
"DE": "德国", "SA": "沙特", "ES": "西班牙", "BR": "巴西", "IT": "意大利",
|
|
"KR": "韩国", "CA": "加拿大", "FR": "法国", "PL": "波兰", "NL": "荷兰",
|
|
"AU": "澳大利亚", "AT": "奥地利", "PT": "葡萄牙", "BE": "比利时", "SE": "瑞典",
|
|
"CZ": "捷克", "RO": "罗马尼亚", "HU": "匈牙利", "SK": "斯洛伐克", "DK": "丹麦",
|
|
"LV": "拉脱维亚", "LT": "立陶宛", "IE": "爱尔兰", "CN": "中国", "HR": "克罗地亚",
|
|
"SI": "斯洛文尼亚", "EE": "爱沙尼亚", "GR": "希腊", "FI": "芬兰", "LU": "卢森堡",
|
|
"BG": "保加利亚", "CY": "塞浦路斯", "MT": "马耳他", "MEX": "墨西哥",
|
|
"NO": "挪威", "IS": "冰岛", "TR": "土耳其", "AE": "阿联酋", "IL": "以色列",
|
|
"NZ": "新西兰", "SG": "新加坡", "MY": "马来西亚", "TH": "泰国", "PH": "菲律宾",
|
|
"ID": "印度尼西亚", "VN": "越南", "IN": "印度", "RU": "俄罗斯", "UA": "乌克兰",
|
|
"RS": "塞尔维亚",
|
|
}
|
|
|
|
_EN_CN: dict[str, str] = {
|
|
"UNITED STATES": "美国", "UNITED KINGDOM": "英国", "MEXICO": "墨西哥",
|
|
"SAUDI ARABIA": "沙特", "JAPAN": "日本", "GERMANY": "德国",
|
|
"SPAIN": "西班牙", "BRAZIL": "巴西", "ITALY": "意大利", "CANADA": "加拿大",
|
|
"FRANCE": "法国", "POLAND": "波兰", "NETHERLANDS": "荷兰", "AUSTRALIA": "澳大利亚",
|
|
"CROATIA": "克罗地亚", "CZECH REPUBLIC": "捷克", "ESTADOS UNIDOS": "美国",
|
|
"SLOVAKIA (SLOVAK REPUBLIC)": "斯洛伐克", "ROMANIA": "罗马尼亚",
|
|
"HUNGARY": "匈牙利", "BELGIUM": "比利时", "SWEDEN": "瑞典", "DENMARK": "丹麦",
|
|
"FINLAND": "芬兰", "IRELAND": "爱尔兰", "PORTUGAL": "葡萄牙", "AUSTRIA": "奥地利",
|
|
"GREECE": "希腊", "LUXEMBOURG": "卢森堡", "BULGARIA": "保加利亚",
|
|
"ESTONIA": "爱沙尼亚", "LATVIA": "拉脱维亚", "LITHUANIA": "立陶宛",
|
|
"SLOVENIA": "斯洛文尼亚",
|
|
}
|
|
|
|
_CN_NORMALIZE: dict[str, str] = {
|
|
"澳洲": "澳大利亚",
|
|
}
|
|
|
|
|
|
def normalize_country_region(raw: str | None, factory_name: str | None = None) -> str | None:
|
|
"""将国家/地区原始值统一归一为中文国名。"""
|
|
if factory_name == "广州印美达":
|
|
return "中国"
|
|
if not raw:
|
|
return None
|
|
trimmed = raw.strip()
|
|
if not trimmed:
|
|
return None
|
|
if trimmed in _WAREHOUSE_MAP:
|
|
return _WAREHOUSE_MAP[trimmed]
|
|
if trimmed in _DIRECT_MAP:
|
|
return _DIRECT_MAP[trimmed]
|
|
if trimmed in _PL_ALIASES:
|
|
return "波兰"
|
|
if trimmed in _CN_NORMALIZE:
|
|
return _CN_NORMALIZE[trimmed]
|
|
upper = trimmed.upper()
|
|
if upper in _US_ALIASES:
|
|
return "美国"
|
|
if upper in _MX_ALIASES:
|
|
return "墨西哥"
|
|
if upper in _CA_ALIASES:
|
|
return "加拿大"
|
|
if upper in _ISO2_CN:
|
|
return _ISO2_CN[upper]
|
|
if upper in _EN_CN:
|
|
return _EN_CN[upper]
|
|
return trimmed
|
|
|
|
|
|
# ---------- 统一写入模型 ----------
|
|
|
|
class OrderUpsertInput:
|
|
__slots__ = (
|
|
"orderNo", "outOrderNo", "courierSource", "courierName", "courierNo",
|
|
"merchantId", "countryRegion", "factoryName", "systemName", "systemStatus",
|
|
"platformName", "platformStatus", "createdAt", "orderedAt", "confirmedAt",
|
|
"completedAt", "shipDeadline",
|
|
"consigneeName", "destinationCountry", "stateProvince", "city",
|
|
"addressLine1", "addressLine2", "addressLine3", "postalCode", "storeName",
|
|
"labelPdfUrl",
|
|
)
|
|
|
|
def __init__(self, **kwargs: Any) -> None:
|
|
for k in self.__slots__:
|
|
setattr(self, k, kwargs.get(k))
|
|
|
|
|
|
class GoodsUpsertInput:
|
|
__slots__ = (
|
|
"orderNo", "productionNo", "createdAt", "outOrderNo", "itemName",
|
|
"itemColor", "itemSize", "itemStyleNo", "quantity", "status",
|
|
)
|
|
|
|
def __init__(self, **kwargs: Any) -> None:
|
|
for k in self.__slots__:
|
|
setattr(self, k, kwargs.get(k))
|
|
|
|
|
|
# ---------- 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_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)
|
|
is_terminal = system_status in ("已发货", "已关闭", "已取消")
|
|
completed_at: datetime | None = None
|
|
if is_terminal:
|
|
completed_at = _parse(raw.shipping_time, "") or _parse(raw.finish_time, "")
|
|
ship_deadline = _parse(raw.except_last_shipping_time, "")
|
|
courier_source = RIIN_EXPRESS_SOURCE_MAP.get(raw.express_source) if raw.express_source is not None else None
|
|
factory_name = raw.distributor_push_target_name
|
|
|
|
return OrderUpsertInput(
|
|
orderNo=platform_oid,
|
|
outOrderNo=raw.source_platform_oid,
|
|
courierSource=courier_source,
|
|
courierName=raw.delivery_courier,
|
|
courierNo=raw.courier_number,
|
|
merchantId=raw.sub_name,
|
|
countryRegion=normalize_country_region(raw.receiver_country, factory_name),
|
|
factoryName=factory_name,
|
|
systemName="RIIN",
|
|
systemStatus=system_status,
|
|
platformName=RIIN_PLATFORM_TYPE_MAP.get(raw.platform_type, str(raw.platform_type)) if raw.platform_type is not None else None,
|
|
platformStatus=RIIN_PLATFORM_STATUS_MAP.get(raw.platform_order_status, raw.platform_order_status) if raw.platform_order_status else None,
|
|
createdAt=safe_date(raw.order_create_time, f"RIIN order {platform_oid} orderCreateTime"),
|
|
orderedAt=ordered_at,
|
|
confirmedAt=confirmed_at,
|
|
completedAt=completed_at,
|
|
shipDeadline=ship_deadline,
|
|
consigneeName=raw.consignee_name,
|
|
destinationCountry=raw.receiver_country,
|
|
stateProvince=raw.receiver_province,
|
|
city=raw.receiver_city,
|
|
addressLine1=raw.address,
|
|
addressLine2=raw.receiver_district,
|
|
addressLine3=raw.receiver_town,
|
|
postalCode=raw.post_code,
|
|
storeName=raw.platform_shop_name,
|
|
labelPdfUrl=raw.waybill,
|
|
)
|
|
|
|
|
|
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
|
|
style_no = raw.style_code or item_name
|
|
return GoodsUpsertInput(
|
|
orderNo=parent_order_no,
|
|
productionNo=raw.id,
|
|
createdAt=parent_created_at,
|
|
outOrderNo=raw.platform_oid,
|
|
itemName=item_name,
|
|
itemColor=raw.color_code,
|
|
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),
|
|
)
|
|
|
|
|
|
# ---------- SDS 映射 ----------
|
|
|
|
def map_sds_channel_type(channel_type: Any) -> str | None:
|
|
if not isinstance(channel_type, str):
|
|
return 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_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)
|
|
is_terminal = system_status in ("已完成", "已取消")
|
|
completed_at: datetime | None = None
|
|
if is_terminal and raw.finish_time and raw.finish_time > 0:
|
|
completed_at = _from_epoch_ms(raw.finish_time)
|
|
ship_deadline = _from_epoch_ms(raw.latest_ship_time) if raw.latest_ship_time else None
|
|
|
|
courier_source = pre_courier
|
|
if courier_source is None and raw.logistics_id is not None:
|
|
try:
|
|
lid = int(raw.logistics_id)
|
|
except (TypeError, ValueError):
|
|
lid = None
|
|
if lid is not None and lid in channel_type_map:
|
|
courier_source = channel_type_map[lid]
|
|
|
|
courier_no: str | None = None
|
|
label_pdf_url: str | None = None
|
|
if raw.parcel_list:
|
|
first = raw.parcel_list[0]
|
|
try:
|
|
carriage = first["logistics"]["carriage"]
|
|
courier_no = carriage.get("no")
|
|
label_pdf_url = carriage.get("laberPdf") or carriage.get("labelPdf")
|
|
except (TypeError, KeyError, IndexError):
|
|
pass
|
|
|
|
factory_name: str | None = None
|
|
for item in raw.items or []:
|
|
fac = getattr(item, "factory", None)
|
|
fname = fac.get("name") if isinstance(fac, dict) else getattr(fac, "name", None)
|
|
if fname:
|
|
factory_name = fname
|
|
break
|
|
|
|
confirmed_at = _parse(getattr(raw, "gmt_confirm_time", None), f"SDS order {raw.no} gmtConfirmTime")
|
|
area_name = raw.issuing_bay_area.name if raw.issuing_bay_area else None
|
|
addr = raw.address
|
|
return OrderUpsertInput(
|
|
orderNo=raw.no,
|
|
outOrderNo=raw.out_order_no,
|
|
courierSource=courier_source,
|
|
courierName=raw.carriage_name,
|
|
courierNo=courier_no,
|
|
merchantId=raw.merchant.name if raw.merchant else None,
|
|
countryRegion=normalize_country_region(area_name, factory_name),
|
|
factoryName=factory_name,
|
|
systemName="SDS",
|
|
systemStatus=system_status,
|
|
platformName=raw.merchant_store.platform_name if raw.merchant_store else None,
|
|
platformStatus=raw.online_order_status_name,
|
|
createdAt=safe_date(raw.gmt_created, f"SDS order {raw.no}"),
|
|
orderedAt=ordered_at,
|
|
confirmedAt=confirmed_at,
|
|
completedAt=completed_at,
|
|
shipDeadline=ship_deadline,
|
|
consigneeName=addr.consignee if addr else None,
|
|
destinationCountry=addr.country if addr else None,
|
|
stateProvince=addr.province if addr else None,
|
|
city=addr.city if addr else None,
|
|
addressLine1=addr.address_detail1 if addr else None,
|
|
addressLine2=addr.address_detail2 if addr else None,
|
|
addressLine3=addr.address_detail3 if addr else None,
|
|
postalCode=addr.postcode if addr else None,
|
|
storeName=raw.merchant_store.name if raw.merchant_store else None,
|
|
labelPdfUrl=label_pdf_url,
|
|
)
|
|
|
|
|
|
_STYLE_CODE_RE = re.compile(r"[A-Za-z]{1,6}\d{2,4}[A-Za-z]{0,2}|\d{2,6}[A-Za-z]{1,2}")
|
|
|
|
|
|
def _parse_sds_style_no(name: str | None) -> str | None:
|
|
"""从商品名提取款号,如 '英国（包邮）180g纯棉T恤-GBTM001-单面印花' → 'GBTM001';
|
|
裸码名(如 'DF009')直接返回本身。逐 '-' 段扫描,避免误中段外字符。"""
|
|
if not name:
|
|
return None
|
|
for seg in name.split("-"):
|
|
seg = seg.strip()
|
|
m = _STYLE_CODE_RE.fullmatch(seg)
|
|
if m:
|
|
return seg.upper()
|
|
return None
|
|
|
|
|
|
def map_sds_goods(raw: Any, parent_order_no: str, parent_created_at: datetime) -> GoodsUpsertInput:
|
|
product_name = raw.product_name
|
|
color_name = None
|
|
block = raw.product_color_block
|
|
if block is not None:
|
|
color_name = getattr(block, "color_name", None) or (block.get("colorName") if isinstance(block, dict) else None)
|
|
return GoodsUpsertInput(
|
|
orderNo=parent_order_no,
|
|
productionNo=raw.no or raw.id,
|
|
createdAt=parent_created_at,
|
|
outOrderNo=raw.out_order_item_id,
|
|
itemName=product_name,
|
|
itemColor=color_name,
|
|
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),
|
|
)
|
|
|
|
|
|
def _parse(value: Any, context: str) -> datetime | None:
|
|
if not value:
|
|
return None
|
|
if isinstance(value, (int, float)):
|
|
return _from_epoch_ms(value)
|
|
try:
|
|
return safe_date(value, context)
|
|
except ValueError:
|
|
return None
|