--- a/src/eric7/RemoteServerInterface/EricServerFileSystemInterface.py Fri Feb 23 16:52:01 2024 +0100 +++ b/src/eric7/RemoteServerInterface/EricServerFileSystemInterface.py Mon Feb 26 10:41:10 2024 +0100 @@ -20,6 +20,18 @@ from eric7.SystemUtilities import FileSystemUtilities +class EricServerNotConnectedError(OSError): + """ + Class defining a special OSError indicating a missing server connection. + """ + + def __init__(self): + """ + Constructor + """ + super().__init("Not connected to an 'eric-ide' server.") + + class EricServerFileSystemInterface(QObject): """ Class implementing the file system interface to the eric-ide server. @@ -27,6 +39,8 @@ _MagicCheck = re.compile("([*?[])") + NotConnectedMessage = "Not connected to an 'eric-ide' server." + def __init__(self, serverInterface): """ Constructor @@ -143,7 +157,7 @@ loop.exec() - return cwd + return FileSystemUtilities.remoteFileName(cwd) def chdir(self, directory): """ @@ -187,7 +201,7 @@ return ok, error else: - return False, "Not connected to an 'eric-ide' server." + return False, EricServerFileSystemInterface.NotConnectedMessage def listdir(self, directory=""): """ @@ -196,7 +210,7 @@ @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 + @return tuple containing the listed directory, the path separator and the directory listing. Each directory listing entry contains a dictionary with the relevant data. @rtype tuple of (str, str, dict) @@ -330,7 +344,7 @@ if not ok: raise OSError(error) - return result + return [FileSystemUtilities.remoteFileName(r) for r in result] def glob(self, pathname, recursive=False, includeHidden=False): """ @@ -587,7 +601,7 @@ return ok, error else: - return False, "Not connected to an 'eric-ide' server." + return False, EricServerFileSystemInterface.NotConnectedMessage def makedirs(self, directory, exist_ok=False): """ @@ -638,7 +652,7 @@ return ok, error else: - return False, "Not connected to an 'eric-ide' server." + return False, EricServerFileSystemInterface.NotConnectedMessage def rmdir(self, directory): """ @@ -682,7 +696,7 @@ return ok, error else: - return False, "Not connected to an 'eric-ide' server." + return False, EricServerFileSystemInterface.NotConnectedMessage def replace(self, oldName, newName): """ @@ -731,7 +745,7 @@ return ok, error else: - return False, "Not connected to an 'eric-ide' server." + return False, EricServerFileSystemInterface.NotConnectedMessage def remove(self, filename): """ @@ -775,7 +789,7 @@ return ok, error else: - return False, "Not connected to an 'eric-ide' server." + return False, EricServerFileSystemInterface.NotConnectedMessage def expanduser(self, name): """ @@ -836,6 +850,41 @@ """ return self.__serverPathSep + def isabs(self, p): + """ + Public method to chack a path for being an absolute path. + + @param p path to be checked + @type str + @return flag indicating an absolute path + @rtype bool + """ + if self.__serverInterface.isServerConnected(): + if self.__serverPathSep == "\\": + s = FileSystemUtilities.plainFileName(p)[:3].replace("/", "\\") + return s.startswith("\\)") or s.startswith(":\\", 1) + else: + return FileSystemUtilities.plainFileName(p).startswith("/") + else: + return os.path.isabs(p) + + def abspath(self, p): + """ + Public method to convert the given path to an absolute path. + + @param p path to be converted + @type str + @return absolute path + @rtype str + """ + if self.__serverInterface.isServerConnected(): + p = FileSystemUtilities.plainFileName(p) + if not self.isabs(p): + p = self.join(self.getcwd(), p) + return FileSystemUtilities.remoteFileName(p) + else: + return os.path.abspath(p) + def join(self, a, *p): """ Public method to join two or more path name components using the path separator @@ -990,6 +1039,8 @@ @type bool (optional) @return bytes data read from the eric-ide server @rtype bytes + @exception EricServerNotConnectedError raised to indicate a missing server + connection @exception OSError raised in case the server reported an issue """ loop = QEventLoop() @@ -1019,7 +1070,7 @@ loop.quit() if not self.__serverInterface.isServerConnected(): - raise OSError("Not connected to an 'eric-ide' server.") + raise EricServerNotConnectedError() else: self.__serverInterface.sendJson( @@ -1049,6 +1100,8 @@ @param withBackup flag indicating to create a backup file first (defaults to False) @type bool (optional) + @exception EricServerNotConnectedError raised to indicate a missing server + connection @exception OSError raised in case the server reported an issue """ loop = QEventLoop() @@ -1073,7 +1126,7 @@ loop.quit() if not self.__serverInterface.isServerConnected(): - raise OSError("Not connected to an 'eric-ide' server.") + raise EricServerNotConnectedError else: self.__serverInterface.sendJson( @@ -1156,6 +1209,19 @@ ####################################################################### def shutilCopy(self, srcName, dstName): + """ + Public method to copy a source file to a given destination file or directory. + + @param srcName name of the source file + @type str + @param dstName name of the destination file or directory + @type str + @return name of the destination file + @rtype str + @exception EricServerNotConnectedError raised to indicate a missing server + connection + @exception OSError raised to indicate an issue + """ loop = QEventLoop() ok = False error = "" @@ -1181,7 +1247,7 @@ loop.quit() if not self.__serverInterface.isServerConnected(): - raise OSError("Not connected to an 'eric-ide' server.") + raise EricServerNotConnectedError else: self.__serverInterface.sendJson( @@ -1199,3 +1265,102 @@ raise OSError(error) return dst + + def shutilRmtree(self, pathname, ignore_errors=False): + """ + Public method to delete an entire directory tree. + + @param pathname name of the directory to be deleted + @type str + @param ignore_errors flag indicating to ignore error resulting from failed + removals (defaults to False) + @type bool (optional) + @exception EricServerNotConnectedError raised to indicate a missing server + connection + @exception OSError raised to indicate 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 == "ShutilRmtree": + ok = params["ok"] + if not ok: + error = params["error"] + loop.quit() + + if not self.__serverInterface.isServerConnected(): + raise EricServerNotConnectedError + + else: + self.__serverInterface.sendJson( + category=EricRequestCategory.FileSystem, + request="ShutilRmtree", + params={ + "name": FileSystemUtilities.plainFileName(pathname), + "ignore_errors": ignore_errors, + }, + callback=callback, + ) + + loop.exec() + if not ok: + raise OSError(error) + + ####################################################################### + ## Utility methods. + ####################################################################### + + def compactPath(self, longPath, width, measure=len): + """ + Public method to return a compacted path fitting inside the given width. + + @param longPath path to be compacted + @type str + @param width width for the compacted path + @type int + @param measure reference to a function used to measure the length of the + string (defaults to len) + @type function (optional) + @return compacted path + @rtype str + """ + if measure(longPath) <= width: + return longPath + + ellipsis = "..." + + head, tail = self.split(longPath) + mid = len(head) // 2 + head1 = head[:mid] + head2 = head[mid:] + while head1: + # head1 is same size as head2 or one shorter + cpath = self.join(f"{head1}{ellipsis}{head2}", tail) + if measure(cpath) <= width: + return cpath + head1 = head1[:-1] + head2 = head2[1:] + cpath = self.join(ellipsis, tail) + if measure(cpath) <= width: + return cpath + remoteMarker = FileSystemUtilities.remoteFileName("") + if width <= len(remoteMarker): + return f"{remoteMarker}{ellipsis}{tail}" + while tail: + cpath = f"{remoteMarker}{ellipsis}{tail}" + if measure(cpath) <= width: + return cpath + tail = tail[1:] + return ""