1688 店铺全量商品接口实战：从 memberId 解析、分页优化到数据完整性闭环

import time
import hashlib
import requests
import json
import redis
from urllib.parse import quote
from typing import Dict, List, Optional
class AlibabaSellerItemAPI:
    def __init__(self, app_key: str, app_secret: str):
        self.app_key = app_key
        self.app_secret = app_secret
        self.api_url = "https://gw.open.1688.com/openapi/param2/2/portals.open/api/"
        self.session = self._init_session()
        # 缓存memberId与SKU数据（1688解析成本高，缓存24小时）
        self.redis = redis.Redis(host='localhost', port=6379, db=2)
        self.cache_expire = 86400
    def _init_session(self) -> requests.Session:
        """初始化会话池：早年没做连接池，并发时频繁断连，现在稳定多了"""
        session = requests.Session()
        adapter = requests.adapters.HTTPAdapter(
            pool_connections=20, pool_maxsize=100, max_retries=3
        )
        session.mount('https://', adapter)
        return session
    def _generate_sign(self, params: Dict) -> str:
        """生成1688签名：关键坑点——参数要URL编码，中文不编码必错"""
        # 1. 过滤空值，按ASCII升序排序
        valid_params = {k: v for k, v in params.items() if v is not None}
        sorted_params = sorted(valid_params.items(), key=lambda x: x[0])
        # 2. 拼接并URL编码：1688要求比淘宝严格，每个值都要编码
        query_str = '&'.join([f'{k}={quote(str(v), safe="")}' for k, v in sorted_params])
        # 3. 首尾加secret，MD5加密转大写
        sign_str = self.app_secret + query_str + self.app_secret
        return hashlib.md5(sign_str.encode('utf-8')).hexdigest().upper()
    def get_member_id_by_url(self, shop_url: str) -> Optional[str]:
        """从店铺URL解析memberId：早年手动复制常错，封装后准确率100%"""
        cache_key = f"shop_url:{shop_url}"
        if cached_id := self.redis.get(cache_key):
            return cached_id.decode()
        
        # 1688店铺URL有3种格式，需适配解析
        if "memberId=" in shop_url:
            member_id = shop_url.split("memberId=")[1].split("&")[0]
        elif "shop/" in shop_url:
            member_id = shop_url.split("shop/")[1].split(".")[0]
        else:
            # 复杂URL调用解析接口
            params = {
                "method": "alibaba.shop.get",
                "app_key": self.app_key,
                "timestamp": str(int(time.time() * 1000)),  # 13位毫秒级
                "format": "json",
                "v": "2.0",
                "shop_url": shop_url
            }
            params["sign"] = self._generate_sign(params)
            try:
                response = self.session.get(self.api_url, params=params, timeout=(5, 15))
                result = response.json()
                if "error_response" in result:
                    print(f"解析失败: {result['error_response']['msg']}")
                    return None
                member_id = result["shop_get_response"]["shop"]["member_id"]
            except Exception as e:
                print(f"解析异常: {str(e)}")
                return None
        
        self.redis.setex(cache_key, self.cache_expire, member_id)
        return member_id

from concurrent.futures import ThreadPoolExecutor, as_completed
def _fetch_page_items(self, member_id: str, page_num: int, start_time: str = None) -> List[Dict]:
    """拉取单页商品：处理1688分页超时与空数据"""
    params = {
        "method": "alibaba.seller.items.list.get",
        "app_key": self.app_key,
        "timestamp": str(int(time.time() * 1000)),
        "format": "json",
        "v": "2.0",
        "member_id": member_id,
        "pageNum": page_num,
        "pageSize": 40,  # 1688最大40，别改大
        "offline": "false",  # 只拉在售商品
        "fields": "item_id,title,wholesalePrice,moq,sales,stock,skuIds,modified_time"
    }
    # 按更新时间切片，解决50页限制
    if start_time:
        params["start_time"] = start_time
    params["sign"] = self._generate_sign(params)
    try:
        response = self.session.get(self.api_url, params=params, timeout=(8, 20))
        result = response.json()
        if "error_response" in result:
            err_msg = result["error_response"]["msg"]
            print(f"分页{page_num}错误: {err_msg}")
            # 1001参数错直接返回，5002系统忙重试
            return [] if "1001" in err_msg else None
        # 解析商品，补全SKU规格
        raw_items = result.get("seller_items_list_get_response", {}).get("items", {}).get("item", [])
        if not raw_items:
            return []
        # 补全SKU详情（1688主接口不含SKU规格）
        for item in raw_items:
            sku_list = []
            for sku_id in item["skuIds"].split(","):
                if sku_detail := self._get_sku_detail(sku_id):
                    sku_list.append(sku_detail)
            item["sku_list"] = sku_list
        return raw_items
    except Exception as e:
        print(f"分页{page_num}异常: {str(e)}")
        return None
def get_all_shop_items(self, shop_identifier: str, is_url: bool = True) -> List[Dict]:
    """全量拉取：按类目+时间分段，突破50页限制"""
    member_id = shop_identifier if not is_url else self.get_member_id_by_url(shop_identifier)
    if not member_id:
        return []
    
    # 先获取店铺类目，按类目分段拉取
    categories = self._get_shop_categories(member_id)
    all_items = []
    
    # 4线程最优（测过6线程会触发限流）
    with ThreadPoolExecutor(max_workers=4) as executor:
        futures = [executor.submit(self._fetch_category_items, member_id, cat["cid"]) 
                   for cat in categories]
        for future in as_completed(futures):
            if category_items := future.result():
                all_items.extend(category_items)
    
    # 去重（跨类目可能有重复）
    seen_ids = set()
    return [item for item in all_items if (item_id := item.get("item_id")) not in seen_ids and not seen_ids.add(item_id)]
def _fetch_category_items(self, member_id: str, cid: str) -> List[Dict]:
    """拉取单个类目的所有商品，处理50页限制"""
    items = []
    page_num = 1
    max_page = 50
    last_modified = None
    
    while page_num <= max_page:
        # 重试3次（1688偶尔抽风）
        retry = 0
        page_items = None
        while retry < 3:
            # 到50页且有数据，用最后修改时间续拉
            if page_num == max_page and last_modified:
                page_items = self._fetch_page_items(member_id, 1, last_modified)
                if page_items:
                    items.extend(page_items)
                    page_num = 1
                    last_modified = items[-1]["modified_time"]
                    continue
            page_items = self._fetch_page_items(member_id, page_num)
            if page_items is not None:
                break
            time.sleep(2)
            retry += 1
        
        if not page_items:
            break
        items.extend(page_items)
        last_modified = items[-1]["modified_time"]
        page_num += 1
        time.sleep(0.8)  # 控制频率，避免限流
    return items

def verify_item_completeness(self, member_id: str, fetched_items: List[Dict]) -> Dict:
    """三重校验：官方计数+类目总和+SKU完整性"""
    # 1. 调用1688计数接口拿官方总数
    official_count = 0
    try:
        params = {
            "method": "alibaba.seller.items.count.get",
            "app_key": self.app_key,
            "timestamp": str(int(time.time() * 1000)),
            "format": "json",
            "v": "2.0",
            "member_id": member_id,
            "offline": "false"
        }
        params["sign"] = self._generate_sign(params)
        response = self.session.get(self.api_url, params=params, timeout=(5, 10))
        result = response.json()
        if "error_response" not in result:
            official_count = result["seller_items_count_get_response"]["total_count"]
    except Exception as e:
        print(f"计数接口异常: {str(e)}")
    # 2. 校验SKU完整性（B2B商品无SKU占比不能超3%）
    no_sku_count = sum(1 for item in fetched_items if not item.get("sku_list"))
    sku_complete_rate = 1 - (no_sku_count / len(fetched_items)) if fetched_items else 0
    # 3. 校验批发价完整性（必填字段不能缺失）
    price_missing = sum(1 for item in fetched_items if not item.get("wholesalePrice"))
    price_complete_rate = 1 - (price_missing / len(fetched_items)) if fetched_items else 0
    # 结果返回：允许5个误差，双率≥97%算合格
    fetched_count = len(fetched_items)
    return {
        "fetched_count": fetched_count,
        "official_count": official_count,
        "sku_complete_rate": round(sku_complete_rate * 100, 1),
        "price_complete_rate": round(price_complete_rate * 100, 1),
        "is_complete": (abs(fetched_count - official_count) <= 5 
                        and sku_complete_rate >= 0.97 
                        and price_complete_rate >= 0.97)
    }

if __name__ == "__main__":
    # 初始化客户端（替换成自己的key和secret）
    alibaba_api = AlibabaSellerItemAPI("your_app_key", "your_app_secret")
    
    # 1. 全量拉取店铺商品（传入店铺URL或memberId）
    print("===== 全量拉取商品 =====")
    shop_url = "https://shop12345678.1688.com"
    all_items = alibaba_api.get_all_shop_items(shop_url, is_url=True)
    print(f"拉取商品总数: {len(all_items)}")
    
    # 2. 完整性校验
    print("\n===== 数据完整性校验 =====")
    member_id = alibaba_api.get_member_id_by_url(shop_url)
    verify_res = alibaba_api.verify_item_completeness(member_id, all_items)
    print(f"官方总数: {verify_res['official_count']} | 拉取数: {verify_res['fetched_count']}")
    print(f"SKU完整率: {verify_res['sku_complete_rate']}% | 价格完整率: {verify_res['price_complete_rate']}%")
    print(f"是否完整: {'是' if verify_res['is_complete'] else '否'}")
    
    # 3. 打印示例商品（带B2B核心数据）
    if all_items:
        print("\n===== 示例商品数据 =====")
        sample = all_items[0]
        print(f"商品ID: {sample['item_id']} | 标题: {sample['title']}")
        print(f"批发价: {sample['wholesalePrice']}元 | 起订量: {sample['moq']}件")
        print(f"SKU数量: {len(sample['sku_list'])} | 销量: {sample['sales']}件")