path: root/plugins/DigitalLibrary/src/DFFileExportAndUploadManager.py

# Copyright (c) 2021 Ultimaker B.V.
# Cura is released under the terms of the LGPLv3 or higher.
import json
import threading
from json import JSONDecodeError
from typing import List, Dict, Any, Callable, Union, Optional

from PyQt6.QtCore import QUrl
from PyQt6.QtGui import QDesktopServices
from PyQt6.QtNetwork import QNetworkReply

from UM.FileHandler.FileHandler import FileHandler
from UM.Logger import Logger
from UM.Message import Message
from UM.Scene.SceneNode import SceneNode
from cura.CuraApplication import CuraApplication
from .BackwardsCompatibleMessage import getBackwardsCompatibleMessage
from .DFLibraryFileUploadRequest import DFLibraryFileUploadRequest
from .DFLibraryFileUploadResponse import DFLibraryFileUploadResponse
from .DFPrintJobUploadRequest import DFPrintJobUploadRequest
from .DFPrintJobUploadResponse import DFPrintJobUploadResponse
from .DigitalFactoryApiClient import DigitalFactoryApiClient
from .ExportFileJob import ExportFileJob


class DFFileExportAndUploadManager:
    """
    Class responsible for exporting the scene and uploading the exported data to the Digital Factory Library. Since 3mf
    and UFP files may need to be uploaded at the same time, this class keeps a single progress and success message for
    both files and updates those messages according to the progress of both the file job uploads.
    """
    def __init__(self, file_handlers: Dict[str, FileHandler],
                 nodes: List[SceneNode],
                 library_project_id: str,
                 library_project_name: str,
                 file_name: str,
                 formats: List[str],
                 on_upload_error: Callable[[], Any],
                 on_upload_success: Callable[[], Any],
                 on_upload_finished: Callable[[], Any],
                 on_upload_progress: Callable[[int], Any]) -> None:

        self._file_handlers: Dict[str, FileHandler] = file_handlers
        self._nodes: List[SceneNode] = nodes
        self._library_project_id: str = library_project_id
        self._library_project_name: str = library_project_name
        self._file_name: str = file_name
        self._upload_jobs: List[ExportFileJob] = []
        self._formats: List[str] = formats
        self._api = DigitalFactoryApiClient(application = CuraApplication.getInstance(), on_error = lambda error: Logger.log("e", str(error)))
        self._source_file_id: Optional[str] = None

        # Functions of the parent class that should be called based on the upload process output
        self._on_upload_error = on_upload_error
        self._on_upload_success = on_upload_success
        self._on_upload_finished = on_upload_finished
        self._on_upload_progress = on_upload_progress

        # Lock used for updating the progress message (since the progress is changed by two parallel upload jobs) or
        # show the success message (once both upload jobs are done)
        self._message_lock = threading.Lock()

        self._file_upload_job_metadata: Dict[str, Dict[str, Any]] = self.initializeFileUploadJobMetadata()

        self.progress_message = Message(
                title = "Uploading...",
                text = "Uploading files to '{}'".format(self._library_project_name),
                progress = -1,
                lifetime = 0,
                dismissable = False,
                use_inactivity_timer = False
        )

        self._generic_success_message = getBackwardsCompatibleMessage(
                text = "Your {} uploaded to '{}'.".format("file was" if len(self._file_upload_job_metadata) <= 1 else "files were", self._library_project_name),
                title = "Upload successful",
                lifetime = 30,
                message_type_str = "POSITIVE"
        )
        self._generic_success_message.addAction(
                "open_df_project",
                "Open project",
                "open-folder", "Open the project containing the file in Digital Library"
        )
        self._generic_success_message.actionTriggered.connect(self._onMessageActionTriggered)

    def _onCuraProjectFileExported(self, job: ExportFileJob) -> None:
        """Handler for when the DF Library workspace file (3MF) has been created locally.

        It can now be sent over the Digital Factory API.
        """
        if not job.getOutput():
            self._onJobExportError(job.getFileName())
            return
        self._file_upload_job_metadata[job.getFileName()]["export_job_output"] = job.getOutput()
        request = DFLibraryFileUploadRequest(
                content_type = job.getMimeType(),
                file_name = job.getFileName(),
                file_size = len(job.getOutput()),
                library_project_id = self._library_project_id
        )
        self._api.requestUpload3MF(request, on_finished = self._uploadFileData, on_error = self._onRequestUploadCuraProjectFileFailed)

    def _onPrintFileExported(self, job: ExportFileJob) -> None:
        """Handler for when the DF Library print job file (UFP) has been created locally.

        It can now be sent over the Digital Factory API.
        """
        if not job.getOutput():
            self._onJobExportError(job.getFileName())
            return
        self._file_upload_job_metadata[job.getFileName()]["export_job_output"] = job.getOutput()
        request = DFPrintJobUploadRequest(
                content_type = job.getMimeType(),
                job_name = job.getFileName(),
                file_size = len(job.getOutput()),
                library_project_id = self._library_project_id,
                source_file_id = self._source_file_id
        )
        self._api.requestUploadUFP(request, on_finished = self._uploadFileData, on_error = self._onRequestUploadPrintFileFailed)

    def _uploadFileData(self, file_upload_response: Union[DFLibraryFileUploadResponse, DFPrintJobUploadResponse]) -> None:
        """Uploads the exported file data after the file or print job upload has been registered at the Digital Factory
        Library API.

        :param file_upload_response: The response received from the Digital Factory Library API.
        """
        if isinstance(file_upload_response, DFLibraryFileUploadResponse):
            file_name = file_upload_response.file_name

            # store the `file_id` so it can be as `source_file_id` when uploading the print file
            self._source_file_id = file_upload_response.file_id
        elif isinstance(file_upload_response, DFPrintJobUploadResponse):
            file_name = file_upload_response.job_name if file_upload_response.job_name is not None else ""
        else:
            Logger.log("e", "Wrong response type received. Aborting uploading file to the Digital Library")
            return
        with self._message_lock:
            self.progress_message.show()
        self._file_upload_job_metadata[file_name]["file_upload_response"] = file_upload_response
        job_output = self._file_upload_job_metadata[file_name]["export_job_output"]

        with self._message_lock:
            self._file_upload_job_metadata[file_name]["upload_status"] = "uploading"

        self._api.uploadExportedFileData(file_upload_response,
                                         job_output,
                                         on_finished = self._onFileUploadFinished,
                                         on_success = self._onUploadSuccess,
                                         on_progress = self._onUploadProgress,
                                         on_error = self._onUploadError)

        self._handleNextUploadJob()

    def _onUploadProgress(self, filename: str, progress: int) -> None:
        """
        Updates the progress message according to the total progress of the two files and displays it to the user. It is
        made thread-safe with a lock, since the progress can be updated by two separate upload jobs

        :param filename: The name of the file for which we have progress (including the extension).
        :param progress: The progress percentage
        """
        with self._message_lock:
            self._file_upload_job_metadata[filename]["upload_progress"] = progress
            self._file_upload_job_metadata[filename]["upload_status"] = "uploading"
            total_progress = self.getTotalProgress()
            self.progress_message.setProgress(total_progress)
            self.progress_message.show()
        self._on_upload_progress(progress)

    def _onUploadSuccess(self, filename: str) -> None:
        """
        Sets the upload status to success and the progress of the file with the given filename to 100%. This function is
        should be called only if the file has uploaded all of its data successfully (i.e. no error occurred during the
        upload process).

        :param filename: The name of the file that was uploaded successfully (including the extension).
        """
        with self._message_lock:
            self._file_upload_job_metadata[filename]["upload_status"] = "success"
            self._file_upload_job_metadata[filename]["upload_progress"] = 100
        self._on_upload_success()

    def _onFileUploadFinished(self, filename: str) -> None:
        """
        Callback that makes sure the correct messages are displayed according to the statuses of the individual jobs.

        This function is called whenever an upload job has finished, regardless if it had errors or was successful.
        Both jobs have to have finished for the messages to show.

        :param filename: The name of the file that has finished uploading (including the extension).
        """
        with self._message_lock:

            # All files have finished their uploading process
            if all([(file_upload_job["upload_progress"] == 100 and file_upload_job["upload_status"] != "uploading") for file_upload_job in self._file_upload_job_metadata.values()]):

                # Reset and hide the progress message
                self.progress_message.setProgress(-1)
                self.progress_message.hide()

                # All files were successfully uploaded.
                if all([(file_upload_job["upload_status"] == "success") for file_upload_job in self._file_upload_job_metadata.values()]):
                    # Show a single generic success message for all files
                    self._generic_success_message.show()
                else:  # One or more files failed to upload.
                    # Show individual messages for each file, according to their statuses
                    for filename, upload_job_metadata in self._file_upload_job_metadata.items():
                        if upload_job_metadata["upload_status"] == "success":
                            upload_job_metadata["file_upload_success_message"].show()
                        else:
                            upload_job_metadata["file_upload_failed_message"].show()

                # Call the parent's finished function
                self._on_upload_finished()

    def _onJobExportError(self, filename: str) -> None:
        """
        Displays an appropriate message when the process to export a file fails.

        :param filename: The name of the file that failed to be exported (including the extension).
        """
        Logger.log("d", "Error while exporting file '{}'".format(filename))
        with self._message_lock:
            # Set the progress to 100% when the upload job fails, to avoid having the progress message stuck
            self._file_upload_job_metadata[filename]["upload_status"] = "failed"
            self._file_upload_job_metadata[filename]["upload_progress"] = 100
            self._file_upload_job_metadata[filename]["file_upload_failed_message"] = getBackwardsCompatibleMessage(
                    text = "Failed to export the file '{}'. The upload process is aborted.".format(filename),
                    title = "Export error",
                    message_type_str = "ERROR",
                    lifetime = 30
            )
        self._on_upload_error()
        self._onFileUploadFinished(filename)

    def _onRequestUploadCuraProjectFileFailed(self, reply: "QNetworkReply", network_error: "QNetworkReply.NetworkError") -> None:
        """
        Displays an appropriate message when the request to upload the Cura project file (.3mf) to the Digital Library fails.
        This means that something went wrong with the initial request to create a "file" entry in the digital library.
        """
        reply_string = bytes(reply.readAll()).decode()
        filename_3mf = self._file_name + ".3mf"
        Logger.log("d", "An error occurred while uploading the Cura project file '{}' to the Digital Library project '{}': {}".format(filename_3mf, self._library_project_id, reply_string))
        with self._message_lock:
            # Set the progress to 100% when the upload job fails, to avoid having the progress message stuck
            self._file_upload_job_metadata[filename_3mf]["upload_status"] = "failed"
            self._file_upload_job_metadata[filename_3mf]["upload_progress"] = 100

            human_readable_error = self.extractErrorTitle(reply_string)
            self._file_upload_job_metadata[filename_3mf]["file_upload_failed_message"] = getBackwardsCompatibleMessage(
                    text = "Failed to upload the file '{}' to '{}'. {}".format(filename_3mf, self._library_project_name, human_readable_error),
                    title = "File upload error",
                    message_type_str = "ERROR",
                    lifetime = 30
            )
        self._on_upload_error()
        self._onFileUploadFinished(filename_3mf)

    def _onRequestUploadPrintFileFailed(self, reply: "QNetworkReply", network_error: "QNetworkReply.NetworkError") -> None:
        """
        Displays an appropriate message when the request to upload the print file (.ufp) to the Digital Library fails.
        This means that something went wrong with the initial request to create a "file" entry in the digital library.
        """
        reply_string = bytes(reply.readAll()).decode()
        filename_ufp = self._file_name + ".ufp"
        Logger.log("d", "An error occurred while uploading the print job file '{}' to the Digital Library project '{}': {}".format(filename_ufp, self._library_project_id, reply_string))
        with self._message_lock:
            # Set the progress to 100% when the upload job fails, to avoid having the progress message stuck
            self._file_upload_job_metadata[filename_ufp]["upload_status"] = "failed"
            self._file_upload_job_metadata[filename_ufp]["upload_progress"] = 100

            human_readable_error = self.extractErrorTitle(reply_string)
            self._file_upload_job_metadata[filename_ufp]["file_upload_failed_message"] = getBackwardsCompatibleMessage(
                    title = "File upload error",
                    text = "Failed to upload the file '{}' to '{}'. {}".format(filename_ufp, self._library_project_name, human_readable_error),
                    message_type_str = "ERROR",
                    lifetime = 30
            )
        self._on_upload_error()
        self._onFileUploadFinished(filename_ufp)

    @staticmethod
    def extractErrorTitle(reply_body: Optional[str]) -> str:
        error_title = ""
        if reply_body:
            try:
                reply_dict = json.loads(reply_body)
            except JSONDecodeError:
                Logger.logException("w", "Unable to extract title from reply body")
                return error_title
            if "errors" in reply_dict and len(reply_dict["errors"]) >= 1 and "title" in reply_dict["errors"][0]:
                error_title = reply_dict["errors"][0]["title"]
        return error_title

    def _onUploadError(self, filename: str, reply: "QNetworkReply", error: "QNetworkReply.NetworkError") -> None:
        """
        Displays the given message if uploading the mesh has failed due to a generic error (i.e. lost connection).
        If one of the two files fail, this error function will set its progress as finished, to make sure that the
        progress message doesn't get stuck.

        :param filename: The name of the file that failed to upload (including the extension).
        """
        reply_string = bytes(reply.readAll()).decode()
        Logger.log("d", "Error while uploading '{}' to the Digital Library project '{}'. Reply: {}".format(filename, self._library_project_id, reply_string))
        with self._message_lock:
            # Set the progress to 100% when the upload job fails, to avoid having the progress message stuck
            self._file_upload_job_metadata[filename]["upload_status"] = "failed"
            self._file_upload_job_metadata[filename]["upload_progress"] = 100
            human_readable_error = self.extractErrorTitle(reply_string)
            self._file_upload_job_metadata[filename]["file_upload_failed_message"] = getBackwardsCompatibleMessage(
                    title = "File upload error",
                    text = "Failed to upload the file '{}' to '{}'. {}".format(self._file_name, self._library_project_name, human_readable_error),
                    message_type_str = "ERROR",
                    lifetime = 30
            )

        self._on_upload_error()

    def getTotalProgress(self) -> int:
        """
        Returns the total upload progress of all the upload jobs

        :return: The average progress percentage
        """
        return int(sum([file_upload_job["upload_progress"] for file_upload_job in self._file_upload_job_metadata.values()]) / len(self._file_upload_job_metadata.values()))

    def _onMessageActionTriggered(self, message, action):
        if action == "open_df_project":
            project_url = "{}/app/library/project/{}?wait_for_new_files=true&utm_source=cura&utm_medium=software&utm_campaign=saved-library-file-message".format(CuraApplication.getInstance().ultimakerDigitalFactoryUrl, self._library_project_id)
            QDesktopServices.openUrl(QUrl(project_url))
            message.hide()

    def start(self) -> None:
        self._handleNextUploadJob()

    def _handleNextUploadJob(self):
        match self._upload_jobs:
            case [job, *jobs]:
                job.start()
                self._upload_jobs = jobs

    def initializeFileUploadJobMetadata(self) -> Dict[str, Any]:
        metadata = {}
        self._upload_jobs = []
        if "3mf" in self._formats and "3mf" in self._file_handlers and self._file_handlers["3mf"]:
            filename_3mf = self._file_name + ".3mf"
            metadata[filename_3mf] = {
                "export_job_output"   : None,
                "upload_progress"     : -1,
                "upload_status"       : "",
                "file_upload_response": None,
                "file_upload_success_message": getBackwardsCompatibleMessage(
                    text = "'{}' was uploaded to '{}'.".format(filename_3mf, self._library_project_name),
                    title = "Upload successful",
                    message_type_str = "POSITIVE",
                    lifetime = 30
                ),
                "file_upload_failed_message": getBackwardsCompatibleMessage(
                    text = "Failed to upload the file '{}' to '{}'.".format(filename_3mf, self._library_project_name),
                    title = "File upload error",
                    message_type_str = "ERROR",
                    lifetime = 30
                )
            }
            job_3mf = ExportFileJob(self._file_handlers["3mf"], self._nodes, self._file_name, "3mf")
            job_3mf.finished.connect(self._onCuraProjectFileExported)
            self._upload_jobs.append(job_3mf)

        if "ufp" in self._formats and "ufp" in self._file_handlers and self._file_handlers["ufp"]:
            filename_ufp = self._file_name + ".ufp"
            metadata[filename_ufp] = {
                "export_job_output"   : None,
                "upload_progress"     : -1,
                "upload_status"       : "",
                "file_upload_response": None,
                "file_upload_success_message": getBackwardsCompatibleMessage(
                    text = "'{}' was uploaded to '{}'.".format(filename_ufp, self._library_project_name),
                    title = "Upload successful",
                    message_type_str = "POSITIVE",
                    lifetime = 30,
                ),
                "file_upload_failed_message": getBackwardsCompatibleMessage(
                    text = "Failed to upload the file '{}' to '{}'.".format(filename_ufp, self._library_project_name),
                    title = "File upload error",
                    message_type_str = "ERROR",
                    lifetime = 30
                )
            }
            job_ufp = ExportFileJob(self._file_handlers["ufp"], self._nodes, self._file_name, "ufp")
            job_ufp.finished.connect(self._onPrintFileExported)
            self._upload_jobs.append(job_ufp)
        return metadata