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

#!/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=f"{os.path.expanduser('~')}/.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)