478 lines
16 KiB
Python
478 lines
16 KiB
Python
"""
|
|
文件浏览器组件
|
|
"""
|
|
import os
|
|
from typing import Optional, List
|
|
|
|
from PyQt5.QtWidgets import (
|
|
QWidget, QVBoxLayout, QHBoxLayout, QTreeView,
|
|
QTableView, QHeaderView, QMenu, QAction,
|
|
QMessageBox, QApplication, QSplitter, QLabel,
|
|
QPushButton, QProgressBar
|
|
)
|
|
from PyQt5.QtCore import (
|
|
Qt, QSortFilterProxyModel, QModelIndex,
|
|
QThread, pyqtSignal, QAbstractTableModel
|
|
)
|
|
from PyQt5.QtGui import QFont, QColor, QIcon, QCursor
|
|
|
|
from file_scanner import FileScanner, FileItem, ScanProgress
|
|
from utils import format_size, safe_delete
|
|
|
|
|
|
class FileTableModel(QAbstractTableModel):
|
|
"""文件表格数据模型"""
|
|
|
|
HEADERS = ["名称", "大小", "修改时间", "权限", "所有者"]
|
|
|
|
def __init__(self, parent=None):
|
|
super().__init__(parent)
|
|
self.files: List[FileItem] = []
|
|
self.current_path = ""
|
|
|
|
def rowCount(self, parent=QModelIndex()):
|
|
return len(self.files)
|
|
|
|
def columnCount(self, parent=QModelIndex()):
|
|
return len(self.HEADERS)
|
|
|
|
def data(self, index, role=Qt.DisplayRole):
|
|
if not index.isValid() or index.row() >= len(self.files):
|
|
return None
|
|
|
|
file_item = self.files[index.row()]
|
|
column = index.column()
|
|
|
|
if role == Qt.DisplayRole:
|
|
if column == 0:
|
|
return f"{file_item.icon} {file_item.name}"
|
|
elif column == 1:
|
|
return file_item.size_formatted
|
|
elif column == 2:
|
|
return file_item.modified_time_formatted
|
|
elif column == 3:
|
|
return file_item.permissions
|
|
elif column == 4:
|
|
return f"{file_item.owner}:{file_item.group}"
|
|
|
|
elif role == Qt.UserRole:
|
|
# 返回原始数据用于排序
|
|
if column == 0:
|
|
return file_item.name
|
|
elif column == 1:
|
|
return file_item.size
|
|
elif column == 2:
|
|
return file_item.modified_time
|
|
elif column == 3:
|
|
return file_item.permissions
|
|
elif column == 4:
|
|
return file_item.owner
|
|
|
|
elif role == Qt.TextAlignmentRole:
|
|
if column == 1:
|
|
return Qt.AlignRight | Qt.AlignVCenter
|
|
|
|
elif role == Qt.ForegroundRole:
|
|
if column == 1:
|
|
if file_item.size > 1024 * 1024 * 1024: # > 1GB
|
|
return QColor(255, 0, 0)
|
|
elif file_item.size > 100 * 1024 * 1024: # > 100MB
|
|
return QColor(255, 165, 0)
|
|
|
|
return None
|
|
|
|
def headerData(self, section, orientation, role=Qt.DisplayRole):
|
|
if orientation == Qt.Horizontal and role == Qt.DisplayRole:
|
|
return self.HEADERS[section]
|
|
return None
|
|
|
|
def set_files(self, files: List[FileItem]):
|
|
"""设置文件列表"""
|
|
self.beginResetModel()
|
|
self.files = files
|
|
self.endResetModel()
|
|
|
|
def get_file_at(self, row: int) -> Optional[FileItem]:
|
|
"""获取指定行的文件信息"""
|
|
if 0 <= row < len(self.files):
|
|
return self.files[row]
|
|
return None
|
|
|
|
|
|
class SortProxyModel(QSortFilterProxyModel):
|
|
"""排序代理模型"""
|
|
|
|
def __init__(self, parent=None):
|
|
super().__init__(parent)
|
|
self.setSortRole(Qt.UserRole)
|
|
|
|
def lessThan(self, left, right):
|
|
"""自定义排序逻辑"""
|
|
left_data = self.sourceModel().data(left, Qt.UserRole)
|
|
right_data = self.sourceModel().data(right, Qt.UserRole)
|
|
|
|
if left_data is None:
|
|
return True
|
|
if right_data is None:
|
|
return False
|
|
|
|
# 目录始终排在前面
|
|
left_item = self.sourceModel().get_file_at(left.row())
|
|
right_item = self.sourceModel().get_file_at(right.row())
|
|
|
|
if left_item and right_item:
|
|
if left_item.is_dir != right_item.is_dir:
|
|
return left_item.is_dir
|
|
|
|
try:
|
|
return left_data < right_data
|
|
except TypeError:
|
|
return str(left_data) < str(right_data)
|
|
|
|
|
|
class DirectoryLoadThread(QThread):
|
|
"""目录加载线程"""
|
|
finished = pyqtSignal(list, str)
|
|
error = pyqtSignal(str)
|
|
|
|
def __init__(self, scanner: FileScanner, path: str):
|
|
super().__init__()
|
|
self.scanner = scanner
|
|
self.path = path
|
|
|
|
def run(self):
|
|
try:
|
|
files = self.scanner.list_directory(self.path)
|
|
self.finished.emit(files, self.path)
|
|
except Exception as e:
|
|
self.error.emit(str(e))
|
|
|
|
|
|
class FileBrowser(QWidget):
|
|
"""文件浏览器组件"""
|
|
|
|
# 信号
|
|
file_selected = pyqtSignal(FileItem) # 文件被选中
|
|
directory_changed = pyqtSignal(str) # 目录改变
|
|
files_deleted = pyqtSignal(int) # 文件被删除（释放的空间）
|
|
|
|
def __init__(self, scanner: FileScanner, sudo_password: Optional[str] = None, parent=None):
|
|
super().__init__(parent)
|
|
self.scanner = scanner
|
|
self.sudo_password = sudo_password
|
|
self.current_path = ""
|
|
self.path_history = []
|
|
self.history_index = -1
|
|
self.init_ui()
|
|
|
|
def init_ui(self):
|
|
"""初始化界面"""
|
|
layout = QVBoxLayout(self)
|
|
layout.setContentsMargins(0, 0, 0, 0)
|
|
layout.setSpacing(5)
|
|
|
|
# 工具栏
|
|
toolbar_layout = QHBoxLayout()
|
|
toolbar_layout.setSpacing(5)
|
|
|
|
# 后退按钮
|
|
self.back_button = QPushButton("◀")
|
|
self.back_button.setFixedSize(30, 30)
|
|
self.back_button.setToolTip("后退")
|
|
self.back_button.clicked.connect(self.go_back)
|
|
self.back_button.setEnabled(False)
|
|
toolbar_layout.addWidget(self.back_button)
|
|
|
|
# 前进按钮
|
|
self.forward_button = QPushButton("▶")
|
|
self.forward_button.setFixedSize(30, 30)
|
|
self.forward_button.setToolTip("前进")
|
|
self.forward_button.clicked.connect(self.go_forward)
|
|
self.forward_button.setEnabled(False)
|
|
toolbar_layout.addWidget(self.forward_button)
|
|
|
|
# 上级目录按钮
|
|
self.up_button = QPushButton("⬆")
|
|
self.up_button.setFixedSize(30, 30)
|
|
self.up_button.setToolTip("上级目录")
|
|
self.up_button.clicked.connect(self.go_up)
|
|
toolbar_layout.addWidget(self.up_button)
|
|
|
|
# 刷新按钮
|
|
self.refresh_button = QPushButton("🔄")
|
|
self.refresh_button.setFixedSize(30, 30)
|
|
self.refresh_button.setToolTip("刷新")
|
|
self.refresh_button.clicked.connect(self.refresh)
|
|
toolbar_layout.addWidget(self.refresh_button)
|
|
|
|
# 当前路径显示
|
|
self.path_label = QLabel()
|
|
self.path_label.setStyleSheet(
|
|
"QLabel { background-color: #f0f0f0; padding: 5px; border: 1px solid #ccc; }"
|
|
)
|
|
toolbar_layout.addWidget(self.path_label, 1)
|
|
|
|
# 加载状态
|
|
self.loading_label = QLabel("⏳")
|
|
self.loading_label.setFixedSize(30, 30)
|
|
self.loading_label.setAlignment(Qt.AlignCenter)
|
|
self.loading_label.hide()
|
|
toolbar_layout.addWidget(self.loading_label)
|
|
|
|
layout.addLayout(toolbar_layout)
|
|
|
|
# 文件表格
|
|
self.table_view = QTableView()
|
|
self.table_view.setSelectionBehavior(QTableView.SelectRows)
|
|
self.table_view.setSelectionMode(QTableView.ExtendedSelection)
|
|
self.table_view.setEditTriggers(QTableView.NoEditTriggers)
|
|
self.table_view.setSortingEnabled(True)
|
|
self.table_view.setContextMenuPolicy(Qt.CustomContextMenu)
|
|
self.table_view.customContextMenuRequested.connect(self.show_context_menu)
|
|
self.table_view.doubleClicked.connect(self.on_item_double_clicked)
|
|
self.table_view.clicked.connect(self.on_item_clicked)
|
|
|
|
# 设置数据模型
|
|
self.model = FileTableModel()
|
|
self.proxy_model = SortProxyModel()
|
|
self.proxy_model.setSourceModel(self.model)
|
|
self.table_view.setModel(self.proxy_model)
|
|
|
|
# 设置列宽
|
|
header = self.table_view.horizontalHeader()
|
|
header.setSectionResizeMode(0, QHeaderView.Stretch)
|
|
header.setSectionResizeMode(1, QHeaderView.ResizeToContents)
|
|
header.setSectionResizeMode(2, QHeaderView.ResizeToContents)
|
|
header.setSectionResizeMode(3, QHeaderView.ResizeToContents)
|
|
header.setSectionResizeMode(4, QHeaderView.ResizeToContents)
|
|
|
|
# 默认按大小降序排序
|
|
self.table_view.sortByColumn(1, Qt.DescendingOrder)
|
|
|
|
layout.addWidget(self.table_view)
|
|
|
|
# 状态栏
|
|
status_layout = QHBoxLayout()
|
|
|
|
self.status_label = QLabel("就绪")
|
|
status_layout.addWidget(self.status_label)
|
|
|
|
status_layout.addStretch()
|
|
|
|
self.item_count_label = QLabel()
|
|
status_layout.addWidget(self.item_count_label)
|
|
|
|
layout.addLayout(status_layout)
|
|
|
|
def load_directory(self, path: str):
|
|
"""加载目录内容"""
|
|
if not os.path.isdir(path):
|
|
QMessageBox.warning(self, "错误", f"目录不存在：{path}")
|
|
return
|
|
|
|
self.current_path = path
|
|
|
|
# 更新路径历史
|
|
if self.history_index < len(self.path_history) - 1:
|
|
self.path_history = self.path_history[:self.history_index + 1]
|
|
|
|
if not self.path_history or self.path_history[-1] != path:
|
|
self.path_history.append(path)
|
|
self.history_index = len(self.path_history) - 1
|
|
|
|
# 更新UI
|
|
self.update_navigation_buttons()
|
|
self.path_label.setText(path)
|
|
self.loading_label.show()
|
|
self.status_label.setText(f"正在加载 {path}...")
|
|
|
|
# 发送目录改变信号
|
|
self.directory_changed.emit(path)
|
|
|
|
# 在后台线程加载
|
|
self.load_thread = DirectoryLoadThread(self.scanner, path)
|
|
self.load_thread.finished.connect(self.on_directory_loaded)
|
|
self.load_thread.error.connect(self.on_load_error)
|
|
self.load_thread.start()
|
|
|
|
def on_directory_loaded(self, files: List[FileItem], path: str):
|
|
"""目录加载完成"""
|
|
if path != self.current_path:
|
|
return # 已经切换到其他目录
|
|
|
|
self.loading_label.hide()
|
|
self.model.set_files(files)
|
|
|
|
# 更新状态栏
|
|
total_size = sum(f.size for f in files)
|
|
dir_count = sum(1 for f in files if f.is_dir)
|
|
file_count = len(files) - dir_count
|
|
|
|
self.status_label.setText(
|
|
f"共 {dir_count} 个目录, {file_count} 个文件, "
|
|
f"总计 {format_size(total_size)}"
|
|
)
|
|
self.item_count_label.setText(f"{len(files)} 个项目")
|
|
|
|
def on_load_error(self, error_message: str):
|
|
"""加载出错"""
|
|
self.loading_label.hide()
|
|
self.status_label.setText(f"加载失败：{error_message}")
|
|
QMessageBox.critical(self, "错误", f"加载目录失败：\n{error_message}")
|
|
|
|
def on_item_clicked(self, index: QModelIndex):
|
|
"""点击项目"""
|
|
source_index = self.proxy_model.mapToSource(index)
|
|
file_item = self.model.get_file_at(source_index.row())
|
|
|
|
if file_item:
|
|
self.file_selected.emit(file_item)
|
|
|
|
def on_item_double_clicked(self, index: QModelIndex):
|
|
"""双击项目"""
|
|
source_index = self.proxy_model.mapToSource(index)
|
|
file_item = self.model.get_file_at(source_index.row())
|
|
|
|
if file_item and file_item.is_dir:
|
|
self.load_directory(file_item.path)
|
|
|
|
def show_context_menu(self, position):
|
|
"""显示右键菜单"""
|
|
# 获取选中的项目
|
|
selected_indexes = self.table_view.selectedIndexes()
|
|
|
|
if not selected_indexes:
|
|
return
|
|
|
|
# 获取唯一的行号
|
|
rows = set()
|
|
for index in selected_indexes:
|
|
rows.add(index.row())
|
|
|
|
# 获取文件信息
|
|
selected_files = []
|
|
for row in rows:
|
|
source_index = self.proxy_model.mapToSource(
|
|
self.proxy_model.index(row, 0)
|
|
)
|
|
file_item = self.model.get_file_at(source_index.row())
|
|
if file_item:
|
|
selected_files.append(file_item)
|
|
|
|
if not selected_files:
|
|
return
|
|
|
|
# 创建菜单
|
|
menu = QMenu(self)
|
|
|
|
# 删除动作
|
|
delete_action = QAction(f"🗑️ 删除 ({len(selected_files)} 个项目)", self)
|
|
delete_action.triggered.connect(lambda: self.delete_files(selected_files))
|
|
menu.addAction(delete_action)
|
|
|
|
# 如果是单个目录，添加进入目录动作
|
|
if len(selected_files) == 1 and selected_files[0].is_dir:
|
|
menu.addSeparator()
|
|
enter_action = QAction("📂 进入目录", self)
|
|
enter_action.triggered.connect(
|
|
lambda: self.load_directory(selected_files[0].path)
|
|
)
|
|
menu.addAction(enter_action)
|
|
|
|
# 显示菜单
|
|
menu.exec_(QCursor.pos())
|
|
|
|
def delete_files(self, files: List[FileItem]):
|
|
"""删除文件"""
|
|
from gui.confirm_dialog import DeleteConfirmDialog
|
|
|
|
# 计算总大小
|
|
total_size = sum(f.size for f in files)
|
|
|
|
# 检查是否包含目录
|
|
has_dir = any(f.is_dir for f in files)
|
|
|
|
# 显示确认对话框
|
|
paths = [f.path for f in files]
|
|
dialog = DeleteConfirmDialog(paths, total_size, has_dir, self)
|
|
|
|
if dialog.exec_() == DeleteConfirmDialog.Accepted and dialog.is_confirmed():
|
|
# 执行删除
|
|
deleted_size = 0
|
|
errors = []
|
|
|
|
for file_item in files:
|
|
success, message = safe_delete(
|
|
file_item.path,
|
|
use_sudo=True,
|
|
sudo_password=self.sudo_password
|
|
)
|
|
|
|
if success:
|
|
deleted_size += file_item.size
|
|
else:
|
|
errors.append(f"{file_item.name}: {message}")
|
|
|
|
# 显示结果
|
|
if errors:
|
|
QMessageBox.warning(
|
|
self,
|
|
"删除完成（有错误）",
|
|
f"成功删除 {len(files) - len(errors)} 个项目，\n"
|
|
f"释放 {format_size(deleted_size)} 空间。\n\n"
|
|
f"以下项目删除失败：\n" + "\n".join(errors[:10])
|
|
)
|
|
else:
|
|
QMessageBox.information(
|
|
self,
|
|
"删除成功",
|
|
f"成功删除 {len(files)} 个项目，\n"
|
|
f"释放 {format_size(deleted_size)} 空间。"
|
|
)
|
|
|
|
# 发送删除信号
|
|
self.files_deleted.emit(deleted_size)
|
|
|
|
# 刷新当前目录
|
|
self.refresh()
|
|
|
|
def go_back(self):
|
|
"""后退"""
|
|
if self.history_index > 0:
|
|
self.history_index -= 1
|
|
path = self.path_history[self.history_index]
|
|
self.load_directory(path)
|
|
|
|
def go_forward(self):
|
|
"""前进"""
|
|
if self.history_index < len(self.path_history) - 1:
|
|
self.history_index += 1
|
|
path = self.path_history[self.history_index]
|
|
self.load_directory(path)
|
|
|
|
def go_up(self):
|
|
"""上级目录"""
|
|
if self.current_path:
|
|
parent_path = os.path.dirname(self.current_path)
|
|
if parent_path != self.current_path:
|
|
self.load_directory(parent_path)
|
|
|
|
def refresh(self):
|
|
"""刷新当前目录"""
|
|
if self.current_path:
|
|
self.load_directory(self.current_path)
|
|
|
|
def update_navigation_buttons(self):
|
|
"""更新导航按钮状态"""
|
|
self.back_button.setEnabled(self.history_index > 0)
|
|
self.forward_button.setEnabled(
|
|
self.history_index < len(self.path_history) - 1
|
|
)
|
|
self.up_button.setEnabled(
|
|
self.current_path and self.current_path != '/'
|
|
)
|
|
|
|
def get_current_path(self) -> str:
|
|
"""获取当前路径"""
|
|
return self.current_path
|