cloud-sync/cloud_sync/ui_qt.py

"""Qt progress UI for cloud-sync.

Supports both PySide6 (preferred — LGPL, official Qt binding) and PyQt6
(fallback — GPL/commercial). Same code runs on both because their
QtWidgets / QtCore APIs are interchangeable for our subset.

This module never imports Qt at top level. ``import_qt()`` raises
ImportError if neither binding is available; the factory in ``ui.py``
catches that and falls back to the tkinter window.

Threading model: ``QApplication`` runs on the main thread (started by
``run_with`` via ``QDialog.exec``); the restic worker runs on a daemon
``threading.Thread``; cross-thread UI updates go via a ``Signal`` on a
bridge ``QObject`` so Qt enqueues them onto the main thread.
"""

from __future__ import annotations

import sys
import threading
from typing import Any, Callable


def import_qt() -> tuple[Any, Any, Any]:
    """Return (QtWidgets, QtCore, Signal). Raises ImportError if neither
    PySide6 nor PyQt6 is installed."""
    try:
        from PySide6 import QtCore, QtWidgets  # type: ignore
        return QtWidgets, QtCore, QtCore.Signal
    except ImportError:
        pass
    try:
        from PyQt6 import QtCore, QtWidgets  # type: ignore
        return QtWidgets, QtCore, QtCore.pyqtSignal
    except ImportError as e:
        raise ImportError(
            "neither PySide6 nor PyQt6 is installed; "
            "pip install 'cloud-sync[qt]' or pip install PySide6"
        ) from e


def _import_qtgui() -> Any:
    """Return whichever QtGui module is available (PySide6 or PyQt6)."""
    try:
        from PySide6 import QtGui  # type: ignore
        return QtGui
    except ImportError:
        from PyQt6 import QtGui  # type: ignore
        return QtGui


def _apply_prism_dark(app: Any) -> None:
    """Apply a Prism-Launcher-inspired dark palette + Fusion style.

    Palette RGB values are facts taken from
    PrismLauncher/launcher/ui/themes/DarkTheme.cpp (GPL-3.0). Numerical
    color values themselves aren't copyrightable; only the surrounding
    code is. We reimplement the same look from scratch under MIT.

    Override via ``PRISM_THEME=off`` to keep the system / Qt-default
    palette (useful if the user's desktop theme handles dark mode).
    """
    import os
    if os.environ.get("PRISM_THEME") == "off":
        return
    G = _import_qtgui()
    app.setStyle("Fusion")
    p = G.QPalette()
    Role = G.QPalette.ColorRole
    white = G.QColor("white")
    p.setColor(Role.Window, G.QColor(49, 49, 49))
    p.setColor(Role.WindowText, white)
    p.setColor(Role.Base, G.QColor(34, 34, 34))
    p.setColor(Role.AlternateBase, G.QColor(42, 42, 42))
    p.setColor(Role.ToolTipBase, white)
    p.setColor(Role.ToolTipText, white)
    p.setColor(Role.Text, white)
    p.setColor(Role.Button, G.QColor(48, 48, 48))
    p.setColor(Role.ButtonText, white)
    p.setColor(Role.BrightText, G.QColor("red"))
    p.setColor(Role.Link, G.QColor(47, 163, 198))
    p.setColor(Role.Highlight, G.QColor(150, 219, 89))  # Prism's signature green
    p.setColor(Role.HighlightedText, G.QColor("black"))
    p.setColor(Role.PlaceholderText, G.QColor("darkGray"))
    app.setPalette(p)


class QtProgressWindow:
    """Modal Qt dialog: title + status + indeterminate progress + Cancel."""

    def __init__(self) -> None:
        QtWidgets, QtCore, Signal = import_qt()
        self._QtCore = QtCore

        # Bridge so worker thread can update UI via a queued signal.
        class _Bridge(QtCore.QObject):  # type: ignore[misc, valid-type]
            status_changed = Signal(str)
            finished = Signal(int)

        app_existed = QtWidgets.QApplication.instance() is not None
        self._app = QtWidgets.QApplication.instance() or QtWidgets.QApplication(
            sys.argv
        )
        if not app_existed:
            _apply_prism_dark(self._app)

        self._dialog = QtWidgets.QDialog()
        self._dialog.setWindowTitle("Cloud sync")
        self._dialog.setFixedSize(440, 160)
        # Block ESC + window close X → mark cancelled, don't accept
        self._dialog.setWindowFlag(
            QtCore.Qt.WindowType.WindowContextHelpButtonHint, False
        )

        layout = QtWidgets.QVBoxLayout(self._dialog)
        layout.setContentsMargins(20, 20, 20, 20)

        self._title_label = QtWidgets.QLabel("Working…")
        font = self._title_label.font()
        font.setBold(True)
        font.setPointSize(font.pointSize() + 1)
        self._title_label.setFont(font)
        layout.addWidget(self._title_label)

        self._status_label = QtWidgets.QLabel("Starting…")
        layout.addWidget(self._status_label)

        self._bar = QtWidgets.QProgressBar()
        self._bar.setRange(0, 0)  # indeterminate
        self._bar.setTextVisible(False)
        layout.addWidget(self._bar)

        button_row = QtWidgets.QHBoxLayout()
        button_row.addStretch(1)
        self._cancel_btn = QtWidgets.QPushButton("Cancel")
        button_row.addWidget(self._cancel_btn)
        layout.addLayout(button_row)

        self._bridge = _Bridge()
        self._bridge.status_changed.connect(self._status_label.setText)
        self._bridge.finished.connect(self._on_finished)
        self._cancel_btn.clicked.connect(self._on_cancel)
        self._dialog.rejected.connect(self._on_cancel)

        self._cancelled = False
        self._worker_rc: int | None = None
        self._worker_exc: BaseException | None = None

    # -- public API ----------------------------------------------------

    def set_status(self, msg: str) -> None:
        self._bridge.status_changed.emit(msg)

    def is_cancelled(self) -> bool:
        return self._cancelled

    def run_with(self, worker: Callable[[], int], title: str) -> int:
        self._title_label.setText(title)
        self._status_label.setText("Starting…")

        def thread_target() -> None:
            try:
                rc = worker()
            except BaseException as e:  # noqa: BLE001
                self._worker_exc = e
                rc = 1
            self._worker_rc = rc
            self._bridge.finished.emit(rc)

        t = threading.Thread(target=thread_target, daemon=True)
        t.start()
        self._dialog.exec()

        if self._worker_exc is not None:
            raise self._worker_exc
        return self._worker_rc if self._worker_rc is not None else 1

    # -- internals -----------------------------------------------------

    def _on_cancel(self) -> None:
        self._cancelled = True
        self._status_label.setText("Cancelling…")
        self._cancel_btn.setEnabled(False)

    def _on_finished(self, _rc: int) -> None:
        self._dialog.accept()