妙手 ERP 采集箱管理 Skill

API 封装概览

| 方法 | 妙手接口 | 说明 | |------|----------|------| | list_collect_boxes() | collectbox.list | 查询公共采集箱列表、分页、筛选 | | get_collect_box(box_id) | collectbox.get | 获取单个采集箱详情 | | create_collect_box(data) | collectbox.add | 新增商品到公共采集箱 | | update_collect_box(box_id, data) | collectbox.edit | 编辑采集箱商品信息 | | delete_collect_box(box_id) | collectbox.delete | 删除采集箱商品 | | batch_import(items) | collectbox.batch_add | 批量导入 (分批、幂等、进度回调) | | preview_source_link(url) | source.preview | 预览来源链接 (校验有效性) | | fetch_source_detail(url) | source.fetch | 抓取来源详情 (标题、图片、SKU、价格等) |

快速开始

安装配置

from miaoshou_erp import MiaoShouClient, CollectBoxManager

# 初始化客户端
client = MiaoShouClient(
    app_key="your_app_key",
    app_secret="your_app_secret",
    access_token="your_access_token",  # 或通过 OAuth 获取
    sandbox=False,  # True=沙箱环境
)

# 采集箱管理器
collect_box = CollectBoxManager(client)

基础操作

# 1. 查询采集箱列表
boxes = await collect_box.list_collect_boxes(
    page=1,
    page_size=50,
    status="normal",  # normal/deleted
    keyword="瑜伽裤",
)
print(f"共 {boxes.total} 条，当前页 {len(boxes.items)} 条")

# 2. 新增商品到采集箱
new_item = {
    "source_url": "https://www.temu.com/sg/product-12345.html",
    "title": "高腰无痕瑜伽裤 女 速干 显瘦",
    "category_id": 12345,  # 妙手类目 ID
    "brand": "自有品牌",
    "images": [
        "https://cdn.example.com/img1.jpg",
        "https://cdn.example.com/img2.jpg",
    ],
    "skus": [
        {"specs": {"颜色": "黑色", "尺码": "S"}, "price": "89.00", "stock": 50, "sku_code": "YGK-S-BLK"},
        {"specs": {"颜色": "黑色", "尺码": "M"}, "price": "89.00", "stock": 50, "sku_code": "YGK-M-BLK"},
    ],
    "description": "<p>高弹面料，不勒肉不下滑...</p>",
    "attrs": {"材质": "尼龙+氨纶", "产地": "中国"},
}

result = await collect_box.create_collect_box(new_item)
print(f"创建成功: box_id={result.box_id}")

# 3. 批量导入 (自动分批、重试、进度)
items = [...]  # 100 条商品数据
progress = await collect_box.batch_import(
    items,
    batch_size=20,
    on_progress=lambda done, total: print(f"进度: {done}/{total}"),
    on_error=lambda item, err: logger.error(f"导入失败: {item}, {err}"),
)
print(f"成功: {progress.success}, 失败: {progress.failed}")

# 4. 来源链接预览 + 抓取 (上架前必做)
preview = await collect_box.preview_source_link("https://www.temu.com/xxx.html")
if preview.valid:
    detail = await collect_box.fetch_source_detail("https://www.temu.com/xxx.html")
    # 用 detail 覆盖/补全 new_item 字段
    print(f"抓取到 {len(detail.images)} 张图，{len(detail.skus)} 个 SKU")

核心模型定义

from pydantic import BaseModel, Field, HttpUrl
from typing import List, Optional, Dict, Any
from enum import Enum


class CollectBoxStatus(str, Enum):
    NORMAL = "normal"
    DELETED = "deleted"


class CollectBoxItem(BaseModel):
    box_id: Optional[str] = None
    source_url: HttpUrl
    title: str = Field(max_length=100)
    category_id: int
    brand: Optional[str] = None
    images: List[HttpUrl] = Field(min_length=1, max_length=9)
    skus: List["SKUItem"]
    description: Optional[str] = None
    attrs: Dict[str, str] = Field(default_factory=dict)
    # 妙手特有字段
    source_type: str = "temu"  # temu/1688/other
    is_cross_border: bool = True
    shipping_template_id: Optional[int] = None


class SKUItem(BaseModel):
    specs: Dict[str, str]  # {"颜色": "黑色", "尺码": "M"}
    price: str  # 字符串格式价格，保留两位小数
    stock: int = Field(ge=0)
    sku_code: Optional[str] = None  # 商家编码
    barcode: Optional[str] = None
    cost_price: Optional[str] = None


class BatchImportResult(BaseModel):
    total: int
    success: int
    failed: int
    errors: List[Dict[str, Any]]
    box_ids: List[str]

限流与重试策略

妙手 ERP API 限流规则（官方文档 + 实测）：

QPS: 单应用 ≤ 50 QPS（峰值），建议常态 ≤ 20 QPS
并发: 单 IP ≤ 10 并发连接
日配额: 部分接口有日调用上限 (如 source.fetch 5000 次/天)

本 Skill 内置策略：

from tenacity import retry, stop_after_attempt, wait_exponential, retry_if_exception_type

RETRY_CONFIG = {
    "stop": stop_after_attempt(3),
    "wait": wait_exponential(multiplier=1, min=2, max=30),
    "retry": retry_if_exception_type((RateLimitError, TimeoutError, ServerError)),
    "before_sleep": log_retry_attempt,
}

# 令牌桶限流器 (全局单例)
rate_limiter = TokenBucket(rate=20, burst=30)  # 20 QPS, 突发 30

async def _request_with_limit(self, method: str, path: str, **kwargs):
    await self.rate_limiter.acquire()
    return await self._request_with_retry(method, path, **kwargs)

错误分类与处理

| 错误码 | 含义 | 处理建议 | |--------|------|----------| | 40001 | 参数校验失败 | 检查必填字段、格式、长度 | | 40002 | 来源链接无效 | 预览接口校验、更换链接 | | 40003 | 类目不存在/不匹配 | 调用类目推荐接口、人工核对 | | 40101 | 签名验证失败 | 检查 AppSecret、时间戳、参数排序 | | 40102 | Token 过期/无效 | 刷新 Access Token | | 40301 | 无权限操作 | 确认授权范围、店铺绑定 | | 42901 | 触发限流 | 指数退避重试、降低并发 | | 50001 | 系统内部错误 | 记录日志、人工介入、稍后重试 |

class MiaoShouError(Exception):
    def __init__(self, code: str, message: str, data: Any = None):
        self.code = code
        self.message = message
        self.data = data
        super().__init__(f"[{code}] {message}")


ERROR_MAP = {
    "40001": ValidationError,
    "40002": InvalidSourceUrlError,
    "40003": CategoryMismatchError,
    "40101": SignatureError,
    "40102": TokenExpiredError,
    "40301": PermissionDeniedError,
    "42901": RateLimitError,
    "50001": ServerError,
}

def handle_response(response: Dict) -> Any:
    if response.get("code") == "0":
        return response.get("data")
    error_class = ERROR_MAP.get(response.get("code"), MiaoShouError)
    raise error_class(response["code"], response.get("msg", "Unknown error"), response.get("data"))

幂等设计

批量导入必须幂等，防止重复提交造成脏数据：

import hashlib
import json

def generate_idempotency_key(item: CollectBoxItem) -> str:
    """基于核心字段生成幂等键"""
    core = {
        "source_url": str(item.source_url),
        "title": item.title,
        "skus": [(k, v) for sku in item.skus for k, v in sku.specs.items()],
    }
    return hashlib.md5(json.dumps(core, sort_keys=True).encode()).hexdigest()

# 使用方式
async def batch_import(self, items: List[CollectBoxItem], ...):
    for item in items:
        idempotency_key = generate_idempotency_key(item)
        # 检查 Redis 是否已处理
        if await redis.exists(f"idempotent:{idempotency_key}"):
            continue  # 跳过已处理
        # 处理...
        await redis.setex(f"idempotent:{idempotency_key}", 86400, "done")

实战最佳实践

1. 采集 → 入箱标准流程

async def collect_and_import(temu_urls: List[str], category_map: Dict[str, int]):
    """TEMU 链接批量采集入妙手采集箱"""
    results = []
    
    for url in temu_urls:
        # 步骤 1: 预览校验
        preview = await collect_box.preview_source_link(url)
        if not preview.valid:
            results.append({"url": url, "status": "invalid_source", "reason": preview.reason})
            continue
        
        # 步骤 2: 抓取详情
        detail = await collect_box.fetch_source_link(url)
        
        # 步骤 3: 数据标准化 + 业务规则
        item = CollectBoxItem(
            source_url=url,
            title=optimize_title(detail.title),  # 关键词植入、合规过滤
            category_id=category_map.get(detail.category_name, default_category),
            images=detail.images[:9],
            skus=transform_skus(detail.skus),   # 价格公式、库存预设、规格标准化
            description=clean_html(detail.description),
            attrs=extract_attrs(detail.attrs),
        )
        
        # 步骤 4: 幂等创建
        try:
            result = await collect_box.create_collect_box(item.model_dump())
            results.append({"url": url, "status": "success", "box_id": result.box_id})
        except MiaoShouError as e:
            results.append({"url": url, "status": "failed", "error": str(e)})
    
    return results

2. 价格公式引擎

def calculate_sale_price(cost: float, formula: str = "cost * 2 + 4") -> str:
    """支持表达式的价格计算"""
    # 安全评估表达式
    allowed_names = {"cost": cost, "round": round, "max": max, "min": min}
    price = eval(formula, {"__builtins__": {}}, allowed_names)
    return f"{round(price, 2):.2f}"

# 使用
for sku in item.skus:
    sku.price = calculate_sale_price(float(sku.cost_price or sku.price))

3. 图片托管与 CDN

async def upload_images_to_cdn(local_paths: List[str]) -> List[str]:
    """上传本地图片到 CloudStudio/R2/OSS，返回 CDN URL"""
    urls = []
    for path in local_paths:
        # 1. 压缩优化 (WebP, 质量 85%, 长边 ≤ 2048)
        optimized = optimize_image(path)
        # 2. 上传
        url = await cdn_client.upload(optimized, key=f"products/{sku_code}/{filename}")
        urls.append(url)
    return urls

测试

# tests/test_collectbox.py
import pytest
from unittest.mock import AsyncMock, patch
from miaoshou_erp import CollectBoxManager


@pytest.fixture
def mock_client():
    client = AsyncMock()
    client.post = AsyncMock(return_value={"code": "0", "data": {"box_id": "test_123"}})
    return client


@pytest.mark.asyncio
async def test_create_collect_box(mock_client):
    manager = CollectBoxManager(mock_client)
    item = {"source_url": "https://temu.com/x", "title": "Test", "category_id": 1, "skus": [{"specs": {}, "price": "10", "stock": 1}]}
    
    result = await manager.create_collect_box(item)
    
    assert result.box_id == "test_123"
    mock_client.post.assert_called_once()
    # 验证请求参数
    call_args = mock_client.post.call_args
    assert call_args[0][0] == "collectbox.add"
    assert call_args[1]["json"]["title"] == "Test"

更新日志

v1.3.0 (2026-08-20)

新增 source.preview / source.fetch 封装
优化批量导入进度回调、错误分类
增加幂等键生成工具函数
适配妙手 ERP API v2.1 新字段

v1.2.0 (2026-07-10)

令牌桶限流器重构，支持多实例共享
新增沙箱环境切换
完善错误码映射表

v1.1.0 (2026-06-15)

批量导入支持断点续传
新增图片 CDN 上传集成示例
类目映射缓存机制

v1.0.0 (2026-05-20)

核心 CRUD 封装
基础重试/限流
Pydantic 模型定义

常见问题

下载链接失效怎么办？
请优先切换其他网盘渠道，并反馈失效链接。

会员能下载所有资料吗？
页面标注为会员专享的资料支持会员下载，具体以当前页面显示为准。

妙手 ERP 采集箱管理 Skill

Skill / 插件详情