×

1688 item_get_app 接口深度分析及 Python 实现

万邦科技Lex 万邦科技Lex 发表于2025-09-08 10:22:37 浏览247 评论0

抢沙发发表评论

                     注册账号免费测试1688API数据接口

1688 平台的 item_get_app 接口是获取商品原始详情数据的核心接口,专门针对移动端应用场景设计。与普通的 item_get 接口相比,它返回的数据结构更贴近 1688 APP 端展示的原始格式,包含更丰富的 B2B 场景特有字段,如批发价格、起订量、供应商信息等,对采购决策和供应链分析具有重要价值。

一、接口核心特性分析

1. 接口功能与定位

  • 核心功能:获取 1688 商品的原始详情数据,包括商品基础信息、价格体系(含批发价)、规格参数、供应商信息、交易数据等

  • 数据特点

    • 与 1688 APP 端数据结构一致,保留移动端特有的展示字段

    • 包含 B2B 特有的批发价格、起订量、混批规则等信息

    • 提供供应商详细信息,如经营年限、交易量、认证资质等

    • 包含实时库存、发货地、物流模板等供应链关键数据

  • 应用场景

    • 采购决策支持系统

    • 供应商评估与筛选

    • 市场价格监测

    • 竞品分析与比较

    • 供应链优化与风险控制

2. 认证机制

1688 开放平台采用 appkey + access_token 的认证方式:


  • 开发者在 1688 开放平台注册应用,获取 appkey 和 appsecret

  • 通过 appkey 和 appsecret 获取 access_token(通常有效期为 24 小时)

  • 每次接口调用需携带有效 access_token

  • item_get_app 属于高级接口,需要单独申请权限,企业账号权限更高

3. 核心参数与响应结构

请求参数

参数名类型是否必填说明
offer_idString商品 ID(1688 中称为 offer_id)
access_tokenString访问令牌
member_idString采购商 ID,用于获取个性化价格
provinceString省份名称,用于获取区域化价格和物流信息
fieldsString需要返回的字段,默认返回全部字段

响应核心字段

  • 商品基础信息:商品 ID、名称、标题、详情描述、类目信息

  • 价格体系:批发价、零售价、阶梯价格、混批规则、折扣信息

  • 规格参数:SKU 规格、属性组合、起订量、库存

  • 供应商信息:供应商 ID、名称、经营年限、交易量、认证信息

  • 物流信息:发货地、运费模板、预计发货时间

  • 交易数据:30 天成交量、买家数、重复采购率

  • 多媒体信息:主图、详情图、视频、规格图

二、Python 脚本实现

以下是调用 1688 item_get_app 接口的完整 Python 实现,包含令牌获取、接口调用、数据解析及 B2B 场景特有分析功能:
import requests
import time
import json
import logging
import re
from typing import Dict, Optional, List
from requests.exceptions import RequestException

# 配置日志
logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s - %(levelname)s - %(message)s"
)

class Alibaba1688ItemGetAppAPI:
    def __init__(self, appkey: str, appsecret: str):
        """
        初始化1688商品详情API客户端
        :param appkey: 1688开放平台appkey
        :param appsecret: 1688开放平台appsecret
        """
        self.appkey = appkey
        self.appsecret = appsecret
        self.base_url = "https://gw.open.1688.com/openapi"
        self.access_token = None
        self.token_expires_at = 0  # token过期时间戳
        self.session = requests.Session()
        self.session.headers.update({
            "Content-Type": "application/x-www-form-urlencoded",
            "User-Agent": "AlibabaApp/9.1.0 (iPhone; iOS 14.4; Scale/2.00)"
        })

    def _get_access_token(self) -> Optional[str]:
        """获取访问令牌"""
        # 检查token是否有效
        if self.access_token and self.token_expires_at > time.time() + 60:
            return self.access_token
            
        logging.info("获取新的access_token")
        params = {
            "method": "alibaba.oauth2.getToken",
            "client_id": self.appkey,
            "client_secret": self.appsecret,
            "grant_type": "client_credentials",
            "format": "json"
        }
        
        try:
            response = self.session.get(f"{self.base_url}/gateway.do", params=params, timeout=10)
            response.raise_for_status()
            result = response.json()
            
            if "error_response" in result:
                logging.error(f"获取access_token失败: {result['error_response']['msg']} (错误码: {result['error_response']['code']})")
                return None
                
            self.access_token = result["access_token"]
            self.token_expires_at = time.time() + result.get("expires_in", 86400)  # 默认为24小时
            return self.access_token
                
        except RequestException as e:
            logging.error(f"获取access_token请求异常: {str(e)}")
            return None

    def get_item_raw_data(self, 
                         offer_id: str, 
                         member_id: Optional[str] = None,
                         province: Optional[str] = None,
                         fields: Optional[str] = None) -> Optional[Dict]:
        """
        获取商品原始详情数据
        :param offer_id: 商品ID(1688中称为offer_id)
        :param member_id: 采购商ID
        :param province: 省份名称
        :param fields: 需要返回的字段
        :return: 商品原始数据
        """
        # 获取有效的access_token
        if not self._get_access_token():
            return None
            
        params = {
            "method": "alibaba.item.get.app",
            "client_id": self.appkey,
            "access_token": self.access_token,
            "offer_id": offer_id,
            "format": "json",
            "v": "1.0",
            "timestamp": time.strftime("%Y-%m-%d %H:%M:%S")
        }
        
        # 可选参数
        if member_id:
            params["member_id"] = member_id
        if province:
            params["province"] = province
        if fields:
            params["fields"] = fields
            
        try:
            response = self.session.get(f"{self.base_url}/gateway.do", params=params, timeout=20)
            response.raise_for_status()
            result = response.json()
            
            if "error_response" in result:
                logging.error(f"获取商品数据失败: {result['error_response']['msg']} (错误码: {result['error_response']['code']})")
                return None
                
            item_response = result.get("alibaba_item_get_app_response", {})
            item_data = item_response.get("result", {})
            
            if not item_data:
                logging.warning("未获取到商品数据")
                return None
                
            # 格式化原始数据
            return self._process_raw_data(item_data)
            
        except RequestException as e:
            logging.error(f"获取商品数据请求异常: {str(e)}")
            return None
        except json.JSONDecodeError:
            logging.error(f"商品数据响应解析失败: {response.text[:200]}...")
            return None

    def _process_raw_data(self, raw_data: Dict) -> Dict:
        """处理原始数据,提取关键信息并格式化"""
        # 基础信息提取
        base_info = {
            "offer_id": raw_data.get("offer_id"),
            "title": raw_data.get("title"),
            "sub_title": raw_data.get("sub_title"),
            "detail_url": raw_data.get("detail_url"),
            "category": {
                "cid": raw_data.get("category_id"),
                "name": raw_data.get("category_name"),
                "level1_cid": raw_data.get("level1_category_id"),
                "level1_name": raw_data.get("level1_category_name"),
                "level2_cid": raw_data.get("level2_category_id"),
                "level2_name": raw_data.get("level2_category_name")
            },
            "tags": raw_data.get("tags", "").split(","),
            "creation_time": raw_data.get("creation_time"),
            "modify_time": raw_data.get("modify_time")
        }
        
        # 价格信息提取(B2B特有的阶梯价格体系)
        price_info = {
            "retail_price": self._safe_float(raw_data.get("retail_price")),  # 零售价
            "wholesale_prices": self._format_wholesale_prices(raw_data.get("wholesale_price_list", [])),  # 批发价列表
            "mix_batch": {  # 混批规则
                "support": raw_data.get("support_mix_batch", False),
                "min_amount": self._safe_float(raw_data.get("mix_batch_min_amount")),
                "min_quantity": self._safe_int(raw_data.get("mix_batch_min_quantity"))
            },
            "currency": raw_data.get("currency", "CNY"),
            "price_unit": raw_data.get("price_unit", "个")
        }
        
        # 库存与规格信息提取
        sku_info = {
            "total_sku": self._safe_int(raw_data.get("total_sku")),
            "skus": self._format_skus(raw_data.get("skus", [])),
            "specs": self._format_specs(raw_data.get("specs", [])),
            "total_stock": self._safe_int(raw_data.get("total_stock")),
            "sales_count": self._safe_int(raw_data.get("sales_count_30d")),  # 30天销量
            "unit": raw_data.get("unit", "个")
        }
        
        # 供应商信息提取(B2B核心信息)
        supplier_info = {
            "supplier_id": raw_data.get("supplier_id"),
            "supplier_name": raw_data.get("supplier_name"),
            "company_name": raw_data.get("company_name"),
            "years": self._safe_int(raw_data.get("operating_years")),  # 经营年限
            "main_products": raw_data.get("main_products", "").split(";"),
            "location": raw_data.get("location"),
            "transaction": {
                "turnover": self._safe_float(raw_data.get("annual_turnover")),  # 年交易额
                "rating": self._safe_float(raw_data.get("supplier_rating")),  # 供应商评分
                "repeat_rate": self._safe_float(raw_data.get("repeat_purchase_rate")),  # 重复采购率
                "buyer_count": self._safe_int(raw_data.get("buyer_count_30d"))  # 30天买家数
            },
            "authentication": {
                "real_name": raw_data.get("real_name_authentication", False),
                "company": raw_data.get("company_authentication", False),
                "gold_supplier": raw_data.get("is_gold_supplier", False),  # 是否金牌供应商
                "assessed_supplier": raw_data.get("is_assessed_supplier", False)  # 是否实力商家
            },
            "contact": {
                "phone": raw_data.get("contact_phone"),
                "online_status": raw_data.get("online_status")
            }
        }
        
        # 物流信息提取
        logistics_info = {
            "delivery_location": raw_data.get("delivery_location"),  # 发货地
            "freight_template": raw_data.get("freight_template_name"),
            "delivery_time": raw_data.get("delivery_time"),  # 发货时间
            "transport_modes": raw_data.get("transport_modes", "").split(","),  # 运输方式
            "min_delivery_days": self._safe_int(raw_data.get("min_delivery_days")),
            "max_delivery_days": self._safe_int(raw_data.get("max_delivery_days"))
        }
        
        # 多媒体信息提取
        media_info = {
            "main_images": raw_data.get("main_image_list", []),
            "detail_images": raw_data.get("detail_image_list", []),
            "video_url": raw_data.get("video_url"),
            "video_cover": raw_data.get("video_cover")
        }
        
        # 详情描述提取
        description = {
            "detail": self._clean_html(raw_data.get("detail", "")),
            "short_description": raw_data.get("short_description")
        }
        
        return {
            "base_info": base_info,
            "price_info": price_info,
            "sku_info": sku_info,
            "supplier_info": supplier_info,
            "logistics_info": logistics_info,
            "media_info": media_info,
            "description": description,
            "raw_data": raw_data  # 保留原始数据
        }
    
    def _safe_float(self, value) -> float:
        """安全转换为float"""
        try:
            return float(value) if value is not None else 0.0
        except (ValueError, TypeError):
            return 0.0
    
    def _safe_int(self, value) -> int:
        """安全转换为int"""
        try:
            return int(value) if value is not None else 0
        except (ValueError, TypeError):
            return 0
    
    def _format_wholesale_prices(self, prices: List[Dict]) -> List[Dict]:
        """格式化批发价格列表(阶梯价格)"""
        formatted = []
        for price in prices:
            formatted.append({
                "min_quantity": self._safe_int(price.get("min_quantity")),
                "max_quantity": self._safe_int(price.get("max_quantity")),
                "price": self._safe_float(price.get("price")),
                "discount": self._safe_float(price.get("discount"))  # 折扣
            })
        # 按起订量排序
        return sorted(formatted, key=lambda x: x["min_quantity"])
    
    def _format_skus(self, skus: List[Dict]) -> List[Dict]:
        """格式化SKU列表"""
        formatted = []
        for sku in skus:
            # 处理SKU的阶梯价格
            sku_prices = []
            if sku.get("price_list"):
                for p in sku.get("price_list"):
                    sku_prices.append({
                        "min_quantity": self._safe_int(p.get("min_quantity")),
                        "price": self._safe_float(p.get("price"))
                    })
            
            formatted.append({
                "sku_id": sku.get("sku_id"),
                "specs": sku.get("specs", []),  # 规格组合
                "price": self._safe_float(sku.get("price")),
                "wholesale_prices": sku_prices,  # SKU级别的阶梯价格
                "stock": self._safe_int(sku.get("stock")),
                "sales_count": self._safe_int(sku.get("sales_count")),
                "image_url": sku.get("image_url")
            })
        return formatted
    
    def _format_specs(self, specs: List[Dict]) -> List[Dict]:
        """格式化规格参数列表"""
        formatted = []
        for spec in specs:
            formatted.append({
                "name": spec.get("name"),
                "values": [
                    {
                        "name": val.get("name"),
                        "image_url": val.get("image_url"),
                        "spec_id": val.get("spec_id")
                    } for val in spec.get("values", [])
                ]
            })
        return formatted
    
    def _clean_html(self, html: str) -> str:
        """清理HTML内容,提取纯文本"""
        if not html:
            return ""
        # 去除HTML标签
        clean = re.sub(r'<.*?>', ' ', html)
        # 去除多余空格和换行
        clean = re.sub(r'\s+', ' ', clean).strip()
        return clean

    def analyze_b2b_value(self, item_data: Dict) -> Dict:
        """分析商品的B2B采购价值,生成评估报告"""
        if not item_data:
            return {}
            
        # 价格优势评分(0-10分)
        price_advantage = 0
        if item_data["price_info"]["wholesale_prices"]:
            # 基于批发价与零售价的差异计算
            retail_price = item_data["price_info"]["retail_price"]
            lowest_wholesale = item_data["price_info"]["wholesale_prices"][0]["price"]
            
            if retail_price > 0 and lowest_wholesale > 0:
                discount_rate = (retail_price - lowest_wholesale) / retail_price
                price_advantage = min(10, round(discount_rate * 10))
        
        # 供应商可靠性评分(0-10分)
        supplier_score = 0
        supplier = item_data["supplier_info"]
        if supplier["authentication"]["company"]:
            supplier_score += 3
        if supplier["authentication"]["gold_supplier"]:
            supplier_score += 2
        if supplier["authentication"]["assessed_supplier"]:
            supplier_score += 2
        if supplier["years"] >= 3:
            supplier_score += 2
        if supplier["transaction"]["repeat_rate"] > 30:
            supplier_score += 1
        
        # 采购灵活性评分(0-10分)
        flexibility_score = 0
        if item_data["price_info"]["mix_batch"]["support"]:
            flexibility_score += 5
        # 基于起订量评估
        min_order = item_data["price_info"]["mix_batch"]["min_quantity"] or float('inf')
        if min_order <= 5:
            flexibility_score += 5
        elif min_order <= 20:
            flexibility_score += 3
        elif min_order <= 100:
            flexibility_score += 1
        
        # 物流评分(0-10分)
        logistics_score = 0
        logistics = item_data["logistics_info"]
        if logistics["min_delivery_days"] <= 2:
            logistics_score += 5
        elif logistics["min_delivery_days"] <= 5:
            logistics_score += 3
        if len(logistics["transport_modes"]) >= 2:
            logistics_score += 3
        if logistics["freight_template"]:
            logistics_score += 2
        
        # 综合评分
        overall_score = round((price_advantage + supplier_score + flexibility_score + logistics_score) / 4, 1)
        
        return {
            "overall_score": overall_score,
            "price_advantage": {
                "score": price_advantage,
                "description": f"批发价最低为零售价的{round(item_data['price_info']['wholesale_prices'][0]['price']/item_data['price_info']['retail_price']*100, 1)}%" if item_data["price_info"]["retail_price"] > 0 else ""
            },
            "supplier_reliability": {
                "score": supplier_score,
                "description": f"{supplier['years']}年经营,重复采购率{supplier['transaction']['repeat_rate']}%"
            },
            "purchase_flexibility": {
                "score": flexibility_score,
                "description": f"{'支持' if item_data['price_info']['mix_batch']['support'] else '不支持'}混批,最低起订量{item_data['price_info']['mix_batch']['min_quantity'] or '未知'}"
            },
            "logistics_capability": {
                "score": logistics_score,
                "description": f"发货地{logistics['delivery_location']},{logistics['min_delivery_days']}-{logistics['max_delivery_days']}天送达"
            },
            "risk_assessment": {
                "high_risk": overall_score < 3,
                "medium_risk": 3 <= overall_score < 6,
                "low_risk": overall_score >= 6
            }
        }


# 示例调用
if __name__ == "__main__":
    # 替换为实际的appkey和appsecret(从1688开放平台获取)
    APPKEY = "your_appkey"
    APPSECRET = "your_appsecret"
    # 替换为目标商品offer_id
    OFFER_ID = "61234567890"
    
    # 初始化API客户端
    api = Alibaba1688ItemGetAppAPI(APPKEY, APPSECRET)
    
    # 获取商品原始数据
    item_data = api.get_item_raw_data(
        offer_id=OFFER_ID,
        # member_id="your_member_id",  # 可选,采购商ID
        # province="浙江省",  # 可选,省份
        # fields="offer_id,title,price,stock"  # 可选,指定需要的字段
    )
    
    if item_data:
        print(f"=== 1688商品详情 (offer_id: {OFFER_ID}) ===")
        print(f"商品名称: {item_data['base_info']['title']}")
        print(f"类目: {item_data['base_info']['category']['level1_name']} > {item_data['base_info']['category']['level2_name']}")
        print(f"供应商: {item_data['supplier_info']['supplier_name']} ({item_data['supplier_info']['company_name']})")
        print(f"经营年限: {item_data['supplier_info']['years']}年")
        print(f"30天销量: {item_data['sku_info']['sales_count']}件")
        print(f"30天买家数: {item_data['supplier_info']['transaction']['buyer_count']}")
        print(f"重复采购率: {item_data['supplier_info']['transaction']['repeat_rate']}%")
        
        # 价格信息
        print("\n价格信息:")
        print(f"  零售价: {item_data['price_info']['retail_price']}元/{item_data['price_info']['price_unit']}")
        print("  批发价:")
        for i, wholesale in enumerate(item_data['price_info']['wholesale_prices'], 1):
            max_qty = f"-{wholesale['max_quantity']}" if wholesale['max_quantity'] else "+"
            print(f"    {wholesale['min_quantity']}{max_qty}件: {wholesale['price']}元/{item_data['price_info']['price_unit']}")
        
        # 混批信息
        print("\n采购规则:")
        mix_batch = item_data['price_info']['mix_batch']
        print(f"  混批: {'支持' if mix_batch['support'] else '不支持'}")
        if mix_batch['support']:
            print(f"    最低金额: {mix_batch['min_amount']}元")
            print(f"    最低数量: {mix_batch['min_quantity']}件")
        
        # 物流信息
        print("\n物流信息:")
        print(f"  发货地: {item_data['logistics_info']['delivery_location']}")
        print(f"  预计发货时间: {item_data['logistics_info']['delivery_time']}")
        print(f"  运输方式: {', '.join(item_data['logistics_info']['transport_modes'])}")
        print(f"  预计送达时间: {item_data['logistics_info']['min_delivery_days']}-{item_data['logistics_info']['max_delivery_days']}天")
        
        # 规格信息
        if item_data['sku_info']['skus']:
            print("\n规格信息:")
            for i, sku in enumerate(item_data['sku_info']['skus'][:3], 1):
                specs = ", ".join([f"{s['name']}:{s['value']}" for s in sku['specs']]) if sku['specs'] else "标准"
                print(f"  {i}. {specs}: {sku['price']}元, 库存{sku['stock']}件")
        
        # B2B价值分析
        b2b_analysis = api.analyze_b2b_value(item_data)
        print("\n=== B2B采购价值评估 ===")
        print(f"综合评分: {b2b_analysis['overall_score']}/10分")
        print(f"价格优势: {b2b_analysis['price_advantage']['score']}/10分 {b2b_analysis['price_advantage']['description']}")
        print(f"供应商可靠性: {b2b_analysis['supplier_reliability']['score']}/10分 {b2b_analysis['supplier_reliability']['description']}")
        print(f"采购灵活性: {b2b_analysis['purchase_flexibility']['score']}/10分 {b2b_analysis['purchase_flexibility']['description']}")
        print(f"物流能力: {b2b_analysis['logistics_capability']['score']}/10分 {b2b_analysis['logistics_capability']['description']}")
        print(f"风险评估: {'高风险' if b2b_analysis['risk_assessment']['high_risk'] else '中等风险' if b2b_analysis['risk_assessment']['medium_risk'] else '低风险'}")

三、接口调用注意事项

1. 调用限制与规范

  • QPS 限制:1688 开放平台对商品详情接口的 QPS 限制通常为 5-10 次 / 秒

  • 权限差异:企业认证账号比个人账号能获取更详细的供应商数据和价格信息

  • 数据缓存:建议缓存获取的数据(缓存时间 30-60 分钟),减少重复调用

  • 地区差异:不同地区可能有不同的物流政策和价格,需正确设置 province 参数

  • 字段权限:部分敏感字段(如供应商联系方式)需要额外申请权限

2. 常见错误及解决方案

错误码说明解决方案
401未授权或 token 无效重新获取 access_token,检查权限是否正确
403权限不足升级账号类型,申请商品详情接口的完整权限
404商品不存在确认 offer_id 是否正确,1688 商品 ID 通常为 10-12 位数字
429调用频率超限降低调用频率,实现请求限流
500服务器内部错误实现重试机制,最多 3 次,间隔指数退避
110参数错误检查 offer_id 格式是否正确

3. 数据解析要点

  • 价格体系:1688 采用阶梯价格体系,需特别处理不同采购量对应的价格

  • 供应商评估:重点关注经营年限、重复采购率、认证资质等供应商指标

  • 库存状态:部分商品显示的是总库存,部分显示 SKU 级库存,需区分处理

  • 混批规则:混批支持情况和具体规则对采购决策至关重要

  • 物流信息:发货地和物流模板直接影响采购成本和到货时间

四、应用场景与扩展建议

典型应用场景

  • 智能采购系统:基于商品详情数据和供应商评估实现自动化采购决策

  • 供应商管理平台:整合商品数据和供应商信息,构建供应商评估体系

  • 价格监测工具:实时监控商品价格变化,把握最佳采购时机

  • 竞品分析系统:对比同类商品的价格、规格、供应商等信息

  • 供应链优化工具:基于物流信息和供应商分布优化采购布局

扩展建议

  • 实现价格趋势分析:定期获取价格数据,分析价格波动规律

  • 开发供应商匹配系统:基于采购需求自动匹配最合适的供应商

  • 构建采购风险评估模型:结合供应商信息和商品数据评估采购风险

  • 实现多地区价格对比:分析不同地区的价格和物流成本差异

  • 开发批量采购计算器:根据阶梯价格计算最优采购量和总成本

  • 构建商品相似度匹配:基于规格参数和描述实现同类商品自动匹配


通过合理使用 1688 item_get_app 接口,开发者可以构建针对 B2B 场景的商品分析和采购决策系统,充分利用 1688 平台的批发特性,为企业采购提供数据支持。使用时需特别关注 B2B 场景特有的阶梯价格、起订量、供应商评估等维度,以获取更有价值的分析结果。


群贤毕至

访客