in source/soca/cluster_web_ui/views/my_files.py [0:0]
def index():
try:
timestamp = datetime.datetime.utcnow().strftime("%s")
path = request.args.get("path", None)
logger.info(f"Accessing {path}")
ts = request.args.get("ts", None)
if ts is None:
if path is None:
return redirect("/my_files?ts="+timestamp)
else:
return redirect("/my_files?path="+path+"&ts=" + timestamp)
filesystem = {}
breadcrumb = {}
if path is None:
path = config.Config.USER_HOME + "/" + session["user"]
else:
path = path
# Clean Path
if path != "/":
if path.endswith("/"):
return redirect("/my_files?path=" + path[:-1])
if ".." in path:
return redirect("/my_files")
if user_has_permission(path, "read", "folder") is False:
if path == config.Config.USER_HOME + "/" + session["user"]:
flash("We cannot access to your own home directory. Please ask a admin to rollback your folder ACLs to 750")
return redirect("/")
else:
flash("You are not authorized to access this location and/or this path is restricted by the HPC admin. If you recently changed the permissions, please allow up to 10 minutes for sync.", "error")
return redirect("/my_files")
# Build breadcrumb
count = 1
for level in path.split("/"):
if level == "":
breadcrumb["/"] = "root"
else:
breadcrumb["/".join(path.split('/')[:count])] = level
count += 1
# Retrieve files/folders
if CACHE_FOLDER_CONTENT_PREFIX + path not in cache.keys():
is_cached = False
try:
for entry in os.scandir(path):
if not entry.name.startswith("."):
try:
filesystem[entry.name] = {"path": path + "/" + entry.name,
"uid": encrypt(path + "/" + entry.name, entry.stat().st_size)["message"],
"type": "folder" if entry.is_dir() else "file",
"st_size": convert_size(entry.stat().st_size),
"st_size_default": entry.stat().st_size,
"st_mtime": entry.stat().st_mtime
}
except Exception as err:
# most likely symbolic link pointing to wrong location
flash("{} returned an error and cannot be displayed: {}".format(entry.name, err))
cache[CACHE_FOLDER_CONTENT_PREFIX + path] = filesystem
except Exception as err:
if err.errno == errno.EPERM:
flash("Sorry we could not access this location due to a permission error. If you recently changed the permissions, please allow up to 10 minutes for sync.", "error")
elif err.errno == errno.ENOENT:
flash("Could not locate the directory. Did you delete it ?", "error")
else:
flash("Could not locate the directory: " + str (err), "error")
return redirect("/my_files")
else:
is_cached = True
filesystem = cache[CACHE_FOLDER_CONTENT_PREFIX + path]
get_all_uid = [file_info['uid'] for file_info in filesystem.values() if file_info["type"] == "file"]
return render_template('my_files.html', user=session["user"],
filesystem=OrderedDict(sorted(filesystem.items(), key=lambda t: t[0].lower())),
get_all_uid=base64.b64encode(",".join(get_all_uid).encode()).decode(),
get_all_uid_count=len(get_all_uid),
breadcrumb=breadcrumb,
max_upload_size=config.Config.MAX_UPLOAD_FILE,
max_upload_timeout=config.Config.MAX_UPLOAD_TIMEOUT,
max_online_preview=config.Config.MAX_SIZE_ONLINE_PREVIEW,
default_cache_time=config.Config.DEFAULT_CACHE_TIME,
path=path,
page="my_files",
is_cached=is_cached,
timestamp=timestamp)
except Exception as err:
flash("Error, this path probably does not exist. "+str(err), "error")
logger.error(err)
return redirect("/my_files")