claude-timemachine
d4c76a4e37
Login dialog (prompt_login_qt): Modal asking for a discord_id:password token. Echo masked. Validates format before accepting. 'Skip cloud sync' returns None so caller can bypass sync without blocking the launch. Conflict dialog (prompt_conflict_qt): Modal with three radio options — keep_local, use_remote, cancel. Shown when local files were modified since the last pulled snapshot AND the remote has a newer snapshot from another host (divergence state.json detection happens in sync.pull; not wired yet). Both reuse the Prism dark palette via _apply_prism_dark. Tk fallbacks not implemented — Qt is the path most players will hit. Next: state.json (last_pulled_snapshot_id), divergence detection in sync.pull, integration with these dialogs.
355 lines
12 KiB
Python
355 lines
12 KiB
Python
"""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)
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def prompt_conflict_qt(
|
|
local_summary: str,
|
|
remote_summary: str,
|
|
) -> str:
|
|
"""Ask how to resolve a divergence.
|
|
|
|
Returns one of: ``"keep_local"``, ``"use_remote"``, ``"cancel"``.
|
|
"""
|
|
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 — conflict")
|
|
dialog.setFixedSize(520, 320)
|
|
|
|
layout = QtWidgets.QVBoxLayout(dialog)
|
|
layout.setContentsMargins(20, 20, 20, 20)
|
|
layout.setSpacing(10)
|
|
|
|
title = QtWidgets.QLabel("Local and cloud diverged")
|
|
font = title.font()
|
|
font.setBold(True)
|
|
font.setPointSize(font.pointSize() + 2)
|
|
title.setFont(font)
|
|
layout.addWidget(title)
|
|
|
|
summary = QtWidgets.QLabel(
|
|
f"Local changes since last sync:\n {local_summary}\n\n"
|
|
f"Cloud has newer changes from another machine:\n {remote_summary}\n\n"
|
|
f"What should we do?"
|
|
)
|
|
summary.setWordWrap(True)
|
|
layout.addWidget(summary)
|
|
|
|
keep_local = QtWidgets.QRadioButton(
|
|
"Keep local — skip pull. Next exit will overwrite cloud."
|
|
)
|
|
use_remote = QtWidgets.QRadioButton(
|
|
"Use cloud — overwrite local files with the cloud version."
|
|
)
|
|
cancel_launch = QtWidgets.QRadioButton(
|
|
"Cancel launch — don't start Minecraft until I resolve manually."
|
|
)
|
|
keep_local.setChecked(True)
|
|
layout.addWidget(keep_local)
|
|
layout.addWidget(use_remote)
|
|
layout.addWidget(cancel_launch)
|
|
|
|
layout.addStretch(1)
|
|
|
|
button_row = QtWidgets.QHBoxLayout()
|
|
button_row.addStretch(1)
|
|
confirm = QtWidgets.QPushButton("Continue")
|
|
confirm.setDefault(True)
|
|
button_row.addWidget(confirm)
|
|
layout.addLayout(button_row)
|
|
|
|
chosen: dict[str, str] = {"value": "cancel"}
|
|
|
|
def on_confirm() -> None:
|
|
if keep_local.isChecked():
|
|
chosen["value"] = "keep_local"
|
|
elif use_remote.isChecked():
|
|
chosen["value"] = "use_remote"
|
|
else:
|
|
chosen["value"] = "cancel"
|
|
dialog.accept()
|
|
|
|
confirm.clicked.connect(on_confirm)
|
|
dialog.exec()
|
|
return chosen["value"]
|