当当商品详情数据采集，当当API接口（json数据返回）

import requestsimport jsonimport timeimport randomfrom typing import Dict, Optionalclass DangdangSpider:
    def __init__(self):
        # 随机User-Agent池，降低反爬风险
        self.user_agents = [
            "Mozilla/5.0 (Windows NT 10.0; Win64; x64) Chrome/125.0.0.0 Safari/537.36",
            "Mozilla/5.0 (Macintosh; Intel Mac OS X 14_5) Firefox/126.0",
            "Mozilla/5.0 (iPhone; CPU iPhone OS 17_5 like Mac OS X) Mobile/15E148"
        ]
        self.headers = {
            "User-Agent": random.choice(self.user_agents),
            "Referer": "https://product.dangdang.com/",
            "Accept-Language": "zh-CN,zh;q=0.9"
        }

    def _fetch(self, url: str, params: Dict) -> Optional[Dict]:
        """通用请求方法，处理异常和JSON解析"""
        try:
            # 随机延迟1-2秒，避免高频请求
            time.sleep(random.uniform(1, 2))
            response = requests.get(
                url=url,
                params=params,
                headers=self.headers,
                timeout=15
            )
            response.raise_for_status()  # 触发4xx/5xx错误
            return response.json()
        except requests.exceptions.RequestException as e:
            print(f"请求错误: {str(e)}")
        except json.JSONDecodeError:
            print("返回数据不是有效JSON")
        return None

    def get_product_base(self, product_id: str) -> Optional[Dict]:
        """获取商品基础信息（名称、价格、作者等）"""
        url = "https://product.dangdang.com/index.php?r=product/getProductInfo"
        params = {"productId": product_id}
        data = self._fetch(url, params)
        return data.get("data") if data and data.get("status") == 1 else None

    def get_product_params(self, product_id: str) -> Optional[Dict]:
        """获取商品规格参数（如书籍开本、页数）"""
        url = "https://product.dangdang.com/index.php?r=product/getProductParam"
        params = {"productId": product_id}
        data = self._fetch(url, params)
        return data.get("data") if data and data.get("status") == 1 else None

    def get_product_comments(self, product_id: str, page: int = 1) -> Optional[Dict]:
        """获取商品评论（分页，每页20条）"""
        url = "https://product.dangdang.com/index.php?r=comment/list"
        params = {
            "productId": product_id,
            "page": page,
            "size": 20,
            "sortType": 1  # 1=按时间排序，2=按 helpful 排序
        }
        data = self._fetch(url, params)
        return data.get("data") if data and data.get("status") == 1 else None

    def collect_all(self, product_id: str) -> Dict:
        """整合所有信息，返回完整数据"""
        return {
            "base_info": self.get_product_base(product_id),
            "params": self.get_product_params(product_id),
            "comments_page1": self.get_product_comments(product_id, page=1)
        }# 示例使用if __name__ == "__main__":
    spider = DangdangSpider()
    product_id = "29383160"  # 示例商品ID（可替换）
    
    # 采集完整数据
    product_data = spider.collect_all(product_id)
    
    # 打印关键信息
    if product_data["base_info"]:
        print("===== 基础信息 =====")
        base = product_data["base_info"]
        print(f"商品名称：{base.get('product_name')}")
        print(f"原价：{base.get('original_price')} | 当前价：{base.get('current_price')}")
        print(f"作者：{base.get('author')} | 出版社：{base.get('publisher')}")
        print(f"好评率：{base.get('good_comment_rate')}%")
    
    if product_data["params"]:
        print("\n===== 规格参数 =====")
        for param in product_data["params"].get("param_list", []):
            print(f"{param.get('name')}: {param.get('value')}")  # 如"开本: 16开"
    
    if product_data["comments_page1"]:
        print("\n===== 评论示例 =====")
        comments = product_data["comments_page1"]["list"]
        if comments:
            print(f"评论总数：{product_data['comments_page1']['total']}")
            print(f"用户评论：{comments[0].get('comment_content')[:80]}...")