693 lines
22 KiB
Python
693 lines
22 KiB
Python
"""
|
|
中间图片预览面板 - 优化版本
|
|
主要优化:
|
|
1. SQLite查询移到后台线程
|
|
2. 使用线程池管理图片加载
|
|
3. 添加取消机制,避免翻页时的线程冲突
|
|
4. 正确清理图片对象,避免内存泄漏
|
|
"""
|
|
import customtkinter as ctk
|
|
from tkinter import messagebox
|
|
from typing import List, Dict, Callable, Optional
|
|
import threading
|
|
import sys
|
|
import os
|
|
from concurrent.futures import ThreadPoolExecutor, Future
|
|
from collections import defaultdict
|
|
|
|
# 添加父目录到路径
|
|
sys.path.append(os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
|
|
from config import GRID_COLUMNS, THUMBNAIL_SIZE, CARD_PADDING, ITEMS_PER_PAGE
|
|
from utils.image_loader import load_image_from_url, create_placeholder_image, get_thread_pool
|
|
|
|
|
|
class ImageCard(ctk.CTkFrame):
|
|
"""图片卡片组件 - 优化版本"""
|
|
|
|
def __init__(
|
|
self,
|
|
parent,
|
|
data: Dict,
|
|
is_downloaded: bool,
|
|
on_check_changed: Callable,
|
|
on_preview: Callable,
|
|
**kwargs
|
|
):
|
|
super().__init__(parent, **kwargs)
|
|
|
|
self.data = data
|
|
self.is_downloaded = is_downloaded
|
|
self.on_check_changed = on_check_changed
|
|
self.on_preview = on_preview
|
|
|
|
# 勾选状态
|
|
self.check_var = ctk.BooleanVar(value=False)
|
|
|
|
# 图片加载Future对象(用于取消)
|
|
self.load_future: Optional[Future] = None
|
|
|
|
# 标记是否已销毁
|
|
self._destroyed = False
|
|
|
|
self._create_widgets()
|
|
self._load_thumbnail()
|
|
|
|
def _create_widgets(self):
|
|
"""创建卡片组件"""
|
|
# 勾选框
|
|
self.checkbox = ctk.CTkCheckBox(
|
|
self,
|
|
text="",
|
|
variable=self.check_var,
|
|
command=self._on_check,
|
|
width=20
|
|
)
|
|
self.checkbox.pack(pady=(5, 0), padx=5, anchor="w")
|
|
|
|
# 图片容器
|
|
self.image_label = ctk.CTkLabel(
|
|
self,
|
|
text="加载中...",
|
|
width=THUMBNAIL_SIZE,
|
|
height=THUMBNAIL_SIZE
|
|
)
|
|
self.image_label.pack(pady=5, padx=5)
|
|
self.image_label.bind("<Button-1>", lambda e: self.on_preview(self.data))
|
|
|
|
# 物品名称(加粗)
|
|
name_label = ctk.CTkLabel(
|
|
self,
|
|
text=self.data['goods_name'],
|
|
font=("Arial", 11, "bold"),
|
|
wraplength=THUMBNAIL_SIZE - 10
|
|
)
|
|
name_label.pack(pady=(0, 2), padx=5)
|
|
|
|
# 采集时间
|
|
time_label = ctk.CTkLabel(
|
|
self,
|
|
text=self.data['create_time'],
|
|
font=("Arial", 9),
|
|
text_color="gray"
|
|
)
|
|
time_label.pack(pady=(0, 2), padx=5)
|
|
|
|
# 设备编码
|
|
if self.data['equipment_code']:
|
|
equipment_label = ctk.CTkLabel(
|
|
self,
|
|
text=f"设备: {self.data['equipment_code']}",
|
|
font=("Arial", 9),
|
|
text_color="gray"
|
|
)
|
|
equipment_label.pack(pady=(0, 2), padx=5)
|
|
|
|
# 已下载标记
|
|
if self.is_downloaded:
|
|
downloaded_label = ctk.CTkLabel(
|
|
self,
|
|
text="✅ 已下载",
|
|
font=("Arial", 10),
|
|
text_color="green"
|
|
)
|
|
downloaded_label.pack(pady=(0, 5), padx=5)
|
|
|
|
# 查看大图按钮
|
|
preview_btn = ctk.CTkButton(
|
|
self,
|
|
text="查看大图",
|
|
command=lambda: self.on_preview(self.data),
|
|
width=100,
|
|
height=25,
|
|
font=("Arial", 10)
|
|
)
|
|
preview_btn.pack(pady=(0, 5), padx=5)
|
|
|
|
def _load_thumbnail(self):
|
|
"""使用线程池异步加载缩略图"""
|
|
def load():
|
|
# 检查是否已销毁
|
|
if self._destroyed:
|
|
return
|
|
|
|
image = load_image_from_url(
|
|
self.data['image_url'],
|
|
size=(THUMBNAIL_SIZE, THUMBNAIL_SIZE)
|
|
)
|
|
|
|
# 在主线程更新UI
|
|
if not self._destroyed:
|
|
self.after(0, lambda: self._set_image(image))
|
|
|
|
# 使用全局线程池
|
|
thread_pool = get_thread_pool()
|
|
self.load_future = thread_pool.submit(load)
|
|
|
|
def _set_image(self, image):
|
|
"""设置图片(在主线程调用)"""
|
|
if self._destroyed:
|
|
return
|
|
|
|
try:
|
|
if image:
|
|
self.image_label.configure(image=image, text="")
|
|
self.image_label.image = image # 保持引用
|
|
else:
|
|
placeholder = create_placeholder_image(
|
|
size=(THUMBNAIL_SIZE, THUMBNAIL_SIZE)
|
|
)
|
|
self.image_label.configure(image=placeholder, text="")
|
|
self.image_label.image = placeholder
|
|
except Exception as e:
|
|
print(f"设置图片失败: {e}")
|
|
|
|
def _on_check(self):
|
|
"""勾选框改变"""
|
|
self.on_check_changed(self.data['id'], self.check_var.get())
|
|
|
|
def is_checked(self) -> bool:
|
|
"""是否被勾选"""
|
|
return self.check_var.get()
|
|
|
|
def set_checked(self, checked: bool):
|
|
"""设置勾选状态"""
|
|
self.check_var.set(checked)
|
|
|
|
def cancel_loading(self):
|
|
"""取消图片加载"""
|
|
if self.load_future and not self.load_future.done():
|
|
self.load_future.cancel()
|
|
|
|
def cleanup(self):
|
|
"""清理资源"""
|
|
self._destroyed = True
|
|
self.cancel_loading()
|
|
|
|
# 清理图片引用
|
|
if hasattr(self.image_label, 'image'):
|
|
self.image_label.image = None
|
|
|
|
|
|
class PreviewPanel(ctk.CTkFrame):
|
|
"""中间图片预览面板 - 优化版本"""
|
|
|
|
def __init__(
|
|
self,
|
|
parent,
|
|
sqlite_manager,
|
|
on_download_callback: Callable,
|
|
mysql_manager=None,
|
|
on_delete_callback: Callable = None,
|
|
**kwargs
|
|
):
|
|
super().__init__(parent, **kwargs)
|
|
|
|
self.sqlite_manager = sqlite_manager
|
|
self.mysql_manager = mysql_manager
|
|
self.on_download_callback = on_download_callback
|
|
self.on_delete_callback = on_delete_callback
|
|
|
|
# 数据
|
|
self.current_data = []
|
|
self.total_count = 0
|
|
self.current_page = 1
|
|
self.selected_ids = set()
|
|
|
|
# 下载状态缓存(避免重复查询SQLite)
|
|
self.download_status_cache = {}
|
|
|
|
# 当前的ImageCard列表
|
|
self.image_cards: List[ImageCard] = []
|
|
|
|
self._create_widgets()
|
|
|
|
def _create_widgets(self):
|
|
"""创建UI组件"""
|
|
# 标题栏
|
|
header_frame = ctk.CTkFrame(self)
|
|
header_frame.pack(fill="x", padx=15, pady=(15, 10))
|
|
|
|
title_label = ctk.CTkLabel(
|
|
header_frame,
|
|
text="图片预览",
|
|
font=("Arial", 16, "bold")
|
|
)
|
|
title_label.pack(side="left", padx=10)
|
|
|
|
# 统计信息
|
|
self.count_label = ctk.CTkLabel(
|
|
header_frame,
|
|
text="共 0 张 | 已选 0 张",
|
|
font=("Arial", 12),
|
|
text_color="gray"
|
|
)
|
|
self.count_label.pack(side="left", padx=20)
|
|
|
|
# 图片网格(可滚动)
|
|
self.scroll_frame = ctk.CTkScrollableFrame(self)
|
|
self.scroll_frame.pack(fill="both", expand=True, padx=15, pady=(0, 10))
|
|
|
|
# 网格容器
|
|
self.grid_frame = ctk.CTkFrame(self.scroll_frame)
|
|
self.grid_frame.pack(fill="both", expand=True)
|
|
|
|
# 底部操作栏
|
|
bottom_frame = ctk.CTkFrame(self)
|
|
bottom_frame.pack(fill="x", padx=15, pady=(0, 15))
|
|
|
|
# 左侧按钮
|
|
left_buttons = ctk.CTkFrame(bottom_frame)
|
|
left_buttons.pack(side="left", padx=10, pady=10)
|
|
|
|
self.select_all_btn = ctk.CTkButton(
|
|
left_buttons,
|
|
text="全选",
|
|
command=self._select_all,
|
|
width=80,
|
|
height=30
|
|
)
|
|
self.select_all_btn.pack(side="left", padx=5)
|
|
|
|
self.deselect_all_btn = ctk.CTkButton(
|
|
left_buttons,
|
|
text="取消全选",
|
|
command=self._deselect_all,
|
|
width=80,
|
|
height=30,
|
|
fg_color="gray",
|
|
hover_color="darkgray"
|
|
)
|
|
self.deselect_all_btn.pack(side="left", padx=5)
|
|
|
|
self.download_btn = ctk.CTkButton(
|
|
left_buttons,
|
|
text="批量下载",
|
|
command=self._on_download,
|
|
width=100,
|
|
height=30,
|
|
fg_color="green",
|
|
hover_color="darkgreen"
|
|
)
|
|
self.download_btn.pack(side="left", padx=5)
|
|
|
|
self.delete_btn = ctk.CTkButton(
|
|
left_buttons,
|
|
text="🗑️ 删除选中",
|
|
command=self._on_delete,
|
|
width=100,
|
|
height=30,
|
|
fg_color="red",
|
|
hover_color="darkred"
|
|
)
|
|
self.delete_btn.pack(side="left", padx=5)
|
|
|
|
# 分页控件
|
|
page_frame = ctk.CTkFrame(bottom_frame)
|
|
page_frame.pack(side="right", padx=10, pady=10)
|
|
|
|
self.first_page_btn = ctk.CTkButton(
|
|
page_frame,
|
|
text="首页",
|
|
command=self._first_page,
|
|
width=70,
|
|
height=30,
|
|
fg_color="#1f538d",
|
|
hover_color="#14375e"
|
|
)
|
|
self.first_page_btn.pack(side="left", padx=5)
|
|
|
|
self.prev_btn = ctk.CTkButton(
|
|
page_frame,
|
|
text="上一页",
|
|
command=self._prev_page,
|
|
width=80,
|
|
height=30
|
|
)
|
|
self.prev_btn.pack(side="left", padx=5)
|
|
|
|
self.page_label = ctk.CTkLabel(
|
|
page_frame,
|
|
text="第 1 页",
|
|
font=("Arial", 11)
|
|
)
|
|
self.page_label.pack(side="left", padx=10)
|
|
|
|
self.next_btn = ctk.CTkButton(
|
|
page_frame,
|
|
text="下一页",
|
|
command=self._next_page,
|
|
width=80,
|
|
height=30
|
|
)
|
|
self.next_btn.pack(side="left", padx=5)
|
|
|
|
def load_data(self, data_list: List[Dict], total_count: int, page: int = 1):
|
|
"""加载图片数据"""
|
|
self.current_data = data_list
|
|
self.total_count = total_count
|
|
self.current_page = page
|
|
self.selected_ids.clear()
|
|
|
|
# 在后台线程查询下载状态
|
|
def query_download_status():
|
|
try:
|
|
# 批量查询下载状态
|
|
status_dict = {}
|
|
for data in data_list:
|
|
is_downloaded = self.sqlite_manager.is_downloaded(
|
|
data['datasource'],
|
|
data['image_url']
|
|
)
|
|
status_dict[data['id']] = is_downloaded
|
|
|
|
# 在主线程更新UI
|
|
self.after(0, lambda: self._update_grid_with_status(status_dict))
|
|
|
|
except Exception as e:
|
|
print(f"查询下载状态失败: {e}")
|
|
# 失败时使用空字典
|
|
self.after(0, lambda: self._update_grid_with_status({}))
|
|
|
|
# 先显示加载中,然后在后台查询
|
|
self._show_loading()
|
|
threading.Thread(target=query_download_status, daemon=True).start()
|
|
|
|
def _show_loading(self):
|
|
"""显示加载中状态"""
|
|
# 清空现有卡片
|
|
self._clear_grid()
|
|
|
|
loading_label = ctk.CTkLabel(
|
|
self.grid_frame,
|
|
text="加载中...",
|
|
font=("Arial", 14),
|
|
text_color="gray"
|
|
)
|
|
loading_label.grid(row=0, column=0, pady=50, padx=50)
|
|
|
|
def _update_grid_with_status(self, status_dict: Dict[int, bool]):
|
|
"""使用下载状态更新网格"""
|
|
self.download_status_cache = status_dict
|
|
self._update_grid()
|
|
self._update_stats()
|
|
self._update_pagination()
|
|
|
|
def _update_grid(self):
|
|
"""更新图片网格"""
|
|
# 清空现有卡片
|
|
self._clear_grid()
|
|
|
|
if not self.current_data:
|
|
no_data_label = ctk.CTkLabel(
|
|
self.grid_frame,
|
|
text="暂无数据",
|
|
font=("Arial", 14),
|
|
text_color="gray"
|
|
)
|
|
no_data_label.grid(row=0, column=0, pady=50, padx=50)
|
|
return
|
|
|
|
# 创建网格卡片
|
|
for i, data in enumerate(self.current_data):
|
|
row = i // GRID_COLUMNS
|
|
col = i % GRID_COLUMNS
|
|
|
|
# 从缓存获取下载状态
|
|
is_downloaded = self.download_status_cache.get(data['id'], False)
|
|
|
|
card = ImageCard(
|
|
self.grid_frame,
|
|
data=data,
|
|
is_downloaded=is_downloaded,
|
|
on_check_changed=self._on_check_changed,
|
|
on_preview=self._on_preview
|
|
)
|
|
card.grid(
|
|
row=row,
|
|
column=col,
|
|
padx=CARD_PADDING,
|
|
pady=CARD_PADDING,
|
|
sticky="nsew"
|
|
)
|
|
|
|
# 保存引用
|
|
self.image_cards.append(card)
|
|
|
|
# 配置网格列权重
|
|
for col in range(GRID_COLUMNS):
|
|
self.grid_frame.grid_columnconfigure(col, weight=1)
|
|
|
|
def _clear_grid(self):
|
|
"""清空网格(正确清理资源)"""
|
|
# 先取消所有图片加载
|
|
for card in self.image_cards:
|
|
card.cleanup()
|
|
|
|
# 清空列表
|
|
self.image_cards.clear()
|
|
|
|
# 销毁所有子组件
|
|
for widget in self.grid_frame.winfo_children():
|
|
widget.destroy()
|
|
|
|
def _update_stats(self):
|
|
"""更新统计信息"""
|
|
selected_count = len(self.selected_ids)
|
|
self.count_label.configure(
|
|
text=f"共 {self.total_count} 张 | 已选 {selected_count} 张"
|
|
)
|
|
|
|
def _update_pagination(self):
|
|
"""更新分页控件"""
|
|
total_pages = (self.total_count + ITEMS_PER_PAGE - 1) // ITEMS_PER_PAGE
|
|
if total_pages == 0:
|
|
total_pages = 1
|
|
|
|
self.page_label.configure(text=f"第 {self.current_page}/{total_pages} 页")
|
|
|
|
# 更新按钮状态
|
|
self.first_page_btn.configure(state="normal" if self.current_page > 1 else "disabled")
|
|
self.prev_btn.configure(state="normal" if self.current_page > 1 else "disabled")
|
|
self.next_btn.configure(state="normal" if self.current_page < total_pages else "disabled")
|
|
|
|
def _on_check_changed(self, item_id: int, checked: bool):
|
|
"""勾选框改变"""
|
|
if checked:
|
|
self.selected_ids.add(item_id)
|
|
else:
|
|
self.selected_ids.discard(item_id)
|
|
|
|
self._update_stats()
|
|
|
|
def _select_all(self):
|
|
"""全选"""
|
|
for card in self.image_cards:
|
|
card.set_checked(True)
|
|
self.selected_ids.add(card.data['id'])
|
|
|
|
self._update_stats()
|
|
|
|
def _deselect_all(self):
|
|
"""取消全选"""
|
|
for card in self.image_cards:
|
|
card.set_checked(False)
|
|
|
|
self.selected_ids.clear()
|
|
self._update_stats()
|
|
|
|
def _on_download(self):
|
|
"""批量下载"""
|
|
if not self.selected_ids:
|
|
messagebox.showwarning("提示", "请先选择要下载的图片")
|
|
return
|
|
|
|
# 获取选中的图片数据
|
|
selected_data = [
|
|
data for data in self.current_data
|
|
if data['id'] in self.selected_ids
|
|
]
|
|
|
|
# 调用下载回调
|
|
if self.on_download_callback:
|
|
self.on_download_callback(selected_data)
|
|
|
|
def _on_delete(self):
|
|
"""删除选中的图片"""
|
|
if not self.selected_ids:
|
|
messagebox.showwarning("提示", "请先选择要删除的图片")
|
|
return
|
|
|
|
# 获取选中的图片数据
|
|
selected_data = [
|
|
data for data in self.current_data
|
|
if data['id'] in self.selected_ids
|
|
]
|
|
|
|
# 二次确认
|
|
count = len(selected_data)
|
|
confirm = messagebox.askyesno(
|
|
"确认删除",
|
|
f"确定要删除选中的 {count} 张图片吗?\n\n"
|
|
"此操作将:\n"
|
|
"1. 从MySQL数据库中删除记录\n"
|
|
"2. 删除已下载的本地文件(如有)\n\n"
|
|
"删除后无法恢复!",
|
|
icon='warning'
|
|
)
|
|
|
|
if not confirm:
|
|
return
|
|
|
|
# 在后台线程执行删除
|
|
def delete_thread():
|
|
try:
|
|
from config import DELETE_MODE
|
|
|
|
success_count = 0
|
|
failed_ids = []
|
|
|
|
for data in selected_data:
|
|
try:
|
|
# 模式B: 仅删除MySQL
|
|
if DELETE_MODE == 'remote':
|
|
# 删除MySQL记录
|
|
if self.mysql_manager:
|
|
deleted = self.mysql_manager.delete_records(
|
|
data['datasource'],
|
|
[data['id']]
|
|
)
|
|
if deleted > 0:
|
|
# 删除SQLite记录和本地文件
|
|
self.sqlite_manager.delete_download_record(
|
|
data['datasource'],
|
|
data['image_url'],
|
|
delete_file=True
|
|
)
|
|
success_count += 1
|
|
else:
|
|
failed_ids.append(data['id'])
|
|
else:
|
|
failed_ids.append(data['id'])
|
|
except Exception as e:
|
|
print(f"删除记录失败 (ID: {data['id']}): {e}")
|
|
failed_ids.append(data['id'])
|
|
|
|
# 更新UI(在主线程)
|
|
self.after(0, lambda: self._on_delete_complete(success_count, len(failed_ids)))
|
|
|
|
except Exception as e:
|
|
print(f"删除操作失败: {e}")
|
|
self.after(0, lambda: messagebox.showerror("错误", f"删除失败: {e}"))
|
|
|
|
threading.Thread(target=delete_thread, daemon=True).start()
|
|
|
|
def _on_delete_complete(self, success_count: int, failed_count: int):
|
|
"""删除完成回调"""
|
|
# 显示结果
|
|
if failed_count == 0:
|
|
messagebox.showinfo("删除成功", f"成功删除 {success_count} 张图片")
|
|
else:
|
|
messagebox.showwarning(
|
|
"部分删除失败",
|
|
f"成功删除 {success_count} 张图片\n失败 {failed_count} 张"
|
|
)
|
|
|
|
# 清空选中
|
|
self.selected_ids.clear()
|
|
|
|
# 调用删除回调,刷新数据
|
|
if self.on_delete_callback:
|
|
self.on_delete_callback()
|
|
|
|
def _on_preview(self, data: Dict):
|
|
"""查看大图"""
|
|
# 创建预览窗口
|
|
preview_window = ctk.CTkToplevel(self)
|
|
preview_window.title(f"预览 - {data['goods_name']}")
|
|
preview_window.geometry("800x700")
|
|
|
|
# 窗口属性
|
|
preview_window.transient(self)
|
|
preview_window.grab_set()
|
|
preview_window.lift()
|
|
preview_window.focus_set()
|
|
|
|
# 居中显示
|
|
preview_window.update_idletasks()
|
|
x = (preview_window.winfo_screenwidth() // 2) - 400
|
|
y = (preview_window.winfo_screenheight() // 2) - 350
|
|
preview_window.geometry(f"800x700+{x}+{y}")
|
|
|
|
# 信息框架
|
|
info_frame = ctk.CTkFrame(preview_window)
|
|
info_frame.pack(fill="x", padx=20, pady=(20, 10))
|
|
|
|
info_text = f"""
|
|
物品名称: {data['goods_name']}
|
|
采集时间: {data['create_time']}
|
|
设备编码: {data['equipment_code']}
|
|
数据源: {data['datasource']}
|
|
""".strip()
|
|
|
|
info_label = ctk.CTkLabel(
|
|
info_frame,
|
|
text=info_text,
|
|
font=("Arial", 12),
|
|
justify="left"
|
|
)
|
|
info_label.pack(pady=10, padx=10)
|
|
|
|
# 图片容器
|
|
image_container = ctk.CTkLabel(
|
|
preview_window,
|
|
text="加载中...",
|
|
width=760,
|
|
height=550
|
|
)
|
|
image_container.pack(pady=10, padx=20)
|
|
|
|
# 异步加载大图
|
|
def load_large_image():
|
|
image = load_image_from_url(data['image_url'], size=(760, 550))
|
|
if image:
|
|
image_container.configure(image=image, text="")
|
|
image_container.image = image
|
|
else:
|
|
image_container.configure(text="图片加载失败")
|
|
|
|
threading.Thread(target=load_large_image, daemon=True).start()
|
|
|
|
def _first_page(self):
|
|
"""首页"""
|
|
if self.current_page > 1:
|
|
# 需要重新查询数据(在主应用中处理)
|
|
pass
|
|
|
|
def _prev_page(self):
|
|
"""上一页"""
|
|
if self.current_page > 1:
|
|
# 需要重新查询数据(在主应用中处理)
|
|
pass
|
|
|
|
def _next_page(self):
|
|
"""下一页"""
|
|
total_pages = (self.total_count + ITEMS_PER_PAGE - 1) // ITEMS_PER_PAGE
|
|
if self.current_page < total_pages:
|
|
# 需要重新查询数据(在主应用中处理)
|
|
pass
|
|
|
|
def get_current_page(self) -> int:
|
|
"""获取当前页码"""
|
|
return self.current_page
|
|
|
|
def clear(self):
|
|
"""清空数据"""
|
|
self.current_data = []
|
|
self.total_count = 0
|
|
self.current_page = 1
|
|
self.selected_ids.clear()
|
|
self.download_status_cache.clear()
|
|
self._clear_grid()
|
|
self._update_stats()
|
|
self._update_pagination()
|