import sys
import requests
import json
import os
import re
import uuid
import hashlib
import time
import threading
from concurrent.futures import ThreadPoolExecutor, as_completed
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 *
from requests.adapters import HTTPAdapter
from urllib3.util.retry import Retry
# --- 图片预览选择对话框 ---
class ImageSelectionDialog(QDialog):
"""图片选择对话框 - 纯主线程，无定时器，无子线程"""
def __init__(self, image_urls_with_notice, parent=None):
super().__init__(parent)
self.image_urls_with_notice = image_urls_with_notice
self.selected_url = None
self.selected_index = -1
self.skip = False
self.image_cache = {}
self.init_ui()
# 直接加载，不用定时器
self.load_images()
def init_ui(self):
self.setWindowTitle("选择验机图片")
self.setGeometry(200, 200, 900, 600)
main_layout = QVBoxLayout(self)
hint_label = QLabel("请选择要识别的图片（双击预览大图）:")
hint_label.setStyleSheet("font-weight: bold; font-size: 12px; padding: 5px;")
main_layout.addWidget(hint_label)
self.list_widget = QListWidget()
self.list_widget.setIconSize(QSize(150, 100))
self.list_widget.setViewMode(QListWidget.IconMode)
self.list_widget.setResizeMode(QListWidget.Adjust)
self.list_widget.setGridSize(QSize(180, 140))
self.list_widget.setSpacing(10)
main_layout.addWidget(self.list_widget)
btn_layout = QHBoxLayout()
self.preview_btn = QPushButton("🔍 预览大图")
self.preview_btn.clicked.connect(self.preview_selected)
btn_layout.addWidget(self.preview_btn)
self.select_btn = QPushButton("✅ 选择此图片识别")
self.select_btn.clicked.connect(self.select_image)
self.select_btn.setStyleSheet("background-color: #28a745; color: white;")
btn_layout.addWidget(self.select_btn)
self.skip_btn = QPushButton("⏭️ 跳过此订单")
self.skip_btn.clicked.connect(self.skip_order)
self.skip_btn.setStyleSheet("background-color: #dc3545; color: white;")
btn_layout.addWidget(self.skip_btn)
self.cancel_btn = QPushButton("❌ 取消全部")
self.cancel_btn.clicked.connect(self.reject)
self.cancel_btn.setStyleSheet("background-color: #6c757d; color: white;")
btn_layout.addWidget(self.cancel_btn)
main_layout.addLayout(btn_layout)
self.status_label = QLabel("正在加载图片...")
self.status_label.setStyleSheet("color: #ffa500; padding: 5px;")
main_layout.addWidget(self.status_label)
# 直接连接信号，不用QueuedConnection
self.list_widget.itemSelectionChanged.connect(self.on_selection_changed)
self.list_widget.itemDoubleClicked.connect(self.preview_selected)
def load_images(self):
"""直接在主线程加载图片"""
if not self.image_urls_with_notice:
self.status_label.setText("没有找到图片")
self.status_label.setStyleSheet("color: #dc3545; padding: 5px;")
return
self.status_label.setText(f"正在加载图片... 共 {len(self.image_urls_with_notice)} 张")
QApplication.processEvents()
# 先创建所有列表项
for i, (notice, url) in enumerate(self.image_urls_with_notice):
display_text = notice[:25] + "..." if len(notice) > 25 else notice
item = QListWidgetItem()
item.setText(display_text)
item.setToolTip(f"完整信息: {notice}\nURL: {url}")
item.setData(Qt.UserRole, url)
item.setData(Qt.UserRole + 1, i)
item.setData(Qt.UserRole + 2, notice)
item.setIcon(self.style().standardIcon(QStyle.SP_FileIcon))
self.list_widget.addItem(item)
# 直接在主线程下载图片
loaded_count = 0
for i, (notice, url) in enumerate(self.image_urls_with_notice):
try:
response = requests.get(url, timeout=10)
if response.status_code == 200:
img_data = response.content
pixmap = QPixmap()
pixmap.loadFromData(img_data)
if not pixmap.isNull():
scaled = pixmap.scaled(150, 100, Qt.KeepAspectRatio, Qt.SmoothTransformation)
item = self.list_widget.item(i)
if item:
item.setIcon(QIcon(scaled))
self.image_cache[url] = pixmap
loaded_count += 1
self.status_label.setText(f"加载图片 {i+1}/{len(self.image_urls_with_notice)}")
QApplication.processEvents()
except Exception as e:
print(f"加载图片失败: {e}")
self.status_label.setText(f"加载完成，共加载 {loaded_count}/{len(self.image_urls_with_notice)} 张图片")
self.status_label.setStyleSheet("color: #28a745; padding: 5px;")
if self.list_widget.count() > 0:
self.list_widget.setCurrentRow(0)
def on_selection_changed(self):
selected = self.list_widget.currentItem()
if selected:
notice = selected.data(Qt.UserRole + 2)
self.status_label.setText(f"已选择: {notice[:50]}...")
self.status_label.setStyleSheet("color: #0078d7; padding: 5px;")
else:
self.status_label.setText("请选择一个图片")
self.status_label.setStyleSheet("color: #666; padding: 5px;")
def preview_selected(self):
selected = self.list_widget.currentItem()
if not selected:
QMessageBox.warning(self, "提示", "请先选择一个图片")
return
url = selected.data(Qt.UserRole)
notice = selected.data(Qt.UserRole + 2)
preview_dialog = QDialog(self)
preview_dialog.setWindowTitle(f"图片预览: {notice[:30]}...")
preview_dialog.setGeometry(300, 300, 800, 700)
layout = QVBoxLayout(preview_dialog)
scroll_area = QScrollArea()
scroll_area.setWidgetResizable(True)
label = QLabel()
label.setAlignment(Qt.AlignCenter)
label.setStyleSheet("background-color: #f0f0f0; padding: 10px;")
if url in self.image_cache:
pixmap = self.image_cache[url]
if pixmap and not pixmap.isNull():
scaled = pixmap.scaled(700, 600, Qt.KeepAspectRatio, Qt.SmoothTransformation)
label.setPixmap(scaled)
else:
label.setText("图片加载失败")
else:
try:
response = requests.get(url, timeout=10)
if response.status_code == 200:
pixmap = QPixmap()
pixmap.loadFromData(response.content)
if not pixmap.isNull():
self.image_cache[url] = pixmap
scaled = pixmap.scaled(700, 600, Qt.KeepAspectRatio, Qt.SmoothTransformation)
label.setPixmap(scaled)
else:
label.setText("图片加载失败")
else:
label.setText("图片加载失败")
except Exception as e:
label.setText(f"加载失败")
scroll_area.setWidget(label)
layout.addWidget(scroll_area)
close_btn = QPushButton("关闭")
close_btn.clicked.connect(preview_dialog.accept)
close_btn.setFixedHeight(35)
layout.addWidget(close_btn)
preview_dialog.exec_()
def select_image(self):
selected = self.list_widget.currentItem()
if not selected:
QMessageBox.warning(self, "提示", "请先选择一个图片")
return
self.selected_url = selected.data(Qt.UserRole)
self.selected_index = selected.data(Qt.UserRole + 1)
self.accept()
def skip_order(self):
reply = QMessageBox.question(self, "确认跳过",
"确定要跳过此订单吗？",
QMessageBox.Yes | QMessageBox.No,
QMessageBox.No)
if reply == QMessageBox.Yes:
self.skip = True
self.accept()
def get_selected_url(self):
return self.selected_url
def is_skipped(self):
return self.skip
# --- 核心功能类 ---
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"
self.order_detail_cache = {}
self.ocr_cache = {}
self.cache_lock = threading.Lock()
self.session = requests.Session()
retry_strategy = Retry(
total=2,
backoff_factor=0.5,
status_forcelist=[429, 500, 502, 503, 504],
)
adapter = HTTPAdapter(
pool_connections=20,
pool_maxsize=20,
max_retries=retry_strategy
)
self.session.mount('http://', adapter)
self.session.mount('https://', adapter)
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 = self.session.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):
with self.cache_lock:
if order_no in self.order_detail_cache:
return self.order_detail_cache[order_no], None
snatch_url = "https://wirelessgate.aihuishou.com/jdx-rc-service/front/bid/snatch"
snatch_payload = {"orderNo": order_no}
try:
snatch_response = self.session.post(snatch_url, headers=self.headers, json=snatch_payload, timeout=15)
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 = self.session.get(detail_url, headers=self.headers, timeout=15)
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)
with self.cache_lock:
self.order_detail_cache[order_no] = order_detail
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
stack = [json_data]
keyword_lower = notice_keyword.lower()
while stack:
obj = stack.pop()
if isinstance(obj, dict):
if 'notice' in obj and 'originalImage' in obj:
if keyword_lower in obj['notice'].lower():
return obj['originalImage']
for value in obj.values():
stack.append(value)
elif isinstance(obj, list):
stack.extend(obj)
return None
def get_all_images_with_notice(self, json_data):
images = []
stack = [json_data]
while stack:
obj = stack.pop()
if isinstance(obj, dict):
if 'notice' in obj and 'originalImage' in obj:
images.append((obj['notice'], obj['originalImage']))
for value in obj.values():
stack.append(value)
elif isinstance(obj, list):
stack.extend(obj)
return images
def get_all_image_notices(self, json_data):
notices = []
stack = [json_data]
while stack:
obj = stack.pop()
if isinstance(obj, dict):
if 'notice' in obj and 'image' in obj:
notices.append(obj['notice'])
for value in obj.values():
stack.append(value)
elif isinstance(obj, list):
stack.extend(obj)
return list(set(notices))
def ocr_by_url(self, image_url):
cache_key = hashlib.md5(image_url.encode()).hexdigest()
with self.cache_lock:
if cache_key in self.ocr_cache:
return self.ocr_cache[cache_key]
try:
payload = {"url": image_url}
response = self.session.post(self.ocr_url, json=payload, timeout=30)
if response.status_code == 200:
ocr_result = response.json()
if ocr_result.get('success'):
with self.cache_lock:
self.ocr_cache[cache_key] = ocr_result
return ocr_result
return None
except Exception as e:
return None
def extract_text_from_ocr_result(self, ocr_result):
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):
if not text:
return None
pattern1 = r'[（(]\s*(\d+)\s*次\s*循环\s*[）)]'
match = re.search(pattern1, text)
if match:
return int(match.group(1))
pattern2 = r'充电次数(\d+)次'
match = re.search(pattern2, text)
if match:
return int(match.group(1))
pattern3 = r'(\d+)次'
match = re.search(pattern3, text)
if match:
return int(match.group(1))
return None
def find_best_image(self, order_detail, parent_widget=None):
"""查找最佳图片，如果找不到则让用户选择"""
keywords = [
"苹果沙漏验机报告",
"沙漏图",
"苹果验机软件",
"苹果沙漏验机基本信息",
"验机照片",
"验机报告"
]
# 先尝试用关键词自动匹配
for keyword in keywords:
img = self.find_original_image_by_notice(order_detail, keyword)
if img:
return img, "auto"
# 自动匹配失败，获取所有图片让用户选择
all_images = self.get_all_images_with_notice(order_detail)
if not all_images:
return None, "no_images"
# 如果只有一张图片，直接返回
if len(all_images) == 1:
return all_images[0][1], "auto_single"
# 多张图片，让用户选择
if parent_widget:
# 使用信号槽方式在主线程显示对话框
result_container = {"action": None, "url": None}
event = threading.Event()
def show_dialog():
try:
dialog = ImageSelectionDialog(all_images, parent_widget)
if dialog.exec_() == QDialog.Accepted:
if dialog.is_skipped():
result_container["action"] = "skip"
else:
selected_url = dialog.get_selected_url()
if selected_url:
result_container["action"] = "selected"
result_container["url"] = selected_url
else:
result_container["action"] = "cancelled"
else:
result_container["action"] = "cancelled"
except Exception as e:
print(f"显示对话框出错: {e}")
result_container["action"] = "error"
finally:
event.set()
# 在主线程执行
if QThread.currentThread() != QCoreApplication.instance().thread():
# 在子线程中，使用invokeMethod
QMetaObject.invokeMethod(parent_widget, "show_image_selection_dialog",
Qt.BlockingQueuedConnection,
Q_ARG(list, all_images),
Q_ARG(dict, result_container),
Q_ARG(threading.Event, event))
else:
# 已经在主线程，直接调用
show_dialog()
# 等待对话框完成
event.wait()
action = result_container.get("action")
if action == "selected":
return result_container.get("url"), "user_selected"
elif action == "skip":
return None, "skipped"
else:
return None, "cancelled"
# 没有父窗口，返回第一张图片
return all_images[0][1], "auto_first"
# --- 后台工作线程 ---
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(100)
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)
error = pyqtSignal(int, str)
def __init__(self, order_manager, order_no, index, parent_widget=None):
super().__init__()
self.order_manager = order_manager
self.order_no = order_no
self.index = index
self.parent_widget = parent_widget
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
image_notices = self.order_manager.get_all_image_notices(order_detail)
# 传入 parent_widget 以便显示对话框
report_img, source = self.order_manager.find_best_image(order_detail, self.parent_widget)
# 处理用户取消或跳过的情况
if source in ["cancelled", "skipped"]:
self.error.emit(self.index, "用户取消选择" if source == "cancelled" else "用户跳过订单")
return
if not report_img:
self.finished.emit(self.index, None, None, image_notices)
return
ocr_result = self.order_manager.ocr_by_url(report_img)
if not ocr_result:
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))
# --- 主窗口 ---
class MainWindow(QMainWindow):
def __init__(self):
super().__init__()
self.order_manager = None
self.order_list = []
self.order_details_cache = {}
self.charge_counts = {}
self.is_batch_processing = False
self.batch_queue = []
self.current_batch_index = 0
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;
}
QPushButton#batch_btn {
background-color: #28a745;
}
QPushButton#batch_btn:hover {
background-color: #218838;
}
QPushButton#batch_btn: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;
}
QProgressBar {
border: 1px solid #ccc;
border-radius: 4px;
text-align: center;
}
QProgressBar::chunk {
background-color: #0078d7;
border-radius: 4px;
}
""")
central_widget = QWidget()
self.setCentralWidget(central_widget)
main_layout = QVBoxLayout(central_widget)
main_layout.setSpacing(10)
config_group = QGroupBox("配置")
config_layout = QGridLayout()
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)
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)
config_layout.addWidget(QLabel("OCR地址:"), 1, 0)
self.ocr_input = QLineEdit()
self.ocr_input.setText("http://111.229.45.38:19100/ocr/url")
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)
self.batch_btn = QPushButton("📋 批量查询充电次数")
self.batch_btn.setObjectName("batch_btn")
self.batch_btn.clicked.connect(self.batch_check_charge_counts)
self.batch_btn.setFixedHeight(35)
self.batch_btn.setEnabled(False)
btn_layout.addWidget(self.batch_btn)
btn_layout.addStretch()
config_layout.addLayout(btn_layout, 3, 0, 1, 3)
config_group.setLayout(config_layout)
main_layout.addWidget(config_group)
list_group = QGroupBox("订单列表")
list_layout = QVBoxLayout()
self.progress_bar = QProgressBar()
self.progress_bar.setVisible(False)
list_layout.addWidget(self.progress_bar)
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, 120)
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("就绪")
@pyqtSlot(list, dict, threading.Event)
def show_image_selection_dialog(self, all_images, result_container, event):
"""在主线程显示图片选择对话框（供invokeMethod调用）"""
try:
dialog = ImageSelectionDialog(all_images, self)
if dialog.exec_() == QDialog.Accepted:
if dialog.is_skipped():
result_container["action"] = "skip"
else:
selected_url = dialog.get_selected_url()
if selected_url:
result_container["action"] = "selected"
result_container["url"] = selected_url
else:
result_container["action"] = "cancelled"
else:
result_container["action"] = "cancelled"
except Exception as e:
print(f"显示图片选择对话框出错: {e}")
result_container["action"] = "error"
finally:
event.set()
def toggle_token_visibility(self):
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",
"PROGRESS": "#00bcd4"
}
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.batch_btn.setEnabled(False)
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.batch_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.batch_btn.setEnabled(False)
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 = 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.setFocusPolicy(Qt.NoFocus)
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")
self.status_bar.showMessage(f"正在查询订单 {order_no} 的充电次数...")
btn = self.findChild(QPushButton, f"check_btn_{index}")
if btn:
btn.setEnabled(False)
btn.setText("稍等")
# 传入 self 作为 parent_widget
self.charge_worker = ChargeQueryWorker(self.order_manager, order_no, index, self)
self.charge_worker.finished.connect(self.on_charge_queried)
self.charge_worker.error.connect(self.on_charge_error)
self.charge_worker.start()
def batch_check_charge_counts(self):
"""串行批量查询充电次数（保留弹窗功能）"""
if not self.order_list:
QMessageBox.warning(self, "提示", "请先搜索订单")
return
if self.is_batch_processing:
return
pending_indices = []
for i, order in enumerate(self.order_list):
order_no = order.get('orderNo')
if order_no not in self.charge_counts:
pending_indices.append(i)
if not pending_indices:
self.log_message("所有订单已查询完成", "SUCCESS")
QMessageBox.information(self, "提示", "所有订单已查询完成")
return
self.is_batch_processing = True
self.batch_queue = pending_indices
self.current_batch_index = 0
self.batch_btn.setEnabled(False)
self.search_btn.setEnabled(False)
self.refresh_btn.setEnabled(False)
self.progress_bar.setVisible(True)
self.progress_bar.setMaximum(len(pending_indices))
self.progress_bar.setValue(0)
self.log_message(f"开始批量查询 {len(pending_indices)} 个订单", "PROGRESS")
self.process_next_batch()
def process_next_batch(self):
if self.current_batch_index >= len(self.batch_queue):
self.on_batch_complete()
return
index = self.batch_queue[self.current_batch_index]
order_no = self.order_list[index].get('orderNo')
self.progress_bar.setValue(self.current_batch_index + 1)
self.status_bar.showMessage(f"处理中: {self.current_batch_index + 1}/{len(self.batch_queue)}")
btn = self.findChild(QPushButton, f"check_btn_{index}")
if btn:
btn.setEnabled(False)
btn.setText("稍等")
# 传入 self 作为 parent_widget
self.batch_worker = ChargeQueryWorker(self.order_manager, order_no, index, self)
self.batch_worker.finished.connect(self.on_batch_item_complete)
self.batch_worker.error.connect(self.on_batch_item_error)
self.batch_worker.start()
def on_batch_item_complete(self, index, charge_count, ocr_result, image_notices):
order_no = self.order_list[index].get('orderNo')
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.log_message(f"[{self.current_batch_index + 1}/{len(self.batch_queue)}] 订单 {order_no} 充电次数: {charge_count}", "SUCCESS")
else:
self.log_message(f"[{self.current_batch_index + 1}/{len(self.batch_queue)}] 订单 {order_no} 未找到充电次数", "WARNING")
self.current_batch_index += 1
QTimer.singleShot(300, self.process_next_batch)
def on_batch_item_error(self, index, error_msg):
order_no = self.order_list[index].get('orderNo')
self.log_message(f"[{self.current_batch_index + 1}/{len(self.batch_queue)}] 订单 {order_no} 查询失败: {error_msg}", "ERROR")
btn = self.findChild(QPushButton, f"check_btn_{index}")
if btn:
btn.setEnabled(True)
btn.setText("查询")
self.current_batch_index += 1
QTimer.singleShot(300, self.process_next_batch)
def on_batch_complete(self):
self.is_batch_processing = False
self.batch_queue = []
self.batch_btn.setEnabled(True)
self.search_btn.setEnabled(True)
self.refresh_btn.setEnabled(True)
self.progress_bar.setVisible(False)
success_count = sum(1 for v in self.charge_counts.values() if v is not None)
total_count = len(self.charge_counts)
self.status_bar.showMessage(f"批量查询完成，成功 {success_count}/{total_count}")
self.log_message(f"批量查询完成，成功 {success_count}/{total_count}", "SUCCESS")
QMessageBox.information(self, "完成", f"批量查询完成!\n\n成功找到充电次数: {success_count}/{total_count}")
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}", "INFO")
else:
self.log_message(f"OCR文本为空", "WARNING")
if not image_notices:
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}")
if error_msg != "用户取消选择" and 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;")
def closeEvent(self, event):
event.accept()
# --- 主程序入口 ---
def main():
app = QApplication(sys.argv)
app.setStyle('Fusion')
window = MainWindow()
window.show()
sys.exit(app.exec_())
if __name__ == "__main__":
main()