从京东商品详情 API 到结构化数据：关键字段提取与清洗策略

import requests
import json
import re
from bs4 import BeautifulSoup
import pandas as pd
from datetime import datetime

class JDProductAPI:
    def __init__(self, app_key, app_secret):
        self.app_key = app_key
        self.app_secret = app_secret
        self.base_url = "https://api.jd.com/routerjson"
    
    def get_sign(self, params):
        # 生成签名（京东 API 要求的签名算法）
        # 此处省略签名生成的具体实现，实际使用时需要根据京东开放平台的文档进行实现
        # 签名算法通常涉及对参数进行排序、拼接、加密等操作
        return "generated_sign"
    
    def get_product_detail(self, sku_id):
        # 获取商品详情
        params = {
            "app_key": self.app_key,
            "method": "jingdong.item.jingdong.get",
            "timestamp": datetime.now().strftime("%Y-%m-%d %H:%M:%S"),
            "format": "json",
            "v": "1.0",
            "skuId": sku_id
        }
        # 生成签名
        params["sign"] = self.get_sign(params)
        # 发送请求
        response = requests.get(self.base_url, params=params)
        if response.status_code == 200:
            return response.json()
        else:
            print(f"请求失败，状态码：{response.status_code}")
            return None

class DataProcessor:
    def extract_key_fields(self, product_data):
        # 提取关键字段
        if not product_data or "jingdong_item_jingdong_get_response" not in product_data:
            return None
        
        result = product_data["jingdong_item_jingdong_get_response"]
        if "result" not in result:
            return None
        
        product_info = result["result"]
        key_fields = {
            "skuId": product_info.get("skuId", ""),
            "name": product_info.get("name", ""),
            "price": product_info.get("price", 0),
            "marketPrice": product_info.get("marketPrice", 0),
            "brandName": product_info.get("brandName", ""),
            "categoryId": product_info.get("categoryId", ""),
            "categoryName": product_info.get("categoryName", ""),
            "mainImage": product_info.get("mainImage", ""),
            "imageList": product_info.get("imageList", []),
            "detail": product_info.get("detail", ""),
            "parameterList": product_info.get("parameterList", []),
            "salesCount": product_info.get("salesCount", 0),
            "commentCount": product_info.get("commentCount", 0),
            "shopName": product_info.get("shopName", ""),
            "shopId": product_info.get("shopId", "")
        }
        return key_fields
    
    def clean_data(self, data):
        # 清洗数据
        if not data:
            return None
        
        # 处理缺失值
        # 对于价格字段，如果缺失，填充为 0
        data["price"] = data["price"] if data["price"] is not None else 0
        data["marketPrice"] = data["marketPrice"] if data["marketPrice"] is not None else 0
        
        # 对于销量和评论数字段，如果缺失，填充为 0
        data["salesCount"] = data["salesCount"] if data["salesCount"] is not None else 0
        data["commentCount"] = data["commentCount"] if data["commentCount"] is not None else 0
        
        # 处理文本字段
        # 去除商品名称中的特殊字符
        data["name"] = re.sub(r"[^\u4e00-\u9fa5a-zA-Z0-9]", "", data["name"])
        
        # 去除商品详情中的 HTML 标签
        if data["detail"]:
            soup = BeautifulSoup(data["detail"], "html.parser")
            data["detail"] = soup.get_text().strip()
        
        # 处理规格参数列表
        if data["parameterList"]:
            cleaned_params = []
            for param in data["parameterList"]:
                param_name = param.get("name", "").strip()
                param_value = param.get("value", "").strip()
                if param_name and param_value:
                    cleaned_params.append({"name": param_name, "value": param_value})
            data["parameterList"] = cleaned_params
        
        # 数据类型转换
        # 将价格字段转换为浮点数
        data["price"] = float(data["price"]) if data["price"] != "" else 0.0
        data["marketPrice"] = float(data["marketPrice"]) if data["marketPrice"] != "" else 0.0
        
        # 将销量和评论数字段转换为整数
        data["salesCount"] = int(data["salesCount"]) if data["salesCount"] != "" else 0
        data["commentCount"] = int(data["commentCount"]) if data["commentCount"] != "" else 0
        
        # 将分类 ID 和店铺 ID 转换为字符串
        data["categoryId"] = str(data["categoryId"]) if data["categoryId"] != "" else ""
        data["shopId"] = str(data["shopId"]) if data["shopId"] != "" else ""
        
        return data

# 示例用法
if __name__ == "__main__":
    # 替换为自己的 AppKey 和 AppSecret
    app_key = "your_app_key"
    app_secret = "your_app_secret"
    
    # 初始化 API 客户端和数据处理器
    jd_api = JDProductAPI(app_key, app_secret)
    processor = DataProcessor()
    
    # 要查询的商品 SKU ID
    sku_id = "100000000000"
    
    # 获取商品详情
    product_data = jd_api.get_product_detail(sku_id)
    
    if product_data:
        # 提取关键字段
        key_fields = processor.extract_key_fields(product_data)
        
        if key_fields:
            # 清洗数据
            cleaned_data = processor.clean_data(key_fields)
            
            # 打印清洗后的数据
            print(json.dumps(cleaned_data, ensure_ascii=False, indent=2))
            
            # 可以将清洗后的数据保存到数据库或文件中
            # 例如，保存到 CSV 文件
            df = pd.DataFrame([cleaned_data])
            df.to_csv("jd_product_detail.csv", index=False, encoding="utf-8-sig")
            print("数据已保存到 jd_product_detail.csv 文件中")
        else:
            print("提取关键字段失败")
    else:
        print("获取商品详情失败")