Conditionally enable new importing in GUI

pylib/anki/collection.py

@@ -10,6 +10,7 @@ from anki import (
     collection_pb2,
     config_pb2,
     generic_pb2,
+    import_export_pb2,
     links_pb2,
     search_pb2,
     stats_pb2,
@@ -32,6 +33,7 @@ OpChangesAfterUndo = collection_pb2.OpChangesAfterUndo
 BrowserRow = search_pb2.BrowserRow
 BrowserColumns = search_pb2.BrowserColumns
 StripHtmlMode = card_rendering_pb2.StripHtmlRequest
+ImportLogWithChanges = import_export_pb2.ImportAnkiPackageResponse
 
 import copy
 import os
@@ -259,14 +261,6 @@ class Collection(DeprecatedNamesMixin):
             self._clear_caches()
             self.db = None
 
-    def export_collection(
-        self, out_path: str, include_media: bool, legacy: bool
-    ) -> None:
-        self.close_for_full_sync()
-        self._backend.export_collection_package(
-            out_path=out_path, include_media=include_media, legacy=legacy
-        )
-
     def rollback(self) -> None:
         self._clear_caches()
         self.db.rollback()
@@ -321,6 +315,15 @@ class Collection(DeprecatedNamesMixin):
         else:
             return -1
 
+    def legacy_checkpoint_pending(self) -> bool:
+        return (
+            self._have_outstanding_checkpoint()
+            and time.time() - self._last_checkpoint_at < 300
+        )
+
+    # Import/export
+    ##########################################################################
+
     def create_backup(
         self,
         *,
@@ -353,12 +356,17 @@ class Collection(DeprecatedNamesMixin):
         "Throws if backup creation failed."
         self._backend.await_backup_completion()
 
-    def legacy_checkpoint_pending(self) -> bool:
+    def export_collection(
-        return (
+        self, out_path: str, include_media: bool, legacy: bool
-            self._have_outstanding_checkpoint()
+    ) -> None:
-            and time.time() - self._last_checkpoint_at < 300
+        self.close_for_full_sync()
+        self._backend.export_collection_package(
+            out_path=out_path, include_media=include_media, legacy=legacy
         )
 
+    def import_anki_package(self, path: str) -> ImportLogWithChanges:
+        return self._backend.import_anki_package(package_path=path)
+
     # Object helpers
     ##########################################################################
 

qt/aqt/import_export.py

@@ -0,0 +1,124 @@
+# Copyright: Ankitects Pty Ltd and contributors
+# License: GNU AGPL, version 3 or later; http://www.gnu.org/licenses/agpl.html
+
+from __future__ import annotations
+
+from concurrent.futures import Future
+
+import aqt.main
+from anki.errors import Interrupted
+from aqt.operations import (
+    ClosedCollectionOpWithBackendProgress,
+    CollectionOpWithBackendProgress,
+)
+from aqt.qt import *
+from aqt.utils import askUser, getFile, showInfo, showText, showWarning, tooltip, tr
+
+
+def import_file(mw: aqt.main.AnkiQt) -> None:
+    if not (path := get_file_path(mw)):
+        return
+
+    filename = os.path.basename(path).lower()
+    if filename.endswith(".anki"):
+        showInfo(tr.importing_anki_files_are_from_a_very())
+    elif filename.endswith(".anki2"):
+        showInfo(tr.importing_anki2_files_are_not_directly_importable())
+    elif is_collection_package(filename):
+        maybe_import_collection_package(mw, path)
+    elif filename.endswith(".apkg"):
+        import_anki_package(mw, path)
+    else:
+        raise NotImplementedError
+
+
+def get_file_path(mw: aqt.main.AnkiQt) -> str | None:
+    if file := getFile(
+        mw,
+        tr.actions_import(),
+        None,
+        key="import",
+        filter=tr.importing_packaged_anki_deckcollection_apkg_colpkg_zip(),
+    ):
+        return str(file)
+    return None
+
+
+def is_collection_package(filename: str) -> bool:
+    return (
+        filename == "collection.apkg"
+        or (filename.startswith("backup-") and filename.endswith(".apkg"))
+        or filename.endswith(".colpkg")
+    )
+
+
+def maybe_import_collection_package(mw: aqt.main.AnkiQt, path: str) -> None:
+    if askUser(
+        tr.importing_this_will_delete_your_existing_collection(),
+        msgfunc=QMessageBox.warning,
+        defaultno=True,
+    ):
+        import_collection_package(mw, path)
+
+
+def import_collection_package(mw: aqt.main.AnkiQt, file: str) -> None:
+    def on_success(_future: Future) -> None:
+        mw.loadCollection()
+        tooltip(tr.importing_importing_complete())
+
+    def on_failure(err: Exception) -> None:
+        mw.loadCollection()
+        if not isinstance(err, Interrupted):
+            showWarning(str(err))
+
+    import_collection_package_op(mw, file).success(on_success).failure(
+        on_failure
+    ).run_in_background()
+
+
+def import_collection_package_op(
+    mw: aqt.main.AnkiQt, path: str
+) -> ClosedCollectionOpWithBackendProgress:
+    def op() -> None:
+        col_path = mw.pm.collectionPath()
+        media_folder = os.path.join(mw.pm.profileFolder(), "collection.media")
+        mw.backend.import_collection_package(
+            col_path=col_path, backup_path=path, media_folder=media_folder
+        )
+
+    return ClosedCollectionOpWithBackendProgress(
+        parent=mw,
+        op=op,
+        key="importing",
+    )
+
+
+def import_anki_package(mw: aqt.main.AnkiQt, path: str) -> None:
+    CollectionOpWithBackendProgress(
+        parent=mw,
+        op=lambda col: col.import_anki_package(path),
+        key="importing",
+    ).success(show_import_log).run_in_background()
+
+
+def show_import_log(future: Future) -> None:
+    log = future.log  # type: ignore
+    total = len(log.conflicting) + len(log.updated) + len(log.new) + len(log.duplicate)
+
+    text = f"""{tr.importing_notes_found_in_file(val=total)}
+{tr.importing_notes_that_could_not_be_imported(val=len(log.conflicting))}
+{tr.importing_notes_updated_as_file_had_newer(val=len(log.updated))}
+{tr.importing_notes_added_from_file(val=len(log.new))}
+{tr.importing_notes_skipped_as_theyre_already_in(val=len(log.duplicate))}
+
+{log_rows(log.conflicting, tr.importing_skipped())}
+{log_rows(log.updated, tr.importing_updated())}
+{log_rows(log.new, tr.adding_added())}
+{log_rows(log.duplicate, tr.importing_identical())}
+    """
+
+    showText(text, plain_text_edit=True)
+
+
+def log_rows(rows: list, action: str) -> str:
+    return "\n".join(f"[{action}] {', '.join(note.fields)}" for note in rows)

qt/aqt/importing.py

@@ -11,11 +11,10 @@ import anki.importing as importing
 import aqt.deckchooser
 import aqt.forms
 import aqt.modelchooser
-from anki.errors import Interrupted
 from anki.importing.anki2 import V2ImportIntoV1
 from anki.importing.apkg import AnkiPackageImporter
+from aqt.import_export import import_collection_package
 from aqt.main import AnkiQt, gui_hooks
-from aqt.operations import ClosedCollectionOpWithBackendProgress
 from aqt.qt import *
 from aqt.utils import (
     HelpPage,
@@ -439,31 +438,6 @@ def setupApkgImport(mw: AnkiQt, importer: AnkiPackageImporter) -> bool:
         msgfunc=QMessageBox.warning,
         defaultno=True,
     ):
-        run_full_apkg_import(mw, importer.file)
+        import_collection_package(mw, importer.file)
 
     return False
-
-
-def run_full_apkg_import(mw: AnkiQt, file: str) -> None:
-    def on_success(_future: Future) -> None:
-        mw.loadCollection()
-        tooltip(tr.importing_importing_complete())
-
-    def on_failure(err: Exception) -> None:
-        mw.loadCollection()
-        if not isinstance(err, Interrupted):
-            showWarning(str(err))
-
-    ClosedCollectionOpWithBackendProgress(
-        parent=mw,
-        op=lambda: full_apkg_import(mw, file),
-        key="importing",
-    ).success(on_success).failure(on_failure).run_in_background()
-
-
-def full_apkg_import(mw: AnkiQt, file: str) -> None:
-    col_path = mw.pm.collectionPath()
-    media_folder = os.path.join(mw.pm.profileFolder(), "collection.media")
-    mw.backend.import_collection_package(
-        col_path=col_path, backup_path=file, media_folder=media_folder
-    )

qt/aqt/main.py

@@ -47,10 +47,11 @@ from aqt.addons import DownloadLogEntry, check_and_prompt_for_updates, show_log_
 from aqt.dbcheck import check_db
 from aqt.emptycards import show_empty_cards
 from aqt.flags import FlagManager
+from aqt.import_export import import_collection_package_op, import_file
 from aqt.legacy import install_pylib_legacy
 from aqt.mediacheck import check_media_db
 from aqt.mediasync import MediaSyncer
-from aqt.operations import ClosedCollectionOpWithBackendProgress, QueryOp
+from aqt.operations import QueryOp
 from aqt.operations.collection import redo, undo
 from aqt.operations.deck import set_current_deck
 from aqt.profiles import ProfileManager as ProfileManagerType
@@ -402,16 +403,10 @@ class AnkiQt(QMainWindow):
         )
 
     def _openBackup(self, path: str) -> None:
-        import aqt.importing
-
         self.restoring_backup = True
         showInfo(tr.qt_misc_automatic_syncing_and_backups_have_been())
 
-        ClosedCollectionOpWithBackendProgress(
+        import_collection_package_op(self, path).success(
-            parent=self,
-            op=lambda: aqt.importing.full_apkg_import(self, path),
-            key="importing",
-        ).success(
             lambda _: self.onOpenProfile(
                 callback=lambda: self.col.mod_schema(check=False)
             )
@@ -1188,7 +1183,10 @@ title="{}" {}>{}</button>""".format(
     def onImport(self) -> None:
         import aqt.importing
 
-        aqt.importing.onImport(self)
+        if os.getenv("ANKI_BACKEND_IMPORT_EXPORT"):
+            import_file(self)
+        else:
+            aqt.importing.onImport(self)
 
     def onExport(self, did: DeckId | None = None) -> None:
         import aqt.exporting

qt/aqt/operations/__init__.py

@@ -11,6 +11,7 @@ import aqt.gui_hooks
 import aqt.main
 from anki.collection import (
     Collection,
+    ImportLogWithChanges,
     OpChanges,
     OpChangesAfterUndo,
     OpChangesWithCount,
@@ -35,6 +36,7 @@ ResultWithChanges = TypeVar(
         OpChangesWithCount,
         OpChangesWithId,
         OpChangesAfterUndo,
+        ImportLogWithChanges,
         HasChangesProperty,
     ],
 )