From d92196da5346cba0e36d46882599f20c57c8e15a Mon Sep 17 00:00:00 2001 From: Remco Burema Date: Wed, 12 Feb 2025 17:22:27 +0100 Subject: [PATCH] Exclude plugins available in Marketplace from backups. part of CURA-12156 --- cura/API/Backups.py | 15 +++---- cura/Backups/Backup.py | 35 ++++++++++++--- cura/Backups/BackupsManager.py | 9 ++-- plugins/CuraDrive/src/CreateBackupJob.py | 54 +++++++++++++++++++----- 4 files changed, 87 insertions(+), 26 deletions(-) diff --git a/cura/API/Backups.py b/cura/API/Backups.py index 1940d38a36..07001d22e2 100644 --- a/cura/API/Backups.py +++ b/cura/API/Backups.py @@ -1,4 +1,4 @@ -# Copyright (c) 2018 Ultimaker B.V. +# Copyright (c) 2025 UltiMaker # Cura is released under the terms of the LGPLv3 or higher. from typing import Tuple, Optional, TYPE_CHECKING, Dict, Any @@ -9,14 +9,10 @@ if TYPE_CHECKING: class Backups: - """The back-ups API provides a version-proof bridge between Cura's - - BackupManager and plug-ins that hook into it. + """The back-ups API provides a version-proof bridge between Cura's BackupManager and plug-ins that hook into it. Usage: - .. code-block:: python - from cura.API import CuraAPI api = CuraAPI() api.backups.createBackup() @@ -26,13 +22,13 @@ class Backups: def __init__(self, application: "CuraApplication") -> None: self.manager = BackupsManager(application) - def createBackup(self) -> Tuple[Optional[bytes], Optional[Dict[str, Any]]]: + def createBackup(self, available_remote_plugins: frozenset[str] = frozenset()) -> Tuple[Optional[bytes], Optional[Dict[str, Any]]]: """Create a new back-up using the BackupsManager. :return: Tuple containing a ZIP file with the back-up data and a dict with metadata about the back-up. """ - return self.manager.createBackup() + return self.manager.createBackup(available_remote_plugins) def restoreBackup(self, zip_file: bytes, meta_data: Dict[str, Any]) -> None: """Restore a back-up using the BackupsManager. @@ -42,3 +38,6 @@ class Backups: """ return self.manager.restoreBackup(zip_file, meta_data) + + def shouldReinstallDownloadablePlugins(self) -> bool: + return self.manager.shouldReinstallDownloadablePlugins() diff --git a/cura/Backups/Backup.py b/cura/Backups/Backup.py index 19655df531..9f35e54ef1 100644 --- a/cura/Backups/Backup.py +++ b/cura/Backups/Backup.py @@ -1,5 +1,6 @@ -# Copyright (c) 2021 Ultimaker B.V. +# Copyright (c) 2025 UltiMaker # Cura is released under the terms of the LGPLv3 or higher. +import json import io import os @@ -13,6 +14,7 @@ from UM import i18nCatalog from UM.Logger import Logger from UM.Message import Message from UM.Platform import Platform +from UM.PluginRegistry import PluginRegistry from UM.Resources import Resources from UM.Version import Version @@ -30,10 +32,14 @@ class Backup: """These files should be ignored when making a backup.""" IGNORED_FOLDERS = [] # type: List[str] + """These folders should be ignored when making a backup.""" SECRETS_SETTINGS = ["general/ultimaker_auth_data"] """Secret preferences that need to obfuscated when making a backup of Cura""" + TO_INSTALL_FILE = "packages.json" + """File that contains the 'to_install' dictionary, that manages plugins to be installed on next startup.""" + catalog = i18nCatalog("cura") """Re-use translation catalog""" @@ -42,7 +48,7 @@ class Backup: self.zip_file = zip_file # type: Optional[bytes] self.meta_data = meta_data # type: Optional[Dict[str, str]] - def makeFromCurrent(self) -> None: + def makeFromCurrent(self, available_remote_plugins: frozenset[str] = frozenset()) -> None: """Create a back-up from the current user config folder.""" cura_release = self._application.getVersion() @@ -92,21 +98,40 @@ class Backup: # Restore the obfuscated settings self._illuminate(**secrets) - def _makeArchive(self, buffer: "io.BytesIO", root_path: str) -> Optional[ZipFile]: + def _fillToInstallsJson(self, file_path: str, reinstall_on_restore: dict[str, str], archive: ZipFile) -> None: + pass # TODO! + + def _findRedownloadablePlugins(self, available_remote_plugins: frozenset) -> dict[str, str]: + """ Find all plugins that should be able to be reinstalled from the Marketplace. + + :param plugins_path: Path to all plugins in the user-space. + :return: Set of all package-id's of plugins that can be reinstalled from the Marketplace. + """ + plugin_reg = PluginRegistry.getInstance() + id = "id" + return {v["location"]: v[id] for v in plugin_reg.getAllMetaData() + if v[id] in available_remote_plugins and not plugin_reg.isBundledPlugin(v[id])} + + def _makeArchive(self, buffer: "io.BytesIO", root_path: str, available_remote_plugins: frozenset) -> Optional[ZipFile]: """Make a full archive from the given root path with the given name. :param root_path: The root directory to archive recursively. :return: The archive as bytes. """ ignore_string = re.compile("|".join(self.IGNORED_FILES + self.IGNORED_FOLDERS)) + reinstall_instead_plugins = self._findRedownloadablePlugins(available_remote_plugins) try: archive = ZipFile(buffer, "w", ZIP_DEFLATED) - for root, folders, files in os.walk(root_path): + for root, folders, files in os.walk(root_path, topdown=True): + folders[:] = [f for f in folders if f not in reinstall_instead_plugins] for item_name in folders + files: absolute_path = os.path.join(root, item_name) if ignore_string.search(absolute_path): continue - archive.write(absolute_path, absolute_path[len(root_path) + len(os.sep):]) + if item_name == self.TO_INSTALL_FILE: + self._fillToInstallsJson(absolute_path, reinstall_instead_plugins, archive) + else: + archive.write(absolute_path, absolute_path[len(root_path) + len(os.sep):]) archive.close() return archive except (IOError, OSError, BadZipfile) as error: diff --git a/cura/Backups/BackupsManager.py b/cura/Backups/BackupsManager.py index 6c4670edb6..90dfc5e34e 100644 --- a/cura/Backups/BackupsManager.py +++ b/cura/Backups/BackupsManager.py @@ -1,4 +1,4 @@ -# Copyright (c) 2018 Ultimaker B.V. +# Copyright (c) 2025 UltiMaker # Cura is released under the terms of the LGPLv3 or higher. from typing import Dict, Optional, Tuple, TYPE_CHECKING @@ -22,7 +22,10 @@ class BackupsManager: def __init__(self, application: "CuraApplication") -> None: self._application = application - def createBackup(self) -> Tuple[Optional[bytes], Optional[Dict[str, str]]]: + def shouldReinstallDownloadablePlugins(self) -> bool: + return True + + def createBackup(self, available_remote_plugins: frozenset[str] = frozenset()) -> Tuple[Optional[bytes], Optional[Dict[str, str]]]: """ Get a back-up of the current configuration. @@ -31,7 +34,7 @@ class BackupsManager: self._disableAutoSave() backup = Backup(self._application) - backup.makeFromCurrent() + backup.makeFromCurrent(available_remote_plugins if self.shouldReinstallDownloadablePlugins() else frozenset()) self._enableAutoSave() # We don't return a Backup here because we want plugins only to interact with our API and not full objects. return backup.zip_file, backup.meta_data diff --git a/plugins/CuraDrive/src/CreateBackupJob.py b/plugins/CuraDrive/src/CreateBackupJob.py index 7d772769ed..cdd1d569c7 100644 --- a/plugins/CuraDrive/src/CreateBackupJob.py +++ b/plugins/CuraDrive/src/CreateBackupJob.py @@ -1,4 +1,4 @@ -# Copyright (c) 2020 Ultimaker B.V. +# Copyright (c) 2025 UltiMaker # Cura is released under the terms of the LGPLv3 or higher. import json import threading @@ -13,11 +13,14 @@ from UM.Message import Message from UM.TaskManagement.HttpRequestManager import HttpRequestManager from UM.TaskManagement.HttpRequestScope import JsonDecoratorScope from UM.i18n import i18nCatalog +from cura.ApplicationMetadata import CuraSDKVersion from cura.CuraApplication import CuraApplication from cura.UltimakerCloud.UltimakerCloudScope import UltimakerCloudScope +import cura.UltimakerCloud.UltimakerCloudConstants as UltimakerCloudConstants catalog = i18nCatalog("cura") +PACKAGES_URL = f"{UltimakerCloudConstants.CuraCloudAPIRoot}/cura-packages/v{UltimakerCloudConstants.CuraCloudAPIVersion}/cura/v{CuraSDKVersion}/packages" class CreateBackupJob(Job): """Creates backup zip, requests upload url and uploads the backup file to cloud storage.""" @@ -40,23 +43,54 @@ class CreateBackupJob(Job): self._job_done = threading.Event() """Set when the job completes. Does not indicate success.""" self.backup_upload_error_message = "" - """After the job completes, an empty string indicates success. Othrerwise, the value is a translated message.""" + """After the job completes, an empty string indicates success. Otherwise, the value is a translated message.""" + + def _setPluginFetchErrorMessage(self, error_msg: str) -> None: + Logger.error(f"Fetching plugins for backup resulted in error: {error_msg}") + self.backup_upload_error_message = "Couldn't update currently available plugins, backup stopped." + self._upload_message.hide() + self._job_done.set() def run(self) -> None: - upload_message = Message(catalog.i18nc("@info:backup_status", "Creating your backup..."), + self._upload_message = Message(catalog.i18nc("@info:backup_status", "Fetch re-downloadable package-ids..."), title = self.MESSAGE_TITLE, progress = -1) - upload_message.show() + self._upload_message.show() + CuraApplication.getInstance().processEvents() + + if CuraApplication.getInstance().getCuraAPI().backups.shouldReinstallDownloadablePlugins(): + request_url = f"{PACKAGES_URL}?package_type=plugin" + scope = JsonDecoratorScope(UltimakerCloudScope(CuraApplication.getInstance())) + HttpRequestManager.getInstance().get( + request_url, + scope=scope, + callback=self._continueRun, + error_callback=lambda reply, error: self._setPluginFetchErrorMessage(str(error)), + ) + else: + self._continueRun() + + def _continueRun(self, reply: "QNetworkReply" = None) -> None: + if reply is not None: + response_data = HttpRequestManager.readJSON(reply) + if "data" not in response_data: + self._setPluginFetchErrorMessage(f"Missing 'data' from response. Keys in response: {response_data.keys()}") + return + available_remote_plugins = frozenset({v["package_id"] for v in response_data["data"]}) + else: + available_remote_plugins = frozenset() + + self._upload_message.setText(catalog.i18nc("@info:backup_status", "Creating your backup...")) CuraApplication.getInstance().processEvents() cura_api = CuraApplication.getInstance().getCuraAPI() - self._backup_zip, backup_meta_data = cura_api.backups.createBackup() + self._backup_zip, backup_meta_data = cura_api.backups.createBackup(available_remote_plugins) if not self._backup_zip or not backup_meta_data: self.backup_upload_error_message = catalog.i18nc("@info:backup_status", "There was an error while creating your backup.") - upload_message.hide() + self._upload_message.hide() return - upload_message.setText(catalog.i18nc("@info:backup_status", "Uploading your backup...")) + self._upload_message.setText(catalog.i18nc("@info:backup_status", "Uploading your backup...")) CuraApplication.getInstance().processEvents() # Create an upload entry for the backup. @@ -66,11 +100,11 @@ class CreateBackupJob(Job): self._job_done.wait() if self.backup_upload_error_message == "": - upload_message.setText(catalog.i18nc("@info:backup_status", "Your backup has finished uploading.")) - upload_message.setProgress(None) # Hide progress bar + self._upload_message.setText(catalog.i18nc("@info:backup_status", "Your backup has finished uploading.")) + self._upload_message.setProgress(None) # Hide progress bar else: # some error occurred. This error is presented to the user by DrivePluginExtension - upload_message.hide() + self._upload_message.hide() def _requestUploadSlot(self, backup_metadata: Dict[str, Any], backup_size: int) -> None: """Request a backup upload slot from the API.