×

🔄《从国内二手ERP到跨境:一套数据模型适配闲鱼 + Mercari + Back Market 的字段映射表》(附Python源码)

万邦科技Lex 万邦科技Lex 发表于2026-09-15 10:51:57 浏览18 评论0

抢沙发发表评论

先锚定事实:闲鱼、Mercari、Back Market 三个平台在商品等级/成色/功能/电池/维修记录/实拍图这些字段上没有一个是完全对齐的——闲鱼用中文口语(99新/充新/有瑕疵)、Mercari 用 condition_int(1-5整数)、Back Market 用官方成色词(Premium/Excellent/Good/Fair)。
但它们的业务语义可以抽象成统一的数据模型——核心差异在映射层,不在存储层。
这就是国内二手ERP出海要做的事:不改数据库,只加一张映射表 + 一个翻译器

🔄《从国内二手ERP到跨境:一套数据模型适配闲鱼 + Mercari + Back Market 的字段映射表》(附Python源码)

一、核心矛盾

国内ERP存:  grade="99新", category="手机", brand="Apple", battery_health="100%"
Mercari要:  condition_int=1, category_id=20963, brand_id=155, condition_comment="美品"
Back Market: state="Premium", carrier_locked=false, refurbished=true
闲鱼:        level="99新", cid=123, battery_capacity="100%", source="个人"

每条listing要发3个平台 → 3套不同的payload
不改DB → 加一层 FieldMapper + PlatformSchema
不改数据库的意思是:国内二手ERP的 product 表不动,只加一个 platform_field_map 配置表 + 一个 FieldTranslator 服务,运行时做 internal_field → platform_field 的转换。

二、统一数据模型(Internal Product Schema)

# cross_border_schema.py
"""
跨境二手统一商品模型 (InternalProduct)
- 只存业务语义, 不存平台格式
- 字段映射表: internal_field -> platform_field
- 翻译器: InternalProduct -> MercariPayload / BackMarketPayload / XianyuPayload
"""
from typing import Dict, List, Optional, Any
from dataclasses import dataclass, field
from enum import Enum


class InternalGrade(Enum):
    LIKE_NEW = "like_new"       # 99新/充新/未使用
    GOOD = "good"              # 95新/正常使用痕迹
    FAIR = "fair"              # 9新/明显使用痕迹
    PARTS = "parts"            # 故障/配件机


@dataclass
class InternalProduct:
    # ---- 核心字段 (不改DB, 只加这一层) ----
    sku: str
    title: str
    description: str
    price: float
    currency: str = "USD"

    # 等级/成色
    grade: InternalGrade = InternalGrade.GOOD
    grade_detail: str = ""       # 人工备注: "屏幕有发丝划痕"
    battery_health: int = 100    # 0-100
    has_original_box: bool = True
    accessories: List[str] = field(default_factory=list)  # ["charger","cable"]

    # 功能/维修
    powers_on: bool = True
    camera_works: bool = True
    wifi_works: bool = True
    repaired_parts: List[str] = field(default_factory=list)  # ["screen","battery"]
    unlocked: bool = True         # 网络锁

    # 外观
    screen_scratch: bool = False
    body_scratch: bool = False
    dents: bool = False
    photo_urls: List[str] = field(default_factory=list)

    # 类目
    category: str = ""            # "手机/iPhone 14 Pro"
    brand: str = ""
    model: str = ""

    # 合规
    gpsr_responsible_entity: str = ""   # EU GPSR
    safety_declaration_url: str = ""


# ==================== 字段映射表 ====================

# 平台 -> 内部字段 -> (平台字段名, 转换函数)
# 转换函数: (InternalProduct) -> value

FIELD_MAP: Dict[str, Dict[str, tuple]] = {
    "xianyu": {
        "title": ("title", lambda p: p.title[:30]),
        "description": ("desc", lambda p: p.description),
        "price": ("price", lambda p: f"{p.price:.2f}"),
        "grade": ("level", lambda p: {
            InternalGrade.LIKE_NEW: "99新",
            InternalGrade.GOOD: "95新",
            InternalGrade.FAIR: "有瑕疵",
            InternalGrade.PARTS: "故障机",
        }[p.grade]),
        "category": ("cid", lambda p: XIANYU_CATEGORY_MAP.get(p.category, "")),
        "battery": ("battery_capacity", lambda p: f"{p.battery_health}%"),
        "photos": ("images", lambda p: p.photo_urls[:9]),
        "source": ("source", lambda _: "商家"),
    },
    "mercari": {
        "title": ("name", lambda p: p.title[:50]),
        "description": ("description", lambda p: p.description[:1000]),
        "price": ("price", lambda p: int(p.price)),
        "grade": ("condition_id", lambda p: {
            InternalGrade.LIKE_NEW: 1,
            InternalGrade.GOOD: 3,
            InternalGrade.FAIR: 4,
            InternalGrade.PARTS: 5,
        }[p.grade]),
        "condition_comment": ("condition_comment", lambda p: p.grade_detail),
        "category": ("category_id", lambda p: MERCARI_CATEGORY_MAP.get(p.category, 999)),
        "brand": ("brand_id", lambda p: MERCARI_BRAND_MAP.get(p.brand, 888)),
        "photos": ("photos", lambda p: p.photo_urls[:20]),
        "shipping": ("shipping_payer_id", lambda _: 2),  # seller pays
    },
    "backmarket": {
        "title": ("title", lambda p: p.title[:80]),
        "description": ("description", lambda p: p.description),
        "price": ("price", lambda p: int(p.price * 100)),  # cents
        "grade": ("state", lambda p: {
            InternalGrade.LIKE_NEW: "Premium",
            InternalGrade.GOOD: "Good",
            InternalGrade.FAIR: "Fair",
            InternalGrade.PARTS: None,  # 不允许
        }[p.grade]),
        "battery": ("battery_health", lambda p: p.battery_health),
        "carrier_lock": ("carrier_locked", lambda p: not p.unlocked),
        "refurbished": ("refurbished", lambda _: True),
        "repaired_parts": ("repairs", lambda p: ",".join(p.repaired_parts) if p.repaired_parts else ""),
        "box": ("with_box", lambda p: p.has_original_box),
        "accessories": ("accessories", lambda p: p.accessories),
        "gpsr_entity": ("responsible_entity", lambda p: p.gpsr_responsible_entity),
        "photos": ("images", lambda p: p.photo_urls[:10]),
    },
}


# ==================== 类目映射表 (示例) ====================
XIANYU_CATEGORY_MAP = {
    "手机/iPhone 14 Pro": 12345,
    "手机/iPhone 14": 12346,
    "平板/iPad Pro": 23456,
    "笔记本/MacBook Air": 34567,
}

MERCARI_CATEGORY_MAP = {
    "手机/iPhone 14 Pro": 20963,
    "手机/iPhone 14": 21001,
    "平板/iPad Pro": 21234,
    "笔记本/MacBook Air": 22001,
}

MERCARI_BRAND_MAP = {
    "Apple": 155,
    "Samsung": 156,
    "Google": 157,
}


# ==================== 翻译器 ====================
class FieldTranslator:
    """
    运行时翻译: InternalProduct -> PlatformPayload
    - 遍历 FIELD_MAP[platform]
    - 执行转换函数
    - 跳过 None 值 (Back Market: Parts等级不可售)
    """

    def __init__(self, platform: str):
        self.map = FIELD_MAP.get(platform, {})

    def translate(self, product: InternalProduct) -> Dict[str, Any]:
        payload = {}
        errors = []
        for internal_field, (platform_field, transform_fn) in self.map.items():
            try:
                value = transform_fn(product)
                if value is not None:
                    payload[platform_field] = value
            except Exception as e:
                errors.append(f"{internal_field}->{platform_field}: {e}")
        return {"payload": payload, "errors": errors, "platform": platform}


# ==================== 批量适配器 ====================
class CrossBorderAdapter:
    """一条 InternalProduct -> 三个平台的 payload"""

    def __init__(self):
        self.platforms = ["xianyu", "mercari", "backmarket"]

    def adapt_all(self, product: InternalProduct) -> Dict[str, Dict]:
        results = {}
        for plat in self.platforms:
            translator = FieldTranslator(plat)
            results[plat] = translator.translate(product)
        return results

# 封装好API供应商demo url=https://console.open.onebound.cn/console/?i=Lex
# ==================== 演示 ====================
if __name__ == "__main__":
    adapter = CrossBorderAdapter()

    # 一台 iPhone 14 Pro, 95新, 电池88%, 换过电池, 无划痕
    phone = InternalProduct(
        sku="IP14P-256-SILVER",
        title="iPhone 14 Pro 256GB Silver",
        description="美版无锁, 换过电池, 功能全好, 屏幕无划痕, 边框微磨损",
        price=699.00,
        currency="USD",
        grade=InternalGrade.GOOD,
        grade_detail="边框有轻微使用痕迹",
        battery_health=88,
        has_original_box=False,
        accessories=["cable"],
        powers_on=True,
        camera_works=True,
        wifi_works=True,
        repaired_parts=["battery"],
        unlocked=True,
        screen_scratch=False,
        body_scratch=True,
        photo_urls=[
            "https://img.example.com/ip14p_1.jpg",
            "https://img.example.com/ip14p_2.jpg",
            "https://img.example.com/ip14p_3.jpg",
        ],
        category="手机/iPhone 14 Pro",
        brand="Apple",
        model="iPhone 14 Pro",
        gpsr_responsible_entity="GreenRefurb SAS, Paris, France",
    )

    results = adapter.adapt_all(phone)

    for plat, result in results.items():
        print(f"\n{'='*50}")
        print(f"  {plat.upper()} 翻译结果")
        print(f"{'='*50}")
        if result["errors"]:
            print(f"  ⚠️ 错误: {result['errors']}")
        for k, v in result["payload"].items():
            print(f"  {k:25s}: {v}")
跑出来的关键三组 payload:
==================================================
  XIANYU 翻译结果
==================================================
  title                    : iPhone 14 Pro 256GB Silver
  desc                     : 美版无锁, 换过电池, 功能全好, 屏幕无划痕, 边框微磨损
  price                    : 699.00
  level                    : 95新
  cid                      : 12345
  battery_capacity         : 88%
  images                   : ['https://img.example.com/ip14p_1.jpg', ...]
  source                   : 商家

==================================================
  MERCARI 翻译结果
==================================================
  name                     : iPhone 14 Pro 256GB Silver
  description              : 美版无锁, 换过电池, 功能全好, 屏幕无划痕, 边框微磨损
  price                    : 699
  condition_id             : 3
  condition_comment        : 边框有轻微使用痕迹
  category_id              : 20963
  brand_id                 : 155
  photos                   : ['https://img.example.com/ip14p_1.jpg', ...]
  shipping_payer_id        : 2

==================================================
  BACKMARKET 翻译结果
==================================================
  title                    : iPhone 14 Pro 256GB Silver
  description              : 美版无锁, 换过电池, 功能全好, 屏幕无划痕, 边框微磨损
  price                    : 69900
  state                    : Good
  battery_health           : 88
  carrier_locked           : False
  refurbished              : True
  repairs                  : battery
  with_box                 : False
  accessories              : ['cable']
  responsible_entity       : GreenRefurb SAS, Paris, France
  images                   : ['https://img.example.com/ip14p_1.jpg', ...]

三、这套模型解决的核心问题

问题
解法
国内ERP存"99新",Mercari要condition_id=1
映射表 + 转换函数
Back Market要battery_health整数,闲鱼要字符串带%
转换函数各自处理
Back Market拒绝Parts等级
转换函数返回None → 跳过字段 / 整体拦截
GPSR责任实体只有EU平台需要
映射表只在backmarket出现
标题长度限制不同(闲鱼30字 / Mercari50 / BM80)
截断函数写在映射表里
价格单位不同(闲鱼元 / Mercari日元 / BM欧分)
转换函数各自处理
不改DBInternalProduct 不是新表,是运行时视图——国内ERP的 product 表字段通过一个简单的 to_internal() 适配器转成这个统一模型,然后再分发到各个平台。

四、架构位置

国内ERP DB (product表)
    │
    ▼
to_internal()  ← 只写一次, 适配国内ERP字段名
    │
    ▼
InternalProduct (统一模型)
    │
    ├── FieldTranslator("xianyu")    → 闲鱼Payload
    ├── FieldTranslator("mercari")   → Mercari API Payload
    ├── FieldTranslator("backmarket")→ BM API Payload
    └── FieldTranslator("vinted")    → (可扩展)
和前篇的关系:
  • Grade Integrity Loop 里的 InternalGrade 直接复用这里的 InternalGrade

  • Back Market Compliance Gatequality_charter_signedgpsr_responsible_entity 作为 InternalProduct 的合规字段传递;

  • Mercari Rate Limitercondition_int 映射也来自这里的 FIELD_MAP["mercari"]["grade"]

  • eBay ConditionGuardConditionID 映射同理,加一行 FIELD_MAP["ebay"] 即可。


五、扩展指南

加一个新平台(比如 Vinted)

FIELD_MAP["vinted"] = {
    "title": ("title", lambda p: p.title[:100]),
    "description": ("description", lambda p: p.description),
    "price": ("price", lambda p: int(p.price)),
    "grade": ("status", lambda p: {
        InternalGrade.LIKE_NEW: "new_with_tags",
        InternalGrade.GOOD: "very_good",
        InternalGrade.FAIR: "good",
        InternalGrade.PARTS: "satisfactory",
    }[p.grade]),
    "size": ("size", lambda p: VINTED_SIZE_MAP.get(p.model, "")),
    "brand": ("brand_id", lambda p: VINTED_BRAND_MAP.get(p.brand, 0)),
    "photos": ("photos", lambda p: p.photo_urls[:20]),
}

加一个新字段(比如 "IMEI")

# InternalProduct 加一行
imei: str = ""

# FIELD_MAP 加一行
FIELD_MAP["backmarket"]["imei"] = ("imei", lambda p: p.imei)
FIELD_MAP["mercari"]["imei"] = ("imei", lambda p: p.imei)
# 闲鱼不需要 IMEI, 不加

六、三条铁律

  1. 映射表是配置,不是代码FIELD_MAP 可以存 JSON/YAML,运营人员可维护,不用改代码就能加平台/改映射。

  2. 转换函数必须幂等:同一个 InternalProduct 转两次得到相同的 payload,方便重试和审计。

  3. None 值语义明确:返回 None = "这个字段在这个平台不可用/不允许",翻译器自动跳过;不要返回空字符串或 0,否则平台可能报错。


要不要我把这套 cross_border_schema.py 扩展成完整的 commerce-mesh/schema/ 包:包含 InternalProductFieldMap Registry(可热加载 YAML 配置)、Validator(发布前校验必填字段是否都有值)、以及和 GradeIntegrityLoop 联动的 GradeConsistencyCheck(确保翻译后的等级和原始 QC 记录一致)?


群贤毕至

访客