1688 包装运费尺寸接口开发实战指南

sign = MD5(app_secret + key1value1 + key2value2 + ... + app_secret).toUpperCase()

import time
import hashlib
import urllib.parse
import requests
import json
from typing import Dict, Optional
class Official1688API:
    """
    1688 官方开放平台 API 客户端
    用于获取商品包装、运费、尺寸等基础信息
    """
    
    GATEWAY_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
    
    def _generate_sign(self, params: dict) -> str:
        """
        生成 1688 官方 API 的 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]
        )
        
        # 拼接签名字符串
        sign_str = self.app_secret
        for key, value in sorted_params:
            encoded_value = urllib.parse.quote(str(value), safe='')
            sign_str += f"{key}{encoded_value}"
        sign_str += self.app_secret
        
        # MD5 加密并转大写
        return hashlib.md5(sign_str.encode('utf-8')).hexdigest().upper()
    
    def get_product_shipping_info(self, product_id: str, fields: str = None) -> Dict:
        """
        获取商品包装、运费、尺寸信息
        
        :param product_id: 1688 商品 ID
        :param fields: 指定返回字段，如 "shippingInfo,saleInfo,attributes"
        :return: 解析后的包装信息
        """
        timestamp = time.strftime("%Y-%m-%d %H:%M:%S", time.localtime())
        
        params = {
            "app_key": self.app_key,
            "method": "com.alibaba.product.alibaba.product.get",
            "timestamp": timestamp,
            "v": "2.0",
            "format": "json",
            "sign_method": "md5",
            "access_token": self.access_token,
            "productId": product_id,
        }
        
        if fields:
            params["fields"] = fields
        
        # 生成签名
        params["sign"] = self._generate_sign(params)
        
        try:
            response = requests.post(self.GATEWAY_URL, data=params, timeout=30)
            response.raise_for_status()
            data = response.json()
            
            # 解析包装相关字段
            return self._parse_shipping_data(data, product_id)
            
        except requests.exceptions.RequestException as e:
            return {
                "success": False,
                "error": f"请求异常: {str(e)}",
                "product_id": product_id
            }
    
    def _parse_shipping_data(self, raw_data: dict, product_id: str) -> Dict:
        """
        从官方 API 返回数据中解析包装、运费、尺寸信息
        """
        product_info = raw_data.get("productinfo", {})
        shipping = product_info.get("shippingInfo", {})
        sale_info = product_info.get("saleInfo", {})
        attributes = product_info.get("attributes", [])
        
        # 从 attributes 中提取包装相关属性
        packaging_attrs = {}
        for attr in attributes:
            attr_name = attr.get("attributeName", "")
            if any(kw in attr_name for kw in ["包装", "重量", "体积", "尺寸", "运费"]):
                packaging_attrs[attr_name] = attr.get("value")
        
        return {
            "success": True,
            "source": "official_api",
            "product_id": product_id,
            "product_title": product_info.get("subject"),
            
            # 核心包装/尺寸字段
            "unit_weight_kg": shipping.get("unitWeight"),           # 单位重量（千克）
            "package_size_cm": shipping.get("packageSize"),         # 包装尺寸（长x宽x高）
            "volume_cm3": shipping.get("volume"),                   # 体积（立方厘米）
            
            # 运费相关
            "freight_template_id": shipping.get("freightTemplateID"),  # 运费模板 ID
            "send_goods_address": shipping.get("sendGoodsAddress"),    # 发货地址
            
            # 销售信息
            "min_order_quantity": sale_info.get("minOrderQuantity"),   # 最小起订量
            "unit": sale_info.get("unit"),                             # 计量单位
            "price_ranges": sale_info.get("priceRanges"),              # 阶梯价格
            
            # 扩展属性
            "packaging_attributes": packaging_attrs,
            "raw_data": raw_data
        }
# ==================== 使用示例 ====================
if __name__ == "__main__":
    api = Official1688API(
        app_key="your_app_key",
        app_secret="your_app_secret",
        access_token="your_access_token"
    )
    
    result = api.get_product_shipping_info(
        product_id="619899292404",
        fields="productID,subject,shippingInfo,saleInfo,attributes"
    )
    
    print(json.dumps(result, ensure_ascii=False, indent=2))

import requests
import re
from typing import Dict, Optional
class ThirdParty1688API:
    """
    1688 第三方聚合 API 客户端
    支持 AliPrice、VV-Tool 等多种数据源
    """
    
    def __init__(self, api_key: str, provider: str = "aliprice"):
        self.api_key = api_key
        self.provider = provider
        
        # 不同提供商的 API 端点
        self.endpoints = {
            "aliprice": "https://www.aliprice.com/items/itemDetail",
            "vv_tool": "http://api.vv-tool.com/tool/erps/product1688get",
            "o0b_cn": "https://o0b.cn/ibrad/1688/item_get/"
        }
    
    def get_full_packaging_info(self, product_id: str) -> Dict:
        """
        获取完整的包装、运费、尺寸信息
        """
        try:
            if self.provider == "aliprice":
                data = self._call_aliprice(product_id)
            elif self.provider == "vv_tool":
                data = self._call_vv_tool(product_id)
            else:
                data = self._call_generic(product_id)
            
            return self._normalize_packaging_data(data, product_id)
            
        except Exception as e:
            return {
                "success": False,
                "error": str(e),
                "product_id": product_id
            }
    
    def _call_aliprice(self, product_id: str) -> dict:
        """调用 AliPrice API"""
        headers = {
            "Authorization": f"Bearer {self.api_key}",
            "Content-Type": "application/json"
        }
        payload = {
            "offerId": product_id,
            "includePackaging": True,
            "includeSku": True
        }
        response = requests.post(
            self.endpoints["aliprice"],
            headers=headers,
            json=payload,
            timeout=30
        )
        response.raise_for_status()
        return response.json()
    
    def _call_vv_tool(self, product_id: str) -> dict:
        """调用 VV-Tool API"""
        headers = {"Authorization": f"Bearer {self.api_key}"}
        params = {"productId": product_id}
        response = requests.get(
            self.endpoints["vv_tool"],
            headers=headers,
            params=params,
            timeout=30
        )
        response.raise_for_status()
        return response.json()
    
    def _normalize_packaging_data(self, raw_data: dict, product_id: str) -> Dict:
        """
        标准化不同来源的包装数据
        """
        # 适配不同响应结构
        item = raw_data.get("item", raw_data)
        
        # 提取重量（统一转换为 kg）
        net_weight = self._extract_weight(item.get("weight") or item.get("unitWeight"))
        gross_weight = self._extract_weight(item.get("grossWeight"))
        
        # 提取尺寸
        unit_size = item.get("packingSize") or item.get("packageSize")
        carton_size = item.get("cartonSize")
        
        # 提取运费
        post_fee = item.get("post_fee", 0)
        express_fee = item.get("express_fee", 0)
        ems_fee = item.get("ems_fee", 0)
        
        # 计算物流指标
        logistics = {}
        if carton_size and gross_weight:
            logistics = self._calculate_logistics_metrics(carton_size, gross_weight)
        
        return {
            "success": True,
            "source": f"third_party_{self.provider}",
            "product_id": product_id,
            "title": item.get("title"),
            
            # 重量信息
            "net_weight_kg": net_weight,
            "gross_weight_kg": gross_weight,
            
            # 尺寸信息
            "unit_package_size": unit_size,      # 单品包装尺寸，如 "10*20*5cm"
            "carton_dimensions": carton_size,    # 外箱尺寸，如 "50*40*30cm"
            "volume_cm3": item.get("volume"),
            
            # 包装方式
            "packing_type": item.get("packing") or item.get("packagingType"),
            "quantity_per_carton": item.get("cartonQty"),
            
            # 运费信息
            "shipping_fees": {
                "post_fee": post_fee,           # 普通邮费
                "express_fee": express_fee,     # 快递费
                "ems_fee": ems_fee,             # EMS 费用
                "shipping_to": item.get("shipping_to")  # 发货至
            },
            
            # 物流分析
            "logistics_analysis": logistics,
            
            "raw_data": raw_data
        }
    
    @staticmethod
    def _extract_weight(weight_val) -> Optional[float]:
        """从字符串中提取数值型重量，统一为 kg"""
        if not weight_val:
            return None
        if isinstance(weight_val, (int, float)):
            return float(weight_val)
        
        # 处理字符串如 "0.8kg"、"1.2KG"、"800g"
        match = re.search(r'(\d+\.?\d*)', str(weight_val))
        if not match:
            return None
        
        value = float(match.group(1))
        weight_str = str(weight_val).lower()
        
        if 'g' in weight_str and 'kg' not in weight_str:
            value = value / 1000  # 克转千克
        
        return round(value, 3)
    
    @staticmethod
    def _calculate_logistics_metrics(carton_size: str, actual_weight: float) -> Dict:
        """
        基于包装尺寸和重量计算物流指标
        """
        try:
            # 解析尺寸（格式：50*40*30cm 或 50x40x30）
            dims = [float(x) for x in re.split(r'[xX*×]', carton_size.replace('cm', '').strip())]
            if len(dims) != 3:
                return {}
            
            length, width, height = dims
            
            # 计算体积重（快递行业标准：体积重 = 长*宽*高 / 5000）
            volume_weight = (length * width * height) / 5000
            
            # 计费重量 = max(体积重, 实际重量)
            chargeable_weight = max(volume_weight, actual_weight)
            
            return {
                "carton_length_cm": length,
                "carton_width_cm": width,
                "carton_height_cm": height,
                "carton_volume_m3": round(length * width * height / 1000000, 4),
                "volume_weight_kg": round(volume_weight, 2),
                "actual_weight_kg": actual_weight,
                "chargeable_weight_kg": round(chargeable_weight, 2),
                "dimensional_factor": "1:5000",
                "suitable_shipping": "express" if chargeable_weight < 20 else "freight",
                "stackable": height < 60  # 高度小于60cm适合堆叠
            }
        except Exception:
            return {}
# ==================== 使用示例 ====================
if __name__ == "__main__":
    api = ThirdParty1688API(
        api_key="your_third_party_key",
        provider="aliprice"
    )
    
    result = api.get_full_packaging_info("1234567890")
    
    if result["success"]:
        print(f"商品: {result['title']}")
        print(f"净重: {result['net_weight_kg']} kg")
        print(f"毛重: {result['gross_weight_kg']} kg")
        print(f"包装方式: {result['packing_type']}")
        print(f"单品尺寸: {result['unit_package_size']}")
        print(f"外箱尺寸: {result['carton_dimensions']}")
        print(f"邮费: ¥{result['shipping_fees']['post_fee']}")
        
        logistics = result.get("logistics_analysis", {})
        if logistics:
            print(f"\n物流分析:")
            print(f"  体积重: {logistics['volume_weight_kg']} kg")
            print(f"  计费重量: {logistics['chargeable_weight_kg']} kg")
            print(f"  推荐运输方式: {'快递' if logistics['suitable_shipping'] == 'express' else '货运'}")
    else:
        print(f"获取失败: {result['error']}")

class LogisticsOrderAPI:
    """
    1688 物流订单查询 API
    用于获取已成交订单的实际物流信息
    """
    
    GATEWAY_URL = "https://gw.open.1688.com/openapi/param2/1/com.alibaba.logistics/alibaba.trade.getLogisticsInfos.buyerView"
    
    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
    
    def get_logistics_info(self, order_id: str, fields: str = None) -> Dict:
        """
        获取订单物流详情
        
        :param order_id: 1688 订单号
        :param fields: 返回字段，如 "company.name,sender,receiver,sendgood"
        """
        from official_api import Official1688API  # 复用签名方法
        
        timestamp = time.strftime("%Y-%m-%d %H:%M:%S", time.localtime())
        
        params = {
            "app_key": self.app_key,
            "method": "com.alibaba.logistics.alibaba.trade.getLogisticsInfos.buyerView",
            "timestamp": timestamp,
            "v": "1",
            "format": "json",
            "sign_method": "md5",
            "access_token": self.access_token,
            "orderId": order_id,
            "webSite": "1688"
        }
        
        if fields:
            params["fields"] = fields
        
        # 生成签名（复用 Official1688API 的签名逻辑）
        signer = Official1688API(self.app_key, self.app_secret, self.access_token)
        params["sign"] = signer._generate_sign(params)
        
        try:
            response = requests.post(self.GATEWAY_URL, data=params, timeout=30)
            response.raise_for_status()
            data = response.json()
            
            if data.get("success"):
                return {
                    "success": True,
                    "order_id": order_id,
                    "logistics_list": data.get("result", []),
                    "raw_data": data
                }
            else:
                return {
                    "success": False,
                    "error": data.get("errorMessage"),
                    "error_code": data.get("errorCode")
                }
                
        except requests.exceptions.RequestException as e:
            return {
                "success": False,
                "error": f"请求异常: {str(e)}"
            }

class SmartFreightCalculator:
    """
    1688 智能运费计算引擎
    整合官方 API + 第三方 API，自动选择最优数据源
    """
    
    def __init__(self, official_api: Official1688API = None, 
                 third_party_api: ThirdParty1688API = None):
        self.official = official_api
        self.third_party = third_party_api
    
    def calculate_freight(self, product_id: str, quantity: int = 1, 
                        destination: str = None) -> Dict:
        """
        计算商品运费
        
        :param product_id: 1688 商品 ID
        :param quantity: 购买数量
        :param destination: 目的地（用于运费模板匹配）
        """
        # 优先使用第三方 API（数据更完整）
        if self.third_party:
            packaging = self.third_party.get_full_packaging_info(product_id)
            if packaging.get("success"):
                return self._calculate_from_third_party(packaging, quantity, destination)
        
        # 回退到官方 API
        if self.official:
            shipping = self.official.get_product_shipping_info(product_id)
            if shipping.get("success"):
                return self._calculate_from_official(shipping, quantity, destination)
        
        return {"success": False, "error": "无可用数据源"}
    
    def _calculate_from_third_party(self, data: dict, quantity: int, destination: str) -> Dict:
        """基于第三方 API 数据计算运费"""
        fees = data.get("shipping_fees", {})
        logistics = data.get("logistics_analysis", {})
        
        # 基础运费（取邮费或快递费）
        base_fee = fees.get("express_fee", 0) or fees.get("post_fee", 0)
        
        # 按数量计算（假设阶梯运费）
        total_fee = base_fee * max(1, quantity / (data.get("quantity_per_carton") or 1))
        
        # 按重量计算（快递续重）
        chargeable_weight = logistics.get("chargeable_weight_kg", 0) * quantity
        weight_fee = self._estimate_weight_fee(chargeable_weight)
        
        return {
            "success": True,
            "product_id": data["product_id"],
            "quantity": quantity,
            "destination": destination,
            "freight_details": {
                "base_fee": round(base_fee, 2),
                "weight_fee": round(weight_fee, 2),
                "estimated_total": round(max(total_fee, weight_fee), 2)
            },
            "packaging_info": {
                "unit_weight": data.get("net_weight_kg"),
                "total_weight": round(data.get("gross_weight_kg", 0) * quantity, 2),
                "chargeable_weight": round(chargeable_weight, 2)
            }
        }
    
    @staticmethod
    def _estimate_weight_fee(weight_kg: float) -> float:
        """估算续重运费（简化模型）"""
        if weight_kg <= 1:
            return 8.0  # 首重
        return 8.0 + (weight_kg - 1) * 4.0  # 续重 4元/kg
# ==================== 完整使用示例 ====================
if __name__ == "__main__":
    # 初始化 API 客户端
    official = Official1688API(
        app_key="your_app_key",
        app_secret="your_app_secret",
        access_token="your_token"
    )
    
    third_party = ThirdParty1688API(
        api_key="your_third_party_key",
        provider="aliprice"
    )
    
    # 创建运费计算器
    calculator = SmartFreightCalculator(official, third_party)
    
    # 计算运费
    result = calculator.calculate_freight(
        product_id="1234567890",
        quantity=100,
        destination="广东省深圳市"
    )
    
    print(json.dumps(result, ensure_ascii=False, indent=2))