import sys
import requests
import json
import os
import re
import uuid
import hashlib
from Crypto.Cipher import AES
from Crypto.Util.Padding import unpad
from typing import Optional, List, Dict, Any
from PyQt5.QtWidgets import *
from PyQt5.QtCore import *
from PyQt5.QtGui import *
# --- 核心功能类 ---
class OrderManager:
def __init__(self, access_token, ocr_url):
self.access_token = access_token
self.ocr_url = ocr_url
self.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': access_token,
'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',
}
self.key = b"S9u978T13NLCGc5W"
self.iv = b"X83yWMD9iKhLxfwX"
def decrypt_order_data(self, hex_string):
"""解密订单数据"""
encrypted_data = bytes.fromhex(hex_string)
cipher = AES.new(self.key, AES.MODE_CBC, self.iv)
decrypted_data = unpad(cipher.decrypt(encrypted_data), AES.block_size)
return decrypted_data.decode('utf-8')
def get_pending_order_list(self, page_size=10):
"""获取待抢订单列表"""
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=self.headers, json=payload, timeout=30)
response.raise_for_status()
result = response.json()
if result.get('code') != 200:
return None, result.get('resultMessage', '未知错误')
return result.get('data', []), None
except Exception as e:
return None, str(e)
def get_order_detail(self, order_no):
"""获取订单详情"""
# 获取BiddingNo
snatch_url = "https://wirelessgate.aihuishou.com/jdx-rc-service/front/bid/snatch"
snatch_payload = {"orderNo": order_no}
try:
snatch_response = requests.post(snatch_url, headers=self.headers, json=snatch_payload, timeout=30)
snatch_response.raise_for_status()
snatch_data = snatch_response.json()
if snatch_data.get('code') != 200:
return None, snatch_data.get('resultMessage', '获取BiddingNo失败')
bidding_no = snatch_data.get('data')
if not bidding_no:
return None, "获取BiddingNo为空"
except Exception as e:
return None, str(e)
# 获取订单详情
detail_url = f"https://wirelessgate.aihuishou.com/jdx-qa-service/app/ka/v3/order/detail?orderNo={order_no}&biddingNo={bidding_no}"
try:
detail_response = requests.get(detail_url, headers=self.headers, timeout=30)
detail_response.raise_for_status()
detail_data = detail_response.json()
if detail_data.get('code') != 200:
return None, detail_data.get('resultMessage', '获取订单详情失败')
encrypted_hex = detail_data.get('data')
if not encrypted_hex:
return None, "未获取到加密数据"
decrypted_json_string = self.decrypt_order_data(encrypted_hex)
order_detail = json.loads(decrypted_json_string)
return order_detail, None
except Exception as e:
return None, str(e)
def find_original_image_by_notice(self, json_data, notice_keyword):
"""查找图片链接"""
if isinstance(json_data, str):
try:
json_data = json.loads(json_data)
except json.JSONDecodeError:
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)
def get_all_image_notices(self, json_data):
"""获取所有图片的notice信息"""
notices = []
def recursive_search(obj):
if isinstance(obj, dict):
if 'notice' in obj and 'image' in obj:
notices.append(obj['notice'])
for value in obj.values():
recursive_search(value)
elif isinstance(obj, list):
for item in obj:
recursive_search(item)
recursive_search(json_data)
return list(set(notices)) # 去重
def download_and_ocr_image(self, image_url):
"""下载图片并OCR识别"""
try:
# 下载图片
response = requests.get(image_url, timeout=30)
response.raise_for_status()
# 创建temp目录
os.makedirs('./temp', exist_ok=True)
# 保存临时文件
url_path = image_url.split('?')[0]
file_ext = os.path.splitext(url_path)[1]
if not file_ext:
file_ext = '.jpg'
random_filename = f"{uuid.uuid4().hex}{file_ext}"
temp_path = os.path.join('./temp', random_filename)
with open(temp_path, 'wb') as f:
f.write(response.content)
# 发送到OCR服务
with open(temp_path, 'rb') as f:
files = {'image': (random_filename, f, 'image/jpeg')}
ocr_response = requests.post(self.ocr_url, files=files, timeout=60)
# 清理临时文件
if os.path.exists(temp_path):
os.remove(temp_path)
if ocr_response.status_code == 200:
ocr_result = ocr_response.json()
if ocr_result.get('success'):
return ocr_result
return None
except Exception as e:
return None
def extract_text_from_ocr_result(self, ocr_result):
"""提取OCR文本"""
if not ocr_result or not ocr_result.get('texts'):
return ''
texts = ''
for item in ocr_result['texts']:
rec_texts = item.get('rec_texts', [])
for text in rec_texts:
texts = texts + text
return texts
def extract_charge_count(self, text):
"""提取充电次数"""
pattern = r'充电次数(\d+)次'
match = re.search(pattern, text)
if match:
return int(match.group(1))
return None
# --- GUI主窗口 ---
class MainWindow(QMainWindow):
def __init__(self):
super().__init__()
self.order_manager = None
self.order_list = []
self.order_details_cache = {} # 缓存订单详情
self.charge_counts = {} # 缓存充电次数
self.init_ui()
def init_ui(self):
self.setWindowTitle('爱回收订单查询工具')
self.setGeometry(100, 100, 1200, 800)
# 设置样式
self.setStyleSheet("""
QMainWindow {
background-color: #f0f0f0;
}
QGroupBox {
font-weight: bold;
border: 1px solid #ccc;
border-radius: 5px;
margin-top: 10px;
padding-top: 10px;
}
QGroupBox::title {
subcontrol-origin: margin;
left: 10px;
padding: 0 5px 0 5px;
}
QPushButton {
background-color: #0078d7;
color: white;
border: none;
padding: 8px 16px;
border-radius: 4px;
font-weight: bold;
}
QPushButton:hover {
background-color: #005a9e;
}
QPushButton:disabled {
background-color: #cccccc;
}
QLineEdit, QTextEdit {
border: 1px solid #ccc;
border-radius: 4px;
padding: 5px;
}
QTableWidget {
border: 1px solid #ccc;
border-radius: 4px;
gridline-color: #e0e0e0;
}
QTableWidget::item {
padding: 5px;
}
QHeaderView::section {
background-color: #e8e8e8;
padding: 5px;
border: 1px solid #ccc;
font-weight: bold;
}
QStatusBar {
background-color: #f0f0f0;
color: #333;
}
#log_text {
background-color: #1e1e1e;
color: #d4d4d4;
font-family: 'Consolas', 'Courier New', monospace;
font-size: 11px;
}
""")
# 中央部件
central_widget = QWidget()
self.setCentralWidget(central_widget)
main_layout = QVBoxLayout(central_widget)
main_layout.setSpacing(10)
# --- 顶部配置区域 ---
config_group = QGroupBox("配置")
config_layout = QGridLayout()
# Token输入
config_layout.addWidget(QLabel("Access Token:"), 0, 0)
self.token_input = QLineEdit()
self.token_input.setText("jdx9ce52336abee4cf78db56567dc259512")
self.token_input.setEchoMode(QLineEdit.Password)
config_layout.addWidget(self.token_input, 0, 1)
# 显示/隐藏Token按钮
self.show_token_btn = QPushButton("显示")
self.show_token_btn.setFixedWidth(60)
self.show_token_btn.clicked.connect(self.toggle_token_visibility)
config_layout.addWidget(self.show_token_btn, 0, 2)
# OCR地址输入
config_layout.addWidget(QLabel("OCR地址:"), 1, 0)
self.ocr_input = QLineEdit()
self.ocr_input.setText("http://192.168.0.110:10001/ocr")
config_layout.addWidget(self.ocr_input, 1, 1, 1, 2)
# 操作按钮
btn_layout = QHBoxLayout()
self.search_btn = QPushButton("🔍 搜索订单")
self.search_btn.clicked.connect(self.search_orders)
self.search_btn.setFixedHeight(35)
btn_layout.addWidget(self.search_btn)
self.refresh_btn = QPushButton("🔄 刷新")
self.refresh_btn.clicked.connect(self.refresh_orders)
self.refresh_btn.setFixedHeight(35)
btn_layout.addWidget(self.refresh_btn)
btn_layout.addStretch()
config_layout.addLayout(btn_layout, 2, 0, 1, 3)
config_group.setLayout(config_layout)
main_layout.addWidget(config_group)
# --- 订单列表 ---
list_group = QGroupBox("订单列表")
list_layout = QVBoxLayout()
# 表格
self.table = QTableWidget()
self.table.setColumnCount(7)
self.table.setHorizontalHeaderLabels(["序号", "订单号", "商品信息", "IMEI", "预估价格", "门店", "充电次数"])
self.table.setColumnWidth(0, 50)
self.table.setColumnWidth(1, 200)
self.table.setColumnWidth(2, 250)
self.table.setColumnWidth(3, 150)
self.table.setColumnWidth(4, 100)
self.table.setColumnWidth(5, 200)
self.table.setColumnWidth(6, 100)
self.table.horizontalHeader().setStretchLastSection(True)
self.table.verticalHeader().setVisible(False)
list_layout.addWidget(self.table)
list_group.setLayout(list_layout)
main_layout.addWidget(list_group)
# --- 日志区域 ---
log_group = QGroupBox("日志")
log_layout = QVBoxLayout()
# 日志文本框
self.log_text = QTextEdit()
self.log_text.setObjectName("log_text")
self.log_text.setReadOnly(True)
self.log_text.setMaximumHeight(200)
log_layout.addWidget(self.log_text)
# 日志控制按钮
log_btn_layout = QHBoxLayout()
self.clear_log_btn = QPushButton("清空日志")
self.clear_log_btn.clicked.connect(self.clear_log)
log_btn_layout.addWidget(self.clear_log_btn)
log_btn_layout.addStretch()
log_layout.addLayout(log_btn_layout)
log_group.setLayout(log_layout)
main_layout.addWidget(log_group)
# --- 底部状态栏 ---
self.status_bar = QStatusBar()
self.setStatusBar(self.status_bar)
self.status_bar.showMessage("就绪")
def toggle_token_visibility(self):
"""切换Token显示/隐藏"""
if self.token_input.echoMode() == QLineEdit.Password:
self.token_input.setEchoMode(QLineEdit.Normal)
self.show_token_btn.setText("隐藏")
else:
self.token_input.setEchoMode(QLineEdit.Password)
self.show_token_btn.setText("显示")
def log_message(self, message, level="INFO"):
"""添加日志消息"""
timestamp = QDateTime.currentDateTime().toString("yyyy-MM-dd hh:mm:ss")
color_map = {
"INFO": "#d4d4d4",
"WARNING": "#ffa500",
"ERROR": "#ff4444",
"SUCCESS": "#4caf50"
}
color = color_map.get(level, "#d4d4d4")
html = f'[{timestamp}] [{level}] {message}
'
self.log_text.append(html)
# 滚动到底部
scrollbar = self.log_text.verticalScrollBar()
scrollbar.setValue(scrollbar.maximum())
def clear_log(self):
"""清空日志"""
self.log_text.clear()
def search_orders(self):
"""搜索订单"""
access_token = self.token_input.text().strip()
ocr_url = self.ocr_input.text().strip()
if not access_token:
QMessageBox.warning(self, "提示", "请输入Access Token")
return
if not ocr_url:
QMessageBox.warning(self, "提示", "请输入OCR地址")
return
self.log_message("开始搜索订单...", "INFO")
# 初始化订单管理器
self.order_manager = OrderManager(access_token, ocr_url)
# 清空缓存
self.order_details_cache = {}
self.charge_counts = {}
self.status_bar.showMessage("正在获取订单列表...")
self.search_btn.setEnabled(False)
# 在后台线程执行
self.worker = OrderListWorker(self.order_manager)
self.worker.finished.connect(self.on_orders_loaded)
self.worker.error.connect(self.on_error)
self.worker.start()
def refresh_orders(self):
"""刷新订单列表"""
if self.order_manager:
self.search_orders()
else:
QMessageBox.warning(self, "提示", "请先搜索订单")
def on_orders_loaded(self, order_list):
"""订单列表加载完成"""
self.order_list = order_list
self.display_orders()
self.search_btn.setEnabled(True)
self.status_bar.showMessage(f"成功加载 {len(order_list)} 条订单")
self.log_message(f"成功加载 {len(order_list)} 条订单", "SUCCESS")
def on_error(self, error_msg):
"""错误处理"""
self.search_btn.setEnabled(True)
self.status_bar.showMessage("加载失败")
self.log_message(f"获取订单列表失败: {error_msg}", "ERROR")
QMessageBox.critical(self, "错误", f"获取订单列表失败:\n{error_msg}")
def display_orders(self):
"""显示订单列表"""
self.table.setRowCount(len(self.order_list))
for i, order in enumerate(self.order_list):
# 序号
self.table.setItem(i, 0, QTableWidgetItem(str(i + 1)))
# 订单号
order_no = order.get('orderNo', '')
self.table.setItem(i, 1, QTableWidgetItem(order_no))
# 商品信息
brand_info = order.get('brandInfo', '')
self.table.setItem(i, 2, QTableWidgetItem(brand_info))
# IMEI
imei = order.get('imei', '')
self.table.setItem(i, 3, QTableWidgetItem(imei))
# 预估价格
price = order.get('price', '')
price_str = str(price) if price else '待定'
self.table.setItem(i, 4, QTableWidgetItem(price_str))
# 门店
title = order.get('title', '')
self.table.setItem(i, 5, QTableWidgetItem(title))
# 充电次数（初始为空，显示"查询"按钮）
charge_widget = QWidget()
charge_layout = QHBoxLayout(charge_widget)
charge_layout.setContentsMargins(5, 2, 5, 2)
charge_layout.setSpacing(5)
charge_label = QLabel("未查询")
charge_label.setObjectName(f"charge_label_{i}")
charge_layout.addWidget(charge_label)
check_btn = QPushButton("查询")
check_btn.setFixedSize(50, 25)
check_btn.setObjectName(f"check_btn_{i}")
check_btn.clicked.connect(lambda checked, idx=i: self.check_charge_count(idx))
charge_layout.addWidget(check_btn)
charge_widget.setLayout(charge_layout)
self.table.setCellWidget(i, 6, charge_widget)
def check_charge_count(self, index):
"""查询充电次数"""
if not self.order_manager:
return
if index >= len(self.order_list):
return
order_no = self.order_list[index].get('orderNo')
if not order_no:
return
self.log_message(f"开始查询订单 {order_no} 的充电次数", "INFO")
# 检查是否已缓存
if order_no in self.charge_counts:
count = self.charge_counts[order_no]
self.update_charge_display(index, count)
self.log_message(f"订单 {order_no} 充电次数: {count} (来自缓存)", "INFO")
return
self.status_bar.showMessage(f"正在查询订单 {order_no} 的充电次数...")
# 禁用按钮
btn = self.findChild(QPushButton, f"check_btn_{index}")
if btn:
btn.setEnabled(False)
btn.setText("查询中")
# 在后台线程执行
self.charge_worker = ChargeQueryWorker(self.order_manager, order_no, index)
self.charge_worker.finished.connect(self.on_charge_queried)
self.charge_worker.error.connect(self.on_charge_error)
self.charge_worker.start()
def on_charge_queried(self, index, charge_count, ocr_result, image_notices):
"""充电次数查询完成"""
order_no = self.order_list[index].get('orderNo')
# 如果没有找到充电次数，输出详细信息
if charge_count is None:
if ocr_result:
self.log_message(f"订单 {order_no} 未找到充电次数，OCR识别结果:", "WARNING")
ocr_text = self.order_manager.extract_text_from_ocr_result(ocr_result)
if ocr_text:
self.log_message(f"OCR文本: {ocr_text[:200]}...", "INFO")
else:
self.log_message(f"OCR文本为空", "WARNING")
else:
self.log_message(f"订单 {order_no} OCR识别失败", "ERROR")
if image_notices:
self.log_message(f"订单 {order_no} 找到的图片notice: {', '.join(image_notices)}", "INFO")
else:
self.log_message(f"订单 {order_no} 未找到验机图片", "ERROR")
else:
self.log_message(f"订单 {order_no} 充电次数: {charge_count}", "SUCCESS")
self.charge_counts[order_no] = charge_count
self.update_charge_display(index, charge_count)
btn = self.findChild(QPushButton, f"check_btn_{index}")
if btn:
btn.setEnabled(True)
btn.setText("查询")
if charge_count is not None:
self.status_bar.showMessage(f"订单 {order_no} 充电次数: {charge_count}")
else:
self.status_bar.showMessage(f"订单 {order_no} 未找到充电次数信息")
def on_charge_error(self, index, error_msg):
"""充电次数查询错误"""
order_no = self.order_list[index].get('orderNo')
self.log_message(f"订单 {order_no} 查询失败: {error_msg}", "ERROR")
btn = self.findChild(QPushButton, f"check_btn_{index}")
if btn:
btn.setEnabled(True)
btn.setText("查询")
self.status_bar.showMessage(f"查询失败: {error_msg}")
QMessageBox.warning(self, "查询失败", f"获取订单详情失败:\n{error_msg}")
def update_charge_display(self, index, charge_count):
"""更新充电次数显示"""
widget = self.table.cellWidget(index, 6)
if widget:
label = widget.findChild(QLabel)
if label:
if charge_count is not None:
label.setText(f"{charge_count} 次")
label.setStyleSheet("color: #28a745; font-weight: bold;")
else:
label.setText("未找到")
label.setStyleSheet("color: #dc3545;")
# --- 后台工作线程 ---
class OrderListWorker(QThread):
finished = pyqtSignal(list)
error = pyqtSignal(str)
def __init__(self, order_manager):
super().__init__()
self.order_manager = order_manager
def run(self):
try:
order_list, error = self.order_manager.get_pending_order_list()
if error:
self.error.emit(error)
else:
self.finished.emit(order_list)
except Exception as e:
self.error.emit(str(e))
class ChargeQueryWorker(QThread):
finished = pyqtSignal(int, object, object, list) # index, charge_count, ocr_result, image_notices
error = pyqtSignal(int, str)
def __init__(self, order_manager, order_no, index):
super().__init__()
self.order_manager = order_manager
self.order_no = order_no
self.index = index
def run(self):
try:
# 获取订单详情
order_detail, error = self.order_manager.get_order_detail(self.order_no)
if error:
self.error.emit(self.index, error)
return
# 查找所有图片的notice
image_notices = self.order_manager.get_all_image_notices(order_detail)
# 查找验机图片
report_img = self.order_manager.find_original_image_by_notice(order_detail, "苹果沙漏验机报告")
if report_img is None:
report_img = self.order_manager.find_original_image_by_notice(order_detail, "苹果沙漏验机基本信息")
if report_img is None:
report_img = self.order_manager.find_original_image_by_notice(order_detail, "验机照片")
if report_img is None:
report_img = self.order_manager.find_original_image_by_notice(order_detail, "苹果验机软件")
if not report_img:
# 没有找到图片，返回notice列表
self.finished.emit(self.index, None, None, image_notices)
return
# OCR识别
ocr_result = self.order_manager.download_and_ocr_image(report_img)
if not ocr_result:
# OCR失败，返回notice列表
self.finished.emit(self.index, None, None, image_notices)
return
# 提取充电次数
ocr_text = self.order_manager.extract_text_from_ocr_result(ocr_result)
charge_count = self.order_manager.extract_charge_count(ocr_text)
self.finished.emit(self.index, charge_count, ocr_result, image_notices)
except Exception as e:
self.error.emit(self.index, str(e))
# --- 主程序入口 ---
def main():
app = QApplication(sys.argv)
app.setStyle('Fusion')
# 设置应用图标
app.setWindowIcon(QIcon())
window = MainWindow()
window.show()
sys.exit(app.exec_())
if __name__ == "__main__":
main()