zj/diskmanager
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

fix bug45

Browse Source
This commit is contained in:
zj
2026-02-02 22:31:02 +08:00
parent ac809c08df
commit 3eecfc4db0
13 changed files with 651 additions and 290 deletions
Download Patch File Download Diff File Expand all files Collapse all files

362
disk_operations.py
View File

@@ -3,12 +3,13 @@ import logging
import re
import os
from PySide6.QtWidgets import QMessageBox, QInputDialog
from system_info import SystemInfoManager
logger = logging.getLogger(__name__)
class DiskOperations:
def __init__(self):
pass
def __init__(self, system_manager: SystemInfoManager): # NEW: 接收 system_manager 实例
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):
@@ -18,6 +19,7 @@ class DiskOperations:
: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)
"""
@@ -50,10 +52,17 @@ class DiskOperations:
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 or \
(isinstance(suppress_critical_dialog_on_stderr_match, tuple) and \
any(s in stderr_output for s in suppress_critical_dialog_on_stderr_match))):
# Determine if the error dialog should be suppressed by this function
should_suppress_dialog_here = 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_here = True
elif isinstance(suppress_critical_dialog_on_stderr_match, tuple):
if any(s in stderr_output for s in suppress_critical_dialog_on_stderr_match):
should_suppress_dialog_here = True
if should_suppress_dialog_here:
logger.info(f"错误信息 '{stderr_output}' 匹配抑制条件，不显示关键错误对话框。")
else:
QMessageBox.critical(None, "错误", f"{error_message}\n错误详情: {stderr_output}")
@@ -68,162 +77,173 @@ class DiskOperations:
return False, "", str(e)
def _add_to_fstab(self, device_path, mount_point, fstype, uuid):
"""
将设备的挂载信息添加到 /etc/fstab。
使用 UUID 识别设备以提高稳定性。
"""
if not uuid or not mount_point or not fstype:
logger.error(f"无法将设备 {device_path} 添加到 fstab：缺少 UUID/挂载点/文件系统类型。")
QMessageBox.warning(None, "警告", f"无法将设备 {device_path} 添加到 fstab：缺少 UUID/挂载点/文件系统类型。")
return False
"""
将设备的挂载信息添加到 /etc/fstab。
使用 UUID 识别设备以提高稳定性。
"""
if not uuid or not mount_point or not fstype:
logger.error(f"无法将设备 {device_path} 添加到 fstab：缺少 UUID/挂载点/文件系统类型。")
QMessageBox.warning(None, "警告", f"无法将设备 {device_path} 添加到 fstab：缺少 UUID/挂载点/文件系统类型。")
return False
fstab_entry = f"UUID={uuid} {mount_point} {fstype} defaults 0 2"
fstab_path = "/etc/fstab"
fstab_entry = f"UUID={uuid} {mount_point} {fstype} defaults 0 2"
fstab_path = "/etc/fstab"
try:
# 检查 fstab 中是否已存在相同 UUID 或挂载点的条目
with open(fstab_path, 'r') as f:
fstab_content = f.readlines()
try:
# 检查 fstab 中是否已存在相同 UUID 的条目
# 使用 _execute_shell_command 来读取 fstab，尽管通常不需要 sudo
# 这里为了简化，直接读取文件，但写入时使用 _execute_shell_command
with open(fstab_path, 'r') as f:
fstab_content = f.readlines()
for line in fstab_content:
if f"UUID={uuid}" in line:
logger.warning(f"设备 {device_path} (UUID={uuid}) 已存在于 fstab 中。跳过添加。")
QMessageBox.information(None, "信息", f"设备 {device_path} (UUID={uuid}) 已存在于 fstab 中。")
return True # 认为成功，因为目标已达成
# 检查挂载点是否已存在 (可能被其他设备使用)
if mount_point in line.split():
logger.warning(f"挂载点 {mount_point} 已存在于 fstab 中。跳过添加。")
QMessageBox.information(None, "信息", f"挂载点 {mount_point} 已存在于 fstab 中。")
return True # 认为成功，因为目标已达成
for line in fstab_content:
if f"UUID={uuid}" in line:
logger.warning(f"设备 {device_path} (UUID={uuid}) 已存在于 fstab 中。跳过添加。")
QMessageBox.information(None, "信息", f"设备 {device_path} (UUID={uuid}) 已存在于 fstab 中。")
return True # 认为成功，因为目标已达成
# 如果不存在，则追加到 fstab
with open(fstab_path, 'a') as f:
f.write(fstab_entry + '\n')
logger.info(f"已将 {fstab_entry} 添加到 {fstab_path}。")
QMessageBox.information(None, "成功", f"设备 {device_path} 已成功添加到 /etc/fstab。")
return True
except Exception as e:
logger.error(f"写入 {fstab_path} 失败: {e}")
QMessageBox.critical(None, "错误", f"写入 {fstab_path} 失败: {e}")
return False
# 如果不存在，则追加到 fstab
# 使用 _execute_shell_command for sudo write
success, _, stderr = self._execute_shell_command(
["sh", "-c", f"echo '{fstab_entry}' >> {fstab_path}"],
f"将 {device_path} 添加到 {fstab_path} 失败",
root_privilege=True
)
if success:
logger.info(f"已将 {fstab_entry} 添加到 {fstab_path}。")
QMessageBox.information(None, "成功", f"设备 {device_path} 已成功添加到 /etc/fstab。")
return True
else:
return False # Error handled by _execute_shell_command
except Exception as e:
logger.error(f"处理 {fstab_path} 失败: {e}")
QMessageBox.critical(None, "错误", f"处理 {fstab_path} 失败: {e}")
return False
def _remove_fstab_entry(self, device_path):
"""
从 /etc/fstab 中移除指定设备的条目。
此方法需要 SystemInfoManager 来获取设备的 UUID。
"""
success, stdout, stderr = self._execute_shell_command(
["lsblk", "-no", "UUID", device_path],
f"获取设备 {device_path} 的 UUID 失败",
root_privilege=False, # lsblk 通常不需要 sudo
suppress_critical_dialog_on_stderr_match=("找不到或无法访问", "No such device or address") # 匹配中英文错误
)
if not success:
logger.warning(f"无法获取设备 {device_path} 的 UUID，无法从 fstab 中移除。错误: {stderr}")
return False
"""
从 /etc/fstab 中移除指定设备的条目。
此方法需要 SystemInfoManager 来获取设备的 UUID。
"""
# NEW: 使用 system_manager 获取 UUID，它会处理路径解析
device_details = self.system_manager.get_device_details_by_path(device_path)
if not device_details or not device_details.get('uuid'):
logger.warning(f"无法获取设备 {device_path} 的 UUID，无法从 fstab 中移除。")
return False
uuid = device_details.get('uuid')
uuid = stdout.strip()
if not uuid:
logger.warning(f"设备 {device_path} 没有 UUID，无法从 fstab 中移除。")
return False
fstab_path = "/etc/fstab"
try:
with open(fstab_path, 'r') as f:
lines = f.readlines()
new_lines = []
removed = False
for line in lines:
if f"UUID={uuid}" in line:
logger.info(f"从 {fstab_path} 中移除了条目: {line.strip()}")
removed = True
else:
new_lines.append(line)
if removed:
# 写入临时文件，然后替换原文件
with open(fstab_path + ".tmp", 'w') as f_tmp:
f_tmp.writelines(new_lines)
os.rename(fstab_path + ".tmp", fstab_path)
logger.info(f"已从 {fstab_path} 中移除 UUID={uuid} 的条目。")
fstab_path = "/etc/fstab"
# Use sed for robust removal with sudo
# suppress_critical_dialog_on_stderr_match is added to handle cases where fstab might not exist
# or sed reports no changes, which is not a critical error for removal.
command = ["sed", "-i", f"/UUID={uuid}/d", fstab_path]
success, _, stderr = self._execute_shell_command(
command,
f"从 {fstab_path} 中删除 UUID={uuid} 的条目失败",
root_privilege=True,
suppress_critical_dialog_on_stderr_match=(
f"sed: {fstab_path}: No such file or directory", # English
f"sed: {fstab_path}: 没有那个文件或目录", # Chinese
"no changes were made" # if sed finds nothing to delete
)
)
if success:
logger.info(f"已从 {fstab_path} 中删除 UUID={uuid} 的条目。")
return True
else:
logger.info(f"fstab 中未找到 UUID={uuid} 的条目。")
return False
except Exception as e:
logger.error(f"修改 {fstab_path} 失败: {e}")
QMessageBox.critical(None, "错误", f"修改 {fstab_path} 失败: {e}")
return False
# If sed failed, it might be because the entry wasn't found, which is fine.
# _execute_shell_command would have suppressed the dialog if it matched the suppress_critical_dialog_on_stderr_match.
# So, if we reach here, it's either a real error (dialog shown by _execute_shell_command)
# or a suppressed "no changes" type of error. In both cases, if no real error, we return True.
if any(s in stderr for s in (
f"sed: {fstab_path}: No such file or directory",
f"sed: {fstab_path}: 没有那个文件或目录",
"no changes were made",
"No such file or directory" # more general check
)):
logger.info(f"fstab 中未找到 UUID={uuid} 的条目，无需删除。")
return True # Consider it a success if the entry is not there
return False # Other errors are already handled by _execute_shell_command
def mount_partition(self, device_path, mount_point, add_to_fstab=False):
"""
挂载指定设备到指定挂载点。
:param device_path: 要挂载的设备路径。
:param mount_point: 挂载点。
:param add_to_fstab: 是否添加到 /etc/fstab。
:return: True 如果成功，否则 False。
"""
if not os.path.exists(mount_point):
try:
os.makedirs(mount_point)
logger.info(f"创建挂载点目录: {mount_point}")
except OSError as e:
QMessageBox.critical(None, "错误", f"创建挂载点目录 {mount_point} 失败: {e}")
logger.error(f"创建挂载点目录 {mount_point} 失败: {e}")
"""
挂载指定设备到指定挂载点。
:param device_path: 要挂载的设备路径。
:param mount_point: 挂载点。
:param add_to_fstab: 是否添加到 /etc/fstab。
:return: True 如果成功，否则 False。
"""
if not os.path.exists(mount_point):
try:
os.makedirs(mount_point)
logger.info(f"创建挂载点目录: {mount_point}")
except OSError as e:
QMessageBox.critical(None, "错误", f"创建挂载点目录 {mount_point} 失败: {e}")
logger.error(f"创建挂载点目录 {mount_point} 失败: {e}")
return False
logger.info(f"尝试挂载设备 {device_path} 到 {mount_point}。")
success, _, stderr = self._execute_shell_command(
["mount", device_path, mount_point],
f"挂载设备 {device_path} 失败"
)
if success:
QMessageBox.information(None, "成功", f"设备 {device_path} 已成功挂载到 {mount_point}。")
if add_to_fstab:
# NEW: 使用 self.system_manager 获取 fstype 和 UUID
device_details = self.system_manager.get_device_details_by_path(device_path)
if device_details:
fstype = device_details.get('fstype')
uuid = device_details.get('uuid')
if fstype and uuid:
self._add_to_fstab(device_path, mount_point, fstype, uuid)
else:
logger.error(f"无法获取设备 {device_path} 的文件系统类型或 UUID 以添加到 fstab。")
QMessageBox.warning(None, "警告", f"设备 {device_path} 已挂载，但无法获取文件系统类型或 UUID 以添加到 fstab。")
else:
logger.error(f"无法获取设备 {device_path} 的详细信息以添加到 fstab。")
QMessageBox.warning(None, "警告", f"设备 {device_path} 已挂载，但无法获取详细信息以添加到 fstab。")
return True
else:
return False
logger.info(f"尝试挂载设备 {device_path} 到 {mount_point}。")
success, _, stderr = self._execute_shell_command(
["mount", device_path, mount_point],
f"挂载设备 {device_path} 失败"
)
if success:
QMessageBox.information(None, "成功", f"设备 {device_path} 已成功挂载到 {mount_point}。")
if add_to_fstab:
# 为了获取 fstype 和 UUID，需要再次调用 lsblk
success_details, stdout_details, stderr_details = self._execute_shell_command(
["lsblk", "-no", "FSTYPE,UUID", device_path],
f"获取设备 {device_path} 的文件系统类型和 UUID 失败",
root_privilege=False,
suppress_critical_dialog_on_stderr_match=("找不到或无法访问", "No such device or address")
)
if success_details:
fstype, uuid = stdout_details.strip().split()
self._add_to_fstab(device_path, mount_point, fstype, uuid)
else:
logger.error(f"无法获取设备 {device_path} 的详细信息以添加到 fstab: {stderr_details}")
QMessageBox.warning(None, "警告", f"设备 {device_path} 已挂载，但无法获取详细信息以添加到 fstab。")
return True
else:
return False
def unmount_partition(self, device_path, show_dialog_on_error=True):
"""
卸载指定设备。
:param device_path: 要卸载的设备路径。
:param show_dialog_on_error: 是否在发生错误时显示对话框。
:return: True 如果成功，否则 False。
"""
logger.info(f"尝试卸载设备 {device_path}。")
try:
"""
卸载指定设备。
:param device_path: 要卸载的设备路径。
:param show_dialog_on_error: 是否在发生错误时显示对话框。
:return: True 如果成功，否则 False。
"""
logger.info(f"尝试卸载设备 {device_path}。")
# 定义表示设备已未挂载的错误信息（中英文）
# 增加了 "未指定挂载点"
already_unmounted_errors = ("not mounted", "未挂载", "未指定挂载点")
# 调用 _execute_shell_command，并告诉它在遇到“已未挂载”错误时，不要弹出其自身的关键错误对话框。
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
suppress_critical_dialog_on_stderr_match=already_unmounted_errors
)
if success:
# 如果命令成功执行，则卸载成功。
QMessageBox.information(None, "成功", f"设备 {device_path} 已成功卸载。")
return True
else:
if show_dialog_on_error:
QMessageBox.critical(None, "错误", f"卸载设备 {device_path} 失败。\n错误详情: {stderr}")
return False
except Exception as e:
logger.error(f"卸载设备 {device_path} 时发生异常: {e}")
if show_dialog_on_error:
QMessageBox.critical(None, "错误", f"卸载设备 {device_path} 时发生异常: {e}")
return False
# 如果命令失败，检查是否是因为设备已经未挂载或挂载点未指定。
is_already_unmounted_error = any(s in stderr for s in already_unmounted_errors)
if is_already_unmounted_error:
logger.info(f"设备 {device_path} 已经处于未挂载状态（或挂载点未指定），无需重复卸载。")
# 这种情况下，操作结果符合预期（设备已是未挂载），不弹出对话框，并返回 True。
return True
else:
# 对于其他类型的卸载失败（例如设备忙、权限不足等），
# _execute_shell_command 应该已经弹出了关键错误对话框
# （因为它没有匹配到 `already_unmounted_errors` 进行抑制）。
# 所以，这里我们不需要再次弹出对话框，直接返回 False。
return False
def get_disk_free_space_info_mib(self, disk_path, total_disk_mib):
"""
@@ -234,13 +254,19 @@ class DiskOperations:
如果磁盘有分区表但没有空闲空间，返回 (None, None)。
"""
logger.debug(f"尝试获取磁盘 {disk_path} 的空闲空间信息 (MiB)。")
# suppress_critical_dialog_on_stderr_match is added to handle cases where parted might complain about
# an unrecognized disk label, which is expected for a fresh disk.
success, stdout, stderr = self._execute_shell_command(
["parted", "-s", disk_path, "unit", "MiB", "print", "free"],
f"获取磁盘 {disk_path} 分区信息失败",
root_privilege=True
root_privilege=True,
suppress_critical_dialog_on_stderr_match=("无法辨识的磁盘卷标", "unrecognized disk label")
)
if not success:
# If parted failed and it wasn't due to an unrecognized label (handled by suppress_critical_dialog_on_stderr_match),
# then _execute_shell_command would have shown a dialog.
# We just log and return None.
logger.error(f"获取磁盘 {disk_path} 空闲空间信息失败: {stderr}")
return None, None
@@ -294,23 +320,31 @@ class DiskOperations:
# 1. 检查磁盘是否有分区表
has_partition_table = False
try:
success_check, stdout_check, stderr_check = self._execute_shell_command(
["parted", "-s", disk_path, "print"],
f"检查磁盘 {disk_path} 分区表失败",
suppress_critical_dialog_on_stderr_match=("无法辨识的磁盘卷标", "unrecognized disk label"),
root_privilege=True
)
if success_check:
if "Partition Table: unknown" not in stdout_check and "分区表：unknown" not in stdout_check:
has_partition_table = True
else:
logger.info(f"parted print 报告磁盘 {disk_path} 的分区表为 'unknown'。")
# Use suppress_critical_dialog_on_stderr_match for "unrecognized disk label" when checking
success_check, stdout_check, stderr_check = self._execute_shell_command(
["parted", "-s", disk_path, "print"],
f"检查磁盘 {disk_path} 分区表失败",
suppress_critical_dialog_on_stderr_match=("无法辨识的磁盘卷标", "unrecognized disk label"),
root_privilege=True
)
# If success_check is False, it means _execute_shell_command encountered an error.
# If that error was "unrecognized disk label", it was suppressed, and we treat it as no partition table.
# If it was another error, a dialog was shown by _execute_shell_command, and we should stop.
if not success_check:
# Check if the failure was due to "unrecognized disk label" (which means no partition table)
if any(s in stderr_check for s in ("无法辨识的磁盘卷标", "unrecognized disk label")):
logger.info(f"磁盘 {disk_path} 没有可识别的分区表。")
has_partition_table = False # Explicitly set to False
else:
logger.info(f"parted print 命令失败，但可能不是因为没有分区表，错误信息: {stderr_check}")
except Exception as e:
logger.error(f"检查磁盘 {disk_path} 分区表时发生异常: {e}")
pass
logger.error(f"检查磁盘 {disk_path} 分区表时发生非预期的错误: {stderr_check}")
return False # Other critical error, stop operation.
else: # success_check is True
if "Partition Table: unknown" not in stdout_check and "分区表：unknown" not in stdout_check:
has_partition_table = True
else:
logger.info(f"parted print 报告磁盘 {disk_path} 的分区表为 'unknown'。")
has_partition_table = False
actual_start_mib_for_parted = 0.0
@@ -374,12 +408,12 @@ class DiskOperations:
create_cmd,
f"在 {disk_path} 上创建分区失败"
)
if not success:
if success:
QMessageBox.information(None, "成功", f"在 {disk_path} 上成功创建了 {size_for_log} 的分区。")
return True
else:
return False
QMessageBox.information(None, "成功", f"在 {disk_path} 上成功创建了 {size_for_log} 的分区。")
return True
def delete_partition(self, device_path):
"""
删除指定分区。
@@ -394,9 +428,12 @@ class DiskOperations:
return False
# 尝试卸载分区
self.unmount_partition(device_path, show_dialog_on_error=False) # 静默卸载
# The unmount_partition method now handles "not mounted" gracefully without dialog and returns True.
# So, we just call it.
self.unmount_partition(device_path, show_dialog_on_error=False) # show_dialog_on_error=False means no dialog for other errors too.
# 从 fstab 中移除条目
# _remove_fstab_entry also handles "not found" gracefully.
self._remove_fstab_entry(device_path)
# 获取父磁盘和分区号
@@ -472,4 +509,3 @@ class DiskOperations:
return True
else:
return False