""" 工具函数模块 提供 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')