94 lines
3.0 KiB
Plaintext
94 lines
3.0 KiB
Plaintext
|
#!/usr/bin/env python3
|
||
|
|
|
||
|
|
import argparse
|
||
|
|
import os
|
||
|
|
import requests
|
||
|
|
import requests.auth
|
||
|
|
import urllib
|
||
|
|
import urllib.parse
|
||
|
|
import yaml
|
||
|
|
|
||
|
|
|
||
|
|
class WebDavManager:
|
||
|
|
|
||
|
|
def __init__(self, url, user, pswd):
|
||
|
|
self.__url = url
|
||
|
|
self.__auth = requests.auth.HTTPBasicAuth(user, pswd)
|
||
|
|
|
||
|
|
@staticmethod
|
||
|
|
def __remote_path(dstdir, dst):
|
||
|
|
path = []
|
||
|
|
if dstdir is not None:
|
||
|
|
path.append(dstdir.strip('/'))
|
||
|
|
path.append(dst.strip('/'))
|
||
|
|
return '/'.join(path)
|
||
|
|
|
||
|
|
def __url_path(self, dstdir, dst):
|
||
|
|
return '/'.join([
|
||
|
|
self.__url.strip('/'),
|
||
|
|
urllib.parse.quote(self.__remote_path(dstdir, dst)),
|
||
|
|
])
|
||
|
|
|
||
|
|
def mkdir(self, dstdir, dir):
|
||
|
|
print(f"mkdir: {self.__remote_path(dstdir, dir)}", flush=True)
|
||
|
|
|
||
|
|
rc = requests.request("MKCOL", self.__url_path(dstdir, dir), auth=self.__auth)
|
||
|
|
print(f"response: {rc}", flush=True)
|
||
|
|
|
||
|
|
if ((rc.status_code // 100) != 2) and (rc.status_code != 405):
|
||
|
|
raise RuntimeError(f"unexpected response ({rc}) to MKCOL ({rc.url}):\n{rc.text}")
|
||
|
|
|
||
|
|
def upload_file(self, srcdir, dstdir, file):
|
||
|
|
print(f"upload: {self.__remote_path(dstdir, file)}", flush=True)
|
||
|
|
|
||
|
|
with open(os.path.join(srcdir, file), "rb") as fobj:
|
||
|
|
rc = requests.put(self.__url_path(dstdir, file), auth=self.__auth, data=fobj)
|
||
|
|
print(f"response: {rc}", flush=True)
|
||
|
|
|
||
|
|
if (rc.status_code // 100) != 2:
|
||
|
|
raise RuntimeError(f"unexpected response ({rc}) to PUT ({rc.url}):\n{rc.text}")
|
||
|
|
|
||
|
|
@staticmethod
|
||
|
|
def __walkraise(error):
|
||
|
|
raise error
|
||
|
|
|
||
|
|
def upload_dir(self, target):
|
||
|
|
prefix = os.path.dirname(os.path.realpath(target))
|
||
|
|
self.mkdir(None, os.path.relpath(target, prefix))
|
||
|
|
|
||
|
|
for root, dirs, files in os.walk(target, topdown=True, onerror=self.__walkraise):
|
||
|
|
for dir in dirs:
|
||
|
|
self.mkdir(os.path.relpath(root, prefix), dir)
|
||
|
|
for file in files:
|
||
|
|
self.upload_file(root, os.path.relpath(root, prefix), file)
|
||
|
|
|
||
|
|
def upload(self, target):
|
||
|
|
if os.path.isfile(target):
|
||
|
|
self.upload_file(
|
||
|
|
os.path.dirname(os.path.realpath(args.target)),
|
||
|
|
None,
|
||
|
|
os.path.basename(args.target),
|
||
|
|
)
|
||
|
|
else:
|
||
|
|
assert os.path.isdir(target)
|
||
|
|
self.upload_dir(target)
|
||
|
|
|
||
|
|
if __name__ == "__main__":
|
||
|
|
parser = argparse.ArgumentParser(description="upload files to nextcloud")
|
||
|
|
|
||
|
|
parser.add_argument("--config", type=str, default="~/.config/nextcloud-upload/config.yml",
|
||
|
|
help="path to configuration")
|
||
|
|
parser.add_argument("target", type=str, help="file or folder to upload")
|
||
|
|
|
||
|
|
args = parser.parse_args()
|
||
|
|
|
||
|
|
with open(args.config, encoding="utf-8") as config_file:
|
||
|
|
config = yaml.safe_load(config_file)
|
||
|
|
|
||
|
|
for key in ["url", "user", "pswd"]:
|
||
|
|
if key not in config:
|
||
|
|
raise KeyError(f"{key} must be present in {args.config}")
|
||
|
|
|
||
|
|
webdav = WebDavManager(config["url"], config["user"], config["pswd"])
|
||
|
|
webdav.upload(args.target)
|