From 73e4b08dbd789c76f8f1b33d3ee5ab053ae292f6 Mon Sep 17 00:00:00 2001 From: Terranom674 Date: Tue, 18 Aug 2026 17:04:20 +0200 Subject: [PATCH] 0.9.5.1: benutzerbezogenes Manifest ohne Showcase-View --- runtime/lib/build_user_manifest.py | 142 +++++++++++++++++++++++++++++ 1 file changed, 142 insertions(+) create mode 100644 runtime/lib/build_user_manifest.py diff --git a/runtime/lib/build_user_manifest.py b/runtime/lib/build_user_manifest.py new file mode 100644 index 0000000..51a0065 --- /dev/null +++ b/runtime/lib/build_user_manifest.py @@ -0,0 +1,142 @@ +#!/usr/bin/env python3 +"""Build a manifest from one connection's explicitly selected user filesystem roots. + +This mode deliberately does not read the legacy Showcase source view. The storage +configuration must already resolve to the authenticated Nextcloud user's local +home/files tree. Only the selected include prefixes are exposed to Piwigo. +""" + +from __future__ import annotations + +import argparse +import hashlib +import json +import sys +import tempfile +from pathlib import Path, PurePosixPath + + +def safe_relative(value: str) -> str: + value = str(value).strip("/") + parts = PurePosixPath(value).parts + if ".." in parts: + raise ValueError(f"unsafe relative path: {value!r}") + return value + + +def contained(root: Path, relative: str) -> Path: + relative = safe_relative(relative) + candidate = root.joinpath(*PurePosixPath(relative).parts) if relative else root + root_real = root.resolve() + candidate_real = candidate.resolve() + try: + candidate_real.relative_to(root_real) + except ValueError as error: + raise ValueError(f"path escapes configured user root: {relative!r}") from error + return candidate + + +def read_config(path: Path) -> list[tuple[str, str, Path, str]]: + rows: list[tuple[str, str, Path, str]] = [] + with path.open(encoding="utf-8") as handle: + for number, raw in enumerate(handle, 1): + line = raw.rstrip("\n") + if not line or line.startswith("#"): + continue + fields = line.split("\t") + if len(fields) not in {3, 4}: + raise ValueError( + f"{path}:{number}: expected storage_id, source_prefix, local_mount and optional include_prefix" + ) + storage_id, source_prefix, local_mount = fields[:3] + include_prefix = fields[3] if len(fields) == 4 else "" + source_prefix = safe_relative(source_prefix) + include_prefix = safe_relative(include_prefix) + mount = Path(local_mount) + if not storage_id.strip(): + raise ValueError(f"{path}:{number}: storage_id is empty") + if not mount.is_absolute(): + raise ValueError(f"{path}:{number}: local_mount must be absolute") + rows.append((storage_id.strip(), source_prefix, mount, include_prefix)) + if not rows: + raise ValueError("no user storage mappings configured") + return rows + + +def stable_id(source: Path) -> str: + digest = hashlib.sha256(str(source.resolve()).encode("utf-8")).hexdigest()[:24] + return f"user-{digest}" + + +def manifest_entry(source: Path) -> str: + if source.is_symlink(): + raise ValueError(f"selected source must not be a symlink: {source}") + if source.is_dir(): + item_type = "folder" + elif source.is_file(): + item_type = "file" + else: + raise FileNotFoundError(f"selected source is unavailable: {source}") + name = source.name + if not name: + raise ValueError(f"selected source has no display name: {source}") + for value in (name, str(source)): + if any(char in value for char in ("\t", "\n", "\r")): + raise ValueError(f"selected source contains unsupported control characters: {source}") + return f"{stable_id(source)}\t{item_type}\t{name}\t{source}" + + +def build(storage_config: Path, output: Path) -> dict[str, object]: + mappings = read_config(storage_config) + entries: dict[str, str] = {} + + for _storage_id, source_prefix, mount, include_prefix in mappings: + if not mount.is_dir(): + raise FileNotFoundError(f"user storage mount unavailable: {mount}") + user_root = contained(mount, source_prefix) + if not user_root.is_dir(): + raise FileNotFoundError(f"user files root unavailable: {user_root}") + + if include_prefix: + selected = contained(user_root, include_prefix) + line = manifest_entry(selected) + entries[str(selected.resolve())] = line + continue + + # Empty selection means the user's root. Expose its direct children as + # Piwigo roots instead of creating an artificial "files" album. + for child in sorted(user_root.iterdir(), key=lambda item: (item.name.casefold(), item.name)): + if child.is_symlink(): + continue + if not (child.is_dir() or child.is_file()): + continue + line = manifest_entry(child) + entries[str(child.resolve())] = line + + if not entries: + raise RuntimeError("the selected Nextcloud directories contain no readable files or folders") + + output.parent.mkdir(parents=True, exist_ok=True) + with tempfile.NamedTemporaryFile("w", encoding="utf-8", dir=output.parent, delete=False) as handle: + for line in entries.values(): + handle.write(line + "\n") + temporary = Path(handle.name) + temporary.replace(output) + return {"roots": len(entries), "manifest": str(output)} + + +def main() -> int: + parser = argparse.ArgumentParser() + parser.add_argument("--storage-config", required=True, type=Path) + parser.add_argument("--output", required=True, type=Path) + args = parser.parse_args() + try: + print(json.dumps(build(args.storage_config, args.output), ensure_ascii=False)) + except Exception as error: + print(f"user-manifest: {error}", file=sys.stderr) + return 1 + return 0 + + +if __name__ == "__main__": + raise SystemExit(main())