561 lines
19 KiB
Python
561 lines
19 KiB
Python
"""
|
|
主窗口
|
|
应用程序的主界面
|
|
"""
|
|
from PyQt5.QtWidgets import (
|
|
QMainWindow, QWidget, QVBoxLayout, QHBoxLayout,
|
|
QTabWidget, QStatusBar, QMenuBar, QMenu, QAction,
|
|
QMessageBox, QFileDialog, QToolBar
|
|
)
|
|
from PyQt5.QtCore import Qt, pyqtSignal
|
|
from PyQt5.QtGui import QIcon, QKeySequence
|
|
|
|
from config import Config
|
|
from controllers.field_controller import FieldController
|
|
from controllers.message_controller import MessageController
|
|
from controllers.mapping_controller import MappingController
|
|
from controllers.codegen_controller import CodeGenController
|
|
from utils.logger import get_logger
|
|
|
|
logger = get_logger(__name__)
|
|
|
|
|
|
class MainWindow(QMainWindow):
|
|
"""主窗口类"""
|
|
|
|
def __init__(self):
|
|
super().__init__()
|
|
|
|
# 初始化控制器
|
|
self.field_controller = FieldController()
|
|
self.message_controller = MessageController()
|
|
self.mapping_controller = MappingController()
|
|
self.codegen_controller = CodeGenController()
|
|
|
|
self.status_bar = QStatusBar(parent=self)
|
|
# 存储当前选中的对象
|
|
self.current_field = None
|
|
self.current_message = None
|
|
self.current_mapping = None
|
|
|
|
# 初始化UI
|
|
self.init_ui()
|
|
|
|
logger.info("Main window initialized")
|
|
|
|
def init_ui(self):
|
|
"""初始化用户界面"""
|
|
# 设置窗口属性
|
|
self.setWindowTitle(f"{Config.APP_NAME} v{Config.APP_VERSION}")
|
|
self.setGeometry(100, 100, Config.WINDOW_WIDTH, Config.WINDOW_HEIGHT)
|
|
self.setMinimumSize(Config.WINDOW_MIN_WIDTH, Config.WINDOW_MIN_HEIGHT)
|
|
|
|
# 创建中心部件
|
|
central_widget = QWidget()
|
|
self.setCentralWidget(central_widget)
|
|
|
|
# 创建主布局
|
|
main_layout = QVBoxLayout(central_widget)
|
|
main_layout.setContentsMargins(0, 0, 0, 0)
|
|
|
|
# 创建工具栏
|
|
self.create_toolbar()
|
|
|
|
# 创建标签页
|
|
self.tab_widget = QTabWidget()
|
|
self.tab_widget.currentChanged.connect(self.on_tab_changed)
|
|
main_layout.addWidget(self.tab_widget)
|
|
|
|
# 添加各个功能标签页
|
|
self.create_field_tab()
|
|
self.create_message_tab()
|
|
self.create_mapping_tab()
|
|
self.create_codegen_tab()
|
|
|
|
# 创建菜单栏
|
|
self.create_menu_bar()
|
|
|
|
# 创建状态栏
|
|
self.status_bar = QStatusBar()
|
|
self.setStatusBar(self.status_bar)
|
|
self.status_bar.showMessage("就绪")
|
|
|
|
# 加载样式表
|
|
self.load_stylesheet()
|
|
|
|
def create_toolbar(self):
|
|
"""创建工具栏"""
|
|
toolbar = QToolBar("主工具栏")
|
|
toolbar.setMovable(False)
|
|
self.addToolBar(toolbar)
|
|
|
|
# 新建操作
|
|
new_field_action = QAction("新建字段", self)
|
|
new_field_action.setShortcut("Ctrl+N")
|
|
new_field_action.setStatusTip("创建新字段")
|
|
new_field_action.triggered.connect(self.on_new_field)
|
|
toolbar.addAction(new_field_action)
|
|
|
|
new_message_action = QAction("新建消息", self)
|
|
new_message_action.setStatusTip("创建新消息")
|
|
new_message_action.triggered.connect(self.on_new_message)
|
|
toolbar.addAction(new_message_action)
|
|
|
|
new_mapping_action = QAction("新建映射", self)
|
|
new_mapping_action.setStatusTip("创建新映射")
|
|
new_mapping_action.triggered.connect(self.on_new_mapping)
|
|
toolbar.addAction(new_mapping_action)
|
|
|
|
toolbar.addSeparator()
|
|
|
|
# 代码生成
|
|
codegen_action = QAction("生成代码", self)
|
|
codegen_action.setShortcut("Ctrl+G")
|
|
codegen_action.setStatusTip("生成转换代码")
|
|
codegen_action.triggered.connect(self.on_generate_code)
|
|
toolbar.addAction(codegen_action)
|
|
|
|
toolbar.addSeparator()
|
|
|
|
# 导入导出
|
|
import_action = QAction("导入", self)
|
|
import_action.setShortcut("Ctrl+I")
|
|
import_action.triggered.connect(self.on_import)
|
|
toolbar.addAction(import_action)
|
|
|
|
export_action = QAction("导出", self)
|
|
export_action.setShortcut("Ctrl+E")
|
|
export_action.triggered.connect(self.on_export)
|
|
toolbar.addAction(export_action)
|
|
|
|
def create_field_tab(self):
|
|
"""创建字段管理标签页"""
|
|
from ui.widgets.field_tree_widget import FieldManagementWidget
|
|
|
|
self.field_widget = FieldManagementWidget(self.field_controller)
|
|
self.field_widget.field_selected.connect(self.on_field_selected)
|
|
self.tab_widget.addTab(self.field_widget, "字段管理")
|
|
|
|
def create_message_tab(self):
|
|
"""创建消息管理标签页"""
|
|
from ui.widgets.message_tree_widget import MessageManagementWidget
|
|
|
|
self.message_widget = MessageManagementWidget(self.message_controller)
|
|
self.message_widget.message_selected.connect(self.on_message_selected)
|
|
self.tab_widget.addTab(self.message_widget, "消息管理")
|
|
|
|
def create_mapping_tab(self):
|
|
"""创建映射管理标签页"""
|
|
from ui.widgets.mapping_graph_widget import MappingManagementWidget
|
|
|
|
self.mapping_widget = MappingManagementWidget(
|
|
self.mapping_controller,
|
|
self.field_controller
|
|
)
|
|
self.mapping_widget.mapping_selected.connect(self.on_mapping_selected)
|
|
self.tab_widget.addTab(self.mapping_widget, "映射管理")
|
|
|
|
def create_codegen_tab(self):
|
|
"""创建代码生成标签页"""
|
|
from ui.widgets.codegen_widget import CodeGenWidget
|
|
|
|
self.codegen_widget = CodeGenWidget(
|
|
self.codegen_controller,
|
|
self.message_controller
|
|
)
|
|
self.tab_widget.addTab(self.codegen_widget, "代码生成")
|
|
|
|
def create_menu_bar(self):
|
|
"""创建菜单栏"""
|
|
menubar = self.menuBar()
|
|
|
|
# 文件菜单
|
|
file_menu = menubar.addMenu("文件(&F)")
|
|
|
|
new_menu = file_menu.addMenu("新建(&N)")
|
|
|
|
new_field_action = QAction("字段(&F)", self)
|
|
new_field_action.setShortcut("Ctrl+N")
|
|
new_field_action.triggered.connect(self.on_new_field)
|
|
new_menu.addAction(new_field_action)
|
|
|
|
new_message_action = QAction("消息(&M)", self)
|
|
new_message_action.triggered.connect(self.on_new_message)
|
|
new_menu.addAction(new_message_action)
|
|
|
|
new_mapping_action = QAction("映射(&P)", self)
|
|
new_mapping_action.triggered.connect(self.on_new_mapping)
|
|
new_menu.addAction(new_mapping_action)
|
|
|
|
file_menu.addSeparator()
|
|
|
|
import_action = QAction("导入(&I)", self)
|
|
import_action.setShortcut("Ctrl+I")
|
|
import_action.triggered.connect(self.on_import)
|
|
file_menu.addAction(import_action)
|
|
|
|
export_action = QAction("导出(&E)", self)
|
|
export_action.setShortcut("Ctrl+E")
|
|
export_action.triggered.connect(self.on_export)
|
|
file_menu.addAction(export_action)
|
|
|
|
file_menu.addSeparator()
|
|
|
|
exit_action = QAction("退出(&X)", self)
|
|
exit_action.setShortcut("Ctrl+Q")
|
|
exit_action.triggered.connect(self.close)
|
|
file_menu.addAction(exit_action)
|
|
|
|
# 编辑菜单
|
|
edit_menu = menubar.addMenu("编辑(&E)")
|
|
|
|
delete_action = QAction("删除(&D)", self)
|
|
delete_action.setShortcut(QKeySequence.Delete)
|
|
delete_action.triggered.connect(self.on_delete_current)
|
|
edit_menu.addAction(delete_action)
|
|
|
|
edit_menu.addSeparator()
|
|
|
|
refresh_action = QAction("刷新(&R)", self)
|
|
refresh_action.setShortcut("F5")
|
|
refresh_action.triggered.connect(self.on_refresh)
|
|
edit_menu.addAction(refresh_action)
|
|
|
|
# 视图菜单
|
|
view_menu = menubar.addMenu("视图(&V)")
|
|
|
|
# 工具菜单
|
|
tools_menu = menubar.addMenu("工具(&T)")
|
|
|
|
codegen_action = QAction("生成代码(&G)", self)
|
|
codegen_action.setShortcut("Ctrl+G")
|
|
codegen_action.triggered.connect(self.on_generate_code)
|
|
tools_menu.addAction(codegen_action)
|
|
|
|
tools_menu.addSeparator()
|
|
|
|
validate_action = QAction("验证数据(&V)", self)
|
|
validate_action.triggered.connect(self.on_validate_data)
|
|
tools_menu.addAction(validate_action)
|
|
|
|
# 帮助菜单
|
|
help_menu = menubar.addMenu("帮助(&H)")
|
|
|
|
user_guide_action = QAction("用户指南(&U)", self)
|
|
user_guide_action.setShortcut("F1")
|
|
user_guide_action.triggered.connect(self.on_user_guide)
|
|
help_menu.addAction(user_guide_action)
|
|
|
|
help_menu.addSeparator()
|
|
|
|
about_action = QAction("关于(&A)", self)
|
|
about_action.triggered.connect(self.on_about)
|
|
help_menu.addAction(about_action)
|
|
|
|
def load_stylesheet(self):
|
|
"""加载样式表"""
|
|
try:
|
|
style_path = Config.BASE_DIR / "ui" / "styles.qss"
|
|
if style_path.exists():
|
|
with open(style_path, 'r', encoding='utf-8') as f:
|
|
self.setStyleSheet(f.read())
|
|
logger.info("Stylesheet loaded")
|
|
except Exception as e:
|
|
logger.warning(f"Failed to load stylesheet: {e}")
|
|
|
|
# 槽函数
|
|
|
|
def on_tab_changed(self, index):
|
|
"""标签页切换"""
|
|
tab_names = ["字段管理", "消息管理", "映射管理", "代码生成"]
|
|
if index < len(tab_names):
|
|
self.status_bar.showMessage(f"当前: {tab_names[index]}")
|
|
|
|
def on_field_selected(self, field):
|
|
"""字段被选中"""
|
|
self.current_field = field
|
|
self.status_bar.showMessage(f"选中字段: {field.full_name}")
|
|
|
|
def on_message_selected(self, message):
|
|
"""消息被选中"""
|
|
self.current_message = message
|
|
self.status_bar.showMessage(f"选中消息: {message.full_name}")
|
|
|
|
def on_mapping_selected(self, mapping):
|
|
"""映射被选中"""
|
|
self.current_mapping = mapping
|
|
self.status_bar.showMessage(f"选中映射")
|
|
|
|
def on_new_field(self):
|
|
"""新建字段"""
|
|
from ui.dialogs.field_dialog import FieldDialog
|
|
|
|
dialog = FieldDialog(self)
|
|
dialog.field_saved.connect(self.on_field_saved)
|
|
dialog.exec_()
|
|
|
|
def on_new_message(self):
|
|
"""新建消息"""
|
|
from ui.dialogs.message_dialog import MessageDialog
|
|
|
|
dialog = MessageDialog(self)
|
|
dialog.message_saved.connect(self.on_message_saved)
|
|
dialog.exec_()
|
|
|
|
def on_new_mapping(self):
|
|
"""新建映射"""
|
|
from ui.dialogs.mapping_dialog import MappingDialog
|
|
|
|
dialog = MappingDialog(self)
|
|
dialog.mapping_saved.connect(self.on_mapping_saved)
|
|
dialog.exec_()
|
|
|
|
def on_field_saved(self, field):
|
|
"""字段保存后"""
|
|
success, msg, field_id = self.field_controller.create_field(field.to_dict())
|
|
if success:
|
|
QMessageBox.information(self, "成功", msg)
|
|
self.field_widget.load_fields()
|
|
else:
|
|
QMessageBox.warning(self, "失败", msg)
|
|
|
|
def on_message_saved(self, message):
|
|
"""消息保存后"""
|
|
success, msg, message_id = self.message_controller.create_message(message.to_dict())
|
|
if success:
|
|
QMessageBox.information(self, "成功", msg)
|
|
self.message_widget.load_messages()
|
|
else:
|
|
QMessageBox.warning(self, "失败", msg)
|
|
|
|
def on_mapping_saved(self, mapping):
|
|
"""映射保存后"""
|
|
from models.mapping import OperatorMapping, CodeMapping
|
|
|
|
if isinstance(mapping, OperatorMapping):
|
|
from services.mapping_service import MappingService
|
|
service = MappingService()
|
|
success, msg, mapping_id = service.create_operator_mapping(mapping)
|
|
elif isinstance(mapping, CodeMapping):
|
|
from services.mapping_service import MappingService
|
|
service = MappingService()
|
|
success, msg, mapping_id = service.create_code_mapping(mapping)
|
|
else:
|
|
success, msg = False, "未知的映射类型"
|
|
|
|
if success:
|
|
QMessageBox.information(self, "成功", msg)
|
|
self.mapping_widget.load_mappings()
|
|
else:
|
|
QMessageBox.warning(self, "失败", msg)
|
|
|
|
def on_delete_current(self):
|
|
"""删除当前选中项"""
|
|
current_tab = self.tab_widget.currentIndex()
|
|
|
|
if current_tab == 0: # 字段管理
|
|
if self.current_field:
|
|
reply = QMessageBox.question(
|
|
self,
|
|
"确认删除",
|
|
f"确定要删除字段 '{self.current_field.full_name}' 吗？",
|
|
QMessageBox.Yes | QMessageBox.No
|
|
)
|
|
if reply == QMessageBox.Yes:
|
|
success, msg = self.field_controller.delete_field(self.current_field.id)
|
|
if success:
|
|
QMessageBox.information(self, "成功", msg)
|
|
self.field_widget.load_fields()
|
|
else:
|
|
QMessageBox.warning(self, "失败", msg)
|
|
|
|
elif current_tab == 1: # 消息管理
|
|
if self.current_message:
|
|
reply = QMessageBox.question(
|
|
self,
|
|
"确认删除",
|
|
f"确定要删除消息 '{self.current_message.full_name}' 吗？",
|
|
QMessageBox.Yes | QMessageBox.No
|
|
)
|
|
if reply == QMessageBox.Yes:
|
|
success, msg = self.message_controller.delete_message(self.current_message.id)
|
|
if success:
|
|
QMessageBox.information(self, "成功", msg)
|
|
self.message_widget.load_messages()
|
|
else:
|
|
QMessageBox.warning(self, "失败", msg)
|
|
|
|
def on_refresh(self):
|
|
"""刷新当前标签页"""
|
|
current_tab = self.tab_widget.currentIndex()
|
|
|
|
if current_tab == 0:
|
|
self.field_widget.load_fields()
|
|
elif current_tab == 1:
|
|
self.message_widget.load_messages()
|
|
elif current_tab == 2:
|
|
self.mapping_widget.load_mappings()
|
|
|
|
self.status_bar.showMessage("已刷新")
|
|
|
|
def on_generate_code(self):
|
|
"""生成代码"""
|
|
from ui.dialogs.codegen_dialog import CodeGenDialog
|
|
|
|
# 检查是否有选中的消息
|
|
source_message = None
|
|
target_message = None
|
|
|
|
if self.current_message:
|
|
reply = QMessageBox.question(
|
|
self,
|
|
"选择消息",
|
|
f"是否使用当前选中的消息 '{self.current_message.full_name}' 作为源消息？",
|
|
QMessageBox.Yes | QMessageBox.No
|
|
)
|
|
if reply == QMessageBox.Yes:
|
|
source_message = self.current_message
|
|
|
|
dialog = CodeGenDialog(self, source_message, target_message)
|
|
dialog.code_generated.connect(self.on_code_generated)
|
|
dialog.exec_()
|
|
|
|
def on_code_generated(self, files):
|
|
"""代码生成完成"""
|
|
self.status_bar.showMessage(f"代码生成成功，共 {len(files)} 个文件")
|
|
|
|
def on_import(self):
|
|
"""导入数据"""
|
|
file_path, selected_filter = QFileDialog.getOpenFileName(
|
|
self,
|
|
"导入数据",
|
|
"",
|
|
"JSON文件 (*.json);;CSV文件 (*.csv);;XML文件 (*.xml);;所有文件 (*.*)"
|
|
)
|
|
|
|
if file_path:
|
|
# 根据文件类型确定格式
|
|
if file_path.endswith('.json'):
|
|
format_type = 'json'
|
|
elif file_path.endswith('.csv'):
|
|
format_type = 'csv'
|
|
elif file_path.endswith('.xml'):
|
|
format_type = 'xml'
|
|
else:
|
|
QMessageBox.warning(self, "警告", "不支持的文件格式")
|
|
return
|
|
|
|
# 导入字段
|
|
success, msg, fields = self.field_controller.import_fields(file_path, format_type)
|
|
|
|
if success and fields:
|
|
# 批量创建
|
|
success_count, fail_count, errors = self.field_controller.batch_create_fields(fields)
|
|
|
|
result_msg = f"导入完成\n成功: {success_count}\n失败: {fail_count}"
|
|
if errors:
|
|
result_msg += f"\n\n错误信息:\n" + "\n".join(errors[:5])
|
|
|
|
QMessageBox.information(self, "导入结果", result_msg)
|
|
self.field_widget.load_fields()
|
|
else:
|
|
QMessageBox.warning(self, "导入失败", msg)
|
|
|
|
def on_export(self):
|
|
"""导出数据"""
|
|
file_path, selected_filter = QFileDialog.getSaveFileName(
|
|
self,
|
|
"导出数据",
|
|
"",
|
|
"JSON文件 (*.json);;CSV文件 (*.csv);;XML文件 (*.xml)"
|
|
)
|
|
|
|
if file_path:
|
|
# 根据文件类型确定格式
|
|
if file_path.endswith('.json'):
|
|
format_type = 'json'
|
|
elif file_path.endswith('.csv'):
|
|
format_type = 'csv'
|
|
elif file_path.endswith('.xml'):
|
|
format_type = 'xml'
|
|
else:
|
|
format_type = 'json'
|
|
file_path += '.json'
|
|
|
|
# 获取所有字段
|
|
fields, _ = self.field_controller.get_all_fields(page=1, page_size=10000)
|
|
|
|
if fields:
|
|
success, msg = self.field_controller.export_fields(
|
|
fields,
|
|
file_path,
|
|
format_type
|
|
)
|
|
|
|
if success:
|
|
QMessageBox.information(self, "成功", msg)
|
|
else:
|
|
QMessageBox.warning(self, "失败", msg)
|
|
else:
|
|
QMessageBox.warning(self, "警告", "没有可导出的数据")
|
|
|
|
def on_validate_data(self):
|
|
"""验证数据"""
|
|
# 验证映射图
|
|
graph = self.mapping_controller.get_mapping_graph()
|
|
if graph:
|
|
from services.graph_service import GraphService
|
|
graph_service = GraphService()
|
|
is_valid, errors = graph_service.validate_graph(graph)
|
|
|
|
if is_valid:
|
|
QMessageBox.information(self, "验证结果", "数据验证通过，未发现问题")
|
|
else:
|
|
error_msg = "发现以下问题:\n\n" + "\n".join(errors)
|
|
QMessageBox.warning(self, "验证结果", error_msg)
|
|
else:
|
|
QMessageBox.information(self, "提示", "映射图为空")
|
|
|
|
def on_user_guide(self):
|
|
"""显示用户指南"""
|
|
QMessageBox.information(
|
|
self,
|
|
"用户指南",
|
|
"用户指南功能\n\n"
|
|
"详细文档请查看 docs/user_guide.md"
|
|
)
|
|
|
|
def on_about(self):
|
|
"""显示关于对话框"""
|
|
QMessageBox.about(
|
|
self,
|
|
"关于",
|
|
f"<h2>{Config.APP_NAME}</h2>"
|
|
f"<p>版本: {Config.APP_VERSION}</p>"
|
|
f"<p>作者: {Config.APP_AUTHOR}</p>"
|
|
f"<p>用于实现不同系统之间的协议转换和互操作性测试</p>"
|
|
f"<p><b>功能特性:</b></p>"
|
|
f"<ul>"
|
|
f"<li>字段管理</li>"
|
|
f"<li>消息定义</li>"
|
|
f"<li>映射建模</li>"
|
|
f"<li>代码生成</li>"
|
|
f"</ul>"
|
|
)
|
|
|
|
def closeEvent(self, event):
|
|
"""窗口关闭事件"""
|
|
reply = QMessageBox.question(
|
|
self,
|
|
"确认退出",
|
|
"确定要退出应用程序吗？",
|
|
QMessageBox.Yes | QMessageBox.No,
|
|
QMessageBox.No
|
|
)
|
|
|
|
if reply == QMessageBox.Yes:
|
|
logger.info("Application closing")
|
|
event.accept()
|
|
else:
|
|
event.ignore()
|