日本电商 API 对接实战：从雅虎代拍到煤炉秒切的技术架构设计

from abc import ABC, abstractmethod
from dataclasses import dataclass
from typing import List, Optional

@dataclass
class AuctionItem:
    item_id: str
    title: str
    current_price: float
    currency: str
    end_time: datetime
    seller_id: str
    platform: str  # yahoo, mercari, rakuten, amazon

class PlatformAdapter(ABC):
    @abstractmethod
    def search_items(self, keyword: str, page: int = 1) -> List[AuctionItem]:
        pass
    
    @abstractmethod
    def get_item_detail(self, item_id: str) -> Optional[AuctionItem]:
        pass
    
    @abstractmethod
    def place_bid(self, item_id: str, price: float) -> bool:
        pass

class TokenManager:
    def __init__(self):
        self._token_store = {}
        self._lock = threading.Lock()
    
    def get_token(self, platform: str) -> str:
        with self._lock:
            token_info = self._token_store.get(platform)
            if not token_info or self._is_expired(token_info):
                token_info = self._refresh_token(platform)
                self._token_store[platform] = token_info
            return token_info["access_token"]
    
    def _is_expired(self, token_info: dict) -> bool:
        # 提前5分钟刷新，避免边界问题
        return token_info["expires_at"] - time.time() < 300

def place_bid_with_idempotency(self, item_id: str, price: float, request_id: str) -> bool:
    redis_key = f"bid:idempotent:{request_id}"
    
    # 尝试设置锁，成功说明是第一次处理
    if redis_client.set(redis_key, "processing", ex=3600, nx=True):
        try:
            result = self.adapter.place_bid(item_id, price)
            redis_client.set(redis_key, "success" if result else "failed", ex=86400)
            return result
        except Exception as e:
            redis_client.delete(redis_key)
            raise e
    else:
        # 已经处理过，等待结果
        for _ in range(10):
            status = redis_client.get(redis_key)
            if status == "success":
                return True
            elif status == "failed":
                return False
            time.sleep(0.5)
        raise TimeoutError("等待幂等结果超时")