296 lines
7.4 KiB
Python
296 lines
7.4 KiB
Python
"""
|
|
工具函数模块
|
|
提供 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')
|