"""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() # --------------------------------------------------------------------------- # Login dialog (first-run token capture) # --------------------------------------------------------------------------- def prompt_login_qt() -> str | None: """Modal dialog to collect a fresh ``discord_id:password`` token. Returned string is the validated raw token (caller writes to disk). Returns ``None`` if the user picked "Skip cloud sync" or closed the dialog — caller should treat that as "don't sync, don't block launch." """ QtWidgets, QtCore, _ = import_qt() app_existed = QtWidgets.QApplication.instance() is not None app = QtWidgets.QApplication.instance() or QtWidgets.QApplication(sys.argv) if not app_existed: _apply_prism_dark(app) dialog = QtWidgets.QDialog() dialog.setWindowTitle("Cloud sync — first time setup") dialog.setFixedSize(480, 260) layout = QtWidgets.QVBoxLayout(dialog) layout.setContentsMargins(20, 20, 20, 20) layout.setSpacing(10) title = QtWidgets.QLabel("Connect to cloud save") font = title.font() font.setBold(True) font.setPointSize(font.pointSize() + 2) title.setFont(font) layout.addWidget(title) body = QtWidgets.QLabel( "In Discord, message the bot:\n" " /cloud register\n\n" "It will DM you a one-line token. Paste it below:" ) body.setWordWrap(True) layout.addWidget(body) field = QtWidgets.QLineEdit() field.setPlaceholderText("123456789012345678:a1b2c3d4…") field.setEchoMode(QtWidgets.QLineEdit.EchoMode.Password) layout.addWidget(field) error = QtWidgets.QLabel("") error.setStyleSheet("color: #ff6b6b;") layout.addWidget(error) layout.addStretch(1) button_row = QtWidgets.QHBoxLayout() skip = QtWidgets.QPushButton("Skip cloud sync") button_row.addWidget(skip) button_row.addStretch(1) save = QtWidgets.QPushButton("Save and continue") save.setDefault(True) button_row.addWidget(save) layout.addLayout(button_row) chosen: dict[str, str | None] = {"value": None} def on_save() -> None: token = field.text().strip() if ":" not in token: error.setText("Token must be discord_id:password") return head, sep, tail = token.partition(":") if not head.isdigit() or not tail: error.setText("discord_id must be numeric; password must be non-empty") return chosen["value"] = token dialog.accept() def on_skip() -> None: chosen["value"] = None dialog.reject() save.clicked.connect(on_save) skip.clicked.connect(on_skip) field.returnPressed.connect(on_save) dialog.exec() return chosen["value"] # --------------------------------------------------------------------------- # Conflict dialog (local-modified-since-last-sync + remote-newer) # --------------------------------------------------------------------------- _CONFLICT_QSS = """ QDialog { background: #313131; } QLabel#title { color: white; font-size: 20pt; font-weight: bold; letter-spacing: 2px; } QLabel#warning { color: #313131; background: #d6dde6; border-radius: 16px; font-size: 18pt; font-weight: bold; qproperty-alignment: AlignCenter; min-width: 32px; max-width: 32px; min-height: 32px; max-height: 32px; } QLabel#body { color: #c8c8c8; font-size: 10pt; } QLabel#note { color: #8a8a8a; font-size: 9pt; } QFrame#card { background: #2a2a2a; border: 1px solid transparent; border-radius: 4px; } QFrame#card:hover { background: #383838; border: 1px solid #96db59; } QLabel#cardTitle { color: white; font-size: 13pt; font-weight: bold; background: transparent; } QLabel#cardSubtitle { color: #b8b8b8; font-size: 9pt; background: transparent; } QLabel#cardIcon { color: white; font-size: 24pt; background: transparent; } QPushButton#cancel { background: #303030; color: white; border: 1px solid #4a4a4a; border-radius: 2px; padding: 8px 24px; font-size: 10pt; } QPushButton#cancel:hover { background: #3a3a3a; border-color: #96db59; } QPushButton#cancel:pressed { background: #222222; } """ def prompt_conflict_qt( local_modified: str, remote_modified: str, save_label: str = "Minecraft save", ) -> str: """Steam-Cloud-Conflict-styled dialog. Args: local_modified: human-readable "Saturday, February 12 at 12:28 AM" remote_modified: same, but for the cloud snapshot save_label: noun phrase for body copy (e.g. "Minecraft save"). Returns one of: ``"keep_local"``, ``"use_remote"``, ``"cancel"``. Visually mirrors Steam's Cloud Conflict modal: header with warning glyph + uppercase title, two large card buttons (cloud / local), Cancel bottom-right. Card click commits and dismisses. """ QtWidgets, QtCore, Signal = import_qt() app_existed = QtWidgets.QApplication.instance() is not None app = QtWidgets.QApplication.instance() or QtWidgets.QApplication(sys.argv) if not app_existed: _apply_prism_dark(app) class _Card(QtWidgets.QFrame): clicked = Signal() def __init__(self, glyph: str, title: str, subtitle: str) -> None: super().__init__() self.setObjectName("card") self.setCursor(QtCore.Qt.CursorShape.PointingHandCursor) self.setSizePolicy( QtWidgets.QSizePolicy.Policy.Expanding, QtWidgets.QSizePolicy.Policy.Fixed, ) row = QtWidgets.QHBoxLayout(self) row.setContentsMargins(18, 14, 18, 14) row.setSpacing(16) icon = QtWidgets.QLabel(glyph) icon.setObjectName("cardIcon") icon.setFixedWidth(40) icon.setAlignment(QtCore.Qt.AlignmentFlag.AlignCenter) row.addWidget(icon) text_col = QtWidgets.QVBoxLayout() text_col.setSpacing(2) t = QtWidgets.QLabel(title) t.setObjectName("cardTitle") s = QtWidgets.QLabel(subtitle) s.setObjectName("cardSubtitle") text_col.addWidget(t) text_col.addWidget(s) row.addLayout(text_col, 1) def mousePressEvent(self, _event: Any) -> None: # noqa: N802 self.clicked.emit() dialog = QtWidgets.QDialog() dialog.setWindowTitle("Cloud sync — conflict") dialog.setFixedSize(640, 460) dialog.setStyleSheet(_CONFLICT_QSS) outer = QtWidgets.QVBoxLayout(dialog) outer.setContentsMargins(28, 24, 28, 20) outer.setSpacing(14) header = QtWidgets.QHBoxLayout() header.setSpacing(12) warning = QtWidgets.QLabel("!") warning.setObjectName("warning") header.addWidget(warning) title = QtWidgets.QLabel("CLOUD CONFLICT") title.setObjectName("title") header.addWidget(title) header.addStretch(1) outer.addLayout(header) body = QtWidgets.QLabel( f"Your local {save_label} conflicts with what is stored in the cloud. " f"Whichever save data you choose to keep will be synced to this device " f"and the cloud. The option you choose not to keep will be overwritten." ) body.setObjectName("body") body.setWordWrap(True) outer.addWidget(body) outer.addSpacing(6) cloud_card = _Card("☁", "Cloud Save", f"Modified {remote_modified}") local_card = _Card("▢", "Local Save", f"Modified {local_modified}") outer.addWidget(cloud_card) outer.addWidget(local_card) outer.addStretch(1) foot = QtWidgets.QHBoxLayout() note = QtWidgets.QLabel( "Note: You will need to choose a version to keep before launching the game." ) note.setObjectName("note") note.setWordWrap(True) foot.addWidget(note, 1) cancel = QtWidgets.QPushButton("Cancel") cancel.setObjectName("cancel") foot.addWidget(cancel) outer.addLayout(foot) chosen: dict[str, str] = {"value": "cancel"} def pick_remote() -> None: chosen["value"] = "use_remote" dialog.accept() def pick_local() -> None: chosen["value"] = "keep_local" dialog.accept() def pick_cancel() -> None: chosen["value"] = "cancel" dialog.reject() cloud_card.clicked.connect(pick_remote) local_card.clicked.connect(pick_local) cancel.clicked.connect(pick_cancel) dialog.exec() return chosen["value"]