363 lines
17 KiB
Python
363 lines
17 KiB
Python
# lvm_operations.py
|
||
import subprocess
|
||
import logging
|
||
from PySide6.QtWidgets import QMessageBox
|
||
|
||
logger = logging.getLogger(__name__)
|
||
|
||
class LvmOperations:
|
||
def __init__(self):
|
||
pass
|
||
|
||
def _execute_shell_command(self, command_list, error_message, root_privilege=True,
|
||
suppress_critical_dialog_on_stderr_match=None, input_data=None,
|
||
show_dialog=True, expected_non_zero_exit_codes=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)。
|
||
:param show_dialog: 如果为 False，则不显示关键错误对话框。
|
||
:param expected_non_zero_exit_codes: 预期为非零但仍视为成功的退出码列表。
|
||
:return: (True/False, stdout_str, stderr_str)
|
||
"""
|
||
if not all(isinstance(arg, str) for arg in command_list):
|
||
logger.error(f"命令列表包含非字符串元素: {command_list}")
|
||
if show_dialog:
|
||
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}")
|
||
|
||
# NEW LOGIC: Check if the exit code is expected as non-critical
|
||
if expected_non_zero_exit_codes and e.returncode in expected_non_zero_exit_codes:
|
||
logger.info(f"命令 '{full_cmd_str}' 以预期非零退出码 {e.returncode} 结束，仍视为成功。")
|
||
return True, e.stdout.strip(), stderr_output # Treat as success
|
||
|
||
|
||
should_suppress_dialog = False
|
||
if suppress_critical_dialog_on_stderr_match:
|
||
if isinstance(suppress_critical_dialog_on_stderr_match, str):
|
||
if suppress_critical_dialog_on_stderr_match in stderr_output:
|
||
should_suppress_dialog = True
|
||
elif isinstance(suppress_critical_dialog_on_stderr_match, (list, tuple)):
|
||
for pattern in suppress_critical_dialog_on_stderr_match:
|
||
if pattern in stderr_output:
|
||
should_suppress_dialog = True
|
||
break
|
||
|
||
if show_dialog and not should_suppress_dialog:
|
||
QMessageBox.critical(None, "错误", f"{error_message}\n错误详情: {stderr_output}")
|
||
|
||
return False, e.stdout.strip(), stderr_output
|
||
except FileNotFoundError:
|
||
if show_dialog:
|
||
QMessageBox.critical(None, "错误", f"命令 '{command_list[0]}' 未找到。请确保已安装相关工具。")
|
||
logger.error(f"命令 '{command_list[0]}' 未找到。")
|
||
return False, "", f"命令 '{command_list[0]}' 未找到。"
|
||
except Exception as e:
|
||
if show_dialog:
|
||
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 或 /dev/sdb。
|
||
: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],
|
||
f"在 {device_path} 上创建物理卷失败",
|
||
suppress_critical_dialog_on_stderr_match="device is partitioned"
|
||
)
|
||
|
||
if success:
|
||
QMessageBox.information(None, "成功", f"物理卷已在 {device_path} 上成功创建。")
|
||
return True
|
||
else:
|
||
if "device is partitioned" in stderr:
|
||
wipe_reply = QMessageBox.question(
|
||
None, "设备已分区",
|
||
f"设备 {device_path} 已分区。您是否要擦除设备上的所有分区表，"
|
||
f"并将其整个用于物理卷？此操作将导致所有数据丢失！",
|
||
QMessageBox.Yes | QMessageBox.No, QMessageBox.No
|
||
)
|
||
if wipe_reply == QMessageBox.Yes:
|
||
logger.warning(f"用户选择擦除 {device_path} 上的分区表并创建物理卷。")
|
||
mklabel_success, _, mklabel_stderr = self._execute_shell_command(
|
||
["parted", "-s", device_path, "mklabel", "gpt"],
|
||
f"擦除 {device_path} 上的分区表失败"
|
||
)
|
||
if mklabel_success:
|
||
logger.info(f"已成功擦除 {device_path} 上的分区表。重试创建物理卷。")
|
||
retry_success, _, retry_stderr = self._execute_shell_command(
|
||
["pvcreate", "-y", device_path],
|
||
f"在 {device_path} 上创建物理卷失败 (重试)"
|
||
)
|
||
if retry_success:
|
||
QMessageBox.information(None, "成功", f"物理卷已在 {device_path} 上成功创建。")
|
||
return True
|
||
else:
|
||
return False
|
||
else:
|
||
return False
|
||
else:
|
||
logger.info(f"用户取消了擦除 {device_path} 分区表的操作。")
|
||
QMessageBox.information(None, "信息", f"未在已分区设备 {device_path} 上创建物理卷。")
|
||
return False
|
||
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}。")
|
||
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
|
||
|
||
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", "-y", "-l", "100%FREE", "-n", lv_name, vg_name]
|
||
logger.info(f"尝试在卷组 {vg_name} 中使用最大可用空间创建逻辑卷 {lv_name}。")
|
||
else:
|
||
create_cmd = ["lvcreate", "-y", "-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}。")
|
||
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
|