import timeimport hashlibimport requestsimport jsonfrom typing import Dict, List, Optionalfrom concurrent.futures import ThreadPoolExecutor, as_completedclass JdProductAPI:    def __init__(self, app_key: str, app_secret: str):        self.app_key = app_key        self.app_secret = app_secret        self.api_url = "京东开放平台接口地址"  # 按官方文档配置        self.session = self._init_session()    def _init_session(self) -> requests.Session:        """初始化会话池，减少连接开销"""        session = requests.Session()        adapter = requests.adapters.HTTPAdapter(            pool_connections=15, pool_maxsize=80, max_retries=3        )        session.mount('https://', adapter)        return session    def _generate_sign(self, params: Dict) -> str:        """生成京东签名（处理毫秒级时间戳的坑）"""        # 坑点1：必须按参数名ASCII升序排序，普通dict会乱序        sorted_params = sorted(params.items(), key=lambda x: x[0])        # 坑点2：拼接格式为"key=value&key=value"，首尾加app_secret        sign_str = "&".join([f"{k}={v}" for k, v in sorted_params])        sign_str = f"{self.app_secret}{sign_str}{self.app_secret}"        # 坑点3：MD5加密后需转大写        return hashlib.md5(sign_str.encode("utf-8")).hexdigest().upper()    def _fetch_page_items(self, start_time: str, end_time: str, page: int = 1, page_size: int = 30) -> List[Dict]:        """单页商品拉取（基础方法）"""        params = {            "method": "jingdong.ware.get",            "app_key": self.app_key,            "timestamp": str(int(time.time() * 1000)),  # 京东需毫秒级时间戳            "format": "json",            "v": "2.0",            "sign_method": "md5",            "page": str(page),            "pageSize": str(page_size),            "startModified": start_time,            "endModified": end_time,            "fields": "wareId,title,price,stock,specList,modifiedTime,wareStatus"        }        params["sign"] = self._generate_sign(params)        try:            response = self.session.get(self.api_url, params=params, timeout=(5, 18))            result = response.json()            if result.get("code") != 0:                print(f"单页拉取失败: {result.get('message', '未知错误')}")                return []            return result.get("data", {}).get("wareList", [])        except Exception as e:            print(f"单页拉取异常: {str(e)}")            return []

def parse_spec_structure(self, raw_specs: List[Dict]) -> List[Dict]:        """        递归解析京东多规格结构        :param raw_specs: 接口返回的原始specList        :return: 扁平化的规格列表        """        parsed_specs = []        for spec in raw_specs:            # 基础规格信息            base_spec = {                "specId": spec.get("specId", ""),                "specName": spec.get("specName", ""),                "specValue": spec.get("specValue", ""),                "children": []  # 子规格容器            }            # 递归处理子规格（若有）            if child_specs := spec.get("childSpecList"):                base_spec["children"] = self.parse_spec_structure(child_specs)            parsed_specs.append(base_spec)        return parsed_specs    def get_ware_with_parsed_spec(self, ware_id: str) -> Optional[Dict]:        """获取商品详情+解析后规格"""        params = {            "method": "jingdong.ware.get",            "app_key": self.app_key,            "timestamp": str(int(time.time() * 1000)),            "format": "json",            "v": "2.0",            "sign_method": "md5",            "wareId": ware_id,            "fields": "wareId,title,price,stock,specList,modifiedTime,wareStatus"        }        params["sign"] = self._generate_sign(params)        try:            response = self.session.get(self.api_url, params=params, timeout=(5, 15))            result = response.json()            if result.get("code") != 0:                print(f"商品详情拉取失败: {result.get('message')}")                return None            ware_data = result.get("data", {}).get("ware", {})            # 解析规格并替换原始字段            if raw_specs := ware_data.get("specList"):                ware_data["parsedSpecList"] = self.parse_spec_structure(raw_specs)                del ware_data["specList"]  # 删除原始嵌套字段            # 处理价格区间（拆分为起始价/最高价）            if price_range := ware_data.get("price"):                price_split = price_range.split("-")                ware_data["startPrice"] = float(price_split[0])                ware_data["endPrice"] = float(price_split[1]) if len(price_split) > 1 else ware_data["startPrice"]                del ware_data["price"]            return ware_data        except Exception as e:            print(f"商品详情异常: {str(e)}")            return None

def verify_ware_completeness(self, fetched_wares: List[Dict], start_time: str, end_time: str) -> Dict:        """        双重校验商品完整性：官方计数+字段核对        :param fetched_wares: 已拉取商品列表        :param start_time/end_time: 时间范围        :return: 校验结果        """        # 1. 获取官方总计数        official_count = 0        try:            params = {                "method": "jingdong.ware.count.get",                "app_key": self.app_key,                "timestamp": str(int(time.time() * 1000)),                "format": "json",                "v": "2.0",                "sign_method": "md5",                "startModified": start_time,                "endModified": end_time            }            params["sign"] = self._generate_sign(params)            response = self.session.get(self.api_url, params=params, timeout=(3, 10))            result = response.json()            if result.get("code") == 0:                official_count = result.get("data", {}).get("totalCount", 0)        except Exception as e:            print(f"官方计数获取异常: {str(e)}")        # 2. 字段完整性核对（必选字段：wareId、title、startPrice、wareStatus）        required_fields = ["wareId", "title", "startPrice", "wareStatus"]        incomplete_wares = []        for ware in fetched_wares:            missing_fields = [f for f in required_fields if f not in ware or not ware[f]]            if missing_fields:                incomplete_wares.append({                    "wareId": ware.get("wareId", "未知ID"),                    "missingFields": missing_fields                })        # 3. 生成校验结果        fetched_count = len(fetched_wares)        return {            "officialCount": official_count,            "fetchedCount": fetched_count,            "completenessRate": round(fetched_count / official_count * 100, 1) if official_count != 0 else 0,            "incompleteWares": incomplete_wares,            "isAcceptable": abs(fetched_count - official_count) <= 3 and len(incomplete_wares) <= 2        }

# tasks.py（Celery分布式任务）from celery import Celeryimport jsonfrom jd_api import JdProductAPI  # 导入上文的JdProductAPI类app = Celery('jd_ware_tasks', broker='redis://localhost:6379/0')@app.task(bind=True, max_retries=3)def fetch_time_segment(self, start_time: str, end_time: str, config: dict) -> int:    """按时间分段拉取商品的分布式任务"""    api = JdProductAPI(config["app_key"], config["app_secret"])    try:        # 分页拉取当前时间段商品        page = 1        total_fetched = 0        while True:            wares = api._fetch_page_items(start_time, end_time, page=page, page_size=30)            if not wares:                break            # 存储结果（按时间分段存文件）            with open(f"jd_wares_{start_time.replace(' ', '_')}_{end_time.replace(' ', '_')}_page{page}.json", "w") as f:                json.dump(wares, f, ensure_ascii=False)            total_fetched += len(wares)            page += 1            time.sleep(0.4)  # 控制频率        return total_fetched    except Exception as e:        # 失败6秒后重试，最多3次        self.retry(exc=e, countdown=6)

if __name__ == "__main__":    # 初始化客户端（替换为合规申请的key/secret）    jd_api = JdProductAPI(app_key="your_jd_app_key", app_secret="your_jd_app_secret")    # 1. 按时间分段拉取（示例：2024年1月1日-2024年1月7日）    print("===== 按时间分段拉取 =====")    time_segments = [        ("2024-01-01 00:00:00", "2024-01-02 23:59:59"),        ("2024-01-03 00:00:00", "2024-01-04 23:59:59"),        ("2024-01-05 00:00:00", "2024-01-07 23:59:59")    ]    total_wares = []    with ThreadPoolExecutor(max_workers=4) as executor:        futures = [executor.submit(jd_api._fetch_page_items, s, e) for s, e in time_segments]        for future in as_completed(futures):            total_wares.extend(future.result())    print(f"总拉取商品数: {len(total_wares)}")    # 2. 完整性校验    print("\n===== 完整性校验 =====")    verify_result = jd_api.verify_ware_completeness(        total_wares, 
        start_time="2024-01-01 00:00:00", 
        end_time="2024-01-07 23:59:59"    )    print(f"校验结果: 官方计数{verify_result['officialCount']} | 拉取计数{verify_result['fetchedCount']} | 完整率{verify_result['completenessRate']}%")    if verify_result["incompleteWares"]:        print(f"不完整商品数: {len(verify_result['incompleteWares'])}（示例：{verify_result['incompleteWares'][0]}）")    # 3. 单商品详情+规格解析    print("\n===== 单商品详情+规格解析 =====")    sample_ware = jd_api.get_ware_with_parsed_spec(ware_id="100012014970")  # 示例商品ID    if sample_ware:        print(f"商品ID: {sample_ware['wareId']}")        print(f"商品标题: {sample_ware['title']}")        print(f"价格区间: {sample_ware['startPrice']}-{sample_ware['endPrice']}元")        print(f"解析后规格（前2层）: {json.dumps(sample_ware['parsedSpecList'][:2], ensure_ascii=False, indent=2)}")