148 lines
6.2 KiB
Python
148 lines
6.2 KiB
Python
# system_info.py
|
||
import subprocess
|
||
import json
|
||
import os
|
||
import logging # 导入 logging 模块
|
||
|
||
logger = logging.getLogger(__name__) # 获取当前模块的 logger 实例
|
||
|
||
class SystemInfoManager:
|
||
def __init__(self):
|
||
pass
|
||
|
||
def _run_command(self, cmd, root_privilege=False, check_output=True):
|
||
"""
|
||
执行shell命令并返回stdout和stderr。
|
||
如果需要root权限，会尝试使用sudo。
|
||
所有输出和错误都会通过 logger 记录。
|
||
"""
|
||
full_cmd = []
|
||
if root_privilege:
|
||
# 检查当前是否已经是root用户
|
||
if os.geteuid() != 0:
|
||
full_cmd.append("sudo")
|
||
|
||
full_cmd.extend(cmd)
|
||
cmd_str = ' '.join(full_cmd) # 用于日志记录的完整命令字符串
|
||
|
||
logger.info(f"执行命令: {cmd_str}")
|
||
|
||
try:
|
||
# text=True 自动解码为字符串，encoding='utf-8' 确保正确处理中文
|
||
# check=check_output 表示如果命令返回非零退出码，则抛出CalledProcessError
|
||
result = subprocess.run(
|
||
full_cmd,
|
||
capture_output=True,
|
||
text=True,
|
||
check=check_output,
|
||
encoding='utf-8',
|
||
# 设置LANG环境变量，确保命令输出使用UTF-8编码，避免解析问题
|
||
env=dict(os.environ, LANG="en_US.UTF-8")
|
||
)
|
||
if result.stdout:
|
||
logger.debug(f"命令输出 (stdout):\n{result.stdout.strip()}")
|
||
if result.stderr:
|
||
logger.warning(f"命令输出 (stderr):\n{result.stderr.strip()}")
|
||
return result.stdout.strip(), result.stderr.strip()
|
||
except subprocess.CalledProcessError as e:
|
||
# 捕获命令执行失败的情况
|
||
error_msg = f"命令执行失败: {cmd_str}\n" \
|
||
f"退出码: {e.returncode}\n" \
|
||
f"标准输出: {e.stdout}\n" \
|
||
f"标准错误: {e.stderr}"
|
||
logger.error(error_msg) # 使用 logger 记录错误
|
||
raise ValueError(error_msg) # 抛出更易于处理的异常
|
||
except FileNotFoundError:
|
||
error_msg = f"命令未找到: {full_cmd[0]}。请确保已安装。"
|
||
logger.error(error_msg)
|
||
raise FileNotFoundError(error_msg)
|
||
except Exception as e:
|
||
error_msg = f"执行命令时发生未知错误: {e}"
|
||
logger.error(error_msg)
|
||
raise RuntimeError(error_msg)
|
||
|
||
def get_block_devices(self):
|
||
"""
|
||
获取所有块设备的信息，以JSON格式返回。
|
||
使用 lsblk -J 命令。
|
||
"""
|
||
try:
|
||
stdout, _ = self._run_command(["lsblk", "-J", "-o", "NAME,FSTYPE,SIZE,MOUNTPOINT,RO,TYPE,UUID,PARTUUID,VENDOR,MODEL,SERIAL,MAJ:MIN,PKNAME"], root_privilege=False)
|
||
data = json.loads(stdout)
|
||
logger.info("成功获取块设备信息。")
|
||
return data.get('blockdevices', [])
|
||
except Exception as e:
|
||
logger.error(f"获取块设备信息失败: {e}") # 使用 logger 记录错误
|
||
return []
|
||
|
||
def get_mdadm_arrays(self):
|
||
"""
|
||
获取所有RAID阵列的详细信息。
|
||
使用 mdadm --detail --scan 命令。
|
||
"""
|
||
try:
|
||
stdout, _ = self._run_command(["mdadm", "--detail", "--scan"], root_privilege=False)
|
||
logger.info("成功获取RAID阵列信息。")
|
||
return stdout
|
||
except Exception as e:
|
||
logger.error(f"获取RAID阵列信息失败: {e}")
|
||
return "无法获取RAID阵列信息。"
|
||
|
||
def get_lvm_info(self):
|
||
"""
|
||
获取LVM的物理卷、卷组、逻辑卷信息。
|
||
使用 pvs, vgs, lvs 命令，并尝试获取JSON格式。
|
||
"""
|
||
lvm_info = {}
|
||
try:
|
||
stdout_pvs, _ = self._run_command(["pvs", "--reportformat", "json"], root_privilege=False)
|
||
lvm_info['pvs'] = json.loads(stdout_pvs).get('report', [])[0].get('pv', [])
|
||
logger.info("成功获取物理卷信息。")
|
||
except Exception as e:
|
||
logger.error(f"获取物理卷信息失败: {e}")
|
||
lvm_info['pvs'] = []
|
||
|
||
try:
|
||
stdout_vgs, _ = self._run_command(["vgs", "--reportformat", "json"], root_privilege=False)
|
||
lvm_info['vgs'] = json.loads(stdout_vgs).get('report', [])[0].get('vg', [])
|
||
logger.info("成功获取卷组信息。")
|
||
except Exception as e:
|
||
logger.error(f"获取卷组信息失败: {e}")
|
||
lvm_info['vgs'] = []
|
||
|
||
try:
|
||
stdout_lvs, _ = self._run_command(["lvs", "--reportformat", "json"], root_privilege=False)
|
||
lvm_info['lvs'] = json.loads(stdout_lvs).get('report', [])[0].get('lv', [])
|
||
logger.info("成功获取逻辑卷信息。")
|
||
except Exception as e:
|
||
logger.error(f"获取逻辑卷信息失败: {e}")
|
||
lvm_info['lvs'] = []
|
||
|
||
return lvm_info
|
||
|
||
# 示例用法 (可以在此模块中添加测试代码)
|
||
if __name__ == "__main__":
|
||
# 为了在单独运行时也能看到日志输出，这里简单配置一下
|
||
logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s')
|
||
manager = SystemInfoManager()
|
||
logger.info("--- 块设备信息 ---")
|
||
devices = manager.get_block_devices()
|
||
for dev in devices:
|
||
logger.info(f" NAME: {dev.get('name')}, TYPE: {dev.get('type')}, SIZE: {dev.get('size')}, MOUNTPOINT: {dev.get('mountpoint')}")
|
||
|
||
logger.info("\n--- RAID 阵列信息 ---")
|
||
raid_info = manager.get_mdadm_arrays()
|
||
logger.info(raid_info)
|
||
|
||
logger.info("\n--- LVM 信息 ---")
|
||
lvm_info = manager.get_lvm_info()
|
||
logger.info("物理卷 (PVs):")
|
||
for pv in lvm_info['pvs']:
|
||
logger.info(f" {pv.get('pv_name')} (VG: {pv.get('vg_name')}, Size: {pv.get('pv_size')})")
|
||
logger.info("卷组 (VGs):")
|
||
for vg in lvm_info['vgs']:
|
||
logger.info(f" {vg.get('vg_name')} (Size: {vg.get('vg_size')}, PVs: {vg.get('pv_count')}, LVs: {vg.get('lv_count')})")
|
||
logger.info("逻辑卷 (LVs):")
|
||
for lv in lvm_info['lvs']:
|
||
logger.info(f" {lv.get('lv_name')} (VG: {lv.get('vg_name')}, Size: {lv.get('lv_size')})")
|