万邦api博客

主流电商平台（淘宝/1688/京东/拼多多）商品比价与数据分析实战指南

Sat, 06 Jun 2026 17:32:08 +0800

在电商运营中，比价和价格数据分析是制定定价策略、优化选品、提升利润的核心手段。淘宝、1688、京东、拼多多四大平台定位不同、用户画像不同、价格体系也不同——同一商品在不同平台的价格差异可能高达 30% 以上。本文将从数据采集、比价方法、数据分析模型到实战应用，系统讲解如何搭建一套多平台商品比价与数据分析体系。

一、四大平台价格特征与定位差异

点击测试获取key

sandbox:///mnt/agents/output/price_comparison_analysis.png

表格

平台	定位	价格特征	核心用户	比价价值
淘宝/天猫	综合零售	价格中等，品牌溢价明显，促销频繁	全年龄段，注重品质与性价比	基准价格参考，品牌定价锚点
京东	品质电商	价格偏高，自营正品保障，物流快	一二线城市，高消费能力	品质溢价上限，服务价值量化
拼多多	下沉市场	价格最低，百亿补贴力度大，白牌为主	价格敏感型用户	价格底线参考，供应链成本探测
1688	B2B批发	价格最低（批发价），起订量要求	商家、代采、跨境卖家	进货成本基准，利润空间测算

从上图的价格分布可以看出：1688 作为批发平台价格最低，京东作为品质平台价格最高，淘宝和拼多多处于中间区间。这种价格梯度为商家提供了清晰的定价参考框架。

二、数据采集：比价系统的基石

2.1 三种主流采集方式

表格

方式	原理	优点	缺点	适用场景
官方 API	调用平台开放接口获取结构化数据	稳定、合法、数据精准	需申请权限，字段受限，有调用频率限制	品牌方、大型卖家
第三方数据服务	接入万邦、鲸昔等聚合 API	一次对接多平台，开发成本低	需付费，数据延迟 3-5 分钟	中小卖家、快速验证
爬虫采集	模拟浏览器抓取页面数据	灵活、字段完整	反爬严格，IP 易被封，法律风险	有技术团队的企业

2.2 各平台官方 API 对比

表格

平台	核心比价接口	数据更新频率	调用限制	认证要求
淘宝/天猫	`taobao.item.price.get`、`taobao.item.search`	实时	500次/秒	企业开发者 + OAuth
京东	`jingdong.ware.price.get`、`jingdong.ware.search`	实时	200ms/次	企业开发者 + AppKey
拼多多	`pdd.goods.price.check`、`pdd.goods.search`	5分钟	视套餐	企业开发者
1688	`alibaba.product.search`、`alibaba.product.get`	5分钟	视套餐	企业开发者

关键注意点：

1688 的 API 数据同步延迟可能超 24 小时，选择服务商时需确认数据时效性，否则可能导致定价失误，平均损失占月营收的 5%-8%
淘宝反爬机制极强，非 API 方式需处理滑块验证、字体反爬、CSS 偏移等

三、核心比价数据分析模型

3.1 价格带分析

sandbox:///mnt/agents/output/price_strategy_dashboard.png

价格带是指同一品类在不同平台的价格区间分布。通过价格带分析，可以明确自身产品的定价位置：

Python

# 价格带分析示例def price_band_analysis(category, platform_prices):
    """
    分析某品类在各平台的价格分布
    platform_prices: {平台: [价格列表]}
    """
    results = {}
    for platform, prices in platform_prices.items():
        results[platform] = {
            "min": min(prices),
            "max": max(prices),
            "median": np.median(prices),
            "q1": np.percentile(prices, 25),   # 下四分位
            "q3": np.percentile(prices, 75),   # 上四分位
            "mean": np.mean(prices)
        }
    return results# 示例输出# 数码3C品类：# 淘宝: {min: 280, max: 380, median: 320, q1: 300, q3: 340}# 京东: {min: 300, max: 400, median: 340, q1: 320, q3: 360}# 拼多多: {min: 240, max: 340, median: 280, q1: 260, q3: 300}# 1688: {min: 180, max: 280, median: 220, q1: 200, q3: 240}

分析结论：

若你的数码产品定价 > 京东中位数，需有极强的品牌溢价或服务差异化
若定价 < 拼多多中位数，需警惕是否为假冒伪劣或亏本引流
淘宝中位数 ± 10% 是最安全的定价区间

3.2 价格弹性分析

价格弹性衡量的是价格变化 1% 带来的销量变化百分比，是制定促销策略的核心依据：

Ed=%ΔP%ΔQ=(P1+P2)/2P2−P1(Q1+Q2)/2Q2−Q1 

表格

弹性类型	弹性系数	特征	定价策略
富有弹性	\|Ed\| > 1	降价带来销量大幅增长	适合薄利多销，如拼多多
单位弹性	\|Ed\| = 1	降价幅度 = 销量增长幅度	利润最大化点
缺乏弹性	\|Ed\| < 1	降价对销量刺激有限	适合维持高价，如京东自营

从上图的价格弹性曲线可以看出：拼多多的价格敏感度最高（降价 20% 销量增长 45%），京东的价格敏感度最低（降价 20% 销量增长 28%）。这反映了两个平台用户群体的消费心理差异。

3.3 促销周期价格分析

大促期间的价格策略直接影响全年利润。通过历史数据分析，可以识别各平台的促销规律：

表格

促销节点	淘宝/天猫	京东	拼多多	1688
618 大促	满 300 减 50，预售定金膨胀	百亿补贴 + 京东秒杀	百亿补贴直降，无套路	批发满减，起订量降低
双 11	跨店满减 + 红包雨	京贴 + 品类券叠加	百亿补贴 + 限时秒杀	年终清库存，量大从优
日常	聚划算、淘金币	京东秒杀、闪购	百亿补贴常驻	阶梯价、混批优惠

关键发现：

京东大促降价幅度最大（日常 220 元 → 618 大促 165 元，降幅 25%）
1688 价格波动最小，但起订量要求会随促销降低
拼多多日常价已接近其他平台大促价，这是其核心竞争力

四、比价系统的技术架构

4.1 系统架构设计

plain

┌─────────────────────────────────────────────────────────────┐
│                    多平台比价系统架构                          │
├─────────────────────────────────────────────────────────────┤
│  数据采集层                                                  │
│  ├─ 淘宝 API / 爬虫  → 商品标题、价格、销量、评价、SKU        │
│  ├─ 京东 API / 爬虫  → 自营价、第三方价、促销信息、库存        │
│  ├─ 拼多多 API       → 团购价、单买价、百亿补贴价             │
│  └─ 1688 API         → 批发价、起订量、混批规则、运费          │
├─────────────────────────────────────────────────────────────┤
│  数据清洗层                                                  │
│  ├─ 去重：同一商品多链接合并                                  │
│  ├─ 标准化：统一货币单位、规格单位、时间格式                   │
│  ├─ 异常值处理：剔除 0 元、99999 元等异常价格                 │
│  └─ 到手价计算：叠加满减、优惠券、红包后的真实支付价格          │
├─────────────────────────────────────────────────────────────┤
│  存储层                                                      │
│  ├─ 时序数据库（InfluxDB）：价格历史数据                       │
│  ├─ 关系数据库（PostgreSQL）：商品基础信息、SKU 关系          │
│  └─ 缓存（Redis）：热点商品价格实时缓存                        │
├─────────────────────────────────────────────────────────────┤
│  分析引擎层                                                  │
│  ├─ 价格趋势预测：ARIMA / LSTM 时序模型                      │
│  ├─ 竞品关联：标题相似度 + 图片哈希匹配同款                    │
│  ├─ 定价建议：基于成本 + 竞品价格 + 弹性系数的动态定价         │
│  └─ 异常检测：突然降价/涨价识别，触发告警                     │
├─────────────────────────────────────────────────────────────┤
│  应用层                                                      │
│  ├─ 可视化看板：价格走势、溢价率热力图、波动率对比             │
│  ├─ 告警推送：微信/钉钉/邮件，价格突破阈值实时通知            │
│  └─ 报表导出：Excel / PDF，支持按品类、店铺、时间维度筛选       │
└─────────────────────────────────────────────────────────────┘

4.2 核心代码：多平台比价引擎

Python

import requestsimport pandas as pdfrom datetime import datetimeimport hashlibimport hmacclass MultiPlatformPriceComparator:
    """多平台商品比价引擎"""
    
    def __init__(self, api_keys):
        self.api_keys = api_keys
        self.platform_apis = {
            "taobao": "https://api.taobao.com/router/rest",
            "jd": "https://api.jd.com/routerjson",
            "pdd": "https://api.pinduoduo.com/api/router",
            "1688": "https://api.1688.com/openapi/"
        }
    
    def fetch_price(self, platform, keyword_or_sku):
        """获取单平台商品价格"""
        if platform == "taobao":
            return self._fetch_taobao(keyword_or_sku)
        elif platform == "jd":
            return self._fetch_jd(keyword_or_sku)
        elif platform == "pdd":
            return self._fetch_pdd(keyword_or_sku)
        elif platform == "1688":
            return self._fetch_1688(keyword_or_sku)
    
    def _fetch_taobao(self, keyword):
        """淘宝价格采集"""
        params = {
            "method": "taobao.item.search",
            "app_key": self.api_keys["taobao"]["app_key"],
            "q": keyword,
            "fields": "num_iid,title,price,sold_quantity,nick,pic_url",
            "page_size": 20
        }
        # 签名逻辑...
        resp = requests.get(self.platform_apis["taobao"], params=params, timeout=10)
        items = resp.json().get("items", [])
        return [{
            "platform": "淘宝",
            "title": item["title"],
            "price": float(item["price"]),
            "sales": int(item.get("sold_quantity", 0)),
            "shop": item["nick"],
            "url": f"https://item.taobao.com/item.htm?id={item['num_iid']}"
        } for item in items]
    
    def _calculate_real_price(self, item):
        """计算到手价（含促销）"""
        base_price = item["price"]
        # 叠加满减
        if item.get("discount"):
            base_price *= (1 - item["discount"])
        # 叠加优惠券
        if item.get("coupon"):
            base_price -= item["coupon"]
        # 叠加平台补贴
        if item.get("subsidy"):
            base_price -= item["subsidy"]
        return round(max(base_price, 0.01), 2)
    
    def compare_same_product(self, product_identifiers):
        """
        跨平台同款比价
        product_identifiers: {
            "taobao": "sku_id_1",
            "jd": "sku_id_2", 
            "pdd": "sku_id_3",
            "1688": "sku_id_4"
        }
        """
        results = []
        for platform, sku_id in product_identifiers.items():
            data = self.fetch_price(platform, sku_id)
            if data:
                item = data[0]
                item["real_price"] = self._calculate_real_price(item)
                results.append(item)
        
        # 计算价差
        if results:
            prices = [r["real_price"] for r in results]
            min_price = min(prices)
            for r in results:
                r["premium"] = round((r["real_price"] - min_price) / min_price * 100, 2)
        
        return pd.DataFrame(results)
    
    def generate_price_report(self, comparison_df):
        """生成比价分析报告"""
        report = {
            "generated_at": datetime.now().strftime("%Y-%m-%d %H:%M:%S"),
            "product_count": len(comparison_df),
            "price_stats": {
                "highest": comparison_df["real_price"].max(),
                "lowest": comparison_df["real_price"].min(),
                "mean": comparison_df["real_price"].mean(),
                "median": comparison_df["real_price"].median(),
                "std": comparison_df["real_price"].std()
            },
            "premium_analysis": comparison_df.groupby("platform")["premium"].mean().to_dict(),
            "recommendations": []
        }
        
        # 定价建议
        lowest_platform = comparison_df.loc[comparison_df["real_price"].idxmin(), "platform"]
        report["recommendations"].append(
            f"进货渠道建议：{lowest_platform} 价格最低，可作为主要采购来源"
        )
        
        # 如果某平台溢价率 > 30%，建议关注是否为品牌溢价或异常定价
        high_premium = comparison_df[comparison_df["premium"] > 30]
        if not high_premium.empty:
            report["recommendations"].append(
                f"高溢价平台：{', '.join(high_premium['platform'].tolist())}，建议核查是否为正品或存在品牌溢价"
            )
        
        return report

五、实战应用场景

5.1 品牌方控价

品牌方需要监控全网经销商是否遵守最低零售价（MAP）政策：

Python

def brand_price_control(brand_name, min_price_map):
    """
    品牌控价监控
    min_price_map: {sku_id: 最低零售价}
    """
    violations = []
    for sku_id, min_price in min_price_map.items():
        # 采集全网该 SKU 价格
        prices = fetch_all_platform_prices(sku_id)
        for p in prices:
            if p["real_price"] < min_price:
                violations.append({
                    "sku_id": sku_id,
                    "platform": p["platform"],
                    "shop": p["shop"],
                    "current_price": p["real_price"],
                    "min_price": min_price,
                    "violation_rate": round((min_price - p["real_price"]) / min_price * 100, 2),
                    "screenshot": capture_screenshot(p["url"]),
                    "timestamp": datetime.now()
                })
    return violations

输出结果可用于：

自动发送律师函或平台投诉
生成月度控价报告提交给管理层
识别低价倾销的经销商，优化渠道管理

5.2 选品定价决策

通过比价数据辅助新品定价：

Python

def pricing_recommendation(cost_price, category, target_platform):
    """
    基于竞品数据的定价建议
    """
    # 获取同品类竞品数据
    competitors = fetch_category_prices(category, target_platform)
    
    # 计算价格分位
    p25 = np.percentile(competitors, 25)
    p50 = np.percentile(competitors, 50)
    p75 = np.percentile(competitors, 75)
    
    # 定价策略
    if cost_price < p25:
        # 成本优势大，可低价抢占市场
        recommended = p25 * 0.95
        strategy = "渗透定价"
    elif cost_price < p50:
        # 成本中等，跟随市场
        recommended = p50
        strategy = "竞争定价"
    else:
        # 成本高，需差异化
        recommended = p75 * 1.05
        strategy = "撇脂定价"
    
    margin = (recommended - cost_price) / recommended * 100
    
    return {
        "recommended_price": round(recommended, 2),
        "strategy": strategy,
        "expected_margin": round(margin, 2),
        "competitor_range": f"{p25} - {p75}",
        "confidence": "高" if len(competitors) > 50 else "中"
    }

5.3 跨境代采决策

1688 作为货源平台，其价格直接决定跨境卖家的利润空间：

Python

def cross_border_sourcing_analysis(product_keyword):
    """
    跨境代采利润分析
    """
    # 获取 1688 批发价
    _1688_items = fetch_1688_prices(product_keyword)
    
    # 获取目标市场零售价（如 Shopee 东南亚）
    shopee_items = fetch_shopee_prices(product_keyword)
    
    analysis = []
    for item_1688 in _1688_items:
        cost = item_1688["price"]  # 1688 批发价
        moq = item_1688["min_order"]  # 起订量
        
        # 找同款在 Shopee 的售价
        matching_shopee = find_matching_item(item_1688, shopee_items)
        if matching_shopee:
            selling_price = matching_shopee["price"]
            
            # 计算利润
            logistics_cost = selling_price * 0.15  # 物流约 15%
            platform_fee = selling_price * 0.05   # 平台佣金约 5%
            profit = selling_price - cost - logistics_cost - platform_fee
            margin = profit / selling_price * 100
            
            analysis.append({
                "product": item_1688["title"],
                "1688_price": cost,
                "shopee_price": selling_price,
                "moq": moq,
                "logistics": round(logistics_cost, 2),
                "platform_fee": round(platform_fee, 2),
                "profit": round(profit, 2),
                "margin": round(margin, 2),
                "recommendation": "推荐" if margin > 30 else "谨慎"
            })
    
    return sorted(analysis, key=lambda x: x["margin"], reverse=True)

六、数据可视化与报表

6.1 核心监控指标

表格

指标	计算方式	业务含义
价格溢价率	(平台价格 - 最低价) / 最低价 × 100%	该平台相对于最优采购价的溢价程度
价格波动率	日价格变化的标准差	价格稳定性，波动大说明促销频繁或竞争激烈
价差幅度	最高价 - 最低价	跨平台套利空间
价格弹性系数	销量变化% / 价格变化%	用户对价格敏感度
到手价准确率	正确计算促销后价格的占比	系统算法可靠性

6.2 可视化看板设计

建议搭建以下看板模块：

实时价格大屏：四平台同款商品实时价格对比，自动刷新
价格趋势图：选定 SKU 近 30 天价格走势，标注促销节点
溢价率热力图：按月份 × 平台展示溢价率变化
告警列表：突破阈值的价格异常，支持一键跳转处理
选品推荐榜：基于价差和销量的潜力商品排行

七、常见问题与解决方案

表格

问题	原因	解决方案
价格数据延迟	API 缓存或服务商同步慢	选择数据同步延迟 < 5 分钟的服务商，或自建爬虫补充
到手价计算不准	促销活动复杂，规则多变	接入平台促销 API，或训练 NLP 模型解析活动规则
同款匹配困难	标题差异大，图片不同	结合标题相似度（余弦相似度）+ 图片感知哈希（pHash）
IP 被封	请求频率过高	代理 IP 池 + 请求频率随机化（2-5 秒间隔）
数据量过大	监控 SKU 数量多	分库分表 + 冷热数据分离，历史数据归档到对象存储
法律合规风险	爬虫可能违反平台协议	优先使用官方 API，爬虫需遵守 robots.txt，数据脱敏处理

八、总结

多平台商品比价与数据分析是电商精细化运营的核心能力。通过系统化的数据采集、清洗、分析和可视化，商家可以：

摸清价格底线：以 1688 批发价为成本基准，拼多多为零售底线
锚定品牌溢价：以京东价格为品质溢价上限，淘宝为大众市场参考
捕捉套利机会：发现跨平台价差 > 20% 的商品，快速调整采购或定价策略
优化促销节奏：基于历史数据分析，在最佳时机推出最优折扣

一文通，第三方接口如何实现批量上货，主流平台[淘宝|京东|1688|抖音）和跨境平台

Fri, 05 Jun 2026 11:54:11 +0800

一、国内主流平台批量上货接口

1. 淘宝 / 天猫（TOP 开放平台）

淘宝官方提供了完整的商品生命周期管理 API，支持从发布到上架的全流程自动化。

表格

核心接口	功能	说明
`taobao.item.add`	发布新商品	最核心接口，需传类目ID、标题、价格、SKU、主图等
`taobao.item.update`	编辑已有商品	修改价格、库存、标题、详情等
`taobao.item.shelf` / `taobao.item.unshelf`	上架/下架	控制商品售卖状态
`taobao.item.img.upload`	上传图片	主图、详情图需先上传获取图片URL
`taobao.itemcats.get`	获取类目树	发布前必须确定正确的类目ID
`taobao.item.propimg.upload`	上传属性图	SKU颜色/规格对应的图片

接入流程：

在淘宝开放平台注册企业开发者账号（个人权限受限）
创建应用，申请 "商品发布" 权限（需审核）
商家通过 OAuth 2.0 授权你的应用访问其店铺
获取 Access Token 后调用接口

关键注意点：

类目必须准确：先用 taobao.itemcats.get 查类目ID，不同类目要求的属性字段不同
SKU 规格处理：颜色/尺寸组合需生成 sku_properties 字段，格式如 1627207:3232481;20509:28383
图片预上传：所有图片必须先调用上传接口获取淘宝图片空间的 URL，不能直接贴外部链接
智能发布：2026年淘宝推出了新商品管理API，支持智能推荐类目、自动抠白底图，能大幅降低发布门槛

2. 京东（JOS 开放平台）

京东的商品发布体系与淘宝类似，但字段命名和类目体系不同。

表格

核心接口	功能
`jingdong.ware.write.add`	添加新商品
`jingdong.ware.write.update`	修改商品信息
`jingdong.ware.write.upOrDown`	商品上下架
`jingdong.image.write.upload`	图片上传
`jingdong.category.read.findAttrsByCategoryId`	获取类目属性

接入流程：

京东开放平台注册企业开发者
创建应用，申请 "商品管理" 权限
商家授权后获取 token

关键差异：

京东对售后服务模板有强制要求，发布时必须关联售后模板ID
京东自营和 POP（第三方）店铺的接口权限不同，需确认店铺类型

3. 拼多多（PDD Open API）

拼多多接口相对简单，免费额度较高，适合中小商家。

表格

核心接口	功能
`pdd.goods.add`	创建新商品
`pdd.goods.update`	修改商品
`pdd.goods.sku.price.update`	批量修改SKU价格
`pdd.stock.update`	批量修改库存
`pdd.goods.image.upload`	图片上传

关键注意点：

拼多多对商品标题有敏感词过滤，发布前建议做合规校验
多规格商品（SKU）需通过 sku_list 数组传入，每个 SKU 包含规格名、价格、库存、外部编码
支持通过 outer_id（商家自定义编码）做系统间关联，方便后续库存同步

4. 1688（阿里巴巴开放平台）

B2B 批发场景，接口侧重起订量、批发价、货期等字段。

表格

核心接口	功能
`alibaba.product.add`	发布产品
`alibaba.product.edit`	编辑产品
`alibaba.product.repost`	重发产品（刷新排名）
`alibaba.category.get`	获取类目

关键差异：

必须填写最小起订量（MOQ）和批发价格区间
支持产品分组和运费模板关联

5. 抖音电商（抖店开放平台）

表格

核心接口	功能
`product.add`	创建商品
`product.edit`	编辑商品
`product.list`	获取商品列表
`product.spec.add`	添加规格

关键注意点：

抖音对商品类目和资质审核严格，部分类目需要品牌授权或行业资质
视频/直播带货场景下，商品需额外绑定到直播间或短视频

二、跨境平台批量上货接口

1. Shopee / Lazada

表格

接口	功能
`/api/v2/product/add_item`	创建商品
`/api/v2/product/update_item`	更新商品
`/api/v2/product/get_item_base_info`	获取商品信息

注意：

多站点（SG/MY/ID/TH/PH/VN）需分别授权，每个站点独立调用
2026年平台算法升级，API 批量上传的商品在 SEO 初始权重上可能低于"模拟人工录入"的商品，部分卖家会结合 RPA 方案规避

2. TikTok Shop

表格

接口	功能
`product.create`	创建商品
`product.update`	更新商品
`product.list`	获取商品列表

注意：

跨境店和本土店的字段要求不同，需区分报关参数和本地库存
需校验目标市场合规政策（如欧盟 CE 认证标注）

3. 亚马逊（SP-API）

表格

接口	功能
`POST /listings/2021-08-01/items`	创建/更新 Listing
`GET /catalog/2022-04-01/items`	获取商品目录信息
`POST /productTypeDefinitions`	获取商品类型定义（用于校验属性）

注意：

需 AWS 账号 + 卖家店铺授权，签名使用 AWS SigV4，接入复杂度最高
多语言/多站点（US/EU/JP）需分别处理货币单位和语言

三、批量上货的技术实现方案

方案一：官方 API 直连（推荐）点击测试

适合有开发能力、追求稳定合规的团队。

plain

你的上货软件 → 调用各平台官方 API → 批量发布商品

核心代码框架（Python 示例）：

Python

import requestsimport hashlibimport timeimport hmacclass TaobaoUploader:
    def __init__(self, app_key, app_secret, access_token):
        self.app_key = app_key
        self.app_secret = app_secret
        self.access_token = access_token
        self.gateway = "https://gw.api.taobao.com/router/rest"
    
    def _sign(self, params):
        # 参数排序 + HMAC-SHA256 签名
        sorted_params = sorted(params.items())
        query = ''.join(f"{k}{v}" for k, v in sorted_params)
        sign = hmac.new(
            self.app_secret.encode(),
            query.encode(),
            hashlib.sha256        ).hexdigest().upper()
        return sign    
    def add_item(self, item_data):
        """发布单个商品"""
        params = {
            "method": "taobao.item.add",
            "app_key": self.app_key,
            "timestamp": time.strftime("%Y-%m-%d %H:%M:%S"),
            "format": "json",
            "v": "2.0",
            "sign_method": "hmac-sha256",
            "session": self.access_token,
            # 商品必填字段
            "num_iid": "",  # 新发布留空
            "title": item_data["title"],
            "price": item_data["price"],
            "num": item_data["stock"],  # 库存
            "cid": item_data["category_id"],  # 类目ID
            "desc": item_data["description"],
            "props": item_data["properties"],  # 商品属性
            "sku_properties": item_data.get("sku_props", ""),
            "sku_quantities": item_data.get("sku_quantities", ""),
            "sku_prices": item_data.get("sku_prices", ""),
            "sku_outer_ids": item_data.get("sku_outer_ids", ""),
        }
        params["sign"] = self._sign(params)
        
        resp = requests.post(self.gateway, data=params, timeout=30)
        return resp.json()
    
    def batch_upload(self, items):
        """批量上货：循环调用，控制频率"""
        results = []
        for idx, item in enumerate(items):
            try:
                result = self.add_item(item)
                results.append({"index": idx, "status": "success", "data": result})
                # 控制频率：每10秒发一个，避免限流
                if idx < len(items) - 1:
                    time.sleep(10)
            except Exception as e:
                results.append({"index": idx, "status": "failed", "error": str(e)})
        return results

方案二：第三方 ERP / SaaS 对接

如果你不想自己对接每个平台，可以接入成熟的电商 ERP，它们已经封装好了多平台 API：

表格

工具	支持平台	特点
店小秘	淘宝、拼多多、Shopee、Lazada、TikTok 等	全球用户超150万，跨平台采集+一键转化格式
芒果店长	多平台	轻量化，适合中小卖家
聚水潭	国内主流平台	深度对接库存、订单、供应链
万里牛	跨境+国内	跨境场景成熟

这些工具通过官方 API 实现数据传输，你只需按它们的文档对接一次，就能实现多平台批量上货。

方案三：RPA 模拟人工（特殊场景）

部分平台（如 Shopee、敦煌网）对 API 批量行为有风控，API 上货可能导致新品权重降低。此时可用 RPA（机器人流程自动化） 模拟真人操作：

通过视觉识别和行为模拟，在卖家后台页面自动填写商品信息
配合指纹浏览器，随机化鼠标轨迹和输入速度
被平台算法识别为"真实活跃用户"，SEO 初始权重更高

四、批量上货的关键注意事项

表格

问题	解决方案
API 权限申请被拒	企业资质 + 详细使用场景说明，强调"提升商家运营效率"而非"数据采集"
类目匹配错误	先用类目查询接口获取完整类目树，建立自有类目映射表
图片上传失败	压缩至 2MB 以内，格式 JPG/PNG，主图统一 800×800px，先上传获取平台图片 URL
SKU 规格混乱	建立标准化的 SKU 编码体系（如 `颜色_尺码_款式`），通过 `outer_id` 关联
频繁调用被限流	令牌桶算法控制并发，单应用淘宝约 500次/秒，京东 ≤200ms/次，超限用指数退避重试
跨平台字段不兼容	设计"标准商品数据模型"，通过字段映射表转换到各平台格式
商品被判定重复铺货	标题做语义改写，主图差异化处理，避免完全相同的商品信息
库存超卖	发布时预留安全库存，通过库存同步 API 实时扣减

五、推荐接入路径

如果你现在已有上货软件，建议按以下步骤推进：

确定目标平台：先选 1-2 个核心平台（如淘宝+拼多多）跑通，再扩展
申请开发者账号：以企业身份注册各平台开放平台，申请"商品发布"权限
构建标准数据模型：统一你软件内部的商品字段，再映射到各平台
图片/资源预处理：搭建图片服务器或接入 OSS，实现图片自动压缩、上传、获取 URL
灰度测试：先用 10-20 个商品测试发布流程，验证字段映射和类目匹配
批量调度：用 APScheduler 或 Celery 实现定时/批量任务，加入失败重试和告警
监控与优化：跟踪发布成功率、审核通过率、商品曝光数据，持续优化

小白必看！用AI+现成工具实现1688选品监控自动化

Wed, 03 Jun 2026 14:29:19 +0800

一、先泼一盆冷水：为什么普通人玩不了官方API？

很多教程一上来就教你注册1688开放平台、申请API、写代码调用。但真相是：

表格

门槛	实际情况	你的处境
企业资质	1688核心API要求营业执照+对公账户认证，个人开发者仅能获得基础字段，QPS极低	❌ 个人/小卖家直接被挡在门外
技术门槛	需要理解签名算法（HMAC-MD5/SHA1）、OAuth2.0授权、JSON解析	❌ 零基础完全看不懂
合规风险	自己写爬虫容易被封IP、账号受限，甚至面临法律风险	❌ 得不偿失
维护成本	1688页面结构经常改版，代码需要反复修改	❌ 没有技术团队根本维护不了

基础准入条件已明确要求企业主体资质，个人开发者无法申请核心API权限。必备资质包括：加盖公章的营业执照扫描件、法人身份证、企业对公账户信息。

所以，本文不讲官方API，只讲小白能立刻上手的替代方案。

二、方案总览：三条零技术路线，总有一条适合你

表格

方案	适合人群	成本	难度	效果
方案A：1688官方AI工具	所有1688用户	免费/低价	⭐	基础选品+趋势洞察
方案B：第三方AI选品插件	跨境电商卖家	免费试用+付费	⭐⭐	跨平台数据+利润测算
方案C：AI助手+现成接口服务	想深度自动化的用户	按次计费（约10元/次）	⭐⭐⭐	全自动监控+批量选品

三、方案A：1688官方AI工具（完全免费，开箱即用）

3.1 1688 AI版App — 你的随身选品助手

2025年底，1688推出了AI版App，这是官方给普通用户的最强福利。

它能做什么？

📸 图片智能搜索：看到爆款图？直接上传，AI秒找1688同款/相似款
🤖 专属Agent配置：设定你的选品偏好（价格区间、类目、起订量），AI每天自动推送匹配商品
📊 全网趋势洞察：自动抓取小红书、抖音热词，匹配1688近7天动销靠前的商品
🏷️ 智能标签筛选：自动打上"高复购""低退货率""支持密文面单"等实战标签

怎么用？（3步搞定）

下载安装：在应用商店搜索"1688"，下载最新版App
配置Agent：登录后点击右上角设置图标 → 选择"商品偏好""商家偏好""服务偏好" → 保存
开始使用：

首页每天自动推送"为你精选"的选品资讯
点击搜索框旁的相机图标，上传图片即可图搜
在"一件代发选品频道"可按条件筛选

1688在珠宝饰品类目正式推出AI智能选品助手，能自动抓取小红书、抖音的实时热词，在1688站内匹配近7天动销靠前的商品，还会打上"高复购""低退货率"等实战组合标签。

3.2 1688诚信通AI版 — 商家端智能体（适合有店铺的）

如果你是1688商家，2026年1月起可以使用诚信通AI版。数据显示：用了AI的商家比没用的，交易额涨了73%、复购率涨了31%。

核心功能：

10秒做选品开款
AI根据评价自动改款建议
智能客服自动回复询盘
成本核算与报价优化

四、方案B：第三方AI选品插件（跨境电商必备）点击测试

如果你做跨境（亚马逊、TikTok、Shopee等），需要将1688货源与海外平台数据打通，推荐以下工具：

4.1 店雷达 — 1688站内选品神器

定位：专注1688选品、跨境选品、同行分析

核心功能：

🔍 全网图搜找同款
📈 同行单品/店铺透视
🌍 跨境选品导航
💰 多平台比价

怎么用？

访问店雷达官网注册账号
安装浏览器插件（Chrome/Edge）
打开1688商品页，插件自动显示：

该商品在亚马逊的同款售价
预估利润空间
供应商信用评级
历史价格走势

店雷达是一款专为电商人打造的选品与供应链管理工具，核心聚焦1688选品、跨境选品、同行分析三大场景，无需复杂操作，就能解决选品难、找厂难、盯竞品难的问题。

4.2 Sorftime — 跨境全链路AI选品（进阶推荐）

定位：跨境电商AI数据供应链基础设施

为什么推荐？

2025年12月行业首发1688插件，打通1688货源与亚马逊销售数据
支持"一句话选品"：告诉AI你的需求，10分钟输出完整报告
单次选品成本约10元，3天工作量压缩到10分钟

核心功能一览：

表格

功能	说明	你的收益
🔥 1688同款匹配	一键查找亚马逊爆款在1688的同款	不用手动翻页找货源
💰 跨平台利润测算	输入1688拿货价，自动算FBA成本、佣金、利润	避免"看起来赚钱实际亏钱"
🤖 MCP工作流	AI自动调用23+工具，全自动执行	彻底解放双手
📊 竞品监控	实时监控竞品销量、评论、排名	及时调整策略
🔥 隐赚指数	识别"上架<6个月冲入Top20"的潜力款	抢占蓝海先机

效率对比：

表格

维度	传统方式	Sorftime MCP工作流
启动方式	手动搜索+人工筛选	一句话指令，AI自动执行
数据获取	复制粘贴，耗时3天	AI自动调用，10分钟
1688货源匹配	人工查找，漏检率高	算法精准匹配，准确率提升3倍
利润测算	Excel手动计算	自动生成含FBA成本的利润报告

Sorftime支持AI Agent驾驶舱，"一句话选品"降低了工具学习门槛，同时覆盖40+平台便于未来业务扩展。60万+付费用户规模也验证了其产品成熟度。

实操示例：

plain

你输入："帮我找1688上适合做亚马逊宠物用品的货源，要求：
1. 支持一件代发
2. 1688价格15元以内
3. 亚马逊售价能到15美元以上
4. 供应商诚信通3年以上"

AI输出：
✅ 匹配到23个商品
✅ 自动过滤掉专利风险款
✅ 利润测算：最高毛利率68%
✅ 推荐Top3供应商及联系方式

五、方案C：AI助手+现成接口服务（深度自动化）

如果你已经有一定的业务规模，想要7×24小时自动监控价格、库存、竞品动态，但又不想雇程序员，可以使用第三方封装好的API服务。

5.1 什么是"现成接口服务"？

简单说，就是有技术团队已经帮你完成了：

✅ 企业资质认证
✅ 官方API权限申请
✅ 签名算法封装
✅ 反爬处理

你只需要像"点外卖"一样，调用他们封装好的接口，传入参数就能拿到数据。

5.2 典型使用场景

场景1：爆款图秒找货源

plain

你看到抖音上某款"折叠收纳箱"很火
→ 复制商品图片链接
→ 调用图搜接口
→ 3秒内返回1688上的同款/相似款，按相似度排序
→ 自动筛选出销量>100、支持一件代发的优质货源

场景2：价格自动监控

plain

你锁定了5款潜力商品
→ 设置每30分钟自动查询一次价格
→ 价格下降时，微信/钉钉自动推送提醒
→ 库存不足时，自动预警补货

场景3：批量选品导出

plain

输入10个关键词
→ 自动翻页抓取每个关键词前5页商品
→ 按销量、价格、供应商等级筛选
→ 一键导出Excel，直接用于店群铺货

5.3 完全零代码的实现方式

方式一：使用AI助手（如ChatGPT/Claude/通义千问）

把以下提示词发给AI，让它帮你生成可运行的代码：

plain

请帮我写一段Python代码，实现以下功能：
1. 调用1688商品搜索API（接口地址：https://api-gw.xxx.cn/1688/item_search）
2. 搜索关键词"无线蓝牙耳机"
3. 返回前20个商品的标题、价格、销量、链接
4. 将结果保存为Excel文件

参数说明：
- key: 我的API Key（我会替换）
- secret: 我的API Secret（我会替换）
- q: 搜索关键词
- page: 页码
- page_size: 每页数量

要求：代码要完整可运行，有错误处理，有注释说明。

AI会生成完整代码，你只需要：

复制代码到本地
替换API Key和Secret
双击运行

方式二：使用可视化工具（如Postman/Apifox）

如果你连Python都不想装，可以用API调试工具：

打开Postman（免费软件）
新建一个GET请求
输入接口地址
在Params里填入key、secret、keyword等参数
点击Send，直接看到返回的JSON数据
用在线JSON转Excel工具导出

5.4 避坑指南

不要用爬虫脚本替代API：爬虫容易被风控，API调用合规稳定，长期可用。合理设置监控间隔：避免频繁调用，建议10-30分钟一次，既保证数据实时，又避免触发接口限制。

表格

❌ 错误做法	✅ 正确做法
自己写爬虫扒1688	使用官方API或第三方封装接口
每秒调用10次接口	每10-30分钟调用一次
用免费代理IP	使用正规接口服务商
存储用户隐私数据	只采集公开商品信息

六、三种方案怎么选？一张图看懂

plain

┌─────────────────────────────────────────────────────────────┐
│                    你的情况是什么？                           │
└─────────────────────────────────────────────────────────────┘
                              │
            ┌─────────────────┼─────────────────┐
            ▼                 ▼                 ▼
      【纯1688采购】    【跨境电商卖家】    【想深度自动化】
            │                 │                 │
            ▼                 ▼                 ▼
      用1688 AI版App    用Sorftime/店雷达   用第三方API服务
      + 诚信通AI版       + 1688官方AI工具    + AI助手生成代码
            │                 │                 │
      成本：免费~低价    成本：10元/次起      成本：按调用量计费
      难度：⭐           难度：⭐⭐          难度：⭐⭐⭐
      效果：基础选品      效果：全链路决策    效果：7×24监控

七、立即行动：今天就能开始的3个步骤

Step 1：下载1688 AI版App（5分钟）

应用商店搜索"1688"
注册/登录账号
配置你的选品偏好Agent
上传一张爆款图试试图搜功能

Step 2：安装一个选品插件（10分钟）

跨境卖家：访问Sorftime或店雷达官网
注册账号，安装浏览器插件
打开1688任意商品页，体验数据透视

Step 3：设置你的第一个监控（15分钟）

选定3-5款潜力商品
记录商品链接和当前价格
设置每天检查一次的提醒（手机日历/闹钟）
一周后对比价格变化，判断供应商稳定性

八、常见问题FAQ

Q1：我是完全的小白，真的不用学代码吗？

真的不用。本文推荐的方案A和方案B都是纯图形界面操作，像刷抖音一样简单。方案C需要复制粘贴几行代码，但AI已经帮你写好了。

Q2：这些工具收费吗？

1688官方AI工具基础功能免费；第三方插件通常有免费试用版，高级功能付费；API服务按调用量计费，单次约0.1-10元不等。

Q3：个人没有营业执照怎么办？

这正是本文的核心价值——绕过企业资质门槛。使用官方AI工具不需要任何资质；使用第三方插件只需要注册账号；使用API服务由服务商统一处理资质。

Q4：数据准确吗？

官方工具数据最准（实时）；第三方工具通常每日更新，延迟在24小时内；API服务取决于服务商的更新频率。建议重要决策前人工复核一次。

Q5：会被封号吗？

使用官方工具和正规第三方服务不会被封号。但自己写爬虫、用破解插件、高频刷接口都有风险。

九、写在最后

2026年的1688选品，已经不是"谁能找到货"的比拼，而是"谁能最快判断货的价值"的效率战争。

你不需要成为程序员，不需要注册公司，甚至不需要花大钱——善用AI工具，一个普通人就能拥有过去只有大企业才具备的数据分析能力。

1688图搜+关键词搜索商品API：从入门到实战，让你效率翻倍

Tue, 02 Jun 2026 16:45:14 +0800

一、概述

1688作为国内最大的B2B批发采购平台，拥有海量商品数据。对于电商开发者、选品工具和供应链系统而言，图片搜索（图搜/拍立淘）和关键词搜索是两种最高效的找货方式。本文将带你从0到1掌握1688图搜API和关键词搜索API的完整调用流程，附赠实战代码，让你的开发效率翻倍。

二、准备工作：账号与权限

2.1 注册1688开放平台账号

1688开放平台主要面向企业开发者，个人开发者权限受限。

访问 1688平台
使用企业支付宝账号注册并完成企业实名认证
提交营业执照、法人身份证等资质

2.2 创建应用获取凭证

在控制台创建应用后，你将获得：

App Key：应用唯一标识
App Secret：签名密钥（严格保密）

2.3 申请API权限

根据业务需求申请对应接口权限：

表格

接口类型	接口名称	功能描述
图搜API	`alibaba.image.search`	上传图片搜索相似商品
关键词搜索	`com.alibaba.product.search` / `alibaba.wholesale.goods.search`	按关键词搜索商品列表
商品详情	`alibaba.product.get`	获取商品详细信息
店铺商品	`alibaba.product.getSellerProductList`	获取指定店铺全部商品

⚠️ 注意：部分接口需要通过OAuth2.0授权获取 access_token，特别是涉及店铺数据的操作。

三、签名生成机制

1688开放平台采用 HMAC-SHA1 或 MD5 签名算法，所有请求参数必须按规则生成签名。

3.1 签名规则

排序：将所有参数（除 sign 外）按参数名 ASCII 码升序排列
拼接：格式为 appSecret + key1value1 + key2value2 + ... + appSecret
加密：使用 MD5 加密，结果转大写

3.2 Python签名工具类

Python

import hashlibimport hmacimport timeimport urllib.parsefrom datetime import datetimeclass AlibabaApiSigner:
    """1688开放平台API签名工具类"""
    
    def __init__(self, app_key: str, app_secret: str):
        self.app_key = app_key
        self.app_secret = app_secret
        self.base_url = "https://api.1688.com/router/rest"
    
    def generate_md5_sign(self, params: dict) -> str:
        """
        生成MD5签名（适用于大多数1688接口）
        """
        # 过滤空值和sign参数
        filtered = {k: v for k, v in params.items() 
                   if v is not None and k != 'sign'}
        
        # 按key升序排列
        sorted_items = sorted(filtered.items(), key=lambda x: x[0])
        
        # 拼接字符串
        sign_content = self.app_secret        for key, value in sorted_items:
            sign_content += f"{key}{value}"
        sign_content += self.app_secret        
        # MD5加密并转大写
        return hashlib.md5(sign_content.encode('utf-8')).hexdigest().upper()
    
    def generate_hmac_sign(self, params: dict) -> str:
        """
        生成HMAC-SHA1签名（部分接口使用）
        """
        filtered = {k: v for k, v in params.items() 
                   if v is not None and k != 'sign'}
        sorted_items = sorted(filtered.items(), key=lambda x: x[0])
        
        sign_content = self.app_secret        for key, value in sorted_items:
            sign_content += f"{key}{value}"
        sign_content += self.app_secret        
        # HMAC-SHA1加密
        sign = hmac.new(
            self.app_secret.encode('utf-8'),
            sign_content.encode('utf-8'),
            hashlib.sha1        ).hexdigest().upper()
        return sign    
    def build_base_params(self) -> dict:
        """构建基础请求参数"""
        return {
            'app_key': self.app_key,
            'timestamp': datetime.now().strftime('%Y-%m-%d %H:%M:%S'),
            'format': 'json',
            'v': '2.0',
            'sign_method': 'md5'
        }

四、关键词搜索商品API实战

4.1 接口概述

关键词搜索是1688最核心的找货方式，支持按关键词、价格区间、类目、排序等多维度筛选。

接口地址：https://api.1688.com/router/rest
请求方式：POST
数据格式：application/x-www-form-urlencoded

4.2 核心参数说明

表格

参数名	类型	必选	说明
`method`	String	是	接口方法名，如 `com.alibaba.product.search`
`app_key`	String	是	应用Key
`sign`	String	是	签名
`timestamp`	String	是	时间戳
`q` / `keyword`	String	是	搜索关键词
`page` / `page_no`	Number	否	页码，默认1
`pageSize` / `page_size`	Number	否	每页数量，默认20
`priceStart`	Number	否	价格区间起始
`priceEnd`	Number	否	价格区间结束
`categoryId`	Number	否	类目ID过滤
`sort`	String	否	排序方式：`price_asc`/`price_desc`/`sale`

4.3 完整调用代码

Python

import requestsimport jsonimport timefrom typing import List, Dict, Optionalclass AlibabaKeywordSearch:
    """1688关键词搜索商品API封装"""
    
    def __init__(self, app_key: str, app_secret: str):
        self.signer = AlibabaApiSigner(app_key, app_secret)
        self.api_url = "https://api.1688.com/router/rest"
    
    def search_products(
        self,
        keyword: str,
        page: int = 1,
        page_size: int = 20,
        price_start: Optional[float] = None,
        price_end: Optional[float] = None,
        category_id: Optional[int] = None,
        sort: str = "default"
    ) -> Dict:
        """
        关键词搜索商品
        
        :param keyword: 搜索关键词，如"无线蓝牙耳机"
        :param page: 页码
        :param page_size: 每页数量（建议不超过50）
        :param price_start: 最低价格
        :param price_end: 最高价格
        :param category_id: 类目ID过滤
        :param sort: 排序方式
        :return: 搜索结果字典
        """
        # 构建基础参数
        params = self.signer.build_base_params()
        params['method'] = 'com.alibaba.product.search'  # 或 alibaba.wholesale.goods.search
        
        # 业务参数
        params['q'] = keyword
        params['page'] = page
        params['pageSize'] = page_size        
        if price_start is not None:
            params['priceStart'] = price_start        if price_end is not None:
            params['priceEnd'] = price_end        if category_id is not None:
            params['categoryId'] = category_id        if sort != "default":
            params['sort'] = sort        
        # 生成签名
        params['sign'] = self.signer.generate_md5_sign(params)
        
        # 发送请求
        try:
            response = requests.post(
                self.api_url,
                data=params,
                headers={'Content-Type': 'application/x-www-form-urlencoded'},
                timeout=30
            )
            response.raise_for_status()
            return response.json()
        except requests.RequestException as e:
            print(f"请求异常: {e}")
            return {"error": str(e)}
    
    def parse_search_results(self, response: Dict) -> List[Dict]:
        """
        解析搜索结果，提取关键字段
        """
        results = []
        
        # 根据实际返回结构解析（不同接口返回结构可能不同）
        if 'result' in response:
            products = response['result'].get('result', [])
        elif 'productSearchResponse' in response:
            products = response['productSearchResponse'].get('products', [])
        else:
            # 适配不同返回结构
            products = response.get('products', [])
        
        for product in products:
            item = {
                'product_id': product.get('productId') or product.get('offerId'),
                'title': product.get('subject') or product.get('title'),
                'price': product.get('price') or product.get('salePrice'),
                'image_url': product.get('imageUrl') or product.get('mainImage'),
                'supplier_name': product.get('companyName') or product.get('supplier'),
                'min_order': product.get('minOrderQuantity', 1),
                'sale_count': product.get('saleCount', 0),
                'detail_url': product.get('detailUrl', ''),
                'location': product.get('location', '')
            }
            results.append(item)
        
        return results    
    def batch_search(
        self,
        keywords: List[str],
        pages: int = 3,
        delay: float = 1.5
    ) -> Dict[str, List[Dict]]:
        """
        批量关键词搜索（带限流控制）
        
        :param keywords: 关键词列表
        :param pages: 每个关键词搜索页数
        :param delay: 请求间隔（秒）
        :return: {keyword: [products]}
        """
        all_results = {}
        
        for keyword in keywords:
            print(f"🔍 正在搜索: {keyword}")
            keyword_results = []
            
            for page in range(1, pages + 1):
                response = self.search_products(
                    keyword=keyword,
                    page=page,
                    page_size=50
                )
                
                products = self.parse_search_results(response)
                if not products:
                    break
                    
                keyword_results.extend(products)
                print(f"  第{page}页获取 {len(products)} 条数据")
                
                # 限流：避免触发QPS限制
                if page < pages:
                    time.sleep(delay)
            
            all_results[keyword] = keyword_results            print(f"✅ {keyword} 共获取 {len(keyword_results)} 条数据\n")
        
        return all_results# ========== 实战演示 ==========if __name__ == '__main__':
    APP_KEY = 'your_app_key'
    APP_SECRET = 'your_app_secret'
    
    searcher = AlibabaKeywordSearch(APP_KEY, APP_SECRET)
    
    # 单关键词搜索
    print("=" * 60)
    print("【单关键词搜索示例】")
    print("=" * 60)
    
    result = searcher.search_products(
        keyword="无线蓝牙耳机",
        page=1,
        page_size=10,
        price_start=10,
        price_end=100,
        sort="price_asc"
    )
    
    products = searcher.parse_search_results(result)
    for i, p in enumerate(products[:5], 1):
        print(f"{i}. {p['title'][:30]}... | ¥{p['price']} | 销量:{p['sale_count']}")
    
    # 批量关键词搜索
    print("\n" + "=" * 60)
    print("【批量关键词搜索示例】")
    print("=" * 60)
    
    keywords = ["T恤批发", "手机壳", "数据线", "保温杯"]
    batch_results = searcher.batch_search(keywords, pages=2, delay=2)
    
    for kw, items in batch_results.items():
        print(f"📦 {kw}: {len(items)} 条商品")

五、图搜（以图搜货/拍立淘）API实战点击测试

5.1 接口概述

1688图搜API（拍立淘/以图搜货）是官方图像搜品接口，支持通过图片URL或Base64编码输入图片，秒级返回同款/相似商品的结构化数据。

核心能力：

上传商品图片 → 返回同款/相似商品
返回商品ID、标题、价格、供应商等结构化数据
支持图片URL或Base64两种输入方式

5.2 核心参数说明

表格

参数名	类型	必选	说明
`method`	String	是	接口方法名，如 `alibaba.image.search`
`image_url`	String	条件	图片网络URL（与image_content二选一）
`image_content`	String	条件	图片Base64编码（与image_url二选一）
`category_id`	Number	否	指定搜索类目，提高精准度
`keyword`	String	否	配合图片的关键词辅助搜索

5.3 完整调用代码

Python

import base64import requestsfrom typing import Optionalclass AlibabaImageSearch:
    """1688图搜（拍立淘）API封装"""
    
    def __init__(self, app_key: str, app_secret: str):
        self.signer = AlibabaApiSigner(app_key, app_secret)
        self.api_url = "https://api.1688.com/router/rest"
    
    def search_by_url(
        self,
        image_url: str,
        category_id: Optional[int] = None,
        keyword: Optional[str] = None
    ) -> dict:
        """
        通过图片URL搜索商品
        
        :param image_url: 图片的网络地址
        :param category_id: 指定类目ID（可选）
        :param keyword: 辅助关键词（可选）
        :return: 搜索结果
        """
        params = self.signer.build_base_params()
        params['method'] = 'alibaba.image.search'
        params['image_url'] = image_url        
        if category_id:
            params['category_id'] = category_id        if keyword:
            params['keyword'] = keyword
        
        params['sign'] = self.signer.generate_md5_sign(params)
        
        try:
            response = requests.post(
                self.api_url,
                data=params,
                timeout=30
            )
            return response.json()
        except Exception as e:
            return {"error": str(e)}
    
    def search_by_file(
        self,
        image_path: str,
        category_id: Optional[int] = None,
        keyword: Optional[str] = None
    ) -> dict:
        """
        通过本地图片文件搜索商品（Base64编码上传）
        
        :param image_path: 本地图片路径
        :param category_id: 指定类目ID
        :param keyword: 辅助关键词
        :return: 搜索结果
        """
        # 读取图片并Base64编码
        try:
            with open(image_path, 'rb') as f:
                image_bytes = f.read()
                image_base64 = base64.b64encode(image_bytes).decode('utf-8')
        except Exception as e:
            return {"error": f"图片读取失败: {e}"}
        
        params = self.signer.build_base_params()
        params['method'] = 'alibaba.image.search'
        params['image_content'] = image_base64        
        if category_id:
            params['category_id'] = category_id        if keyword:
            params['keyword'] = keyword
        
        params['sign'] = self.signer.generate_md5_sign(params)
        
        try:
            response = requests.post(
                self.api_url,
                data=params,
                timeout=60  # 图搜可能需要更长时间
            )
            return response.json()
        except Exception as e:
            return {"error": str(e)}
    
    def parse_image_results(self, response: dict) -> list:
        """
        解析图搜结果
        """
        results = []
        
        # 适配不同返回结构
        if 'imageSearchResponse' in response:
            items = response['imageSearchResponse'].get('items', [])
        elif 'result' in response:
            items = response['result'].get('items', [])
        else:
            items = response.get('items', [])
        
        for item in items:
            result = {
                'product_id': item.get('productId') or item.get('offerId'),
                'title': item.get('subject') or item.get('title'),
                'price': item.get('price'),
                'image_url': item.get('imageUrl') or item.get('mainImage'),
                'similarity': item.get('similarity', 0),  # 相似度分数
                'supplier': item.get('companyName') or item.get('supplier'),
                'detail_url': item.get('detailUrl', ''),
                'min_order': item.get('minOrderQuantity', 1)
            }
            results.append(result)
        
        return results# ========== 图搜实战演示 ==========if __name__ == '__main__':
    APP_KEY = 'your_app_key'
    APP_SECRET = 'your_app_secret'
    
    image_searcher = AlibabaImageSearch(APP_KEY, APP_SECRET)
    
    print("=" * 60)
    print("【图搜实战：通过图片URL搜索】")
    print("=" * 60)
    
    # 方式1：通过图片URL搜索
    image_url = "https://example.com/your-product-image.jpg"
    result = image_searcher.search_by_url(
        image_url=image_url,
        category_id=1512,  # 可选：限定在手机数码类目
        keyword="蓝牙耳机"
    )
    
    items = image_searcher.parse_image_results(result)
    print(f"找到 {len(items)} 个相似商品：")
    for i, item in enumerate(items[:5], 1):
        print(f"{i}. {item['title'][:35]}... | ¥{item['price']} | 相似度:{item['similarity']}%")
    
    # 方式2：通过本地文件搜索
    print("\n" + "=" * 60)
    print("【图搜实战：通过本地图片文件搜索】")
    print("=" * 60)
    
    # result = image_searcher.search_by_file(
    #     image_path="/path/to/your/image.jpg",
    #     keyword="手机壳"
    # )
    # items = image_searcher.parse_image_results(result)

六、高级实战：关键词+图搜组合策略

6.1 组合搜索场景

在实际业务中，单一搜索方式往往不够精准。以下是几种高效的组合策略：

表格

场景	策略	效果
找同款货源	先关键词搜索 → 用图搜验证	精准定位源头工厂
竞品分析	图搜找到竞品 → 关键词搜索同类	全面了解市场行情
价格监控	图搜锁定商品 → 定期关键词比价	实时跟踪价格波动
选品拓展	关键词发现趋势 → 图搜找相似款	快速扩展产品线

6.2 组合搜索实战代码

Python

class AlibabaSearchCombo:
    """1688组合搜索策略：关键词+图搜联动"""
    
    def __init__(self, app_key: str, app_secret: str):
        self.keyword_searcher = AlibabaKeywordSearch(app_key, app_secret)
        self.image_searcher = AlibabaImageSearch(app_key, app_secret)
    
    def find_source_factory(self, product_title: str) -> list:
        """
        找同款货源：先关键词搜索，再用图搜验证
        
        :param product_title: 商品标题
        :return: 源头工厂列表
        """
        print(f"🔍 步骤1：关键词搜索 '{product_title}'")
        
        # 第一步：关键词搜索获取候选商品
        kw_result = self.keyword_searcher.search_products(
            keyword=product_title,
            page=1,
            page_size=10,
            sort="sale"  # 按销量排序
        )
        candidates = self.keyword_searcher.parse_search_results(kw_result)
        
        if not candidates:
            return []
        
        # 取第一个商品的图片进行图搜
        first_product = candidates[0]
        image_url = first_product.get('image_url')
        
        if not image_url:
            return candidates        
        print(f"🖼️ 步骤2：用图搜验证 '{first_product['title'][:20]}...'")
        
        # 第二步：图搜找同款
        img_result = self.image_searcher.search_by_url(
            image_url=image_url,
            keyword=product_title.split()[0]  # 取第一个词作为辅助关键词
        )
        similar_items = self.image_searcher.parse_image_results(img_result)
        
        # 合并结果，去重
        all_items = candidates + similar_items
        seen_ids = set()
        unique_items = []
        
        for item in all_items:
            pid = item.get('product_id')
            if pid and pid not in seen_ids:
                seen_ids.add(pid)
                unique_items.append(item)
        
        # 按价格升序排列（找最便宜的源头）
        unique_items.sort(key=lambda x: float(x.get('price', 0) or 0))
        
        return unique_items    
    def price_comparison(self, image_url: str, keyword: str) -> dict:
        """
        价格监控：图搜锁定商品 + 关键词搜索比价
        
        :param image_url: 商品图片URL
        :param keyword: 商品关键词
        :return: 比价结果
        """
        # 图搜获取同款
        img_result = self.image_searcher.search_by_url(image_url, keyword=keyword)
        image_items = self.image_searcher.parse_image_results(img_result)
        
        # 关键词搜索获取同类
        kw_result = self.keyword_searcher.search_products(
            keyword=keyword,
            page=1,
            page_size=50
        )
        keyword_items = self.keyword_searcher.parse_search_results(kw_result)
        
        # 价格分析
        all_prices = []
        for item in image_items + keyword_items:
            price = item.get('price')
            if price and isinstance(price, (int, float)):
                all_prices.append(float(price))
        
        if not all_prices:
            return {"error": "未获取到价格数据"}
        
        return {
            "image_search_count": len(image_items),
            "keyword_search_count": len(keyword_items),
            "min_price": min(all_prices),
            "max_price": max(all_prices),
            "avg_price": round(sum(all_prices) / len(all_prices), 2),
            "median_price": round(sorted(all_prices)[len(all_prices)//2], 2),
            "cheapest_item": min(image_items + keyword_items, 
                                key=lambda x: float(x.get('price', float('inf')) or float('inf')))
        }# ========== 组合搜索实战 ==========if __name__ == '__main__':
    APP_KEY = 'your_app_key'
    APP_SECRET = 'your_app_secret'
    
    combo = AlibabaSearchCombo(APP_KEY, APP_SECRET)
    
    # 实战1：找同款货源
    print("=" * 60)
    print("【实战1：找同款货源】")
    print("=" * 60)
    
    sources = combo.find_source_factory("无线蓝牙耳机 降噪")
    print(f"找到 {len(sources)} 个源头：")
    for i, s in enumerate(sources[:5], 1):
        print(f"{i}. {s['title'][:30]}... | ¥{s['price']} | {s['supplier']}")
    
    # 实战2：价格监控
    print("\n" + "=" * 60)
    print("【实战2：价格监控】")
    print("=" * 60)
    
    # price_analysis = combo.price_comparison(
    #     image_url="https://example.com/product.jpg",
    #     keyword="蓝牙耳机"
    # )
    # print(f"最低价格: ¥{price_analysis['min_price']}")
    # print(f"平均价格: ¥{price_analysis['avg_price']}")
    # print(f"最便宜的商品: {price_analysis['cheapest_item']['title']}")

七、性能优化与效率翻倍技巧

7.1 分页与批量处理

Python

class BatchProcessor:
    """批量处理优化"""
    
    def __init__(self, searcher: AlibabaKeywordSearch):
        self.searcher = searcher    
    def fetch_all_pages(
        self,
        keyword: str,
        max_pages: int = 10,
        page_size: int = 50,
        delay: float = 1.2
    ) -> list:
        """
        获取关键词的所有分页数据
        
        :param max_pages: 最大页数
        :param delay: 请求间隔（遵守QPS限制）
        """
        all_products = []
        
        for page in range(1, max_pages + 1):
            result = self.searcher.search_products(
                keyword=keyword,
                page=page,
                page_size=page_size            )
            products = self.searcher.parse_search_results(result)
            
            if not products:
                break
            
            all_products.extend(products)
            print(f"  第{page}页: {len(products)}条，累计:{len(all_products)}条")
            
            # 智能延时：根据返回数据量调整
            time.sleep(delay)
        
        return all_products

7.2 缓存策略

Python

import jsonfrom functools import lru_cacheclass CachedSearcher(AlibabaKeywordSearch):
    """带缓存的搜索器"""
    
    def __init__(self, app_key: str, app_secret: str, cache_file: str = "search_cache.json"):
        super().__init__(app_key, app_secret)
        self.cache_file = cache_file
        self._load_cache()
    
    def _load_cache(self):
        try:
            with open(self.cache_file, 'r', encoding='utf-8') as f:
                self.cache = json.load(f)
        except FileNotFoundError:
            self.cache = {}
    
    def _save_cache(self):
        with open(self.cache_file, 'w', encoding='utf-8') as f:
            json.dump(self.cache, f, ensure_ascii=False, indent=2)
    
    def search_products(self, **kwargs):
        cache_key = json.dumps(kwargs, sort_keys=True)
        
        if cache_key in self.cache:
            print("📦 命中缓存")
            return self.cache[cache_key]
        
        result = super().search_products(**kwargs)
        self.cache[cache_key] = result
        self._save_cache()
        return result

7.3 并发请求（注意QPS限制）

Python

from concurrent.futures import ThreadPoolExecutor, as_completeddef parallel_search(searcher, keywords: list, max_workers: int = 3) -> dict:
    """
    并发搜索多个关键词（需确保不超过QPS限制）
    
    :param max_workers: 并发数，建议不超过3
    """
    results = {}
    
    with ThreadPoolExecutor(max_workers=max_workers) as executor:
        future_to_kw = {
            executor.submit(searcher.search_products, kw, page=1, page_size=20): kw 
            for kw in keywords        }
        
        for future in as_completed(future_to_kw):
            kw = future_to_kw[future]
            try:
                result = future.result()
                results[kw] = searcher.parse_search_results(result)
            except Exception as e:
                results[kw] = {"error": str(e)}
    
    return results

八、错误处理与常见问题

8.1 常见错误码

表格

错误码	说明	解决方案
`isv.invalid-parameter`	参数错误	检查必填参数和格式
`isv.permission-api-forbidden`	API权限不足	申请对应接口权限
`isv.permission-ip-limit`	IP受限	配置服务器IP白名单
`isp.call-limited`	调用频率超限	降低QPS，增加延时
`isv.invalid-signature`	签名错误	检查签名算法和参数排序
`isv.item-not-exist`	商品不存在	检查商品ID是否有效

8.2 完整错误处理封装

Python

class ApiErrorHandler:
    """API错误处理工具"""
    
    @staticmethod
    def handle_response(response: dict) -> dict:
        """
        统一处理API响应
        """
        if 'error_response' in response:
            error = response['error_response']
            code = error.get('code', 'unknown')
            msg = error.get('msg', '未知错误')
            
            error_map = {
                'isv.invalid-signature': "签名错误，请检查App Secret和签名算法",
                'isv.permission-api-forbidden': "API权限不足，请在开放平台申请权限",
                'isp.call-limited': "调用频率超限，请降低请求频率",
                'isv.invalid-parameter': f"参数错误: {msg}",
            }
            
            return {
                'success': False,
                'error_code': code,
                'error_msg': error_map.get(code, msg),
                'raw': response            }
        
        return {'success': True, 'data': response}

九、总结

表格

能力	关键词搜索	图搜（拍立淘）
输入方式	文字关键词	图片URL/Base64
适用场景	批量找货、市场分析	找同款、竞品追踪
精准度	依赖关键词质量	图像匹配，更精准
返回速度	快（秒级）	稍慢（需图像处理）
最佳实践	组合使用，先用关键词筛选，再用图搜验证

调用淘宝API接口获取商品类目：完整开发指南

Tue, 02 Jun 2026 14:53:27 +0800

一、概述

淘宝商品类目是电商平台最基础的数据结构之一，它决定了商品在平台上的展示方式、属性规则以及搜索推荐逻辑。淘宝开放平台提供了多个类目相关API接口，开发者可以通过这些接口获取完整的类目体系、类目属性及属性值等信息。本文将详细介绍如何注册开发者账号、申请权限、生成签名并调用类目相关API。

二、准备工作

2.1 注册淘宝开放平台账号

访问淘宝开放平台官网
点击注册，使用淘宝账号登录并完成实名认证
进入控制台，创建新应用

2.2 创建应用并获取密钥

在开放平台控制台中：

进入「应用管理」→「创建应用」
填写应用名称、描述、图标等信息
选择应用类型（自用型或他用型）
填写使用场景和目的
提交审核，审核通过后将获得：

App Key：应用标识
App Secret：应用密钥（用于签名）

2.3 申请API权限

在应用详情页中找到「接口权限」模块，申请以下类目相关接口权限：

taobao.itemcats.get — 获取可供发布的类目列表
taobao.item.cat.get — 获取单个类目详情
taobao.itemprops.get — 获取标准商品类目属性
taobao.itempropvalues.get — 获取标准类目属性值
taobao.itemcats.authorize.get — 查询商家被授权的品牌和类目

⚠️ 注意：淘宝开放平台会根据应用情况、资质及安全策略进行审核，部分接口可能需要商家授权才能调用。

三、核心API接口详解

3.1 接口概览

表格

接口名称	功能描述	是否需要授权
`taobao.itemcats.get`	获取可供发布的标准类目列表	否
`taobao.item.cat.get`	获取单个类目详细信息	否
`taobao.itemprops.get`	获取类目的属性列表	否
`taobao.itempropvalues.get`	获取类目的属性值列表	否
`taobao.itemcats.authorize.get`	查询商家被授权的品牌和类目	是

四、签名生成机制

淘宝API采用 HMAC-MD5 签名算法，所有请求参数必须经过签名验证。签名步骤如下：

4.1 签名规则

排序：将所有参数（除 sign 外）按参数名 ASCII 码升序排列
拼接：将排序后的参数按 key+value 格式拼接成字符串，首尾各加一次 App Secret
加密：使用 HMAC-MD5 算法加密，结果转大写

4.2 Python签名示例

Python

import hmacimport hashlibimport urllib.parsedef generate_sign(params: dict, app_secret: str) -> str:
    """
    生成淘宝API请求签名（HMAC-MD5）
    
    :param params: 请求参数（不含sign）
    :param app_secret: 应用密钥
    :return: 签名字符串（大写）
    """
    # 1. 过滤空值参数，按key排序
    sorted_params = sorted(
        [(k, v) for k, v in params.items() if v is not None and k != 'sign'],
        key=lambda x: x[0]
    )
    
    # 2. 拼接字符串：secret + key1value1 + key2value2 + ... + secret
    sign_content = app_secret    for key, value in sorted_params:
        sign_content += f"{key}{value}"
    sign_content += app_secret    
    # 3. HMAC-MD5加密
    sign = hmac.new(
        app_secret.encode('utf-8'),
        sign_content.encode('utf-8'),
        hashlib.md5    ).hexdigest().upper()
    
    return sign

五、完整调用示例

5.1 获取一级类目列表（taobao.itemcats.get）

该接口用于获取可供发布的标准类目，通过递归调用可获取完整的类目树。

Python

import requestsimport timeimport json# ========== 配置信息 ==========APP_KEY = 'your_app_key'          # 替换为你的App KeyAPP_SECRET = 'your_app_secret'    # 替换为你的App SecretAPI_URL = 'https://eco.taobao.com/router/rest'# ========== 签名生成函数 ==========def generate_sign(params, app_secret):
    sorted_params = sorted(
        [(k, v) for k, v in params.items() if v is not None and k != 'sign'],
        key=lambda x: x[0]
    )
    sign_content = app_secret    for key, value in sorted_params:
        sign_content += f"{key}{value}"
    sign_content += app_secret    
    import hmac, hashlib
    sign = hmac.new(
        app_secret.encode('utf-8'),
        sign_content.encode('utf-8'),
        hashlib.md5    ).hexdigest().upper()
    return sign# ========== 获取类目列表 ==========def get_itemcats(parent_cid=0, fields=None):
    """
    调用 taobao.itemcats.get 获取类目列表
    
    :param parent_cid: 父类目ID，0表示获取一级类目
    :param fields: 返回字段，默认获取常用字段
    :return: 类目列表
    """
    if fields is None:
        fields = "cid,parent_cid,name,is_parent,status,sort_order"
    
    params = {
        'app_key': APP_KEY,
        'method': 'taobao.itemcats.get',
        'timestamp': time.strftime('%Y-%m-%d %H:%M:%S'),
        'format': 'json',
        'v': '2.0',
        'sign_method': 'hmac-md5',
        'fields': fields,
        'parent_cid': parent_cid    }
    
    # 生成签名
    params['sign'] = generate_sign(params, APP_SECRET)
    
    # 发送请求
    response = requests.post(API_URL, data=params, timeout=30)
    result = response.json()
    
    # 解析结果
    if 'itemcats_get_response' in result:
        itemcats = result['itemcats_get_response'].get('item_cats', {}).get('item_cat', [])
        return itemcats    else:
        error = result.get('error_response', {})
        print(f"调用失败: {error.get('msg', '未知错误')}")
        return None# ========== 递归获取完整类目树 ==========def build_category_tree(parent_cid=0, max_depth=3, current_depth=0):
    """
    递归构建类目树
    
    :param parent_cid: 父类目ID
    :param max_depth: 最大递归深度
    :param current_depth: 当前深度
    :return: 类目树结构
    """
    if current_depth >= max_depth:
        return []
    
    categories = get_itemcats(parent_cid)
    if not categories:
        return []
    
    tree = []
    for cat in categories:
        node = {
            'cid': cat.get('cid'),
            'name': cat.get('name'),
            'parent_cid': cat.get('parent_cid'),
            'is_parent': cat.get('is_parent'),
            'status': cat.get('status'),
            'sort_order': cat.get('sort_order'),
            'children': []
        }
        
        # 如果是父类目，递归获取子类目
        if cat.get('is_parent') and current_depth < max_depth - 1:
            # 添加延时避免触发限流（个人开发者通常1QPS）
            time.sleep(1.2)
            node['children'] = build_category_tree(
                cat.get('cid'), max_depth, current_depth + 1
            )
        
        tree.append(node)
    
    return tree# ========== 主程序 ==========if __name__ == '__main__':
    print("=" * 50)
    print("淘宝商品类目获取工具")
    print("=" * 50)
    
    # 获取一级类目
    print("\n【一级类目列表】")
    top_categories = get_itemcats(parent_cid=0)
    for cat in top_categories[:5]:  # 只展示前5个
        print(f"  📁 [{cat['cid']}] {cat['name']} "
              f"{'(有子类目)' if cat.get('is_parent') else '(叶子类目)'}")
    
    # 获取某个一级类目下的二级类目（以女装为例：cid=50008163）
    print("\n【女装类目下的二级类目】")
    sub_categories = get_itemcats(parent_cid=50008163)
    if sub_categories:
        for cat in sub_categories[:5]:
            print(f"  📂 [{cat['cid']}] {cat['name']}")
    
    # 构建完整类目树（示例：只递归2层）
    print("\n【构建类目树（深度2）】")
    # tree = build_category_tree(parent_cid=0, max_depth=2)
    # print(json.dumps(tree, ensure_ascii=False, indent=2))

5.2 获取单个类目详情（taobao.item.cat.get）

Python

def get_item_cat_detail(cid):
    """
    获取单个类目的详细信息
    
    :param cid: 类目ID
    :return: 类目详情
    """
    params = {
        'app_key': APP_KEY,
        'method': 'taobao.item.cat.get',
        'timestamp': time.strftime('%Y-%m-%d %H:%M:%S'),
        'format': 'json',
        'v': '2.0',
        'sign_method': 'hmac-md5',
        'cid': cid    }
    
    params['sign'] = generate_sign(params, APP_SECRET)
    
    response = requests.post(API_URL, data=params, timeout=30)
    result = response.json()
    
    if 'item_cat_get_response' in result:
        return result['item_cat_get_response'].get('item_cat', {})
    else:
        error = result.get('error_response', {})
        print(f"获取失败: {error.get('msg', '未知错误')}")
        return None# 示例：获取女装类目详情cat_detail = get_item_cat_detail(50008163)if cat_detail:
    print(f"类目名称: {cat_detail.get('name')}")
    print(f"父类目ID: {cat_detail.get('parent_cid')}")
    print(f"是否为父类目: {cat_detail.get('is_parent')}")

5.3 获取类目属性（taobao.itemprops.get）

Python

def get_item_props(cid, fields=None):
    """
    获取类目的属性列表
    
    :param cid: 类目ID
    :param fields: 返回字段
    :return: 属性列表
    """
    if fields is None:
        fields = "pid,name,must,multi,prop_values,is_key_prop,is_sale_prop"
    
    params = {
        'app_key': APP_KEY,
        'method': 'taobao.itemprops.get',
        'timestamp': time.strftime('%Y-%m-%d %H:%M:%S'),
        'format': 'json',
        'v': '2.0',
        'sign_method': 'hmac-md5',
        'fields': fields,
        'cid': cid    }
    
    params['sign'] = generate_sign(params, APP_SECRET)
    
    response = requests.post(API_URL, data=params, timeout=30)
    result = response.json()
    
    if 'itemprops_get_response' in result:
        return result['itemprops_get_response'].get('item_props', {}).get('item_prop', [])
    else:
        error = result.get('error_response', {})
        print(f"获取失败: {error.get('msg', '未知错误')}")
        return None# 示例：获取手机类目（cid=1512）的属性print("\n【手机类目属性列表】")props = get_item_props(1512)if props:
    for prop in props[:5]:
        print(f"  🔧 [{prop['pid']}] {prop['name']} "
              f"{'(必填)' if prop.get('must') else '(选填)'} "
              f"{'(多选)' if prop.get('multi') else '(单选)'}")

5.4 获取类目属性值（taobao.itempropvalues.get）点击测试获取测试key

Python

def get_item_prop_values(cid, pvs, fields=None):
    """
    获取类目的属性值列表
    
    :param cid: 类目ID
    :param pvs: 属性ID，多个用逗号分隔，如 "139248429"
    :param fields: 返回字段
    :return: 属性值列表
    """
    if fields is None:
        fields = "cid,pid,prop_name,vid,name,name_alias,status,sort_order"
    
    params = {
        'app_key': APP_KEY,
        'method': 'taobao.itempropvalues.get',
        'timestamp': time.strftime('%Y-%m-%d %H:%M:%S'),
        'format': 'json',
        'v': '2.0',
        'sign_method': 'hmac-md5',
        'fields': fields,
        'cid': cid,
        'pvs': pvs    }
    
    params['sign'] = generate_sign(params, APP_SECRET)
    
    response = requests.post(API_URL, data=params, timeout=30)
    result = response.json()
    
    if 'itempropvalues_get_response' in result:
        return result['itempropvalues_get_response'].get('prop_values', {}).get('prop_value', [])
    else:
        error = result.get('error_response', {})
        print(f"获取失败: {error.get('msg', '未知错误')}")
        return None

六、返回字段说明

6.1 taobao.itemcats.get 返回字段

表格

字段名	类型	说明
`cid`	Number	类目ID
`parent_cid`	Number	父类目ID
`name`	String	类目名称
`is_parent`	Boolean	是否为父类目（true表示有子类目）
`status`	String	类目状态（normal/normal）
`sort_order`	Number	排序值

6.2 类目属性字段

表格

字段名	类型	说明
`pid`	Number	属性ID
`name`	String	属性名称
`must`	Boolean	是否必填
`multi`	Boolean	是否多选
`is_key_prop`	Boolean	是否关键属性（如品牌、型号）
`is_sale_prop`	Boolean	是否销售属性（影响SKU）

七、关键注意事项

7.1 调用频率限制

淘宝开放平台对接口调用有 QPS（每秒查询率）限制，个人开发者通常限制为 1 QPS。建议在代码中加入延时控制：

Python

import time# 每次调用间隔至少1.2秒time.sleep(1.2)

7.2 叶子类目

商品必须发布在叶子类目（is_parent = false）下。获取叶子类目的方法是通过递归调用 taobao.itemcats.get，直到 is_parent 为 false。

7.3 管控类目

部分类目（如食品、酒类、书籍、保健品、农药、医疗等）需要卖家提供相关资质才能发布，调用 taobao.itemcats.authorize.get 可查询商家是否有权限发布。

7.4 错误处理

常见错误码及处理方式：

表格

错误码	说明	解决方案
`isv.invalid-parameter`	参数错误	检查参数格式和必填项
`isv.permission-ip-limit`	IP受限	在开放平台配置服务器IP白名单
`isv.permission-api-forbidden`	API权限不足	申请对应接口权限
`isv.trade-not-exist`	交易不存在	检查传入的参数值

八、Java调用示例

java

import java.util.*;import javax.crypto.Mac;import javax.crypto.spec.SecretKeySpec;import org.apache.commons.codec.binary.Hex;import java.net.URLEncoder;public class TaobaoCategoryApi {
    
    private static final String APP_KEY = "your_app_key";
    private static final String APP_SECRET = "your_app_secret";
    private static final String API_URL = "https://eco.taobao.com/router/rest";
    
    /**
     * 生成HMAC-MD5签名
     */
    public static String generateSign(Map<String, String> params, String appSecret) throws Exception {
        List<String> keys = new ArrayList<>(params.keySet());
        Collections.sort(keys);
        
        StringBuilder sb = new StringBuilder();
        sb.append(appSecret);
        for (String key : keys) {
            String value = params.get(key);
            if (value != null && !value.isEmpty()) {
                sb.append(key).append(value);
            }
        }
        sb.append(appSecret);
        
        Mac mac = Mac.getInstance("HmacMD5");
        SecretKeySpec secretKey = new SecretKeySpec(appSecret.getBytes("UTF-8"), "HmacMD5");
        mac.init(secretKey);
        byte[] bytes = mac.doFinal(sb.toString().getBytes("UTF-8"));
        return Hex.encodeHexString(bytes).toUpperCase();
    }
    
    /**
     * 获取类目列表
     */
    public static String getItemCats(long parentCid) throws Exception {
        Map<String, String> params = new HashMap<>();
        params.put("app_key", APP_KEY);
        params.put("method", "taobao.itemcats.get");
        params.put("timestamp", new java.text.SimpleDateFormat("yyyy-MM-dd HH:mm:ss").format(new Date()));
        params.put("format", "json");
        params.put("v", "2.0");
        params.put("sign_method", "hmac-md5");
        params.put("fields", "cid,parent_cid,name,is_parent,status,sort_order");
        params.put("parent_cid", String.valueOf(parentCid));
        
        String sign = generateSign(params, APP_SECRET);
        params.put("sign", sign);
        
        // 构造请求URL
        StringBuilder urlBuilder = new StringBuilder(API_URL + "?");
        for (Map.Entry<String, String> entry : params.entrySet()) {
            urlBuilder.append(entry.getKey())
                     .append("=")
                     .append(URLEncoder.encode(entry.getValue(), "UTF-8"))
                     .append("&");
        }
        
        // 发送HTTP请求（此处使用HttpClient或URLConnection）
        // ...
        return urlBuilder.toString();
    }
    
    public static void main(String[] args) throws Exception {
        System.out.println(getItemCats(0));
    }}

九、总结

表格

步骤	内容
1. 注册	在淘宝开放平台注册开发者账号并完成实名认证
2. 创建应用	创建应用并获取 App Key 和 App Secret
3. 申请权限	申请类目相关API接口权限
4. 生成签名	使用 HMAC-MD5 算法生成请求签名
5. 调用接口	通过 HTTP POST/GET 发送请求到 `https://eco.taobao.com/router/rest`
6. 解析数据	解析JSON响应，构建类目树结构

电商平台api：1688 阿里巴巴平台商品详情全解析

Thu, 28 May 2026 15:23:51 +0800

1688（阿里巴巴中国站）作为国内最大的 B2B 批发平台，其商品详情数据涵盖了从基础信息到供应链属性的完整维度。本文将深入解析 1688 商品详情的全量数据字段，并对比主流电商平台的 API 接口体系，帮助开发者全面掌握 B2B 电商数据采集技术。

一、1688 商品详情数据结构全景图

1688 商品详情不同于淘宝/天猫的 C2C/B2C 模式，其核心特征是批发属性——起订量、阶梯价、代发支持、工厂信息等字段是 1688 独有的数据维度。

plain

复制

┌─────────────────────────────────────────────────────────────┐
│                    1688 商品详情数据结构                       │
├─────────────────────────────────────────────────────────────┤
│  1. 基础信息层                                               │
│     ├── 商品ID (num_iid / productId)                         │
│     ├── 标题 (subject / title)                               │
│     ├── 副标题 / 营销文案                                     │
│     ├── 主图列表 (imageUrls / item_imgs)                     │
│     ├── 商品视频 (video_url)                                  │
│     └── 商品状态 (status: 上架/下架/删除)                      │
├─────────────────────────────────────────────────────────────┤
│  2. 价格体系层（1688 核心）                                   │
│     ├── 参考价 / 划线价 (reference_price)                     │
│     ├── 售价 (price)                                          │
│     ├── 阶梯批发价 (priceRanges)                              │
│     │   └── [{startQuantity: 1, price: 10.0}, ...]           │
│     ├── 代发价 (agent_price / dropship_price)                 │
│     ├── 最小起订量 (moq / minOrderQuantity)                   │
│     └── 计量单位 (unit: 件/套/箱)                             │
├─────────────────────────────────────────────────────────────┤
│  3. SKU 规格层                                               │
│     ├── 规格属性 (specs: 颜色/尺码/材质...)                    │
│     ├── SKU 映射表 (skuMap)                                   │
│     │   └── "红色;M": {skuId, price, stock, specId}          │
│     └── SKU 图片 (sku_images)                                 │
├─────────────────────────────────────────────────────────────┤
│  4. 库存物流层                                               │
│     ├── 总库存 (stock / quantity)                             │
│     ├── 可售数量 (amountOnSale)                               │
│     ├── 发货地 (location / sendGoodsAddress)                  │
│     ├── 运费模板 (freightTemplateID)                          │
│     ├── 重量体积 (weight / volume / packageSize)              │
│     └── 发货时效 (deliveryTime)                               │
├─────────────────────────────────────────────────────────────┤
│  5. 供应商信息层（1688 特色）                                 │
│     ├── 供应商名称 (supplier / sellerNick)                    │
│     ├── 供应商ID (seller_id)                                  │
│     ├── 诚信通等级 (creditLevel / supplier_level)             │
│     ├── 实力商家/超级工厂标识                                  │
│     ├── 经营模式 (manufacturer / trader)                      │
│     ├── 工厂信息 (factoryInfo)                                │
│     └── 近 90 天成交额 / 回头率                                │
├─────────────────────────────────────────────────────────────┤
│  6. 交易服务层                                               │
│     ├── 是否支持混批 (isSupportMix)                           │
│     ├── 是否支持一件代发 (isSupportAgent / dropship)          │
│     ├── 支付方式 (creditPayment / assurance)                  │
│     ├── 服务保障 (7天包退 / 材质保障...)                       │
│     └── 跨境专供标识 (crossBorder)                            │
├─────────────────────────────────────────────────────────────┤
│  7. 商品属性层                                               │
│     ├── 类目属性 (categoryProps)                              │
│     ├── 自定义属性 (attributes)                               │
│     ├── 商品描述 (description / detail)                       │
│     └── 商品参数 (paramList)                                  │
├─────────────────────────────────────────────────────────────┤
│  8. 营销数据层                                               │
│     ├── 30 天销量 (saleCount / sales)                         │
│     ├── 总销量 (totalSales)                                   │
│     ├── 评价数 (evaluateCount)                                │
│     ├── 好评率 (goodRate)                                     │
│     └── 收藏数 (favCount)                                     │
└─────────────────────────────────────────────────────────────┘

二、官方 API 接口详解：`alibaba.product.get`

2.1 接口基本信息

表格

属性	说明
接口地址	`https://gw.open.1688.com/openapi/param2/1/com.alibaba.product/alibaba.product.get`
请求方式	POST / GET
数据格式	`application/x-www-form-urlencoded`
返回格式	JSON
接口版本	2.0（2026 年主推，1.0 已不返回核心字段）

2.2 请求参数

表格

参数名	类型	必选	说明
`app_key`	String	是	应用唯一标识
`method`	String	是	固定值 `com.alibaba.product.alibaba.product.get`
`timestamp`	String	是	时间戳（`yyyy-MM-dd HH:mm:ss`）
`v`	String	是	API 版本：`2.0`
`format`	String	是	返回格式：`json`
`sign`	String	是	MD5 签名
`access_token`	String	是	OAuth2.0 授权令牌
`productId`	Long	是	1688 商品数字 ID
`fields`	String	否	指定返回字段，逗号分隔

2.3 核心返回字段全解析

JSON

复制

{
  "result": {
    "productId": 619899292404,
    "subject": "2026新款蓝牙耳机无线降噪运动入耳式超长续航",
    "description": "<html>...</html>",
    "imageUrl": "https://cbu01.alicdn.com/img/ibank/xxx.jpg",
    "detailPage": "https://detail.1688.com/offer/619899292404.html",
    "status": "published",
    
    "priceRanges": [
      {"startQuantity": 1, "price": 35.00},
      {"startQuantity": 50, "price": 32.00},
      {"startQuantity": 200, "price": 28.50},
      {"startQuantity": 500, "price": 25.00}
    ],
    
    "saleInfo": {
      "minOrderQuantity": 2,
      "unit": "件",
      "amountOnSale": 5000,
      "priceRanges": [...]
    },
    
    "skuInfo": {
      "skuMap": {
        "黑色;标准版": {
          "skuId": "4862151234567",
          "price": 35.00,
          "stock": 1200
        },
        "白色;顶配版": {
          "skuId": "4862151234568",
          "price": 55.00,
          "stock": 800
        }
      },
      "specs": [
        {
          "specId": "1627207",
          "name": "颜色",
          "values": [
            {"valueId": "28320", "name": "黑色"},
            {"valueId": "28321", "name": "白色"}
          ]
        },
        {
          "specId": "20509",
          "name": "版本",
          "values": [
            {"valueId": "28324", "name": "标准版"},
            {"valueId": "28325", "name": "顶配版"}
          ]
        }
      ]
    },
    
    "shippingInfo": {
      "freightTemplateID": 11754104,
      "unitWeight": 0.15,
      "packageSize": "10x8x5",
      "volume": 400,
      "sendGoodsAddress": {
        "province": "广东",
        "city": "深圳"
      }
    },
    
    "supplier": {
      "supplierId": "b2b-1234567890",
      "companyName": "深圳市xxx电子有限公司",
      "creditLevel": "AAAAA",
      "isFactory": true,
      "mainProducts": ["蓝牙耳机", "音箱"]
    }
  }}

三、Python 完整实战：商品详情全字段采集，获取账号测试

Python

复制

# -*- coding: utf-8 -*-"""
1688 商品详情全字段采集器
支持官方 API 和第三方聚合 API
"""import requestsimport hashlibimport timeimport jsonimport urllib.parsefrom typing import Dict, List, Optionalfrom dataclasses import dataclass, asdictfrom datetime import datetime@dataclassclass PriceRange:
    """阶梯价格"""
    start_quantity: int
    price: float@dataclassclass SkuItem:
    """SKU 规格"""
    sku_id: str
    properties: str           # 如"黑色;标准版"
    price: float
    stock: int@dataclassclass ProductSpec:
    """商品规格定义"""
    spec_id: str
    name: str                 # 如"颜色"、"尺码"
    values: List[Dict]        # [{valueId, name}, ...]@dataclassclass ShippingInfo:
    """物流信息"""
    freight_template_id: int
    unit_weight: float        # 单位重量(kg)
    package_size: str         # 包装尺寸(cm)
    volume: int               # 体积(cm³)
    province: str
    city: str@dataclassclass SupplierInfo:
    """供应商信息"""
    supplier_id: str
    company_name: str
    credit_level: str
    is_factory: bool
    main_products: List[str]@dataclassclass ProductDetail:
    """1688 商品详情完整结构"""
    # 基础信息
    product_id: str
    title: str
    description: str
    main_image: str
    images: List[str]
    detail_url: str
    status: str
    
    # 价格体系
    reference_price: float
    price_ranges: List[PriceRange]
    min_order_quantity: int
    unit: str
    
    # SKU 信息
    skus: List[SkuItem]
    specs: List[ProductSpec]
    
    # 库存物流
    total_stock: int
    amount_on_sale: int
    shipping: ShippingInfo    
    # 供应商
    supplier: SupplierInfo    
    # 交易服务
    support_mix: bool
    support_agent: bool          # 一件代发
    support_credit: bool         # 账期支付
    
    # 营销数据
    sales_30d: int
    total_sales: int
    evaluate_count: int
    good_rate: float
    
    # 时间戳
    create_time: str
    update_time: strclass Alibaba1688DetailAPI:
    """
    1688 商品详情 API 客户端
    """
    
    OFFICIAL_URL = "https://gw.open.1688.com/openapi/param2/1/com.alibaba.product/alibaba.product.get"
    
    def __init__(self, app_key: str, app_secret: str, access_token: str):
        self.app_key = app_key
        self.app_secret = app_secret
        self.access_token = access_token
        self.session = requests.Session()
        
    def _generate_sign(self, params: dict) -> str:
        """
        生成 MD5 签名
        规则：app_secret + 排序后参数拼接 + app_secret，整体 MD5 大写
        """
        sorted_params = sorted(
            [(k, v) for k, v in params.items() if v is not None and k != 'sign'],
            key=lambda x: x[0]
        )
        param_str = ''.join([f"{k}{urllib.parse.quote_plus(str(v))}" for k, v in sorted_params])
        sign_str = f"{self.app_secret}{param_str}{self.app_secret}"
        return hashlib.md5(sign_str.encode('utf-8')).hexdigest().upper()
    
    def get_product_detail(
        self,
        product_id: str,
        fields: Optional[str] = None
    ) -> Optional[ProductDetail]:
        """
        获取商品详情（全字段）
        
        Args:
            product_id: 1688 商品数字 ID
            fields: 指定返回字段，None 则返回全部
        
        Returns:
            ProductDetail 对象
        """
        # 默认请求全量字段
        if fields is None:
            fields = (
                "productId,subject,description,imageUrl,detailPage,status,"
                "priceRanges,saleInfo,skuInfo,shippingInfo,"
                "supplier,creditInfo,tradeServiceInfo,"
                "createTime,lastUpdateTime"
            )
        
        params = {
            "app_key": self.app_key,
            "method": "com.alibaba.product.alibaba.product.get",
            "timestamp": time.strftime("%Y-%m-%d %H:%M:%S"),
            "v": "2.0",
            "format": "json",
            "access_token": self.access_token,
            "productId": product_id,
            "fields": fields        }
        
        params["sign"] = self._generate_sign(params)
        
        try:
            response = self.session.post(
                self.OFFICIAL_URL,
                data=params,
                timeout=30
            )
            response.raise_for_status()
            data = response.json()
            
            if data.get("success"):
                return self._parse_product(data.get("result", {}))
            else:
                print(f"API 错误: {data.get('errorMessage', '未知错误')}")
                return None
                
        except requests.exceptions.RequestException as e:
            print(f"请求异常: {e}")
            return None
    
    def _parse_product(self, data: dict) -> ProductDetail:
        """解析 API 响应为结构化对象"""
        
        # 解析价格区间
        price_ranges = []
        for pr in data.get("priceRanges", []):
            price_ranges.append(PriceRange(
                start_quantity=pr.get("startQuantity", 1),
                price=float(pr.get("price", 0))
            ))
        
        # 解析 SKU
        skus = []
        sku_info = data.get("skuInfo", {})
        sku_map = sku_info.get("skuMap", {})
        for props, sku_data in sku_map.items():
            skus.append(SkuItem(
                sku_id=str(sku_data.get("skuId", "")),
                properties=props,
                price=float(sku_data.get("price", 0)),
                stock=int(sku_data.get("stock", 0))
            ))
        
        # 解析规格定义
        specs = []
        for spec in sku_info.get("specs", []):
            specs.append(ProductSpec(
                spec_id=str(spec.get("specId", "")),
                name=spec.get("name", ""),
                values=spec.get("values", [])
            ))
        
        # 解析物流信息
        shipping = data.get("shippingInfo", {})
        address = shipping.get("sendGoodsAddress", {})
        shipping_info = ShippingInfo(
            freight_template_id=shipping.get("freightTemplateID", 0),
            unit_weight=float(shipping.get("unitWeight", 0) or 0),
            package_size=shipping.get("packageSize", ""),
            volume=int(shipping.get("volume", 0) or 0),
            province=address.get("province", ""),
            city=address.get("city", "")
        )
        
        # 解析供应商
        supplier = data.get("supplier", {})
        supplier_info = SupplierInfo(
            supplier_id=supplier.get("supplierId", ""),
            company_name=supplier.get("companyName", ""),
            credit_level=supplier.get("creditLevel", ""),
            is_factory=supplier.get("isFactory", False),
            main_products=supplier.get("mainProducts", [])
        )
        
        # 交易服务
        trade_service = data.get("tradeServiceInfo", {})
        
        return ProductDetail(
            product_id=str(data.get("productId", "")),
            title=data.get("subject", ""),
            description=data.get("description", ""),
            main_image=data.get("imageUrl", ""),
            images=data.get("imageUrls", [data.get("imageUrl", "")]),
            detail_url=data.get("detailPage", ""),
            status=data.get("status", ""),
            
            reference_price=float(data.get("referencePrice", 0) or 0),
            price_ranges=price_ranges,
            min_order_quantity=data.get("saleInfo", {}).get("minOrderQuantity", 1),
            unit=data.get("saleInfo", {}).get("unit", "件"),
            
            skus=skus,
            specs=specs,
            
            total_stock=int(data.get("stock", 0) or 0),
            amount_on_sale=int(data.get("saleInfo", {}).get("amountOnSale", 0) or 0),
            shipping=shipping_info,
            
            supplier=supplier_info,
            
            support_mix=trade_service.get("supportMix", False),
            support_agent=trade_service.get("supportAgent", False),
            support_credit=trade_service.get("supportCredit", False),
            
            sales_30d=int(data.get("saleCount", 0) or 0),
            total_sales=int(data.get("totalSales", 0) or 0),
            evaluate_count=int(data.get("evaluateCount", 0) or 0),
            good_rate=float(data.get("goodRate", 0) or 0),
            
            create_time=data.get("createTime", ""),
            update_time=data.get("lastUpdateTime", "")
        )
    
    def export_to_json(self, product: ProductDetail, filename: str = None):
        """导出商品详情为 JSON 文件"""
        if filename is None:
            filename = f"product_{product.product_id}_{datetime.now().strftime('%Y%m%d')}.json"
        
        with open(filename, 'w', encoding='utf-8') as f:
            json.dump(asdict(product), f, ensure_ascii=False, indent=2)
        
        print(f"已导出: {filename}")
        return filename# ==================== 使用示例 ====================if __name__ == "__main__":
    api = Alibaba1688DetailAPI(
        app_key="your_app_key",
        app_secret="your_app_secret",
        access_token="your_access_token"
    )
    
    # 获取商品详情
    product_id = "619899292404"
    detail = api.get_product_detail(product_id)
    
    if detail:
        print("=" * 60)
        print(f"商品详情: {detail.title}")
        print("=" * 60)
        
        # 基础信息
        print(f"\n【基础信息】")
        print(f"商品ID: {detail.product_id}")
        print(f"状态: {detail.status}")
        print(f"链接: {detail.detail_url}")
        
        # 价格体系
        print(f"\n【价格体系】")
        print(f"起订量: {detail.min_order_quantity} {detail.unit}")
        print("阶梯批发价:")
        for pr in detail.price_ranges:
            print(f"  ≥{pr.start_quantity} {detail.unit}: ¥{pr.price}")
        
        # SKU 信息
        print(f"\n【SKU 规格】")
        for spec in detail.specs:
            values = [v["name"] for v in spec.values]
            print(f"  {spec.name}: {', '.join(values)}")
        
        print(f"\nSKU 明细:")
        for sku in detail.skus:
            print(f"  {sku.properties} → ¥{sku.price} (库存:{sku.stock})")
        
        # 物流信息
        print(f"\n【物流信息】")
        print(f"发货地: {detail.shipping.province} {detail.shipping.city}")
        print(f"重量: {detail.shipping.unit_weight} kg")
        print(f"尺寸: {detail.shipping.package_size} cm")
        print(f"体积: {detail.shipping.volume} cm³")
        
        # 供应商
        print(f"\n【供应商信息】")
        print(f"公司: {detail.supplier.company_name}")
        print(f"诚信通: {detail.supplier.credit_level}")
        print(f"工厂店: {'是' if detail.supplier.is_factory else '否'}")
        print(f"主营: {', '.join(detail.supplier.main_products)}")
        
        # 交易服务
        print(f"\n【交易服务】")
        print(f"支持混批: {'是' if detail.support_mix else '否'}")
        print(f"支持代发: {'是' if detail.support_agent else '否'}")
        print(f"支持账期: {'是' if detail.support_credit else '否'}")
        
        # 营销数据
        print(f"\n【营销数据】")
        print(f"30天销量: {detail.sales_30d}")
        print(f"总销量: {detail.total_sales}")
        print(f"评价数: {detail.evaluate_count}")
        print(f"好评率: {detail.good_rate}%")
        
        # 导出 JSON
        api.export_to_json(detail)

四、主流电商平台 API 接口对比

4.1 平台接口体系概览

表格

平台	核心详情接口	特色字段	认证要求
1688	`alibaba.product.get`	阶梯价、起订量、代发价、工厂信息	企业认证
淘宝/天猫	`taobao.item.get`	优惠券、天猫积分、物流模板	个人/企业
京东	`jd.item.get`	京东价、PLUS 价、自营标识	企业认证
拼多多	`pdd.goods.detail.get`	拼单价、百亿补贴、成团价	企业认证
抖音电商	`douyin.product.detail`	直播价、达人佣金、短视频挂车	企业认证
亚马逊	`GetMatchingProduct`	ASIN、BuyBox、FBA 费用	开发者账号

4.2 商品详情字段对比

Python

复制

# 各平台商品详情字段映射对比PLATFORM_FIELD_MAP = {
    "1688": {
        "商品ID": "productId / num_iid",
        "标题": "subject",
        "售价": "priceRanges[0].price",
        "批发阶梯价": "priceRanges",
        "起订量": "saleInfo.minOrderQuantity",
        "代发价": "tradeServiceInfo.agentPrice",
        "供应商": "supplier.companyName",
        "诚信通等级": "supplier.creditLevel",
        "工厂标识": "supplier.isFactory",
        "发货地": "shippingInfo.sendGoodsAddress",
        "重量": "shippingInfo.unitWeight",
        "SKU价格": "skuInfo.skuMap.*.price",
        "30天销量": "saleCount",
        "支持混批": "tradeServiceInfo.supportMix",
        "支持代发": "tradeServiceInfo.supportAgent",
    },
    
    "taobao": {
        "商品ID": "num_iid",
        "标题": "title",
        "售价": "price",
        "原价": "original_price",
        "优惠券": "coupon_info",
        "店铺名": "nick",
        "店铺等级": "seller_credit_level",
        "发货地": "location",
        "邮费": "post_fee",
        "SKU价格": "skus.*.price",
        "月销量": "sold_quantity",
        "库存": "num",
        "天猫标识": "user_type",
    },
    
    "jd": {
        "商品ID": "skuId",
        "标题": "name",
        "京东价": "jdPrice",
        "PLUS价": "plusPrice",
        "自营标识": "isSelf",
        "店铺名": "shopName",
        "好评率": "goodRate",
        "评论数": "commentCount",
        "配送": "deliveryInfo",
        "产地": "productArea",
    },
    
    "pdd": {
        "商品ID": "goods_id",
        "标题": "goods_name",
        "拼单价": "min_group_price",
        "单买价": "min_normal_price",
        "百亿补贴": "is_billion_subsidy",
        "店铺名": "mall_name",
        "销量": "sales_tip",
        "成团人数": "group_required",
    }}

4.3 各平台 API 接入差异

表格

维度	1688	淘宝/天猫	京东	拼多多
开发者类型	企业为主	个人/企业	企业	企业
审核周期	1-2 工作日	即时	3-5 工作日	1-3 工作日
免费额度	50 次/分钟	100 次/分钟	有限	有限
签名算法	MD5	MD5/HMAC	HMAC-SHA256	MD5
数据更新延迟	~2 分钟	~1 分钟	~5 分钟	~3 分钟
特色能力	批发价/工厂直供	淘客佣金/直播	自营物流	社交电商裂变

五、第三方聚合 API 方案

由于官方 API 存在权限限制和字段分散的问题，实际开发中常使用第三方聚合 API 获取更完整的商品详情数据。

5.1 第三方 API 优势

表格

优势	说明
字段完整	整合包装信息、代发价、工厂资质等分散字段
无需企业认证	个人开发者即可使用
多平台统一	一套接口对接淘宝、1688、拼多多等多个平台
额外数据	提供物流分析、利润计算等增值字段

5.2 第三方 API 调用示例

Python

复制

import requestsclass ThirdParty1688API:
    """
    第三方 1688 聚合 API 客户端
    示例：使用 AliPrice / VV-Tool 等服务
    """
    
    def __init__(self, api_key: str, base_url: str = "https://api.example.com"):
        self.api_key = api_key
        self.base_url = base_url        
    def get_item_detail(self, num_iid: str) -> dict:
        """
        获取商品详情（含完整包装信息）
        """
        url = f"{self.base_url}/1688/item_get"
        params = {
            "key": self.api_key,
            "num_iid": num_iid,
            "lang": "zh-CN"
        }
        
        response = requests.get(url, params=params, timeout=30)
        data = response.json()
        
        if data.get("code") == 200:
            item = data.get("item", {})
            return {
                "product_id": item.get("num_iid"),
                "title": item.get("title"),
                "price": item.get("price"),
                "original_price": item.get("original_price"),
                "promotion_price": item.get("promotion_price"),
                
                # 包装信息（第三方 API 特有）
                "weight": item.get("weight"),              # 净重
                "gross_weight": item.get("grossWeight"),   # 毛重
                "packing": item.get("packing"),            # 包装方式
                "packing_size": item.get("packingSize"),   # 单品包装尺寸
                "carton_size": item.get("cartonSize"),     # 外箱尺寸
                "carton_qty": item.get("cartonQty"),       # 装箱数量
                
                # 代发信息
                "agent_price": item.get("agent_price"),
                "agent_fee": item.get("agent_fee"),
                "dropship_support": item.get("is_support_dropship"),
                
                # 供应商
                "supplier": item.get("seller_nick"),
                "supplier_level": item.get("seller_level"),
                "factory_cert": item.get("factory_certification"),
                
                # SKU
                "skus": item.get("skus", []),
                "props_list": item.get("props_list", {}),
                
                # 图片
                "images": item.get("item_imgs", []),
                "desc_images": item.get("desc_imgs", []),
            }
        
        return {"error": data.get("msg", "请求失败")}

六、数据应用场景

6.1 选品分析

Python

复制

def analyze_product(product: ProductDetail) -> dict:
    """
    商品选品分析
    """
    # 计算利润率
    if product.price_ranges and product.price_ranges[0].price > 0:
        cost_price = product.price_ranges[0].price
        suggested_retail = cost_price * 2.5  # 建议零售价
        profit_margin = (suggested_retail - cost_price) / suggested_retail        
        return {
            "product_id": product.product_id,
            "title": product.title,
            "cost_price": cost_price,
            "suggested_retail": suggested_retail,
            "profit_margin": f"{profit_margin*100:.1f}%",
            "moq": product.min_order_quantity,
            "is_dropship": product.support_agent,
            "supplier_level": product.supplier.credit_level,
            "recommendation": "推荐" if profit_margin > 0.5 and product.support_agent else "谨慎"
        }
    
    return {}

6.2 物流成本估算

Python

复制

def estimate_shipping_cost(product: ProductDetail, destination: str) -> dict:
    """
    估算物流成本
    """
    shipping = product.shipping    
    # 解析包装尺寸
    if shipping.package_size:
        dims = [float(x) for x in shipping.package_size.split('x')]
        if len(dims) == 3:
            l, w, h = dims
            volume_weight = (l * w * h) / 5000  # 体积重系数 1:5000
            
            return {
                "actual_weight": shipping.unit_weight,
                "volume_weight": round(volume_weight, 2),
                "chargeable_weight": max(shipping.unit_weight, volume_weight),
                "volume": shipping.volume,
                "destination": destination,
                "estimated_cost": "需调用运费模板 API"
            }
    
    return {}

七、关键注意事项

表格

注意事项	说明
企业认证	1688 官方 API 主要面向企业，个人权限受限
版本选择	2026 年使用 2.0 版本，1.0 不返回核心字段
IP 白名单	必须在开放平台配置请求服务器 IP
字段分散	包装信息等字段分布在多个子结构中，需完整解析
数据更新	商品信息实时变动，建议缓存 5-10 分钟
合规使用	仅用于自有业务分析，不得转售或恶意爬取

八、总结

表格

能力维度	1688 特色	主流电商共性
价格体系	阶梯批发价、代发价、起订量	单件售价、促销价
供应链	工厂信息、诚信通、实力商家	店铺等级、自营标识
交易模式	混批、账期、跨境专供	零售、优惠券、积分
物流	重量体积、运费模板、发货地	快递、自提、次日达

1688 的商品详情 API 是 B2B 电商数据的核心入口，掌握其全字段结构和解析方法，对于构建选品系统、供应链管理平台、跨境采购工具等业务场景至关重要。通过官方 API 与第三方聚合 API 的灵活组合，可以最大化数据获取的完整性和开发效率。

Python 获取 1688 商品采集 API 接口 | 工厂货源自动化对接商品信息 | 无需选品

Wed, 27 May 2026 17:41:14 +0800

在跨境电商和供应链管理中，1688 作为国内最大的 B2B 工厂货源平台，拥有海量商品数据。传统的人工选品方式效率低下、容易出错，而通过 API 接口实现自动化采集，可以彻底解放人力，实现"无需选品"的智能货源对接。本文将深入讲解如何使用 Python 对接 1688 商品采集 API，构建工厂货源自动化在跨境电商和供应链管理中，1688 作为国内最大的 B2B 工厂货源平台，拥有海量商品数据。传统的人工选品方式效率低下、容易出错，而通过 API 接口实现自动化采集，可以彻底解放人力，实现"无需选品"的智能货源对接。本文将深入讲解如何使用 Python 对接 1688 商品采集 API，构建工厂货源自动化系统。

一、方案概述：为什么需要自动化采集？

1.1 传统选品的痛点

表格

痛点	影响	解决方案
手动翻页比价	耗时耗力，效率低下	API 批量采集
数据不及时	错过最佳拿货时机	实时监控更新
信息不完整	SKU、库存、代发价遗漏	全字段自动获取
多平台操作	重复劳动，容易出错	一键铺货对接
反爬限制	IP 封禁，数据中断	官方合规接口

1.2 自动化采集的核心优势

通过 API 接口实现 1688 商品采集，可以：

无需人工选品：设定规则后系统自动筛选符合条件的商品
实时数据同步：价格、库存、销量变动自动更新
全字段获取：标题、价格、SKU、主图、详情、销量、代发价一次性获取
多平台铺货：采集后直接上架到淘宝、拼多多、抖音、跨境平台
7×24 小时监控：自动发现爆款、价格预警、库存提醒

二、1688 商品采集 API 体系

1688 开放平台及第三方数据服务提供了丰富的商品采集接口：

表格

接口	功能	适用场景
`item_get`	获取单个商品详情	深度分析、SKU 监控
`item_search`	按关键词搜索商品	批量选品、类目采集
`item_search_img`	按图搜索商品（拍立淘）	找同款、竞品分析
`item_search_shop`	获取店铺所有商品	整店采集、供应商管理
`item_search_suggest`	获取搜索词推荐	SEO 优化、关键词挖掘
`item_fee`	获取商品快递费用	运费核算、利润计算
`seller_info`	获取店铺详情	供应商资质审核
`item_password`	淘口令解析	链接转换、推广分析
`cat_get`	获取商品分类	类目导航、数据归类

三、环境准备与依赖安装

bash复制

# 安装必要的 Python 库 pip install requests pip install pandas          # 数据处理与导出 pip install openpyxl        # Excel 导出支持 pip install schedule        # 定时任务 pip install python-dotenv   # 环境变量管理 pip install loguru          # 日志记录

四、核心实现：商品采集 API 客户端

4.1 基础 API 客户端

Python复制

# -*- coding: utf-8 -*- """ 1688 商品采集 API 客户端 支持：商品详情获取、关键词搜索、店铺采集、图片搜索 """ import requests import hashlib import time import json import os from typing import List, Dict, Optional, Callable from dataclasses import dataclass, asdict from datetime import datetime from urllib.parse import urlencode from loguru import logger @dataclass class ProductItem:     """1688 商品数据结构"""     item_id: str     title: str     price: float     original_price: float     wholesale_price: float      # 批发价     agent_price: float          # 代发价     min_order: int              # 最小起订量     unit: str     stock: int     sales_30d: int              # 30天销量     total_sales: int     location: str               # 发货地     supplier_name: str     supplier_id: str     supplier_level: str         # 诚信通等级     is_support_mix: bool        # 是否支持混批     is_support_agent: bool      # 是否支持一件代发     main_image: str     images: List[str]     detail_url: str     category: str     attributes: Dict[str, str]  # 商品属性     skus: List[Dict]            # SKU 规格     create_time: str     update_time: str class Alibaba1688Collector:     """     1688 商品采集器     支持官方 API 和第三方数据服务     """          def __init__(self, api_key: str, api_secret: str, base_url: Optional[str] = None):         """         初始化采集器                  Args:             api_key: API 调用 Key             api_secret: API 调用密钥             base_url: API 基础地址（默认使用第三方服务地址）         """         self.api_key = api_key         self.api_secret = api_secret         self.base_url = base_url or "https://api.openclaw.com/1688"         self.session = requests.Session()         self.session.headers.update({             "Accept": "application/json",             "Content-Type": "application/x-www-form-urlencoded;charset=utf-8"         })                  # 请求统计         self.request_count = 0         self.success_count = 0                  logger.info(f"1688 采集器初始化完成 | 接口地址: {self.base_url}")     def _generate_sign(self, params: dict) -> str:         """         生成 API 签名（MD5）         规则：参数按 key 排序后拼接 + secret，整体 MD5 大写         """         sorted_params = sorted(params.items(), key=lambda x: x[0])         param_str = "".join([f"{k}{v}" for k, v in sorted_params])         sign_str = f"{self.api_secret}{param_str}{self.api_secret}"         return hashlib.md5(sign_str.encode('utf-8')).hexdigest().upper()     def _request(self, api_name: str, params: dict, timeout: int = 30) -> dict:         """         发送 API 请求                  Args:             api_name: API 接口名称（如 item_get, item_search）             params: 业务参数             timeout: 请求超时时间                  Returns:             API 响应字典         """         # 构建公共参数         common_params = {             "key": self.api_key,             "api_name": api_name,             "timestamp": int(time.time()),             "format": "json",             "lang": "cn"         }         common_params.update(params)                  # 生成签名         common_params["sign"] = self._generate_sign(common_params)                  url = f"{self.base_url}/{api_name}"         self.request_count += 1                  try:             response = self.session.get(url, params=common_params, timeout=timeout)             response.raise_for_status()                          result = response.json()             if result.get("code") == 0 or result.get("code") == 200:                 self.success_count += 1                 logger.debug(f"请求成功 | API: {api_name} | 耗时: {response.elapsed.total_seconds():.2f}s")                 return result.get("data", result)             else:                 logger.warning(f"请求失败 | API: {api_name} | 错误: {result.get('msg', '未知错误')}")                 return {"error": result.get("msg", "请求失败"), "code": result.get("code")}                          except requests.exceptions.Timeout:             logger.error(f"请求超时 | API: {api_name}")             return {"error": "请求超时", "code": -1}         except requests.exceptions.RequestException as e:             logger.error(f"请求异常 | API: {api_name} | {str(e)}")             return {"error": str(e), "code": -2}     def get_item_detail(self, num_iid: str, get_sales: bool = True,                         get_agent: bool = True) -> Optional[ProductItem]:         """         获取单个商品详情                  Args:             num_iid: 1688 商品数字 ID             get_sales: 是否获取销量数据             get_agent: 是否获取代发价格                  Returns:             ProductItem 对象         """         params = {             "num_iid": num_iid,             "sales_data": 1 if get_sales else 0,             "agent": 1 if get_agent else 0         }                  data = self._request("item_get", params)                  if "error" in data:             return None                      return self._parse_product(data)     def search_by_keyword(self, keyword: str, page: int = 1, page_size: int = 20,                           sort: str = "sales", price_start: Optional[float] = None,                           price_end: Optional[float] = None) -> List[ProductItem]:         """         关键词搜索商品                  Args:             keyword: 搜索关键词             page: 页码             page_size: 每页数量（最大 50）             sort: 排序方式（sales:销量, price_asc:价格升序, price_desc:价格降序, credit:信用）             price_start: 价格下限             price_end: 价格上限                  Returns:             ProductItem 列表         """         params = {             "q": keyword,             "page": page,             "page_size": page_size,             "sort": sort         }                  if price_start is not None:             params["start_price"] = price_start         if price_end is not None:             params["end_price"] = price_end                      data = self._request("item_search", params)                  if "error" in data or "items" not in data:             return []                      return [self._parse_product(item) for item in data.get("items", [])]     def search_by_image(self, image_url: str, page: int = 1) -> List[ProductItem]:         """         按图搜索商品（拍立淘）                  Args:             image_url: 图片 URL             page: 页码                  Returns:             ProductItem 列表         """         params = {             "imgid": image_url,             "page": page         }                  data = self._request("item_search_img", params)                  if "error" in data:             return []                      return [self._parse_product(item) for item in data.get("items", [])]     def get_shop_items(self, seller_id: str, page: int = 1,                         page_size: int = 20) -> List[ProductItem]:         """         获取店铺所有商品                  Args:             seller_id: 卖家 ID（供应商 ID）             page: 页码             page_size: 每页数量                  Returns:             ProductItem 列表         """         params = {             "seller_id": seller_id,             "page": page,             "page_size": page_size         }                  data = self._request("item_search_shop", params)                  if "error" in data:             return []                      return [self._parse_product(item) for item in data.get("items", [])]     def _parse_product(self, data: dict) -> ProductItem:         """解析商品数据为结构化对象"""         return ProductItem(             item_id=str(data.get("num_iid", "")),             title=data.get("title", ""),             price=float(data.get("price", 0) or 0),             original_price=float(data.get("original_price", 0) or 0),             wholesale_price=float(data.get("wholesale_price", 0) or 0),             agent_price=float(data.get("agent_price", 0) or 0),             min_order=int(data.get("min_num", 1) or 1),             unit=data.get("unit", "件"),             stock=int(data.get("stock", 0) or 0),             sales_30d=int(data.get("sales", 0) or 0),             total_sales=int(data.get("total_sales", 0) or 0),             location=data.get("location", ""),             supplier_name=data.get("supplier", ""),             supplier_id=str(data.get("seller_id", "")),             supplier_level=data.get("supplier_level", ""),             is_support_mix=data.get("is_support_mix", False),             is_support_agent=data.get("is_support_agent", False),             main_image=data.get("pic_url", ""),             images=data.get("item_imgs", []),             detail_url=data.get("detail_url", ""),             category=data.get("category", ""),             attributes=data.get("props", {}),             skus=data.get("skus", []),             create_time=data.get("created_time", ""),             update_time=data.get("modified_time", "")         )     def get_stats(self) -> dict:         """获取采集统计"""         return {             "total_requests": self.request_count,             "success_requests": self.success_count,             "success_rate": f"{(self.success_count / max(self.request_count, 1) * 100):.1f}%"         } # ==================== 使用示例 ==================== if __name__ == "__main__":     # 初始化采集器（替换为真实凭证）     collector = Alibaba1688Collector(         api_key="your_api_key",         api_secret="your_api_secret"     )          # 示例 1：获取单个商品详情     print("=" * 60)     print("示例 1：获取商品详情")     print("=" * 60)          item_id = "702356889901"  # 替换为真实商品 ID     product = collector.get_item_detail(item_id)          if product:         print(f"商品ID: {product.item_id}")         print(f"标题: {product.title}")         print(f"售价: ¥{product.price}")         print(f"批发价: ¥{product.wholesale_price}")         print(f"代发价: ¥{product.agent_price}")         print(f"起订量: {product.min_order} {product.unit}")         print(f"库存: {product.stock}")         print(f"30天销量: {product.sales_30d}")         print(f"发货地: {product.location}")         print(f"供应商: {product.supplier_name}")         print(f"支持代发: {'是' if product.is_support_agent else '否'}")         print(f"SKU数量: {len(product.skus)}")     else:         print("获取商品详情失败")

五、自动化选品引擎：无需人工筛选

5.1 智能选品规则引擎

Python复制

from typing import List, Callable from dataclasses import dataclass @dataclass class SelectionRule:     """选品规则配置"""     name: str                           # 规则名称     min_price: float = 0               # 最低价格     max_price: float = float('inf')    # 最高价格     min_sales_30d: int = 0             # 最低月销量     min_stock: int = 0                  # 最低库存     must_support_agent: bool = False    # 必须支持一件代发     must_support_mix: bool = False      # 必须支持混批     supplier_levels: List[str] = None   # 供应商等级要求     exclude_keywords: List[str] = None   # 排除关键词     include_keywords: List[str] = None  # 必须包含关键词     min_profit_margin: float = 0.3      # 最低利润率（基于代发价） class AutoSelector:     """     自动化选品器     根据预设规则自动筛选符合条件的商品     """          def __init__(self, collector: Alibaba1688Collector):         self.collector = collector         self.rules: List[SelectionRule] = []         self.selected_products: List[ProductItem] = []              def add_rule(self, rule: SelectionRule):         """添加选品规则"""         self.rules.append(rule)         logger.info(f"添加选品规则: {rule.name}")              def evaluate_product(self, product: ProductItem, rule: SelectionRule) -> bool:         """         评估商品是否符合规则                  Returns:             bool: 是否符合所有条件         """         # 价格区间检查         if not (rule.min_price <= product.price <= rule.max_price):             return False                      # 销量检查         if product.sales_30d < rule.min_sales_30d:             return False                      # 库存检查         if product.stock < rule.min_stock:             return False                      # 一件代发检查         if rule.must_support_agent and not product.is_support_agent:             return False                      # 混批检查         if rule.must_support_mix and not product.is_support_mix:             return False                      # 供应商等级检查         if rule.supplier_levels and product.supplier_level not in rule.supplier_levels:             return False                      # 关键词排除         if rule.exclude_keywords:             for kw in rule.exclude_keywords:                 if kw.lower() in product.title.lower():                     return False                              # 关键词必须包含         if rule.include_keywords:             for kw in rule.include_keywords:                 if kw.lower() not in product.title.lower():                     return False                              # 利润率估算（简化计算）         if rule.min_profit_margin > 0 and product.agent_price > 0:             estimated_retail = product.price * 1.5  # 假设零售价为售价的 1.5 倍             profit_margin = (estimated_retail - product.agent_price) / estimated_retail             if profit_margin < rule.min_profit_margin:                 return False                          return True              def select_by_keyword(self, keyword: str, max_pages: int = 5) -> List[ProductItem]:         """         根据关键词自动选品                  Args:             keyword: 搜索关键词             max_pages: 最大搜索页数                  Returns:             符合规则的商品列表         """         all_selected = []                  for page in range(1, max_pages + 1):             logger.info(f"正在搜索第 {page} 页 | 关键词: {keyword}")                          products = self.collector.search_by_keyword(                 keyword=keyword,                 page=page,                 page_size=50,                 sort="sales"  # 按销量排序，优先爆款             )                          if not products:                 break                              for product in products:                 # 应用所有规则                 for rule in self.rules:                     if self.evaluate_product(product, rule):                         all_selected.append(product)                         logger.info(f"✅ 选中商品: {product.title[:30]}... | "                                   f"销量:{product.sales_30d} | 价格:¥{product.price}")                         break  # 符合任一规则即可                                      # 限流控制             time.sleep(1)                      self.selected_products.extend(all_selected)         logger.info(f"选品完成 | 共选中 {len(all_selected)} 个商品")         return all_selected              def select_by_supplier(self, seller_id: str) -> List[ProductItem]:         """         整店采集并筛选                  Args:             seller_id: 供应商店铺 ID                  Returns:             符合规则的商品列表         """         all_selected = []         page = 1                  while True:             products = self.collector.get_shop_items(seller_id, page=page, page_size=50)                          if not products:                 break                              for product in products:                 for rule in self.rules:                     if self.evaluate_product(product, rule):                         all_selected.append(product)                         break                                      page += 1             time.sleep(1)                      self.selected_products.extend(all_selected)         return all_selected              def export_to_excel(self, filename: str = "selected_products.xlsx"):         """导出选品结果到 Excel"""         import pandas as pd                  data = []         for p in self.selected_products:             data.append({                 "商品ID": p.item_id,                 "标题": p.title,                 "售价": p.price,                 "批发价": p.wholesale_price,                 "代发价": p.agent_price,                 "起订量": p.min_order,                 "库存": p.stock,                 "30天销量": p.sales_30d,                 "发货地": p.location,                 "供应商": p.supplier_name,                 "支持代发": "是" if p.is_support_agent else "否",                 "链接": p.detail_url             })                      df = pd.DataFrame(data)         df.to_excel(filename, index=False, engine='openpyxl')         logger.info(f"选品结果已导出: {filename} | 共 {len(data)} 条记录") # ==================== 自动化选品示例 ==================== if __name__ == "__main__":     collector = Alibaba1688Collector(         api_key="your_api_key",         api_secret="your_api_secret"     )          selector = AutoSelector(collector)          # 配置选品规则：寻找适合一件代发的低价爆款     selector.add_rule(SelectionRule(         name="一件代发爆款",         min_price=10,         max_price=100,         min_sales_30d=100,         min_stock=50,         must_support_agent=True,         supplier_levels=["实力商家", "超级工厂"],         min_profit_margin=0.4     ))          # 配置规则：寻找高利润混批商品     selector.add_rule(SelectionRule(         name="高利润混批",         min_price=50,         max_price=500,         min_sales_30d=50,         must_support_mix=True,         include_keywords=["定制", "批发"],         min_profit_margin=0.5     ))          # 自动选品     results = selector.select_by_keyword("蓝牙耳机", max_pages=3)          # 导出结果     selector.export_to_excel("bluetooth_earphones_selected.xlsx")

六、定时监控与价格预警系统

Python复制

import schedule import time from datetime import datetime from typing import Dict, Callable class PriceMonitor:     """     价格监控与预警系统     自动监控商品价格、库存变动，触发告警     """          def __init__(self, collector: Alibaba1688Collector):         self.collector = collector         self.monitored_items: Dict[str, dict] = {}  # 监控中的商品         self.alert_handlers: List[Callable] = []         self.price_history: Dict[str, List[tuple]] = {}  # 价格历史              def add_item(self, item_id: str, target_price: Optional[float] = None):         """         添加监控商品                  Args:             item_id: 商品 ID             target_price: 目标价格（低于此价格触发提醒）         """         product = self.collector.get_item_detail(item_id)         if product:             self.monitored_items[item_id] = {                 "product": product,                 "target_price": target_price,                 "last_price": product.price,                 "last_stock": product.stock,                 "added_at": datetime.now()             }             self.price_history[item_id] = [(datetime.now(), product.price)]             logger.info(f"添加监控: {product.title[:30]}... | 当前价: ¥{product.price}")                  def add_alert_handler(self, handler: Callable):         """添加告警处理器"""         self.alert_handlers.append(handler)              def _check_changes(self, item_id: str):         """检查商品变动"""         monitor = self.monitored_items[item_id]         old_product = monitor["product"]                  # 重新获取最新数据         new_product = self.collector.get_item_detail(item_id)         if not new_product:             return                      monitor["product"] = new_product                  alerts = []                  # 价格变动检测         if abs(new_product.price - monitor["last_price"]) > 0.01:             change_pct = (new_product.price - monitor["last_price"]) / monitor["last_price"] * 100             direction = "上涨" if change_pct > 0 else "下降"             alerts.append({                 "type": "price_change",                 "title": f"价格{direction} {abs(change_pct):.1f}%",                 "content": f"{new_product.title[:30]}... | "                           f"¥{monitor['last_price']} → ¥{new_product.price}",                 "item_id": item_id,                 "old_price": monitor["last_price"],                 "new_price": new_product.price             })             monitor["last_price"] = new_product.price             self.price_history[item_id].append((datetime.now(), new_product.price))                      # 库存变动检测         if old_product.stock == 0 and new_product.stock > 0:             alerts.append({                 "type": "restock",                 "title": "商品补货",                 "content": f"{new_product.title[:30]}... | 库存: {new_product.stock}",                 "item_id": item_id             })         elif old_product.stock > 0 and new_product.stock == 0:             alerts.append({                 "type": "out_of_stock",                 "title": "商品缺货",                 "content": f"{new_product.title[:30]}...",                 "item_id": item_id             })                      # 目标价格达成         if monitor["target_price"] and new_product.price <= monitor["target_price"]:             alerts.append({                 "type": "target_price",                 "title": "达到目标价格",                 "content": f"{new_product.title[:30]}... | "                           f"当前¥{new_product.price} ≤ 目标¥{monitor['target_price']}",                 "item_id": item_id,                 "target_price": monitor["target_price"]             })                      # 销量激增检测（日销量 > 100）         daily_sales = new_product.sales_30d - old_product.sales_30d         if daily_sales > 100:             alerts.append({                 "type": "sales_surge",                 "title": "销量激增",                 "content": f"{new_product.title[:30]}... | 日增 {daily_sales} 单",                 "item_id": item_id,                 "daily_sales": daily_sales             })                      # 触发告警         for alert in alerts:             self._trigger_alert(alert)                  def _trigger_alert(self, alert: dict):         """触发告警"""         logger.warning(f"【{alert['title']}】{alert['content']}")         for handler in self.alert_handlers:             try:                 handler(alert)             except Exception as e:                 logger.error(f"告警处理器异常: {e}")                      def check_all(self):         """检查所有监控商品"""         logger.info(f"开始批量检查 | 共 {len(self.monitored_items)} 个商品")         for item_id in list(self.monitored_items.keys()):             self._check_changes(item_id)             time.sleep(2)  # 限流                  def start_monitoring(self, interval_minutes: int = 30):         """         启动定时监控                  Args:             interval_minutes: 检查间隔（分钟）         """         schedule.every(interval_minutes).minutes.do(self.check_all)         logger.info(f"定时监控已启动 | 间隔: {interval_minutes} 分钟")                  while True:             schedule.run_pending()             time.sleep(1) # ==================== 告警处理器示例 ==================== def dingtalk_alert(alert: dict):     """钉钉机器人告警"""     # 实现钉钉 Webhook 发送     pass def email_alert(alert: dict):     """邮件告警"""     # 实现邮件发送     pass def wechat_alert(alert: dict):     """企业微信告警"""     # 实现企业微信发送     pass # ==================== 监控示例 ==================== if __name__ == "__main__":     collector = Alibaba1688Collector(         api_key="your_api_key",         api_secret="your_api_secret"     )          monitor = PriceMonitor(collector)          # 添加监控商品     monitor.add_item("702356889901", target_price=15.0)     monitor.add_item("702356889902", target_price=20.0)          # 添加告警处理器     monitor.add_alert_handler(dingtalk_alert)     monitor.add_alert_handler(wechat_alert)          # 启动监控（每 30 分钟检查一次）     monitor.start_monitoring(interval_minutes=30)

七、多平台铺货对接

Python复制

import json class PlatformPublisher:     """     多平台铺货发布器     将采集的 1688 商品信息转换为各平台格式并发布     """          def __init__(self):         self.platforms = {}              def register_platform(self, name: str, publisher: Callable):         """注册平台发布器"""         self.platforms[name] = publisher              def publish(self, product: ProductItem, platforms: List[str]) -> Dict[str, bool]:         """         发布商品到指定平台                  Args:             product: 商品信息             platforms: 目标平台列表                  Returns:             各平台发布结果         """         results = {}         for platform in platforms:             if platform in self.platforms:                 try:                     self.platforms[platform](product)                     results[platform] = True                     logger.info(f"发布成功 | 平台: {platform} | 商品: {product.title[:30]}")                 except Exception as e:                     results[platform] = False                     logger.error(f"发布失败 | 平台: {platform} | {e}")             else:                 results[platform] = False                 logger.warning(f"未注册平台: {platform}")         return results              def generate_taobao_format(self, product: ProductItem) -> dict:         """生成淘宝上架格式"""         return {             "title": product.title,             "price": product.price * 1.5,  # 加价 50%             "original_price": product.price * 2,             "quantity": min(product.stock, 100),             "category": self._map_category(product.category, "taobao"),             "images": product.images[:5],             "detail": self._generate_detail_html(product),             "express_fee": 0  # 包邮         }              def generate_pdd_format(self, product: ProductItem) -> dict:         """生成拼多多上架格式"""         return {             "goods_name": product.title,             "market_price": int(product.price * 2),             "price": int(product.price * 1.3),             "quantity": min(product.stock, 1000),             "carousel_gallery": product.images[:10],             "detail_gallery": product.images,             "category_id": self._map_category(product.category, "pdd")         }              def generate_douyin_format(self, product: ProductItem) -> dict:         """生成抖音小店上架格式"""         return {             "name": product.title[:30],  # 抖音标题限制 30 字             "pic": product.main_image,             "description": product.title,             "price": int(product.price * 1.4),             "stock_num": product.stock,             "category_id": self._map_category(product.category, "douyin"),             "specs": self._convert_skus(product.skus)         }              def _map_category(self, category: str, platform: str) -> str:         """类目映射（简化版，实际需维护映射表）"""         mappings = {             "taobao": {"数码": "14", "服装": "16"},             "pdd": {"数码": "8424", "服装": "12896"},             "douyin": {"数码": "20001", "服装": "20005"}         }         return mappings.get(platform, {}).get(category, "")              def _generate_detail_html(self, product: ProductItem) -> str:         """生成详情页 HTML"""         html = f"<h1>{product.title}</h1>"         html += f"<p>发货地: {product.location}</p>"         html += f"<p>供应商: {product.supplier_name}</p>"         for img in product.images:             html += f'<img src="{img}" />'         return html              def _convert_skus(self, skus: List[dict]) -> List[dict]:         """转换 SKU 格式"""         return [{             "spec_id": s.get("sku_id", ""),             "spec_name": s.get("properties", ""),             "price": s.get("price", 0),             "stock_num": s.get("quantity", 0)         } for s in skus] # ==================== 铺货示例 ==================== if __name__ == "__main__":     publisher = PlatformPublisher()          # 注册各平台发布器（实际需实现具体 API 调用）     publisher.register_platform("taobao", lambda p: print(f"发布到淘宝: {p.title}"))     publisher.register_platform("pdd", lambda p: print(f"发布到拼多多: {p.title}"))     publisher.register_platform("douyin", lambda p: print(f"发布到抖音: {p.title}"))          # 模拟商品     product = ProductItem(         item_id="123456",         title="测试商品",         price=10.0,         original_price=20.0,         wholesale_price=8.0,         agent_price=12.0,         min_order=1,         unit="件",         stock=100,         sales_30d=50,         total_sales=500,         location="义乌",         supplier_name="测试供应商",         supplier_id="789",         supplier_level="实力商家",         is_support_mix=True,         is_support_agent=True,         main_image="http://example.com/1.jpg",         images=["http://example.com/1.jpg", "http://example.com/2.jpg"],         detail_url="http://1688.com/item/123456",         category="数码",         attributes={},         skus=[],         create_time="",         update_time=""     )          # 发布到多平台     results = publisher.publish(product, ["taobao", "pdd", "douyin"])     print(f"发布结果: {results}")

八、完整自动化流程整合

Python复制

class AutoSourcingPipeline:     """     自动化货源对接流水线     实现从采集、选品、监控到铺货的全流程自动化     """          def __init__(self, api_key: str, api_secret: str):         self.collector = Alibaba1688Collector(api_key, api_secret)         self.selector = AutoSelector(self.collector)         self.monitor = PriceMonitor(self.collector)         self.publisher = PlatformPublisher()              def run_full_pipeline(         self,         keywords: List[str],         rules: List[SelectionRule],         target_platforms: List[str],         monitor_interval: int = 30     ):         """         运行完整自动化流程                  Args:             keywords: 搜索关键词列表             rules: 选品规则列表             target_platforms: 目标铺货平台             monitor_interval: 监控间隔（分钟）         """         logger.info("=" * 60)         logger.info("启动 1688 自动化货源对接流水线")         logger.info("=" * 60)                  # 步骤 1：配置选品规则         for rule in rules:             self.selector.add_rule(rule)                      # 步骤 2：批量采集与选品         all_selected = []         for keyword in keywords:             logger.info(f"正在采集关键词: {keyword}")             selected = self.selector.select_by_keyword(keyword, max_pages=5)             all_selected.extend(selected)                      # 去重         unique_items = {p.item_id: p for p in all_selected}         all_selected = list(unique_items.values())                  logger.info(f"选品完成 | 共 {len(all_selected)} 个唯一商品")                  # 步骤 3：添加到监控系统         for product in all_selected:             self.monitor.add_item(product.item_id)                      # 步骤 4：铺货到目标平台         publish_results = []         for product in all_selected:             results = self.publisher.publish(product, target_platforms)             publish_results.append({                 "item_id": product.item_id,                 "title": product.title,                 "results": results             })                      # 步骤 5：启动价格监控         self.monitor.start_monitoring(monitor_interval)                  return {             "total_selected": len(all_selected),             "publish_results": publish_results         } # ==================== 完整运行示例 ==================== if __name__ == "__main__":     # 初始化流水线     pipeline = AutoSourcingPipeline(         api_key="your_api_key",         api_secret="your_api_secret"     )          # 配置选品规则     rules = [         SelectionRule(             name="一件代发低价爆款",             min_price=5,             max_price=50,             min_sales_30d=200,             must_support_agent=True,             min_profit_margin=0.4         ),         SelectionRule(             name="高利润定制款",             min_price=100,             max_price=500,             min_sales_30d=50,             include_keywords=["定制", "logo"],             min_profit_margin=0.5         )     ]          # 运行完整流程     result = pipeline.run_full_pipeline(         keywords=["蓝牙耳机", "手机壳", "数据线"],         rules=rules,         target_platforms=["taobao", "pdd"],         monitor_interval=30     )          print(f"\n流水线运行完成 | 选中 {result['total_selected']} 个商品")

九、关键注意事项

表格

注意事项	说明
API 凭证安全	`api_secret` 严禁硬编码，使用环境变量或密钥管理服务
请求频率控制	默认限流 50 次/分钟，超出会返回 429 错误
数据合规使用	仅用于自有业务分析，不得转售或恶意爬取
商品信息更新	1688 商品信息可能实时变动，建议定时刷新
供应商沟通	一件代发需与供应商确认库存和发货时效
多语言支持	跨境场景可设置 `lang=en` 获取英文数据

十、总结

表格

功能模块	核心能力	实现方式
商品采集	详情/搜索/店铺/图片搜索	`item_get` / `item_search` / `item_search_shop`
智能选品	自动筛选符合条件的商品	`AutoSelector` 规则引擎
价格监控	实时价格/库存变动告警	`PriceMonitor` 定时检查
多平台铺货	一键上架到淘宝/拼多多/抖音	`PlatformPublisher` 格式转换
数据导出	Excel/CSV 批量导出	`pandas` + `openpyxl`

通过这套 Python 自动化方案，你可以实现真正的"无需选品"——设定规则后，系统自动从 1688 工厂货源中筛选、监控、铺货，大幅降低人工成本，提升电商运营效率。

系统。

如遇任何疑问或有进一步的需求，请随时与我私信或者评论联系。

万邦api博客

主流电商平台（淘宝/1688/京东/拼多多）商品比价与数据分析实战指南

一、四大平台价格特征与定位差异

点击测试获取key

二、数据采集：比价系统的基石

2.1 三种主流采集方式

2.2 各平台官方 API 对比

三、核心比价数据分析模型

3.1 价格带分析

3.2 价格弹性分析

3.3 促销周期价格分析

四、比价系统的技术架构

4.1 系统架构设计

4.2 核心代码：多平台比价引擎

五、实战应用场景

5.1 品牌方控价

5.2 选品定价决策

5.3 跨境代采决策

六、数据可视化与报表

6.1 核心监控指标

6.2 可视化看板设计

七、常见问题与解决方案

八、总结

一文通，第三方接口如何实现批量上货，主流平台[淘宝|京东|1688|抖音）和跨境平台

一、国内主流平台批量上货接口

1. 淘宝 / 天猫（TOP 开放平台）

2. 京东（JOS 开放平台）

3. 拼多多（PDD Open API）

4. 1688（阿里巴巴开放平台）

5. 抖音电商（抖店开放平台）

二、跨境平台批量上货接口

1. Shopee / Lazada

2. TikTok Shop

3. 亚马逊（SP-API）

三、批量上货的技术实现方案

方案一：官方 API 直连（推荐）点击测试

方案二：第三方 ERP / SaaS 对接

方案三：RPA 模拟人工（特殊场景）

四、批量上货的关键注意事项

五、推荐接入路径

小白必看！用AI+现成工具实现1688选品监控自动化

一、先泼一盆冷水：为什么普通人玩不了官方API？

二、方案总览：三条零技术路线，总有一条适合你

三、方案A：1688官方AI工具（完全免费，开箱即用）

3.1 1688 AI版App — 你的随身选品助手

3.2 1688诚信通AI版 — 商家端智能体（适合有店铺的）

四、方案B：第三方AI选品插件（跨境电商必备）点击测试

4.1 店雷达 — 1688站内选品神器

4.2 Sorftime — 跨境全链路AI选品（进阶推荐）

五、方案C：AI助手+现成接口服务（深度自动化）

5.1 什么是"现成接口服务"？

5.2 典型使用场景

5.3 完全零代码的实现方式

5.4 避坑指南

六、三种方案怎么选？一张图看懂

七、立即行动：今天就能开始的3个步骤

Step 1：下载1688 AI版App（5分钟）

Step 2：安装一个选品插件（10分钟）

Step 3：设置你的第一个监控（15分钟）

八、常见问题FAQ

九、写在最后

1688图搜+关键词搜索商品API：从入门到实战，让你效率翻倍

一、概述

二、准备工作：账号与权限

2.1 注册1688开放平台账号

2.2 创建应用获取凭证

2.3 申请API权限

三、签名生成机制

3.1 签名规则

3.2 Python签名工具类

四、关键词搜索商品API实战

4.1 接口概述

4.2 核心参数说明

4.3 完整调用代码

五、图搜（以图搜货/拍立淘）API实战 点击测试

5.1 接口概述

5.2 核心参数说明

5.3 完整调用代码

六、高级实战：关键词+图搜组合策略

6.1 组合搜索场景

五、图搜（以图搜货/拍立淘）API实战点击测试

二、官方 API 接口详解：`alibaba.product.get`