This commit is contained in:
zj
2026-02-02 18:38:41 +08:00
parent be5a0bc2c7
commit a90725d178
14 changed files with 2345 additions and 465 deletions

Binary file not shown.

Binary file not shown.

Binary file not shown.

445
dialogs.py Normal file
View File

@@ -0,0 +1,445 @@
# dialogs.py
import os
from PySide6.QtWidgets import (QDialog, QVBoxLayout, QHBoxLayout, QLabel,
QLineEdit, QComboBox, QDoubleSpinBox,
QPushButton, QFormLayout, QFileDialog,
QMessageBox, QCheckBox, QListWidget, QListWidgetItem)
from PySide6.QtCore import Qt
class CreatePartitionDialog(QDialog):
def __init__(self, parent=None, disk_path="", total_disk_mib=0.0, max_available_mib=0.0):
super().__init__(parent)
self.setWindowTitle("创建分区")
self.setMinimumWidth(300)
self.disk_path = disk_path
self.total_disk_mib = total_disk_mib # 磁盘总大小 (MiB)
self.max_available_mib = max_available_mib # 可用空间 (MiB)
self.partition_table_type_combo = QComboBox()
self.partition_table_type_combo.addItems(["gpt", "msdos"])
self.size_spinbox = QDoubleSpinBox()
self.size_spinbox.setMinimum(0.1) # 最小分区大小 0.1 GB
self.size_spinbox.setMaximum(self.max_available_mib / 1024.0) # 将 MiB 转换为 GB 显示
self.size_spinbox.setSuffix(" GB")
self.size_spinbox.setDecimals(2)
self.use_max_space_checkbox = QCheckBox("使用最大可用空间")
self.use_max_space_checkbox.setChecked(True) # 默认选中
self._setup_ui()
self._connect_signals()
self._initialize_state()
def _setup_ui(self):
layout = QVBoxLayout(self)
form_layout = QFormLayout()
form_layout.addRow("磁盘路径:", QLabel(self.disk_path))
form_layout.addRow("分区表类型:", self.partition_table_type_combo)
form_layout.addRow("分区大小:", self.size_spinbox)
form_layout.addRow("", self.use_max_space_checkbox) # 添加复选框
layout.addLayout(form_layout)
button_box = QHBoxLayout()
self.confirm_button = QPushButton("确定")
self.cancel_button = QPushButton("取消")
button_box.addWidget(self.confirm_button)
button_box.addWidget(self.cancel_button)
layout.addLayout(button_box)
def _connect_signals(self):
self.confirm_button.clicked.connect(self.accept)
self.cancel_button.clicked.connect(self.reject)
self.use_max_space_checkbox.stateChanged.connect(self._toggle_size_input)
def _initialize_state(self):
# 根据默认选中状态设置 spinbox
self._toggle_size_input(self.use_max_space_checkbox.checkState())
def _toggle_size_input(self, state):
if state == Qt.Checked:
self.size_spinbox.setDisabled(True)
self.size_spinbox.setValue(self.size_spinbox.maximum()) # 设置为最大可用 GB
else:
self.size_spinbox.setDisabled(False)
# 如果之前是最大值,取消勾选后,恢复到最小值或一个合理值
if self.size_spinbox.value() == self.size_spinbox.maximum():
self.size_spinbox.setValue(self.size_spinbox.minimum())
def get_partition_info(self):
size_gb = self.size_spinbox.value()
use_max_space = self.use_max_space_checkbox.isChecked()
if not use_max_space and size_gb <= 0:
QMessageBox.warning(self, "输入错误", "分区大小必须大于0。")
return None
if not use_max_space and size_gb > self.max_available_mib / 1024.0:
QMessageBox.warning(self, "输入错误", "分区大小不能超过最大可用空间。")
return None
return {
'disk_path': self.disk_path,
'partition_table_type': self.partition_table_type_combo.currentText(),
'size_gb': size_gb,
'total_disk_mib': self.total_disk_mib, # 传递磁盘总大小 (MiB)
'use_max_space': use_max_space # 返回此标志
}
class MountDialog(QDialog):
def __init__(self, parent=None, device_path=""):
super().__init__(parent)
self.setWindowTitle("挂载分区")
self.setMinimumWidth(300)
self.device_path = device_path
self.mount_point_input = QLineEdit()
self.add_to_fstab_checkbox = QCheckBox("开机自动挂载 (添加到 /etc/fstab)") # 新增
self._setup_ui()
self._connect_signals()
def _setup_ui(self):
layout = QVBoxLayout(self)
form_layout = QFormLayout()
form_layout.addRow("设备路径:", QLabel(self.device_path))
form_layout.addRow("挂载点:", self.mount_point_input)
form_layout.addRow("", self.add_to_fstab_checkbox) # 添加复选框
layout.addLayout(form_layout)
button_box = QHBoxLayout()
self.confirm_button = QPushButton("确定")
self.cancel_button = QPushButton("取消")
button_box.addWidget(self.confirm_button)
button_box.addWidget(self.cancel_button)
layout.addLayout(button_box)
# 默认挂载点,可以根据设备名生成
default_mount_point = f"/mnt/{os.path.basename(self.device_path)}"
self.mount_point_input.setText(default_mount_point)
def _connect_signals(self):
self.confirm_button.clicked.connect(self.accept)
self.cancel_button.clicked.connect(self.reject)
def get_mount_info(self):
mount_point = self.mount_point_input.text().strip()
add_to_fstab = self.add_to_fstab_checkbox.isChecked()
if not mount_point:
QMessageBox.warning(self, "输入错误", "挂载点不能为空。")
return None
if not mount_point.startswith('/'):
QMessageBox.warning(self, "输入错误", "挂载点必须是绝对路径 (以 '/' 开头)。")
return None
return {
'mount_point': mount_point,
'add_to_fstab': add_to_fstab
}
class CreateRaidDialog(QDialog):
def __init__(self, parent=None, available_devices=None):
super().__init__(parent)
self.setWindowTitle("创建 RAID 阵列")
self.setMinimumWidth(350)
self.available_devices = available_devices if available_devices is not None else []
self.selected_devices = []
self.raid_level_combo = QComboBox()
self.raid_level_combo.addItems(["raid0", "raid1", "raid5", "raid6", "raid10"])
self.chunk_size_spinbox = QDoubleSpinBox()
self.chunk_size_spinbox.setMinimum(4)
self.chunk_size_spinbox.setMaximum(1024)
self.chunk_size_spinbox.setSingleStep(4)
self.chunk_size_spinbox.setValue(512) # 默认值
self.chunk_size_spinbox.setSuffix(" KB")
self.device_list_widget = QListWidget()
for dev in self.available_devices:
item = QListWidgetItem(dev)
item.setFlags(item.flags() | Qt.ItemIsUserCheckable)
item.setCheckState(Qt.Unchecked)
self.device_list_widget.addItem(item)
self._setup_ui()
self._connect_signals()
def _setup_ui(self):
layout = QVBoxLayout(self)
form_layout = QFormLayout()
form_layout.addRow("RAID 级别:", self.raid_level_combo)
form_layout.addRow("Chunk 大小:", self.chunk_size_spinbox)
form_layout.addRow("选择设备:", self.device_list_widget)
layout.addLayout(form_layout)
button_box = QHBoxLayout()
self.confirm_button = QPushButton("确定")
self.cancel_button = QPushButton("取消")
button_box.addWidget(self.confirm_button)
button_box.addWidget(self.cancel_button)
layout.addLayout(button_box)
def _connect_signals(self):
self.confirm_button.clicked.connect(self.accept)
self.cancel_button.clicked.connect(self.reject)
def get_raid_info(self):
self.selected_devices = []
for i in range(self.device_list_widget.count()):
item = self.device_list_widget.item(i)
if item.checkState() == Qt.Checked:
self.selected_devices.append(item.text())
if not self.selected_devices:
QMessageBox.warning(self, "输入错误", "请选择至少一个设备来创建 RAID 阵列。")
return None
raid_level = self.raid_level_combo.currentText()
num_devices = len(self.selected_devices)
# RAID level specific checks for minimum devices
if raid_level == "raid0" and num_devices < 1: # RAID0 can be 1 device in some contexts, but usually >1
QMessageBox.warning(self, "输入错误", "RAID0 至少需要一个设备。")
return None
elif raid_level == "raid1" and num_devices < 2:
QMessageBox.warning(self, "输入错误", "RAID1 至少需要两个设备。")
return None
elif raid_level == "raid5" and num_devices < 3:
QMessageBox.warning(self, "输入错误", "RAID5 至少需要三个设备。")
return None
elif raid_level == "raid6" and num_devices < 4:
QMessageBox.warning(self, "输入错误", "RAID6 至少需要四个设备。")
return None
elif raid_level == "raid10" and num_devices < 2: # RAID10 needs at least 2 (for 1+0)
QMessageBox.warning(self, "输入错误", "RAID10 至少需要两个设备。")
return None
return {
'devices': self.selected_devices,
'level': self.raid_level_combo.currentText().replace('raid', ''), # 移除 'raid' 前缀
'chunk_size': int(self.chunk_size_spinbox.value()) # KB
}
class CreatePvDialog(QDialog):
def __init__(self, parent=None, available_partitions=None):
super().__init__(parent)
self.setWindowTitle("创建物理卷 (PV)")
self.setMinimumWidth(300)
self.available_partitions = available_partitions if available_partitions is not None else []
self.device_combo_box = QComboBox()
self.device_combo_box.addItems(self.available_partitions)
self._setup_ui()
self._connect_signals()
def _setup_ui(self):
layout = QVBoxLayout(self)
form_layout = QFormLayout()
form_layout.addRow("选择设备:", self.device_combo_box)
layout.addLayout(form_layout)
button_box = QHBoxLayout()
self.confirm_button = QPushButton("确定")
self.cancel_button = QPushButton("取消")
button_box.addWidget(self.confirm_button)
button_box.addWidget(self.cancel_button)
layout.addLayout(button_box)
def _connect_signals(self):
self.confirm_button.clicked.connect(self.accept)
self.cancel_button.clicked.connect(self.reject)
def get_pv_info(self):
device_path = self.device_combo_box.currentText()
if not device_path:
QMessageBox.warning(self, "输入错误", "请选择一个设备来创建物理卷。")
return None
return {'device_path': device_path}
class CreateVgDialog(QDialog):
def __init__(self, parent=None, available_pvs=None):
super().__init__(parent)
self.setWindowTitle("创建卷组 (VG)")
self.setMinimumWidth(300)
self.available_pvs = available_pvs if available_pvs is not None else []
self.selected_pvs = []
self.vg_name_input = QLineEdit()
self.pv_list_widget = QListWidget()
for pv in self.available_pvs:
item = QListWidgetItem(pv)
item.setFlags(item.flags() | Qt.ItemIsUserCheckable)
item.setCheckState(Qt.Unchecked)
self.pv_list_widget.addItem(item)
self._setup_ui()
self._connect_signals()
def _setup_ui(self):
layout = QVBoxLayout(self)
form_layout = QFormLayout()
form_layout.addRow("卷组名称:", self.vg_name_input)
form_layout.addRow("选择物理卷:", self.pv_list_widget)
layout.addLayout(form_layout)
button_box = QHBoxLayout()
self.confirm_button = QPushButton("确定")
self.cancel_button = QPushButton("取消")
button_box.addWidget(self.confirm_button)
button_box.addWidget(self.cancel_button)
layout.addLayout(button_box)
def _connect_signals(self):
self.confirm_button.clicked.connect(self.accept)
self.cancel_button.clicked.connect(self.reject)
def get_vg_info(self):
vg_name = self.vg_name_input.text().strip()
self.selected_pvs = []
for i in range(self.pv_list_widget.count()):
item = self.pv_list_widget.item(i)
if item.checkState() == Qt.Checked:
self.selected_pvs.append(item.text())
if not vg_name:
QMessageBox.warning(self, "输入错误", "卷组名称不能为空。")
return None
if not self.selected_pvs:
QMessageBox.warning(self, "输入错误", "请选择至少一个物理卷来创建卷组。")
return None
return {
'vg_name': vg_name,
'pvs': self.selected_pvs
}
class CreateLvDialog(QDialog):
def __init__(self, parent=None, available_vgs=None, vg_sizes=None):
super().__init__(parent)
self.setWindowTitle("创建逻辑卷")
self.setMinimumWidth(300)
self.available_vgs = available_vgs if available_vgs is not None else []
self.vg_sizes = vg_sizes if vg_sizes is not None else {} # 存储每个 VG 的可用大小 (GB)
self.lv_name_input = QLineEdit()
self.vg_combo_box = QComboBox()
self.lv_size_spinbox = QDoubleSpinBox()
self.use_max_space_checkbox = QCheckBox("使用最大可用空间") # 新增复选框
self._setup_ui()
self._connect_signals()
self._initialize_state() # 初始化状态
def _setup_ui(self):
layout = QVBoxLayout(self)
form_layout = QFormLayout()
form_layout.addRow("逻辑卷名称:", self.lv_name_input)
form_layout.addRow("选择卷组:", self.vg_combo_box)
form_layout.addRow("逻辑卷大小 (GB):", self.lv_size_spinbox)
form_layout.addRow("", self.use_max_space_checkbox) # 添加复选框
layout.addLayout(form_layout)
button_box = QHBoxLayout()
self.confirm_button = QPushButton("确定")
self.cancel_button = QPushButton("取消")
button_box.addWidget(self.confirm_button)
button_box.addWidget(self.cancel_button)
layout.addLayout(button_box)
self.lv_size_spinbox.setMinimum(0.1) # 逻辑卷最小大小,例如 0.1 GB
self.lv_size_spinbox.setSuffix(" GB")
self.lv_size_spinbox.setDecimals(2)
def _connect_signals(self):
self.confirm_button.clicked.connect(self.accept)
self.cancel_button.clicked.connect(self.reject)
self.vg_combo_box.currentIndexChanged.connect(self._update_size_options) # 卷组选择改变时更新大小选项
self.use_max_space_checkbox.stateChanged.connect(self._toggle_size_input) # 复选框状态改变时切换输入
def _initialize_state(self):
self.vg_combo_box.addItems(self.available_vgs)
self._update_size_options() # 首次调用以设置初始状态
self.use_max_space_checkbox.setChecked(True) # 默认选中“使用最大可用空间”
def _update_size_options(self):
"""根据选中的卷组更新逻辑卷大小的选项。"""
selected_vg = self.vg_combo_box.currentText()
max_size_gb = self.vg_sizes.get(selected_vg, 0.0)
# 确保 max_size_gb 至少是 spinbox 的最小值,以防卷组可用空间过小导致 UI 问题
if max_size_gb < self.lv_size_spinbox.minimum():
self.lv_size_spinbox.setMinimum(max_size_gb) # 临时将最小值设为实际最大值
else:
self.lv_size_spinbox.setMinimum(0.1) # 恢复正常最小值
self.lv_size_spinbox.setMaximum(max_size_gb) # 设置最大值
# 如果选中了“使用最大可用空间”,则将 spinbox 值设置为最大值
if self.use_max_space_checkbox.isChecked():
self.lv_size_spinbox.setValue(max_size_gb)
else:
# 如果当前值超过了新的最大值,则调整为新的最大值
if self.lv_size_spinbox.value() > max_size_gb:
self.lv_size_spinbox.setValue(max_size_gb)
# 如果当前值小于新的最小值 (例如VG可用空间变为0最小值被调整为0),则调整
elif self.lv_size_spinbox.value() < self.lv_size_spinbox.minimum():
self.lv_size_spinbox.setValue(self.lv_size_spinbox.minimum())
def _toggle_size_input(self, state):
"""根据“使用最大可用空间”复选框的状态切换大小输入框的启用/禁用状态。"""
if state == Qt.Checked:
self.lv_size_spinbox.setDisabled(True)
self.lv_size_spinbox.setValue(self.lv_size_spinbox.maximum()) # 设置为最大值
else:
self.lv_size_spinbox.setDisabled(False)
# 如果之前是最大值,取消勾选后,恢复到最小值或一个合理值
if self.lv_size_spinbox.value() == self.lv_size_spinbox.maximum():
self.lv_size_spinbox.setValue(self.lv_size_spinbox.minimum())
def get_lv_info(self):
lv_name = self.lv_name_input.text().strip()
vg_name = self.vg_combo_box.currentText()
size_gb = self.lv_size_spinbox.value()
use_max_space = self.use_max_space_checkbox.isChecked() # 获取复选框状态
if not lv_name:
QMessageBox.warning(self, "输入错误", "逻辑卷名称不能为空。")
return None
if not vg_name:
QMessageBox.warning(self, "输入错误", "请选择一个卷组。")
return None
if not use_max_space and size_gb <= 0:
QMessageBox.warning(self, "输入错误", "逻辑卷大小必须大于0。")
return None
if not use_max_space and size_gb > self.vg_sizes.get(vg_name, 0.0):
QMessageBox.warning(self, "输入错误", "逻辑卷大小不能超过卷组的可用空间。")
return None
return {
'lv_name': lv_name,
'vg_name': vg_name,
'size_gb': size_gb,
'use_max_space': use_max_space # 返回此标志
}

View File

@@ -1,237 +1,536 @@
# disk_operations.py # disk_operations.py
import os import subprocess
import logging import logging
import os
import re
from PySide6.QtWidgets import QMessageBox, QInputDialog from PySide6.QtWidgets import QMessageBox, QInputDialog
from system_info import SystemInfoManager # 引入 SystemInfoManager 来复用 _run_command
logger = logging.getLogger(__name__) # 获取当前模块的 logger 实例 # 导入我们自己编写的系统信息管理模块
from system_info import SystemInfoManager
logger = logging.getLogger(__name__)
class DiskOperations: class DiskOperations:
def __init__(self): def __init__(self):
self.system_manager = SystemInfoManager() # 复用 SystemInfoManager 的命令执行器 self.system_manager = SystemInfoManager() # 实例化 SystemInfoManager
def _get_device_path(self, device_name): def _execute_shell_command(self, command_list, error_message, root_privilege=True,
suppress_critical_dialog_on_stderr_match=None, input_data=None):
""" """
辅助函数:将设备名(如 'sda1')转换为完整路径(如 '/dev/sda1' 通用地运行一个 shell 命令,并处理错误
:param command_list: 命令及其参数的列表。
:param error_message: 命令失败时显示给用户的错误消息。
:param root_privilege: 如果为 True则使用 sudo 执行命令。
:param suppress_critical_dialog_on_stderr_match: 如果 stderr 包含此字符串,则不显示关键错误对话框。
:param input_data: 传递给命令stdin的数据 (str)。
:return: (True/False, stdout_str, stderr_str)
""" """
if not device_name.startswith('/dev/'): # 确保 command_list 中的所有项都是字符串
return f'/dev/{device_name}' if not all(isinstance(arg, str) for arg in command_list):
return device_name logger.error(f"命令列表包含非字符串元素: {command_list}")
QMessageBox.critical(None, "错误", f"内部错误:尝试执行的命令包含无效参数。\n命令详情: {command_list}")
return False, "", "内部错误:命令参数类型不正确。"
def mount_partition(self, device_name, mount_point=None): if root_privilege:
""" command_list = ["sudo"] + command_list
挂载指定的分区。
如果未提供挂载点,会尝试查找现有挂载点或提示用户输入。
"""
dev_path = self._get_device_path(device_name)
logger.info(f"尝试挂载设备: {dev_path}")
if not mount_point: full_cmd_str = ' '.join(command_list)
# 尝试从系统信息中获取当前挂载点 logger.debug(f"执行命令: {full_cmd_str}")
devices = self.system_manager.get_block_devices()
found_mount_point = None
for dev in devices:
if dev.get('name') == device_name:
found_mount_point = dev.get('mountpoint')
break
if 'children' in dev:
for child in dev['children']:
if child.get('name') == device_name:
found_mount_point = child.get('mountpoint')
break
if found_mount_point:
break
if found_mount_point and found_mount_point != '[SWAP]' and found_mount_point != '': try:
# 如果设备已经有挂载点并且不是SWAP则直接使用 result = subprocess.run(
mount_point = found_mount_point command_list,
logger.info(f"设备 {dev_path} 已经挂载到 {mount_point}") capture_output=True,
QMessageBox.information(None, "信息", f"设备 {dev_path} 已经挂载到 {mount_point}") text=True,
return True # 已经挂载,视为成功 check=True,
encoding='utf-8',
input=input_data
)
logger.info(f"命令成功: {full_cmd_str}")
return True, result.stdout.strip(), result.stderr.strip()
except subprocess.CalledProcessError as e:
stderr_output = e.stderr.strip()
logger.error(f"命令失败: {full_cmd_str}")
logger.error(f"退出码: {e.returncode}")
logger.error(f"标准输出: {e.stdout.strip()}")
logger.error(f"标准错误: {stderr_output}")
if suppress_critical_dialog_on_stderr_match and \
suppress_critical_dialog_on_stderr_match in stderr_output:
logger.info(f"错误信息 '{stderr_output}' 匹配抑制条件,不显示关键错误对话框。")
else: else:
# 如果没有挂载点或者挂载点是SWAP则提示用户输入 QMessageBox.critical(None, "错误", f"{error_message}\n错误详情: {stderr_output}")
mount_point, ok = QInputDialog.getText(None, "挂载分区", return False, e.stdout.strip(), stderr_output
f"请输入 {dev_path} 的挂载点 (例如: /mnt/data):", except FileNotFoundError:
text=f"/mnt/{device_name}") QMessageBox.critical(None, "错误", f"命令 '{command_list[0]}' 未找到。请确保已安装相关工具。")
if not ok or not mount_point: logger.error(f"命令 '{command_list[0]}' 未找到。")
logger.info("用户取消了挂载操作或未提供挂载点") return False, "", f"命令 '{command_list[0]}' 未找到"
return False
# 确保挂载点目录存在
if not os.path.exists(mount_point):
try:
os.makedirs(mount_point, exist_ok=True)
logger.info(f"创建挂载点目录: {mount_point}")
except OSError as e:
logger.error(f"创建挂载点目录失败 {mount_point}: {e}")
QMessageBox.critical(None, "错误", f"创建挂载点目录失败: {e}")
return False
try:
stdout, stderr = self.system_manager._run_command(["mount", dev_path, mount_point], root_privilege=True)
logger.info(f"成功挂载 {dev_path}{mount_point}")
QMessageBox.information(None, "成功", f"成功挂载 {dev_path}{mount_point}")
return True
except Exception as e: except Exception as e:
logger.error(f"挂载 {dev_path} 失败: {e}") QMessageBox.critical(None, "错误", f"执行命令时发生未知错误: {e}")
QMessageBox.critical(None, "错误", f"挂载 {dev_path} 失败: {e}") logger.error(f"执行命令 {full_cmd_str} 时发生未知错误: {e}")
return False return False, "", str(e)
def unmount_partition(self, device_name): def _get_fstab_path(self):
""" return "/etc/fstab"
卸载指定的分区。
"""
dev_path = self._get_device_path(device_name)
logger.info(f"尝试卸载设备: {dev_path}")
# 尝试从系统信息中获取当前挂载点 def create_partition(self, disk_path, partition_table_type, size_gb, total_disk_mib, use_max_space):
current_mount_point = None """
在指定磁盘上创建分区。
:param disk_path: 磁盘设备路径,例如 /dev/sdb。
:param partition_table_type: 分区表类型,'gpt''msdos'
:param size_gb: 分区大小GB
:param total_disk_mib: 磁盘总大小 (MiB)。
:param use_max_space: 是否使用最大可用空间创建分区。
:return: 新创建的分区路径,如果失败则为 None。
"""
if not isinstance(disk_path, str) or not isinstance(partition_table_type, str):
logger.error(f"尝试创建分区时传入无效的磁盘路径或分区表类型。磁盘: {disk_path} (类型: {type(disk_path)}), 类型: {partition_table_type} (类型: {type(partition_table_type)})")
QMessageBox.critical(None, "错误", "无效的磁盘路径或分区表类型。")
return None
logger.info(f"尝试在 {disk_path} 上创建 {size_gb}GB 的分区,分区表类型为 {partition_table_type}")
# 1. 检查磁盘是否已经有分区表,如果没有则创建
# parted -s /dev/sdb print
try:
stdout, _ = self.system_manager._run_command(["parted", "-s", disk_path, "print"], root_privilege=True)
if "unrecognised disk label" in stdout:
logger.info(f"磁盘 {disk_path} 没有分区表,将创建 {partition_table_type} 分区表。")
success, _, _ = self._execute_shell_command(
["parted", "-s", disk_path, "mklabel", partition_table_type],
f"创建分区表 {partition_table_type} 失败"
)
if not success:
return None
else:
logger.info(f"磁盘 {disk_path} 已有分区表。")
except Exception as e:
logger.error(f"检查或创建分区表失败: {e}")
QMessageBox.critical(None, "错误", f"检查或创建分区表失败: {e}")
return None
# 2. 获取下一个分区的起始扇区
start_sector = self.get_disk_next_partition_start_sector(disk_path)
if start_sector is None:
QMessageBox.critical(None, "错误", f"无法确定磁盘 {disk_path} 的分区起始位置。")
return None
# 3. 构建 parted 命令
parted_cmd = ["parted", "-s", disk_path, "mkpart", "primary"]
# 对于 GPT默认文件系统是 ext2。对于 msdos默认是 ext2。
# parted mkpart primary [fstype] start end
# 我们可以省略 fstype让用户稍后格式化。
# 计算结束位置
start_mib = start_sector * 512 / (1024 * 1024) # 转换为 MiB
if use_max_space:
end_mib = total_disk_mib # 使用磁盘总大小作为结束位置
parted_cmd.extend([f"{start_mib}MiB", f"{end_mib}MiB"])
else:
end_mib = start_mib + size_gb * 1024 # 将 GB 转换为 MiB
parted_cmd.extend([f"{start_mib}MiB", f"{end_mib}MiB"])
# 执行创建分区命令
success, stdout, stderr = self._execute_shell_command(
parted_cmd,
f"{disk_path} 上创建分区失败"
)
if not success:
return None
# 4. 刷新内核分区表
self._execute_shell_command(["partprobe", disk_path], f"刷新 {disk_path} 分区表失败", root_privilege=True)
# 5. 获取新创建的分区路径
# 通常新分区是磁盘的最后一个分区,例如 /dev/sdb1, /dev/sdb2 等
# 我们可以通过 lsblk 再次获取信息,找到最新的分区
try:
# 重新获取设备信息,找到最新的分区
devices = self.system_manager.get_block_devices() devices = self.system_manager.get_block_devices()
new_partition_path = None
for dev in devices: for dev in devices:
if dev.get('name') == device_name: if dev.get('path') == disk_path and dev.get('children'):
current_mount_point = dev.get('mountpoint') # 找到最新的子分区
break latest_partition = None
if 'children' in dev:
for child in dev['children']: for child in dev['children']:
if child.get('name') == device_name: if child.get('type') == 'part':
current_mount_point = child.get('mountpoint') if latest_partition is None or int(child.get('maj:min').split(':')[1]) > int(latest_partition.get('maj:min').split(':')[1]):
break latest_partition = child
if current_mount_point: if latest_partition:
new_partition_path = latest_partition.get('path')
break break
if not current_mount_point or current_mount_point == '[SWAP]' or current_mount_point == '': if new_partition_path:
logger.warning(f"设备 {dev_path} 未挂载或无法确定挂载点,无法卸载。") logger.info(f"成功在 {disk_path} 上创建分区: {new_partition_path}")
QMessageBox.warning(None, "警告", f"设备 {dev_path} 未挂载或无法确定挂载点,无法卸载。") return new_partition_path
return False else:
logger.error(f"{disk_path} 上创建分区成功,但未能确定新分区的设备路径。")
QMessageBox.warning(None, "警告", f"分区创建成功,但未能确定新分区的设备路径。请手动检查。")
return None
except Exception as e:
logger.error(f"获取新创建分区路径失败: {e}")
QMessageBox.critical(None, "错误", f"获取新创建分区路径失败: {e}")
return None
def get_disk_next_partition_start_sector(self, disk_path):
"""
获取磁盘上下一个可用的分区起始扇区。
:param disk_path: 磁盘设备路径,例如 /dev/sdb。
:return: 下一个分区起始扇区(整数),如果失败则为 None。
"""
if not isinstance(disk_path, str):
logger.error(f"传入 get_disk_next_partition_start_sector 的 disk_path 不是字符串: {disk_path} (类型: {type(disk_path)})")
return None
try: try:
# 尝试通过挂载点卸载,通常更可靠 stdout, _ = self.system_manager._run_command(["parted", "-s", disk_path, "unit", "s", "print", "free"], root_privilege=True)
stdout, stderr = self.system_manager._run_command(["umount", current_mount_point], root_privilege=True) # 查找最大的空闲空间
logger.info(f"成功卸载 {current_mount_point} ({dev_path})") free_space_pattern = re.compile(r'^\s*\d+\s+([\d.]+s)\s+([\d.]+s)\s+([\d.]+s)\s+Free Space', re.MULTILINE)
QMessageBox.information(None, "成功", f"成功卸载 {current_mount_point} ({dev_path})") matches = free_space_pattern.findall(stdout)
return True
except Exception as e:
logger.error(f"卸载 {current_mount_point} ({dev_path}) 失败: {e}")
QMessageBox.critical(None, "错误", f"卸载 {current_mount_point} ({dev_path}) 失败: {e}")
return False
def delete_partition(self, device_name): if matches:
# 找到所有空闲空间的起始扇区,并取最小的作为下一个分区的起始
# 或者,更准确地说,找到最后一个分区的结束扇区 + 1
# 简化处理:如果没有分区,从 2048s 开始 (常见对齐)
# 如果有分区,从最后一个分区的结束扇区 + 1 开始
# 先尝试获取所有分区信息
stdout_partitions, _ = self.system_manager._run_command(["parted", "-s", disk_path, "unit", "s", "print"], root_privilege=True)
partition_end_pattern = re.compile(r'^\s*\d+\s+[\d.]+s\s+([\d.]+s)', re.MULTILINE)
partition_end_sectors = [int(float(s.replace('s', ''))) for s in partition_end_pattern.findall(stdout_partitions)]
if partition_end_sectors:
# 最后一个分区的结束扇区 + 1
last_end_sector = max(partition_end_sectors)
# 确保对齐,通常是 1MiB (2048扇区) 或 4MiB (8192扇区)
# 这里我们简单地取下一个 1MiB 对齐的扇区
start_sector = (last_end_sector // 2048 + 1) * 2048
logger.debug(f"磁盘 {disk_path} 最后一个分区结束于 {last_end_sector}s下一个分区起始扇区建议为 {start_sector}s。")
return start_sector
else:
# 没有分区,从 2048s 开始
logger.debug(f"磁盘 {disk_path} 没有现有分区,建议起始扇区为 2048s。")
return 2048 # 2048扇区是 1MiB常见的起始对齐位置
else:
logger.warning(f"无法在磁盘 {disk_path} 上找到空闲空间信息。")
return None
except Exception as e:
logger.error(f"获取磁盘 {disk_path} 下一个分区起始扇区失败: {e}")
return None
def get_disk_next_partition_start_mib(self, disk_path):
"""
获取磁盘上下一个可用的分区起始位置 (MiB)。
:param disk_path: 磁盘设备路径,例如 /dev/sdb。
:return: 下一个分区起始位置 (MiB),如果失败则为 None。
"""
start_sector = self.get_disk_next_partition_start_sector(disk_path)
if start_sector is not None:
return start_sector * 512 / (1024 * 1024) # 转换为 MiB
return None
def delete_partition(self, device_path):
""" """
删除指定的分区。 删除指定的分区。
此操作不可逆,会丢失数据 :param device_path: 要删除的分区路径,例如 /dev/sdb1
:return: True 如果删除成功,否则 False。
""" """
dev_path = self._get_device_path(device_name) if not isinstance(device_path, str):
logger.info(f"尝试删除分区: {dev_path}") logger.error(f"尝试删除非字符串设备路径: {device_path} (类型: {type(device_path)})")
QMessageBox.critical(None, "错误", "无效的设备路径。")
# 安全检查: 确保是分区,而不是整个磁盘
# 简单的检查方法是看设备名是否包含数字 (如 sda1, nvme0n1p1)
if not any(char.isdigit() for char in device_name):
QMessageBox.warning(None, "警告", f"{dev_path} 看起来不是一个分区。为安全起见,不执行删除操作。")
logger.warning(f"尝试删除整个磁盘 {dev_path},已阻止。")
return False return False
reply = QMessageBox.question(None, "确认删除分区", reply = QMessageBox.question(None, "确认删除分区",
f"确定要删除分区 {dev_path} 吗?\n" f"确定要删除分区 {device_path} 吗?此操作不可逆!",
"此操作不可逆,将导致数据丢失!",
QMessageBox.Yes | QMessageBox.No, QMessageBox.No) QMessageBox.Yes | QMessageBox.No, QMessageBox.No)
if reply == QMessageBox.No: if reply == QMessageBox.No:
logger.info(f"用户取消了删除分区 {dev_path} 的操作。") logger.info(f"用户取消了删除分区 {device_path} 的操作。")
return False return False
# 首先尝试卸载分区(如果已挂载) logger.info(f"尝试删除分区: {device_path}")
try:
# umount 如果设备未挂载会返回非零退出码,所以这里 check_output=False
self.system_manager._run_command(["umount", dev_path], root_privilege=True, check_output=False)
logger.info(f"尝试卸载 {dev_path} (如果已挂载)。")
except Exception as e:
logger.warning(f"卸载 {dev_path} 失败或未挂载: {e}")
# 继续执行,因为即使卸载失败也可能能删除
# 使用 parted 来删除分区 # 1. 尝试卸载分区 (静默处理,如果未挂载则不报错)
# parted 需要父磁盘设备名和分区号 self.unmount_partition(device_path, show_dialog_on_error=False)
# 例如,对于 /dev/sda1需要 /dev/sda 和分区号 1
partition_number_str = ''.join(filter(str.isdigit, device_name)) # 从 sda1 提取 "1" # 2. 从 fstab 中移除条目
if not partition_number_str: self._remove_fstab_entry(device_path)
QMessageBox.critical(None, "错误", f"无法从 {device_name} 解析分区号。")
logger.error(f"无法从 {device_name} 解析分区号。") # 3. 获取父磁盘和分区号
match = re.match(r'(/dev/\w+)(\d+)', device_path)
if not match:
QMessageBox.critical(None, "错误", f"无法解析分区路径 {device_path}")
logger.error(f"无法解析分区路径 {device_path}")
return False return False
parent_disk_name = device_name.rstrip(partition_number_str) # 从 sda1 提取 "sda" disk_path = match.group(1)
parent_disk_path = self._get_device_path(parent_disk_name) partition_number = match.group(2)
try: # 4. 执行 parted 命令删除分区
# parted -s /dev/sda rm 1 success, _, stderr = self._execute_shell_command(
stdout, stderr = self.system_manager._run_command( ["parted", "-s", disk_path, "rm", partition_number],
["parted", "-s", parent_disk_path, "rm", partition_number_str], f"删除分区 {device_path} 失败"
root_privilege=True
) )
logger.info(f"成功删除分区 {dev_path}") if success:
QMessageBox.information(None, "成功", f"成功删除分区 {dev_path}") QMessageBox.information(None, "成功", f"分区 {device_path} 已成功删除。")
# 刷新内核分区表
self._execute_shell_command(["partprobe", disk_path], f"刷新 {disk_path} 分区表失败", root_privilege=True)
return True return True
except Exception as e: else:
logger.error(f"删除分区 {dev_path} 失败: {e}")
QMessageBox.critical(None, "错误", f"删除分区 {dev_path} 失败: {e}")
return False return False
def format_partition(self, device_name, fstype=None): def format_partition(self, device_path, fstype=None):
""" """
格式化指定分区。 格式化指定分区。
此操作不可逆,会丢失数据 :param device_path: 要格式化的设备路径,例如 /dev/sdb1
:param fstype: 文件系统类型,例如 'ext4', 'xfs'。如果为 None则弹出对话框让用户选择。
:return: True 如果格式化成功,否则 False。
""" """
dev_path = self._get_device_path(device_name) if not isinstance(device_path, str):
logger.info(f"尝试格式化分区: {dev_path}") logger.error(f"尝试格式化非字符串设备路径: {device_path} (类型: {type(device_path)})")
QMessageBox.critical(None, "错误", "无效的设备路径。")
# 安全检查: 确保是分区
if not any(char.isdigit() for char in device_name):
QMessageBox.warning(None, "警告", f"{dev_path} 看起来不是一个分区。为安全起见,不执行格式化操作。")
logger.warning(f"尝试格式化整个磁盘 {dev_path},已阻止。")
return False return False
if not fstype: # 1. 尝试卸载分区 (静默处理,如果未挂载则不报错)
# 提示用户选择文件系统类型 self.unmount_partition(device_path, show_dialog_on_error=False)
fstypes = ["ext4", "xfs", "fat32", "ntfs"] # 常用文件系统
fstype, ok = QInputDialog.getItem(None, "格式化分区", # 2. 从 fstab 中移除旧条目 (因为格式化后 UUID 会变)
f"请选择 {dev_path} 的文件系统类型:", self._remove_fstab_entry(device_path)
fstypes, 0, False) # 默认选择 ext4
if not ok or not fstype: if fstype is None:
logger.info("用户取消了格式化操作或未选择文件系统。") # 弹出对话框让用户选择文件系统类型
items = ["ext4", "xfs", "fat32", "ntfs"] # 常用文件系统
selected_fstype, ok = QInputDialog.getItem(None, "选择文件系统",
f"请为 {device_path} 选择文件系统类型:",
items, 0, False)
if not ok or not selected_fstype:
logger.info(f"用户取消了格式化 {device_path} 的操作。")
return False return False
fstype = selected_fstype
reply = QMessageBox.question(None, "确认格式化分区", reply = QMessageBox.question(None, "确认格式化分区",
f"确定要格式化分区 {dev_path}{fstype} 吗?\n" f"确定要格式化分区 {device_path}{fstype} 吗?此操作将擦除所有数据!",
"此操作不可逆,将导致数据丢失!",
QMessageBox.Yes | QMessageBox.No, QMessageBox.No) QMessageBox.Yes | QMessageBox.No, QMessageBox.No)
if reply == QMessageBox.No: if reply == QMessageBox.No:
logger.info(f"用户取消了格式化分区 {dev_path} 的操作。") logger.info(f"用户取消了格式化分区 {device_path} 的操作。")
return False return False
# 首先尝试卸载分区(如果已挂载) logger.info(f"尝试格式化设备 {device_path}{fstype}")
try:
self.system_manager._run_command(["umount", dev_path], root_privilege=True, check_output=False)
logger.info(f"尝试卸载 {dev_path} (如果已挂载)。")
except Exception as e:
logger.warning(f"卸载 {dev_path} 失败或未挂载: {e}")
# 执行 mkfs 命令 format_cmd = []
try:
mkfs_cmd = []
if fstype == "ext4": if fstype == "ext4":
mkfs_cmd = ["mkfs.ext4", "-F", dev_path] # -F 强制执行 format_cmd = ["mkfs.ext4", "-F", device_path]
elif fstype == "xfs": elif fstype == "xfs":
mkfs_cmd = ["mkfs.xfs", "-f", dev_path] # -f 强制执行 format_cmd = ["mkfs.xfs", "-f", device_path]
elif fstype == "fat32": elif fstype == "fat32":
mkfs_cmd = ["mkfs.fat", "-F", "32", dev_path] format_cmd = ["mkfs.vfat", "-F", "32", device_path]
elif fstype == "ntfs": elif fstype == "ntfs":
mkfs_cmd = ["mkfs.ntfs", "-f", dev_path] # -f 强制执行 format_cmd = ["mkfs.ntfs", "-f", device_path]
else: else:
raise ValueError(f"不支持的文件系统类型: {fstype}") QMessageBox.critical(None, "错误", f"不支持的文件系统类型: {fstype}")
logger.error(f"不支持的文件系统类型: {fstype}")
return False
stdout, stderr = self.system_manager._run_command(mkfs_cmd, root_privilege=True) success, _, stderr = self._execute_shell_command(
logger.info(f"成功格式化分区 {dev_path}{fstype}") format_cmd,
QMessageBox.information(None, "成功", f"成功格式化分区 {dev_path}{fstype}") f"格式化设备 {device_path}{fstype} 失败"
)
if success:
QMessageBox.information(None, "成功", f"设备 {device_path} 已成功格式化为 {fstype}")
return True
else:
return False
def mount_partition(self, device_path, mount_point, add_to_fstab=False):
"""
挂载一个分区。
:param device_path: 要挂载的设备路径,例如 /dev/sdb1。
:param mount_point: 挂载点,例如 /mnt/data。
:param add_to_fstab: 是否将挂载信息添加到 /etc/fstab。
:return: True 如果挂载成功,否则 False。
"""
if not isinstance(device_path, str) or not isinstance(mount_point, str):
logger.error(f"尝试挂载时传入无效的设备路径或挂载点。设备: {device_path} (类型: {type(device_path)}), 挂载点: {mount_point} (类型: {type(mount_point)})")
QMessageBox.critical(None, "错误", "无效的设备路径或挂载点。")
return False
logger.info(f"尝试挂载设备 {device_path}{mount_point}")
try:
# 确保挂载点存在
if not os.path.exists(mount_point):
logger.debug(f"创建挂载点目录: {mount_point}")
os.makedirs(mount_point, exist_ok=True)
# 执行挂载命令
success, _, stderr = self._execute_shell_command(
["mount", device_path, mount_point],
f"挂载设备 {device_path}{mount_point} 失败"
)
if success:
QMessageBox.information(None, "成功", f"设备 {device_path} 已成功挂载到 {mount_point}")
if add_to_fstab:
self._add_to_fstab(device_path, mount_point)
return True
else:
return False
except Exception as e:
logger.error(f"挂载设备 {device_path} 时发生错误: {e}")
QMessageBox.critical(None, "错误", f"挂载设备 {device_path} 失败。\n错误详情: {e}")
return False
def unmount_partition(self, device_path, show_dialog_on_error=True):
"""
卸载一个分区。
:param device_path: 要卸载的设备路径,例如 /dev/sdb1 或 /dev/md0。
:param show_dialog_on_error: 是否在卸载失败时显示错误对话框。
:return: True 如果卸载成功,否则 False。
"""
if not isinstance(device_path, str):
logger.error(f"尝试卸载非字符串设备路径: {device_path} (类型: {type(device_path)})")
if show_dialog_on_error:
QMessageBox.critical(None, "错误", f"无效的设备路径: {device_path}")
return False
logger.info(f"尝试卸载设备: {device_path}")
try:
# 尝试从 fstab 中移除条目,如果存在的话
# 注意:这里先移除 fstab 条目,是为了防止卸载失败后 fstab 中仍然有无效条目
# 但如果卸载成功fstab 条目也需要移除。
# 如果卸载失败,且 fstab 移除成功,用户需要手动处理。
# 更好的做法可能是:卸载成功后,再移除 fstab 条目。
# 但为了简化逻辑,这里先移除,并假设卸载会成功。
# 实际上_remove_fstab_entry 应该在卸载成功后才调用,或者在删除分区/格式化时调用。
# 对于单纯的卸载操作,不应该自动移除 fstab 条目,除非用户明确要求。
# 考虑到当前场景是“卸载后刷新”通常用户只是想临时卸载fstab条目不应被删除。
# 所以这里暂时注释掉,或者只在删除分区/格式化时调用。
# self._remove_fstab_entry(device_path)
# 执行卸载命令
success, _, stderr = self._execute_shell_command(
["umount", device_path],
f"卸载设备 {device_path} 失败",
suppress_critical_dialog_on_stderr_match="not mounted" if not show_dialog_on_error else None
)
if success:
QMessageBox.information(None, "成功", f"设备 {device_path} 已成功卸载。")
return success
except Exception as e:
logger.error(f"卸载设备 {device_path} 时发生错误: {e}")
if show_dialog_on_error:
QMessageBox.critical(None, "错误", f"卸载设备 {device_path} 失败。\n错误详情: {e}")
return False
def _add_to_fstab(self, device_path, mount_point):
"""
将设备的挂载信息添加到 /etc/fstab。
使用 UUID 方式添加。
"""
if not isinstance(device_path, str) or not isinstance(mount_point, str):
logger.error(f"尝试添加到 fstab 时传入无效的设备路径或挂载点。设备: {device_path} (类型: {type(device_path)}), 挂载点: {mount_point} (类型: {type(mount_point)})")
QMessageBox.critical(None, "错误", "无效的设备路径或挂载点,无法添加到 fstab。")
return False
logger.info(f"尝试将设备 {device_path} (挂载点: {mount_point}) 添加到 /etc/fstab。")
device_details = self.system_manager.get_device_details_by_path(device_path)
uuid = device_details.get('uuid') if device_details else None
fstype = device_details.get('fstype') if device_details else None
if not uuid or not fstype:
logger.error(f"无法获取设备 {device_path} 的 UUID 或文件系统类型,无法添加到 fstab。UUID: {uuid}, FSTYPE: {fstype}")
QMessageBox.critical(None, "错误", f"无法获取设备 {device_path} 的 UUID 或文件系统类型,无法添加到 /etc/fstab。")
return False
fstab_entry = f"UUID={uuid} {mount_point} {fstype} defaults 0 2"
fstab_path = self._get_fstab_path()
try:
# 检查是否已存在相同的挂载点或 UUID 条目
with open(fstab_path, 'r') as f:
lines = f.readlines()
for line in lines:
if f"UUID={uuid}" in line or mount_point in line.split():
logger.warning(f"fstab 中已存在设备 {device_path} (UUID: {uuid}) 或挂载点 {mount_point} 的条目,跳过添加。")
return True # 已经存在,认为成功
# 添加新条目
with open(fstab_path, 'a') as f:
f.write(fstab_entry + '\n')
logger.info(f"成功将 '{fstab_entry}' 添加到 {fstab_path}")
return True return True
except Exception as e: except Exception as e:
logger.error(f"格式化分区 {dev_path} 失败: {e}") logger.error(f"将设备 {device_path} 添加到 /etc/fstab 失败: {e}")
QMessageBox.critical(None, "错误", f"格式化分区 {dev_path} 失败: {e}") QMessageBox.critical(None, "错误", f"将设备 {device_path} 添加到 /etc/fstab 失败。\n错误详情: {e}")
return False
def _remove_fstab_entry(self, device_path):
"""
从 /etc/fstab 中移除指定设备的条目。
:param device_path: 设备路径,例如 /dev/sdb1 或 /dev/md0。
:return: True 如果成功移除或没有找到条目,否则 False。
"""
if not isinstance(device_path, str):
logger.warning(f"尝试移除 fstab 条目时传入非字符串设备路径: {device_path} (类型: {type(device_path)})")
return False
logger.info(f"尝试从 /etc/fstab 移除与 {device_path} 相关的条目。")
fstab_path = self._get_fstab_path()
try:
device_details = self.system_manager.get_device_details_by_path(device_path)
uuid = device_details.get('uuid') if device_details else None
logger.debug(f"尝试移除 fstab 条目,获取到的设备 {device_path} 的 UUID: {uuid}")
# 如果没有 UUID 且不是 /dev/ 路径,则无法可靠地从 fstab 中移除
if not uuid and not device_path.startswith('/dev/'):
logger.warning(f"无法获取设备 {device_path} 的 UUID也无法识别为 /dev/ 路径,跳过 fstab 移除。")
return False
with open(fstab_path, 'r') as f:
lines = f.readlines()
new_lines = []
removed_count = 0
for line in lines:
# 检查是否是注释或空行
if line.strip().startswith('#') or not line.strip():
new_lines.append(line)
continue
is_match = False
# 尝试匹配 UUID
if uuid:
if f"UUID={uuid}" in line:
is_match = True
# 尝试匹配设备路径 (如果 UUID 不存在或不匹配)
if not is_match and device_path:
# 分割行以避免部分匹配,例如 /dev/sda 匹配 /dev/sda1
parts = line.split()
if len(parts) > 0 and parts[0] == device_path:
is_match = True
if is_match:
logger.info(f"{fstab_path} 移除了条目: {line.strip()}")
removed_count += 1
else:
new_lines.append(line)
if removed_count > 0:
with open(fstab_path, 'w') as f:
f.writelines(new_lines)
logger.info(f"成功从 {fstab_path} 移除了 {removed_count} 个条目。")
return True
else:
logger.info(f"{fstab_path} 中未找到与 {device_path} 相关的条目。")
return True # 没有找到也算成功,因为目标是确保没有该条目
except FileNotFoundError:
logger.warning(f"fstab 文件 {fstab_path} 不存在。")
return True
except Exception as e:
logger.error(f"从 /etc/fstab 移除条目时发生错误: {e}")
QMessageBox.critical(None, "错误", f"从 /etc/fstab 移除条目失败。\n错误详情: {e}")
return False return False

View File

@@ -31,7 +31,7 @@ def setup_logging(text_edit_widget: QTextEdit):
""" """
root_logger = logging.getLogger() root_logger = logging.getLogger()
# 设置日志级别,可以根据需要调整 (DEBUG, INFO, WARNING, ERROR, CRITICAL) # 设置日志级别,可以根据需要调整 (DEBUG, INFO, WARNING, ERROR, CRITICAL)
root_logger.setLevel(logging.INFO) root_logger.setLevel(logging.DEBUG)
# 清除现有的处理器,防止重复输出(如果多次调用此函数) # 清除现有的处理器,防止重复输出(如果多次调用此函数)
for handler in root_logger.handlers[:]: for handler in root_logger.handlers[:]:

309
lvm_operations.py Normal file
View File

@@ -0,0 +1,309 @@
import subprocess
import logging
from PySide6.QtWidgets import QMessageBox
logger = logging.getLogger(__name__)
class LvmOperations:
def __init__(self):
pass # LVM操作不直接依赖SystemInfoManager通过参数传递所需信息
def _execute_shell_command(self, command_list, error_message, root_privilege=True,
suppress_critical_dialog_on_stderr_match=None, input_data=None):
"""
通用地运行一个 shell 命令,并处理错误。
:param command_list: 命令及其参数的列表。
:param error_message: 命令失败时显示给用户的错误消息。
:param root_privilege: 如果为 True则使用 sudo 执行命令。
:param suppress_critical_dialog_on_stderr_match: 如果 stderr 包含此字符串,则不显示关键错误对话框。
:param input_data: 传递给命令stdin的数据 (str)。
:return: (True/False, stdout_str, stderr_str)
"""
if not all(isinstance(arg, str) for arg in command_list):
logger.error(f"命令列表包含非字符串元素: {command_list}")
QMessageBox.critical(None, "错误", f"内部错误:尝试执行的命令包含无效参数。\n命令详情: {command_list}")
return False, "", "内部错误:命令参数类型不正确。"
if root_privilege:
command_list = ["sudo"] + command_list
full_cmd_str = ' '.join(command_list)
logger.debug(f"执行命令: {full_cmd_str}")
try:
result = subprocess.run(
command_list,
capture_output=True,
text=True,
check=True,
encoding='utf-8',
input=input_data
)
logger.info(f"命令成功: {full_cmd_str}")
return True, result.stdout.strip(), result.stderr.strip()
except subprocess.CalledProcessError as e:
stderr_output = e.stderr.strip()
logger.error(f"命令失败: {full_cmd_str}")
logger.error(f"退出码: {e.returncode}")
logger.error(f"标准输出: {e.stdout.strip()}")
logger.error(f"标准错误: {stderr_output}")
if suppress_critical_dialog_on_stderr_match and \
suppress_critical_dialog_on_stderr_match in stderr_output:
logger.info(f"错误信息 '{stderr_output}' 匹配抑制条件,不显示关键错误对话框。")
else:
QMessageBox.critical(None, "错误", f"{error_message}\n错误详情: {stderr_output}")
return False, e.stdout.strip(), stderr_output
except FileNotFoundError:
QMessageBox.critical(None, "错误", f"命令 '{command_list[0]}' 未找到。请确保已安装相关工具。")
logger.error(f"命令 '{command_list[0]}' 未找到。")
return False, "", f"命令 '{command_list[0]}' 未找到。"
except Exception as e:
QMessageBox.critical(None, "错误", f"执行命令时发生未知错误: {e}")
logger.error(f"执行命令 {full_cmd_str} 时发生未知错误: {e}")
return False, "", str(e)
def create_pv(self, device_path):
"""
创建物理卷 (PV)。
:param device_path: 设备的路径,例如 /dev/sdb1。
:return: True 如果成功,否则 False。
"""
if not isinstance(device_path, str):
logger.error(f"尝试创建 PV 时传入非字符串设备路径: {device_path} (类型: {type(device_path)})")
QMessageBox.critical(None, "错误", "无效的设备路径。")
return False
reply = QMessageBox.question(None, "确认创建物理卷",
f"您确定要在设备 {device_path} 上创建物理卷吗?此操作将覆盖设备上的数据。",
QMessageBox.Yes | QMessageBox.No, QMessageBox.No)
if reply == QMessageBox.No:
logger.info(f"用户取消了在 {device_path} 上创建物理卷的操作。")
return False
logger.info(f"尝试在 {device_path} 上创建物理卷。")
success, _, stderr = self._execute_shell_command(
["pvcreate", "-y", device_path], # -y 自动确认
f"{device_path} 上创建物理卷失败"
)
if success:
QMessageBox.information(None, "成功", f"物理卷已在 {device_path} 上成功创建。")
return True
else:
return False
def delete_pv(self, device_path):
"""
删除物理卷 (PV)。
:param device_path: 物理卷的路径,例如 /dev/sdb1。
:return: True 如果成功,否则 False。
"""
if not isinstance(device_path, str):
logger.error(f"尝试删除 PV 时传入非字符串设备路径: {device_path} (类型: {type(device_path)})")
QMessageBox.critical(None, "错误", "无效的设备路径。")
return False
reply = QMessageBox.question(None, "确认删除物理卷",
f"您确定要删除物理卷 {device_path} 吗?如果该物理卷属于某个卷组,请先将其从卷组中移除。",
QMessageBox.Yes | QMessageBox.No, QMessageBox.No)
if reply == QMessageBox.No:
logger.info(f"用户取消了删除物理卷 {device_path} 的操作。")
return False
logger.info(f"尝试删除物理卷: {device_path}")
success, _, stderr = self._execute_shell_command(
["pvremove", "-y", device_path],
f"删除物理卷 {device_path} 失败"
)
if success:
QMessageBox.information(None, "成功", f"物理卷 {device_path} 已成功删除。")
return True
else:
return False
def create_vg(self, vg_name, pv_paths):
"""
创建卷组 (VG)。
:param vg_name: 卷组的名称。
:param pv_paths: 组成卷组的物理卷路径列表。
:return: True 如果成功,否则 False。
"""
if not isinstance(vg_name, str) or not isinstance(pv_paths, list) or not all(isinstance(p, str) for p in pv_paths):
logger.error(f"尝试创建 VG 时传入无效参数。VG名称: {vg_name}, PV路径: {pv_paths}")
QMessageBox.critical(None, "错误", "无效的卷组名称或物理卷路径。")
return False
reply = QMessageBox.question(None, "确认创建卷组",
f"您确定要使用物理卷 {', '.join(pv_paths)} 创建卷组 {vg_name} 吗?",
QMessageBox.Yes | QMessageBox.No, QMessageBox.No)
if reply == QMessageBox.No:
logger.info(f"用户取消了创建卷组 {vg_name} 的操作。")
return False
logger.info(f"尝试使用 {', '.join(pv_paths)} 创建卷组 {vg_name}")
command = ["vgcreate", vg_name] + pv_paths
success, _, stderr = self._execute_shell_command(
command,
f"创建卷组 {vg_name} 失败"
)
if success:
QMessageBox.information(None, "成功", f"卷组 {vg_name} 已成功创建。")
return True
else:
return False
def delete_vg(self, vg_name):
"""
删除卷组 (VG)。
:param vg_name: 卷组的名称。
:return: True 如果成功,否则 False。
"""
if not isinstance(vg_name, str):
logger.error(f"尝试删除 VG 时传入非字符串 VG 名称: {vg_name} (类型: {type(vg_name)})")
QMessageBox.critical(None, "错误", "无效的卷组名称。")
return False
reply = QMessageBox.question(None, "确认删除卷组",
f"您确定要删除卷组 {vg_name} 吗?此操作将删除所有属于该卷组的逻辑卷!",
QMessageBox.Yes | QMessageBox.No, QMessageBox.No)
if reply == QMessageBox.No:
logger.info(f"用户取消了删除卷组 {vg_name} 的操作。")
return False
logger.info(f"尝试删除卷组: {vg_name}")
# -y 自动确认, -f 强制删除所有逻辑卷
success, _, stderr = self._execute_shell_command(
["vgremove", "-y", "-f", vg_name],
f"删除卷组 {vg_name} 失败"
)
if success:
QMessageBox.information(None, "成功", f"卷组 {vg_name} 已成功删除。")
return True
else:
return False
def create_lv(self, lv_name, vg_name, size_gb, use_max_space=False):
"""
创建逻辑卷 (LV)。
:param lv_name: 逻辑卷的名称。
:param vg_name: 所属卷组的名称。
:param size_gb: 逻辑卷的大小GB。当 use_max_space 为 True 时,此参数被忽略。
:param use_max_space: 是否使用卷组的最大可用空间。
:return: True 如果成功,否则 False。
"""
if not isinstance(lv_name, str) or not isinstance(vg_name, str):
logger.error(f"尝试创建 LV 时传入无效参数。LV名称: {lv_name}, VG名称: {vg_name}")
QMessageBox.critical(None, "错误", "无效的逻辑卷名称或卷组名称。")
return False
if not use_max_space and (not isinstance(size_gb, (int, float)) or size_gb <= 0):
logger.error(f"尝试创建 LV 时传入无效大小。大小: {size_gb}")
QMessageBox.critical(None, "错误", "逻辑卷大小必须大于0。")
return False
# Confirmation message
confirm_message = f"您确定要在卷组 {vg_name} 中创建逻辑卷 {lv_name} 吗?"
if use_max_space:
confirm_message += "使用卷组所有可用空间。"
else:
confirm_message += f"大小为 {size_gb}GB。"
reply = QMessageBox.question(None, "确认创建逻辑卷",
confirm_message,
QMessageBox.Yes | QMessageBox.No, QMessageBox.No)
if reply == QMessageBox.No:
logger.info(f"用户取消了创建逻辑卷 {lv_name} 的操作。")
return False
if use_max_space:
create_cmd = ["lvcreate", "-l", "100%FREE", "-n", lv_name, vg_name]
logger.info(f"尝试在卷组 {vg_name} 中使用最大可用空间创建逻辑卷 {lv_name}")
else:
create_cmd = ["lvcreate", "-L", f"{size_gb}G", "-n", lv_name, vg_name]
logger.info(f"尝试在卷组 {vg_name} 中创建 {size_gb}GB 的逻辑卷 {lv_name}")
success, _, stderr = self._execute_shell_command(
create_cmd,
f"创建逻辑卷 {lv_name} 失败"
)
if success:
QMessageBox.information(None, "成功", f"逻辑卷 {lv_name} 已在卷组 {vg_name} 中成功创建。")
return True
else:
return False
def delete_lv(self, lv_name, vg_name):
"""
删除逻辑卷 (LV)。
:param lv_name: 逻辑卷的名称。
:param vg_name: 所属卷组的名称。
:return: True 如果成功,否则 False。
"""
if not isinstance(lv_name, str) or not isinstance(vg_name, str):
logger.error(f"尝试删除 LV 时传入非字符串 LV 名称或 VG 名称。LV名称: {lv_name}, VG名称: {vg_name}")
QMessageBox.critical(None, "错误", "无效的逻辑卷名称或卷组名称。")
return False
reply = QMessageBox.question(None, "确认删除逻辑卷",
f"您确定要删除逻辑卷 {vg_name}/{lv_name} 吗?此操作将擦除所有数据!",
QMessageBox.Yes | QMessageBox.No, QMessageBox.No)
if reply == QMessageBox.No:
logger.info(f"用户取消了删除逻辑卷 {vg_name}/{lv_name} 的操作。")
return False
logger.info(f"尝试删除逻辑卷: {vg_name}/{lv_name}")
# -y 自动确认
success, _, stderr = self._execute_shell_command(
["lvremove", "-y", f"{vg_name}/{lv_name}"],
f"删除逻辑卷 {vg_name}/{lv_name} 失败"
)
if success:
QMessageBox.information(None, "成功", f"逻辑卷 {vg_name}/{lv_name} 已成功删除。")
return True
else:
return False
def activate_lv(self, lv_name, vg_name):
"""
激活逻辑卷。
:param lv_name: 逻辑卷的名称。
:param vg_name: 所属卷组的名称。
:return: True 如果成功,否则 False。
"""
if not isinstance(lv_name, str) or not isinstance(vg_name, str):
logger.error(f"尝试激活 LV 时传入非字符串 LV 名称或 VG 名称。LV名称: {lv_name}, VG名称: {vg_name}")
QMessageBox.critical(None, "错误", "无效的逻辑卷名称或卷组名称。")
return False
logger.info(f"尝试激活逻辑卷: {vg_name}/{lv_name}")
success, _, stderr = self._execute_shell_command(
["lvchange", "-ay", f"{vg_name}/{lv_name}"],
f"激活逻辑卷 {vg_name}/{lv_name} 失败"
)
if success:
QMessageBox.information(None, "成功", f"逻辑卷 {vg_name}/{lv_name} 已成功激活。")
return True
else:
return False
def deactivate_lv(self, lv_name, vg_name):
"""
停用逻辑卷。
:param lv_name: 逻辑卷的名称。
:param vg_name: 所属卷组的名称。
:return: True 如果成功,否则 False。
"""
if not isinstance(lv_name, str) or not isinstance(vg_name, str):
logger.error(f"尝试停用 LV 时传入非字符串 LV 名称或 VG 名称。LV名称: {lv_name}, VG名称: {vg_name}")
QMessageBox.critical(None, "错误", "无效的逻辑卷名称或卷组名称。")
return False
logger.info(f"尝试停用逻辑卷: {vg_name}/{lv_name}")
success, _, stderr = self._execute_shell_command(
["lvchange", "-an", f"{vg_name}/{lv_name}"],
f"停用逻辑卷 {vg_name}/{lv_name} 失败"
)
if success:
QMessageBox.information(None, "成功", f"逻辑卷 {vg_name}/{lv_name} 已成功停用。")
return True
else:
return False

View File

@@ -1,15 +1,29 @@
# mainwindow.py # mainwindow.py
import sys import sys
import logging import logging
import re
import os # 导入 os 模块
from PySide6.QtWidgets import (QApplication, QMainWindow, QTreeWidgetItem, from PySide6.QtWidgets import (QApplication, QMainWindow, QTreeWidgetItem,
QMessageBox, QHeaderView, QMenu, QInputDialog) QMessageBox, QHeaderView, QMenu, QInputDialog, QDialog)
from PySide6.QtCore import Qt, QPoint from PySide6.QtCore import Qt, QPoint
# 导入自动生成的 UI 文件
from ui_form import Ui_MainWindow from ui_form import Ui_MainWindow
# 导入我们自己编写的系统信息管理模块
from system_info import SystemInfoManager from system_info import SystemInfoManager
# 导入日志配置
from logger_config import setup_logging, logger from logger_config import setup_logging, logger
# 导入磁盘操作模块
from disk_operations import DiskOperations from disk_operations import DiskOperations
# 导入 RAID 操作模块
from raid_operations import RaidOperations
# 导入 LVM 操作模块
from lvm_operations import LvmOperations
# 导入自定义对话框
from dialogs import (CreatePartitionDialog, MountDialog, CreateRaidDialog,
CreatePvDialog, CreateVgDialog, CreateLvDialog)
class MainWindow(QMainWindow): class MainWindow(QMainWindow):
def __init__(self, parent=None): def __init__(self, parent=None):
@@ -20,17 +34,30 @@ class MainWindow(QMainWindow):
setup_logging(self.ui.logOutputTextEdit) setup_logging(self.ui.logOutputTextEdit)
logger.info("应用程序启动。") logger.info("应用程序启动。")
# 初始化管理器和操作类
self.system_manager = SystemInfoManager() self.system_manager = SystemInfoManager()
self.disk_ops = DiskOperations() self.disk_ops = DiskOperations()
self.raid_ops = RaidOperations()
self.lvm_ops = LvmOperations()
# 连接刷新按钮的信号到槽函数
if hasattr(self.ui, 'refreshButton'): if hasattr(self.ui, 'refreshButton'):
self.ui.refreshButton.clicked.connect(self.refresh_all_info) self.ui.refreshButton.clicked.connect(self.refresh_all_info)
else: else:
logger.warning("Warning: refreshButton not found in UI. Please add it in form.ui and regenerate ui_form.py.") logger.warning("Warning: refreshButton not found in UI. Please add it in form.ui and regenerate ui_form.py.")
# 启用 treeWidget 的自定义上下文菜单
self.ui.treeWidget_block_devices.setContextMenuPolicy(Qt.CustomContextMenu) self.ui.treeWidget_block_devices.setContextMenuPolicy(Qt.CustomContextMenu)
self.ui.treeWidget_block_devices.customContextMenuRequested.connect(self.show_block_device_context_menu) self.ui.treeWidget_block_devices.customContextMenuRequested.connect(self.show_block_device_context_menu)
self.ui.treeWidget_raid.setContextMenuPolicy(Qt.CustomContextMenu)
self.ui.treeWidget_raid.customContextMenuRequested.connect(self.show_raid_context_menu)
self.ui.treeWidget_lvm.setContextMenuPolicy(Qt.CustomContextMenu)
self.ui.treeWidget_lvm.customContextMenuRequested.connect(self.show_lvm_context_menu)
# 初始化时刷新所有数据
self.refresh_all_info() self.refresh_all_info()
logger.info("所有设备信息已初始化加载。") logger.info("所有设备信息已初始化加载。")
@@ -44,26 +71,15 @@ class MainWindow(QMainWindow):
self.refresh_lvm_info() self.refresh_lvm_info()
logger.info("所有设备信息刷新完成。") logger.info("所有设备信息刷新完成。")
# --- 块设备概览 Tab ---
def refresh_block_devices_info(self): def refresh_block_devices_info(self):
"""
刷新块设备信息并显示在 QTreeWidget 中。
"""
self.ui.treeWidget_block_devices.clear() self.ui.treeWidget_block_devices.clear()
columns = [ columns = [
("设备名", 'name'), ("设备名", 'name'), ("类型", 'type'), ("大小", 'size'), ("挂载点", 'mountpoint'),
("类型", 'type'), ("文件系统", 'fstype'), ("只读", 'ro'), ("UUID", 'uuid'), ("PARTUUID", 'partuuid'),
("大小", 'size'), ("厂商", 'vendor'), ("型号", 'model'), ("序列号", 'serial'),
("挂载点", 'mountpoint'), ("主次号", 'maj:min'), ("父设备名", 'pkname'),
("文件系统", 'fstype'),
("只读", 'ro'),
("UUID", 'uuid'),
("PARTUUID", 'partuuid'),
("厂商", 'vendor'),
("型号", 'model'),
("序列号", 'serial'),
("主次号", 'maj:min'),
("父设备名", 'pkname'),
] ]
headers = [col[0] for col in columns] headers = [col[0] for col in columns]
@@ -89,10 +105,6 @@ class MainWindow(QMainWindow):
logger.error(f"刷新块设备信息失败: {e}") logger.error(f"刷新块设备信息失败: {e}")
def _add_device_to_tree(self, parent_item, dev_data): def _add_device_to_tree(self, parent_item, dev_data):
"""
辅助函数,将单个设备及其子设备添加到 QTreeWidget。
parent_item 可以是 QTreeWidget 本身,也可以是另一个 QTreeWidgetItem。
"""
item = QTreeWidgetItem(parent_item) item = QTreeWidgetItem(parent_item)
for i, key in enumerate(self.field_keys): for i, key in enumerate(self.field_keys):
value = dev_data.get(key) value = dev_data.get(key)
@@ -110,16 +122,170 @@ class MainWindow(QMainWindow):
self._add_device_to_tree(item, child) self._add_device_to_tree(item, child)
item.setExpanded(True) item.setExpanded(True)
def show_block_device_context_menu(self, pos: QPoint):
item = self.ui.treeWidget_block_devices.itemAt(pos)
menu = QMenu(self)
# 菜单项:创建 RAID 阵列,创建 PV
create_menu = QMenu("创建...", self)
create_raid_action = create_menu.addAction("创建 RAID 阵列...")
create_raid_action.triggered.connect(self._handle_create_raid_array)
create_pv_action = create_menu.addAction("创建物理卷 (PV)...")
create_pv_action.triggered.connect(self._handle_create_pv)
menu.addMenu(create_menu)
menu.addSeparator()
if item:
dev_data = item.data(0, Qt.UserRole)
if not dev_data:
logger.warning(f"无法获取设备 {item.text(0)} 的详细数据。")
return
device_name = dev_data.get('name')
device_type = dev_data.get('type')
mount_point = dev_data.get('mountpoint')
device_path = dev_data.get('path') # 使用lsblk提供的完整路径
if not device_path: # 如果path字段缺失则尝试从name构造
device_path = f"/dev/{device_name}"
# 针对磁盘 (disk) 的操作
if device_type == 'disk':
create_partition_action = menu.addAction(f"创建分区 {device_path}...")
create_partition_action.triggered.connect(lambda: self._handle_create_partition(device_path, dev_data.get('size'))) # 传递原始大小字符串
menu.addSeparator()
# 挂载/卸载操作 (针对分区 'part' 和 RAID/LVM 逻辑卷,但这里只处理 'part')
if device_type == 'part':
if not mount_point or mount_point == '' or mount_point == 'N/A':
mount_action = menu.addAction(f"挂载 {device_path}...")
mount_action.triggered.connect(lambda: self._handle_mount(device_path))
elif mount_point != '[SWAP]':
unmount_action = menu.addAction(f"卸载 {device_path}")
# FIX: 更改 lambda 表达式,避免被 triggered 信号的参数覆盖
unmount_action.triggered.connect(lambda: self._unmount_and_refresh(device_path))
menu.addSeparator()
# 删除分区和格式化操作 (针对分区 'part')
delete_action = menu.addAction(f"删除分区 {device_path}")
delete_action.triggered.connect(lambda: self._handle_delete_partition(device_path))
format_action = menu.addAction(f"格式化分区 {device_path}...")
format_action.triggered.connect(lambda: self._handle_format_partition(device_path))
if menu.actions():
menu.exec(self.ui.treeWidget_block_devices.mapToGlobal(pos))
else:
logger.info("右键点击了空白区域或设备没有可用的操作。")
def _handle_create_partition(self, disk_path, total_size_str): # 接收原始大小字符串
# 1. 解析磁盘总大小 (MiB)
total_disk_mib = 0.0
logger.debug(f"尝试为磁盘 {disk_path} 创建分区。原始大小字符串: '{total_size_str}'")
if total_size_str:
match = re.match(r'(\d+(\.\d+)?)\s*([KMGT]?B?)', total_size_str, re.IGNORECASE)
if match:
value = float(match.group(1))
unit = match.group(3).upper() if match.group(3) else ''
if unit == 'KB' or unit == 'K': total_disk_mib = value / 1024
elif unit == 'MB' or unit == 'M': total_disk_mib = value
elif unit == 'GB' or unit == 'G': total_disk_mib = value * 1024
elif unit == 'TB' or unit == 'T': total_disk_mib = value * 1024 * 1024
elif unit == 'B':
total_disk_mib = value / (1024 * 1024)
else:
logger.warning(f"无法识别磁盘 {disk_path} 的大小单位: '{unit}' (原始: '{total_size_str}')")
total_disk_mib = 0.0
logger.debug(f"解析后的磁盘总大小 (MiB): {total_disk_mib}")
else:
logger.warning(f"无法解析磁盘 {disk_path} 的大小字符串 '{total_size_str}'。正则表达式不匹配。")
total_disk_mib = 0.0
else:
logger.warning(f"获取磁盘 {disk_path} 的大小字符串为空或None。")
total_disk_mib = 0.0
if total_disk_mib <= 0.0:
QMessageBox.critical(self, "错误", f"无法获取磁盘 {disk_path} 的有效总大小。")
return
# 2. 获取下一个分区的起始位置 (MiB)
start_position_mib = self.disk_ops.get_disk_next_partition_start_mib(disk_path)
if start_position_mib is None: # 如果获取起始位置失败
QMessageBox.critical(self, "错误", f"无法确定磁盘 {disk_path} 的分区起始位置。")
return
# 3. 计算最大可用空间 (MiB)
max_available_mib = total_disk_mib - start_position_mib
if max_available_mib < 0: # 安全检查,理论上不应该发生
max_available_mib = 0.0
# 确保 max_available_mib 至少为 1 MiB以避免 spinbox 最小值大于最大值的问题
# 0.1 GB = 102.4 MiB, so 1 MiB is a safe minimum for fdisk/parted
if max_available_mib < 1.0:
max_available_mib = 1.0 # 最小可分配空间 (1 MiB)
dialog = CreatePartitionDialog(self, disk_path, total_disk_mib, max_available_mib) # 传递 total_disk_mib 和 max_available_mib
if dialog.exec() == QDialog.Accepted:
info = dialog.get_partition_info()
if info: # Check if info is not None (dialog might have returned None on validation error)
# 调用 disk_ops.create_partition传递 total_disk_mib 和 use_max_space 标志
new_partition_path = self.disk_ops.create_partition(
info['disk_path'],
info['partition_table_type'],
info['size_gb'],
info['total_disk_mib'], # 传递磁盘总大小 (MiB)
info['use_max_space'] # 传递是否使用最大空间的标志
)
if new_partition_path:
self.refresh_all_info() # 刷新以显示新创建的分区
# 询问用户是否要格式化新创建的分区
reply = QMessageBox.question(None, "格式化新分区",
f"分区 {new_partition_path} 已成功创建。\n"
"您现在想格式化这个分区吗?",
QMessageBox.Yes | QMessageBox.No, QMessageBox.Yes)
if reply == QMessageBox.Yes:
# 调用格式化函数fstype=None 会弹出选择框
if self.disk_ops.format_partition(new_partition_path, fstype=None):
self.refresh_all_info() # 格式化成功后再次刷新
# else: create_partition already shows an error message
# else: dialog was cancelled, no action needed
def _handle_mount(self, device_path):
dialog = MountDialog(self, device_path)
if dialog.exec() == QDialog.Accepted:
info = dialog.get_mount_info()
if info: # 确保 info 不是 None (用户可能在 MountDialog 中取消了操作)
if self.disk_ops.mount_partition(device_path, info['mount_point'], info['add_to_fstab']): # <--- 传递 add_to_fstab
self.refresh_all_info()
# 新增辅助方法,用于卸载并刷新
def _unmount_and_refresh(self, device_path):
"""Helper method to unmount a device and then refresh all info."""
if self.disk_ops.unmount_partition(device_path, show_dialog_on_error=True):
self.refresh_all_info()
def _handle_delete_partition(self, device_path):
# delete_partition 内部会调用 unmount_partition 和 _remove_fstab_entry
if self.disk_ops.delete_partition(device_path):
self.refresh_all_info()
def _handle_format_partition(self, device_path):
# format_partition 内部会调用 unmount_partition 和 _remove_fstab_entry
if self.disk_ops.format_partition(device_path):
self.refresh_all_info()
# --- RAID 管理 Tab ---
def refresh_raid_info(self): def refresh_raid_info(self):
"""
刷新RAID阵列信息并显示在 QTreeWidget 中。
"""
self.ui.treeWidget_raid.clear() self.ui.treeWidget_raid.clear()
# 定义RAID显示列头
raid_headers = [ raid_headers = [
"阵列设备", "级别", "状态", "大小", "活动设备", "失败设备", "备用设备", "阵列设备", "级别", "状态", "大小", "活动设备", "失败设备", "备用设备",
"总设备数", "UUID", "名称", "Chunk Size" "总设备数", "UUID", "名称", "Chunk Size", "挂载点" # 添加挂载点列
] ]
self.ui.treeWidget_raid.setColumnCount(len(raid_headers)) self.ui.treeWidget_raid.setColumnCount(len(raid_headers))
self.ui.treeWidget_raid.setHeaderLabels(raid_headers) self.ui.treeWidget_raid.setHeaderLabels(raid_headers)
@@ -137,7 +303,15 @@ class MainWindow(QMainWindow):
for array in raid_arrays: for array in raid_arrays:
array_item = QTreeWidgetItem(self.ui.treeWidget_raid) array_item = QTreeWidgetItem(self.ui.treeWidget_raid)
array_item.setText(0, array.get('device', 'N/A')) array_path = array.get('device', 'N/A')
# 确保 array_path 是一个有效的路径,否则上下文菜单会失效
if not array_path or array_path == 'N/A':
logger.warning(f"RAID阵列 '{array.get('name', '未知')}' 的设备路径无效,跳过。")
continue
current_mount_point = self.system_manager.get_mountpoint_for_device(array_path)
array_item.setText(0, array_path)
array_item.setText(1, array.get('level', 'N/A')) array_item.setText(1, array.get('level', 'N/A'))
array_item.setText(2, array.get('state', 'N/A')) array_item.setText(2, array.get('state', 'N/A'))
array_item.setText(3, array.get('array_size', 'N/A')) array_item.setText(3, array.get('array_size', 'N/A'))
@@ -148,18 +322,24 @@ class MainWindow(QMainWindow):
array_item.setText(8, array.get('uuid', 'N/A')) array_item.setText(8, array.get('uuid', 'N/A'))
array_item.setText(9, array.get('name', 'N/A')) array_item.setText(9, array.get('name', 'N/A'))
array_item.setText(10, array.get('chunk_size', 'N/A')) array_item.setText(10, array.get('chunk_size', 'N/A'))
array_item.setText(11, current_mount_point if current_mount_point else "") # 显示挂载点
array_item.setExpanded(True) array_item.setExpanded(True)
# 存储原始数据,用于上下文菜单
# 确保 array_path 字段在存储的数据中是正确的
array_data_for_context = array.copy()
array_data_for_context['device'] = array_path
array_item.setData(0, Qt.UserRole, array_data_for_context)
# 添加成员设备作为子节点 # 添加成员设备作为子节点
for member in array.get('member_devices', []): for member in array.get('member_devices', []):
member_item = QTreeWidgetItem(array_item) member_item = QTreeWidgetItem(array_item)
member_item.setText(0, f" {member.get('device_path', 'N/A')}") # 缩进显示 member_item.setText(0, f" {member.get('device_path', 'N/A')}")
member_item.setText(1, f"成员") member_item.setText(1, f"成员: {member.get('raid_device', 'N/A')}")
member_item.setText(2, member.get('state', 'N/A')) member_item.setText(2, member.get('state', 'N/A'))
# 其他列留空或填充N/A因为是成员设备的特有信息 member_item.setText(3, f"Major: {member.get('major', 'N/A')}, Minor: {member.get('minor', 'N/A')}")
member_item.setText(3, f"RaidDevice: {member.get('raid_device', 'N/A')}") # 成员设备不存储UserRole数据因为操作通常针对整个阵列
# 自动调整列宽
for i in range(len(raid_headers)): for i in range(len(raid_headers)):
self.ui.treeWidget_raid.resizeColumnToContents(i) self.ui.treeWidget_raid.resizeColumnToContents(i)
logger.info("RAID阵列信息刷新成功。") logger.info("RAID阵列信息刷新成功。")
@@ -168,14 +348,92 @@ class MainWindow(QMainWindow):
QMessageBox.critical(self, "错误", f"刷新RAID阵列信息失败: {e}") QMessageBox.critical(self, "错误", f"刷新RAID阵列信息失败: {e}")
logger.error(f"刷新RAID阵列信息失败: {e}") logger.error(f"刷新RAID阵列信息失败: {e}")
def show_raid_context_menu(self, pos: QPoint):
item = self.ui.treeWidget_raid.itemAt(pos)
menu = QMenu(self)
# 始终提供创建 RAID 阵列的选项
create_raid_action = menu.addAction("创建 RAID 阵列...")
create_raid_action.triggered.connect(self._handle_create_raid_array)
menu.addSeparator()
if item and item.parent() is None: # 只有点击的是顶层RAID阵列项
array_data = item.data(0, Qt.UserRole)
if not array_data:
logger.warning(f"无法获取 RAID 阵列 {item.text(0)} 的详细数据。")
return
array_path = array_data.get('device') # e.g., /dev/md126
member_devices = [m.get('device_path') for m in array_data.get('member_devices', [])]
if not array_path or array_path == 'N/A':
logger.warning(f"RAID阵列 '{array_data.get('name', '未知')}' 的设备路径无效,无法显示操作。")
return # 如果路径无效,则不显示任何操作
# Check if RAID array is mounted
current_mount_point = self.system_manager.get_mountpoint_for_device(array_path)
if current_mount_point and current_mount_point != '[SWAP]' and current_mount_point != '':
unmount_action = menu.addAction(f"卸载 {array_path} ({current_mount_point})")
# FIX: 更改 lambda 表达式,避免被 triggered 信号的参数覆盖
unmount_action.triggered.connect(lambda: self._unmount_and_refresh(array_path))
else:
mount_action = menu.addAction(f"挂载 {array_path}...")
mount_action.triggered.connect(lambda: self._handle_mount(array_path))
menu.addSeparator()
stop_action = menu.addAction(f"停止阵列 {array_path}")
stop_action.triggered.connect(lambda: self._handle_stop_raid_array(array_path))
delete_action = menu.addAction(f"删除阵列 {array_path}")
delete_action.triggered.connect(lambda: self._handle_delete_raid_array(array_path, member_devices))
format_action = menu.addAction(f"格式化阵列 {array_path}...")
# RAID 阵列的格式化也直接调用 disk_ops.format_partition
format_action.triggered.connect(lambda: self._handle_format_raid_array(array_path))
if menu.actions():
menu.exec(self.ui.treeWidget_raid.mapToGlobal(pos))
else:
logger.info("右键点击了空白区域或没有可用的RAID操作。")
def _handle_create_raid_array(self):
# 获取所有可用于创建RAID阵列的设备
# get_unallocated_partitions 现在应该返回所有合适的磁盘和分区
available_devices = self.system_manager.get_unallocated_partitions()
if not available_devices:
QMessageBox.warning(self, "警告", "没有可用于创建 RAID 阵列的设备。请确保有未挂载、未被LVM或RAID使用的磁盘或分区。")
return
dialog = CreateRaidDialog(self, available_devices)
if dialog.exec() == QDialog.Accepted:
info = dialog.get_raid_info()
if info: # Check if info is not None
if self.raid_ops.create_raid_array(info['devices'], info['level'], info['chunk_size']):
self.refresh_all_info()
def _handle_stop_raid_array(self, array_path):
if self.raid_ops.stop_raid_array(array_path):
self.refresh_all_info()
def _handle_delete_raid_array(self, array_path, member_devices):
if self.raid_ops.delete_raid_array(array_path, member_devices):
self.refresh_all_info()
def _handle_format_raid_array(self, array_path):
# 格式化RAID阵列与格式化分区类似可以复用DiskOperations中的逻辑
# 或者在RaidOperations中实现一个独立的format方法
# 这里我们直接调用DiskOperations的format_partition方法
# format_partition 内部会调用 unmount_partition(..., show_dialog_on_error=False) 和 _remove_fstab_entry
if self.disk_ops.format_partition(array_path):
self.refresh_all_info()
# --- LVM 管理 Tab ---
def refresh_lvm_info(self): def refresh_lvm_info(self):
"""
刷新LVM信息PVs, VGs, LVs并显示在 QTreeWidget 中。
"""
self.ui.treeWidget_lvm.clear() self.ui.treeWidget_lvm.clear()
# 定义LVM显示列头 lvm_headers = ["名称", "大小", "属性", "UUID", "关联", "空闲/已用", "路径/格式", "挂载点"] # 添加挂载点列
lvm_headers = ["名称", "大小", "属性", "UUID", "关联", "空闲/已用", "路径"]
self.ui.treeWidget_lvm.setColumnCount(len(lvm_headers)) self.ui.treeWidget_lvm.setColumnCount(len(lvm_headers))
self.ui.treeWidget_lvm.setHeaderLabels(lvm_headers) self.ui.treeWidget_lvm.setHeaderLabels(lvm_headers)
@@ -195,35 +453,55 @@ class MainWindow(QMainWindow):
pv_root_item = QTreeWidgetItem(self.ui.treeWidget_lvm) pv_root_item = QTreeWidgetItem(self.ui.treeWidget_lvm)
pv_root_item.setText(0, "物理卷 (PVs)") pv_root_item.setText(0, "物理卷 (PVs)")
pv_root_item.setExpanded(True) pv_root_item.setExpanded(True)
pv_root_item.setData(0, Qt.UserRole, {'type': 'pv_root'}) # 标识根节点
if lvm_data.get('pvs'): if lvm_data.get('pvs'):
for pv in lvm_data['pvs']: for pv in lvm_data['pvs']:
pv_item = QTreeWidgetItem(pv_root_item) pv_item = QTreeWidgetItem(pv_root_item)
pv_item.setText(0, pv.get('pv_name', 'N/A')) pv_name = pv.get('pv_name', 'N/A')
if pv_name.startswith('/dev/'): # Ensure it's a full path
pv_path = pv_name
else:
# Attempt to construct if it's just a name, though pv_name from pvs usually is full path
# Fallback, may not be accurate if system_manager.get_lvm_info() isn't consistent
pv_path = f"/dev/{pv_name}" if pv_name != 'N/A' else 'N/A'
pv_item.setText(0, pv_name)
pv_item.setText(1, pv.get('pv_size', 'N/A')) pv_item.setText(1, pv.get('pv_size', 'N/A'))
pv_item.setText(2, pv.get('pv_attr', 'N/A')) pv_item.setText(2, pv.get('pv_attr', 'N/A'))
pv_item.setText(3, pv.get('pv_uuid', 'N/A')) pv_item.setText(3, pv.get('pv_uuid', 'N/A'))
pv_item.setText(4, f"VG: {pv.get('vg_name', 'N/A')}") pv_item.setText(4, f"VG: {pv.get('vg_name', 'N/A')}")
pv_item.setText(5, f"空闲: {pv.get('pv_free', 'N/A')}") pv_item.setText(5, f"空闲: {pv.get('pv_free', 'N/A')}")
pv_item.setText(6, pv.get('pv_fmt', 'N/A')) # PV格式 pv_item.setText(6, pv.get('pv_fmt', 'N/A'))
pv_item.setText(7, "") # PV没有直接挂载点
# 存储原始数据确保pv_name是正确的设备路径
pv_data_for_context = pv.copy()
pv_data_for_context['pv_name'] = pv_path # Store the full path for context menu
pv_item.setData(0, Qt.UserRole, {'type': 'pv', 'data': pv_data_for_context}) # 存储原始数据
else: else:
item = QTreeWidgetItem(pv_root_item) item = QTreeWidgetItem(pv_root_item)
item.setText(0, "未找到物理卷。") item.setText(0, "未找到物理卷。")
# 卷组 (VGs) # 卷组 (VGs)
vg_root_item = QTreeWidgetItem(self.ui.treeWidget_lvm) vg_root_item = QTreeWidgetItem(self.ui.treeWidget_lvm)
vg_root_item.setText(0, "卷组 (VGs)") vg_root_item.setText(0, "卷组 (VGs)")
vg_root_item.setExpanded(True) vg_root_item.setExpanded(True)
vg_root_item.setData(0, Qt.UserRole, {'type': 'vg_root'})
if lvm_data.get('vgs'): if lvm_data.get('vgs'):
for vg in lvm_data['vgs']: for vg in lvm_data['vgs']:
vg_item = QTreeWidgetItem(vg_root_item) vg_item = QTreeWidgetItem(vg_root_item)
vg_item.setText(0, vg.get('vg_name', 'N/A')) vg_name = vg.get('vg_name', 'N/A')
vg_item.setText(0, vg_name)
vg_item.setText(1, vg.get('vg_size', 'N/A')) vg_item.setText(1, vg.get('vg_size', 'N/A'))
vg_item.setText(2, vg.get('vg_attr', 'N/A')) vg_item.setText(2, vg.get('vg_attr', 'N/A'))
vg_item.setText(3, vg.get('vg_uuid', 'N/A')) vg_item.setText(3, vg.get('vg_uuid', 'N/A'))
vg_item.setText(4, f"PVs: {vg.get('pv_count', 'N/A')}, LVs: {vg.get('lv_count', 'N/A')}") vg_item.setText(4, f"PVs: {vg.get('pv_count', 'N/A')}, LVs: {vg.get('lv_count', 'N/A')}")
vg_item.setText(5, f"空闲: {vg.get('vg_free', 'N/A')}, 已分配: {vg.get('vg_alloc_percent', 'N/A')}%") vg_item.setText(5, f"空闲: {vg.get('vg_free', 'N/A')}, 已分配: {vg.get('vg_alloc_percent', 'N/A')}%")
vg_item.setText(6, vg.get('vg_fmt', 'N/A')) # VG格式 vg_item.setText(6, vg.get('vg_fmt', 'N/A'))
vg_item.setText(7, "") # VG没有直接挂载点
# 存储原始数据确保vg_name是正确的
vg_data_for_context = vg.copy()
vg_data_for_context['vg_name'] = vg_name
vg_item.setData(0, Qt.UserRole, {'type': 'vg', 'data': vg_data_for_context})
else: else:
item = QTreeWidgetItem(vg_root_item) item = QTreeWidgetItem(vg_root_item)
item.setText(0, "未找到卷组。") item.setText(0, "未找到卷组。")
@@ -232,21 +510,45 @@ class MainWindow(QMainWindow):
lv_root_item = QTreeWidgetItem(self.ui.treeWidget_lvm) lv_root_item = QTreeWidgetItem(self.ui.treeWidget_lvm)
lv_root_item.setText(0, "逻辑卷 (LVs)") lv_root_item.setText(0, "逻辑卷 (LVs)")
lv_root_item.setExpanded(True) lv_root_item.setExpanded(True)
lv_root_item.setData(0, Qt.UserRole, {'type': 'lv_root'})
if lvm_data.get('lvs'): if lvm_data.get('lvs'):
for lv in lvm_data['lvs']: for lv in lvm_data['lvs']:
lv_item = QTreeWidgetItem(lv_root_item) lv_item = QTreeWidgetItem(lv_root_item)
lv_item.setText(0, lv.get('lv_name', 'N/A')) lv_name = lv.get('lv_name', 'N/A')
vg_name = lv.get('vg_name', 'N/A')
lv_attr = lv.get('lv_attr', '')
# 确保 lv_path 是一个有效的路径
lv_path = lv.get('lv_path')
if not lv_path or lv_path == 'N/A':
# 如果 lv_path 不存在或为 N/A则尝试从 vg_name 和 lv_name 构造
if vg_name != 'N/A' and lv_name != 'N/A':
lv_path = f"/dev/{vg_name}/{lv_name}"
else:
lv_path = 'N/A' # 实在无法构造则标记为N/A
current_mount_point = self.system_manager.get_mountpoint_for_device(lv_path)
lv_item.setText(0, lv_name)
lv_item.setText(1, lv.get('lv_size', 'N/A')) lv_item.setText(1, lv.get('lv_size', 'N/A'))
lv_item.setText(2, lv.get('lv_attr', 'N/A')) lv_item.setText(2, lv_attr)
lv_item.setText(3, lv.get('lv_uuid', 'N/A')) lv_item.setText(3, lv.get('lv_uuid', 'N/A'))
lv_item.setText(4, f"VG: {lv.get('vg_name', 'N/A')}, Origin: {lv.get('origin', 'N/A')}") lv_item.setText(4, f"VG: {vg_name}, Origin: {lv.get('origin', 'N/A')}")
lv_item.setText(5, f"快照: {lv.get('snap_percent', 'N/A')}%") lv_item.setText(5, f"快照: {lv.get('snap_percent', 'N/A')}%")
lv_item.setText(6, lv.get('lv_path', 'N/A')) lv_item.setText(6, lv_path) # 显示构造或获取的路径
lv_item.setText(7, current_mount_point if current_mount_point else "") # 显示挂载点
# 存储原始数据,确保 lv_path, lv_name, vg_name, lv_attr 都是正确的
lv_data_for_context = lv.copy()
lv_data_for_context['lv_path'] = lv_path
lv_data_for_context['lv_name'] = lv_name
lv_data_for_context['vg_name'] = vg_name
lv_data_for_context['lv_attr'] = lv_attr
lv_item.setData(0, Qt.UserRole, {'type': 'lv', 'data': lv_data_for_context})
else: else:
item = QTreeWidgetItem(lv_root_item) item = QTreeWidgetItem(lv_root_item)
item.setText(0, "未找到逻辑卷。") item.setText(0, "未找到逻辑卷。")
for i in range(len(lvm_headers)): for i in range(len(lvm_headers)):
self.ui.treeWidget_lvm.resizeColumnToContents(i) self.ui.treeWidget_lvm.resizeColumnToContents(i)
logger.info("LVM信息刷新成功。") logger.info("LVM信息刷新成功。")
@@ -255,66 +557,193 @@ class MainWindow(QMainWindow):
QMessageBox.critical(self, "错误", f"刷新LVM信息失败: {e}") QMessageBox.critical(self, "错误", f"刷新LVM信息失败: {e}")
logger.error(f"刷新LVM信息失败: {e}") logger.error(f"刷新LVM信息失败: {e}")
def show_block_device_context_menu(self, pos: QPoint): def show_lvm_context_menu(self, pos: QPoint):
""" item = self.ui.treeWidget_lvm.itemAt(pos)
显示块设备列表的右键上下文菜单。
"""
item = self.ui.treeWidget_block_devices.itemAt(pos)
if item:
dev_data = item.data(0, Qt.UserRole)
if not dev_data:
logger.warning(f"无法获取设备 {item.text(0)} 的详细数据。")
return
device_name = dev_data.get('name')
device_type = dev_data.get('type')
mount_point = dev_data.get('mountpoint')
menu = QMenu(self) menu = QMenu(self)
if device_type in ['part', 'disk']: # 根节点或空白区域的创建菜单
if not mount_point or mount_point == '' or mount_point == 'N/A': create_menu = QMenu("创建...", self)
mount_action = menu.addAction(f"挂载 {device_name}...") create_pv_action = create_menu.addAction("创建物理卷 (PV)...")
mount_action.triggered.connect(lambda: self._handle_mount(device_name)) create_pv_action.triggered.connect(self._handle_create_pv)
elif mount_point != '[SWAP]': create_vg_action = create_menu.addAction("创建卷组 (VG)...")
unmount_action = menu.addAction(f"卸载 {device_name}") create_vg_action.triggered.connect(self._handle_create_vg)
unmount_action.triggered.connect(lambda: self._handle_unmount(device_name)) create_lv_action = create_menu.addAction("创建逻辑卷 (LV)...")
create_lv_action.triggered.connect(self._handle_create_lv)
if menu.actions(): menu.addMenu(create_menu)
menu.addSeparator() menu.addSeparator()
if device_type == 'part': if item:
delete_action = menu.addAction(f"删除分区 {device_name}") item_data = item.data(0, Qt.UserRole)
delete_action.triggered.connect(lambda: self._handle_delete_partition(device_name)) if not item_data:
logger.warning(f"无法获取 LVM 项 {item.text(0)} 的详细数据。")
return
format_action = menu.addAction(f"格式化分区 {device_name}...") item_type = item_data.get('type')
format_action.triggered.connect(lambda: self._handle_format_partition(device_name)) data = item_data.get('data', {})
if item_type == 'pv':
pv_name = data.get('pv_name')
if pv_name and pv_name != 'N/A':
delete_pv_action = menu.addAction(f"删除物理卷 {pv_name}")
delete_pv_action.triggered.connect(lambda: self._handle_delete_pv(pv_name))
elif item_type == 'vg':
vg_name = data.get('vg_name')
if vg_name and vg_name != 'N/A':
delete_vg_action = menu.addAction(f"删除卷组 {vg_name}")
delete_vg_action.triggered.connect(lambda: self._handle_delete_vg(vg_name))
elif item_type == 'lv':
lv_name = data.get('lv_name')
vg_name = data.get('vg_name')
lv_attr = data.get('lv_attr', '')
lv_path = data.get('lv_path') # e.g., /dev/vgname/lvname
# 确保所有关键信息都存在且有效
if lv_name and vg_name and lv_path and lv_path != 'N/A':
# Activation/Deactivation
if 'a' in lv_attr: # 'a' 表示 active
deactivate_lv_action = menu.addAction(f"停用逻辑卷 {lv_name}")
deactivate_lv_action.triggered.connect(lambda: self._handle_deactivate_lv(lv_name, vg_name))
else:
activate_lv_action = menu.addAction(f"激活逻辑卷 {lv_name}")
activate_lv_action.triggered.connect(lambda: self._handle_activate_lv(lv_name, vg_name))
# Mount/Unmount
current_mount_point = self.system_manager.get_mountpoint_for_device(lv_path)
if current_mount_point and current_mount_point != '[SWAP]' and current_mount_point != '':
unmount_lv_action = menu.addAction(f"卸载 {lv_name} ({current_mount_point})")
# FIX: 更改 lambda 表达式,避免被 triggered 信号的参数覆盖
unmount_lv_action.triggered.connect(lambda: self._unmount_and_refresh(lv_path))
else:
mount_lv_action = menu.addAction(f"挂载 {lv_name}...")
mount_lv_action.triggered.connect(lambda: self._handle_mount(lv_path))
menu.addSeparator() # 在挂载/卸载后添加分隔符
# Delete and Format
delete_lv_action = menu.addAction(f"删除逻辑卷 {lv_name}")
delete_lv_action.triggered.connect(lambda: self._handle_delete_lv(lv_name, vg_name))
format_lv_action = menu.addAction(f"格式化逻辑卷 {lv_name}...")
format_lv_action.triggered.connect(lambda: self._handle_format_partition(lv_path))
else:
logger.warning(f"逻辑卷 '{lv_name}' (VG: {vg_name}) 的路径无效无法显示操作。Lv Path: {lv_path}")
if menu.actions(): if menu.actions():
menu.exec(self.ui.treeWidget_block_devices.mapToGlobal(pos)) menu.exec(self.ui.treeWidget_lvm.mapToGlobal(pos))
else: else:
logger.info(f"设备 {device_name} 没有可用的操作。") logger.info("右键点击了空白区域或没有可用的LVM操作。")
def _handle_create_pv(self):
available_partitions = self.system_manager.get_unallocated_partitions()
if not available_partitions:
QMessageBox.warning(self, "警告", "没有可用于创建物理卷的未分配分区。")
return
dialog = CreatePvDialog(self, available_partitions)
if dialog.exec() == QDialog.Accepted:
info = dialog.get_pv_info()
if info: # Check if info is not None
if self.lvm_ops.create_pv(info['device_path']):
self.refresh_all_info()
def _handle_delete_pv(self, device_path):
if self.lvm_ops.delete_pv(device_path):
self.refresh_all_info()
def _handle_create_vg(self):
lvm_info = self.system_manager.get_lvm_info()
available_pvs = []
# Filter PVs that are not part of any VG
for pv in lvm_info.get('pvs', []):
pv_name = pv.get('pv_name')
if pv_name and pv_name != 'N/A' and not pv.get('vg_name'): # Check if pv_name is not associated with any VG
if pv_name.startswith('/dev/'):
available_pvs.append(pv_name)
else: else:
logger.info("右键点击了空白区域。") # Attempt to construct full path if only name is given
available_pvs.append(f"/dev/{pv_name}")
def _handle_mount(self, device_name):
"""处理挂载操作并刷新UI。""" if not available_pvs:
if self.disk_ops.mount_partition(device_name): QMessageBox.warning(self, "警告", "没有可用于创建卷组的物理卷。请确保有未分配给任何卷组的物理卷。")
return
dialog = CreateVgDialog(self, available_pvs)
if dialog.exec() == QDialog.Accepted:
info = dialog.get_vg_info()
if info: # Check if info is not None
if self.lvm_ops.create_vg(info['vg_name'], info['pvs']):
self.refresh_all_info() self.refresh_all_info()
def _handle_unmount(self, device_name): def _handle_delete_vg(self, vg_name):
"""处理卸载操作并刷新UI。""" if self.lvm_ops.delete_vg(vg_name):
if self.disk_ops.unmount_partition(device_name):
self.refresh_all_info() self.refresh_all_info()
def _handle_delete_partition(self, device_name): def _handle_create_lv(self):
"""处理删除分区操作并刷新UI。""" lvm_info = self.system_manager.get_lvm_info()
if self.disk_ops.delete_partition(device_name): available_vgs = []
vg_sizes = {} # 存储每个 VG 的可用大小 (GB)
for vg in lvm_info.get('vgs', []):
vg_name = vg.get('vg_name')
if vg_name and vg_name != 'N/A':
available_vgs.append(vg_name)
free_size_str = vg.get('vg_free', '0B').strip() # 确保去除空白字符
current_vg_size_gb = 0.0 # 为当前VG初始化大小
# LVM输出通常使用 'g', 'm', 't', 'k' 作为单位,而不是 'GB', 'MB'
# 匹配数字部分和可选的单位
match = re.match(r'(\d+\.?\d*)\s*([gmkt])?', free_size_str, re.IGNORECASE)
if match:
value = float(match.group(1))
unit = match.group(2).lower() if match.group(2) else '' # 获取单位,转小写
if unit == 'k': # Kilobytes
current_vg_size_gb = value / (1024 * 1024)
elif unit == 'm': # Megabytes
current_vg_size_gb = value / 1024
elif unit == 'g': # Gigabytes
current_vg_size_gb = value
elif unit == 't': # Terabytes
current_vg_size_gb = value * 1024
elif unit == '': # If no unit, assume GB if it's a large number, or just the value
current_vg_size_gb = value # This might be less accurate, but matches common LVM output
else:
logger.warning(f"未知LVM单位: '{unit}' for '{free_size_str}'")
else:
logger.warning(f"无法解析LVM空闲大小字符串: '{free_size_str}'")
vg_sizes[vg_name] = current_vg_size_gb
if not available_vgs:
QMessageBox.warning(self, "警告", "没有可用于创建逻辑卷的卷组。")
return
dialog = CreateLvDialog(self, available_vgs, vg_sizes) # 传递 vg_sizes
if dialog.exec() == QDialog.Accepted:
info = dialog.get_lv_info()
if info: # 确保 info 不是 None (用户可能在 CreateLvDialog 中取消了操作)
# 传递 use_max_space 标志给 lvm_ops.create_lv
if self.lvm_ops.create_lv(info['lv_name'], info['vg_name'], info['size_gb'], info['use_max_space']):
self.refresh_all_info() self.refresh_all_info()
def _handle_format_partition(self, device_name): def _handle_delete_lv(self, lv_name, vg_name):
"""处理格式化分区操作并刷新UI。""" # 删除 LV 前,也需要确保卸载并从 fstab 移除
if self.disk_ops.format_partition(device_name): # 获取 LV 的完整路径
lv_path = f"/dev/{vg_name}/{lv_name}"
# 尝试卸载并从 fstab 移除 (静默,因为删除操作是主要目的)
self.disk_ops.unmount_partition(lv_path, show_dialog_on_error=False)
self.disk_ops._remove_fstab_entry(lv_path) # 直接调用内部方法,不弹出对话框
if self.lvm_ops.delete_lv(lv_name, vg_name):
self.refresh_all_info()
def _handle_activate_lv(self, lv_name, vg_name):
if self.lvm_ops.activate_lv(lv_name, vg_name):
self.refresh_all_info()
def _handle_deactivate_lv(self, lv_name, vg_name):
if self.lvm_ops.deactivate_lv(lv_name, vg_name):
self.refresh_all_info() self.refresh_all_info()

View File

@@ -2,4 +2,4 @@
name = "PySide Widgets Project" name = "PySide Widgets Project"
[tool.pyside6-project] [tool.pyside6-project]
files = ["disk_operations.py", "form.ui", "logger_config.py", "mainwindow.py", "system_info.py"] files = ["dialogs.py", "disk_operations.py", "form.ui", "logger_config.py", "lvm_operations.py", "mainwindow.py", "raid_operations.py", "system_info.py"]

222
raid_operations.py Normal file
View File

@@ -0,0 +1,222 @@
# raid_operations.py
import logging
import subprocess
from PySide6.QtWidgets import QMessageBox
from system_info import SystemInfoManager
import re # 导入正则表达式模块
logger = logging.getLogger(__name__)
class RaidOperations:
def __init__(self):
self.system_manager = SystemInfoManager()
def _execute_shell_command(self, command_list, error_msg_prefix, suppress_critical_dialog_on_stderr_match=None, input_to_command=None): # <--- 添加 input_to_command 参数
"""
执行一个shell命令并返回stdout和stderr。
这个方法包装了 SystemInfoManager._run_command并统一处理日志和错误消息框。
:param command_list: 命令及其参数的列表。
:param error_msg_prefix: 命令失败时显示给用户的错误消息前缀。
:param suppress_critical_dialog_on_stderr_match: 一个字符串,如果在 stderr 中找到此字符串,
则在 CalledProcessError 发生时,只记录日志,不弹出 QMessagebox.critical。
:param input_to_command: 传递给命令stdin的数据 (str)。
:return: (bool success, str stdout, str stderr)
"""
full_cmd_str = ' '.join(command_list)
logger.info(f"执行命令: {full_cmd_str}")
try:
stdout, stderr = self.system_manager._run_command(
command_list,
root_privilege=True,
check_output=True, # RAID操作通常需要检查输出
input_data=input_to_command # <--- 将 input_to_command 传递给 _run_command
)
if stdout: logger.debug(f"命令 '{full_cmd_str}' 标准输出:\n{stdout.strip()}")
if stderr: logger.debug(f"命令 '{full_cmd_str}' 标准错误:\n{stderr.strip()}")
return True, stdout, stderr
except subprocess.CalledProcessError as e:
logger.error(f"{error_msg_prefix} 命令: {full_cmd_str}")
logger.error(f"退出码: {e.returncode}")
logger.error(f"标准输出: {e.stdout.strip()}")
logger.error(f"标准错误: {e.stderr.strip()}")
# 根据 suppress_critical_dialog_on_stderr_match 参数决定是否弹出错误对话框
if suppress_critical_dialog_on_stderr_match and suppress_critical_dialog_on_stderr_match in e.stderr.strip():
logger.info(f"特定错误 '{suppress_critical_dialog_on_stderr_match}' 匹配,已抑制错误对话框。")
else:
QMessageBox.critical(None, "错误", f"{error_msg_prefix}\n详细信息: {e.stderr.strip()}")
return False, e.stdout, e.stderr
except FileNotFoundError:
logger.error(f"命令 '{command_list[0]}' 未找到。请确保已安装相关工具。")
QMessageBox.critical(None, "错误", f"命令 '{command_list[0]}' 未找到。请确保已安装相关工具。")
return False, "", "Command not found."
except Exception as e:
logger.error(f"{error_msg_prefix} 命令: {full_cmd_str} 发生未知错误: {e}")
QMessageBox.critical(None, "错误", f"{error_msg_prefix}\n未知错误: {e}")
return False, "", str(e)
def create_raid_array(self, devices, level, chunk_size):
"""
创建 RAID 阵列。
:param devices: 成员设备列表,例如 ['/dev/sdb1', '/dev/sdc1']
:param level: RAID 级别,例如 'raid0', 'raid1', 'raid5'
:param chunk_size: Chunk 大小 (KB)
"""
if not devices or len(devices) < 2:
QMessageBox.critical(None, "错误", "创建 RAID 阵列至少需要两个成员设备。")
return False
# 检查 RAID 5 至少需要 3 个设备
if level == "raid5" and len(devices) < 3:
QMessageBox.critical(None, "错误", "RAID5 至少需要三个设备。")
return False
# 确认操作
reply = QMessageBox.question(None, "确认创建 RAID 阵列",
f"你确定要使用设备 {', '.join(devices)} 创建 RAID {level} 阵列吗?\n"
"此操作将销毁设备上的所有数据!",
QMessageBox.Yes | QMessageBox.No, QMessageBox.No)
if reply == QMessageBox.No:
logger.info("用户取消了创建 RAID 阵列的操作。")
return False
logger.info(f"尝试创建 RAID {level} 阵列,成员设备: {', '.join(devices)}, Chunk 大小: {chunk_size}KB。")
# 1. 清除设备上的旧 RAID 超级块(如果有)
for dev in devices:
# 使用 --force 选项,避免交互式提示
clear_cmd = ["mdadm", "--zero-superblock", "--force", dev]
success, _, stderr = self._execute_shell_command(
clear_cmd,
f"清除设备 {dev} 上的旧 RAID 超级块失败",
suppress_critical_dialog_on_stderr_match="No superblocks" # 抑制“没有超级块”的错误提示
)
if success:
logger.info(f"已清除设备 {dev} 上的旧 RAID 超级块。")
else:
# 如果清除失败,但不是因为“没有超级块”,则可能需要进一步处理
if "No superblocks" not in stderr:
QMessageBox.warning(None, "警告", f"清除设备 {dev} 上的旧 RAID 超级块可能失败,但尝试继续。")
# 2. 创建 RAID 阵列
# 默认阵列名为 /dev/md/new_raid可以考虑让用户输入
array_name = "/dev/md/new_raid"
if level == "raid0":
# RAID0 至少2个设备
create_cmd = ["mdadm", "--create", array_name, "--level=raid0",
f"--raid-devices={len(devices)}", f"--chunk={chunk_size}K"] + devices
elif level == "raid1":
# RAID1 至少2个设备
create_cmd = ["mdadm", "--create", array_name, "--level=raid1",
f"--raid-devices={len(devices)}"] + devices
elif level == "raid5":
# RAID5 至少3个设备
create_cmd = ["mdadm", "--create", array_name, "--level=raid5",
f"--raid-devices={len(devices)}"] + devices
else:
QMessageBox.critical(None, "错误", f"不支持的 RAID 级别: {level}")
return False
# 在 --create 命令中添加 --force 选项
create_cmd.insert(2, "--force") # 插入到 --create 后面
# <--- 关键修改:通过 input_to_command 参数传入 'y\n' 来强制 mdadm 接受
if not self._execute_shell_command(create_cmd, f"创建 RAID 阵列失败", input_to_command='y\n')[0]:
return False
logger.info(f"成功创建 RAID {level} 阵列 {array_name}")
QMessageBox.information(None, "成功", f"成功创建 RAID {level} 阵列 {array_name}")
# 3. 刷新 mdadm 配置并等待阵列激活
# 注意:这里使用 '>>' 重定向subprocess.run 无法直接处理 shell 重定向符号
# 需要改成先读取,再写入,或者使用 bash -c "..."
# 暂时先用 bash -c 的方式,更简单
# 旧代码self._execute_shell_command(["mdadm", "--examine", "--scan", ">>", "/etc/mdadm/mdadm.conf"], "更新 mdadm.conf 失败")
# 获取新的 mdadm.conf 内容
examine_scan_cmd = ["mdadm", "--examine", "--scan"]
success_scan, scan_stdout, _ = self._execute_shell_command(examine_scan_cmd, "扫描 mdadm 配置失败")
if success_scan:
# 将扫描结果追加到 mdadm.conf
append_to_conf_cmd = ["bash", "-c", f"echo '{scan_stdout.strip()}' >> /etc/mdadm/mdadm.conf"]
if not self._execute_shell_command(append_to_conf_cmd, "更新 /etc/mdadm/mdadm.conf 失败")[0]:
logger.warning("更新 /etc/mdadm/mdadm.conf 失败。")
else:
logger.warning("未能扫描到 mdadm 配置,跳过更新 mdadm.conf。")
# self._execute_shell_command(["update-initramfs", "-u"], "更新 initramfs 失败")
return True
def stop_raid_array(self, array_path):
"""
停止一个 RAID 阵列。
:param array_path: RAID 阵列的设备路径,例如 /dev/md0
"""
reply = QMessageBox.question(None, "确认停止 RAID 阵列",
f"你确定要停止 RAID 阵列 {array_path} 吗?\n"
"停止阵列将使其无法访问,并可能需要重新组装。",
QMessageBox.Yes | QMessageBox.No, QMessageBox.No)
if reply == QMessageBox.No:
logger.info(f"用户取消了停止 RAID 阵列 {array_path} 的操作。")
return False
logger.info(f"尝试停止 RAID 阵列: {array_path}")
# 尝试卸载阵列(如果已挂载),不显示错误对话框
# 注意:这里需要调用 disk_operations 的 unmount_partition
# 由于 RaidOperations 不直接持有 DiskOperations 实例,需要通过某种方式获取或传递
# 暂时先直接调用 umount 命令,不处理 fstab
# 或者,更好的方式是让 MainWindow 调用 disk_ops.unmount_partition
# 此处简化处理,只执行 umount 命令
self._execute_shell_command(["umount", array_path], f"尝试卸载 {array_path} 失败", suppress_critical_dialog_on_stderr_match="not mounted")
if not self._execute_shell_command(["mdadm", "--stop", array_path], f"停止 RAID 阵列 {array_path} 失败")[0]:
return False
logger.info(f"成功停止 RAID 阵列 {array_path}")
QMessageBox.information(None, "成功", f"成功停止 RAID 阵列 {array_path}")
return True
def delete_raid_array(self, array_path, member_devices):
"""
删除一个 RAID 阵列。
此操作将停止阵列并清除成员设备上的超级块。
:param array_path: RAID 阵列的设备路径,例如 /dev/md0
:param member_devices: 成员设备列表,例如 ['/dev/sdb1', '/dev/sdc1']
"""
reply = QMessageBox.question(None, "确认删除 RAID 阵列",
f"你确定要删除 RAID 阵列 {array_path} 吗?\n"
"此操作将停止阵列并清除成员设备上的 RAID 超级块,数据将无法访问!",
QMessageBox.Yes | QMessageBox.No, QMessageBox.No)
if reply == QMessageBox.No:
logger.info(f"用户取消了删除 RAID 阵列 {array_path} 的操作。")
return False
logger.info(f"尝试删除 RAID 阵列: {array_path}")
# 1. 停止阵列
if not self.stop_raid_array(array_path):
QMessageBox.critical(None, "错误", f"删除 RAID 阵列 {array_path} 失败,因为无法停止阵列。")
return False
# 2. 清除成员设备上的超级块
success_all_cleared = True
for dev in member_devices:
# 使用 --force 选项,避免交互式提示
clear_cmd = ["mdadm", "--zero-superblock", "--force", dev]
if not self._execute_shell_command(clear_cmd, f"清除设备 {dev} 上的 RAID 超级块失败")[0]:
success_all_cleared = False
QMessageBox.warning(None, "警告", f"未能清除设备 {dev} 上的 RAID 超级块,请手动检查。")
if success_all_cleared:
logger.info(f"成功删除 RAID 阵列 {array_path} 并清除了成员设备超级块。")
QMessageBox.information(None, "成功", f"成功删除 RAID 阵列 {array_path}")
return True
else:
QMessageBox.critical(None, "错误", f"删除 RAID 阵列 {array_path} 完成,但部分成员设备超级块未能完全清除。")
return False

View File

@@ -1,8 +1,9 @@
# system_info.py # system_info.py
import subprocess import subprocess
import json import json
import os
import logging import logging
import re
import os # 导入 os 模块
logger = logging.getLogger(__name__) logger = logging.getLogger(__name__)
@@ -10,255 +11,430 @@ class SystemInfoManager:
def __init__(self): def __init__(self):
pass pass
def _run_command(self, cmd, root_privilege=False, check_output=True): def _run_command(self, command_list, root_privilege=False, check_output=True, input_data=None):
""" """
执行shell命令并返回stdout和stderr 通用地运行一个 shell 命令。
如果需要root权限会尝试使用sudo :param command_list: 命令及其参数的列表
所有输出和错误都会通过 logger 记录 :param root_privilege: 如果为 True则使用 sudo 执行命令
:param check_output: 如果为 True则捕获 stdout 和 stderr。如果为 False则不捕获用于需要交互或不关心输出的命令。
:param input_data: 传递给命令stdin的数据 (str)。
:return: (stdout_str, stderr_str)
:raises subprocess.CalledProcessError: 如果命令返回非零退出码。
""" """
full_cmd = []
if root_privilege: if root_privilege:
# 检查当前是否已经是root用户 command_list = ["sudo"] + command_list
if os.geteuid() != 0:
full_cmd.append("sudo")
full_cmd.extend(cmd)
cmd_str = ' '.join(full_cmd)
logger.info(f"执行命令: {cmd_str}")
logger.debug(f"运行命令: {' '.join(command_list)}")
try: try:
result = subprocess.run( result = subprocess.run(
full_cmd, command_list,
capture_output=True, capture_output=check_output,
text=True, text=True,
check=check_output, check=True,
encoding='utf-8', encoding='utf-8',
env=dict(os.environ, LANG="en_US.UTF-8") input=input_data
) )
if result.stdout: return result.stdout if check_output else "", result.stderr if check_output else ""
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: except subprocess.CalledProcessError as e:
error_msg = f"命令执行失败: {cmd_str}\n" \ logger.error(f"命令执行失败: {' '.join(command_list)}")
f"退出码: {e.returncode}\n" \ logger.error(f"退出码: {e.returncode}")
f"标准输出: {e.stdout}\n" \ logger.error(f"标准输出: {e.stdout.strip()}")
f"标准错误: {e.stderr}" logger.error(f"标准错误: {e.stderr.strip()}")
logger.error(error_msg) raise
raise ValueError(error_msg)
except FileNotFoundError: except FileNotFoundError:
error_msg = f"命令未找到: {full_cmd[0]}。请确保已安装。" logger.error(f"命令 '{command_list[0]}' 未找到。请确保已安装相关工具")
logger.error(error_msg) raise
raise FileNotFoundError(error_msg)
except Exception as e: except Exception as e:
error_msg = f"行命令时发生未知错误: {e}" logger.error(f"行命令 {' '.join(command_list)} 时发生未知错误: {e}")
logger.error(error_msg) raise
raise RuntimeError(error_msg)
def get_block_devices(self): def get_block_devices(self):
""" """
获取所有块设备信息以JSON格式返回 使用 lsblk 获取块设备信息。
使用 lsblk -J 命令 返回一个字典列表,每个字典代表一个设备
""" """
cmd = [
"lsblk", "-J", "-o",
"NAME,FSTYPE,SIZE,MOUNTPOINT,RO,TYPE,UUID,PARTUUID,VENDOR,MODEL,SERIAL,MAJ:MIN,PKNAME,PATH"
]
try: 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) stdout, _ = self._run_command(cmd)
data = json.loads(stdout) data = json.loads(stdout)
logger.info("成功获取块设备信息。") devices = data.get('blockdevices', [])
return data.get('blockdevices', [])
def add_path_recursive(dev_list):
for dev in dev_list:
if 'PATH' not in dev and 'NAME' in dev:
dev['PATH'] = f"/dev/{dev['NAME']}"
# Rename PATH to path (lowercase) for consistency
if 'PATH' in dev:
dev['path'] = dev.pop('PATH')
if 'children' in dev:
add_path_recursive(dev['children'])
add_path_recursive(devices)
return devices
except Exception as e: except Exception as e:
logger.error(f"获取块设备信息失败: {e}") logger.error(f"获取块设备信息失败: {e}")
return [] return []
def _find_device_by_path_recursive(self, dev_list, target_path):
"""
Helper to find device data by its path recursively.
Added type check for target_path.
"""
if not isinstance(target_path, str): # 添加类型检查
logger.warning(f"传入 _find_device_by_path_recursive 的 target_path 不是字符串: {target_path} (类型: {type(target_path)})")
return None
for dev in dev_list:
if dev.get('path') == target_path:
return dev
if 'children' in dev:
found = self._find_device_by_path_recursive(dev['children'], target_path)
if found:
return found
return None
def get_mountpoint_for_device(self, device_path):
"""
根据设备路径获取其挂载点。
此方法现在可以正确处理 /dev/md/new_raid 这样的 RAID 阵列别名。
Added type check for device_path.
"""
if not isinstance(device_path, str): # 添加类型检查
logger.warning(f"传入 get_mountpoint_for_device 的 device_path 不是字符串: {device_path} (类型: {type(device_path)})")
return None
devices = self.get_block_devices()
# 1. 尝试直接使用提供的 device_path 在 lsblk 输出中查找
dev_info = self._find_device_by_path_recursive(devices, device_path)
if dev_info:
logger.debug(f"直接从 lsblk 获取到 {device_path} 的挂载点: {dev_info.get('mountpoint')}")
return dev_info.get('mountpoint')
# 2. 如果直接查找失败,并且是 RAID 阵列(例如 /dev/md/new_raid
if device_path.startswith('/dev/md'): # 此处 device_path 已通过类型检查
logger.debug(f"处理 RAID 阵列 {device_path} 以获取挂载点...")
# 获取 RAID 阵列的实际内核设备路径(例如 /dev/md127
actual_md_device_path = self._get_actual_md_device_path(device_path)
logger.debug(f"RAID 阵列 {device_path} 的实际设备路径: {actual_md_device_path}")
if actual_md_device_path:
# 现在,使用实际的内核设备路径从 lsblk 中查找挂载点
actual_dev_info = self._find_device_by_path_recursive(devices, actual_md_device_path)
logger.debug(f"实际设备路径 {actual_md_device_path} 的 lsblk 详情: {actual_dev_info}")
if actual_dev_info:
logger.debug(f"在实际设备 {actual_md_device_path} 上找到了挂载点: {actual_dev_info.get('mountpoint')}")
return actual_dev_info.get('mountpoint')
logger.debug(f"未能获取到 {device_path} 的挂载点。")
return None
def get_mdadm_arrays(self): def get_mdadm_arrays(self):
""" """
获取所有RAID阵列的详细信息。 获取 mdadm RAID 阵列信息。
首先使用 mdadm --detail --scan 获取阵列列表,
然后对每个阵列使用 mdadm --detail 获取更详细的信息。
""" """
all_raid_details = [] cmd = ["mdadm", "--detail", "--scan"]
try: try:
# 1. 获取所有RAID设备的路径 stdout, _ = self._run_command(cmd)
scan_stdout, _ = self._run_command(["mdadm", "--detail", "--scan"], root_privilege=False) arrays = []
array_paths = [] for line in stdout.splitlines():
for line in scan_stdout.splitlines():
if line.startswith("ARRAY"): if line.startswith("ARRAY"):
parts = line.split() parts = line.split(' ')
if len(parts) > 1: array_path = parts[1] # e.g., /dev/md0 or /dev/md/new_raid
array_paths.append(parts[1]) # 例如 /dev/md126 # Use mdadm --detail for more specific info
detail_cmd = ["mdadm", "--detail", array_path]
if not array_paths: detail_stdout, _ = self._run_command(detail_cmd)
array_info = self._parse_mdadm_detail(detail_stdout, array_path)
arrays.append(array_info)
return arrays
except subprocess.CalledProcessError as e:
if "No arrays found" in e.stderr or "No arrays found" in e.stdout: # mdadm --detail --scan might exit 1 if no arrays
logger.info("未找到任何RAID阵列。") logger.info("未找到任何RAID阵列。")
return [] return []
logger.error(f"获取RAID阵列信息失败: {e}")
# 2. 对每个RAID设备获取详细信息 return []
for path in array_paths:
try:
detail_stdout, _ = self._run_command(["mdadm", "--detail", path], root_privilege=False)
parsed_detail = self._parse_mdadm_detail_output(detail_stdout)
parsed_detail['device'] = path # 添加设备路径
all_raid_details.append(parsed_detail)
except Exception as e: except Exception as e:
logger.error(f"获取RAID阵列 {path} 的详细信息失败: {e}") logger.error(f"获取RAID阵列信息失败: {e}")
logger.info("成功获取RAID阵列信息。")
return all_raid_details
except Exception as e:
logger.error(f"获取RAID阵列列表失败: {e}")
return [] return []
def _parse_mdadm_detail_output(self, output_string): def _parse_mdadm_detail(self, detail_output, array_path):
""" """
解析 mdadm --detail 命令的输出字符串,提取关键信息 解析 mdadm --detail 的输出。
""" """
details = { info = {
'device': array_path,
'level': 'N/A', 'level': 'N/A',
'array_size': 'N/A',
'raid_devices': 'N/A',
'total_devices': 'N/A',
'state': 'N/A', 'state': 'N/A',
'array_size': 'N/A',
'active_devices': 'N/A', 'active_devices': 'N/A',
'working_devices': 'N/A',
'failed_devices': 'N/A', 'failed_devices': 'N/A',
'spare_devices': 'N/A', 'spare_devices': 'N/A',
'chunk_size': 'N/A', 'total_devices': 'N/A',
'uuid': 'N/A', 'uuid': 'N/A',
'name': 'N/A', 'name': 'N/A',
'chunk_size': 'N/A',
'member_devices': [] 'member_devices': []
} }
member_devices_section = False member_pattern = re.compile(r'^\s*(\d+)\s+(\d+)\s+(\d+)\s+(.+?)\s+(/dev/.+)$')
lines = output_string.splitlines()
for line in lines: for line in detail_output.splitlines():
line = line.strip() if "Raid Level :" in line:
if not line: info['level'] = line.split(':')[-1].strip()
continue elif "Array Size :" in line:
info['array_size'] = line.split(':')[-1].strip()
elif "State :" in line:
info['state'] = line.split(':')[-1].strip()
elif "Active Devices :" in line:
info['active_devices'] = line.split(':')[-1].strip()
elif "Failed Devices :" in line:
info['failed_devices'] = line.split(':')[-1].strip()
elif "Spare Devices :" in line:
info['spare_devices'] = line.split(':')[-1].strip()
elif "Total Devices :" in line:
info['total_devices'] = line.split(':')[-1].strip()
elif "UUID :" in line:
info['uuid'] = line.split(':')[-1].strip()
elif "Name :" in line:
info['name'] = line.split(':')[-1].strip()
elif "Chunk Size :" in line:
info['chunk_size'] = line.split(':')[-1].strip()
if line.startswith("Number Major Minor RaidDevice State"): # Member devices
member_devices_section = True match = member_pattern.match(line)
continue if match:
member_info = {
if member_devices_section: 'number': match.group(1),
parts = line.split() 'major': match.group(2),
if len(parts) >= 6: # Ensure enough parts for device info 'minor': match.group(3),
# State can be multiple words, e.g., "active sync" 'raid_device': match.group(4),
device_state_parts = parts[4:-1] 'device_path': match.group(5)
# Handle cases where device path might have spaces or be complex, though usually not for /dev/sdX }
device_path = parts[-1] info['member_devices'].append(member_info)
return info
details['member_devices'].append({
'number': parts[0],
'major': parts[1],
'minor': parts[2],
'raid_device': parts[3],
'state': ' '.join(device_state_parts),
'device_path': device_path
})
else:
if ':' in line:
key, value = line.split(':', 1)
key = key.strip().lower().replace(' ', '_')
value = value.strip()
if key == 'raid_level':
details['level'] = value
elif key == 'array_size':
details['array_size'] = value
elif key == 'raid_devices':
details['raid_devices'] = value
elif key == 'total_devices':
details['total_devices'] = value
elif key == 'state':
details['state'] = value
elif key == 'active_devices':
details['active_devices'] = value
elif key == 'working_devices':
details['working_devices'] = value
elif key == 'failed_devices':
details['failed_devices'] = value
elif key == 'spare_devices':
details['spare_devices'] = value
elif key == 'chunk_size':
details['chunk_size'] = value
elif key == 'uuid':
details['uuid'] = value
elif key == 'name':
details['name'] = value
return details
def get_lvm_info(self): def get_lvm_info(self):
""" """
获取LVM物理卷、卷组、逻辑卷信息。 获取 LVM 物理卷 (PVs)、卷组 (VGs) 和逻辑卷 (LVs) 的信息。
使用 pvs, vgs, lvs 命令并尝试获取JSON格式。
""" """
lvm_info = {} lvm_info = {'pvs': [], 'vgs': [], 'lvs': []}
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'] = []
# Get PVs
try: try:
stdout_vgs, _ = self._run_command(["vgs", "--reportformat", "json"], root_privilege=False) stdout, _ = self._run_command(["pvs", "--reportformat", "json"])
lvm_info['vgs'] = json.loads(stdout_vgs).get('report', [])[0].get('vg', []) data = json.loads(stdout)
logger.info("成功获取卷组信息。") if 'report' in data and data['report']:
for pv_data in data['report'][0].get('pv', []):
lvm_info['pvs'].append({
'pv_name': pv_data.get('pv_name'),
'vg_name': pv_data.get('vg_name'),
'pv_uuid': pv_data.get('pv_uuid'),
'pv_size': pv_data.get('pv_size'),
'pv_free': pv_data.get('pv_free'),
'pv_attr': pv_data.get('pv_attr'),
'pv_fmt': pv_data.get('pv_fmt')
})
except subprocess.CalledProcessError as e:
if "No physical volume found" in e.stderr or "No physical volumes found" in e.stdout:
logger.info("未找到任何LVM物理卷。")
else:
logger.error(f"获取LVM物理卷信息失败: {e}")
except Exception as e: except Exception as e:
logger.error(f"获取卷信息失败: {e}") logger.error(f"获取LVM物理卷信息失败: {e}")
lvm_info['vgs'] = []
# Get VGs
try: try:
stdout_lvs, _ = self._run_command(["lvs", "--reportformat", "json"], root_privilege=False) stdout, _ = self._run_command(["vgs", "--reportformat", "json"])
lvm_info['lvs'] = json.loads(stdout_lvs).get('report', [])[0].get('lv', []) data = json.loads(stdout)
logger.info("成功获取逻辑卷信息。") if 'report' in data and data['report']:
for vg_data in data['report'][0].get('vg', []):
lvm_info['vgs'].append({
'vg_name': vg_data.get('vg_name'),
'vg_uuid': vg_data.get('vg_uuid'),
'vg_size': vg_data.get('vg_size'),
'vg_free': vg_data.get('vg_free'),
'vg_attr': vg_data.get('vg_attr'),
'pv_count': vg_data.get('pv_count'),
'lv_count': vg_data.get('lv_count'),
'vg_alloc_percent': vg_data.get('vg_alloc_percent'),
'vg_fmt': vg_data.get('vg_fmt')
})
except subprocess.CalledProcessError as e:
if "No volume group found" in e.stderr or "No volume groups found" in e.stdout:
logger.info("未找到任何LVM卷组。")
else:
logger.error(f"获取LVM卷组信息失败: {e}")
except Exception as e: except Exception as e:
logger.error(f"获取逻辑卷信息失败: {e}") logger.error(f"获取LVM卷组信息失败: {e}")
lvm_info['lvs'] = []
# Get LVs
try:
stdout, _ = self._run_command(["lvs", "--reportformat", "json"])
data = json.loads(stdout)
if 'report' in data and data['report']:
for lv_data in data['report'][0].get('lv', []):
lvm_info['lvs'].append({
'lv_name': lv_data.get('lv_name'),
'vg_name': lv_data.get('vg_name'),
'lv_uuid': lv_data.get('lv_uuid'),
'lv_size': lv_data.get('lv_size'),
'lv_attr': lv_data.get('lv_attr'),
'origin': lv_data.get('origin'),
'snap_percent': lv_data.get('snap_percent'),
'lv_path': lv_data.get('lv_path')
})
except subprocess.CalledProcessError as e:
if "No logical volume found" in e.stderr or "No logical volumes found" in e.stdout:
logger.info("未找到任何LVM逻辑卷。")
else:
logger.error(f"获取LVM逻辑卷信息失败: {e}")
except Exception as e:
logger.error(f"获取LVM逻辑卷信息失败: {e}")
return lvm_info return lvm_info
# 示例用法 (可以在此模块中添加测试代码) def get_unallocated_partitions(self):
if __name__ == "__main__": """
logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s') 获取所有可用于创建RAID阵列或LVM物理卷的设备。
manager = SystemInfoManager() 这些设备包括:
logger.info("--- 块设备信息 ---") 1. 没有分区表的整个磁盘。
devices = manager.get_block_devices() 2. 未挂载的分区包括有文件系统但未挂载的如ext4/xfs等
for dev in devices: 3. 未被LVM或RAID使用的分区或磁盘。
logger.info(f" NAME: {dev.get('name')}, TYPE: {dev.get('type')}, SIZE: {dev.get('size')}, MOUNTPOINT: {dev.get('mountpoint')}") 返回一个设备路径列表,例如 ['/dev/sdb1', '/dev/sdc']。
"""
block_devices = self.get_block_devices()
candidates = []
logger.info("\n--- RAID 阵列信息 ---") def process_device(dev):
raid_info = manager.get_mdadm_arrays() dev_path = dev.get('path')
if raid_info: dev_type = dev.get('type')
for array in raid_info: mountpoint = dev.get('mountpoint')
logger.info(f" Device: {array.get('device')}, Level: {array.get('level')}, State: {array.get('state')}, Size: {array.get('array_size')}") fstype = dev.get('fstype')
logger.info(f" Active: {array.get('active_devices')}, Failed: {array.get('failed_devices')}, Spare: {array.get('spare_devices')}")
for member in array.get('member_devices', []):
logger.info(f" Member: {member.get('device_path')} (State: {member.get('state')})")
else:
logger.info(" 未找到RAID阵列。")
logger.info("\n--- LVM 信息 ---") # 1. 检查是否已挂载 (排除SWAP因为SWAP可以被覆盖但如果活跃则不应动)
lvm_info = manager.get_lvm_info() # 如果是活跃挂载点非SWAP则不是候选
logger.info("物理卷 (PVs):") if mountpoint and mountpoint != '[SWAP]':
if lvm_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')})") # 或者子设备挂载不代表父设备不能用(例如使用整个磁盘)
else: if 'children' in dev:
logger.info(" 未找到物理卷。") for child in dev['children']:
process_device(child)
return
logger.info("卷组 (VGs):") # 2. 检查是否是LVM物理卷或RAID成员
if lvm_info['vgs']: # 如果是LVM PV或RAID成员则它不是新RAID/PV的候选
for vg in lvm_info['vgs']: if fstype in ['LVM2_member', 'linux_raid_member']:
logger.info(f" {vg.get('vg_name')} (Size: {vg.get('vg_size')}, PVs: {vg.get('pv_count')}, LVs: {vg.get('lv_count')})") if 'children' in dev: # 即使是LVM/RAID成员也可能存在子设备例如LVM上的分区虽然不常见
else: for child in dev['children']:
logger.info(" 未找到卷组。") process_device(child)
return
logger.info("逻辑卷 (LVs):") # 3. 如果是整个磁盘且没有分区,则是一个候选
if lvm_info['lvs']: if dev_type == 'disk' and not dev.get('children'):
for lv in lvm_info['lvs']: candidates.append(dev_path)
logger.info(f" {lv.get('lv_name')} (VG: {lv.get('vg_name')}, Size: {lv.get('lv_size')})") # 4. 如果是分区且通过了上述挂载和LVM/RAID检查则是一个候选
elif dev_type == 'part':
candidates.append(dev_path)
# 5. 递归处理子设备 (对于有分区的磁盘)
# 这一步放在最后,确保先对当前设备进行评估
if dev_type == 'disk' and dev.get('children'): # 只有当是磁盘且有子设备时才需要递归
for child in dev['children']:
process_device(child)
for dev in block_devices:
process_device(dev)
# 去重并排序
return sorted(list(set(candidates)))
def _get_actual_md_device_path(self, array_path):
"""
解析 mdadm 阵列名称(例如 /dev/md/new_raid
对应的实际内核设备路径(例如 /dev/md127
通过检查 /dev/md/ 目录下的符号链接来获取。
Added type check for array_path.
"""
if not isinstance(array_path, str): # 添加类型检查
logger.warning(f"传入 _get_actual_md_device_path 的 array_path 不是字符串: {array_path} (类型: {type(array_path)})")
return None
if os.path.exists(array_path):
try:
# os.path.realpath 会解析符号链接,例如 /dev/md/new_raid -> /dev/md127
actual_path = os.path.realpath(array_path)
# 确保解析后仍然是 md 设备(以 /dev/md 开头)
if actual_path.startswith('/dev/md'):
logger.debug(f"已将 RAID 阵列 {array_path} 解析为实际设备路径: {actual_path}")
return actual_path
except Exception as e:
logger.warning(f"无法通过 os.path.realpath 获取 RAID 阵列 {array_path} 的实际设备路径: {e}")
logger.warning(f"RAID 阵列 {array_path} 不存在或无法解析其真实路径。")
return None
def get_device_details_by_path(self, device_path):
"""
根据设备路径获取设备的 UUID 和文件系统类型 (fstype)。
此方法现在可以正确处理 /dev/md/new_raid 这样的 RAID 阵列别名。
Added type check for device_path.
:param device_path: 设备的路径,例如 '/dev/sdb1''/dev/md0''/dev/md/new_raid'
:return: 包含 'uuid''fstype' 的字典,如果未找到则返回 None。
"""
if not isinstance(device_path, str): # 添加类型检查
logger.warning(f"传入 get_device_details_by_path 的 device_path 不是字符串: {device_path} (类型: {type(device_path)})")
return None
devices = self.get_block_devices() # 获取所有块设备信息
# 1. 首先,尝试直接使用提供的 device_path 在 lsblk 输出中查找
# 对于 /dev/md/new_raid 这样的别名,这一步通常会返回 None
lsblk_details = self._find_device_by_path_recursive(devices, device_path)
logger.debug(f"lsblk_details for {device_path} (direct lookup): {lsblk_details}")
if lsblk_details and lsblk_details.get('fstype'): # 即使没有UUID有fstype也算找到部分信息
# 如果直接找到了 fstype就返回 lsblk 提供的 UUID 和 fstype
# 这里的 UUID 应该是文件系统 UUID
logger.debug(f"直接从 lsblk 获取到 {device_path} 的详情: {lsblk_details}")
return {
'uuid': lsblk_details.get('uuid'),
'fstype': lsblk_details.get('fstype')
}
# 2. 如果直接查找失败,并且是 RAID 阵列(例如 /dev/md/new_raid
if device_path.startswith('/dev/md'): # 此处 device_path 已通过类型检查
logger.debug(f"处理 RAID 阵列 {device_path}...")
# 获取 RAID 阵列的实际内核设备路径(例如 /dev/md127
actual_md_device_path = self._get_actual_md_device_path(device_path)
logger.debug(f"RAID 阵列 {device_path} 的实际设备路径: {actual_md_device_path}")
if actual_md_device_path:
# 现在,使用实际的内核设备路径从 lsblk 中查找 fstype 和 UUID (文件系统 UUID)
actual_device_lsblk_details = self._find_device_by_path_recursive(devices, actual_md_device_path)
logger.debug(f"实际设备路径 {actual_md_device_path} 的 lsblk 详情: {actual_device_lsblk_details}")
if actual_device_lsblk_details and actual_device_lsblk_details.get('fstype'):
# 找到了实际 RAID 设备上的文件系统信息
# 此时的 UUID 是文件系统 UUIDfstype 是文件系统类型
logger.debug(f"在实际设备 {actual_md_device_path} 上找到了文件系统详情: {actual_device_lsblk_details}")
return {
'uuid': actual_device_lsblk_details.get('uuid'),
'fstype': actual_device_lsblk_details.get('fstype')
}
else: else:
logger.info(" 未找到逻辑卷。") # RAID 设备存在,但 lsblk 没有报告文件系统 (例如,尚未格式化)
# 此时 fstype 为 None。如果需要我们可以返回 RAID 阵列本身的 UUID但 fstype 仍为 None
logger.warning(f"RAID 阵列 {device_path} (实际设备 {actual_md_device_path}) 未找到文件系统类型。")
# 对于 fstab如果没有 fstype就无法创建条目。
# 此时返回 None让调用者知道无法写入 fstab。
return None
else:
logger.warning(f"无法确定 RAID 阵列 {device_path} 的实际内核设备路径。")
return None # 无法解析实际设备路径,也无法获取 fstype
# 3. 如果仍然没有找到,返回 None
logger.debug(f"未能获取到 {device_path} 的任何详情。")
return None