490 lines
14 KiB
Python
490 lines
14 KiB
Python
"""
|
|
文件扫描模块
|
|
提供目录大小扫描、文件列表等功能
|
|
"""
|
|
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
|