From 337b5c6dc7b36c97a2adc173def1ce4fee57c3f9 Mon Sep 17 00:00:00 2001 From: Terranom674 Date: Thu, 20 Aug 2026 10:33:04 +0200 Subject: [PATCH] Remove legacy DB manifest builder --- runtime/lib/build_manifest.py | 216 ---------------------------------- 1 file changed, 216 deletions(-) delete mode 100644 runtime/lib/build_manifest.py diff --git a/runtime/lib/build_manifest.py b/runtime/lib/build_manifest.py deleted file mode 100644 index 70222e2..0000000 --- a/runtime/lib/build_manifest.py +++ /dev/null @@ -1,216 +0,0 @@ -#!/usr/bin/env python3 -"""Resolve Nextcloud share rows through configured storage adapters.""" - -from __future__ import annotations - -import argparse -import csv -import json -import os -import re -import subprocess -import sys -import tempfile -from pathlib import Path, PurePosixPath - -IDENTIFIER = re.compile(r"^[A-Za-z_][A-Za-z0-9_]*(?:\.[A-Za-z_][A-Za-z0-9_]*)?$") - - -def validate_view(name: str) -> str: - value = str(name).strip() - if not IDENTIFIER.fullmatch(value): - raise ValueError(f"invalid SQL view name: {value!r}") - return value - - -def sql_literal(value: str) -> str: - return "'" + str(value).replace("'", "''") + "'" - - -def read_config(path: Path) -> dict[str, list[tuple[str, Path, str]]]: - result: dict[str, list[tuple[str, Path, str]]] = {} - with path.open(encoding="utf-8") as handle: - for number, line in enumerate(handle, 1): - line = line.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, prefix, mount = fields[:3] - include_prefix = fields[3] if len(fields) == 4 else "" - storage_id = storage_id.strip() - prefix = prefix.strip("/") - include_prefix = include_prefix.strip("/") - if not storage_id: - raise ValueError(f"{path}:{number}: storage_id is empty") - if ".." in PurePosixPath(prefix).parts or ".." in PurePosixPath(include_prefix).parts: - raise ValueError(f"{path}:{number}: unsafe prefix") - result.setdefault(storage_id, []).append((prefix, Path(mount), include_prefix)) - return result - - -def run_query(args: argparse.Namespace, env: dict[str, str], sql: str) -> subprocess.CompletedProcess[str]: - command = [ - "psql", "-X", "-A", "-F", "\t", "-t", "-v", "ON_ERROR_STOP=1", - "-h", args.host, "-p", str(args.port), "-U", args.user, "-d", args.database, "-c", sql, - ] - return subprocess.run(command, env=env, check=False, text=True, capture_output=True) - - -def query_rows(args: argparse.Namespace) -> list[list[str]]: - password = args.password_file.read_text(encoding="utf-8").strip() - env = os.environ.copy() - env["PGPASSWORD"] = password - view = validate_view(args.view) - where = "" - if args.access_user: - where = " WHERE lower(access_user) = lower(" + sql_literal(args.access_user) + ")" - modern_sql = f"SELECT share_id, item_type, display_name, storage_id, source_path FROM {view}{where} ORDER BY share_id" - completed = run_query(args, env, modern_sql) - if completed.returncode == 0: - return list(csv.reader(completed.stdout.splitlines(), delimiter="\t")) - if args.access_user: - raise RuntimeError(completed.stderr.strip() or "Nextcloud user-filtered source query failed") - if "item_type" not in completed.stderr or "does not exist" not in completed.stderr: - raise RuntimeError(completed.stderr.strip() or "Nextcloud source view query failed") - legacy_sql = f"SELECT share_id, display_name, storage_id, source_path FROM {view} ORDER BY share_id" - completed = run_query(args, env, legacy_sql) - if completed.returncode != 0: - raise RuntimeError(completed.stderr.strip() or "legacy Nextcloud source view query failed") - rows: list[list[str]] = [] - for row in csv.reader(completed.stdout.splitlines(), delimiter="\t"): - if len(row) == 4: - share_id, display_name, storage_id, source_path = row - rows.append([share_id, "", display_name, storage_id, source_path]) - else: - rows.append(row) - return rows - - -def contained_join(root: Path, relative: str) -> Path: - parts = PurePosixPath(relative).parts - if relative.startswith("/") or ".." in parts: - raise ValueError(f"unsafe source path: {relative}") - return root.joinpath(*parts) - - -def matches_prefix(path: str, prefix: str) -> bool: - return not prefix or path == prefix or path.startswith(prefix + "/") - - -def resolve_adapter(adapters: dict[str, list[tuple[str, Path, str]]], storage_id: str, source_path: str) -> tuple[str, Path, str] | None: - relative = source_path.strip("/") - matches: list[tuple[str, Path, str]] = [] - for prefix, mount, include_prefix in adapters.get(storage_id, []): - if not matches_prefix(relative, prefix): - continue - mapped_relative = relative[len(prefix):].lstrip("/") if prefix else relative - if not matches_prefix(mapped_relative, include_prefix): - continue - matches.append((prefix, mount, include_prefix)) - if not matches: - return None - matches.sort(key=lambda item: (len(item[0]), len(item[2])), reverse=True) - best_score = (len(matches[0][0]), len(matches[0][2])) - best = {(item[0], str(item[1]), item[2]): item for item in matches if (len(item[0]), len(item[2])) == best_score} - if len(best) != 1: - raise RuntimeError(f"storage adapter is ambiguous for {storage_id}") - return next(iter(best.values())) - - -def build(args: argparse.Namespace) -> dict[str, object]: - validate_view(args.view) - adapters = read_config(args.storage_config) - rows = query_rows(args) - if not rows and not args.allow_empty: - raise RuntimeError("Nextcloud returned no matching sources; refusing an empty manifest") - manifest: list[str] = [] - errors: list[str] = [] - folder_count = 0 - file_count = 0 - for row in rows: - if len(row) != 5: - errors.append(f"invalid database row with {len(row)} columns") - continue - share_id, item_type, display_name, storage_id, source_path = row - item_type = item_type.strip().lower() - relative = source_path.strip("/") - if item_type and item_type not in {"folder", "file"}: - errors.append(f"source {share_id}: unsupported item_type {item_type!r}") - continue - try: - adapter = resolve_adapter(adapters, storage_id, relative) - except RuntimeError as error: - errors.append(f"source {share_id}: {error}") - continue - if not adapter: - continue - prefix, mount, _include_prefix = adapter - if prefix: - relative = relative[len(prefix):].lstrip("/") - source = contained_join(mount, relative) - if not mount.is_mount(): - errors.append(f"source {share_id}: storage mount unavailable: {mount}") - continue - if not item_type: - if source.is_dir(): - item_type = "folder" - elif source.is_file(): - item_type = "file" - else: - errors.append(f"source {share_id}: source unavailable: {source}") - continue - if item_type == "folder": - if not source.is_dir(): - errors.append(f"source {share_id}: source directory unavailable: {source}") - continue - folder_count += 1 - else: - if not source.is_file(): - errors.append(f"source {share_id}: source file unavailable: {source}") - continue - file_count += 1 - if any(char in display_name for char in ("\t", "\n", "\r")): - errors.append(f"source {share_id}: display name contains unsupported control characters") - continue - source_text = str(source) - if any(char in source_text for char in ("\t", "\n", "\r")): - errors.append(f"source {share_id}: source path contains unsupported control characters") - continue - manifest.append(f"{share_id}\t{item_type}\t{display_name.lstrip('/')}\t{source}") - if errors: - raise RuntimeError("; ".join(errors)) - if not manifest and rows and not args.allow_empty: - raise RuntimeError("no Nextcloud sources match the configured storage adapters") - args.output.parent.mkdir(parents=True, exist_ok=True) - with tempfile.NamedTemporaryFile("w", encoding="utf-8", dir=args.output.parent, delete=False) as handle: - handle.write("\n".join(manifest) + ("\n" if manifest else "")) - temporary = Path(handle.name) - temporary.replace(args.output) - return {"sources": len(manifest), "folders": folder_count, "files": file_count, "manifest": str(args.output)} - - -def main() -> int: - parser = argparse.ArgumentParser() - parser.add_argument("--host", required=True) - parser.add_argument("--port", type=int, default=5432) - parser.add_argument("--database", required=True) - parser.add_argument("--user", required=True) - parser.add_argument("--password-file", required=True, type=Path) - parser.add_argument("--view", required=True) - parser.add_argument("--access-user", default="") - parser.add_argument("--storage-config", required=True, type=Path) - parser.add_argument("--output", required=True, type=Path) - parser.add_argument("--allow-empty", action="store_true") - args = parser.parse_args() - try: - print(json.dumps(build(args), ensure_ascii=False)) - except Exception as error: - print(f"manifest: {error}", file=sys.stderr) - return 1 - return 0 - - -if __name__ == "__main__": - raise SystemExit(main())