anki/qt/aqt/mediasync.py

# 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 int_time
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=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:
            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 int_time() - 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()