cmshoppe/app/gui/__init__.py

"""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()