a7812dedc0
The enum changes should work on PyQt 5.x, and are required in PyQt 6.x. They are not supported by the PyQt5 typings however, so we need to run our tests with PyQt6.
224 lines
7.2 KiB
Python
224 lines
7.2 KiB
Python
# Copyright: Ankitects Pty Ltd and contributors
|
|
# License: GNU AGPL, version 3 or later; http://www.gnu.org/licenses/agpl.html
|
|
|
|
from __future__ import annotations
|
|
|
|
import time
|
|
from concurrent.futures import Future
|
|
from dataclasses import dataclass
|
|
from typing import Any, Callable, Union
|
|
|
|
import aqt
|
|
from anki.collection import Progress
|
|
from anki.errors import Interrupted, NetworkError
|
|
from anki.types import assert_exhaustive
|
|
from anki.utils import intTime
|
|
from aqt import gui_hooks
|
|
from aqt.qt import QDialog, QDialogButtonBox, QPushButton, QTextCursor, QTimer, qconnect
|
|
from aqt.utils import disable_help_button, showWarning, tr
|
|
|
|
LogEntry = Union[Progress.MediaSync, str]
|
|
|
|
|
|
@dataclass
|
|
class LogEntryWithTime:
|
|
time: int
|
|
entry: LogEntry
|
|
|
|
|
|
class MediaSyncer:
|
|
def __init__(self, mw: aqt.main.AnkiQt) -> None:
|
|
self.mw = mw
|
|
self._syncing: bool = False
|
|
self._log: list[LogEntryWithTime] = []
|
|
self._progress_timer: QTimer | None = None
|
|
gui_hooks.media_sync_did_start_or_stop.append(self._on_start_stop)
|
|
|
|
def _on_progress(self) -> None:
|
|
progress = self.mw.col.latest_progress()
|
|
if not progress.HasField("media_sync"):
|
|
return
|
|
sync_progress = progress.media_sync
|
|
self._log_and_notify(sync_progress)
|
|
|
|
def start(self) -> None:
|
|
"Start media syncing in the background, if it's not already running."
|
|
if self._syncing:
|
|
return
|
|
|
|
if not self.mw.pm.media_syncing_enabled():
|
|
self._log_and_notify(tr.sync_media_disabled())
|
|
return
|
|
|
|
auth = self.mw.pm.sync_auth()
|
|
if auth is None:
|
|
return
|
|
|
|
self._log_and_notify(tr.sync_media_starting())
|
|
self._syncing = True
|
|
self._progress_timer = self.mw.progress.timer(
|
|
1000, self._on_progress, True, True
|
|
)
|
|
gui_hooks.media_sync_did_start_or_stop(True)
|
|
|
|
def run() -> None:
|
|
self.mw.col.sync_media(auth)
|
|
|
|
self.mw.taskman.run_in_background(run, self._on_finished)
|
|
|
|
def _log_and_notify(self, entry: LogEntry) -> None:
|
|
entry_with_time = LogEntryWithTime(time=intTime(), entry=entry)
|
|
self._log.append(entry_with_time)
|
|
self.mw.taskman.run_on_main(
|
|
lambda: gui_hooks.media_sync_did_progress(entry_with_time)
|
|
)
|
|
|
|
def _on_finished(self, future: Future) -> None:
|
|
self._syncing = False
|
|
if self._progress_timer:
|
|
self._progress_timer.stop()
|
|
self._progress_timer = None
|
|
gui_hooks.media_sync_did_start_or_stop(False)
|
|
|
|
exc = future.exception()
|
|
if exc is not None:
|
|
self._handle_sync_error(exc)
|
|
else:
|
|
self._log_and_notify(tr.sync_media_complete())
|
|
|
|
def _handle_sync_error(self, exc: BaseException) -> None:
|
|
if isinstance(exc, Interrupted):
|
|
self._log_and_notify(tr.sync_media_aborted())
|
|
return
|
|
elif isinstance(exc, NetworkError):
|
|
# avoid popups for network errors
|
|
self._log_and_notify(str(exc))
|
|
return
|
|
|
|
self._log_and_notify(tr.sync_media_failed())
|
|
showWarning(str(exc))
|
|
|
|
def entries(self) -> list[LogEntryWithTime]:
|
|
return self._log
|
|
|
|
def abort(self) -> None:
|
|
if not self.is_syncing():
|
|
return
|
|
self._log_and_notify(tr.sync_media_aborting())
|
|
self.mw.col.set_wants_abort()
|
|
self.mw.col.abort_media_sync()
|
|
|
|
def is_syncing(self) -> bool:
|
|
return self._syncing
|
|
|
|
def _on_start_stop(self, running: bool) -> None:
|
|
self.mw.toolbar.set_sync_active(running)
|
|
|
|
def show_sync_log(self) -> None:
|
|
aqt.dialogs.open("sync_log", self.mw, self)
|
|
|
|
def show_diag_until_finished(self, on_finished: Callable[[], None]) -> None:
|
|
# nothing to do if not syncing
|
|
if not self.is_syncing():
|
|
return on_finished()
|
|
|
|
diag: MediaSyncDialog = aqt.dialogs.open("sync_log", self.mw, self, True)
|
|
diag.show()
|
|
|
|
timer: QTimer | None = None
|
|
|
|
def check_finished() -> None:
|
|
if not self.is_syncing():
|
|
timer.stop()
|
|
on_finished()
|
|
|
|
timer = self.mw.progress.timer(150, check_finished, True, False)
|
|
|
|
def seconds_since_last_sync(self) -> int:
|
|
if self.is_syncing():
|
|
return 0
|
|
|
|
if self._log:
|
|
last = self._log[-1].time
|
|
else:
|
|
last = 0
|
|
return intTime() - last
|
|
|
|
|
|
class MediaSyncDialog(QDialog):
|
|
silentlyClose = True
|
|
|
|
def __init__(
|
|
self, mw: aqt.main.AnkiQt, syncer: MediaSyncer, close_when_done: bool = False
|
|
) -> None:
|
|
super().__init__(mw)
|
|
self.mw = mw
|
|
self._syncer = syncer
|
|
self._close_when_done = close_when_done
|
|
self.form = aqt.forms.synclog.Ui_Dialog()
|
|
self.form.setupUi(self)
|
|
self.setWindowTitle(tr.sync_media_log_title())
|
|
disable_help_button(self)
|
|
self.abort_button = QPushButton(tr.sync_abort_button())
|
|
qconnect(self.abort_button.clicked, self._on_abort)
|
|
self.abort_button.setAutoDefault(False)
|
|
self.form.buttonBox.addButton(
|
|
self.abort_button, QDialogButtonBox.ButtonRole.ActionRole
|
|
)
|
|
self.abort_button.setHidden(not self._syncer.is_syncing())
|
|
|
|
gui_hooks.media_sync_did_progress.append(self._on_log_entry)
|
|
gui_hooks.media_sync_did_start_or_stop.append(self._on_start_stop)
|
|
|
|
self.form.plainTextEdit.setPlainText(
|
|
"\n".join(self._entry_to_text(x) for x in syncer.entries())
|
|
)
|
|
self.form.plainTextEdit.moveCursor(QTextCursor.MoveOperation.End)
|
|
self.show()
|
|
|
|
def reject(self) -> None:
|
|
if self._close_when_done and self._syncer.is_syncing():
|
|
# closing while syncing on close starts an abort
|
|
self._on_abort()
|
|
return
|
|
|
|
aqt.dialogs.markClosed("sync_log")
|
|
QDialog.reject(self)
|
|
|
|
def reopen(
|
|
self, mw: aqt.AnkiQt, syncer: Any, close_when_done: bool = False
|
|
) -> None:
|
|
self._close_when_done = close_when_done
|
|
self.show()
|
|
|
|
def _on_abort(self, *_args: Any) -> None:
|
|
self._syncer.abort()
|
|
self.abort_button.setHidden(True)
|
|
|
|
def _time_and_text(self, stamp: int, text: str) -> str:
|
|
asctime = time.asctime(time.localtime(stamp))
|
|
return f"{asctime}: {text}"
|
|
|
|
def _entry_to_text(self, entry: LogEntryWithTime) -> str:
|
|
if isinstance(entry.entry, str):
|
|
txt = entry.entry
|
|
elif isinstance(entry.entry, Progress.MediaSync):
|
|
txt = self._logentry_to_text(entry.entry)
|
|
else:
|
|
assert_exhaustive(entry.entry)
|
|
return self._time_and_text(entry.time, txt)
|
|
|
|
def _logentry_to_text(self, e: Progress.MediaSync) -> str:
|
|
return f"{e.added}, {e.removed}, {e.checked}"
|
|
|
|
def _on_log_entry(self, entry: LogEntryWithTime) -> None:
|
|
self.form.plainTextEdit.appendPlainText(self._entry_to_text(entry))
|
|
if not self._syncer.is_syncing():
|
|
self.abort_button.setHidden(True)
|
|
|
|
def _on_start_stop(self, running: bool) -> None:
|
|
if not running and self._close_when_done:
|
|
aqt.dialogs.markClosed("sync_log")
|
|
self._close_when_done = False
|
|
self.close()
|