ansible-edda/playbooks/roles/music/org/files/nextcloud-upload

94 lines
3.0 KiB
Plaintext
Raw Normal View History

2023-02-19 23:46:17 +01:00
#!/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)