From 9bbcf5a710ba0b8b404ec05f5046b1521defcfaa Mon Sep 17 00:00:00 2001 From: kushidou Date: Fri, 26 Jun 2026 16:54:55 +0800 Subject: [PATCH] =?UTF-8?q?=E9=A6=96=E6=AC=A1=E4=BB=A3=E7=A0=81=E6=8F=90?= =?UTF-8?q?=E4=BA=A4?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .gitignore | 49 ++++ build.sh | 109 +++++++++ disk_manager.py | 506 ++++++++++++++++++++++++++++++++++++++++ file_scanner.py | 489 ++++++++++++++++++++++++++++++++++++++ gui/__init__.py | 1 + gui/confirm_dialog.py | 290 +++++++++++++++++++++++ gui/file_browser.py | 477 +++++++++++++++++++++++++++++++++++++ gui/main_window.py | 373 +++++++++++++++++++++++++++++ gui/partition_dialog.py | 246 +++++++++++++++++++ gui/sudo_dialog.py | 154 ++++++++++++ main.py | 118 ++++++++++ requirements.txt | 2 + utils.py | 295 +++++++++++++++++++++++ 13 files changed, 3109 insertions(+) create mode 100644 .gitignore create mode 100644 build.sh create mode 100644 disk_manager.py create mode 100644 file_scanner.py create mode 100644 gui/__init__.py create mode 100644 gui/confirm_dialog.py create mode 100644 gui/file_browser.py create mode 100644 gui/main_window.py create mode 100644 gui/partition_dialog.py create mode 100644 gui/sudo_dialog.py create mode 100644 main.py create mode 100644 requirements.txt create mode 100644 utils.py diff --git a/.gitignore b/.gitignore new file mode 100644 index 0000000..05e27a6 --- /dev/null +++ b/.gitignore @@ -0,0 +1,49 @@ +# Python +__pycache__/ +*.py[cod] +*$py.class +*.so +*.egg +*.egg-info/ +dist/ +build/ +eggs/ +*.whl +pip-log.txt +pip-delete-this-directory.txt + +# PyInstaller +build +dist +*.manifest +*.spec +*.exe + +# 虚拟环境 +.venv/ +venv/ +ENV/ +env/ + +# IDE +.vscode/ +.idea/ +*.swp +*.swo +*~ + +# 系统文件 +.DS_Store +Thumbs.db +desktop.ini + +# 日志 +*.log +/tmp/ + +# 临时文件 +*.tmp +*.bak +*.cache + +HANDOVER.md \ No newline at end of file diff --git a/build.sh b/build.sh new file mode 100644 index 0000000..1efad3b --- /dev/null +++ b/build.sh @@ -0,0 +1,109 @@ +#!/bin/bash +# Linux 日志删除助手 - 打包脚本 +# 在 UOS 环境中执行此脚本进行打包 + +set -e # 遇到错误立即退出 + +echo "==========================================" +echo " Linux 日志删除助手 - 打包脚本" +echo "==========================================" +echo "" + +# 颜色定义 +RED='\033[0;31m' +GREEN='\033[0;32m' +YELLOW='\033[1;33m' +NC='\033[0m' # No Color + +# 检查是否为 root 用户 +if [ "$EUID" -ne 0 ]; then + echo -e "${YELLOW}提示：安装依赖需要 sudo 权限${NC}" + SUDO="sudo" +else + SUDO="" +fi + +# 步骤 1: 检查并安装系统依赖 +echo "" +echo "步骤 1/4: 检查并安装系统依赖..." +echo "------------------------------------------" + +# 更新包列表 +$SUDO apt update + +# 安装 Python3 和 pip（如果未安装） +if ! command -v python3 &> /dev/null; then + echo "安装 Python3..." + $SUDO apt install -y python3 +fi + +if ! command -v pip3 &> /dev/null; then + echo "安装 pip3..." + $SUDO apt install -y python3-pip +fi + +# 安装 PyQt5 +echo "安装 PyQt5..." +$SUDO apt install -y python3-pyqt5 + +# 步骤 2: 安装 Python 依赖 +echo "" +echo "步骤 2/4: 安装 Python 依赖..." +echo "------------------------------------------" + +# 安装 PyInstaller +pip3 install --user pyinstaller + +# 确保 ~/.local/bin 在 PATH 中 +export PATH="$HOME/.local/bin:$PATH" + +# 步骤 3: 使用 PyInstaller 打包 +echo "" +echo "步骤 3/4: 打包应用程序..." +echo "------------------------------------------" + +# 获取脚本所在目录 +SCRIPT_DIR="$( cd "$( dirname "${BASH_SOURCE[0]}" )" && pwd )" +cd "$SCRIPT_DIR" + +# 清理旧的构建文件 +rm -rf build dist __pycache__ *.spec + +# 使用 PyInstaller 打包 +pyinstaller \ + --onefile \ + --windowed \ + --name "log-cleaner" \ + --add-data "gui:gui" \ + --hidden-import "PyQt5.QtWidgets" \ + --hidden-import "PyQt5.QtCore" \ + --hidden-import "PyQt5.QtGui" \ + --clean \ + main.py + +# 步骤 4: 完成 +echo "" +echo "步骤 4/4: 打包完成！" +echo "==========================================" +echo "" +echo -e "${GREEN}✓ 打包成功！${NC}" +echo "" +echo "可执行文件位置: dist/log-cleaner" +echo "" +echo "使用方法：" +echo " 1. 将 dist/log-cleaner 复制到 UOS LiveCD 环境" +echo " 2. 赋予执行权限: chmod +x log-cleaner" +echo " 3. 运行: sudo ./log-cleaner" +echo "" +echo "注意事项：" +echo " - 运行时需要 sudo 权限" +echo " - 确保系统已安装 PyQt5 运行时库" +echo "" + +# 检查打包结果 +if [ -f "dist/log-cleaner" ]; then + echo -e "${GREEN}✓ 可执行文件大小: $(du -h dist/log-cleaner | cut -f1)${NC}" +else + echo -e "${RED}✗ 打包失败，请检查错误信息${NC}" + exit 1 +fi diff --git a/disk_manager.py b/disk_manager.py new file mode 100644 index 0000000..76436ab --- /dev/null +++ b/disk_manager.py @@ -0,0 +1,506 @@ +""" +磁盘管理模块 +提供分区检测、挂载、卸载等功能 +""" +import json +import os +import subprocess +import tempfile +from typing import List, Dict, Optional, Tuple +from dataclasses import dataclass +from utils import run_command, format_size + + +@dataclass +class Partition: + """分区信息""" + device: str # 设备名，如 /dev/sda1 + mount_point: str # 挂载点，如 / + fstype: str # 文件系统类型，如 ext4 + size: int # 总大小（字节） + used: int # 已用大小（字节） + available: int # 可用大小（字节） + use_percent: float # 使用百分比 + label: str = "" # 卷标 + is_system: bool = False # 是否为系统盘 + + @property + def size_formatted(self) -> str: + return format_size(self.size) + + @property + def used_formatted(self) -> str: + return format_size(self.used) + + @property + def available_formatted(self) -> str: + return format_size(self.available) + + @property + def use_percent_formatted(self) -> str: + return f"{self.use_percent:.1f}%" + + +class DiskManager: + """磁盘管理器""" + + # 排除的文件系统类型 + EXCLUDED_FS_TYPES = { + 'swap', + 'vfat', # 通常是 EFI 分区 + 'iso9660', # 光盘镜像 + 'squashfs', # LiveCD 文件系统 + 'tmpfs', # 临时文件系统 + 'devtmpfs', # 设备文件系统 + 'sysfs', # 内核虚拟文件系统 + 'proc', # 进程虚拟文件系统 + 'devpts', # 设备终端 + 'securityfs', + 'cgroup', + 'pstore', + 'debugfs', + 'hugetlbfs', + 'mqueue', + 'fusectl', + 'configfs', + 'overlay', # Overlay 文件系统 + } + + # 排除的挂载点前缀 + EXCLUDED_MOUNT_PREFIXES = [ + '/boot/efi', + '/efi', + '/boot/EFI', + '/snap', + '/cdrom', + '/media/cdrom', + ] + + def __init__(self, sudo_password: Optional[str] = None): + self.sudo_password = sudo_password + self.mounted_partitions: Dict[str, str] = {} # device -> mount_point + + def get_partitions(self) -> List[Partition]: + """ + 获取所有可用分区（排除 EFI、swap、LiveCD USB 等） + """ + partitions = [] + + # 获取 LiveCD 设备 + livecd_device = self._detect_livecd_device() + + # 使用 lsblk 获取分区信息 + stdout, stderr, returncode = run_command( + ['lsblk', '-J', '-o', 'NAME,SIZE,TYPE,MOUNTPOINT,FSTYPE,LABEL,PKNAME'], + use_sudo=True, + sudo_password=self.sudo_password + ) + + if returncode != 0: + # 备用方案：解析 /proc/partitions + return self._get_partitions_from_proc() + + try: + data = json.loads(stdout) + devices = data.get('blockdevices', []) + + for device in devices: + self._parse_device(device, partitions, livecd_device) + + except (json.JSONDecodeError, KeyError) as e: + print(f"解析 lsblk 输出失败: {e}") + return self._get_partitions_from_proc() + + # 获取每个分区的使用情况 + self._fill_usage_info(partitions) + + # 按使用率排序（高的在前） + partitions.sort(key=lambda p: p.use_percent, reverse=True) + + return partitions + + def _parse_device( + self, + device: dict, + partitions: List[Partition], + livecd_device: Optional[str] + ): + """递归解析设备信息""" + device_name = device.get('name', '') + device_type = device.get('type', '') + fstype = device.get('fstype', '') + mount_point = device.get('mountpoint', '') + label = device.get('label', '') + size_str = device.get('size', '0') + + # 只处理分区类型 + if device_type == 'part': + device_path = f"/dev/{device_name}" + + # 排除条件 + if self._should_exclude(device_path, fstype, mount_point, livecd_device): + return + + # 解析大小 + size = self._parse_size(size_str) + + partition = Partition( + device=device_path, + mount_point=mount_point or "", + fstype=fstype or "unknown", + size=size, + used=0, + available=0, + use_percent=0.0, + label=label or "", + is_system=(mount_point == '/') + ) + + partitions.append(partition) + + # 递归处理子设备 + children = device.get('children', []) + for child in children: + self._parse_device(child, partitions, livecd_device) + + def _should_exclude( + self, + device: str, + fstype: str, + mount_point: str, + livecd_device: Optional[str] + ) -> bool: + """判断是否应该排除此分区""" + # 排除特定文件系统类型 + if fstype in self.EXCLUDED_FS_TYPES: + return True + + # 排除 EFI 分区 + if mount_point and any(mount_point.startswith(prefix) for prefix in self.EXCLUDED_MOUNT_PREFIXES): + return True + + # 排除 LiveCD USB 设备 + if livecd_device and device.startswith(livecd_device): + return True + + # 排除没有文件系统的设备 + if not fstype: + return True + + return False + + def _parse_size(self, size_str: str) -> int: + """解析 lsblk 输出的大小字符串""" + if not size_str: + return 0 + + size_str = size_str.strip() + + # 提取数字和单位 + try: + if size_str.endswith('T'): + return int(float(size_str[:-1]) * 1024 * 1024 * 1024 * 1024) + elif size_str.endswith('G'): + return int(float(size_str[:-1]) * 1024 * 1024 * 1024) + elif size_str.endswith('M'): + return int(float(size_str[:-1]) * 1024 * 1024) + elif size_str.endswith('K'): + return int(float(size_str[:-1]) * 1024) + else: + return int(size_str) + except ValueError: + return 0 + + def _fill_usage_info(self, partitions: List[Partition]): + """填充分区的使用情况信息""" + # 使用 df 命令获取使用情况 + stdout, stderr, returncode = run_command( + ['df', '-B1'], + use_sudo=True, + sudo_password=self.sudo_password + ) + + if returncode != 0: + return + + # 解析 df 输出 + df_info = {} + for line in stdout.strip().split('\n')[1:]: # 跳过标题行 + parts = line.split() + if len(parts) >= 6: + device = parts[0] + try: + size = int(parts[1]) + used = int(parts[2]) + available = int(parts[3]) + use_percent = float(parts[4].rstrip('%')) + df_info[device] = { + 'size': size, + 'used': used, + 'available': available, + 'use_percent': use_percent + } + except (ValueError, IndexError): + pass + + # 填充分区信息 + for partition in partitions: + if partition.device in df_info: + info = df_info[partition.device] + partition.size = info['size'] + partition.used = info['used'] + partition.available = info['available'] + partition.use_percent = info['use_percent'] + + def _detect_livecd_device(self) -> Optional[str]: + """ + 检测 LiveCD USB 设备 + 返回设备名前缀，如 /dev/sdb + """ + # 方法1: 检查 /proc/cmdline + try: + with open('/proc/cmdline', 'r') as f: + cmdline = f.read() + # 查找 boot= 参数 + for param in cmdline.split(): + if param.startswith('boot='): + boot_device = param.split('=', 1)[1] + # 提取设备名（去掉分区号） + if boot_device.startswith('/dev/'): + return boot_device.rstrip('0123456789') + except Exception: + pass + + # 方法2: 检查根文件系统类型 + try: + stdout, _, _ = run_command(['findmnt', '-n', '-o', 'SOURCE', '/']) + root_device = stdout.strip() + if root_device: + # 如果根文件系统是 overlay 或 tmpfs，很可能是 LiveCD + stdout2, _, _ = run_command(['findmnt', '-n', '-o', 'FSTYPE', '/']) + fstype = stdout2.strip() + if fstype in ('overlay', 'squashfs', 'tmpfs'): + # 查找 USB 设备 + return self._find_usb_device() + except Exception: + pass + + # 方法3: 查找标记为 LiveCD 的设备 + return self._find_usb_device() + + def _find_usb_device(self) -> Optional[str]: + """查找 USB 设备""" + try: + stdout, _, _ = run_command( + ['lsblk', '-J', '-o', 'NAME,TRAN,TYPE'], + use_sudo=True, + sudo_password=self.sudo_password + ) + + data = json.loads(stdout) + devices = data.get('blockdevices', []) + + for device in devices: + if device.get('tran') == 'usb' and device.get('type') == 'disk': + return f"/dev/{device['name']}" + + except Exception: + pass + + return None + + def _get_partitions_from_proc(self) -> List[Partition]: + """从 /proc/partitions 获取分区信息（备用方案）""" + partitions = [] + + try: + stdout, _, returncode = run_command( + ['cat', '/proc/partitions'], + use_sudo=True, + sudo_password=self.sudo_password + ) + + if returncode != 0: + return [] + + livecd_device = self._detect_livecd_device() + + for line in stdout.strip().split('\n')[2:]: # 跳过标题行 + parts = line.split() + if len(parts) >= 4: + name = parts[3] + # 只处理分区（名称以数字结尾） + if name[-1].isdigit(): + device = f"/dev/{name}" + + # 排除 LiveCD 设备 + if livecd_device and device.startswith(livecd_device): + continue + + # 获取文件系统类型 + fstype = self._get_fstype(device) + + # 排除不需要的文件系统类型 + if fstype in self.EXCLUDED_FS_TYPES or not fstype: + continue + + # 获取挂载点 + mount_point = self._get_mount_point(device) + + # 排除 EFI 分区 + if mount_point and any( + mount_point.startswith(prefix) + for prefix in self.EXCLUDED_MOUNT_PREFIXES + ): + continue + + size = int(parts[2]) * 1024 # 转换为字节 + + partition = Partition( + device=device, + mount_point=mount_point or "", + fstype=fstype, + size=size, + used=0, + available=0, + use_percent=0.0, + is_system=(mount_point == '/') + ) + + partitions.append(partition) + + except Exception as e: + print(f"从 /proc/partitions 获取信息失败: {e}") + + # 填充使用情况 + self._fill_usage_info(partitions) + + return partitions + + def _get_fstype(self, device: str) -> str: + """获取设备的文件系统类型""" + stdout, _, _ = run_command( + ['blkid', '-o', 'value', '-s', 'TYPE', device], + use_sudo=True, + sudo_password=self.sudo_password + ) + return stdout.strip() + + def _get_mount_point(self, device: str) -> str: + """获取设备的挂载点""" + stdout, _, _ = run_command( + ['findmnt', '-n', '-o', 'TARGET', device], + use_sudo=True, + sudo_password=self.sudo_password + ) + return stdout.strip() + + def mount_partition(self, partition: Partition, mount_point: Optional[str] = None) -> Tuple[bool, str]: + """ + 挂载分区到临时目录 + + 参数: + partition: 分区信息 + mount_point: 指定挂载点（可选） + + 返回: + (success, mount_point_or_error_message) + """ + # 如果已经挂载，返回现有挂载点 + if partition.mount_point: + return True, partition.mount_point + + # 如果已挂载到临时目录 + if partition.device in self.mounted_partitions: + return True, self.mounted_partitions[partition.device] + + # 创建临时挂载点 + if mount_point is None: + mount_point = tempfile.mkdtemp(prefix='log_cleaner_') + + # 挂载分区 + stdout, stderr, returncode = run_command( + ['mount', partition.device, mount_point], + use_sudo=True, + sudo_password=self.sudo_password + ) + + if returncode == 0: + partition.mount_point = mount_point + self.mounted_partitions[partition.device] = mount_point + return True, mount_point + else: + return False, f"挂载失败: {stderr}" + + def unmount_partition(self, partition: Partition) -> Tuple[bool, str]: + """ + 卸载分区 + + 参数: + partition: 分区信息 + + 返回: + (success, message) + """ + # 如果是系统挂载点，不卸载 + if partition.is_system: + return False, "不能卸载系统分区" + + mount_point = partition.mount_point + + if not mount_point: + return False, "分区未挂载" + + # 如果是临时挂载点，卸载 + if partition.device in self.mounted_partitions: + stdout, stderr, returncode = run_command( + ['umount', mount_point], + use_sudo=True, + sudo_password=self.sudo_password + ) + + if returncode == 0: + del self.mounted_partitions[partition.device] + partition.mount_point = "" + + # 删除临时目录 + try: + os.rmdir(mount_point) + except OSError: + pass + + return True, "卸载成功" + else: + return False, f"卸载失败: {stderr}" + + return False, "分区不是由本工具挂载的" + + def unmount_all(self): + """卸载所有由本工具挂载的分区""" + for device in list(self.mounted_partitions.keys()): + mount_point = self.mounted_partitions[device] + run_command( + ['umount', mount_point], + use_sudo=True, + sudo_password=self.sudo_password + ) + try: + os.rmdir(mount_point) + except OSError: + pass + + self.mounted_partitions.clear() + + def get_partition_display_name(self, partition: Partition) -> str: + """获取分区的显示名称""" + parts = [partition.device] + + if partition.mount_point: + parts.append(f"({partition.mount_point})") + elif partition.label: + parts.append(f"({partition.label})") + + if partition.is_system: + parts.append("[系统盘]") + + return " ".join(parts) diff --git a/file_scanner.py b/file_scanner.py new file mode 100644 index 0000000..c572ee2 --- /dev/null +++ b/file_scanner.py @@ -0,0 +1,489 @@ +""" +文件扫描模块 +提供目录大小扫描、文件列表等功能 +""" +import os +import time +from typing import List, Dict, Optional, Callable, Tuple +from dataclasses import dataclass, field +from datetime import datetime +from utils import format_size, is_virtual_filesystem, run_command + + +@dataclass +class FileItem: + """文件/目录信息""" + name: str # 名称 + path: str # 完整路径 + is_dir: bool # 是否为目录 + size: int # 大小（字节） + modified_time: datetime # 修改时间 + permissions: str # 权限字符串 + owner: str # 所有者 + group: str # 所属组 + + @property + def size_formatted(self) -> str: + return format_size(self.size) + + @property + def modified_time_formatted(self) -> str: + return self.modified_time.strftime("%Y-%m-%d %H:%M") + + @property + def icon(self) -> str: + if self.is_dir: + return "📁" + elif self.name.endswith(('.log', '.log.gz', '.log.1', '.log.2')): + return "📋" + elif self.name.endswith(('.gz', '.tar', '.zip', '.7z', '.rar')): + return "📦" + elif self.name.endswith(('.conf', '.cfg', '.ini', '.yaml', '.yml', '.json')): + return "⚙️" + elif self.name.endswith(('.sh', '.bash', '.py', '.pl')): + return "📜" + elif self.name.endswith(('.txt', '.md', '.rst')): + return "📄" + else: + return "📄" + + +@dataclass +class ScanProgress: + """扫描进度信息""" + current_path: str = "" + scanned_dirs: int = 0 + scanned_files: int = 0 + total_size: int = 0 + elapsed_time: float = 0.0 + is_complete: bool = False + + +class FileScanner: + """文件扫描器""" + + # 虚拟文件系统目录列表 + VIRTUAL_DIRS = [ + '/proc', + '/sys', + '/dev', + '/run', + '/var/run', + '/var/lock', + '/tmp', + ] + + # 系统日志文件模式（不包含子目录中的日志） + SYSTEM_LOG_PATTERNS = [ + '*.log', + '*.log.*', + '*.gz', + 'syslog*', + 'messages*', + 'dmesg*', + 'kern.log*', + 'auth.log*', + 'daemon.log*', + 'debug*', + 'boot.log*', + 'faillog', + 'lastlog', + 'wtmp', + 'btmp', + 'utmp', + ] + + def __init__(self, sudo_password: Optional[str] = None): + self.sudo_password = sudo_password + self._cancel_scan = False + + def scan_directory_sizes( + self, + path: str, + skip_virtual: bool = True, + callback: Optional[Callable[[ScanProgress], None]] = None + ) -> Dict[str, int]: + """ + 扫描目录下每个子目录的大小 + + 参数: + path: 要扫描的目录路径 + skip_virtual: 是否跳过虚拟文件系统目录 + callback: 进度回调函数 + + 返回: + {目录路径: 大小} 字典 + """ + self._cancel_scan = False + dir_sizes: Dict[str, int] = {} + progress = ScanProgress() + start_time = time.time() + + try: + # 获取顶层子目录 + entries = self._list_directory(path) + + for entry in entries: + if self._cancel_scan: + break + + entry_path = os.path.join(path, entry) + + # 跳过虚拟文件系统 + if skip_virtual and self._is_virtual_dir(entry_path): + continue + + # 跳过其他挂载点 + if self._is_other_mountpoint(entry_path, path): + continue + + # 计算目录大小 + size = self._get_directory_size(entry_path, progress, callback) + dir_sizes[entry_path] = size + progress.total_size += size + + except PermissionError: + pass + except Exception as e: + print(f"扫描目录时发生错误: {e}") + + progress.is_complete = True + progress.elapsed_time = time.time() - start_time + + if callback: + callback(progress) + + return dir_sizes + + def list_directory( + self, + path: str, + show_hidden: bool = False, + sort_by: str = 'size', + reverse: bool = True + ) -> List[FileItem]: + """ + 列出目录下的文件和子目录 + + 参数: + path: 目录路径 + show_hidden: 是否显示隐藏文件 + sort_by: 排序方式 (name, size, time) + reverse: 是否逆序 + + 返回: + FileItem 列表 + """ + items: List[FileItem] = [] + + try: + entries = self._list_directory(path) + + for entry in entries: + # 跳过隐藏文件（除非要求显示） + if not show_hidden and entry.startswith('.'): + continue + + entry_path = os.path.join(path, entry) + + try: + item = self._get_file_item(entry_path, entry) + if item: + items.append(item) + except (PermissionError, OSError): + # 跳过无权限访问的文件 + pass + + except PermissionError: + pass + except Exception as e: + print(f"列出目录时发生错误: {e}") + + # 排序 + if sort_by == 'name': + items.sort(key=lambda x: x.name.lower(), reverse=reverse) + elif sort_by == 'size': + items.sort(key=lambda x: x.size, reverse=reverse) + elif sort_by == 'time': + items.sort(key=lambda x: x.modified_time, reverse=reverse) + + # 目录排在前面 + items.sort(key=lambda x: (not x.is_dir, -x.size if reverse else x.size)) + + return items + + def get_system_logs(self, mount_point: str) -> Tuple[List[FileItem], int]: + """ + 获取 /var/log 下的系统日志文件（不包含子目录中的日志） + + 参数: + mount_point: 分区挂载点 + + 返回: + (日志文件列表, 总大小) + """ + log_dir = os.path.join(mount_point, 'var', 'log') + + if not os.path.exists(log_dir): + return [], 0 + + logs: List[FileItem] = [] + total_size = 0 + + try: + entries = self._list_directory(log_dir) + + for entry in entries: + entry_path = os.path.join(log_dir, entry) + + # 只处理文件，不处理子目录 + if not os.path.isfile(entry_path): + continue + + # 检查是否为日志文件 + if self._is_log_file(entry): + try: + item = self._get_file_item(entry_path, entry) + if item: + logs.append(item) + total_size += item.size + except (PermissionError, OSError): + pass + + except PermissionError: + pass + except Exception as e: + print(f"获取系统日志时发生错误: {e}") + + # 按大小排序 + logs.sort(key=lambda x: x.size, reverse=True) + + return logs, total_size + + def cancel_scan(self): + """取消正在进行的扫描""" + self._cancel_scan = True + + def _list_directory(self, path: str) -> List[str]: + """列出目录内容""" + try: + return os.listdir(path) + except PermissionError: + # 尝试使用 sudo + if self.sudo_password: + stdout, _, returncode = run_command( + ['ls', '-a', path], + use_sudo=True, + sudo_password=self.sudo_password + ) + if returncode == 0: + return [f for f in stdout.strip().split('\n') if f and f not in ('.', '..')] + return [] + except OSError: + return [] + + def _get_directory_size( + self, + path: str, + progress: ScanProgress, + callback: Optional[Callable[[ScanProgress], None]] = None + ) -> int: + """递归计算目录大小""" + if self._cancel_scan: + return 0 + + total_size = 0 + + try: + entries = self._list_directory(path) + + for entry in entries: + if self._cancel_scan: + break + + entry_path = os.path.join(path, entry) + + # 跳过虚拟文件系统 + if self._is_virtual_dir(entry_path): + continue + + # 跳过其他挂载点 + if self._is_other_mountpoint(entry_path, path): + continue + + try: + if os.path.islink(entry_path): + # 跳过符号链接 + continue + elif os.path.isdir(entry_path): + # 递归计算子目录 + progress.current_path = entry_path + progress.scanned_dirs += 1 + + if callback and progress.scanned_dirs % 100 == 0: + callback(progress) + + total_size += self._get_directory_size(entry_path, progress, callback) + else: + # 获取文件大小 + try: + file_size = os.path.getsize(entry_path) + total_size += file_size + progress.scanned_files += 1 + progress.total_size += file_size + + if callback and progress.scanned_files % 1000 == 0: + callback(progress) + except OSError: + pass + + except PermissionError: + # 无权限访问 + pass + + except PermissionError: + # 无权限访问目录，尝试使用 sudo + if self.sudo_password: + total_size = self._get_directory_size_sudo(path) + + return total_size + + def _get_directory_size_sudo(self, path: str) -> int: + """使用 sudo 获取目录大小""" + stdout, _, returncode = run_command( + ['du', '-sb', path], + use_sudo=True, + sudo_password=self.sudo_password, + timeout=60 + ) + + if returncode == 0: + try: + return int(stdout.split()[0]) + except (ValueError, IndexError): + pass + + return 0 + + def _get_file_item(self, path: str, name: str) -> Optional[FileItem]: + """获取文件/目录信息""" + try: + stat_info = os.lstat(path) + + # 获取所有者和组 + try: + import pwd + import grp + owner = pwd.getpwuid(stat_info.st_uid).pw_name + group = grp.getgrgid(stat_info.st_gid).gr_name + except (ImportError, KeyError): + owner = str(stat_info.st_uid) + group = str(stat_info.st_gid) + + # 获取权限字符串 + permissions = self._format_permissions(stat_info.st_mode) + + # 如果是目录，获取目录大小（使用 du） + if os.path.isdir(path): + size = self._get_dir_size_fast(path) + else: + size = stat_info.st_size + + return FileItem( + name=name, + path=path, + is_dir=os.path.isdir(path), + size=size, + modified_time=datetime.fromtimestamp(stat_info.st_mtime), + permissions=permissions, + owner=owner, + group=group + ) + + except (OSError, PermissionError): + return None + + def _get_dir_size_fast(self, path: str) -> int: + """快速获取目录大小（使用 du 命令）""" + try: + stdout, _, returncode = run_command( + ['du', '-sb', path], + use_sudo=True, + sudo_password=self.sudo_password, + timeout=10 + ) + + if returncode == 0: + return int(stdout.split()[0]) + except Exception: + pass + + return 0 + + def _format_permissions(self, mode: int) -> str: + """格式化文件权限为字符串""" + perms = [] + + # 文件类型 + if os.path.isdir(mode): + perms.append('d') + elif os.path.islink(mode): + perms.append('l') + else: + perms.append('-') + + # 所有者权限 + perms.append('r' if mode & 0o400 else '-') + perms.append('w' if mode & 0o200 else '-') + perms.append('x' if mode & 0o100 else '-') + + # 组权限 + perms.append('r' if mode & 0o040 else '-') + perms.append('w' if mode & 0o020 else '-') + perms.append('x' if mode & 0o010 else '-') + + # 其他用户权限 + perms.append('r' if mode & 0o004 else '-') + perms.append('w' if mode & 0o002 else '-') + perms.append('x' if mode & 0o001 else '-') + + return ''.join(perms) + + def _is_virtual_dir(self, path: str) -> bool: + """检查是否为虚拟文件系统目录""" + path = os.path.normpath(path) + for vdir in self.VIRTUAL_DIRS: + if path == vdir or path.startswith(vdir + '/'): + return True + return False + + def _is_other_mountpoint(self, path: str, base_path: str) -> bool: + """检查是否为其他挂载点""" + try: + # 如果路径和基础路径的设备号不同，说明是不同的挂载点 + base_stat = os.stat(base_path) + path_stat = os.stat(path) + + if base_stat.st_dev != path_stat.st_dev: + return True + + except OSError: + pass + + return False + + def _is_log_file(self, filename: str) -> bool: + """检查是否为日志文件""" + import fnmatch + + for pattern in self.SYSTEM_LOG_PATTERNS: + if fnmatch.fnmatch(filename, pattern): + return True + + # 检查文件扩展名 + log_extensions = ['.log', '.log.gz', '.log.1', '.log.2', '.log.3'] + for ext in log_extensions: + if filename.endswith(ext): + return True + + return False diff --git a/gui/__init__.py b/gui/__init__.py new file mode 100644 index 0000000..52f0ce8 --- /dev/null +++ b/gui/__init__.py @@ -0,0 +1 @@ +# Linux 日志删除助手 - GUI 模块 diff --git a/gui/confirm_dialog.py b/gui/confirm_dialog.py new file mode 100644 index 0000000..2a842d9 --- /dev/null +++ b/gui/confirm_dialog.py @@ -0,0 +1,290 @@ +""" +删除确认对话框 +""" +from PyQt5.QtWidgets import ( + QDialog, QVBoxLayout, QHBoxLayout, QLabel, + QPushButton, QTextEdit, QCheckBox, QMessageBox +) +from PyQt5.QtCore import Qt +from PyQt5.QtGui import QFont, QColor + +from utils import format_size + + +class DeleteConfirmDialog(QDialog): + """删除确认对话框""" + + def __init__( + self, + paths: list, + total_size: int, + is_dir: bool = False, + parent=None + ): + """ + 初始化对话框 + + 参数: + paths: 要删除的路径列表 + total_size: 总大小（字节） + is_dir: 是否为目录 + parent: 父窗口 + """ + super().__init__(parent) + self.paths = paths + self.total_size = total_size + self.is_dir = is_dir + self.confirmed = False + self.init_ui() + + def init_ui(self): + """初始化界面""" + self.setWindowTitle("确认删除") + self.setMinimumSize(500, 350) + self.setWindowFlags(self.windowFlags() & ~Qt.WindowContextHelpButtonHint) + + # 主布局 + layout = QVBoxLayout(self) + layout.setSpacing(15) + layout.setContentsMargins(20, 20, 20, 20) + + # 警告图标和标题 + header_layout = QHBoxLayout() + + icon_label = QLabel("⚠️") + font = QFont() + font.setPointSize(36) + icon_label.setFont(font) + header_layout.addWidget(icon_label) + + title_layout = QVBoxLayout() + + title_label = QLabel("确认删除") + title_font = QFont() + title_font.setPointSize(14) + title_font.setBold(True) + title_label.setFont(title_font) + title_layout.addWidget(title_label) + + size_label = QLabel(f"将释放 {format_size(self.total_size)} 空间") + size_label.setStyleSheet("color: green;") + title_layout.addWidget(size_label) + + header_layout.addLayout(title_layout) + header_layout.addStretch() + + layout.addLayout(header_layout) + + # 分隔线 + separator = QLabel() + separator.setFrameStyle(QLabel.HLine | QLabel.Sunken) + layout.addWidget(separator) + + # 删除内容列表 + content_label = QLabel("将删除以下内容：") + layout.addWidget(content_label) + + self.content_text = QTextEdit() + self.content_text.setReadOnly(True) + self.content_text.setMaximumHeight(150) + + # 显示删除列表 + content = "" + for path in self.paths[:50]: # 最多显示 50 条 + content += f"• {path}\n" + + if len(self.paths) > 50: + content += f"\n... 还有 {len(self.paths) - 50} 个项目" + + self.content_text.setPlainText(content) + layout.addWidget(self.content_text) + + # 警告文本 + if self.is_dir: + warning_label = QLabel( + "⚠️ 目录将被递归删除，包含其中的所有文件和子目录！" + ) + warning_label.setStyleSheet("color: red; font-weight: bold;") + warning_label.setWordWrap(True) + layout.addWidget(warning_label) + + # 确认复选框 + self.confirm_checkbox = QCheckBox("我已确认要删除以上内容") + self.confirm_checkbox.stateChanged.connect(self.on_checkbox_changed) + layout.addWidget(self.confirm_checkbox) + + # 按钮 + button_layout = QHBoxLayout() + button_layout.addStretch() + + self.cancel_button = QPushButton("取消") + self.cancel_button.clicked.connect(self.reject) + button_layout.addWidget(self.cancel_button) + + self.delete_button = QPushButton("删除") + self.delete_button.setEnabled(False) + self.delete_button.setStyleSheet( + "QPushButton:enabled { background-color: #e74c3c; color: white; }" + ) + self.delete_button.clicked.connect(self.on_delete_clicked) + button_layout.addWidget(self.delete_button) + + layout.addLayout(button_layout) + + def on_checkbox_changed(self, state): + """复选框状态改变""" + self.delete_button.setEnabled(state == Qt.Checked) + + def on_delete_clicked(self): + """删除按钮点击""" + if not self.confirm_checkbox.isChecked(): + return + + # 最终确认 + reply = QMessageBox.critical( + self, + "最终确认", + f"确定要删除 {len(self.paths)} 个项目吗？\n" + f"此操作不可撤销！", + QMessageBox.Yes | QMessageBox.No, + QMessageBox.No + ) + + if reply == QMessageBox.Yes: + self.confirmed = True + self.accept() + + def is_confirmed(self) -> bool: + """是否确认删除""" + return self.confirmed + + +class LogDeleteConfirmDialog(QDialog): + """系统日志删除确认对话框""" + + def __init__( + self, + log_files: list, + total_size: int, + parent=None + ): + """ + 初始化对话框 + + 参数: + log_files: 日志文件列表 + total_size: 总大小（字节） + parent: 父窗口 + """ + super().__init__(parent) + self.log_files = log_files + self.total_size = total_size + self.confirmed = False + self.init_ui() + + def init_ui(self): + """初始化界面""" + self.setWindowTitle("清理系统日志") + self.setMinimumSize(500, 400) + self.setWindowFlags(self.windowFlags() & ~Qt.WindowContextHelpButtonHint) + + # 主布局 + layout = QVBoxLayout(self) + layout.setSpacing(15) + layout.setContentsMargins(20, 20, 20, 20) + + # 标题 + header_layout = QHBoxLayout() + + icon_label = QLabel("📋") + font = QFont() + font.setPointSize(36) + icon_label.setFont(font) + header_layout.addWidget(icon_label) + + title_layout = QVBoxLayout() + + title_label = QLabel("清理系统日志") + title_font = QFont() + title_font.setPointSize(14) + title_font.setBold(True) + title_label.setFont(title_font) + title_layout.addWidget(title_label) + + info_label = QLabel( + f"检测到 /var/log 下有 {len(self.log_files)} 个日志文件，\n" + f"共占用 {format_size(self.total_size)} 空间。" + ) + title_layout.addWidget(info_label) + + header_layout.addLayout(title_layout) + header_layout.addStretch() + + layout.addLayout(header_layout) + + # 分隔线 + separator = QLabel() + separator.setFrameStyle(QLabel.HLine | QLabel.Sunken) + layout.addWidget(separator) + + # 日志文件列表 + content_label = QLabel("将删除以下日志文件：") + layout.addWidget(content_label) + + self.content_text = QTextEdit() + self.content_text.setReadOnly(True) + + # 显示日志文件列表 + content = "" + for log_file in self.log_files: + name = log_file.name if hasattr(log_file, 'name') else str(log_file) + size = format_size(log_file.size) if hasattr(log_file, 'size') else "" + content += f"• {name} ({size})\n" + + self.content_text.setPlainText(content) + layout.addWidget(self.content_text) + + # 说明文本 + note_label = QLabel( + "💡 这些是 /var/log 目录下的日志文件（不包含子目录中的日志）。\n" + "删除后系统会自动创建新的日志文件。" + ) + note_label.setWordWrap(True) + note_label.setStyleSheet("color: gray;") + layout.addWidget(note_label) + + # 按钮 + button_layout = QHBoxLayout() + button_layout.addStretch() + + self.cancel_button = QPushButton("取消") + self.cancel_button.clicked.connect(self.reject) + button_layout.addWidget(self.cancel_button) + + self.delete_button = QPushButton("清理日志") + self.delete_button.setStyleSheet( + "QPushButton { background-color: #3498db; color: white; }" + ) + self.delete_button.clicked.connect(self.on_delete_clicked) + button_layout.addWidget(self.delete_button) + + layout.addLayout(button_layout) + + def on_delete_clicked(self): + """删除按钮点击""" + reply = QMessageBox.question( + self, + "确认清理", + f"确定要清理这 {len(self.log_files)} 个日志文件吗？\n" + f"将释放 {format_size(self.total_size)} 空间。", + QMessageBox.Yes | QMessageBox.No, + QMessageBox.Yes + ) + + if reply == QMessageBox.Yes: + self.confirmed = True + self.accept() + + def is_confirmed(self) -> bool: + """是否确认删除""" + return self.confirmed diff --git a/gui/file_browser.py b/gui/file_browser.py new file mode 100644 index 0000000..9fbdbf8 --- /dev/null +++ b/gui/file_browser.py @@ -0,0 +1,477 @@ +""" +文件浏览器组件 +""" +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 diff --git a/gui/main_window.py b/gui/main_window.py new file mode 100644 index 0000000..9547ed2 --- /dev/null +++ b/gui/main_window.py @@ -0,0 +1,373 @@ +""" +主窗口 +""" +import os +from typing import Optional + +from PyQt5.QtWidgets import ( + QMainWindow, QWidget, QVBoxLayout, QHBoxLayout, + QLabel, QPushButton, QStatusBar, QMessageBox, + QApplication, QSplitter, QFrame +) +from PyQt5.QtCore import Qt, QThread, pyqtSignal +from PyQt5.QtGui import QFont, QCloseEvent + +from disk_manager import DiskManager, Partition +from file_scanner import FileScanner, FileItem, ScanProgress +from utils import format_size, check_sudo_needs_password, verify_sudo_password, safe_delete + +from gui.partition_dialog import PartitionDialog +from gui.file_browser import FileBrowser +from gui.confirm_dialog import LogDeleteConfirmDialog +from gui.sudo_dialog import SudoDialog + + +class DirectorySizeThread(QThread): + """目录大小统计线程""" + progress = pyqtSignal(object) # ScanProgress + finished = pyqtSignal(dict) # {path: size} + error = pyqtSignal(str) + + def __init__(self, scanner: FileScanner, path: str): + super().__init__() + self.scanner = scanner + self.path = path + + def run(self): + try: + sizes = self.scanner.scan_directory_sizes( + self.path, + callback=lambda p: self.progress.emit(p) + ) + self.finished.emit(sizes) + except Exception as e: + self.error.emit(str(e)) + + +class MainWindow(QMainWindow): + """主窗口""" + + def __init__(self, sudo_password: Optional[str] = None): + super().__init__() + self.sudo_password = sudo_password + self.disk_manager = DiskManager(sudo_password) + self.file_scanner = FileScanner(sudo_password) + self.current_partition: Optional[Partition] = None + self.current_mount_point: Optional[str] = None + self.dir_sizes = {} # {path: size} + self.init_ui() + + def init_ui(self): + """初始化界面""" + self.setWindowTitle("Linux 日志删除助手") + self.setMinimumSize(900, 600) + + # 中央部件 + central_widget = QWidget() + self.setCentralWidget(central_widget) + + # 主布局 + main_layout = QVBoxLayout(central_widget) + main_layout.setContentsMargins(10, 10, 10, 10) + main_layout.setSpacing(10) + + # 顶部信息栏 + info_layout = QHBoxLayout() + + # 分区信息 + self.partition_label = QLabel("当前分区：未选择") + font = QFont() + font.setPointSize(11) + self.partition_label.setFont(font) + info_layout.addWidget(self.partition_label) + + info_layout.addStretch() + + # 切换分区按钮 + self.switch_partition_button = QPushButton("🔄 切换分区") + self.switch_partition_button.clicked.connect(self.switch_partition) + info_layout.addWidget(self.switch_partition_button) + + main_layout.addLayout(info_layout) + + # 工具栏 + toolbar_layout = QHBoxLayout() + + # 刷新按钮 + self.refresh_button = QPushButton("🔄 刷新") + self.refresh_button.clicked.connect(self.refresh) + toolbar_layout.addWidget(self.refresh_button) + + # 统计按钮 + self.scan_button = QPushButton("📊 统计目录大小") + self.scan_button.clicked.connect(self.scan_directory_sizes) + toolbar_layout.addWidget(self.scan_button) + + # 快捷删除日志按钮 + self.delete_logs_button = QPushButton("🗑️ 快捷删除系统日志") + self.delete_logs_button.setStyleSheet( + "QPushButton { background-color: #e74c3c; color: white; font-weight: bold; }" + ) + self.delete_logs_button.clicked.connect(self.quick_delete_logs) + self.delete_logs_button.hide() # 默认隐藏，检测到大日志时显示 + toolbar_layout.addWidget(self.delete_logs_button) + + toolbar_layout.addStretch() + + main_layout.addLayout(toolbar_layout) + + # 分隔线 + separator = QFrame() + separator.setFrameStyle(QFrame.HLine | QFrame.Sunken) + main_layout.addWidget(separator) + + # 文件浏览器 + self.file_browser = FileBrowser( + self.file_scanner, + self.sudo_password, + self + ) + self.file_browser.directory_changed.connect(self.on_directory_changed) + self.file_browser.files_deleted.connect(self.on_files_deleted) + main_layout.addWidget(self.file_browser) + + # 状态栏 + self.status_bar = QStatusBar() + self.setStatusBar(self.status_bar) + self.status_bar.showMessage("就绪") + + # 进度条（用于扫描） + self.progress_bar = QLabel() + self.status_bar.addPermanentWidget(self.progress_bar) + + # 默认禁用工具栏按钮 + self.refresh_button.setEnabled(False) + self.scan_button.setEnabled(False) + + def switch_partition(self): + """切换分区""" + dialog = PartitionDialog(self.disk_manager, self) + + if dialog.exec_() == PartitionDialog.Accepted: + partition = dialog.get_selected_partition() + + if partition: + self.load_partition(partition) + + def load_partition(self, partition: Partition): + """加载分区""" + self.current_partition = partition + self.status_bar.showMessage(f"正在挂载分区 {partition.device}...") + + # 挂载分区 + success, result = self.disk_manager.mount_partition(partition) + + if success: + self.current_mount_point = result + + # 更新界面 + self.partition_label.setText( + f"当前分区：{partition.device} " + f"({partition.mount_point or '未挂载'}) " + f"- {partition.used_formatted} / {partition.size_formatted} " + f"({partition.use_percent_formatted})" + ) + + # 启用工具栏按钮 + self.refresh_button.setEnabled(True) + self.scan_button.setEnabled(True) + + # 加载根目录 + self.file_browser.load_directory(self.current_mount_point) + + # 检测系统日志 + self.check_system_logs() + + self.status_bar.showMessage(f"已加载分区 {partition.device}") + else: + QMessageBox.critical( + self, + "错误", + f"挂载分区失败：\n{result}" + ) + self.status_bar.showMessage("挂载分区失败") + + def check_system_logs(self): + """检测系统日志大小""" + if not self.current_mount_point: + return + + logs, total_size = self.file_scanner.get_system_logs(self.current_mount_point) + + # 如果大于 1GB，显示快捷删除按钮 + if total_size > 1024 * 1024 * 1024: + self.delete_logs_button.setText( + f"🗑️ 快捷删除系统日志 ({format_size(total_size)})" + ) + self.delete_logs_button.show() + self._cached_logs = logs + self._cached_logs_size = total_size + else: + self.delete_logs_button.hide() + + def quick_delete_logs(self): + """快捷删除系统日志""" + if not hasattr(self, '_cached_logs') or not self._cached_logs: + return + + dialog = LogDeleteConfirmDialog( + self._cached_logs, + self._cached_logs_size, + self + ) + + if dialog.exec_() == LogDeleteConfirmDialog.Accepted and dialog.is_confirmed(): + # 删除日志文件 + deleted_size = 0 + errors = [] + + for log_file in self._cached_logs: + success, message = safe_delete( + log_file.path, + use_sudo=True, + sudo_password=self.sudo_password + ) + + if success: + deleted_size += log_file.size + else: + errors.append(f"{log_file.name}: {message}") + + # 显示结果 + if errors: + QMessageBox.warning( + self, + "清理完成（有错误）", + f"成功清理 {len(self._cached_logs) - len(errors)} 个日志文件，\n" + f"释放 {format_size(deleted_size)} 空间。\n\n" + f"以下文件清理失败：\n" + "\n".join(errors[:10]) + ) + else: + QMessageBox.information( + self, + "清理成功", + f"成功清理 {len(self._cached_logs)} 个日志文件，\n" + f"释放 {format_size(deleted_size)} 空间。" + ) + + # 刷新 + self.check_system_logs() + self.refresh() + + def scan_directory_sizes(self): + """统计目录大小""" + if not self.current_mount_point: + return + + self.scan_button.setEnabled(False) + self.status_bar.showMessage("正在统计目录大小...") + + # 在后台线程统计 + self.scan_thread = DirectorySizeThread( + self.file_scanner, + self.current_mount_point + ) + self.scan_thread.progress.connect(self.on_scan_progress) + self.scan_thread.finished.connect(self.on_scan_finished) + self.scan_thread.error.connect(self.on_scan_error) + self.scan_thread.start() + + def on_scan_progress(self, progress: ScanProgress): + """扫描进度更新""" + self.progress_bar.setText( + f"已扫描 {progress.scanned_dirs} 个目录, " + f"{progress.scanned_files} 个文件" + ) + + def on_scan_finished(self, sizes: dict): + """扫描完成""" + self.dir_sizes = sizes + self.scan_button.setEnabled(True) + self.progress_bar.clear() + + # 按大小排序 + sorted_sizes = sorted( + sizes.items(), + key=lambda x: x[1], + reverse=True + ) + + # 显示结果 + total_size = sum(sizes.values()) + self.status_bar.showMessage( + f"统计完成，共 {len(sizes)} 个目录，" + f"总大小 {format_size(total_size)}" + ) + + # 可以在这里显示一个统计结果窗口 + self.show_scan_results(sorted_sizes) + + def on_scan_error(self, error_message: str): + """扫描出错""" + self.scan_button.setEnabled(True) + self.progress_bar.clear() + self.status_bar.showMessage(f"统计失败：{error_message}") + + QMessageBox.critical( + self, + "错误", + f"统计目录大小失败：\n{error_message}" + ) + + def show_scan_results(self, sorted_sizes: list): + """显示扫描结果""" + # TODO: 可以创建一个专门的结果窗口 + # 目前先在状态栏显示前几个大目录 + if sorted_sizes: + top_dirs = sorted_sizes[:5] + info = "最大的目录: " + ", ".join( + f"{os.path.basename(p)} ({format_size(s)})" + for p, s in top_dirs + ) + self.status_bar.showMessage(info) + + def on_directory_changed(self, path: str): + """目录改变""" + # 可以在这里更新界面状态 + pass + + def on_files_deleted(self, freed_size: int): + """文件被删除""" + # 更新分区使用情况 + if self.current_partition: + self.current_partition.used -= freed_size + self.current_partition.available += freed_size + + self.partition_label.setText( + f"当前分区：{self.current_partition.device} " + f"({self.current_partition.mount_point or '未挂载'}) " + f"- {self.current_partition.used_formatted} / " + f"{self.current_partition.size_formatted} " + f"({self.current_partition.use_percent_formatted})" + ) + + def refresh(self): + """刷新""" + self.file_browser.refresh() + + def closeEvent(self, event: QCloseEvent): + """关闭事件""" + # 卸载所有分区 + reply = QMessageBox.question( + self, + "确认退出", + "确定要退出吗？\n已挂载的分区将被卸载。", + QMessageBox.Yes | QMessageBox.No, + QMessageBox.Yes + ) + + if reply == QMessageBox.Yes: + self.disk_manager.unmount_all() + event.accept() + else: + event.ignore() diff --git a/gui/partition_dialog.py b/gui/partition_dialog.py new file mode 100644 index 0000000..992beec --- /dev/null +++ b/gui/partition_dialog.py @@ -0,0 +1,246 @@ +""" +分区选择对话框 +""" +from PyQt5.QtWidgets import ( + QDialog, QVBoxLayout, QHBoxLayout, QLabel, + QTableWidget, QTableWidgetItem, QPushButton, + QHeaderView, QMessageBox, QProgressBar +) +from PyQt5.QtCore import Qt, QThread, pyqtSignal +from PyQt5.QtGui import QFont, QColor + +from disk_manager import DiskManager, Partition +from utils import format_size + + +class PartitionLoadThread(QThread): + """分区加载线程""" + finished = pyqtSignal(list) + error = pyqtSignal(str) + + def __init__(self, disk_manager: DiskManager): + super().__init__() + self.disk_manager = disk_manager + + def run(self): + try: + partitions = self.disk_manager.get_partitions() + self.finished.emit(partitions) + except Exception as e: + self.error.emit(str(e)) + + +class PartitionDialog(QDialog): + """分区选择对话框""" + + def __init__(self, disk_manager: DiskManager, parent=None): + super().__init__(parent) + self.disk_manager = disk_manager + self.selected_partition = None + self.partitions = [] + self.init_ui() + self.load_partitions() + + def init_ui(self): + """初始化界面""" + self.setWindowTitle("选择要清理的分区") + self.setMinimumSize(700, 450) + self.setWindowFlags(self.windowFlags() & ~Qt.WindowContextHelpButtonHint) + + # 主布局 + layout = QVBoxLayout(self) + layout.setSpacing(15) + layout.setContentsMargins(20, 20, 20, 20) + + # 标题 + title_label = QLabel("请选择要清理的磁盘分区：") + font = QFont() + font.setPointSize(12) + font.setBold(True) + title_label.setFont(font) + layout.addWidget(title_label) + + # 分区表格 + self.table = QTableWidget() + self.table.setColumnCount(7) + self.table.setHorizontalHeaderLabels([ + "设备", "挂载点", "文件系统", "总容量", "已用", "可用", "使用率" + ]) + + # 设置表格属性 + header = self.table.horizontalHeader() + header.setSectionResizeMode(0, QHeaderView.Stretch) + header.setSectionResizeMode(1, QHeaderView.Stretch) + header.setSectionResizeMode(2, QHeaderView.ResizeToContents) + header.setSectionResizeMode(3, QHeaderView.ResizeToContents) + header.setSectionResizeMode(4, QHeaderView.ResizeToContents) + header.setSectionResizeMode(5, QHeaderView.ResizeToContents) + header.setSectionResizeMode(6, QHeaderView.ResizeToContents) + + self.table.setSelectionBehavior(QTableWidget.SelectRows) + self.table.setSelectionMode(QTableWidget.SingleSelection) + self.table.setEditTriggers(QTableWidget.NoEditTriggers) + self.table.doubleClicked.connect(self.on_table_double_clicked) + + layout.addWidget(self.table) + + # 加载进度条 + self.progress_bar = QProgressBar() + self.progress_bar.setRange(0, 0) # 不确定进度 + self.progress_bar.setTextVisible(False) + layout.addWidget(self.progress_bar) + + # 状态标签 + self.status_label = QLabel("正在检测磁盘分区...") + self.status_label.setAlignment(Qt.AlignCenter) + layout.addWidget(self.status_label) + + # 按钮 + button_layout = QHBoxLayout() + button_layout.addStretch() + + self.refresh_button = QPushButton("🔄 刷新") + self.refresh_button.clicked.connect(self.load_partitions) + button_layout.addWidget(self.refresh_button) + + self.cancel_button = QPushButton("取消") + self.cancel_button.clicked.connect(self.reject) + button_layout.addWidget(self.cancel_button) + + self.ok_button = QPushButton("确定") + self.ok_button.setDefault(True) + self.ok_button.setEnabled(False) + self.ok_button.clicked.connect(self.on_ok_clicked) + button_layout.addWidget(self.ok_button) + + layout.addLayout(button_layout) + + def load_partitions(self): + """加载分区列表""" + self.progress_bar.show() + self.status_label.setText("正在检测磁盘分区...") + self.ok_button.setEnabled(False) + self.refresh_button.setEnabled(False) + + # 在后台线程加载分区 + self.load_thread = PartitionLoadThread(self.disk_manager) + self.load_thread.finished.connect(self.on_partitions_loaded) + self.load_thread.error.connect(self.on_load_error) + self.load_thread.start() + + def on_partitions_loaded(self, partitions): + """分区加载完成""" + self.partitions = partitions + self.progress_bar.hide() + self.refresh_button.setEnabled(True) + + if not partitions: + self.status_label.setText("未检测到可用的分区") + return + + self.status_label.setText(f"检测到 {len(partitions)} 个可用分区") + self.update_table() + + def on_load_error(self, error_message): + """加载出错""" + self.progress_bar.hide() + self.refresh_button.setEnabled(True) + self.status_label.setText(f"加载失败：{error_message}") + + QMessageBox.critical( + self, + "错误", + f"加载分区列表失败：\n{error_message}" + ) + + def update_table(self): + """更新表格内容""" + self.table.setRowCount(len(self.partitions)) + + for row, partition in enumerate(self.partitions): + # 设备名 + device_item = QTableWidgetItem(partition.device) + device_item.setData(Qt.UserRole, row) # 存储索引 + self.table.setItem(row, 0, device_item) + + # 挂载点 + mount_point = partition.mount_point or "未挂载" + mount_item = QTableWidgetItem(mount_point) + self.table.setItem(row, 1, mount_item) + + # 文件系统 + fs_item = QTableWidgetItem(partition.fstype) + self.table.setItem(row, 2, fs_item) + + # 总容量 + size_item = QTableWidgetItem(partition.size_formatted) + size_item.setTextAlignment(Qt.AlignRight | Qt.AlignVCenter) + self.table.setItem(row, 3, size_item) + + # 已用 + used_item = QTableWidgetItem(partition.used_formatted) + used_item.setTextAlignment(Qt.AlignRight | Qt.AlignVCenter) + self.table.setItem(row, 4, used_item) + + # 可用 + available_item = QTableWidgetItem(partition.available_formatted) + available_item.setTextAlignment(Qt.AlignRight | Qt.AlignVCenter) + self.table.setItem(row, 5, available_item) + + # 使用率 + percent_item = QTableWidgetItem(partition.use_percent_formatted) + percent_item.setTextAlignment(Qt.AlignRight | Qt.AlignVCenter) + + # 根据使用率设置颜色 + if partition.use_percent >= 90: + percent_item.setForeground(QColor(255, 0, 0)) # 红色 + elif partition.use_percent >= 80: + percent_item.setForeground(QColor(255, 165, 0)) # 橙色 + elif partition.use_percent >= 70: + percent_item.setForeground(QColor(255, 255, 0)) # 黄色 + + self.table.setItem(row, 6, percent_item) + + # 默认选中第一行 + if self.partitions: + self.table.selectRow(0) + self.on_table_clicked(self.table.model().index(0, 0)) + + def on_table_clicked(self, index): + """表格点击事件""" + row = index.row() + if 0 <= row < len(self.partitions): + self.selected_partition = self.partitions[row] + self.ok_button.setEnabled(True) + + def on_table_double_clicked(self, index): + """表格双击事件""" + row = index.row() + if 0 <= row < len(self.partitions): + self.selected_partition = self.partitions[row] + self.accept() + + def on_ok_clicked(self): + """确定按钮点击""" + if self.selected_partition: + # 如果使用率超过 90%，显示警告 + if self.selected_partition.use_percent >= 90: + reply = QMessageBox.warning( + self, + "警告", + f"分区 {self.selected_partition.device} 使用率已达 " + f"{self.selected_partition.use_percent:.1f}%！\n" + f"可用空间仅剩 {self.selected_partition.available_formatted}。\n\n" + f"是否继续清理？", + QMessageBox.Yes | QMessageBox.No, + QMessageBox.Yes + ) + + if reply != QMessageBox.Yes: + return + + self.accept() + + def get_selected_partition(self) -> Partition: + """获取选中的分区""" + return self.selected_partition diff --git a/gui/sudo_dialog.py b/gui/sudo_dialog.py new file mode 100644 index 0000000..7a07dd6 --- /dev/null +++ b/gui/sudo_dialog.py @@ -0,0 +1,154 @@ +""" +sudo 密码输入对话框 +""" +from PyQt5.QtWidgets import ( + QDialog, QVBoxLayout, QHBoxLayout, QLabel, + QLineEdit, QPushButton, QMessageBox +) +from PyQt5.QtCore import Qt +from PyQt5.QtGui import QFont, QIcon + + +class SudoDialog(QDialog): + """sudo 密码输入对话框""" + + def __init__(self, parent=None): + super().__init__(parent) + self.password = None + self.init_ui() + + def init_ui(self): + """初始化界面""" + self.setWindowTitle("需要 sudo 权限") + self.setFixedSize(400, 200) + self.setWindowFlags(self.windowFlags() & ~Qt.WindowContextHelpButtonHint) + + # 主布局 + layout = QVBoxLayout(self) + layout.setSpacing(15) + layout.setContentsMargins(20, 20, 20, 20) + + # 图标和提示 + icon_label = QLabel("🔒") + icon_label.setAlignment(Qt.AlignCenter) + font = QFont() + font.setPointSize(36) + icon_label.setFont(font) + layout.addWidget(icon_label) + + # 提示文本 + hint_label = QLabel( + "本工具需要 sudo 权限来访问磁盘分区和删除文件。\n" + "请输入 sudo 密码：" + ) + hint_label.setAlignment(Qt.AlignCenter) + hint_label.setWordWrap(True) + layout.addWidget(hint_label) + + # 密码输入框 + self.password_input = QLineEdit() + self.password_input.setEchoMode(QLineEdit.Password) + self.password_input.setPlaceholderText("请输入 sudo 密码") + self.password_input.returnPressed.connect(self.on_ok_clicked) + layout.addWidget(self.password_input) + + # 按钮 + button_layout = QHBoxLayout() + button_layout.addStretch() + + self.cancel_button = QPushButton("取消") + self.cancel_button.clicked.connect(self.on_cancel_clicked) + button_layout.addWidget(self.cancel_button) + + self.ok_button = QPushButton("确定") + self.ok_button.setDefault(True) + self.ok_button.clicked.connect(self.on_ok_clicked) + button_layout.addWidget(self.ok_button) + + layout.addLayout(button_layout) + + # 设置焦点 + self.password_input.setFocus() + + def on_ok_clicked(self): + """确定按钮点击""" + password = self.password_input.text() + + if not password: + QMessageBox.warning( + self, + "警告", + "请输入密码！" + ) + return + + self.password = password + self.accept() + + def on_cancel_clicked(self): + """取消按钮点击""" + self.password = None + self.reject() + + def get_password(self) -> str: + """获取输入的密码""" + return self.password + + +class SudoVerifyDialog(QDialog): + """sudo 密码验证对话框（显示验证进度）""" + + def __init__(self, parent=None): + super().__init__(parent) + self.password = None + self.init_ui() + + def init_ui(self): + """初始化界面""" + self.setWindowTitle("验证 sudo 密码") + self.setFixedSize(300, 150) + self.setWindowFlags(self.windowFlags() & ~Qt.WindowContextHelpButtonHint) + + # 主布局 + layout = QVBoxLayout(self) + layout.setSpacing(15) + layout.setContentsMargins(20, 20, 20, 20) + + # 提示文本 + self.status_label = QLabel("正在验证 sudo 密码...") + self.status_label.setAlignment(Qt.AlignCenter) + layout.addWidget(self.status_label) + + # 进度指示 + self.progress_label = QLabel("⏳") + self.progress_label.setAlignment(Qt.AlignCenter) + font = QFont() + font.setPointSize(24) + self.progress_label.setFont(font) + layout.addWidget(self.progress_label) + + # 取消按钮 + button_layout = QHBoxLayout() + button_layout.addStretch() + + self.cancel_button = QPushButton("取消") + self.cancel_button.clicked.connect(self.reject) + button_layout.addWidget(self.cancel_button) + + layout.addLayout(button_layout) + + def set_status(self, status: str): + """设置状态文本""" + self.status_label.setText(status) + + def set_success(self): + """设置验证成功状态""" + self.progress_label.setText("✅") + self.status_label.setText("验证成功！") + self.cancel_button.setText("关闭") + + def set_failure(self, message: str): + """设置验证失败状态""" + self.progress_label.setText("❌") + self.status_label.setText(f"验证失败：{message}") + self.cancel_button.setText("关闭") diff --git a/main.py b/main.py new file mode 100644 index 0000000..eb4aaf8 --- /dev/null +++ b/main.py @@ -0,0 +1,118 @@ +#!/usr/bin/env python3 +# -*- coding: utf-8 -*- +""" +Linux 日志删除助手 +在 UOS LiveCD 环境中运行，帮助用户快速清理磁盘空间 +""" +import sys +import os + +# 添加当前目录到 Python 路径 +current_dir = os.path.dirname(os.path.abspath(__file__)) +if current_dir not in sys.path: + sys.path.insert(0, current_dir) + +from PyQt5.QtWidgets import QApplication, QMessageBox +from PyQt5.QtCore import Qt +from PyQt5.QtGui import QFont + +from utils import check_sudo_needs_password, get_sudo_password, verify_sudo_password +from gui.main_window import MainWindow +from gui.sudo_dialog import SudoDialog + + +def check_environment(): + """检查运行环境""" + # 检查是否为 Linux 系统 + if sys.platform != 'linux': + QMessageBox.warning( + None, + "环境警告", + "本工具设计用于 Linux 系统，\n" + "当前运行在非 Linux 环境中，部分功能可能无法正常工作。" + ) + + # 检查是否有 root 权限或 sudo + if os.geteuid() != 0: + if check_sudo_needs_password(): + return True # 需要密码，稍后会提示 + else: + # sudo 不需要密码，可以直接使用 + return True + + return True # 已经是 root + + +def get_sudo_access(): + """获取 sudo 权限""" + # 检查是否已经是 root + if os.geteuid() == 0: + return None # 不需要密码 + + # 检查 sudo 是否需要密码 + if not check_sudo_needs_password(): + return None # 不需要密码 + + # 需要密码，显示密码输入对话框 + dialog = SudoDialog() + + if dialog.exec_() == SudoDialog.Accepted: + password = dialog.get_password() + + # 验证密码 + if verify_sudo_password(password): + return password + else: + QMessageBox.critical( + None, + "密码错误", + "sudo 密码不正确，程序将退出。" + ) + return None + else: + # 用户取消 + return None + + +def main(): + """主函数""" + # 设置高 DPI 支持 + QApplication.setAttribute(Qt.AA_EnableHighDpiScaling, True) + QApplication.setAttribute(Qt.AA_UseHighDpiPixmaps, True) + + # 创建应用 + app = QApplication(sys.argv) + + # 设置应用样式 + app.setStyle('Fusion') + + # 设置默认字体 + font = QFont() + font.setPointSize(10) + app.setFont(font) + + # 检查环境 + if not check_environment(): + sys.exit(1) + + # 获取 sudo 权限 + sudo_password = get_sudo_access() + + # 如果需要密码但没有提供，且不是 root，则退出 + if sudo_password is None and os.geteuid() != 0: + if check_sudo_needs_password(): + sys.exit(0) + + # 创建主窗口 + window = MainWindow(sudo_password) + window.show() + + # 启动时自动显示分区选择对话框 + window.switch_partition() + + # 运行应用 + sys.exit(app.exec_()) + + +if __name__ == '__main__': + main() diff --git a/requirements.txt b/requirements.txt new file mode 100644 index 0000000..00126bd --- /dev/null +++ b/requirements.txt @@ -0,0 +1,2 @@ +PyQt5>=5.15.0 +pyinstaller>=5.0 diff --git a/utils.py b/utils.py new file mode 100644 index 0000000..9cadd38 --- /dev/null +++ b/utils.py @@ -0,0 +1,295 @@ +""" +工具函数模块 +提供 sudo 检测、密码获取、命令执行等功能 +""" +import subprocess +import os +import sys +from typing import Tuple, Optional + + +def check_sudo_needs_password() -> bool: + """ + 检测 sudo 是否需要密码 + 返回 True 表示需要密码，False 表示不需要 + """ + try: + # 使用 sudo -n (non-interactive) 检测是否需要密码 + result = subprocess.run( + ['sudo', '-n', 'true'], + capture_output=True, + text=True, + timeout=5 + ) + # 返回码为 0 表示不需要密码 + return result.returncode != 0 + except (subprocess.TimeoutExpired, FileNotFoundError): + # 如果 sudo 命令不存在或超时，假设需要密码 + return True + + +def get_sudo_password() -> Optional[str]: + """ + 通过 GUI 对话框获取 sudo 密码 + 返回密码字符串，如果用户取消则返回 None + """ + try: + from PyQt5.QtWidgets import QInputDialog, QLineEdit + from PyQt5.QtCore import Qt + + password, ok = QInputDialog.getText( + None, + "需要 sudo 权限", + "请输入 sudo 密码：", + QLineEdit.Password + ) + + if ok and password: + return password + return None + except ImportError: + # 如果 PyQt5 不可用，使用命令行方式 + return _get_sudo_password_cli() + + +def _get_sudo_password_cli() -> Optional[str]: + """ + 通过命令行获取 sudo 密码（备用方案） + """ + try: + import getpass + password = getpass.getpass("请输入 sudo 密码: ") + return password if password else None + except (EOFError, KeyboardInterrupt): + return None + + +def verify_sudo_password(password: str) -> bool: + """ + 验证 sudo 密码是否正确 + """ + try: + result = subprocess.run( + ['sudo', '-S', 'true'], + input=password + '\n', + capture_output=True, + text=True, + timeout=5 + ) + return result.returncode == 0 + except (subprocess.TimeoutExpired, FileNotFoundError): + return False + + +def run_command( + cmd: list, + use_sudo: bool = False, + sudo_password: Optional[str] = None, + timeout: int = 30 +) -> Tuple[str, str, int]: + """ + 执行系统命令 + + 参数: + cmd: 命令列表，如 ['ls', '-la'] + use_sudo: 是否使用 sudo + sudo_password: sudo 密码（如果为 None 且 use_sudo=True，则使用 -n 模式） + timeout: 超时时间（秒） + + 返回: + (stdout, stderr, returncode) + """ + try: + if use_sudo: + if sudo_password: + # 使用 -S 从 stdin 读取密码 + full_cmd = ['sudo', '-S'] + cmd + result = subprocess.run( + full_cmd, + input=sudo_password + '\n', + capture_output=True, + text=True, + timeout=timeout + ) + else: + # 使用 -n 非交互模式 + full_cmd = ['sudo', '-n'] + cmd + result = subprocess.run( + full_cmd, + capture_output=True, + text=True, + timeout=timeout + ) + else: + result = subprocess.run( + cmd, + capture_output=True, + text=True, + timeout=timeout + ) + + return result.stdout, result.stderr, result.returncode + + except subprocess.TimeoutExpired: + return "", f"命令超时 ({timeout}秒)", -1 + except FileNotFoundError: + return "", f"命令不存在: {cmd[0]}", -1 + except Exception as e: + return "", str(e), -1 + + +def format_size(size_bytes: int) -> str: + """ + 将字节数格式化为人类可读的大小 + """ + if size_bytes < 0: + return "未知" + + units = ['B', 'KB', 'MB', 'GB', 'TB'] + unit_index = 0 + size = float(size_bytes) + + while size >= 1024.0 and unit_index < len(units) - 1: + size /= 1024.0 + unit_index += 1 + + if unit_index == 0: + return f"{int(size)} B" + else: + return f"{size:.1f} {units[unit_index]}" + + +def is_root() -> bool: + """ + 检查当前用户是否为 root + """ + return os.geteuid() == 0 + + +def get_mount_point(path: str) -> Optional[str]: + """ + 获取路径所在的挂载点 + """ + try: + path = os.path.realpath(path) + while path != '/': + if os.path.ismount(path): + return path + path = os.path.dirname(path) + return '/' + except Exception: + return None + + +def is_virtual_filesystem(path: str) -> bool: + """ + 检查路径是否为虚拟文件系统 + """ + virtual_fs_paths = [ + '/proc', + '/sys', + '/dev', + '/run', + '/tmp', + '/var/run', + '/var/lock', + ] + + # 规范化路径 + path = os.path.normpath(path) + + for vfspath in virtual_fs_paths: + if path == vfspath or path.startswith(vfspath + '/'): + return True + + return False + + +def safe_delete(path: str, use_sudo: bool = True, sudo_password: Optional[str] = None) -> Tuple[bool, str]: + """ + 安全删除文件或目录 + + 参数: + path: 要删除的路径 + use_sudo: 是否使用 sudo + sudo_password: sudo 密码 + + 返回: + (success, message) + """ + if not os.path.exists(path): + return False, f"路径不存在: {path}" + + # 获取删除前的大小 + try: + if os.path.isfile(path): + size = os.path.getsize(path) + else: + size = get_directory_size(path) + except Exception: + size = 0 + + # 执行删除 + try: + if os.path.isfile(path) or os.path.islink(path): + cmd = ['rm', '-f', path] + else: + cmd = ['rm', '-rf', path] + + stdout, stderr, returncode = run_command( + cmd, + use_sudo=use_sudo, + sudo_password=sudo_password, + timeout=300 # 删除大目录可能需要较长时间 + ) + + if returncode == 0: + return True, f"成功删除，释放 {format_size(size)}" + else: + return False, f"删除失败: {stderr}" + + except Exception as e: + return False, f"删除时发生错误: {str(e)}" + + +def get_directory_size(path: str) -> int: + """ + 获取目录大小（字节） + """ + total_size = 0 + try: + for dirpath, dirnames, filenames in os.walk(path): + # 跳过虚拟文件系统 + if is_virtual_filesystem(dirpath): + dirnames.clear() + continue + + for filename in filenames: + filepath = os.path.join(dirpath, filename) + try: + if not os.path.islink(filepath): + total_size += os.path.getsize(filepath) + except (OSError, PermissionError): + pass + except (OSError, PermissionError): + pass + + return total_size + + +def setup_logging(): + """ + 配置日志系统 + """ + import logging + + logging.basicConfig( + level=logging.INFO, + format='%(asctime)s - %(name)s - %(levelname)s - %(message)s', + handlers=[ + logging.StreamHandler(sys.stdout), + logging.FileHandler('/tmp/log_cleaner.log', encoding='utf-8') + ] + ) + + return logging.getLogger('LogCleaner')