mirror of
https://github.com/Terranom674/Piwigo_Bratonien_Tools.git
synced 2026-09-19 12:54:31 +00:00
NC runtime: build manifest from selected Nextcloud file IDs
This commit is contained in:
Terranom674
191
runtime/lib/build_selected_manifest.py
Normal file
191
runtime/lib/build_selected_manifest.py
Normal file
@@ -0,0 +1,191 @@
|
|||||||
|
#!/usr/bin/env python3
|
||||||
|
"""Build a manifest from WebDAV-authorized Nextcloud file IDs.
|
||||||
|
|
||||||
|
The selected roots are resolved through a generic Nextcloud file view and then
|
||||||
|
mapped through configured storage adapters. No user name, path layout or storage
|
||||||
|
protocol is assumed here.
|
||||||
|
"""
|
||||||
|
|
||||||
|
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 safe_relative(value: str) -> str:
|
||||||
|
value = str(value).strip("/")
|
||||||
|
if ".." in PurePosixPath(value).parts:
|
||||||
|
raise ValueError(f"unsafe relative path: {value!r}")
|
||||||
|
return value
|
||||||
|
|
||||||
|
|
||||||
|
def read_roots(path: Path) -> dict[int, str]:
|
||||||
|
roots: dict[int, 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) != 2:
|
||||||
|
raise ValueError(f"{path}:{number}: expected fileid and display_name")
|
||||||
|
fileid_text, display_name = fields
|
||||||
|
if not fileid_text.isdigit() or int(fileid_text) < 1:
|
||||||
|
raise ValueError(f"{path}:{number}: invalid fileid")
|
||||||
|
if any(char in display_name for char in ("\t", "\n", "\r")):
|
||||||
|
raise ValueError(f"{path}:{number}: invalid display name")
|
||||||
|
roots[int(fileid_text)] = display_name or f"Element_{fileid_text}"
|
||||||
|
if not roots:
|
||||||
|
raise ValueError("no selected Nextcloud roots configured")
|
||||||
|
return roots
|
||||||
|
|
||||||
|
|
||||||
|
def read_adapters(path: Path) -> dict[str, list[tuple[str, Path]]]:
|
||||||
|
result: dict[str, list[tuple[str, Path]]] = {}
|
||||||
|
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]
|
||||||
|
storage_id = storage_id.strip()
|
||||||
|
source_prefix = safe_relative(source_prefix)
|
||||||
|
mount = Path(local_mount)
|
||||||
|
if not storage_id:
|
||||||
|
raise ValueError(f"{path}:{number}: empty storage_id")
|
||||||
|
if not mount.is_absolute():
|
||||||
|
raise ValueError(f"{path}:{number}: local_mount must be absolute")
|
||||||
|
result.setdefault(storage_id, []).append((source_prefix, mount))
|
||||||
|
if not result:
|
||||||
|
raise ValueError("no storage adapters configured")
|
||||||
|
return result
|
||||||
|
|
||||||
|
|
||||||
|
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]]], storage_id: str, source_path: str) -> tuple[str, Path]:
|
||||||
|
relative = safe_relative(source_path)
|
||||||
|
matches = [(prefix, mount) for prefix, mount in adapters.get(storage_id, []) if matches_prefix(relative, prefix)]
|
||||||
|
if not matches:
|
||||||
|
raise RuntimeError(f"no storage adapter configured for {storage_id}")
|
||||||
|
matches.sort(key=lambda item: len(item[0]), reverse=True)
|
||||||
|
best_length = len(matches[0][0])
|
||||||
|
best = {(prefix, str(mount)): (prefix, mount) for prefix, mount in matches if len(prefix) == best_length}
|
||||||
|
if len(best) != 1:
|
||||||
|
raise RuntimeError(f"storage adapter is ambiguous for {storage_id}")
|
||||||
|
return next(iter(best.values()))
|
||||||
|
|
||||||
|
|
||||||
|
def query_rows(args: argparse.Namespace, roots: dict[int, str]) -> dict[int, tuple[str, str]]:
|
||||||
|
password = args.password_file.read_text(encoding="utf-8").strip()
|
||||||
|
env = os.environ.copy()
|
||||||
|
env["PGPASSWORD"] = password
|
||||||
|
ids = ",".join(str(fileid) for fileid in sorted(roots))
|
||||||
|
view = validate_view(args.view)
|
||||||
|
sql = f"SELECT fileid, storage_id, source_path FROM {view} WHERE fileid IN ({ids}) ORDER BY fileid"
|
||||||
|
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,
|
||||||
|
]
|
||||||
|
completed = subprocess.run(command, env=env, check=False, text=True, capture_output=True)
|
||||||
|
if completed.returncode != 0:
|
||||||
|
raise RuntimeError(completed.stderr.strip() or "Nextcloud file view query failed")
|
||||||
|
result: dict[int, tuple[str, str]] = {}
|
||||||
|
for row in csv.reader(completed.stdout.splitlines(), delimiter="\t"):
|
||||||
|
if len(row) != 3 or not row[0].isdigit():
|
||||||
|
raise RuntimeError("invalid row returned by Nextcloud file view")
|
||||||
|
result[int(row[0])] = (row[1], row[2])
|
||||||
|
missing = sorted(set(roots) - set(result))
|
||||||
|
if missing:
|
||||||
|
raise RuntimeError("selected Nextcloud file IDs are no longer resolvable: " + ", ".join(map(str, missing)))
|
||||||
|
return result
|
||||||
|
|
||||||
|
|
||||||
|
def contained_join(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"resolved source escapes configured mount: {candidate}") from error
|
||||||
|
return candidate
|
||||||
|
|
||||||
|
|
||||||
|
def build(args: argparse.Namespace) -> dict[str, object]:
|
||||||
|
roots = read_roots(args.roots_config)
|
||||||
|
adapters = read_adapters(args.storage_config)
|
||||||
|
rows = query_rows(args, roots)
|
||||||
|
manifest: list[str] = []
|
||||||
|
|
||||||
|
for fileid, display_name in roots.items():
|
||||||
|
storage_id, source_path = rows[fileid]
|
||||||
|
prefix, mount = resolve_adapter(adapters, storage_id, source_path)
|
||||||
|
if not mount.is_mount():
|
||||||
|
raise RuntimeError(f"storage mount unavailable: {mount}")
|
||||||
|
relative = safe_relative(source_path)
|
||||||
|
if prefix:
|
||||||
|
relative = relative[len(prefix):].lstrip("/")
|
||||||
|
source = contained_join(mount, relative)
|
||||||
|
if source.is_dir():
|
||||||
|
item_type = "folder"
|
||||||
|
elif source.is_file():
|
||||||
|
item_type = "file"
|
||||||
|
else:
|
||||||
|
raise RuntimeError(f"selected source unavailable: {source}")
|
||||||
|
if any(char in str(source) for char in ("\t", "\n", "\r")):
|
||||||
|
raise ValueError("resolved source path contains unsupported control characters")
|
||||||
|
manifest.append(f"fileid:{fileid}\t{item_type}\t{display_name}\t{source}")
|
||||||
|
|
||||||
|
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")
|
||||||
|
temporary = Path(handle.name)
|
||||||
|
temporary.replace(args.output)
|
||||||
|
return {"roots": len(manifest), "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("--storage-config", required=True, type=Path)
|
||||||
|
parser.add_argument("--roots-config", required=True, type=Path)
|
||||||
|
parser.add_argument("--output", required=True, type=Path)
|
||||||
|
args = parser.parse_args()
|
||||||
|
try:
|
||||||
|
print(json.dumps(build(args), ensure_ascii=False))
|
||||||
|
except Exception as error:
|
||||||
|
print(f"selected-manifest: {error}", file=sys.stderr)
|
||||||
|
return 1
|
||||||
|
return 0
|
||||||
|
|
||||||
|
|
||||||
|
if __name__ == "__main__":
|
||||||
|
raise SystemExit(main())
|
||||||
Reference in New Issue
Block a user