b9251290ca
This adds Python 3.9 and 3.10 typing syntax to files that import attributions from __future___. Python 3.9 should be able to cope with the 3.10 syntax, but Python 3.8 will no longer work. On Windows/Mac, install the latest Python 3.9 version from python.org. There are currently no orjson wheels for Python 3.10 on Windows/Mac, which will break the build unless you have Rust installed separately. On Linux, modern distros should have Python 3.9 available already. If you're on an older distro, you'll need to build Python from source first.
222 lines
7.1 KiB
Python
222 lines
7.1 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.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.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()
|