# src/certflow/views/main_window.py
"""主窗口 - 流水线调度中心
统一调度各流程视图，导航顺序按业务流（配置驱动）排列：
- 系统设置 → 基础数据维护 → 计划导入 → 校正队列
- → 查询与补充 → 打印合格证 → 打印历史 → 扫描件生成 → 报告生成
"""
from __future__ import annotations
import contextlib
from typing import TYPE_CHECKING, Any, override
from PySide6.QtCore import QTimer
from PySide6.QtWidgets import (
QApplication,
QHBoxLayout,
QMainWindow,
QMessageBox,
QPushButton,
QStackedWidget,
QStatusBar,
QVBoxLayout,
QWidget,
)
from certflow.actions import dispatch
from certflow.config.settings import get_ui_config
from certflow.controllers.base_controller import BaseController
from certflow.utils.database import DatabaseManager
from certflow.utils.logger import logger
if TYPE_CHECKING:
from certflow.controllers.settings_controller import SettingsController
[文档]
class MainWindow(QMainWindow):
"""主窗口 - 统一调度各流程视图"""
def __init__(
self,
session: Any,
config: dict = None,
db_manager: DatabaseManager | None = None,
settings_controller: SettingsController | None = None,
) -> None:
"""初始化主窗口
Args:
session: SQLAlchemy 会话对象
config: 配置字典
db_manager: 数据库管理器实例（可选，用于传递给子视图）
settings_controller: 设置控制器（可选，用于主题/语言/自动保存）
"""
super().__init__()
self.session = session
self.config = config or {}
self.db_manager = db_manager
self._settings_controller = settings_controller
# 如果有 db_manager，设置全局共享管理器（让所有控制器自动使用）
if self.db_manager:
BaseController.set_shared_db_manager(self.db_manager)
logger.debug("已设置全局共享数据库管理器")
_title, _min_w, _min_h = self._resolve_window_geometry(get_ui_config())
self._window_title = _title
self.setWindowTitle(_title)
self.setMinimumSize(_min_w, _min_h)
# 创建堆叠视图
self.stack = QStackedWidget()
# 视图引用
self.import_dialog = None
self.query_view = None
self.output_view = None
self.report_view = None
self.history_view = None
self.settings_view = None
self._init_views()
self._setup_ui()
self._connect_signals()
# 默认显示查询中心 hub
if self.query_hub:
self.stack.setCurrentWidget(self.query_hub)
self.statusBar().showMessage("就绪")
# ============================================================
# 视图初始化
# ============================================================
def _init_views(self) -> None:
"""初始化所有视图"""
# 导入视图（延迟导入避免循环）
from certflow.views.history_view import HistoryView
from certflow.views.import_page import ImportPage
from certflow.views.output_view import OutputView
from certflow.views.query_hub_view import QueryHubView
from certflow.views.query_view import QueryView
from certflow.views.report_view import ReportView
from certflow.views.settings_view import SettingsView
# 查询视图（堆叠页） - 传入 db_manager（如果有）
if self.db_manager:
self.query_view = QueryView(self.session, db_manager=self.db_manager)
else:
self.query_view = QueryView(self.session)
# 查询中心 hub：销售计划(内嵌 query_view) / 校正队列 / 基础数据
self.query_hub = QueryHubView(
self.query_view,
db_manager=self.db_manager,
session=self.session,
)
self.stack.addWidget(self.query_hub)
# 计划导入堆叠页（销售计划导入 / BOM导入 两个 tab）
self.import_page = ImportPage(
db_manager=self.db_manager,
session=self.session,
)
self.stack.addWidget(self.import_page)
# 统一输出视图（打印/扫描合一，§11.5 阶段3）
if self.db_manager:
self.output_view = OutputView(self.session, self.config, db_manager=self.db_manager)
else:
self.output_view = OutputView(self.session, self.config)
self.stack.addWidget(self.output_view)
# 报告视图
self.report_view = ReportView()
self.stack.addWidget(self.report_view)
# 历史视图
if self.db_manager:
self.history_view = HistoryView(self.session, db_manager=self.db_manager)
else:
self.history_view = HistoryView(self.session)
self.stack.addWidget(self.history_view)
# 设置视图（注入 Controller 以实现保存即生效）
self.settings_view = SettingsView(self.config, controller=self._settings_controller)
self.stack.addWidget(self.settings_view)
logger.info(f"视图初始化完成: 共 {self.stack.count()} 个")
# ============================================================
# UI 构建
# ============================================================
def _setup_ui(self) -> None:
"""设置界面布局"""
central = QWidget()
layout = QVBoxLayout(central)
layout.setContentsMargins(0, 0, 0, 0)
layout.setSpacing(0)
# 顶部导航栏
nav_bar = self._create_nav_bar()
layout.addWidget(nav_bar)
# 内容区域
layout.addWidget(self.stack)
# 状态栏
self.setStatusBar(QStatusBar(self))
self.setCentralWidget(central)
def _create_nav_bar(self) -> Any:
"""创建顶部导航栏按钮（配置驱动：ui.yaml nav_buttons → UIConfig.get_nav_buttons，
action 经 dispatch() 派发到 _nav_* 方法；样式由 ThemeManager 通过 QWidget#navBar 控制）"""
nav_widget = QWidget()
nav_widget.setObjectName("navBar")
nav_widget.setFixedHeight(50)
layout = QHBoxLayout(nav_widget)
layout.setContentsMargins(10, 0, 10, 0)
# Logo
logo = QPushButton("🏭 CertFlow")
logo.setObjectName("navLogo")
logo.setEnabled(False)
layout.addWidget(logo)
layout.addSpacing(20)
# 导航按钮：由 ui.yaml nav_buttons 配置驱动（Phase 2.1）；UIConfig 不可用时回退硬编码。
# 键为 action 名（与注册表一致），点击经 dispatch() 派发到 MainWindow 的 _nav_* 方法。
self._nav_buttons: dict[str, Any] = {} # action -> QPushButton
self._nav_button_specs: dict[
str, Any
] = {} # action -> ButtonConfig（保留原始文本供 i18n 回译）
for spec in self._get_nav_button_specs():
btn = QPushButton(spec.text)
btn.setObjectName("navButton")
btn.setCheckable(True)
if spec.tooltip:
btn.setToolTip(spec.tooltip)
action = spec.action
btn.clicked.connect(lambda _=False, a=action: dispatch(self, a))
layout.addWidget(btn)
self._nav_buttons[action] = btn
self._nav_button_specs[action] = spec
layout.addStretch()
# 系统设置：独立角落齿轮按钮（不计入主 nav 5 键，避免 nav 冗余；
# 与 BLUEPRINT_QUERY_UNIFY.md §1.2 终态一致：设置作为角落按钮保留）
self.settings_gear_btn = QPushButton("⚙️")
self.settings_gear_btn.setObjectName("settingsGearBtn")
self.settings_gear_btn.setToolTip("系统设置")
self.settings_gear_btn.setFixedWidth(40)
self.settings_gear_btn.clicked.connect(lambda _=False: self._nav_to_settings())
layout.addWidget(self.settings_gear_btn)
# 主题切换按钮
self.theme_toggle_btn = QPushButton()
self.theme_toggle_btn.setObjectName("themeToggleBtn")
self.theme_toggle_btn.setToolTip("切换浅色/深色主题")
self.theme_toggle_btn.clicked.connect(self._on_toggle_theme)
self._update_theme_toggle_text()
layout.addWidget(self.theme_toggle_btn)
# 当前用户/状态
self.status_label = QPushButton("✅ 系统就绪")
self.status_label.setEnabled(False)
layout.addWidget(self.status_label)
return nav_widget
def _resolve_window_geometry(self, ui_cfg: Any) -> tuple[str, int, int]:
"""从 ui.yaml ``main_window`` 解析窗口标题与最小尺寸（UI 域单入口，任务 2.2）。
标题支持 ``{app_name}`` / ``{app_version}`` 占位符（见 ``_format_window_title``）；
ui.yaml 不可用或该段缺失时回退到与重构前一致的默认值，保证行为零变化。
"""
mw: dict = {}
if ui_cfg is not False and ui_cfg is not None:
mw = ui_cfg.main_window or {}
raw_title = mw.get("title", "CertFlow - 阀门证书管理系统")
return (
self._format_window_title(raw_title),
mw.get("min_width", 1200),
mw.get("min_height", 800),
)
@staticmethod
def _format_window_title(raw: str) -> str:
"""格式化窗口标题模板；含 ``{app_name}`` / ``{app_version}`` 占位符时替换。
占位符来源为 ``certflow.config.settings`` 的 ``APP_NAME`` / ``APP_VERSION``\n（来自 config.yaml 的 ``app.name`` / ``app.version``，已确认赋值：\n``CertFlow-PySide6`` / ``0.0.1-alpha.1``）；模板不含占位符时原样返回。
"""
if "{" not in raw:
return raw
from certflow.config.settings import APP_NAME, APP_VERSION
try:
return raw.format(app_name=APP_NAME, app_version=APP_VERSION)
except (KeyError, IndexError):
return raw
def _get_nav_button_specs(self) -> list:
"""返回导航按钮配置列表（ButtonConfig）。
唯一来源为 ``ui.yaml``（经 ``UIConfig.get_nav_buttons``，Phase 2.1 配置驱动，
任务 2.2 落实 UI 域单入口）；ui.yaml 不可用或为空时返回空列表，由 UI 显示空白
导航栏（不再保留硬编码回退）。
"""
ui_cfg = get_ui_config()
if ui_cfg is not False and ui_cfg is not None:
buttons = ui_cfg.get_nav_buttons()
if buttons:
return buttons
return []
def _switch_to_view(self, view: Any) -> None:
"""切换到指定视图"""
self.stack.setCurrentWidget(view)
self._update_nav_button_state()
# ============================================================
# 导航 action 承载方法（供统一注册表 dispatch，Phase 2.1）
# 命名约定 _nav_<key>，与 ui.yaml nav_buttons[].action（nav.<key>）对应。
# ============================================================
def _nav_to_settings(self) -> None:
"""导航：系统设置（切换设置视图）"""
self._switch_to_view(self.settings_view)
def _nav_show_import(self) -> None:
"""导航：计划导入（切换堆叠页）"""
self._switch_to_view(self.import_page)
def _nav_to_query(self) -> None:
"""导航：查询与补充（切换查询中心 hub）"""
self._switch_to_view(self.query_hub)
def _nav_to_output(self) -> None:
"""导航：统一输出视图（打印/扫描合一）"""
self._switch_to_view(self.output_view)
def _nav_to_history(self) -> None:
"""导航：打印历史（切换历史视图）"""
self._switch_to_view(self.history_view)
def _nav_to_report(self) -> None:
"""导航：报告生成（切换报告视图）"""
self._switch_to_view(self.report_view)
def _nav_show_lookup_tables(self) -> None:
"""导航：基础数据维护（打开查找表对话框）"""
from certflow.views.lookup_tables_dialog import LookupTablesDialog
dlg = LookupTablesDialog(
db_manager=self.db_manager,
session=self.session,
parent=self,
)
dlg.exec()
def _nav_show_correction_queue(self) -> None:
"""导航：校正队列（打开校正队列对话框）"""
from certflow.views.correction_queue_view import CorrectionQueueView
dlg = CorrectionQueueView(
db_manager=self.db_manager,
session=self.session,
parent=self,
)
dlg.exec()
def _update_nav_button_state(self) -> None:
"""更新导航按钮选中状态（按 action 映射，避免依赖按钮文本）"""
current = self.stack.currentWidget()
# action -> 目标视图（None/弹窗类不保持选中）
view_actions = {
"nav.settings": self.settings_view,
"nav.import": self.import_page,
"nav.query": self.query_hub,
"nav.output": self.output_view,
"nav.history": self.history_view,
"nav.report": self.report_view,
}
dialog_actions: set[str] = set()
for action, btn in self._nav_buttons.items():
if action in dialog_actions:
btn.setChecked(False)
elif action in view_actions:
btn.setChecked(current == view_actions[action])
else:
btn.setChecked(False)
# ============================================================
# 导航操作
# ============================================================
def _on_import_completed(self, config: Any) -> None:
"""导入完成后刷新查询中心（销售计划 + 校正队列 + 基础数据/BOM）。
后台导入在独立连接提交后，本共享会话仍缓存旧对象（identity map），
且 WAL 下悬挂的读事务快照会冻结在导入前，导致 BOM 等查询视图显示
陈旧/双倍数据（见 docs/PATCH_LOG.md 补丁 33）。先结束事务并清空缓存，
再刷新各视图，使查询重读已提交的库。
"""
if self.session is not None:
# 仅 expire_all 不足以打破 WAL 读快照：必须结束悬挂的读事务。
with contextlib.suppress(Exception):
self.session.rollback()
self.session.expire_all()
if self.query_view:
self.query_view.refresh()
if getattr(self, "query_hub", None) is not None:
self.query_hub.refresh()
sheet = (config or {}).get("sheet_name", "")
self.statusBar().showMessage(f"导入完成 - {sheet}")
def _set_nav_locked(self, locked: bool) -> None:
"""导入运行中锁定/解锁导航按钮（避免切走导致导入中断）"""
for btn in self._nav_buttons.values():
btn.setEnabled(not locked)
if getattr(self, "settings_gear_btn", None) is not None:
self.settings_gear_btn.setEnabled(not locked)
[文档]
def show_output_view(
self,
sale_plan_ids: list[int] = None,
certificate_ids: list[int] = None,
target: str | None = None,
) -> None:
"""显示统一输出视图并加载数据（§11.5 阶段3）。
Args:
sale_plan_ids: 待打印的销售计划 ID 列表（委托打印子视图）。
certificate_ids: 待生成扫描件的合格证 ID 列表（委托扫描子视图）。
target: 预置输出目标（"print"/"scan"），缺省不强制切换。
"""
self._switch_to_view(self.output_view)
if target:
self.output_view.set_target(target)
if sale_plan_ids and self.output_view:
self.output_view.load_selected_sale_plan_ids(sale_plan_ids)
elif certificate_ids and self.output_view:
self.output_view.load_certificates(certificate_ids)
# ============================================================
# 信号连接
# ============================================================
def _connect_signals(self) -> None:
"""连接跨视图信号（防止重复连接）"""
if hasattr(self, "_signals_connected") and self._signals_connected:
return
# PySide6 connect() 是幂等的，重复连接同一个 slot 会自动忽略
# 不需要先 disconnect，避免 C++ 层 RuntimeWarning
if self.query_view:
self.query_view.navigate_to_print.connect(self._on_navigate_to_print)
if self.import_page:
self.import_page.imported.connect(self._on_import_completed)
self.import_page.busy_changed.connect(self._set_nav_locked)
if self.output_view:
self.output_view.output_completed.connect(self._on_output_completed)
# ============================================================
# SettingsController 信号连接 — 保存即生效
# ============================================================
if self._settings_controller:
# 主题切换 → 全局样式立即生效（通过 ThemeManager）
self._settings_controller.theme_changed.connect(self._on_theme_changed)
# 语言切换 → 刷新主窗口标题 + 状态栏 + 导航栏文本
self._settings_controller.language_changed.connect(self._on_language_changed)
# 自动保存定时器 → 连接输出视图中打印子视图的自动保存回调
if self.output_view:
self._settings_controller.set_auto_save_callback(self._on_auto_save_from_settings)
# 应用初始主题（从 userconfig 加载）
from certflow.config.settings import get_user_preference
initial_theme = get_user_preference("theme", "light")
self._apply_theme(initial_theme)
self._signals_connected = True
def _on_navigate_to_print(self, sale_plan_ids: list[int]) -> None:
"""从查询视图跳转到统一输出视图（延迟加载，避免信号冲突）"""
self._switch_to_view(self.output_view)
self.output_view.set_target("print")
QTimer.singleShot(
100,
lambda: self.output_view.print_view.load_selected_sale_plan_ids(sale_plan_ids),
)
def _on_output_completed(self, target: str, payload: list) -> None:
"""统一输出完成回调（打印/扫描）。
- target == "print"：沿用原「打印完成后提示是否生成扫描件」交互。
- target == "scan"：扫描件生成完成提示。
"""
if target == "print":
certificate_ids = payload or []
if not certificate_ids:
return
reply = QMessageBox.question(
self,
"生成扫描件",
f"已完成 {len(certificate_ids)} 张合格证打印，是否生成扫描件？",
QMessageBox.StandardButton.Yes | QMessageBox.StandardButton.No,
)
if reply == QMessageBox.StandardButton.Yes:
self.show_output_view(certificate_ids=certificate_ids, target="scan")
elif target == "scan":
output_files = payload or []
self.statusBar().showMessage(f"已生成 {len(output_files)} 个扫描件")
QMessageBox.information(
self,
"生成完成",
f"扫描件生成完成！\n共 {len(output_files)} 个文件",
)
[文档]
@override
def closeEvent(self, event: Any) -> None:
"""窗口关闭事件，清理资源"""
# 如果设置了共享管理器，清除它
if self.db_manager:
BaseController.clear_shared_db_manager()
logger.debug("已清除全局共享数据库管理器")
super().closeEvent(event)
# ============================================================
# SettingsController 信号槽 — 主题 / 语言 / 自动保存
# ============================================================
def _on_theme_changed(self, theme: str) -> None:
"""主题变更时全局应用新样式"""
self._apply_theme(theme)
def _apply_theme(self, theme: str) -> None:
"""应用主题到整个应用"""
from certflow.views.common.theme_manager import ThemeManager
app = QApplication.instance()
if app:
manager = ThemeManager(app, parent=self)
manager.apply_theme(theme)
def _on_toggle_theme(self) -> None:
"""点击导航栏主题按钮切换主题"""
from certflow.config.settings import get_user_preference, set_user_preference
current = get_user_preference("theme", "light")
new_theme = "dark" if current == "light" else "light"
self._apply_theme(new_theme)
set_user_preference("theme", new_theme)
self._update_theme_toggle_text()
logger.info(f"主题已切换为: {new_theme}")
def _update_theme_toggle_text(self) -> None:
"""更新主题切换按钮的图标文本"""
from certflow.config.settings import get_user_preference
theme = get_user_preference("theme", "light")
self.theme_toggle_btn.setText("🌙" if theme == "light" else "☀️")
def _on_language_changed(self, language: str) -> None:
"""语言切换时刷新主窗口文本"""
from certflow.views.common.i18n_manager import I18nManager
i18n = I18nManager(parent=self)
i18n.switch_language(language)
# 刷新窗口标题（与初始标题同源，经 ui.yaml main_window.title 驱动）
self.setWindowTitle(i18n.tr(self._window_title))
# 刷新导航栏按钮文本（按 action 取原始文本回译，避免依赖硬编码文本列表）
for action, btn in self._nav_buttons.items():
spec = self._nav_button_specs.get(action)
if spec:
btn.setText(i18n.tr(spec.text))
# 刷新状态标签
self.status_label.setText(i18n.tr("✅ 系统就绪"))
self.statusBar().showMessage(i18n.tr("就绪"))
# 刷新设置视图
if self.settings_view:
self.settings_view.retranslate_ui()
def _on_auto_save_from_settings(self) -> None:
"""自动保存定时器触发 — 委托输出视图中的打印子视图执行保存"""
if not self.output_view:
return
print_view = self.output_view.print_view
# 获取当前打印数据并保存
try:
data = print_view._get_form_data()
serials = print_view._serial_numbers
if data and serials:
print_view._auto_save(serials, data)
logger.debug("自动保存定时器: 已保存打印记录")
except Exception as e:
logger.error(f"自动保存失败: {e}")