"""PySide6 GUI package entry point.""" from __future__ import annotations import os import sys from dataclasses import replace from .. import appconfig, chrome, db, diagnostics, image_studio_generation, update_check, update_health from ..version import APP_NAME, APP_VERSION, display_name from . import widgets as _widgets from .widgets import * QT_IMPORT_ERROR = _widgets.QT_IMPORT_ERROR QMessageBox = _widgets._RawQMessageBox run_worker = _widgets._raw_run_worker if QT_IMPORT_ERROR is None else _widgets.run_worker if QT_IMPORT_ERROR is None: from .models import ApplyTaskTableModel, GenerateTaskTableModel, TaskTableModel from .workers import ( AccountLoginCheckWorker, AIModelTestWorker, CMHubModelCatalogWorker, CMHubSettingsWorker, ApplyWorker, CollectWorker, GenerateWorker, ImageStudioDownloadOriginalWorker, ImageStudioExportWorker, ImageStudioGenerateJobsWorker, ImageStudioPullImagesWorker, ImageStudioResumeJobsWorker, ProductSuiteAiWriteWorker, ProductSuiteGenerateWorker, ProductSuiteHistoryExportWorker, ProductSuiteImportImagesWorker, ProductTabOpenWorker, WriteBackWorker, ) from .tabs.accounts import AccountDialog, AccountsTab from .tabs.apply import ApplyTab from .tabs.collect import CollectTab from .tabs.generate import GenerateTab from .tabs.image_studio import ImageStudioPreviewDialog, ImageStudioTab from .product_suite_prompt_dialog import ProductSuitePromptDialog from .tabs.product_suite import ProductSuitePreviewDialog, ProductSuiteTab from .tabs.settings import SettingsTab from .main_window import MainWindow from .update_dialog import ForcedUpdateDialog, UpdatePreparationWorker else: class MainWindow(QMainWindow): def __init__(self): raise RuntimeError(f"{APP_NAME} GUI 无法启动：当前 Python 环境未安装 PySide6。") def _ensure_offscreen_for_headless_tests(): if "PYTEST_CURRENT_TEST" in os.environ and "QT_QPA_PLATFORM" not in os.environ: os.environ["QT_QPA_PLATFORM"] = "offscreen" def _write_update_check_diagnostic(message, *, result=None, exc=None): payload = None if result is not None: payload = { "checked": result.checked, "forced": result.forced, "current_version": result.current_version, "latest_version": result.latest_version, "min_supported_version": result.min_supported_version, "download_url": result.download_url, "error": result.error, } try: diagnostics.write_diagnostic_log( message, level="WARNING", step="startup_update_check", payload=payload, exc=exc, ) except Exception: # 版本检查第一版必须失败放行，诊断日志不可写也不能阻断启动。 pass def _show_forced_update_dialog(result, *, parent=None, dialog_factory=None) -> bool: dialog_class = dialog_factory or ForcedUpdateDialog dialog = dialog_class(result, parent=parent) dialog.exec() return False def _run_startup_update_gate(*, checker=None, dialog_factory=None) -> bool: try: result = (checker or update_check.check_for_update)() except Exception as exc: _write_update_check_diagnostic("启动版本检查异常，已允许继续使用", exc=exc) return True if result.error: _write_update_check_diagnostic("启动版本检查失败，已允许继续使用", result=result) if result.forced: failed = update_health.get_failed_release( appconfig.app_base_dir(), result.latest_version, result.sha256, ) if failed: result = replace( result, automatic_update_error=( "该版本自动升级曾失败，已停止重复安装。请等待管理员发布修复版本，或手动安装新版。" ), ) return _show_forced_update_dialog(result, dialog_factory=dialog_factory) return True def main() -> int: if QT_IMPORT_ERROR is not None: print(f"{APP_NAME} GUI 无法启动：当前 Python 环境未安装 PySide6。") return 1 _ensure_offscreen_for_headless_tests() app = QApplication.instance() or QApplication(sys.argv) # Set on the application so the taskbar button, dialogs and message boxes # all inherit it, not just the main window. startup_icon = app_icon() if startup_icon is not None: app.setWindowIcon(startup_icon) health_context = update_health.context_from_argv( sys.argv, appconfig.app_base_dir(), APP_VERSION, ) update_health.write_health(health_context, "process_started") try: appconfig.prepare_data_dir() except appconfig.DataMigrationConflictError as exc: update_health.write_health(health_context, "environment_blocked", str(exc)) QMessageBox.critical(None, "数据迁移冲突", str(exc)) return 1 except appconfig.DataDirectoryWriteError as exc: update_health.write_health(health_context, "environment_blocked", str(exc)) QMessageBox.critical(None, "数据目录不可写", str(exc)) return 1 except appconfig.ConfigError as exc: update_health.write_health(health_context, "environment_blocked", str(exc)) QMessageBox.critical(None, "启动配置错误", str(exc)) return 1 if not _run_startup_update_gate(): return 1 try: startup = chrome.ensure_configured_chrome_path() except (OSError, appconfig.ConfigError) as exc: update_health.write_health(health_context, "environment_blocked", str(exc)) QMessageBox.critical(None, "启动配置错误", str(exc)) return 1 runtime_lease = None try: database_path = appconfig.db_path(startup["config"]) db.init_db(database_path) runtime_lease = image_studio_generation.acquire_startup_recovery_lease( appconfig.data_dir(startup["config"]) ) recovery = image_studio_generation.recover_stale_direct_jobs_at_startup( lease=runtime_lease, path=database_path, ) recovered_count = len(recovery.get("recovered") or []) if recovered_count: diagnostics.write_diagnostic_log( "启动时标记未确认的自定义网关套图任务", level="WARNING", step="image_studio_direct_recovery", payload={"recovered_count": recovered_count}, ) elif recovery.get("skipped"): diagnostics.write_diagnostic_log( "检测到另一个程序实例，跳过自定义网关套图任务恢复", level="INFO", step="image_studio_direct_recovery", ) except Exception as exc: _write_update_check_diagnostic("自定义网关套图任务恢复检查失败", exc=exc) window = MainWindow( config=startup["config"], startup_status=startup["message"], ) if runtime_lease is not None: app.aboutToQuit.connect(runtime_lease.release) window.show() QTimer.singleShot( 0, lambda: update_health.write_health(health_context, "main_window_ready"), ) return app.exec()