150 lines
4.1 KiB
Python
150 lines
4.1 KiB
Python
import os
|
|
from datetime import datetime
|
|
from flask import current_app as app
|
|
try:
|
|
from urllib.request import pathname2url
|
|
from urllib.request import URLopener
|
|
from urllib.request import urlparse
|
|
from urllib.error import HTTPError
|
|
except ImportError:
|
|
from urllib import pathname2url
|
|
from urllib import URLopener
|
|
from urlparse import urlparse
|
|
from urllib2 import HTTPError
|
|
|
|
|
|
def download_url(url,filename):
|
|
downloader = URLopener()
|
|
try:
|
|
downloader.retrieve(url, filename)
|
|
except HTTPError as e:
|
|
return (False, ("%s %s"%(e.code,e.reason), e.code))
|
|
return (True, ("OK", 200 ))
|
|
|
|
|
|
def file_date_human(num):
|
|
return datetime.fromtimestamp(
|
|
num
|
|
).strftime(app.config['DATE_FORMAT'])
|
|
|
|
|
|
def file_stat(path, filename):
|
|
full_path = os.path.join(path, filename)
|
|
s = os.stat(full_path)
|
|
return {
|
|
'size': file_size_MB(s.st_size),
|
|
'mtime': file_date_human(s.st_mtime),
|
|
'name': filename,
|
|
'url': path2url(filename)
|
|
}
|
|
|
|
|
|
def file_size_human(num,HTML=True):
|
|
space = ' ' if HTML else ' '
|
|
for x in [space + 'B', 'KB', 'MB', 'GB', 'TB']:
|
|
if num < 1024.0:
|
|
if x == space + 'B':
|
|
return "%d%s%s" % (num, space, x)
|
|
return "%3.1f%s%s" % (num, space, x)
|
|
num /= 1024.0
|
|
|
|
|
|
def file_size_MB(num):
|
|
return "{:,.2f}".format(num/(1024*1024))
|
|
|
|
|
|
def get_folder_size(path):
|
|
total_size = 0
|
|
for dirpath, dirnames, filenames in os.walk(path):
|
|
for f in filenames:
|
|
fp = os.path.join(dirpath, f)
|
|
total_size += os.path.getsize(fp)
|
|
return total_size
|
|
|
|
|
|
def get_or_none(key,d,none = None):
|
|
if key in d:
|
|
return d[key]
|
|
else:
|
|
return none
|
|
|
|
|
|
def get_script_url(public_url, share, end_point, token = "[TOKEN]"):
|
|
cmd = None
|
|
doc = None
|
|
if get_or_none("direct_links", share) and end_point == "download":
|
|
end_point = "direct"
|
|
url = "%s/script/%s/%s/%s"%(
|
|
public_url,
|
|
end_point,
|
|
share['name'],
|
|
token
|
|
)
|
|
if end_point in ( "download", "direct"):
|
|
cmd = 'curl -s %s | bash /dev/stdin [-f]'%( url, )
|
|
doc = 'Download all files in the share. -f to force overwrite existing files.'
|
|
if end_point == "client":
|
|
cmd = 'python <( curl -s %s )'%( url, )
|
|
doc = 'Console client to download and upload files.'
|
|
if end_point == "upload_split":
|
|
cmd = 'curl -s %s | python - [-s split_size_in_Mb] file_to_upload.ext [second.file.ext]'%( url, )
|
|
doc = 'Upload files to the share. -s to set splitting size.'
|
|
return {'cmd': cmd, 'doc': doc}
|
|
|
|
|
|
def is_path_safe(path):
|
|
if path.startswith("."):
|
|
return False
|
|
if "/." in path:
|
|
return False
|
|
return True
|
|
|
|
|
|
def is_valid_url(url, qualifying = None):
|
|
min_attributes = ('scheme', 'netloc')
|
|
qualifying = min_attributes if qualifying is None else qualifying
|
|
token = urlparse(url)
|
|
return all([getattr(token, qualifying_attr)
|
|
for qualifying_attr in qualifying])
|
|
|
|
|
|
def iter_folder_files(path, recursive = True):
|
|
if recursive:
|
|
for dirpath, dirnames, filenames in os.walk(path, topdown = False):
|
|
relative_path = os.path.relpath(dirpath,path)
|
|
dirnames.sort()
|
|
if "/." in relative_path:
|
|
continue
|
|
if relative_path == ".":
|
|
relative_path = ""
|
|
for f in sorted(filenames):
|
|
if f.startswith("."):
|
|
continue
|
|
fp = os.path.join(relative_path, f)
|
|
yield fp
|
|
else:
|
|
for file in sorted(path):
|
|
fp = os.path.join(path,file)
|
|
if os.path.isdir(fp):
|
|
continue
|
|
if file.startswith("."):
|
|
continue
|
|
yield fp
|
|
|
|
|
|
def path2url(path):
|
|
return pathname2url(path)
|
|
|
|
|
|
def safe_name(s):
|
|
return safe_string(s, "-_")
|
|
|
|
|
|
def safe_path(s):
|
|
return safe_string(s, "-_/")
|
|
|
|
|
|
def safe_string(s, valid):
|
|
""" return a safe string, replace non alnum characters with _ . all characters in valid are considered valid. """
|
|
return "".join([c if c.isalnum() or c in valid else "_" for c in s])
|