# Copyright: Ankitects Pty Ltd and contributors # License: GNU AGPL, version 3 or later; http://www.gnu.org/licenses/agpl.html import json import os import re import shutil import traceback import unicodedata import zipfile from concurrent.futures import Future from typing import Any, Optional import anki.importing as importing import aqt.deckchooser import aqt.forms import aqt.modelchooser from anki.importing.anki2 import V2ImportIntoV1 from anki.importing.apkg import AnkiPackageImporter from aqt import AnkiQt, gui_hooks from aqt.qt import * from aqt.utils import ( HelpPage, askUser, disable_help_button, getFile, getText, openHelp, showInfo, showText, showWarning, tooltip, tr, ) class ChangeMap(QDialog): def __init__(self, mw: AnkiQt, model: dict, current: str) -> None: QDialog.__init__(self, mw, Qt.WindowType.Window) self.mw = mw self.model = model self.frm = aqt.forms.changemap.Ui_ChangeMap() self.frm.setupUi(self) disable_help_button(self) n = 0 setCurrent = False for field in self.model["flds"]: item = QListWidgetItem(tr.importing_map_to(val=field["name"])) self.frm.fields.addItem(item) if current == field["name"]: setCurrent = True self.frm.fields.setCurrentRow(n) n += 1 self.frm.fields.addItem(QListWidgetItem(tr.importing_map_to_tags())) self.frm.fields.addItem(QListWidgetItem(tr.importing_ignore_field())) if not setCurrent: if current == "_tags": self.frm.fields.setCurrentRow(n) else: self.frm.fields.setCurrentRow(n + 1) self.field: Optional[str] = None def getField(self) -> str: self.exec() return self.field def accept(self) -> None: row = self.frm.fields.currentRow() if row < len(self.model["flds"]): self.field = self.model["flds"][row]["name"] elif row == self.frm.fields.count() - 2: self.field = "_tags" else: self.field = None QDialog.accept(self) def reject(self) -> None: self.accept() # called by importFile() when importing a mappable file like .csv # ImportType = Union[Importer,AnkiPackageImporter, TextImporter] class ImportDialog(QDialog): _DEFAULT_FILE_DELIMITER = "\t" def __init__(self, mw: AnkiQt, importer: Any) -> None: QDialog.__init__(self, mw, Qt.WindowType.Window) self.mw = mw self.importer = importer self.frm = aqt.forms.importing.Ui_ImportDialog() self.frm.setupUi(self) qconnect( self.frm.buttonBox.button(QDialogButtonBox.StandardButton.Help).clicked, self.helpRequested, ) disable_help_button(self) self.setupMappingFrame() self.setupOptions() self.modelChanged() self.frm.autoDetect.setVisible(self.importer.needDelimiter) gui_hooks.current_note_type_did_change.append(self.modelChanged) qconnect(self.frm.autoDetect.clicked, self.onDelimiter) self.updateDelimiterButtonText() self.frm.allowHTML.setChecked(self.mw.pm.profile.get("allowHTML", True)) qconnect(self.frm.importMode.currentIndexChanged, self.importModeChanged) self.frm.importMode.setCurrentIndex(self.mw.pm.profile.get("importMode", 1)) self.frm.tagModified.setText(self.mw.pm.profile.get("tagModified", "")) self.frm.tagModified.setCol(self.mw.col) # import button b = QPushButton(tr.actions_import()) self.frm.buttonBox.addButton(b, QDialogButtonBox.ButtonRole.AcceptRole) self.exec() def setupOptions(self) -> None: self.model = self.mw.col.models.current() self.modelChooser = aqt.modelchooser.ModelChooser( self.mw, self.frm.modelArea, label=False ) self.deck = aqt.deckchooser.DeckChooser(self.mw, self.frm.deckArea, label=False) def modelChanged(self, unused: Any = None) -> None: self.importer.model = self.mw.col.models.current() self.importer.initMapping() self.showMapping() def onDelimiter(self) -> None: # Open a modal dialog to enter an delimiter # Todo/Idea Constrain the maximum width, so it doesnt take up that much screen space delim, ok = getText( tr.importing_by_default_anki_will_detect_the(), self, help=HelpPage.IMPORTING, ) # If the modal dialog has been confirmed, update the delimiter if ok: # Check if the entered value is valid and if not fallback to default # at the moment every single character entry as well as '\t' is valid delim = delim if len(delim) > 0 else self._DEFAULT_FILE_DELIMITER delim = delim.replace("\\t", "\t") # un-escape it if len(delim) > 1: showWarning( tr.importing_multicharacter_separators_are_not_supported_please() ) return self.hideMapping() def updateDelim() -> None: self.importer.delimiter = delim self.importer.updateDelimiter() self.updateDelimiterButtonText() self.showMapping(hook=updateDelim) else: # If the operation has been canceled, do not do anything pass def updateDelimiterButtonText(self) -> None: if not self.importer.needDelimiter: return if self.importer.delimiter: d = self.importer.delimiter else: d = self.importer.dialect.delimiter if d == "\t": d = tr.importing_tab() elif d == ",": d = tr.importing_comma() elif d == " ": d = tr.studying_space() elif d == ";": d = tr.importing_semicolon() elif d == ":": d = tr.importing_colon() else: d = repr(d) txt = tr.importing_fields_separated_by(val=d) self.frm.autoDetect.setText(txt) def accept(self) -> None: self.importer.mapping = self.mapping if not self.importer.mappingOk(): showWarning(tr.importing_the_first_field_of_the_note()) return self.importer.importMode = self.frm.importMode.currentIndex() self.mw.pm.profile["importMode"] = self.importer.importMode self.importer.allowHTML = self.frm.allowHTML.isChecked() self.mw.pm.profile["allowHTML"] = self.importer.allowHTML self.importer.tagModified = self.frm.tagModified.text() self.mw.pm.profile["tagModified"] = self.importer.tagModified self.mw.col.set_aux_notetype_config( self.importer.model["id"], "lastDeck", self.deck.selected_deck_id ) self.mw.col.models.save(self.importer.model, updateReqs=False) self.mw.progress.start() self.mw.checkpoint(tr.actions_import()) def on_done(future: Future) -> None: self.mw.progress.finish() try: future.result() except UnicodeDecodeError: showUnicodeWarning() return except Exception as e: msg = f"{tr.importing_failed_debug_info()}\n" err = repr(str(e)) if "1-character string" in err: msg += err elif "invalidTempFolder" in err: msg += self.mw.errorHandler.tempFolderMsg() else: msg += traceback.format_exc() showText(msg) return else: txt = f"{tr.importing_importing_complete()}\n" if self.importer.log: txt += "\n".join(self.importer.log) self.close() showText(txt, plain_text_edit=True) self.mw.reset() self.mw.taskman.run_in_background(self.importer.run, on_done) def setupMappingFrame(self) -> None: # qt seems to have a bug with adding/removing from a grid, so we add # to a separate object and add/remove that instead self.frame = QFrame(self.frm.mappingArea) self.frm.mappingArea.setWidget(self.frame) self.mapbox = QVBoxLayout(self.frame) self.mapbox.setContentsMargins(0, 0, 0, 0) self.mapwidget: Optional[QWidget] = None def hideMapping(self) -> None: self.frm.mappingGroup.hide() def showMapping( self, keepMapping: bool = False, hook: Optional[Callable] = None ) -> None: if hook: hook() if not keepMapping: self.mapping = self.importer.mapping self.frm.mappingGroup.show() assert self.importer.fields() # set up the mapping grid if self.mapwidget: self.mapbox.removeWidget(self.mapwidget) self.mapwidget.deleteLater() self.mapwidget = QWidget() self.mapbox.addWidget(self.mapwidget) self.grid = QGridLayout(self.mapwidget) self.mapwidget.setLayout(self.grid) self.grid.setContentsMargins(3, 3, 3, 3) self.grid.setSpacing(6) fields = self.importer.fields() for num in range(len(self.mapping)): text = tr.importing_field_of_file_is(val=num + 1) self.grid.addWidget(QLabel(text), num, 0) if self.mapping[num] == "_tags": text = tr.importing_mapped_to_tags() elif self.mapping[num]: text = tr.importing_mapped_to(val=self.mapping[num]) else: text = tr.importing_ignored() self.grid.addWidget(QLabel(text), num, 1) button = QPushButton(tr.importing_change()) self.grid.addWidget(button, num, 2) qconnect(button.clicked, lambda _, s=self, n=num: s.changeMappingNum(n)) def changeMappingNum(self, n: int) -> None: f = ChangeMap(self.mw, self.importer.model, self.mapping[n]).getField() try: # make sure we don't have it twice index = self.mapping.index(f) self.mapping[index] = None except ValueError: pass self.mapping[n] = f if getattr(self.importer, "delimiter", False): self.savedDelimiter = self.importer.delimiter def updateDelim() -> None: self.importer.delimiter = self.savedDelimiter self.showMapping(hook=updateDelim, keepMapping=True) else: self.showMapping(keepMapping=True) def reject(self) -> None: self.modelChooser.cleanup() self.deck.cleanup() gui_hooks.current_note_type_did_change.remove(self.modelChanged) QDialog.reject(self) def helpRequested(self) -> None: openHelp(HelpPage.IMPORTING) def importModeChanged(self, newImportMode: int) -> None: if newImportMode == 0: self.frm.tagModified.setEnabled(True) else: self.frm.tagModified.setEnabled(False) def showUnicodeWarning() -> None: """Shorthand to show a standard warning.""" showWarning(tr.importing_selected_file_was_not_in_utf8()) def onImport(mw: AnkiQt) -> None: filt = ";;".join([x[0] for x in importing.importers(mw.col)]) file = getFile(mw, tr.actions_import(), None, key="import", filter=filt) if not file: return file = str(file) head, ext = os.path.splitext(file) ext = ext.lower() if ext == ".anki": showInfo(tr.importing_anki_files_are_from_a_very()) return elif ext == ".anki2": showInfo(tr.importing_anki2_files_are_not_directly_importable()) return importFile(mw, file) def importFile(mw: AnkiQt, file: str) -> None: importerClass = None done = False for i in importing.importers(mw.col): if done: break for mext in re.findall(r"[( ]?\*\.(.+?)[) ]", i[0]): if file.endswith(f".{mext}"): importerClass = i[1] done = True break if not importerClass: # if no matches, assume TSV importerClass = importing.importers(mw.col)[0][1] importer = importerClass(mw.col, file) # need to show import dialog? if importer.needMapper: # make sure we can load the file first mw.progress.start(immediate=True) try: importer.open() mw.progress.finish() diag = ImportDialog(mw, importer) except UnicodeDecodeError: mw.progress.finish() showUnicodeWarning() return except Exception as e: mw.progress.finish() msg = repr(str(e)) if msg == "'unknownFormat'": showWarning(tr.importing_unknown_file_format()) else: msg = f"{tr.importing_failed_debug_info()}\n" msg += str(traceback.format_exc()) showText(msg) return finally: importer.close() else: # if it's an apkg/zip, first test it's a valid file if isinstance(importer, AnkiPackageImporter): try: z = zipfile.ZipFile(importer.file) z.getinfo("collection.anki2") except: showWarning(invalidZipMsg()) return # we need to ask whether to import/replace; if it's # a colpkg file then the rest of the import process # will happen in setupApkgImport() if not setupApkgImport(mw, importer): return # importing non-colpkg files mw.progress.start(immediate=True) def on_done(future: Future) -> None: mw.progress.finish() try: future.result() except zipfile.BadZipfile: showWarning(invalidZipMsg()) except V2ImportIntoV1: showWarning( """\ To import this deck, please click the Update button at the top of the deck list, then try again.""" ) except Exception as e: err = repr(str(e)) if "invalidFile" in err: msg = tr.importing_invalid_file_please_restore_from_backup() showWarning(msg) elif "invalidTempFolder" in err: showWarning(mw.errorHandler.tempFolderMsg()) elif "readonly" in err: showWarning(tr.importing_unable_to_import_from_a_readonly()) else: msg = f"{tr.importing_failed_debug_info()}\n" msg += str(e) showText(msg) else: log = "\n".join(importer.log) if "\n" not in log: tooltip(log) else: showText(log, plain_text_edit=True) mw.reset() mw.taskman.run_in_background(importer.run, on_done) def invalidZipMsg() -> str: return tr.importing_this_file_does_not_appear_to() def setupApkgImport(mw: AnkiQt, importer: AnkiPackageImporter) -> bool: base = os.path.basename(importer.file).lower() full = ( (base == "collection.apkg") or re.match("backup-.*\\.apkg", base) or base.endswith(".colpkg") ) if not full: # adding return True if not mw.restoringBackup and not askUser( tr.importing_this_will_delete_your_existing_collection(), msgfunc=QMessageBox.warning, defaultno=True, ): return False replaceWithApkg(mw, importer.file, mw.restoringBackup) return False def replaceWithApkg(mw: aqt.AnkiQt, file: str, backup: bool) -> None: mw.unloadCollection(lambda: _replaceWithApkg(mw, file, backup)) def _replaceWithApkg(mw: aqt.AnkiQt, filename: str, backup: bool) -> None: mw.progress.start(immediate=True) def do_import() -> None: z = zipfile.ZipFile(filename) # v2 scheduler? colname = "collection.anki21" try: z.getinfo(colname) except KeyError: colname = "collection.anki2" with z.open(colname) as source, open(mw.pm.collectionPath(), "wb") as target: # ignore appears related to https://github.com/python/typeshed/issues/4349 # see if can turn off once issue fix is merged in shutil.copyfileobj(source, target) d = os.path.join(mw.pm.profileFolder(), "collection.media") for n, (cStr, file) in enumerate( json.loads(z.read("media").decode("utf8")).items() ): mw.taskman.run_on_main( lambda n=n: mw.progress.update( # type: ignore tr.importing_processed_media_file(count=n) ) ) size = z.getinfo(cStr).file_size dest = os.path.join(d, unicodedata.normalize("NFC", file)) # if we have a matching file size if os.path.exists(dest) and size == os.stat(dest).st_size: continue data = z.read(cStr) with open(dest, "wb") as file: file.write(data) z.close() def on_done(future: Future) -> None: mw.progress.finish() try: future.result() except Exception as e: print(e) showWarning(tr.importing_the_provided_file_is_not_a()) return if not mw.loadCollection(): return if backup: mw.col.mod_schema(check=False) tooltip(tr.importing_importing_complete()) mw.taskman.run_in_background(do_import, on_done)