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
import fnmatch
import os
import re
import stat

from PyQt6.QtCore import QEventLoop, QObject

from eric7.RemoteServer.EricRequestCategory import EricRequestCategory
from eric7.SystemUtilities import FileSystemUtilities


class EricServerFileSystemInterface(QObject):
    """
    Class implementing the file system interface to the eric-ide server.
    """

    _MagicCheck = re.compile("([*?[])")

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

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

        self.__serverInterface = serverInterface

    def __hasMagic(self, pathname):
        """
        Private method to check, if a given path contains glob style magic characters.

        Note: This was taken from 'glob.glob'.

        @param pathname path name to be checked
        @type str
        @return flag indicating the presence of magic characters
        @rtype bool
        """
        match = self._MagicCheck.search(pathname)
        return match is not None

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

        if self.__serverInterface.isServerConnected():
            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()

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

            loop.exec()
            return ok, error

        else:
            return False, "Not connected to an 'eric-ide' server."

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

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

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

        return listedDirectory, separator, listing

    def direntries(
        self,
        directory,
        filesonly=False,
        pattern=None,
        followsymlinks=True,
        ignore=None,
        recursive=True,
        dirsonly=False,
    ):
        """
        Public method to get a list of all files and directories of a given directory.

        @param directory root of the tree to check
        @type str
        @param filesonly flag indicating that only files are wanted (defaults to False)
        @type bool (optional)
        @param pattern a filename pattern or list of filename patterns to check
            against (defaults to None)
        @type str or list of str (optional)
        @param followsymlinks flag indicating whether symbolic links should be
            followed (defaults to True)
        @type bool (optional)
        @param ignore list of entries to be ignored (defaults to None)
        @type list of str (optional)
        @param recursive flag indicating a recursive search (defaults to True)
        @type bool (optional)
        @param dirsonly flag indicating to return only directories. When True it has
            precedence over the 'filesonly' parameter ((defaults to False)
        @type bool
        @return list of all files and directories in the tree rooted at path.
            The names are expanded to start with the given directory name.
        @rtype list of str
        @exception OSError raised in case the server reported an issue
        """
        loop = QEventLoop()
        ok = False
        error = ""
        result = []

        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 result, ok, error

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

        if self.__serverInterface.isServerConnected():
            self.__serverInterface.sendJson(
                category=EricRequestCategory.FileSystem,
                request="DirEntries",
                params={
                    "directory": FileSystemUtilities.plainFileName(directory),
                    "files_only": filesonly,
                    "pattern": [] if pattern is None else pattern,
                    "follow_symlinks": followsymlinks,
                    "ignore": [] if ignore is None else ignore,
                    "recursive": recursive,
                    "dirs_only": dirsonly,
                },
                callback=callback,
            )

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

        return result

    def glob(self, pathname, recursive=False, includeHidden=False):
        """
        Public method to get a list of of all files matching a given pattern
        like 'glob.glob()'.

        @param pathname path name pattern with simple shell-style wildcards
        @type str
        @param recursive flag indicating a recursive list (defaults to False)
        @type bool (optional)
        @param includeHidden flag indicating to include hidden files (defaults to False)
        @type bool (optional)
        @return list of all files matching the pattern
        @rtype list of str
        """
        result = []

        pathname = FileSystemUtilities.plainFileName(pathname)
        dirname, basename = os.path.split(pathname)
        if dirname and not self.__hasMagic(dirname):
            with contextlib.suppress(OSError):
                entries = self.direntries(
                    dirname, pattern=basename, recursive=recursive, filesonly=True
                )
                if includeHidden:
                    result = entries
                else:
                    result = [e for e in entries if not e.startswith(".")]

        return result

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

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

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

        return stResult

    def isdir(self, name):
        """
        Public method to check, if the given name is a directory.

        @param name name to be checked
        @type str
        @return flag indicating a directory
        @rtype bool
        """
        with contextlib.suppress(KeyError, OSError):
            result = self.stat(name, ["st_mode"])
            return stat.S_ISDIR(result["st_mode"])

        return False

    def isfile(self, name):
        """
        Public method to check, if the given name is a regular file.

        @param name name to be checked
        @type str
        @return flag indicating a regular file
        @rtype bool
        """
        with contextlib.suppress(KeyError, OSError):
            result = self.stat(name, ["st_mode"])
            return stat.S_ISREG(result["st_mode"])

        return False

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

        if self.__serverInterface.isServerConnected():
            self.__serverInterface.sendJson(
                category=EricRequestCategory.FileSystem,
                request="Exists",
                params={"name": FileSystemUtilities.plainFileName(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
        @exception ValueError raised for an illegal modes list
        """
        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()

        if self.__serverInterface.isServerConnected():
            self.__serverInterface.sendJson(
                category=EricRequestCategory.FileSystem,
                request="Access",
                params={
                    "name": FileSystemUtilities.plainFileName(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()

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

            loop.exec()
            return ok, error

        else:
            return False, "Not connected to an 'eric-ide' server."

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

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

            loop.exec()
            return ok, error

        else:
            return False, "Not connected to an 'eric-ide' server."

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

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

            loop.exec()
            return ok, error

        else:
            return False, "Not connected to an 'eric-ide' server."

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

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

            loop.exec()
            return ok, error

        else:
            return False, "Not connected to an 'eric-ide' server."

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

        if not self.__serverInterface.isServerConnected():
            raise OSError("Not connected to an 'eric-ide' server.")

        else:
            self.__serverInterface.sendJson(
                category=EricRequestCategory.FileSystem,
                request="ReadFile",
                params={
                    "filename": FileSystemUtilities.plainFileName(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()

        if not self.__serverInterface.isServerConnected():
            raise OSError("Not connected to an 'eric-ide' server.")

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

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