🏛️《从单体到中台：九家电商API统一适配器的DDD建模实战》（附Python源码）

一、DDD战略设计：限界上下文与统一语言

┌─────────────────────────────────────────────────────────┐
│                  电商中台（核心域）                       │
│                                                         │
│  ┌──────────────┐  ┌──────────────┐  ┌──────────────┐  │
│  │  订单上下文   │  │  商品上下文   │  │  库存上下文   │  │
│  │  OrderContext │  │ ProductCtx   │  │ StockCtx     │  │
│  └──────┬───────┘  └──────┬───────┘  └──────┬───────┘  │
│         │                  │                  │           │
│         ▼                  ▼                  ▼           │
│  ┌────────────────────────────────────────────────────┐  │
│  │          统一适配器（Anti-Corruption Layer）        │  │
│  │                                                    │  │
│  │  ┌──────────┐ ┌──────────┐ ┌──────────┐ ┌──────┐ │  │
│  │  │淘宝适配器│ │京东适配器│ │拼多多适配│ │...   │ │  │
│  │  │TaobaoAdp│ │JdAdapter│ │PddAdapte│ │      │ │  │
│  │  └──────────┘ └──────────┘ └──────────┘ └──────┘ │  │
│  └────────────────────────────────────────────────────┘  │
└─────────────────────────────────────────────────────────┘

统一领域语言（Ubiquitous Language）

二、战术设计：Hexagonal Architecture + 端口适配器

┌─────────────────────────────────────────────────────────┐
│                   领域层（Domain）                        │
│  Order / Product / Stock 实体 + 值对象 + 领域服务       │
└─────────────────────┬───────────────────────────────────┘
                      │ 端口接口（Port Interface）
┌─────────────────────┴───────────────────────────────────┐
│                  应用层（Application）                    │
│  OrderService / ProductService / StockService           │
│  编排领域对象 + 调端口接口                               │
└─────────────────────┬───────────────────────────────────┘
                      │ 适配器实现（Adapter Implementation）
┌─────────────────────┴───────────────────────────────────┐
│                  基础设施层（Infrastructure）             │
│  ┌──────────┐ ┌──────────┐ ┌──────────┐ ┌──────────┐  │
│  │TaobaoRepo│ │JdRepo    │ │PddRepo   │ │...       │  │
│  │(适配器)  │ │(适配器)  │ │(适配器)  │ │          │  │
│  └──────────┘ └──────────┘ └──────────┘ └──────────┘  │
└─────────────────────────────────────────────────────────┘

三、Python：DDD电商中台骨架（可直接运行）

# ecommerce_ddd_middleware.py
"""
九家电商API统一适配器 - DDD建模实战
- 四层架构：领域层 / 应用层 / 基础设施层 / 接口层
- 端口适配器模式：OrderRepository端口 + 9家适配器
- 统一领域语言：Order / Money / Recipient / OrderStatus
- 防腐层：每家平台的方言翻译成统一语言
"""
from abc import ABC, abstractmethod
from dataclasses import dataclass
from enum import Enum
from typing import List, Optional, Dict
from datetime import datetime

# ==================== 领域层 ====================

class OrderStatus(Enum):
    CREATED = "CREATED"           # 已创建
    PAID = "PAID"                 # 已付款
    SHIPPED = "SHIPPED"           # 已发货
    SIGNED = "SIGNED"             # 已签收
    REFUNDING = "REFUNDING"       # 退款中
    CLOSED = "CLOSED"             # 已关闭

@dataclass
class Money:
    amount: float
    currency: str = "CNY"
    
    def __add__(self, other: 'Money') -> 'Money':
        assert self.currency == other.currency
        return Money(self.amount + other.amount, self.currency)

@dataclass
class Recipient:
    name: str
    phone: str
    province: str = ""
    city: str = ""
    district: str = ""
    address: str = ""
    
    def mask_phone(self) -> str:
        """脱敏手机号"""
        if len(self.phone) == 11:
            return self.phone[:3] + "****" + self.phone[-4:]
        return self.phone

@dataclass
class OrderItem:
    sku_id: str
    title: str
    quantity: int
    price: Money

@dataclass
class Order:
    order_id: str
    channel: str          # 平台标识
    shop_id: str          # 店铺ID
    status: OrderStatus
    total: Money
    recipient: Recipient
    items: List[OrderItem]
    created_at: datetime
    updated_at: datetime
    raw_data: Dict = None  # 原始数据（调试用）
    
    def is_paid(self) -> bool:
        return self.status in (OrderStatus.PAID, OrderStatus.SHIPPED, OrderStatus.SIGNED)

# ==================== 端口接口 ====================

class OrderRepository(ABC):
    """订单仓储端口"""
    
    @abstractmethod
    def get_order(self, order_id: str) -> Optional[Order]:
        pass
    
    @abstractmethod
    def list_orders(self, shop_id: str, created_after: datetime,
                    status: Optional[OrderStatus] = None) -> List[Order]:
        pass
    
    @abstractmethod
    def save(self, order: Order) -> bool:
        pass

class ProductRepository(ABC):
    """商品仓储端口"""
    
    @abstractmethod
    def get_product(self, sku_id: str) -> Optional[Dict]:
        pass
    
    @abstractmethod
    def update_stock(self, sku_id: str, quantity: int) -> bool:
        pass

# ==================== 基础设施层：适配器 ====================

# -------- 淘宝适配器 --------
class TaobaoOrderAdapter(OrderRepository):
    """淘宝订单适配器（将淘宝方言翻译为统一语言）"""
    
    STATUS_MAP = {
        "WAIT_BUYER_PAY": OrderStatus.CREATED,
        "WAIT_SELLER_SEND_GOODS": OrderStatus.PAID,
        "WAIT_LOGISTICS_CONFIRM": OrderStatus.SHIPPED,
        "TRADE_FINISHED": OrderStatus.SIGNED,
        "TRADE_CLOSED": OrderStatus.CLOSED,
    }
    
    def get_order(self, order_id: str) -> Optional[Order]:
        # 模拟淘宝API调用
        raw = {
            "tid": order_id,
            "status": "WAIT_SELLER_SEND_GOODS",
            "payment": "99.00",
            "receiver_name": "张三",
            "receiver_mobile": "13800138000",
            "receiver_state": "广东省",
            "receiver_city": "深圳市",
            "receiver_district": "南山区",
            "receiver_address": "科技园路1号",
            "orders": [{"oid": "12345", "title": "商品A", "num": 2, "price": "49.50"}],
            "created": "2026-08-20 10:00:00",
        }
        return self._to_order(raw)
    
    def list_orders(self, shop_id: str, created_after: datetime,
                    status: Optional[OrderStatus] = None) -> List[Order]:
        # 模拟列表查询
        return [self.get_order("tb_001"), self.get_order("tb_002")]
    
    def save(self, order: Order) -> bool:
        print(f"💾 淘宝订单 {order.order_id} 落库")
        return True
    
    def _to_order(self, raw: Dict) -> Order:
        items = []
        for item in raw.get("orders", []):
            items.append(OrderItem(
                sku_id=str(item.get("oid", "")),
                title=item.get("title", ""),
                quantity=int(item.get("num", 0)),
                price=Money(float(item.get("price", 0)), "CNY"),
            ))
        
        return Order(
            order_id=str(raw.get("tid", "")),
            channel="taobao",
            shop_id="tb_shop_001",
            status=self.STATUS_MAP.get(raw.get("status", ""), OrderStatus.CREATED),
            total=Money(float(raw.get("payment", 0)), "CNY"),
            recipient=Recipient(
                name=raw.get("receiver_name", ""),
                phone=raw.get("receiver_mobile", ""),
                province=raw.get("receiver_state", ""),
                city=raw.get("receiver_city", ""),
                district=raw.get("receiver_district", ""),
                address=raw.get("receiver_address", ""),
            ),
            items=items,
            created_at=datetime.strptime(raw.get("created", ""), "%Y-%m-%d %H:%M:%S"),
            updated_at=datetime.now(),
            raw_data=raw,
        )

# -------- 拼多多适配器 --------
class PddOrderAdapter(OrderRepository):
    """拼多多订单适配器"""
    
    STATUS_MAP = {
        "0": OrderStatus.CREATED,
        "1": OrderStatus.PAID,
        "2": OrderStatus.SHIPPED,
        "3": OrderStatus.SIGNED,
        "4": OrderStatus.REFUNDING,
        "5": OrderStatus.CLOSED,
    }
    
    def get_order(self, order_id: str) -> Optional[Order]:
        raw = {
            "order_sn": order_id,
            "order_status": "1",
            "pay_amount": "9900",  # 分为单位
            "receiver_name": "李四",
            "receiver_phone": "13900139000",
            "province": "浙江省",
            "city": "杭州市",
            "town": "西湖区",
            "address": "文三路100号",
            "item_list": [{"sku_id": "sku_001", "goods_name": "商品B", "count": 1, "price": "9900"}],
            "created_time": int(datetime.now().timestamp()),
        }
        return self._to_order(raw)
    
    def list_orders(self, shop_id: str, created_after: datetime,
                    status: Optional[OrderStatus] = None) -> List[Order]:
        return [self.get_order("pdd_001")]
    
    def save(self, order: Order) -> bool:
        print(f"💾 拼多多订单 {order.order_id} 落库")
        return True
    
    def _to_order(self, raw: Dict) -> Order:
        items = []
        for item in raw.get("item_list", []):
            items.append(OrderItem(
                sku_id=item.get("sku_id", ""),
                title=item.get("goods_name", ""),
                quantity=int(item.get("count", 0)),
                price=Money(float(item.get("price", 0)) / 100, "CNY"),
            ))
        
        return Order(
            order_id=raw.get("order_sn", ""),
            channel="pdd",
            shop_id="pdd_shop_001",
            status=self.STATUS_MAP.get(raw.get("order_status", "0"), OrderStatus.CREATED),
            total=Money(float(raw.get("pay_amount", 0)) / 100, "CNY"),
            recipient=Recipient(
                name=raw.get("receiver_name", ""),
                phone=raw.get("receiver_phone", ""),
                province=raw.get("province", ""),
                city=raw.get("city", ""),
                district=raw.get("town", ""),
                address=raw.get("address", ""),
            ),
            items=items,
            created_at=datetime.fromtimestamp(int(raw.get("created_time", 0))),
            updated_at=datetime.now(),
            raw_data=raw,
        )

# -------- 亚马逊适配器 --------
class AmazonOrderAdapter(OrderRepository):
    """亚马逊SP-API订单适配器"""
    
    STATUS_MAP = {
        "PendingAvailability": OrderStatus.CREATED,
        "Pending": OrderStatus.CREATED,
        "Unshipped": OrderStatus.PAID,
        "PartiallyShipped": OrderStatus.SHIPPED,
        "Shipped": OrderStatus.SHIPPED,
        "InvoiceUnconfirmed": OrderStatus.SHIPPED,
        "Canceled": OrderStatus.CLOSED,
        "Unfulfillable": OrderStatus.CLOSED,
    }
    
    def get_order(self, order_id: str) -> Optional[Order]:
        raw = {
            "AmazonOrderId": order_id,
            "OrderStatus": "Unshipped",
            "OrderTotal": {"Amount": "79.99", "CurrencyCode": "USD"},
            "BuyerInfo": {"BuyerName": "John Doe"},
            "ShippingAddress": {
                "Name": "John Doe",
                "Phone": "+1-555-0123",
                "StateOrRegion": "California",
                "City": "San Francisco",
                "AddressLine1": "123 Market St",
            },
            "OrderItems": [{"SellerSKU": "SKU_US_001", "Title": "Product C",
                           "QuantityOrdered": 1, "ItemPrice": {"Amount": "79.99"}}],
            "PurchaseDate": "2026-08-20T10:00:00Z",
        }
        return self._to_order(raw)
    
    def list_orders(self, shop_id: str, created_after: datetime,
                    status: Optional[OrderStatus] = None) -> List[Order]:
        return [self.get_order("amz_001")]
    
    def save(self, order: Order) -> bool:
        print(f"💾 亚马逊订单 {order.order_id} 落库")
        return True
    
    def _to_order(self, raw: Dict) -> Order:
        addr = raw.get("ShippingAddress", {})
        items = []
        for item in raw.get("OrderItems", []):
            items.append(OrderItem(
                sku_id=item.get("SellerSKU", ""),
                title=item.get("Title", ""),
                quantity=int(item.get("QuantityOrdered", 0)),
                price=Money(float(item.get("ItemPrice", {}).get("Amount", 0)), "USD"),
            ))
        
        total = raw.get("OrderTotal", {})
        return Order(
            order_id=raw.get("AmazonOrderId", ""),
            channel="amazon",
            shop_id="amz_shop_001",
            status=self.STATUS_MAP.get(raw.get("OrderStatus", ""), OrderStatus.CREATED),
            total=Money(float(total.get("Amount", 0)), total.get("CurrencyCode", "USD")),
            recipient=Recipient(
                name=addr.get("Name", ""),
                phone=addr.get("Phone", ""),
                province=addr.get("StateOrRegion", ""),
                city=addr.get("City", ""),
                address=addr.get("AddressLine1", ""),
            ),
            items=items,
            created_at=datetime.fromisoformat(raw.get("PurchaseDate", "").replace("Z", "+00:00")),
            updated_at=datetime.now(),
            raw_data=raw,
        )

# ==================== 应用层 ====================

class OrderService:
    """订单应用服务"""
    
    def __init__(self):
        self._repos: Dict[str, OrderRepository] = {}
    
    def register_channel(self, channel: str, repo: OrderRepository):
        self._repos[channel] = repo
    
    def sync_orders(self, channels: List[str], since: datetime) -> List[Order]:
        """同步指定渠道的订单"""
        all_orders = []
        for ch in channels:
            if ch not in self._repos:
                print(f"⚠️ 未注册渠道: {ch}")
                continue
            repo = self._repos[ch]
            try:
                orders = repo.list_orders(ch, since)
                for order in orders:
                    repo.save(order)
                all_orders.extend(orders)
                print(f"✅ {ch}: 同步{len(orders)}单")
            except Exception as e:
                print(f"❌ {ch}: 同步失败 - {e}")
        return all_orders
    
    def get_order(self, channel: str, order_id: str) -> Optional[Order]:
        if channel not in self._repos:
            raise ValueError(f"未注册渠道: {channel}")
        return self._repos[channel].get_order(order_id)

# ==================== 接口层 ====================

class OrderController:
    """订单控制器（对外接口）"""
    
    def __init__(self, service: OrderService):
        self.service = service
    
    def sync_all(self) -> Dict:
        """同步所有渠道订单"""
        channels = list(self.service._repos.keys())
        since = datetime.now().replace(hour=0, minute=0, second=0, microsecond=0)
        orders = self.service.sync_orders(channels, since)
        return {
            "total": len(orders),
            "channels": channels,
            "orders": [{
                "id": o.order_id,
                "channel": o.channel,
                "status": o.status.value,
                "total": f"{o.total.amount} {o.total.currency}",
                "recipient": o.recipient.name,
                "items_count": len(o.items),
            } for o in orders[:5]],  # 只返回前5条
        }
    
    def get_order_detail(self, channel: str, order_id: str) -> Optional[Dict]:
        order = self.service.get_order(channel, order_id)
        if not order:
            return None
        return {
            "id": order.order_id,
            "channel": order.channel,
            "status": order.status.value,
            "total": f"{order.total.amount} {order.total.currency}",
            "recipient": {
                "name": order.recipient.name,
                "phone": order.recipient.mask_phone(),
                "address": f"{order.recipient.province}{order.recipient.city}{order.recipient.district}{order.recipient.address}",
            },
            "items": [{
                "sku": i.sku_id,
                "title": i.title,
                "qty": i.quantity,
                "price": f"{i.price.amount} {i.price.currency}",
            } for i in order.items],
            "created_at": order.created_at.isoformat(),
        }

# ==================== 演示 ====================
if __name__ == "__main__":
    # 1. 注册适配器
    service = OrderService()
    service.register_channel("taobao", TaobaoOrderAdapter())
    service.register_channel("pdd", PddOrderAdapter())
    service.register_channel("amazon", AmazonOrderAdapter())
    
    # 2. 控制器
    controller = OrderController(service)
    
    # 3. 同步所有渠道
    print("=== 同步所有渠道订单 ===")
    result = controller.sync_all()
    print(f"总订单数: {result['total']}")
    for o in result['orders']:
        print(f"  {o['channel']:8} {o['id']:20} {o['status']:10} {o['total']:10} {o['recipient']}")
    
    # 4. 查看订单详情
    print("\n=== 订单详情 ===")
    detail = controller.get_order_detail("amazon", "amz_001")
    if detail:
        print(f"渠道: {detail['channel']}")
        print(f"状态: {detail['status']}")
        print(f"金额: {detail['total']}")
        print(f"收件人: {detail['recipient']['name']} ({detail['recipient']['phone']})")
        print(f"地址: {detail['recipient']['address']}")
        for item in detail['items']:
            print(f"  商品: {item['title']} x{item['qty']} @ {item['price']}")

四、DDD建模的关键设计决策

1. 为什么用Repository模式而不是直接调API？

• 隔离变化：淘宝API从taobao.trades.sold.get换成taobao.trades.sold.getNew，只改TaobaoOrderAdapter，领域层零改

• 测试友好：单元测试注入MockRepository，不依赖真实网络

• 事务一致性：save()方法确保订单落库后才返回

2. 为什么Order是实体不是值对象？

• 有唯一标识order_id，生命周期内状态可变

• 跨平台统一后，业务层只认Order实体，不关心来自哪个平台

3. 为什么用mask_phone()而不是外部脱敏？

• 脱敏是收件人的领域行为，封装在值对象内

• 调用方无需知道脱敏逻辑，直接recipient.mask_phone()

4. 适配器里为什么要有STATUS_MAP？

• 每家平台的状态枚举不同（淘宝用字符串，拼多多用数字，亚马逊用驼峰）

• 适配器的核心职责之一就是翻译方言，STATUS_MAP是防腐层的心脏

五、DDD重构前后的对比

六、和前几篇的衔接

把本篇的OrderRepository端口接口，作为前篇ApiGateway的输出端：

• ApiGateway.call()返回原始JSON → *Adapter._to_order()翻译成Order实体

• TripleGuardClient的熔断逻辑嵌入*Adapter.list_orders()调用前

• CloudResidencyGuard的云内检查嵌入*Adapter.__init__()
  DDD防腐层 + 统一网关 + 三重守卫 + 云内着色 = 九家中台的完整战术实现。