GitHub-Proxy/Cura
1
0
Fork 0
mirror of https://git.mirrors.martin98.com/https://github.com/Ultimaker/Cura synced 2025-06-30 08:35:09 +08:00
Code Issues Packages Projects Releases Wiki Activity

Exclude plugins available in Marketplace from backups.

Browse Source 
part of CURA-12156
This commit is contained in:
Remco Burema 2025-02-12 17:22:27 +01:00
parent 2083a35858
commit d92196da53
4 changed files with 87 additions and 26 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

15
cura/API/Backups.py
View File

@ -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()

35
cura/Backups/Backup.py
View File

@ -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:

9
cura/Backups/BackupsManager.py
View File

@ -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

54
plugins/CuraDrive/src/CreateBackupJob.py
View File

@ -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.