Mercurial Repositories > eric / file revision

# -*- coding: utf-8 -*-

# Copyright (c) 2024 Detlev Offenbach <detlev@die-offenbachs.de>
#

"""
Module implementing the file system interface to the eric-ide server.
"""

import base64
import contextlib

from PyQt6.QtCore import QEventLoop, QObject

from eric7.RemoteServer.EricRequestCategory import EricRequestCategory


# TODO: sanitize all file name with FileSystemUtilities.plainFileName()
class EricServerFileSystemInterface(QObject):
    """
    Class implementing the file system interface to the eric-ide server.
    """

    def __init__(self, serverInterface):
        """
        Constructor

        @param serverInterface reference to the eric-ide server interface
        @type EricServerInterface
        """
        super().__init__(parent=serverInterface)

        self.__serverInterface = serverInterface

    def getcwd(self):
        """
        Public method to get the current working directory of the eric-ide server.

        @return current working directory of the eric-ide server
        @rtype str
        """
        loop = QEventLoop()
        cwd = ""

        def callback(reply, params):
            """
            Function to handle the server reply

            @param reply name of the server reply
            @type str
            @param params dictionary containing the reply data
            @type dict
            """
            nonlocal cwd

            if reply == "Getcwd":
                cwd = params["directory"]
                loop.quit()

        self.__serverInterface.sendJson(
            category=EricRequestCategory.FileSystem,
            request="Getcwd",
            params={},
            callback=callback,
        )

        loop.exec()
        return cwd

    def chdir(self, directory):
        """
        Public method to change the current working directory of the eric-ide server.

        @param directory absolute path of the working directory to change to
        @type str
        @return tuple containing an OK flag and an error string in case of an issue
        @rtype tuple of (bool, str)
        """
        loop = QEventLoop()
        ok = False
        error = ""

        def callback(reply, params):
            """
            Function to handle the server reply

            @param reply name of the server reply
            @type str
            @param params dictionary containing the reply data
            @type dict
            """
            nonlocal ok, error

            if reply == "Chdir":
                ok = params["ok"]
                with contextlib.suppress(KeyError):
                    error = params["error"]
                loop.quit()

        self.__serverInterface.sendJson(
            category=EricRequestCategory.FileSystem,
            request="Chdir",
            params={"directory": directory},
            callback=callback,
        )

        loop.exec()
        return ok, error

    def listdir(self, directory=""):
        """
        Public method to get a directory listing.

        @param directory directory to be listed. An empty directory means to list
            the eric-ide server current directory. (defaults to "")
        @type str (optional)
        @return tuple containing the listed directory, the path separartor and the
            directory listing. Each directory listing entry contains a dictionary
            with the relevant data.
        @rtype tuple of (str, str, dict)
        @exception OSError raised in case the server reported an issue
        """
        if directory is None:
            # sanitize the directory in case it is None
            directory = ""

        loop = QEventLoop()
        ok = False
        error = ""
        listedDirectory = ""
        separator = ""
        listing = []

        def callback(reply, params):
            """
            Function to handle the server reply

            @param reply name of the server reply
            @type str
            @param params dictionary containing the reply data
            @type dict
            """
            nonlocal listedDirectory, listing, separator, ok, error

            if reply == "Listdir":
                ok = params["ok"]
                if ok:
                    listedDirectory = params["directory"]
                    listing = params["listing"]
                    separator = params["separator"]
                else:
                    error = params["error"]
                loop.quit()

        self.__serverInterface.sendJson(
            category=EricRequestCategory.FileSystem,
            request="Listdir",
            params={"directory": directory},
            callback=callback,
        )

        loop.exec()
        if not ok:
            raise OSError(error)

        return listedDirectory, separator, listing

    def stat(self, filename, stNames):
        """
        Public method to get the status of a file.

        @param filename name of the file
        @type str
        @param stNames list of 'stat_result' members to retrieve
        @type list of str
        @return dictionary containing the requested status data
        @rtype dict
        @exception OSError raised in case the server reported an issue
        """
        loop = QEventLoop()
        ok = False
        error = ""
        stResult = {}

        def callback(reply, params):
            """
            Function to handle the server reply

            @param reply name of the server reply
            @type str
            @param params dictionary containing the reply data
            @type dict
            """
            nonlocal ok, error, stResult

            if reply == "Stat":
                ok = params["ok"]
                if ok:
                    stResult = params["result"]
                else:
                    error = params["error"]
                loop.quit()

        self.__serverInterface.sendJson(
            category=EricRequestCategory.FileSystem,
            request="Stat",
            params={"filename": filename, "st_names": stNames},
            callback=callback,
        )

        loop.exec()
        if not ok:
            raise OSError(error)

        return stResult

    def exists(self, name):
        """
        Public method the existence of a file or directory.

        @param name name of the file or directory
        @type str
        @return flag indicating the file existence
        @rtype bool
        """
        loop = QEventLoop()
        nameExists = False

        def callback(reply, params):
            """
            Function to handle the server reply

            @param reply name of the server reply
            @type str
            @param params dictionary containing the reply data
            @type dict
            """
            nonlocal nameExists

            if reply == "Exists":
                nameExists = params["exists"]
                loop.quit()

        self.__serverInterface.sendJson(
            category=EricRequestCategory.FileSystem,
            request="Exists",
            params={"name": name},
            callback=callback,
        )

        loop.exec()
        return nameExists

    def access(self, name, modes):
        """
        Public method to test the given access rights to a file or directory.

        The modes to check for are 'read', 'write' or 'execute' or any combination.

        @param name name of the file or directory
        @type str
        @param modes list of modes to check for
        @type str or list of str
        @return flag indicating the user has the asked for permissions
        @rtype bool
        """
        if not modes:
            raise ValueError(
                "At least one of 'read', 'write' or 'execute' must be specified."
            )

        if isinstance(modes, str):
            # convert to a list with one element
            modes = [modes]

        loop = QEventLoop()
        accessOK = False

        def callback(reply, params):
            """
            Function to handle the server reply

            @param reply name of the server reply
            @type str
            @param params dictionary containing the reply data
            @type dict
            """
            nonlocal accessOK

            if reply == "Access":
                accessOK = params["ok"]
                loop.quit()

        self.__serverInterface.sendJson(
            category=EricRequestCategory.FileSystem,
            request="Access",
            params={"name": name, "modes":modes},
            callback=callback,
        )

        loop.exec()
        return accessOK

    def mkdir(self, directory):
        """
        Public method to create a new directory on the eric-ide server.

        @param directory absolute path of the new directory
        @type str
        @return tuple containing an OK flag and an error string in case of an issue
        @rtype tuple of (bool, str)
        """
        loop = QEventLoop()
        ok = False
        error = ""

        def callback(reply, params):
            """
            Function to handle the server reply

            @param reply name of the server reply
            @type str
            @param params dictionary containing the reply data
            @type dict
            """
            nonlocal ok, error

            if reply == "Mkdir":
                ok = params["ok"]
                with contextlib.suppress(KeyError):
                    error = params["error"]
                loop.quit()

        self.__serverInterface.sendJson(
            category=EricRequestCategory.FileSystem,
            request="Mkdir",
            params={"directory": directory},
            callback=callback,
        )

        loop.exec()
        return ok, error

    def rmdir(self, directory):
        """
        Public method to delete a directory on the eric-ide server.

        @param directory absolute path of the directory
        @type str
        @return tuple containing an OK flag and an error string in case of an issue
        @rtype tuple of (bool, str)
        """
        loop = QEventLoop()
        ok = False
        error = ""

        def callback(reply, params):
            """
            Function to handle the server reply

            @param reply name of the server reply
            @type str
            @param params dictionary containing the reply data
            @type dict
            """
            nonlocal ok, error

            if reply == "Rmdir":
                ok = params["ok"]
                with contextlib.suppress(KeyError):
                    error = params["error"]
                loop.quit()

        self.__serverInterface.sendJson(
            category=EricRequestCategory.FileSystem,
            request="Rmdir",
            params={"directory": directory},
            callback=callback,
        )

        loop.exec()
        return ok, error

    def replace(self, oldName, newName):
        """
        Public method to rename a file or directory.

        @param oldName current name of the file or directory
        @type str
        @param newName new name for the file or directory
        @type str
        @return tuple containing an OK flag and an error string in case of an issue
        @rtype tuple of (bool, str)
        """
        loop = QEventLoop()
        ok = False
        error = ""

        def callback(reply, params):
            """
            Function to handle the server reply

            @param reply name of the server reply
            @type str
            @param params dictionary containing the reply data
            @type dict
            """
            nonlocal ok, error

            if reply == "Replace":
                ok = params["ok"]
                with contextlib.suppress(KeyError):
                    error = params["error"]
                loop.quit()

        self.__serverInterface.sendJson(
            category=EricRequestCategory.FileSystem,
            request="Replace",
            params={"old_name": oldName, "new_name": newName},
            callback=callback,
        )

        loop.exec()
        return ok, error

    def remove(self, filename):
        """
        Public method to delete a file on the eric-ide server.

        @param filename absolute path of the file
        @type str
        @return tuple containing an OK flag and an error string in case of an issue
        @rtype tuple of (bool, str)
        """
        loop = QEventLoop()
        ok = False
        error = ""

        def callback(reply, params):
            """
            Function to handle the server reply

            @param reply name of the server reply
            @type str
            @param params dictionary containing the reply data
            @type dict
            """
            nonlocal ok, error

            if reply == "Remove":
                ok = params["ok"]
                with contextlib.suppress(KeyError):
                    error = params["error"]
                loop.quit()

        self.__serverInterface.sendJson(
            category=EricRequestCategory.FileSystem,
            request="Remove",
            params={"filename": filename},
            callback=callback,
        )

        loop.exec()
        return ok, error

    #######################################################################
    ## Methods for reading and writing files
    #######################################################################

    def readFile(self, filename, create=False):
        """
        Public method to read a file from the eric-ide server.

        @param filename name of the file to read
        @type str
        @param create flag indicating to create an empty file, if it does not exist
            (defaults to False)
        @type bool (optional)
        @return bytes data read from the eric-ide server
        @rtype bytes
        @exception OSError raised in case the server reported an issue
        """
        loop = QEventLoop()
        ok = False
        error = ""
        bText = b""

        def callback(reply, params):
            """
            Function to handle the server reply

            @param reply name of the server reply
            @type str
            @param params dictionary containing the reply data
            @type dict
            """
            nonlocal ok, error, bText

            if reply == "ReadFile":
                ok = params["ok"]
                if ok:
                    bText = base64.b85decode(
                        bytes(params["filedata"], encoding="ascii")
                    )
                else:
                    error = params["error"]
                loop.quit()

        self.__serverInterface.sendJson(
            category=EricRequestCategory.FileSystem,
            request="ReadFile",
            params={"filename": filename, "create": create},
            callback=callback,
        )

        loop.exec()
        if not ok:
            raise OSError(error)

        return bText

    def writeFile(self, filename, data, withBackup=False):
        """
        Public method to write the data to a file on the eric-ide server.

        @param filename name of the file to write
        @type str
        @param data data to be written
        @type bytes
        @param withBackup flag indicating to create a backup file first
            (defaults to False)
        @type bool (optional)
        @exception OSError raised in case the server reported an issue
        """
        loop = QEventLoop()
        ok = False
        error = ""

        def callback(reply, params):
            """
            Function to handle the server reply

            @param reply name of the server reply
            @type str
            @param params dictionary containing the reply data
            @type dict
            """
            nonlocal ok, error

            if reply == "WriteFile":
                ok = params["ok"]
                with contextlib.suppress(KeyError):
                    error = params["error"]
                loop.quit()

        self.__serverInterface.sendJson(
            category=EricRequestCategory.FileSystem,
            request="WriteFile",
            params={
                "filename": filename,
                "filedata": str(base64.b85encode(data), encoding="ascii"),
                "with_backup": withBackup,
            },
            callback=callback,
        )

        loop.exec()
        if not ok:
            raise OSError(error)