380 lines
15 KiB
Python
380 lines
15 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": "日本"}
|
|
|
|
_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) -> 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,
|
|
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(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
|
|
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,
|
|
)
|
|
|
|
|
|
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
|