177 lines
6.4 KiB
Python
177 lines
6.4 KiB
Python
#!/usr/bin/env python
|
||
# -*- coding: utf-8 -*-
|
||
"""
|
||
配置管理器
|
||
|
||
接管主窗口中所有配置读写逻辑：
|
||
- new_config() 清空所有面板配置
|
||
- load_config(file_path) 从 JSON 文件加载配置并回填面板
|
||
- save_config(file_path) 将当前配置保存为 JSON 文件
|
||
- get_current_config() 遍历 PanelFactory 收集配置（懒加载安全）
|
||
|
||
懒加载兼容原则：
|
||
- get_current_config() 仅遍历已加载面板，未加载面板返回空字典 {}
|
||
- 绝不为了拿配置而强行唤醒/渲染所有 Panel
|
||
- 如需全量配置（如保存），调用方应先执行 panel_factory.preload_all()
|
||
"""
|
||
|
||
import json
|
||
import os
|
||
from typing import Dict, Optional
|
||
|
||
from PyQt5.QtCore import QObject
|
||
from PyQt5.QtWidgets import QMessageBox, QFileDialog
|
||
|
||
from src.gui.core.event_bus import global_event_bus
|
||
|
||
|
||
class ConfigManager(QObject):
|
||
"""配置管理器。
|
||
|
||
用法::
|
||
|
||
cfg_mgr = ConfigManager(panel_factory, parent=self)
|
||
cfg_mgr.new_config() # 清空配置
|
||
cfg_mgr.load_config(path) # 加载 JSON
|
||
cfg_mgr.save_config(path) # 保存 JSON
|
||
config = cfg_mgr.get_current_config() # 收集当前配置
|
||
"""
|
||
|
||
def __init__(self, panel_factory, parent=None):
|
||
"""
|
||
Args:
|
||
panel_factory: PanelFactory 实例
|
||
parent: 父 QObject（用于弹窗定位）
|
||
"""
|
||
super().__init__(parent)
|
||
self._panel_factory = panel_factory
|
||
|
||
# ═══════════════════════════════════════════════════════════
|
||
# 公开 API
|
||
# ═══════════════════════════════════════════════════════════
|
||
|
||
def new_config(self):
|
||
"""清空所有面板配置（需用户确认）。"""
|
||
reply = QMessageBox.question(
|
||
self.parent(), "新建配置", "是否清空当前配置？",
|
||
QMessageBox.Yes | QMessageBox.No
|
||
)
|
||
if reply != QMessageBox.Yes:
|
||
return
|
||
|
||
for panel in self._panel_factory.get_loaded_panels().values():
|
||
if hasattr(panel, 'clear_config'):
|
||
panel.clear_config()
|
||
|
||
global_event_bus.publish('LogMessage', {
|
||
'message': '已清空配置',
|
||
'level': 'info',
|
||
})
|
||
|
||
def load_config(self, file_path: str = None):
|
||
"""从 JSON 文件加载配置并回填面板。
|
||
|
||
Args:
|
||
file_path: JSON 文件路径。若为 None，弹出文件选择对话框。
|
||
"""
|
||
if file_path is None:
|
||
file_path, _ = QFileDialog.getOpenFileName(
|
||
self.parent(), "加载配置", "",
|
||
"JSON Files (*.json);;All Files (*.*)"
|
||
)
|
||
if not file_path:
|
||
return
|
||
|
||
try:
|
||
with open(file_path, 'r', encoding='utf-8') as f:
|
||
config = json.load(f)
|
||
except Exception as e:
|
||
QMessageBox.critical(
|
||
self.parent(), "加载失败",
|
||
f"无法读取配置文件:\n{file_path}\n\n错误: {e}"
|
||
)
|
||
return
|
||
|
||
# ★ 验证 JSON 结构：必须是 dict 类型
|
||
if not isinstance(config, dict):
|
||
QMessageBox.critical(
|
||
self.parent(), "配置格式错误",
|
||
f"配置文件格式不正确（期望 JSON 对象，实际为 {type(config).__name__}）。\n请确认选择了正确的配置文件。"
|
||
)
|
||
return
|
||
|
||
# 回填已加载面板
|
||
loaded_count = 0
|
||
for step_id, panel in self._panel_factory.get_loaded_panels().items():
|
||
if step_id in config and hasattr(panel, 'set_config'):
|
||
try:
|
||
panel.set_config(config[step_id])
|
||
loaded_count += 1
|
||
except Exception as e:
|
||
global_event_bus.publish('LogMessage', {
|
||
'message': f'[警告] 面板 {step_id} 加载配置失败: {e}',
|
||
'level': 'warning',
|
||
})
|
||
|
||
global_event_bus.publish('LogMessage', {
|
||
'message': f'已加载配置: {file_path}（回填 {loaded_count} 个面板）',
|
||
'level': 'info',
|
||
})
|
||
|
||
def save_config(self, file_path: str = None):
|
||
"""将当前配置保存为 JSON 文件。
|
||
|
||
注意：保存前会强制加载所有面板（preload_all），确保配置完整。
|
||
|
||
Args:
|
||
file_path: 目标 JSON 文件路径。若为 None，弹出保存对话框。
|
||
"""
|
||
if file_path is None:
|
||
file_path, _ = QFileDialog.getSaveFileName(
|
||
self.parent(), "保存配置", "config.json",
|
||
"JSON Files (*.json);;All Files (*.*)"
|
||
)
|
||
if not file_path:
|
||
return
|
||
|
||
# 保存前强制加载所有面板，确保配置完整
|
||
self._panel_factory.preload_all()
|
||
config = self.get_current_config()
|
||
|
||
try:
|
||
with open(file_path, 'w', encoding='utf-8') as f:
|
||
json.dump(config, f, indent=2, ensure_ascii=False)
|
||
except Exception as e:
|
||
QMessageBox.critical(
|
||
self.parent(), "保存失败",
|
||
f"无法保存配置文件:\n{file_path}\n\n错误: {e}"
|
||
)
|
||
return
|
||
|
||
global_event_bus.publish('LogMessage', {
|
||
'message': f'已保存配置: {file_path}',
|
||
'level': 'info',
|
||
})
|
||
|
||
def get_current_config(self) -> Dict[str, dict]:
|
||
"""收集当前所有步骤的配置。
|
||
|
||
懒加载安全：仅遍历已加载面板，未加载面板返回空字典 {}。
|
||
绝不为了拿配置而强行唤醒/渲染所有 Panel。
|
||
|
||
Returns:
|
||
{step_id: panel_config_dict}
|
||
"""
|
||
config = {}
|
||
for step_id, panel in self._panel_factory.get_loaded_panels().items():
|
||
if hasattr(panel, 'get_config'):
|
||
try:
|
||
config[step_id] = panel.get_config()
|
||
except Exception as e:
|
||
config[step_id] = {}
|
||
global_event_bus.publish('LogMessage', {
|
||
'message': f'[警告] 面板 {step_id} 获取配置失败: {e}',
|
||
'level': 'warning',
|
||
})
|
||
return config
|