1688 平台关键字搜索接口深度开发：从精准筛选到供应链数据分析

import timeimport hashlibimport urllib.parsedef generate_1688_sign(params: dict, app_secret: str) -> str:
    """
    生成1688接口签名
    :param params: 请求参数字典
    :param app_secret: 应用密钥
    :return: 签名字符串
    """
    # 1. 按参数名ASCII升序排序
    sorted_params = sorted(params.items(), key=lambda x: x[0])
    
    # 2. 拼接为key=value&key=value格式
    query_string = urllib.parse.urlencode(sorted_params)
    
    # 3. 拼接appsecret并计算MD5
    sign_str = f"{query_string}{app_secret}"
    sign = hashlib.md5(sign_str.encode('utf-8')).hexdigest().upper()
    
    return sign

import requestsimport jsonfrom typing import Dict, Optional, Any, Listclass AlibabaProductSearchAPI:
    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.0/"
        
    def search_products(self, 
                       keywords: str,
                       page: int = 1,
                       page_size: int = 20,** kwargs) -> Optional[Dict[str, Any]]:
        """
        搜索1688商品
        :param keywords: 搜索关键字
        :param page: 页码
        :param page_size: 每页条数
        :param kwargs: 其他筛选参数
        :return: 搜索结果字典
        """
        # 1. 构建基础参数
        params = {
            "method": "alibaba.product.search",
            "app_key": self.app_key,
            "timestamp": time.strftime("%Y-%m-%d %H:%M:%S"),
            "format": "json",
            "v": "2.0",
            "keywords": keywords,
            "page": page,
            "page_size": page_size,
            **kwargs        }
        
        # 2. 生成签名
        params["sign"] = generate_1688_sign(params, self.app_secret)
        
        try:
            # 3. 发送GET请求
            response = requests.get(
                self.api_url,
                params=params,
                timeout=15
            )
            
            # 4. 处理响应
            response.raise_for_status()
            result = json.loads(response.text)
            
            # 5. 错误处理
            if "error_response" in result:
                error = result["error_response"]
                print(f"接口错误: {error.get('msg')} (错误码: {error.get('code')})")
                return None
                
            return result.get("alibaba_product_search_response", {}).get("result", {})
            
        except requests.exceptions.RequestException as e:
            print(f"HTTP请求异常: {str(e)}")
            return None
        except json.JSONDecodeError as e:
            print(f"JSON解析错误: {str(e)}")
            return None

import pandas as pddef parse_search_results(raw_result: Dict) -> tuple[Optional[pd.DataFrame], Optional[Dict]]:
    """
    解析搜索结果为结构化数据
    :param raw_result: 接口返回的原始数据
    :return: 商品DataFrame和统计信息
    """
    if not raw_result or "products" not in raw_result:
        return None, None
        
    products = []
    for item in raw_result["products"]:
        supplier = item.get("supplier", {})
        transaction = item.get("transaction", {})
        capability = item.get("capability", {})
        
        product_info = {
            # 商品基础信息
            "product_id": item.get("product_id"),
            "title": item.get("title"),
            "price": item.get("price"),
            "price_unit": item.get("price_unit", "件"),
            "image_url": item.get("image_url"),
            
            # 批发特性
            "min_order": item.get("min_order", 1),
            "mix_enable": item.get("mix_enable", False),
            
            # 供应商信息
            "supplier_id": supplier.get("user_id"),
            "company_name": supplier.get("company_name"),
            "supplier_type": supplier.get("supplier_type"),
            "credit_level": supplier.get("credit_level"),
            "trust_year": supplier.get("year", 0),
            
            # 交易数据
            "sales_count": transaction.get("sales_count", 0),
            "satisfaction": transaction.get("satisfaction", 0),
            
            # 供应链能力
            "production_capacity": capability.get("production_capacity", ""),
            "customize_support": capability.get("customize_support", False)
        }
        products.append(product_info)
    
    # 转换为DataFrame
    df = pd.DataFrame(products)
    
    # 数据类型转换
    if not df.empty:
        df["price"] = pd.to_numeric(df["price"], errors="coerce")
        df["min_order"] = pd.to_numeric(df["min_order"], errors="coerce")
        df["sales_count"] = pd.to_numeric(df["sales_count"], errors="coerce")
        df["satisfaction"] = pd.to_numeric(df["satisfaction"], errors="coerce")
        df["trust_year"] = pd.to_numeric(df["trust_year"], errors="coerce")
    
    # 统计信息
    stats = {
        "total_results": raw_result.get("total_results", 0),
        "page": raw_result.get("page", 1),
        "page_size": raw_result.get("page_size", 20),
        "total_pages": (raw_result.get("total_results", 0) + raw_result.get("page_size", 20) - 1) 
                      // raw_result.get("page_size", 20)
    }
    
    return df, stats

def filter_suppliers(df: pd.DataFrame, **filters) -> pd.DataFrame:
    """
    高级筛选供应商商品
    :param df: 商品DataFrame
    :param filters: 筛选条件，如min_trust_year=3, min_satisfaction=95等
    :return: 筛选后的DataFrame
    """
    if df.empty:
        return df
        
    filtered_df = df.copy()
    
    # 应用筛选条件
    if "min_trust_year" in filters:
        filtered_df = filtered_df[filtered_df["trust_year"] >= filters["min_trust_year"]]
        
    if "min_satisfaction" in filters:
        filtered_df = filtered_df[filtered_df["satisfaction"] >= filters["min_satisfaction"]]
        
    if "min_sales" in filters:
        filtered_df = filtered_df[filtered_df["sales_count"] >= filters["min_sales"]]
        
    if "supplier_types" in filters:
        filtered_df = filtered_df[filtered_df["supplier_type"].isin(filters["supplier_types"])]
        
    if "customize_required" in filters and filters["customize_required"]:
        filtered_df = filtered_df[filtered_df["customize_support"] == True]
        
    return filtered_df.reset_index(drop=True)def analyze_supply_chain(df: pd.DataFrame) -> Dict:
    """
    供应链数据分析
    :param df: 商品DataFrame
    :return: 供应链分析结果
    """
    if df.empty:
        return {}
        
    # 价格分布
    price_stats = {
        "avg_price": round(df["price"].mean(), 2),
        "min_price": df["price"].min(),
        "max_price": df["price"].max(),
        "price_range": f"{df['price'].min()}-{df['price'].max()}"
    }
    
    # 供应商类型分布
    supplier_type_dist = df["supplier_type"].value_counts().to_dict()
    
    # 起订量分析
    order_stats = {
        "avg_min_order": round(df["min_order"].mean(), 1),
        "min_order_pct": {
            "≤10": round(len(df[df["min_order"] <= 10])/len(df)*100, 1),
            "11-50": round(len(df[(df["min_order"] >10) & (df["min_order"] <=50)])/len(df)*100, 1),
            ">50": round(len(df[df["min_order"] >50])/len(df)*100, 1)
        }
    }
    
    return {
        "price_analysis": price_stats,
        "supplier_distribution": supplier_type_dist,
        "order_analysis": order_stats,
        "customize_support_rate": round(len(df[df["customize_support"] == True])/len(df)*100, 1)
    }

if __name__ == "__main__":
    # 配置应用信息（替换为实际值）
    APP_KEY = "your_app_key_here"
    APP_SECRET = "your_app_secret_here"
    
    # 初始化API客户端
    search_api = AlibabaProductSearchAPI(APP_KEY, APP_SECRET)
    
    # 搜索参数配置
    KEYWORDS = "纯棉T恤 定制"
    SEARCH_PARAMS = {
        "start_price": 10,        # 最低价格
        "end_price": 50,          # 最高价格
        "sort_type": 3,           # 按销量降序
        "page_size": 50           # 每页50条
    }
    
    # 执行搜索
    print(f"搜索关键字: {KEYWORDS}")
    raw_result = search_api.search_products(
        keywords=KEYWORDS,
        page=1,
        **SEARCH_PARAMS    )
    
    # 解析结果
    if raw_result:
        products_df, stats = parse_search_results(raw_result)
        
        if products_df is not None and not products_df.empty:
            print(f"\n搜索完成: 共找到 {stats['total_results']} 件商品，当前为第 {stats['page']}/{stats['total_pages']} 页")
            
            # 高级筛选：找生产厂家、诚信通≥5年、满意度≥95%
            filtered_df = filter_suppliers(
                products_df,
                min_trust_year=5,
                min_satisfaction=95,
                supplier_types=["生产厂家"],
                customize_required=True  # 需要支持定制
            )
            
            print(f"筛选后剩余 {len(filtered_df)} 件符合条件的商品")
            
            # 显示前5条结果
            print("\nTop 5 商品信息:")
            print(filtered_df[["title", "price", "min_order", "company_name", 
                             "trust_year", "sales_count"]].head())
            
            # 供应链分析
            supply_chain_analysis = analyze_supply_chain(filtered_df)
            print("\n===== 供应链分析结果 =====")
            print(f"价格区间: {supply_chain_analysis['price_analysis']['price_range']}元")
            print(f"平均起订量: {supply_chain_analysis['order_analysis']['avg_min_order']}件")
            print(f"支持定制的比例: {supply_chain_analysis['customize_support_rate']}%")
            print("供应商类型分布:")
            for type_name, count in supply_chain_analysis['supplier_distribution'].items():
                print(f"- {type_name}: {count}家")
            
            # 保存结果
            filtered_df.to_csv("filtered_products.csv", index=False, encoding="utf-8-sig")
            print("\n筛选结果已保存至 filtered_products.csv")

import asyncioimport aiohttpclass AsyncAlibabaSearchAPI(AlibabaProductSearchAPI):
    async def async_search(self, session: aiohttp.ClientSession, keywords: str, page: int,** kwargs) -> Optional[Dict]:
        """异步搜索实现"""
        params = {
            "method": "alibaba.product.search",
            "app_key": self.app_key,
            "timestamp": time.strftime("%Y-%m-%d %H:%M:%S"),
            "format": "json",
            "v": "2.0",
            "keywords": keywords,
            "page": page,
            **kwargs        }
        params["sign"] = generate_1688_sign(params, self.app_secret)
        
        try:
            async with session.get(self.api_url, params=params, timeout=15) as response:
                response.raise_for_status()
                result = await response.json()
                if "error_response" in result:
                    return None
                return result.get("alibaba_product_search_response", {}).get("result", {})
        except Exception as e:
            print(f"异步搜索错误: {str(e)}")
            return None
    
    async def batch_search(self, keywords: str, total_pages: int = 3,** kwargs) -> List[pd.DataFrame]:
        """批量异步搜索多页数据"""
        async with aiohttp.ClientSession() as session:
            tasks = [
                self.async_search(session, keywords, page, **kwargs)
                for page in range(1, total_pages + 1)
            ]
            results = await asyncio.gather(*tasks)
            
            # 解析所有结果
            dataframes = []
            for result in results:
                if result:
                    df, _ = parse_search_results(result)
                    if df is not None and not df.empty:
                        dataframes.append(df)
            
            return dataframes# 使用示例# loop = asyncio.get_event_loop()# async_api = AsyncAlibabaSearchAPI(APP_KEY, APP_SECRET)# dfs = loop.run_until_complete(async_api.batch_search("纯棉T恤", total_pages=3))# combined_df = pd.concat(dfs, ignore_index=True)

import redisimport picklefrom datetime import timedeltaclass CachedSearchAPI(AlibabaProductSearchAPI):
    def __init__(self, app_key, app_secret, redis_host="localhost", redis_port=6379):
        super().__init__(app_key, app_secret)
        self.redis = redis.Redis(host=redis_host, port=redis_port, db=0)
        
    def generate_cache_key(self, keywords: str,** kwargs) -> str:
        """生成缓存键"""
        sorted_kwargs = sorted(kwargs.items())
        kwargs_str = "_".join([f"{k}_{v}" for k, v in sorted_kwargs])
        return f"1688_search:{keywords}:{kwargs_str}"
        
    def search_with_cache(self, keywords: str, expire=3600, **kwargs) -> Optional[Dict]:
        """带缓存的搜索"""
        cache_key = self.generate_cache_key(keywords,** kwargs)
        
        # 尝试从缓存获取
        cached_data = self.redis.get(cache_key)
        if cached_data:
            return pickle.loads(cached_data)
            
        # 调用接口
        result = self.search_products(keywords, **kwargs)
        
        # 存入缓存
        if result:
            self.redis.setex(cache_key, timedelta(seconds=expire), pickle.dumps(result))
            
        return result

from tenacity import retry, stop_after_attempt, wait_exponential, retry_if_exception_typeclass ReliableSearchAPI(AlibabaProductSearchAPI):
    @retry(
        stop=stop_after_attempt(3),  # 最多重试3次
        wait=wait_exponential(multiplier=1, min=2, max=10),  # 指数退避
        retry=retry_if_exception_type((requests.exceptions.RequestException, json.JSONDecodeError))
    )
    def search_with_retry(self,** kwargs) -> Optional[Dict]:
        """带重试机制的搜索"""
        return super().search_products(**kwargs)# 限流实现import timefrom collections import dequeclass RateLimitedSearchAPI(AlibabaProductSearchAPI):
    def __init__(self, app_key, app_secret, max_calls=100, period=60):
        super().__init__(app_key, app_secret)
        self.max_calls = max_calls  # 每分钟最大调用次数
        self.period = period
        self.calls = deque()
        
    def acquire(self):
        """获取调用许可"""
        now = time.time()
        # 移除过期的调用记录
        while self.calls and now - self.calls[0] > self.period:
            self.calls.popleft()
        # 如果达到最大调用次数，等待
        if len(self.calls) >= self.max_calls:
            sleep_time = self.period - (now - self.calls[0])
            time.sleep(sleep_time + 0.1)
            self.acquire()
        self.calls.append(now)
        
    def search_with_rate_limit(self, **kwargs) -> Optional[Dict]:
        """限流搜索"""
        self.acquire()
        return super().search_products(** kwargs)