293 lines
9.0 KiB
Python
293 lines
9.0 KiB
Python
#!/usr/bin/env python3
|
|
# -*- coding: utf-8 -*-
|
|
"""
|
|
依赖检查模块
|
|
检查系统是否安装了必要的文件系统工具和其他依赖
|
|
"""
|
|
|
|
import subprocess
|
|
import logging
|
|
from typing import Dict, List, Tuple, Optional
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
|
|
class DependencyChecker:
|
|
"""依赖检查器"""
|
|
|
|
# 定义依赖项及其检查命令
|
|
DEPENDENCIES = {
|
|
# 文件系统工具
|
|
'ext4': {
|
|
'name': 'ext4 文件系统支持',
|
|
'commands': ['mkfs.ext4', 'fsck.ext4', 'resize2fs'],
|
|
'package': {
|
|
'centos': 'e2fsprogs',
|
|
'arch': 'e2fsprogs',
|
|
'ubuntu': 'e2fsprogs'
|
|
}
|
|
},
|
|
'xfs': {
|
|
'name': 'XFS 文件系统支持',
|
|
'commands': ['mkfs.xfs', 'xfs_repair', 'xfs_growfs'],
|
|
'package': {
|
|
'centos': 'xfsprogs',
|
|
'arch': 'xfsprogs',
|
|
'ubuntu': 'xfsprogs'
|
|
}
|
|
},
|
|
'ntfs': {
|
|
'name': 'NTFS 文件系统支持',
|
|
'commands': ['mkfs.ntfs', 'ntfsfix'],
|
|
'package': {
|
|
'centos': 'ntfs-3g',
|
|
'arch': 'ntfs-3g',
|
|
'ubuntu': 'ntfs-3g'
|
|
}
|
|
},
|
|
'fat32': {
|
|
'name': 'FAT32/VFAT 文件系统支持',
|
|
'commands': ['mkfs.vfat', 'fsck.vfat'],
|
|
'package': {
|
|
'centos': 'dosfstools',
|
|
'arch': 'dosfstools',
|
|
'ubuntu': 'dosfstools'
|
|
}
|
|
},
|
|
# 分区工具
|
|
'parted': {
|
|
'name': '分区工具',
|
|
'commands': ['parted', 'partprobe'],
|
|
'package': {
|
|
'centos': 'parted',
|
|
'arch': 'parted',
|
|
'ubuntu': 'parted'
|
|
}
|
|
},
|
|
# RAID 工具
|
|
'mdadm': {
|
|
'name': 'RAID 管理工具',
|
|
'commands': ['mdadm'],
|
|
'package': {
|
|
'centos': 'mdadm',
|
|
'arch': 'mdadm',
|
|
'ubuntu': 'mdadm'
|
|
}
|
|
},
|
|
# LVM 工具
|
|
'lvm': {
|
|
'name': 'LVM 逻辑卷管理',
|
|
'commands': ['pvcreate', 'vgcreate', 'lvcreate', 'pvs', 'vgs', 'lvs'],
|
|
'package': {
|
|
'centos': 'lvm2',
|
|
'arch': 'lvm2',
|
|
'ubuntu': 'lvm2'
|
|
}
|
|
},
|
|
# SMART 监控
|
|
'smartctl': {
|
|
'name': 'SMART 磁盘健康监控',
|
|
'commands': ['smartctl'],
|
|
'package': {
|
|
'centos': 'smartmontools',
|
|
'arch': 'smartmontools',
|
|
'ubuntu': 'smartmontools'
|
|
},
|
|
'optional': True
|
|
},
|
|
# 其他工具
|
|
'lsof': {
|
|
'name': '进程占用查询',
|
|
'commands': ['lsof', 'fuser'],
|
|
'package': {
|
|
'centos': 'lsof, psmisc',
|
|
'arch': 'lsof, psmisc',
|
|
'ubuntu': 'lsof, psmisc'
|
|
},
|
|
'optional': True
|
|
},
|
|
'sfdisk': {
|
|
'name': '分区表备份/恢复',
|
|
'commands': ['sfdisk'],
|
|
'package': {
|
|
'centos': 'util-linux',
|
|
'arch': 'util-linux',
|
|
'ubuntu': 'util-linux'
|
|
},
|
|
'optional': True
|
|
}
|
|
}
|
|
|
|
def __init__(self):
|
|
self._cache = {}
|
|
self._os_type = self._detect_os()
|
|
|
|
def _detect_os(self) -> str:
|
|
"""检测操作系统类型"""
|
|
try:
|
|
with open('/etc/os-release', 'r') as f:
|
|
content = f.read().lower()
|
|
if 'centos' in content or 'rhel' in content or 'fedora' in content:
|
|
return 'centos'
|
|
elif 'arch' in content:
|
|
return 'arch'
|
|
elif 'ubuntu' in content or 'debian' in content:
|
|
return 'ubuntu'
|
|
except Exception:
|
|
pass
|
|
return 'centos' # 默认
|
|
|
|
def _command_exists(self, cmd: str) -> bool:
|
|
"""检查命令是否存在"""
|
|
try:
|
|
result = subprocess.run(
|
|
['which', cmd],
|
|
stdout=subprocess.PIPE,
|
|
stderr=subprocess.PIPE,
|
|
check=False
|
|
)
|
|
return result.returncode == 0
|
|
except Exception:
|
|
return False
|
|
|
|
def check_dependency(self, key: str) -> Tuple[bool, List[str], List[str]]:
|
|
"""
|
|
检查单个依赖项
|
|
返回: (是否全部安装, 已安装的命令列表, 未安装的命令列表)
|
|
"""
|
|
if key in self._cache:
|
|
return self._cache[key]
|
|
|
|
dep = self.DEPENDENCIES.get(key)
|
|
if not dep:
|
|
return False, [], []
|
|
|
|
installed = []
|
|
missing = []
|
|
|
|
for cmd in dep['commands']:
|
|
if self._command_exists(cmd):
|
|
installed.append(cmd)
|
|
else:
|
|
missing.append(cmd)
|
|
|
|
result = (len(missing) == 0, installed, missing)
|
|
self._cache[key] = result
|
|
return result
|
|
|
|
def check_all(self) -> Dict[str, dict]:
|
|
"""
|
|
检查所有依赖项
|
|
返回: 依赖项状态字典
|
|
"""
|
|
results = {}
|
|
for key, dep in self.DEPENDENCIES.items():
|
|
is_complete, installed, missing = self.check_dependency(key)
|
|
results[key] = {
|
|
'name': dep['name'],
|
|
'is_complete': is_complete,
|
|
'installed': installed,
|
|
'missing': missing,
|
|
'package': dep['package'].get(self._os_type, dep['package']['centos']),
|
|
'optional': dep.get('optional', False),
|
|
'commands': dep['commands']
|
|
}
|
|
return results
|
|
|
|
def get_install_commands(self, results: Dict[str, dict] = None) -> Dict[str, List[str]]:
|
|
"""
|
|
获取安装命令
|
|
返回: {包名: [依赖项名称列表]}
|
|
"""
|
|
if results is None:
|
|
results = self.check_all()
|
|
|
|
packages = {}
|
|
for key, status in results.items():
|
|
if not status['is_complete'] and not status['optional']:
|
|
pkg = status['package']
|
|
if pkg not in packages:
|
|
packages[pkg] = []
|
|
packages[pkg].append(status['name'])
|
|
|
|
return packages
|
|
|
|
def get_summary(self) -> Dict[str, any]:
|
|
"""
|
|
获取依赖检查摘要
|
|
"""
|
|
results = self.check_all()
|
|
|
|
total = len(results)
|
|
complete = sum(1 for r in results.values() if r['is_complete'])
|
|
optional_missing = sum(1 for r in results.values() if not r['is_complete'] and r['optional'])
|
|
required_missing = sum(1 for r in results.values() if not r['is_complete'] and not r['optional'])
|
|
|
|
# 按类别分组
|
|
categories = {
|
|
'filesystem': ['ext4', 'xfs', 'ntfs', 'fat32'],
|
|
'partition': ['parted'],
|
|
'raid': ['mdadm'],
|
|
'lvm': ['lvm'],
|
|
'other': ['smartctl', 'lsof', 'sfdisk']
|
|
}
|
|
|
|
category_status = {}
|
|
for cat, keys in categories.items():
|
|
cat_results = {k: results[k] for k in keys if k in results}
|
|
category_status[cat] = cat_results
|
|
|
|
return {
|
|
'total': total,
|
|
'complete': complete,
|
|
'optional_missing': optional_missing,
|
|
'required_missing': required_missing,
|
|
'results': results,
|
|
'categories': category_status,
|
|
'install_commands': self.get_install_commands(results),
|
|
'os_type': self._os_type
|
|
}
|
|
|
|
|
|
# 全局实例
|
|
_checker = None
|
|
|
|
def get_dependency_checker() -> DependencyChecker:
|
|
"""获取全局依赖检查器实例"""
|
|
global _checker
|
|
if _checker is None:
|
|
_checker = DependencyChecker()
|
|
return _checker
|
|
|
|
|
|
if __name__ == '__main__':
|
|
# 测试
|
|
logging.basicConfig(level=logging.INFO)
|
|
checker = get_dependency_checker()
|
|
summary = checker.get_summary()
|
|
|
|
print("=" * 60)
|
|
print("依赖检查报告")
|
|
print("=" * 60)
|
|
print(f"操作系统: {summary['os_type']}")
|
|
print(f"总依赖项: {summary['total']}")
|
|
print(f"已安装: {summary['complete']}")
|
|
print(f"必需但缺失: {summary['required_missing']}")
|
|
print(f"可选缺失: {summary['optional_missing']}")
|
|
print()
|
|
|
|
for key, status in summary['results'].items():
|
|
icon = "✓" if status['is_complete'] else "○" if status['optional'] else "✗"
|
|
print(f"{icon} {status['name']}")
|
|
if not status['is_complete']:
|
|
print(f" 缺失命令: {', '.join(status['missing'])}")
|
|
print(f" 安装包: {status['package']}")
|
|
|
|
print()
|
|
print("安装命令:")
|
|
if summary['install_commands']:
|
|
for pkg, names in summary['install_commands'].items():
|
|
print(f" {pkg} # {', '.join(names)}")
|
|
else:
|
|
print(" 所有必需依赖已安装")
|