import requests import json from Crypto.Cipher import AES from Crypto.Util.Padding import unpad from typing import Optional, List, Dict, Any # --- 1. 解密函数 (从JS代码转换) --- def decrypt_order_data(hex_string): """ 使用AES-CBC模式解密订单详情数据 """ key = b"S9u978T13NLCGc5W" # 密钥 iv = b"X83yWMD9iKhLxfwX" # 偏移量 (IV) # 将十六进制字符串转换为字节 encrypted_data = bytes.fromhex(hex_string) # 创建AES-CBC解密器 cipher = AES.new(key, AES.MODE_CBC, iv) # 解密并去除填充 (PKCS7) decrypted_data = unpad(cipher.decrypt(encrypted_data), AES.block_size) # 返回UTF-8字符串 return decrypted_data.decode('utf-8') # --- 2. 请求头配置 --- HEADERS = { 'Connection': 'keep-alive', 'content-type': 'application/json', 'Version': '2.50.0', 'timestamp': '1784094158706', 'Distinct-Id': '1784057137891-1059356-086b42dd38b0b98-24491897', 'vaccine': 'aa454c048e8843247cebf378d1a9a0fed759027f329caf68fd900654da4232d0', 'U-A': 'JiDaXiaWeApp/23958 (:)', 'x-version': 'ITC-77385', 'ahs-gate-language': 'zh_CN', 'APP-ID': 'jdx135624', 'platform': 'weapp', 'Register-Flow-Source': 'RECYCLER', 'ahs-language': 'zh-CN', 'access_token': 'jdx9ce52336abee4cf78db56567dc259512', 'User-Agent': 'Mozilla/5.0 (iPhone; CPU iPhone OS 26_5_2 like Mac OS X) AppleWebKit/605.1.15 (KHTML, like Gecko) Mobile/15E148 MicroMessenger/8.0.75(0x18004b3c) NetType/WIFI Language/zh_CN', 'Referer': 'https://servicewechat.com/wx7e6573879e91b1ac/28/page-frame.html', } # --- 3. 获取待抢订单列表 --- def get_pending_order_list(page_size: int = 10) -> Optional[List[Dict[str, Any]]]: """ 获取待抢订单列表 Args: page_size: 每页数量,默认10条 Returns: 订单列表,失败返回None """ url = "https://wirelessgate.aihuishou.com/jdx-rc-bff-service/app/order/pendingGrab/list" payload = { "pageSize": page_size, "status": -1, "isApplyForHandle": None, "applyPhotoStatus": None, "productIds": [265024, 225362, 225361, 225358, 225366, 173473, 161415, 161414, 161413, 161412, 121769, 121439, 121436, 121435, 98526, 98528, 98527, 98525, 43513, 43512, 43511, 43510, 36047, 36045, 36044, 36046, 32292, 32291, 32290, 27640, 27639, 27637, 25827, 25680, 25679, 23423, 23422, 66428, 34701, 20079, 17752, 17726, 17462, 17461, 17459, 17458, 17457, 17460, 17455, 2247, 2246], "sourceIds": None, "keyword": None, "timeObj": None, "rcOrderResult": None } try: response = requests.post(url, headers=HEADERS, json=payload) response.raise_for_status() result = response.json() if result.get('code') != 200: print(f"获取订单列表失败: {result.get('resultMessage', '未知错误')}") return None order_list = result.get('data', []) print(f"成功获取 {len(order_list)} 条待抢订单") return order_list except requests.RequestException as e: print(f"请求订单列表失败: {e}") return None except json.JSONDecodeError as e: print(f"解析订单列表JSON失败: {e}") return None # --- 4. 显示订单列表 --- def display_order_list(order_list: List[Dict[str, Any]]): """ 美化显示订单列表 """ if not order_list: print("暂无待抢订单") return print("\n" + "="*80) print("待抢订单列表:") print("="*80) for idx, order in enumerate(order_list, 1): print(f"\n[{idx}] 订单号: {order.get('orderNo', 'N/A')}") print(f" 商品: {order.get('brandInfo', 'N/A')}") print(f" IMEI: {order.get('imei', 'N/A')}") print(f" 预估价格: {order.get('price', '待定')}") print(f" 门店: {order.get('title', 'N/A')}") print(f" 截止时间: {order.get('biddingDeadlineDt', 'N/A')}") # 显示标签 labels = order.get('labels', []) if labels: label_names = [label.get('labelName', '') for label in labels] print(f" 标签: {', '.join(label_names)}") # 显示提示 tips = order.get('tips', {}) if tips: print(f" 提示: {tips.get('tipStr', 'N/A')}") print("\n" + "="*80) # --- 5. 从订单列表中提取订单号 --- def select_order_by_index(order_list: List[Dict[str, Any]], index: int) -> Optional[str]: """ 根据索引从订单列表中提取订单号 Args: order_list: 订单列表 index: 索引(从1开始) Returns: 订单号,失败返回None """ if not order_list: print("订单列表为空") return None if index < 1 or index > len(order_list): print(f"索引超出范围,请输入1-{len(order_list)}之间的数字") return None order_no = order_list[index - 1].get('orderNo') if not order_no: print(f"第{index}个订单缺少订单号") return None print(f"已选择订单: {order_no}") return order_no # --- 6. 获取订单详情(原有功能) --- def get_order_detail(order_no: str) -> Optional[Dict[str, Any]]: """ 获取订单详情 Args: order_no: 订单号 Returns: 解密后的订单详情,失败返回None """ # 第一步:获取BiddingNo snatch_url = "https://wirelessgate.aihuishou.com/jdx-rc-service/front/bid/snatch" snatch_payload = {"orderNo": order_no} print(f"\n正在获取订单 {order_no} 的BiddingNo...") try: snatch_response = requests.post(snatch_url, headers=HEADERS, json=snatch_payload) snatch_response.raise_for_status() snatch_data = snatch_response.json() if snatch_data.get('code') != 200: print(f"获取BiddingNo失败: {snatch_data.get('resultMessage', '未知错误')}") return None bidding_no = snatch_data.get('data') if not bidding_no: print("获取BiddingNo为空") return None print(f"成功获取 BiddingNo: {bidding_no}") except requests.RequestException as e: print(f"请求BiddingNo失败: {e}") return None except json.JSONDecodeError as e: print(f"解析BiddingNo响应失败: {e}") return None # 第二步:请求订单详情(加密数据) detail_url = f"https://wirelessgate.aihuishou.com/jdx-qa-service/app/ka/v3/order/detail?orderNo={order_no}&biddingNo={bidding_no}" print("正在请求订单详情...") try: detail_response = requests.get(detail_url, headers=HEADERS) detail_response.raise_for_status() detail_data = detail_response.json() if detail_data.get('code') != 200: print(f"请求订单详情失败: {detail_data.get('resultMessage', '未知错误')}") return None encrypted_hex = detail_data.get('data') if not encrypted_hex: print("未获取到加密数据") return None # 第三步:解密数据 print("正在解密数据...") decrypted_json_string = decrypt_order_data(encrypted_hex) order_detail = json.loads(decrypted_json_string) return order_detail except requests.RequestException as e: print(f"请求订单详情失败: {e}") return None except json.JSONDecodeError as e: print(f"解析订单详情JSON失败: {e}") return None except Exception as e: print(f"解密或解析失败: {e}") return None # --- 7. 查找图片链接(原有功能) --- def find_original_image_by_notice(json_data, notice_keyword): """ 在复杂的嵌套JSON中,根据notice字段的关键词查找并返回对应的originalImage链接 """ if isinstance(json_data, str): try: json_data = json.loads(json_data) except json.JSONDecodeError: print("错误:输入的字符串不是有效的JSON格式。") return None def recursive_search(obj): if isinstance(obj, dict): if 'notice' in obj and 'originalImage' in obj: if notice_keyword.lower() in obj['notice'].lower(): return obj['originalImage'] for value in obj.values(): result = recursive_search(value) if result: return result elif isinstance(obj, list): for item in obj: result = recursive_search(item) if result: return result return None return recursive_search(json_data) # --- 8. 主流程(新增列表获取功能) --- def main(): # 步骤1:获取待抢订单列表 print("正在获取待抢订单列表...") order_list = get_pending_order_list(page_size=10) if not order_list: print("无法获取订单列表,程序退出") return # 步骤2:显示列表 display_order_list(order_list) # 步骤3:让用户选择要查看的订单 while True: try: choice = input("\n请输入要查看详情的订单序号 (1-{},输入0退出): ".format(len(order_list))) choice = int(choice.strip()) if choice == 0: print("程序退出") return if 1 <= choice <= len(order_list): break else: print(f"请输入1-{len(order_list)}之间的数字") except ValueError: print("请输入有效的数字") except KeyboardInterrupt: print("\n程序被中断") return # 步骤4:获取选中的订单号 order_no = select_order_by_index(order_list, choice) if not order_no: print("获取订单号失败") return # 步骤5:获取订单详情 order_detail = get_order_detail(order_no) if not order_detail: print("获取订单详情失败") return # 步骤6:输出结果 print("\n=== 解密后的订单详情 ===") print(json.dumps(order_detail, indent=2, ensure_ascii=False)) # 步骤7:查找图片链接 report_img = find_original_image_by_notice(order_detail, "苹果沙漏验机报告") if report_img is None: report_img = find_original_image_by_notice(order_detail, "验机照片") if report_img: print(f"\n找到验机报告图片: {report_img}") else: print("\n未找到验机报告图片") if __name__ == "__main__": main()