Files
inkreach-prefect/flows/shared/mappers.py
T

318 lines
12 KiB
Python

"""数据映射:RIIN / SDS 原始订单 → 统一 Order/Goods 写入格式。
忠实移植自 inkreach-bi-auth-mng 的 sync/mappers/{riin,sds,normalize-country}.ts。
"""
from __future__ import annotations
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": "退款中",
}
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": "自提",
}
# ---------- 日期解析 ----------
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": "日本"}
_EXCLUDED_COUNTRIES = {
"Croatia", "Czech Republic", "丹麦", "保加利亚", "克罗地亚", "卢森堡", "奥地利", "希腊",
"拉脱维亚", "捷克", "斯洛伐克", "斯洛文尼亚", "比利时", "法国", "爱尔兰", "爱沙尼亚",
"瑞典", "瑞士", "立陶宛", "罗马尼亚", "芬兰", "荷兰", "Denmark", "Austria", "Greece",
"Latvia", "Slovakia", "Slovenia", "Belgium", "France", "Ireland", "Estonia", "Sweden",
"Switzerland", "Lithuania", "Romania", "Finland", "Netherlands", "Luxembourg", "Bulgaria",
}
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 trimmed in _EXCLUDED_COUNTRIES:
return None
upper = trimmed.upper()
if upper in _US_ALIASES:
return "美国"
if upper in _MX_ALIASES:
return "墨西哥"
if upper in _CA_ALIASES:
return "加拿大"
if trimmed in _PL_ALIASES:
return "波兰"
if trimmed in _WAREHOUSE_MAP:
return _WAREHOUSE_MAP[trimmed]
if raw in _DIRECT_MAP:
return _DIRECT_MAP[raw]
return raw
# ---------- 统一写入模型 ----------
class OrderUpsertInput:
__slots__ = (
"orderNo", "outOrderNo", "courierSource", "courierName", "courierNo",
"merchantId", "countryRegion", "factoryName", "systemName", "systemStatus",
"platformName", "platformStatus", "createdAt", "orderedAt", "confirmedAt",
"completedAt", "shipDeadline",
)
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) -> 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) if raw.order_status is not None else None
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,
)
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
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=item_name,
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,
)
# ---------- 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) -> 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) 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:
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
if raw.parcel_list:
first = raw.parcel_list[0]
try:
courier_no = first["logistics"]["carriage"]["no"]
except (TypeError, KeyError, IndexError):
courier_no = None
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
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,
)
def _parse_sds_style_no(name: str | None) -> str | None:
if not name:
return None
parts = name.split("()")
if len(parts) < 3:
return None
subs = parts[2].split("-")
if len(subs) == 2:
return subs[0] or None
if len(subs) >= 3:
return subs[1] or None
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=SDS_GOODS_STATUS_MAP.get(raw.status) if raw.status is not None else None,
)
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