开发者指南：利用 1688API 构建商品信息采集系统

pip install requests python-dotenv pandas

import time
import hashlib
import hmac
import json
import requests
from dotenv import load_dotenv
import os

# 加载环境变量（建议将敏感信息存入.env文件）
load_dotenv()
APP_KEY = os.getenv("APP_KEY")
APP_SECRET = os.getenv("APP_SECRET")
API_URL = "https://gw.open.1688.com/openapi/param2/2.0/"

class Ali1688Client:
    def __init__(self, app_key, app_secret):
        self.app_key = app_key
        self.app_secret = app_secret
        self.session = requests.Session()

    def _sign(self, params):
        """生成API签名"""
        sorted_params = sorted(params.items(), key=lambda x: x[0])
        sign_str = self.app_secret + ''.join([f"{k}{v}" for k, v in sorted_params]) + self.app_secret
        return hmac.new(
            self.app_secret.encode('utf-8'),
            sign_str.encode('utf-8'),
            hashlib.sha1
        ).hexdigest().upper()

    def call_api(self, api_name, params=None):
        """通用API调用方法"""
        if not params:
            params = {}
            
        # 公共参数
        public_params = {
            "app_key": self.app_key,
            "format": "json",
            "v": "2.0",
            "timestamp": time.strftime("%Y-%m-%d %H:%M:%S"),
            "sign_method": "hmac-sha1"
        }
        
        # 合并参数
        all_params = {**public_params,** params}
        all_params["sign"] = self._sign(all_params)
        
        # 发送请求
        try:
            response = self.session.get(f"{API_URL}{api_name}", params=all_params, timeout=10)
            result = json.loads(response.text)
            if "error_response" in result:
                print(f"API错误: {result['error_response']['msg']}")
                return None
            return result
        except Exception as e:
            print(f"请求异常: {str(e)}")
            return None

import pandas as pd
from datetime import datetime

class ProductCollector:
    def __init__(self, client):
        self.client = client
        self.data_path = "product_data"
        os.makedirs(self.data_path, exist_ok=True)

    def search_products(self, keyword, page=1, page_size=40):
        """搜索商品"""
        params = {
            "keywords": keyword,
            "page_no": page,
            "page_size": page_size,
            "sort": "new_desc"  # 按新品排序
        }
        return self.client.call_api("alibaba.item.search", params)

    def get_product_detail(self, product_id):
        """获取商品详情"""
        params = {
            "offer_id": product_id,
            "fields": "offerId,title,picUrl,price,minPrice,maxPrice,"
                      "tradePrice,quantity,categoryId,sellerId,"
                      "province,city,viewCount,saleCount"
        }
        return self.client.call_api("alibaba.item.offer.get", params)

    def batch_collect(self, keyword, total_pages=5):
        """批量采集商品数据"""
        all_products = []
        
        for page in range(1, total_pages + 1):
            print(f"采集第{page}页数据...")
            search_result = self.search_products(keyword, page)
            
            if not search_result or "result" not in search_result:
                continue
                
            for item in search_result["result"]["toReturn"]:
                # 获取商品详情
                detail = self.get_product_detail(item["offerId"])
                if detail and "result" in detail:
                    product_info = detail["result"]
                    all_products.append({
                        "id": product_info.get("offerId"),
                        "title": product_info.get("title"),
                        "price": product_info.get("price"),
                        "min_price": product_info.get("minPrice"),
                        "max_price": product_info.get("maxPrice"),
                        "sale_count": product_info.get("saleCount"),
                        "view_count": product_info.get("viewCount"),
                        "location": f"{product_info.get('province')}-{product_info.get('city')}",
                        "pic_url": product_info.get("picUrl"),
                        "collect_time": datetime.now().strftime("%Y-%m-%d %H:%M:%S")
                    })
            
            # 控制请求频率，避免触发限流
            time.sleep(2)
        
        # 保存数据
        df = pd.DataFrame(all_products)
        filename = f"{self.data_path}/{keyword}_{datetime.now().strftime('%Y%m%d')}.csv"
        df.to_csv(filename, index=False, encoding="utf-8-sig")
        print(f"数据采集完成，共{len(all_products)}条，已保存至{filename}")
        return df

if __name__ == "__main__":
    # 初始化客户端
    client = Ali1688Client(APP_KEY, APP_SECRET)
    
    # 初始化采集器
    collector = ProductCollector(client)
    
    # 批量采集"笔记本电脑"商品数据（5页）
    collector.batch_collect("笔记本电脑", total_pages=5)