代码解析：咸鱼item_get接口深度剖析与实战指南

import hashlibfrom urllib.parse import urlencodeclass XianyuSignGenerator:
    """
    某鱼签名生成器
    签名规则：
    1. 过滤空值和sign字段
    2. 按key ASCII码升序排序
    3. 拼接key+value字符串
    4. 末尾追加app_secret
    5. MD5加密并转为小写
    """
    
    def __init__(self, app_secret: str):
        self.app_secret = app_secret    
    def generate(self, params: dict) -> str:
        """
        @param params: 待签名的参数字典
        @return: 32位小写MD5签名
        """
        # 1. 过滤空值和sign字段
        filtered_params = {
            k: v for k, v in params.items() 
            if v is not None and k != 'sign'
        }
        
        # 2. 按key升序排序（ASCII码）
        sorted_items = sorted(filtered_params.items(), key=lambda x: x[0])
        
        # 3. 拼接key+value字符串
        sign_str = ''.join(f"{key}{value}" for key, value in sorted_items)
        
        # 4. 追加app_secret
        sign_str += self.app_secret        
        # 5. MD5加密
        md5_hash = hashlib.md5(sign_str.encode('utf-8')).hexdigest()
        
        return md5_hash.lower()# 单元测试示例def test_sign_generator():
    generator = XianyuSignGenerator("test_secret_123")
    params = {
        "method": "xianyu.item.get",
        "app_key": "12345678",
        "item_id": "658767890123",
        "timestamp": "1731923400"
    }
    sign = generator.generate(params)
    assert sign == "a1b2c3d4e5f6g7h8i9j0k1l2m3n4o5p6"  # 预期值
    print(f"生成签名: {sign}")if __name__ == "__main__":
    test_sign_generator()

import crypto from 'crypto';/**
 * 某鱼签名生成类
 */class XianyuSignGenerator {
  constructor(appSecret) {
    this.appSecret = appSecret;
  }

  /**
   * 生成MD5签名
   * @param {Object} params - 参数对象
   * @returns {string} 32位小写签名
   */
  generate(params) {
    // 1. 过滤空值和sign字段
    const filtered = Object.entries(params)
      .filter(([k, v]) => v !== null && v !== undefined && k !== 'sign')
      .reduce((acc, [k, v]) => ({ ...acc, [k]: String(v) }), {});

    // 2. 按key升序排序
    const sortedKeys = Object.keys(filtered).sort();
    
    // 3. 拼接key+value
    const signStr = sortedKeys.map(key => `${key}${filtered[key]}`).join('') + this.appSecret;
    
    // 4. MD5加密并转小写
    return crypto.createHash('md5')
      .update(signStr, 'utf8')
      .digest('hex')
      .toLowerCase();
  }}// 单元测试const generator = new XianyuSignGenerator("test_secret_123");const params = {
  method: "xianyu.item.get",
  app_key: "12345678",
  item_id: "658767890123",
  timestamp: "1731923400"};console.log("生成签名:", generator.generate(params));

import requestsimport timefrom typing import Dict, Optional, Tuplefrom dataclasses import dataclassfrom enum import Enumclass ApiErrorCode(Enum):
    """错误码枚举"""
    SUCCESS = 200
    PARAM_ERROR = 400
    UNAUTHORIZED = 401
    FORBIDDEN = 403
    NOT_FOUND = 404
    RATE_LIMIT = 429
    SERVER_ERROR = 500@dataclassclass ApiResponse:
    """统一响应结构"""
    success: bool
    data: Optional[Dict] = None
    error_code: Optional[int] = None
    error_msg: Optional[str] = None
    request_id: Optional[str] = Noneclass XianyuItemClient:
    """
    生产级某鱼商品详情客户端
    特性：
    - 自动签名
    - 指数退避重试
    - 熔断降级
    - 调用统计
    """
    
    def __init__(self, app_key: str, app_secret: str, access_token: str):
        self.base_url = "https://api.xianyu.com/router/rest"
        self.app_key = app_key
        self.sign_generator = XianyuSignGenerator(app_secret)
        self.access_token = access_token
        self.session = requests.Session()
        # 配置连接池
        adapter = requests.adapters.HTTPAdapter(
            pool_connections=20,
            pool_maxsize=50,
            max_retries=3
        )
        self.session.mount("https://", adapter)
        self.session.timeout = 10  # 全局超时
    
    def get_item_detail(
        self, 
        item_id: str, 
        fields: Optional[str] = None,
        max_retries: int = 3
    ) -> ApiResponse:
        """
        获取商品详情（带重试机制）
        
        @param item_id: 商品ID
        @param fields: 指定字段，逗号分隔
        @param max_retries: 最大重试次数
        @return: ApiResponse对象
        """
        params = {
            "method": "xianyu.item.get",
            "app_key": self.app_key,
            "access_token": self.access_token,
            "item_id": item_id,
            "timestamp": str(int(time.time())),
            "v": "2.0",
            "format": "json"
        }
        if fields:
            params["fields"] = fields        
        # 生成签名
        params["sign"] = self.sign_generator.generate(params)
        
        # 指数退避重试
        for attempt in range(max_retries):
            try:
                response = self.session.post(self.base_url, data=params)
                response.raise_for_status()
                
                result = response.json()
                return self._parse_response(result)
                
            except requests.Timeout:
                if attempt == max_retries - 1:
                    return ApiResponse(
                        success=False,
                        error_code=504,
                        error_msg="请求超时"
                    )
                time.sleep(2 ** attempt)  # 指数退避
                
            except requests.RequestException as e:
                return ApiResponse(
                    success=False,
                    error_code=500,
                    error_msg=f"网络异常: {str(e)}"
                )
    
    def _parse_response(self, result: Dict) -> ApiResponse:
        """解析响应数据"""
        code = result.get("code")
        
        if code == ApiErrorCode.SUCCESS.value:
            return ApiResponse(
                success=True,
                data=result.get("data"),
                request_id=result.get("data", {}).get("request_id")
            )
        else:
            return ApiResponse(
                success=False,
                error_code=code,
                error_msg=result.get("message"),
                request_id=result.get("request_id")
            )# 使用示例if __name__ == "__main__":
    client = XianyuItemClient(
        app_key="your_app_key",
        app_secret="your_app_secret",
        access_token="your_access_token"
    )
    
    resp = client.get_item_detail("658767890123", "title,price,desc")
    if resp.success:
        item = resp.data['item']
        print(f"商品标题: {item['title']}")
        print(f"当前价格: ¥{item['price']['current']}")
    else:
        print(f"请求失败: {resp.error_code} - {resp.error_msg}")

import axios, { AxiosInstance, AxiosError } from 'axios';import { HttpsAgent } from 'agentkeepalive';interface ItemDetail {
  item_id: string;
  title: string;
  desc: string;
  price: { current: string; original: string };
  images: string[];
  seller: { user_id: string; nick: string; credit_score: number };}interface ApiResponse<T> {
  success: boolean;
  data?: T;
  error_code?: number;
  error_msg?: string;
  request_id?: string;}export class XianyuItemClient {
  private readonly baseURL = 'https://api.xianyu.com/router/rest';
  private httpClient: AxiosInstance;
  private signGenerator: XianyuSignGenerator;

  constructor(
    private appKey: string,
    appSecret: string,
    private accessToken: string
  ) {
    this.signGenerator = new XianyuSignGenerator(appSecret);
    
    // 配置高性能HTTP客户端
    this.httpClient = axios.create({
      baseURL: this.baseURL,
      method: 'POST',
      timeout: 10000,
      // 使用keepalive提升性能
      httpAgent: new HttpsAgent({
        maxSockets: 50,
        maxFreeSockets: 20,
        timeout: 60000,
        freeSocketTimeout: 30000
      }),
      headers: {
        'Content-Type': 'application/x-www-form-urlencoded;charset=utf-8',
        'User-Agent': 'Xianyu-API-Client/1.0'
      }
    });
  }

  /**
   * 获取商品详情
   */
  async getItemDetail(
    itemId: string,
    fields?: string,
    maxRetries = 3
  ): Promise<ApiResponse<ItemDetail>> {
    const params = {
      method: 'xianyu.item.get',
      app_key: this.appKey,
      access_token: this.accessToken,
      item_id: itemId,
      timestamp: Math.floor(Date.now() / 1000).toString(),
      v: '2.0',
      format: 'json'
    };

    if (fields) {
      params.fields = fields;
    }
    
    params.sign = this.signGenerator.generate(params);

    // 指数退避重试策略
    for (let attempt = 0; attempt < maxRetries; attempt++) {
      try {
        const response = await this.httpClient.post('', new URLSearchParams(params));
        return this.parseResponse(response.data);
      } catch (error) {
        const axiosError = error as AxiosError;
        
        // 超时重试
        if (axiosError.code === 'ECONNABORTED' && attempt < maxRetries - 1) {
          const delay = Math.pow(2, attempt) * 1000;
          await this.sleep(delay);
          continue;
        }
        
        // 其他网络错误
        return {
          success: false,
          error_code: 500,
          error_msg: axiosError.message        };
      }
    }
    
    return {
      success: false,
      error_code: 504,
      error_msg: '请求超时'
    };
  }

  private parseResponse<T>(result: any): ApiResponse<T> {
    const code = result.code;
    
    if (code === 200) {
      return {
        success: true,
        data: result.data.item,
        request_id: result.data.request_id      };
    } else {
      return {
        success: false,
        error_code: code,
        error_msg: result.message,
        request_id: result.request_id      };
    }
  }

  private sleep(ms: number): Promise<void> {
    return new Promise(resolve => setTimeout(resolve, ms));
  }}// 使用示例const client = new XianyuItemClient('app_key', 'app_secret', 'access_token');async function demo() {
  const resp = await client.getItemDetail('658767890123', 'title,price');
  
  if (resp.success) {
    console.log(`商品: ${resp.data?.title}`);
    console.log(`价格: ¥${resp.data?.price.current}`);
  } else {
    console.error(`失败: ${resp.error_code} - ${resp.error_msg}`);
  }}demo();

from functools import wrapsimport threadingfrom datetime import datetime, timedeltaclass CircuitBreaker:
    """
    熔断器模式实现
    当失败率达到阈值时，快速失败，保护上游服务
    """
    
    def __init__(self, failure_threshold=5, recovery_timeout=60):
        self.failure_threshold = failure_threshold
        self.recovery_timeout = recovery_timeout
        self._failure_count = 0
        self._last_failure_time = None
        self._state = "CLOSED"  # CLOSED, OPEN, HALF_OPEN
        self._lock = threading.Lock()
    
    def call(self, func, *args, **kwargs):
        with self._lock:
            if self._state == "OPEN":
                if datetime.now() - self._last_failure_time > timedelta(seconds=self.recovery_timeout):
                    self._state = "HALF_OPEN"
                else:
                    raise Exception("Circuit breaker is OPEN")
        
        try:
            result = func(*args, **kwargs)
            self._on_success()
            return result        except Exception as e:
            self._on_failure()
            raise e    
    def _on_success(self):
        with self._lock:
            self._failure_count = 0
            self._state = "CLOSED"
    
    def _on_failure(self):
        with self._lock:
            self._failure_count += 1
            self._last_failure_time = datetime.now()
            
            if self._failure_count >= self.failure_threshold:
                self._state = "OPEN"# 在客户端中集成熔断器class XianyuItemClientWithCircuitBreaker(XianyuItemClient):
    def __init__(self, *args, **kwargs):
        super().__init__(*args, **kwargs)
        self.circuit_breaker = CircuitBreaker()
    
    def get_item_detail(self, item_id: str, **kwargs):
        return self.circuit_breaker.call(super().get_item_detail, item_id, **kwargs)

import asynciofrom concurrent.futures import ThreadPoolExecutorfrom typing import Listclass BatchItemSync:
    """批量商品同步器"""
    
    def __init__(self, client: XianyuItemClient, max_workers: int = 5):
        self.client = client
        self.max_workers = max_workers
        self.executor = ThreadPoolExecutor(max_workers=max_workers)
    
    def sync_batch(self, item_ids: List[str]) -> List[ApiResponse]:
        """
        批量同步商品信息
        使用线程池控制并发，避免限流
        """
        futures = [
            self.executor.submit(self.client.get_item_detail, item_id)
            for item_id in item_ids        ]
        
        results = []
        for future in futures:
            try:
                resp = future.result(timeout=15)
                results.append(resp)
            except Exception as e:
                results.append(ApiResponse(
                    success=False,
                    error_code=500,
                    error_msg=str(e)
                ))
        
        return results    
    async def sync_batch_async(self, item_ids: List[str]):
        """异步版本（使用aiohttp）"""
        # 异步实现代码略，原理相同
        pass# 使用示例syncer = BatchItemSync(client)item_ids = ["658767890123", "658767890124", "658767890125"]results = syncer.sync_batch(item_ids)for idx, resp in enumerate(results):
    print(f"商品{idx+1}: {'成功' if resp.success else '失败'}")

# 动态User-Agent池USER_AGENTS = [
    "XianyuApp/7.0.10 (iPhone; iOS 15.6; Scale/3.00)",
    "XianyuApp/7.0.8 (Linux; Android 12; SM-G9980)",
    # ...]def get_random_ua():
    return random.choice(USER_AGENTS)# 在请求头中设置self.session.headers["User-Agent"] = get_random_ua()

import randomimport timedef dynamic_delay(min_ms=1000, max_ms=3000):
    """随机延迟，模拟真人操作"""
    delay = random.randint(min_ms, max_ms) / 1000.0
    time.sleep(delay)# 在调用间插入延迟for item_id in item_list:
    resp = client.get_item_detail(item_id)
    dynamic_delay()

def validate_item_data(item: dict) -> bool:
    """校验返回数据的完整性"""
    required_fields = ['item_id', 'title', 'price', 'status']
    for field in required_fields:
        if field not in item:
            return False
    
    # 价格字段必须合法
    try:
        float(item['price']['current'])
    except (KeyError, ValueError):
        return False
    
    return True

import loggingimport jsonclass MonitoredClient(XianyuItemClient):
    def __init__(self, *args, **kwargs):
        super().__init__(*args, **kwargs)
        self.logger = logging.getLogger("xianyu.api")
    
    def get_item_detail(self, item_id: str, **kwargs):
        start_time = time.time()
        
        try:
            resp = super().get_item_detail(item_id, **kwargs)
            duration = time.time() - start_time            
            # 记录成功日志
            self.logger.info(json.dumps({
                "event": "api_call",
                "item_id": item_id,
                "success": resp.success,
                "duration_ms": round(duration * 1000, 2),
                "error_code": resp.error_code if not resp.success else None
            }))
            
            return resp            
        except Exception as e:
            duration = time.time() - start_time
            self.logger.error(json.dumps({
                "event": "api_exception",
                "item_id": item_id,
                "duration_ms": round(duration * 1000, 2),
                "error": str(e)
            }))
            raise