# lvm_operations_tkinter.py import logging import subprocess from tkinter import messagebox from system_info import SystemInfoManager logger = logging.getLogger(__name__) class LvmOperations: """LVM 操作类 - Tkinter 版本""" def __init__(self, system_manager: SystemInfoManager = None): self.system_manager = system_manager 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): """执行 shell 命令""" full_cmd_str = ' '.join(command_list) logger.info(f"执行命令: {full_cmd_str}") try: if self.system_manager: stdout, stderr = self.system_manager._run_command( command_list, root_privilege=root_privilege, check_output=True, input_data=input_data ) else: result = subprocess.run( ["sudo"] + command_list if root_privilege else command_list, stdout=subprocess.PIPE, stderr=subprocess.PIPE, encoding='utf-8', check=True, input=input_data, stdin=subprocess.PIPE if input_data else None ) stdout = result.stdout stderr = result.stderr 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: stderr_output = e.stderr.strip() if e.stderr else "" logger.error(f"{error_message} 命令: {full_cmd_str}") logger.error(f"退出码: {e.returncode}") logger.error(f"标准输出: {e.stdout.strip() if e.stdout else ''}") logger.error(f"标准错误: {stderr_output}") 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 should_suppress_dialog: logger.info(f"特定错误 '{stderr_output}' 匹配抑制条件，已抑制错误对话框。") elif show_dialog: messagebox.showerror("错误", f"{error_message}\n错误详情: {stderr_output}") return False, e.stdout if e.stdout else "", stderr_output except FileNotFoundError: logger.error(f"命令 '{command_list[0]}' 未找到。请确保已安装相关工具。") if show_dialog: messagebox.showerror("错误", f"命令 '{command_list[0]}' 未找到。请确保已安装相关工具。") return False, "", "Command not found." except Exception as e: logger.error(f"执行命令时发生未知错误: {e}") if show_dialog: messagebox.showerror("错误", f"执行命令时发生未知错误: {e}") return False, "", str(e) def create_pv(self, device_path): """创建物理卷""" if not device_path or not device_path.startswith('/dev/'): messagebox.showerror("错误", "无效的设备路径。") return False if not messagebox.askyesno("确认创建物理卷", f"您确定要在 {device_path} 上创建物理卷吗？\n" "此操作将擦除设备上的数据！", default=messagebox.NO): logger.info(f"用户取消了在 {device_path} 上创建物理卷的操作。") return False logger.info(f"尝试在 {device_path} 上创建物理卷。") success, stdout, stderr = self._execute_shell_command( ["pvcreate", "-y", device_path], f"在 {device_path} 上创建物理卷失败" ) if success: messagebox.showinfo("成功", f"物理卷已在 {device_path} 上成功创建。") return True else: if "partition" in stderr.lower() or "分区" in stderr: if messagebox.askyesno("警告", f"设备 {device_path} 似乎已有分区。\n" "是否继续擦除并创建物理卷？", default=messagebox.NO): success_wipe, _, _ = self._execute_shell_command( ["wipefs", "-a", device_path], f"擦除设备 {device_path} 失败" ) if success_wipe: success_pv, _, stderr_pv = self._execute_shell_command( ["pvcreate", "-y", device_path], f"在 {device_path} 上创建物理卷失败" ) if success_pv: messagebox.showinfo("成功", f"物理卷已在 {device_path} 上成功创建。") return True else: messagebox.showerror("错误", f"在 {device_path} 上创建物理卷失败: {stderr_pv}") else: messagebox.showinfo("信息", f"未在已分区设备 {device_path} 上创建物理卷。") return False def delete_pv(self, device_path): """删除物理卷""" if not device_path or not device_path.startswith('/dev/'): messagebox.showerror("错误", "无效的设备路径。") return False if not messagebox.askyesno("确认删除物理卷", f"您确定要删除物理卷 {device_path} 吗？", default=messagebox.NO): logger.info(f"用户取消了删除物理卷 {device_path} 的操作。") return False logger.info(f"尝试删除物理卷 {device_path}。") success, stdout, stderr = self._execute_shell_command( ["pvremove", "-y", device_path], f"删除物理卷 {device_path} 失败" ) if success: messagebox.showinfo("成功", f"物理卷 {device_path} 已成功删除。") return True return False def create_vg(self, vg_name, pvs): """创建卷组""" if not vg_name or not pvs: messagebox.showerror("错误", "无效的卷组名称或物理卷路径。") return False if not messagebox.askyesno("确认创建卷组", f"您确定要创建卷组 {vg_name} 吗？\n" f"使用物理卷: {', '.join(pvs)}", default=messagebox.NO): logger.info(f"用户取消了创建卷组 {vg_name} 的操作。") return False logger.info(f"尝试创建卷组 {vg_name}，使用物理卷: {pvs}") cmd = ["vgcreate", vg_name] + pvs success, stdout, stderr = self._execute_shell_command( cmd, f"创建卷组 {vg_name} 失败" ) if success: messagebox.showinfo("成功", f"卷组 {vg_name} 已成功创建。") return True return False def delete_vg(self, vg_name): """删除卷组""" if not vg_name: messagebox.showerror("错误", "无效的卷组名称。") return False if not messagebox.askyesno("确认删除卷组", f"您确定要删除卷组 {vg_name} 吗？\n" "此操作将删除该卷组及其所有逻辑卷！", default=messagebox.NO): logger.info(f"用户取消了删除卷组 {vg_name} 的操作。") return False logger.info(f"尝试删除卷组 {vg_name}。") success, stdout, stderr = self._execute_shell_command( ["vgremove", "-y", vg_name], f"删除卷组 {vg_name} 失败" ) if success: messagebox.showinfo("成功", f"卷组 {vg_name} 已成功删除。") return True return False def create_lv(self, lv_name, vg_name, size_gb, use_max_space=False): """创建逻辑卷""" if not lv_name or not vg_name: messagebox.showerror("错误", "无效的逻辑卷名称或卷组名称。") return False if not use_max_space and size_gb <= 0: messagebox.showerror("错误", "逻辑卷大小必须大于0。") return False size_str = "100%FREE" if use_max_space else f"{size_gb}G" if not messagebox.askyesno("确认创建逻辑卷", f"您确定要在卷组 {vg_name} 中创建逻辑卷 {lv_name} 吗？\n" f"大小: {size_str}", default=messagebox.NO): logger.info(f"用户取消了创建逻辑卷 {lv_name} 的操作。") return False logger.info(f"尝试在卷组 {vg_name} 中创建逻辑卷 {lv_name}，大小: {size_str}") cmd = ["lvcreate", "-y", "-n", lv_name, "-L", size_str, vg_name] if not use_max_space else \ ["lvcreate", "-y", "-n", lv_name, "-l", "100%FREE", vg_name] success, stdout, stderr = self._execute_shell_command( cmd, f"创建逻辑卷 {lv_name} 失败" ) if success: messagebox.showinfo("成功", f"逻辑卷 {lv_name} 已在卷组 {vg_name} 中成功创建。") return True return False def delete_lv(self, lv_name, vg_name): """删除逻辑卷""" if not lv_name or not vg_name: messagebox.showerror("错误", "无效的逻辑卷名称或卷组名称。") return False if not messagebox.askyesno("确认删除逻辑卷", f"您确定要删除逻辑卷 {vg_name}/{lv_name} 吗？\n" "此操作将删除该逻辑卷及其所有数据！", default=messagebox.NO): logger.info(f"用户取消了删除逻辑卷 {lv_name} 的操作。") return False logger.info(f"尝试删除逻辑卷 {vg_name}/{lv_name}。") success, stdout, stderr = self._execute_shell_command( ["lvremove", "-y", f"{vg_name}/{lv_name}"], f"删除逻辑卷 {vg_name}/{lv_name} 失败" ) if success: messagebox.showinfo("成功", f"逻辑卷 {vg_name}/{lv_name} 已成功删除。") return True return False def activate_lv(self, lv_name, vg_name): """激活逻辑卷""" if not lv_name or not vg_name: messagebox.showerror("错误", "无效的逻辑卷名称或卷组名称。") return False logger.info(f"尝试激活逻辑卷 {vg_name}/{lv_name}。") success, stdout, stderr = self._execute_shell_command( ["lvchange", "-ay", f"{vg_name}/{lv_name}"], f"激活逻辑卷 {vg_name}/{lv_name} 失败" ) if success: messagebox.showinfo("成功", f"逻辑卷 {vg_name}/{lv_name} 已成功激活。") return True return False def deactivate_lv(self, lv_name, vg_name): """停用逻辑卷""" if not lv_name or not vg_name: messagebox.showerror("错误", "无效的逻辑卷名称或卷组名称。") return False logger.info(f"尝试停用逻辑卷 {vg_name}/{lv_name}。") success, stdout, stderr = self._execute_shell_command( ["lvchange", "-an", f"{vg_name}/{lv_name}"], f"停用逻辑卷 {vg_name}/{lv_name} 失败" ) if success: messagebox.showinfo("成功", f"逻辑卷 {vg_name}/{lv_name} 已成功停用。") return True return False