Skip to content
Snippets Groups Projects
Commit 563ee073 authored by Yngve Levinsen's avatar Yngve Levinsen
Browse files

added ess.nextcloud() class

bumped version to 2.5.0
parent 541e4d8a
No related branches found
Tags 2.5.0
No related merge requests found
Pipeline #20665 passed
__all__ = ["fieldmap", "installed", "lib_tw", "SP_Relativity", "TraceWin", "TTF"]
__version__ = "2.4.2"
from . import TraceWin
from . import installed
from . import lib_tw
from . import fieldmap
from .nextcloud import nextcloud
__all__ = [
"fieldmap",
"installed",
"lib_tw",
"SP_Relativity",
"TraceWin",
"TTF",
"nextcloud",
]
__version__ = "2.5.0"
import os
import getpass
import webdav3.client as wc
import tempfile
import shutil
class nextcloud:
def _request_password(self, username):
self.password = getpass.getpass("Password for {}:".format(username))
def __init__(self, username=None):
self.server = "https://nextcloud.esss.lu.se"
self.username = username
while self.username is None:
for key in ["JUPYTERHUB_USER", "USER", "LOGNAME"]:
if key in os.environ:
self.username = os.environ[key]
self._request_password(self.username)
self.tmpfolder = tempfile.mkdtemp()
def __del__(self):
shutil.rmtree(self.tmpfolder)
def connect(self):
"""
Connect to the server.
This must be called after a new class has been initialized.
"""
options = {
"webdav_hostname": self.server,
"webdav_login": self.username,
"webdav_password": self.password,
"webdav_root": "/remote.php/webdav",
}
self.client = wc.Client(options)
def get_resource(self, filepath):
"""
Copy the file in filepath to a local temporary folder
:param filepath: path to the file
:return: full path to the local copy of the file
"""
check = self.client.check(filepath)
if check:
folder_path = os.path.join(self.tmpfolder, os.path.dirname(filepath))
if not os.path.exists(folder_path):
os.makedirs(folder_path)
local_filepath = os.path.join(folder_path, os.path.basename(filepath))
self.client.download_sync(remote_path=filepath, local_path=local_filepath)
return local_filepath
def get_filelist(self, filepath="."):
"""
Lists files in a folder
:param filepath: path to list of files from
:return: list of files in filepath
"""
check = self.client.check(filepath)
if check:
return self.client.list(filepath)
else:
return []
def _globsearch(self, current, remainders):
"""
A recursive search through a list of paths that are expected to be a folder path.
Each folder/file may include regular expressions used in re.search()
Start a call to this function by self._globsearch('', folders)
:param current: The current file path
:param remainders: the remaining list of folders in the glob search
:return: A list of full paths to all matching files
"""
import re
paths = []
for i in range(len(remainders)):
folder = remainders[i]
if "*" not in folder:
current = os.path.join(current, folder)
else:
for path in self.client.list(current):
if re.search(folder, path):
if i + 1 == len(remainders):
paths.append(os.path.join(current, path))
else:
new_current = os.path.join(current, path)
paths.extend(
self._globsearch(new_current, remainders[i + 1 :])
)
break
return paths
def get_filelist_glob(self, filepath):
"""
Lists files in a path which can include * (may match multiple folders)
:param filepath: glob file path to look for files in
:return: list of all files
"""
# First we need to split in a list of each folder
folders = []
while 1:
filepath, folder = os.path.split(filepath)
if folder != "":
folders.append(folder)
else:
if filepath != "":
folders.append(filepath)
break
folders.reverse()
all_paths = self._globsearch("", folders)
return all_paths
0% Loading or .
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment