速卖通全球运营利器：商品详情接口多语言 + 合规 + 物流适配技术全解析

┌─────────────┐    ┌─────────────────────────────┐    ┌─────────────┐
│  接入层     │    │          适配层             │    │  业务层     │
│（API网关）  │───▶│ 多语言引擎 | 合规校验器      │───▶│（订单/商品）│
└─────────────┘    │ 物流适配模块 | 性能优化组件   │    └─────────────┘
                   └─────────────────────────────┘

from abc import ABC, abstractmethod
from typing import Dict, Optional
import redis
import requests
# 抽象翻译器基类
class TranslatorBase(ABC):
    @abstractmethod
    def translate(self, source_text: str, target_lang: str) -> str:
        pass
# 商家自定义翻译实现
class MerchantTranslator(TranslatorBase):
    def __init__(self, merchant_trans_data: Dict):
        self.trans_data = merchant_trans_data  # 商家提供的多语言映射表
    def translate(self, source_text: str, target_lang: str) -> str:
        # 优先返回商家自定义翻译
        return self.trans_data.get(target_lang, source_text)
# 自动翻译实现（对接速卖通官方接口）
class AutoTranslator(TranslatorBase):
    def __init__(self, app_key: str):
        self.app_key = app_key
        self.redis_client = redis.Redis(host="localhost", port=6379, db=0)
        self.expire_seconds = 86400  # 翻译结果缓存1天
    def translate(self, source_text: str, target_lang: str) -> str:
        cache_key = f"trans:{source_text}:{target_lang}"
        # 缓存命中直接返回
        if cached := self.redis_client.get(cache_key):
            return cached.decode()
        # 调用官方翻译接口
        resp = requests.get(
            "https://api.aliexpress.com/translate",
            params={
                "app_key": self.app_key,
                "text": source_text,
                "target": target_lang,
                "source": "en"  # 统一以英文为中转语言
            },
            timeout=3
        )
        result = resp.json()["translated_text"]
        self.redis_client.setex(cache_key, self.expire_seconds, result)
        return result
# 多语言引擎入口
class MultiLangEngine:
    def __init__(self, merchant_trans_data: Dict, app_key: str):
        self.merchant_trans = MerchantTranslator(merchant_trans_data)
        self.auto_trans = AutoTranslator(app_key)
    def get_content(self, base_content: Dict, target_lang: str) -> str:
        # 优先级：商家翻译 > 自动翻译 > 英文默认
        if target_lang in base_content:
            return self.merchant_trans.translate(base_content[target_lang], target_lang)
        return self.auto_trans.translate(base_content.get("en", ""), target_lang)

def format_currency(amount: float, target_country: str) -> str:
    """按目标国家格式标准化货币显示"""
    country_currency = {
        "US": ("$", "%.2f", ","),  # 美国：$1,234.56
        "DE": ("€", "%.2f", "."),  # 德国：1.234,56€
        "RU": ("₽", "%.2f", " "),  # 俄罗斯：1 234,56₽
        # 其他30+国家配置省略
    }
    symbol, fmt, sep = country_currency.get(target_country, ("$", "%.2f", ","))
    formatted = fmt % amount
    if sep in (",", " "):
        integer_part, decimal_part = formatted.split(".")
        # 千位分隔符处理
        reversed_int = integer_part[::-1]
        grouped = sep.join([reversed_int[i:i+3] for i in range(0, len(reversed_int), 3)])
        integer_part = grouped[::-1]
        formatted = f"{integer_part}.{decimal_part}"
    # 货币符号位置适配（部分国家符号在右侧）
    return f"{symbol}{formatted}" if target_country in ["US", "CN"] else f"{formatted}{symbol}"

{
  "EU": {
    "required_certs": ["CE", "RoHS"],
    "prohibited_materials": ["asbestos", "lead"],
    "data_requirements": ["product_origin", "importer_info"]
  },
  "US": {
    "required_certs": ["FDA", "FCC"],
    "prohibited_materials": ["phthalates"],
    "data_requirements": ["hs_code", "country_of_origin"]
  }
}

class ComplianceValidator:
    def __init__(self, rule_path: str):
        with open(rule_path, "r", encoding="utf-8") as f:
            self.rules = json.load(f)
    def validate(self, product_data: Dict, target_country: str) -> Dict:
        """返回合规校验结果与整改建议"""
        if target_country not in self.rules:
            return {"passed": True, "message": "No specific compliance requirements"}
        rule = self.rules[target_country]
        issues = []
        # 认证检查
        missing_certs = [c for c in rule["required_certs"] if c not in product_data.get("certs", [])]
        if missing_certs:
            issues.append(f"Missing certifications: {', '.join(missing_certs)}")
        # 成分检查
        forbidden_materials = [m for m in rule["prohibited_materials"] 
                              if m in product_data.get("materials", [])]
        if forbidden_materials:
            issues.append(f"Forbidden materials: {', '.join(forbidden_materials)}")
        # 数据完整性检查
        missing_fields = [f for f in rule["data_requirements"] if f not in product_data]
        if missing_fields:
            issues.append(f"Missing required fields: {', '.join(missing_fields)}")
        return {
            "passed": len(issues) == 0,
            "message": "; ".join(issues) if issues else "Compliance check passed",
            "suggestions": self._get_suggestions(issues, target_country)
        }
    def _get_suggestions(self, issues: list, country: str) -> list:
        """生成针对性整改建议"""
        suggestions = []
        if any("Missing certifications" in i for i in issues):
            suggestions.append(f"Apply for required certifications via Aliexpress Compliance Center (avg 7-10 days)")
        if any("Forbidden materials" in i for i in issues):
            suggestions.append(f"Replace with compliant alternatives (see Aliexpress Material Guide for {country})")
        return suggestions

def match_logistics_plan(product_id: str, target_country: str) -> Dict:
    """
    物流方案匹配优先级：
    1. 目标国本地仓 > 2. 区域中转仓 > 3. 国内直发
    """
    # 1. 检查本地仓库存
    if local_stock := check_stock(product_id, target_country):
        return {
            "logistics_type": "local_warehouse",
            "delivery_time": "3-7 business days",
            "cost": calculate_local_cost(product_id, target_country),
            "tracking_support": True
        }
    # 2. 检查区域仓库存（如东南亚→新加坡仓，欧洲→德国仓）
    region = get_region_by_country(target_country)
    if region_stock := check_stock(product_id, region):
        return {
            "logistics_type": "regional_warehouse",
            "delivery_time": "5-12 business days",
            "cost": calculate_region_cost(product_id, region, target_country),
            "tracking_support": True
        }
    # 3. 国内直发方案
    return {
        "logistics_type": "direct_shipping",
        "delivery_time": "10-25 business days",
        "cost": calculate_direct_cost(product_id, target_country),
        "tracking_support": False
    }

┌─────────────┐  优先命中  ┌─────────────┐  其次命中  ┌─────────────┐
│  本地缓存   │◄─────────►│ 区域节点缓存 │◄─────────►│ 中心缓存池  │
│（内存，10s）│           │（Redis，1h） │           │（Redis，24h）│
└─────────────┘           └─────────────┘           └─────────────┘

import boto3
def get_secret(secret_name: str) -> str:
    client = boto3.client('secretsmanager', region_name='us-east-1')
    return client.get_secret_value(SecretId=secret_name)['SecretString']