#!/usr/bin/env python3 """Build a placeholder-backed local source tree from Nextcloud WebDAV. This creates only tiny placeholder files plus a metadata mapping; no Nextcloud original media is downloaded. The authenticated Nextcloud user is never used as an album name. """ from __future__ import annotations import argparse import base64 import getpass import json import os import shutil import ssl import sys import tempfile import urllib.error import urllib.parse import urllib.request import xml.etree.ElementTree as ET from pathlib import Path, PurePosixPath DAV = "DAV:" OC = "http://owncloud.org/ns" SUPPORTED_IMAGE_EXTENSIONS = {".jpg", ".jpeg", ".png", ".gif", ".webp"} PLACEHOLDER = base64.b64decode("R0lGODlhAQABAIAAAAAAAP///ywAAAAAAQABAAACAUwAOw==") def fail(message: str) -> None: raise RuntimeError(message) def validate_relative(value: str) -> str: value = str(value).strip("/") parts = PurePosixPath(value).parts if ".." in parts: raise ValueError(f"unsafe WebDAV path: {value!r}") return value def quote_path(value: str) -> str: value = validate_relative(value) return "/".join(urllib.parse.quote(part, safe="") for part in PurePosixPath(value).parts) def safe_local_name(name: str) -> str: if not name or name in {".", ".."} or "/" in name or "\x00" in name: raise ValueError(f"unsafe remote name: {name!r}") return name class WebDavClient: def __init__(self, base_url: str, user: str, password: str, timeout: int = 30) -> None: self.base_url = base_url.rstrip("/") self.user = user self.password = password self.timeout = timeout token = base64.b64encode(f"{user}:{password}".encode("utf-8")).decode("ascii") self.auth_header = f"Basic {token}" self.context = ssl.create_default_context() def collection_url(self, relative: str) -> str: user = urllib.parse.quote(self.user, safe="") suffix = quote_path(relative) url = f"{self.base_url}/remote.php/dav/files/{user}/" if suffix: url += suffix + "/" return url def list_collection(self, relative: str) -> tuple[dict[str, object], list[dict[str, object]]]: relative = validate_relative(relative) url = self.collection_url(relative) body = ( '' '' '' '' ).encode("utf-8") request = urllib.request.Request(url, data=body, method="PROPFIND") request.add_header("Authorization", self.auth_header) request.add_header("Depth", "1") request.add_header("Content-Type", "application/xml; charset=utf-8") try: with urllib.request.urlopen(request, timeout=self.timeout, context=self.context) as response: status = response.status payload = response.read() except urllib.error.HTTPError as error: if error.code in {401, 403}: fail("Nextcloud rejected the WebDAV credentials or directory access") fail(f"Nextcloud PROPFIND failed with HTTP {error.code}") except urllib.error.URLError as error: fail(f"Nextcloud WebDAV is unreachable: {error.reason}") if status != 207: fail(f"Nextcloud PROPFIND returned HTTP {status}") try: root = ET.fromstring(payload) except ET.ParseError as error: raise RuntimeError("Nextcloud returned invalid WebDAV XML") from error base_path = urllib.parse.unquote(urllib.parse.urlparse(url).path).rstrip("/") current: dict[str, object] | None = None children: list[dict[str, object]] = [] for response in root.findall(f"{{{DAV}}}response"): href = response.findtext(f"{{{DAV}}}href", default="") href_path = urllib.parse.unquote(urllib.parse.urlparse(href).path).rstrip("/") prop = None for propstat in response.findall(f"{{{DAV}}}propstat"): status_text = propstat.findtext(f"{{{DAV}}}status", default="") if " 200 " in status_text: prop = propstat.find(f"{{{DAV}}}prop") if prop is not None: break if prop is None: continue display = prop.findtext(f"{{{DAV}}}displayname", default="") fileid_text = prop.findtext(f"{{{OC}}}fileid", default="").strip() resource_type = prop.find(f"{{{DAV}}}resourcetype") is_dir = resource_type is not None and resource_type.find(f"{{{DAV}}}collection") is not None item = { "display_name": display, "fileid": int(fileid_text) if fileid_text.isdigit() else 0, "is_dir": is_dir, "content_type": prop.findtext(f"{{{DAV}}}getcontenttype", default=""), "size": int(prop.findtext(f"{{{DAV}}}getcontentlength", default="0") or 0), "etag": prop.findtext(f"{{{DAV}}}getetag", default="").strip('"'), } if href_path == base_path: current = item continue if display: children.append(item) if current is None or int(current.get("fileid", 0)) < 1: fail(f"Nextcloud returned no stable fileid for {relative or '/'}") return current, children def link_placeholder(seed: Path, target: Path) -> None: target.parent.mkdir(parents=True, exist_ok=True) try: os.link(seed, target) except OSError: target.write_bytes(PLACEHOLDER) def build_root(client: WebDavClient, remote_root: str, local_root: Path, seed: Path, mapping: dict[str, dict[str, object]]) -> tuple[int, int, int]: files = 0 folders = 0 skipped = 0 stack: list[tuple[str, Path]] = [(validate_relative(remote_root), local_root)] visited: set[str] = set() while stack: remote_dir, local_dir = stack.pop() if remote_dir in visited: continue visited.add(remote_dir) current, children = client.list_collection(remote_dir) local_dir.mkdir(parents=True, exist_ok=True) folders += 1 mapping[str(local_dir)] = { "kind": "folder", "fileid": int(current["fileid"]), "webdav_path": remote_dir, "display_name": str(current.get("display_name", "")), } for child in children: name = safe_local_name(str(child["display_name"])) child_remote = name if remote_dir == "" else f"{remote_dir}/{name}" child_local = local_dir / name if bool(child["is_dir"]): stack.append((child_remote, child_local)) continue extension = Path(name).suffix.lower() if extension not in SUPPORTED_IMAGE_EXTENSIONS: skipped += 1 continue link_placeholder(seed, child_local) files += 1 mapping[str(child_local)] = { "kind": "file", "fileid": int(child.get("fileid", 0)), "webdav_path": child_remote, "display_name": name, "content_type": str(child.get("content_type", "")), "size": int(child.get("size", 0)), "etag": str(child.get("etag", "")), } return files, folders, skipped def atomic_json(path: Path, payload: object) -> None: path.parent.mkdir(parents=True, exist_ok=True) with tempfile.NamedTemporaryFile("w", encoding="utf-8", dir=path.parent, delete=False) as handle: json.dump(payload, handle, ensure_ascii=False, indent=2, sort_keys=True) handle.write("\n") temporary = Path(handle.name) temporary.replace(path) def atomic_text(path: Path, text: str) -> None: path.parent.mkdir(parents=True, exist_ok=True) with tempfile.NamedTemporaryFile("w", encoding="utf-8", dir=path.parent, delete=False) as handle: handle.write(text) temporary = Path(handle.name) temporary.replace(path) def main() -> int: parser = argparse.ArgumentParser() parser.add_argument("--base-url", required=True) parser.add_argument("--user", required=True) parser.add_argument("--password-file", type=Path) parser.add_argument("--root", action="append", required=True, help="WebDAV path relative to the authenticated user's files root") parser.add_argument("--source-dir", required=True, type=Path) parser.add_argument("--manifest", required=True, type=Path) parser.add_argument("--mapping", required=True, type=Path) parser.add_argument("--timeout", type=int, default=30) args = parser.parse_args() try: if args.password_file: password = args.password_file.read_text(encoding="utf-8").rstrip("\r\n") else: password = getpass.getpass("Nextcloud password: ") if not password: fail("Nextcloud password is empty") source_dir = args.source_dir.resolve() staging = source_dir.with_name(f".{source_dir.name}.next") previous = source_dir.with_name(f".{source_dir.name}.previous") seed = source_dir.parent / ".bratonien-webdav-placeholder.gif" source_dir.parent.mkdir(parents=True, exist_ok=True) seed.write_bytes(PLACEHOLDER) os.chmod(seed, 0o644) if staging.exists(): shutil.rmtree(staging) staging.mkdir(parents=True) client = WebDavClient(args.base_url, args.user, password, max(1, args.timeout)) mapping: dict[str, dict[str, object]] = {} manifest: list[str] = [] total_files = total_folders = total_skipped = 0 used_fileids: set[int] = set() for remote_root_raw in args.root: remote_root = validate_relative(remote_root_raw) current, root_children = client.list_collection(remote_root) fileid = int(current["fileid"]) if fileid in used_fileids: fail(f"duplicate selected Nextcloud root fileid: {fileid}") used_fileids.add(fileid) local_name = f"root-{fileid}" local_root = staging / local_name files, folders, skipped = build_root(client, remote_root, local_root, seed, mapping) total_files += files total_folders += folders total_skipped += skipped if remote_root == "": for child in sorted(root_children, key=lambda item: str(item.get("display_name", "")).casefold()): name = safe_local_name(str(child.get("display_name", ""))) child_fileid = int(child.get("fileid", 0)) if child_fileid < 1: fail(f"Nextcloud returned no stable fileid for root child {name!r}") child_path = source_dir / local_name / name if bool(child.get("is_dir")): manifest.append(f"webdav:{child_fileid}\tfolder\t{name}\t{child_path}") elif Path(name).suffix.lower() in SUPPORTED_IMAGE_EXTENSIONS: manifest.append(f"webdav:{child_fileid}\tfile\t{name}\t{child_path}") continue display = str(current.get("display_name", "")).strip() or PurePosixPath(remote_root).name manifest.append(f"webdav:{fileid}\tfolder\t{display}\t{source_dir / local_name}") if previous.exists(): shutil.rmtree(previous) if source_dir.exists(): source_dir.rename(previous) try: staging.rename(source_dir) except Exception: if source_dir.exists(): shutil.rmtree(source_dir) if previous.exists(): previous.rename(source_dir) raise if previous.exists(): shutil.rmtree(previous) final_mapping: dict[str, dict[str, object]] = {} staging_text = str(staging) source_text = str(source_dir) for path, data in mapping.items(): final_path = source_text + path[len(staging_text):] if path.startswith(staging_text) else path final_mapping[final_path] = data atomic_text(args.manifest, "\n".join(manifest) + "\n") atomic_json(args.mapping, { "version": 1, "base_url": args.base_url.rstrip("/"), "user": args.user, "files": final_mapping, }) print(json.dumps({ "roots": len(args.root), "files": total_files, "folders": total_folders, "skipped": total_skipped, "source_dir": str(source_dir), "manifest": str(args.manifest), "mapping": str(args.mapping), }, ensure_ascii=False)) return 0 except Exception as error: print(f"webdav-placeholder: {error}", file=sys.stderr) return 1 if __name__ == "__main__": raise SystemExit(main())