anki/qt/aqt/mediasync.py

226 lines
7.2 KiB
Python
Raw Normal View History

# 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
import aqt.forms
import aqt.main
from anki.collection import Progress
from anki.errors import Interrupted, NetworkError
from anki.types import assert_exhaustive
PEP8 for rest of pylib (#1451) * PEP8 dbproxy.py * PEP8 errors.py * PEP8 httpclient.py * PEP8 lang.py * PEP8 latex.py * Add decorator to deprectate key words * Make replacement for deprecated attribute optional * Use new helper `_print_replacement_warning()` * PEP8 media.py * PEP8 rsbackend.py * PEP8 sound.py * PEP8 stdmodels.py * PEP8 storage.py * PEP8 sync.py * PEP8 tags.py * PEP8 template.py * PEP8 types.py * Fix DeprecatedNamesMixinForModule The class methods need to be overridden with instance methods, so every module has its own dicts. * Use `# pylint: disable=invalid-name` instead of id * PEP8 utils.py * Only decorate `__getattr__` with `@no_type_check` * Fix mypy issue with snakecase Importing it from `anki._vendor` raises attribute errors. * Format * Remove inheritance of DeprecatedNamesMixin There's almost no shared code now and overriding classmethods with instance methods raises mypy issues. * Fix traceback frames of deprecation warnings * remove fn/TimedLog (dae) Neither Anki nor add-ons appear to have been using it * fix some issues with stringcase use (dae) - the wheel was depending on the PyPI version instead of our vendored version - _vendor:stringcase should not have been listed in the anki py_library. We already include the sources in py_srcs, and need to refer to them directly. By listing _vendor:stringcase as well, we were making a top-level stringcase library available, which would have only worked for distributing because the wheel definition was also incorrect. - mypy errors are what caused me to mistakenly add the above - they were because the type: ignore at the top of stringcase.py was causing mypy to completely ignore the file, so it was not aware of any attributes it contained.
2021-10-25 06:50:13 +02:00
from anki.utils import int_time
from aqt import gui_hooks
2020-06-12 18:45:04 +02:00
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:
2021-02-01 14:28:21 +01:00
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)
2021-02-01 14:28:21 +01:00
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():
2021-03-26 04:48:26 +01:00
self._log_and_notify(tr.sync_media_disabled())
return
2020-05-30 04:28:22 +02:00
auth = self.mw.pm.sync_auth()
if auth is None:
return
2021-03-26 04:48:26 +01:00
self._log_and_notify(tr.sync_media_starting())
self._syncing = True
self._progress_timer = self.mw.progress.timer(
1000, self._on_progress, True, True, parent=self.mw
)
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:
PEP8 for rest of pylib (#1451) * PEP8 dbproxy.py * PEP8 errors.py * PEP8 httpclient.py * PEP8 lang.py * PEP8 latex.py * Add decorator to deprectate key words * Make replacement for deprecated attribute optional * Use new helper `_print_replacement_warning()` * PEP8 media.py * PEP8 rsbackend.py * PEP8 sound.py * PEP8 stdmodels.py * PEP8 storage.py * PEP8 sync.py * PEP8 tags.py * PEP8 template.py * PEP8 types.py * Fix DeprecatedNamesMixinForModule The class methods need to be overridden with instance methods, so every module has its own dicts. * Use `# pylint: disable=invalid-name` instead of id * PEP8 utils.py * Only decorate `__getattr__` with `@no_type_check` * Fix mypy issue with snakecase Importing it from `anki._vendor` raises attribute errors. * Format * Remove inheritance of DeprecatedNamesMixin There's almost no shared code now and overriding classmethods with instance methods raises mypy issues. * Fix traceback frames of deprecation warnings * remove fn/TimedLog (dae) Neither Anki nor add-ons appear to have been using it * fix some issues with stringcase use (dae) - the wheel was depending on the PyPI version instead of our vendored version - _vendor:stringcase should not have been listed in the anki py_library. We already include the sources in py_srcs, and need to refer to them directly. By listing _vendor:stringcase as well, we were making a top-level stringcase library available, which would have only worked for distributing because the wheel definition was also incorrect. - mypy errors are what caused me to mistakenly add the above - they were because the type: ignore at the top of stringcase.py was causing mypy to completely ignore the file, so it was not aware of any attributes it contained.
2021-10-25 06:50:13 +02:00
entry_with_time = LogEntryWithTime(time=int_time(), 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:
2021-03-26 04:48:26 +01:00
self._log_and_notify(tr.sync_media_complete())
2021-02-01 14:28:21 +01:00
def _handle_sync_error(self, exc: BaseException) -> None:
2020-02-04 06:16:11 +01:00
if isinstance(exc, Interrupted):
2021-03-26 04:48:26 +01:00
self._log_and_notify(tr.sync_media_aborted())
2020-02-04 06:16:11 +01:00
return
elif isinstance(exc, NetworkError):
# avoid popups for network errors
self._log_and_notify(str(exc))
return
2020-02-04 06:16:11 +01:00
2021-03-26 04:48:26 +01:00
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
2021-03-26 04:48:26 +01:00
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
2020-02-27 04:27:58 +01:00
def _on_start_stop(self, running: bool) -> None:
self.mw.toolbar.set_sync_active(running)
2021-02-01 14:28:21 +01:00
def show_sync_log(self) -> None:
aqt.dialogs.open("sync_log", self.mw, self)
2021-02-01 14:28:21 +01:00
def show_diag_until_finished(self, on_finished: Callable[[], None]) -> None:
2020-02-05 03:23:15 +01:00
# nothing to do if not syncing
if not self.is_syncing():
return on_finished()
2020-02-05 03:23:15 +01:00
diag: MediaSyncDialog = aqt.dialogs.open("sync_log", self.mw, self, True)
diag.show()
timer: QTimer | None = None
2021-02-01 14:28:21 +01:00
def check_finished() -> None:
if not self.is_syncing():
timer.stop()
on_finished()
timer = self.mw.progress.timer(150, check_finished, True, False, parent=self.mw)
2020-02-05 03:23:15 +01:00
2020-02-05 03:38:36 +01:00
def seconds_since_last_sync(self) -> int:
if self.is_syncing():
return 0
if self._log:
last = self._log[-1].time
else:
last = 0
PEP8 for rest of pylib (#1451) * PEP8 dbproxy.py * PEP8 errors.py * PEP8 httpclient.py * PEP8 lang.py * PEP8 latex.py * Add decorator to deprectate key words * Make replacement for deprecated attribute optional * Use new helper `_print_replacement_warning()` * PEP8 media.py * PEP8 rsbackend.py * PEP8 sound.py * PEP8 stdmodels.py * PEP8 storage.py * PEP8 sync.py * PEP8 tags.py * PEP8 template.py * PEP8 types.py * Fix DeprecatedNamesMixinForModule The class methods need to be overridden with instance methods, so every module has its own dicts. * Use `# pylint: disable=invalid-name` instead of id * PEP8 utils.py * Only decorate `__getattr__` with `@no_type_check` * Fix mypy issue with snakecase Importing it from `anki._vendor` raises attribute errors. * Format * Remove inheritance of DeprecatedNamesMixin There's almost no shared code now and overriding classmethods with instance methods raises mypy issues. * Fix traceback frames of deprecation warnings * remove fn/TimedLog (dae) Neither Anki nor add-ons appear to have been using it * fix some issues with stringcase use (dae) - the wheel was depending on the PyPI version instead of our vendored version - _vendor:stringcase should not have been listed in the anki py_library. We already include the sources in py_srcs, and need to refer to them directly. By listing _vendor:stringcase as well, we were making a top-level stringcase library available, which would have only worked for distributing because the wheel definition was also incorrect. - mypy errors are what caused me to mistakenly add the above - they were because the type: ignore at the top of stringcase.py was causing mypy to completely ignore the file, so it was not aware of any attributes it contained.
2021-10-25 06:50:13 +02:00
return int_time() - last
2020-02-05 03:38:36 +01:00
class MediaSyncDialog(QDialog):
silentlyClose = True
2020-02-05 03:23:15 +01:00
def __init__(
self, mw: aqt.main.AnkiQt, syncer: MediaSyncer, close_when_done: bool = False
) -> None:
super().__init__(mw)
self.mw = mw
self._syncer = syncer
2020-02-05 03:23:15 +01:00
self._close_when_done = close_when_done
self.form = aqt.forms.synclog.Ui_Dialog()
self.form.setupUi(self)
2021-03-26 04:48:26 +01:00
self.setWindowTitle(tr.sync_media_log_title())
disable_help_button(self)
2021-03-26 04:48:26 +01:00
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)
2020-02-05 03:23:15 +01:00
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()
2020-02-05 03:23:15 +01:00
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")
2020-02-05 03:23:15 +01:00
QDialog.reject(self)
def reopen(
self, mw: aqt.AnkiQt, syncer: Any, close_when_done: bool = False
) -> None:
2020-02-05 03:23:15 +01:00
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}"
2021-02-01 14:28:21 +01:00
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}"
2021-02-01 14:28:21 +01:00
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)
2020-02-05 03:23:15 +01:00
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()