""" What an upload will do, decided before anything is sent — shared by `put` and the interactive mode, so both refuse the same names and the same collisions. A name already in the destination folder, compared without case as the node and the Files tab compare it, or a name the node does not accept (`names.py`), is found here; what to do about it (refuse the run, skip it, ask) is the caller's. """ import os import stat from dataclasses import dataclass from pathlib import Path from meshbay_common.paths import fold, long_path from ..errors import USAGE, CliError from ..output import safe from .names import same_name, upload_name_problem from .tree import Tree @dataclass class UploadFile: local: Path rel: list[str] # remote parts below the destination folder size: int def plan_upload(sources: list[str], recursive: bool): """Folders to create (parents first) and files to send, as remote parts.""" dirs: list[list[str]] = [] files: list[UploadFile] = [] problems: list[str] = [] seen: dict[str, str] = {} for src in sources: path = Path(src).expanduser() try: st = os.stat(long_path(path)) except OSError as e: problems.append(f"{src}: {e.strerror}") continue top = path.resolve().name if src.rstrip("/\\") in (".", "..") else path.name key = fold(top) if key in seen: problems.append(f"{src} and {seen[key]} would have the same name there") continue seen[key] = src if stat.S_ISREG(st.st_mode): files.append(UploadFile(path, [top], st.st_size)) elif stat.S_ISDIR(st.st_mode): if not recursive: raise CliError(USAGE, f"{safe(src)} is a folder; use -R.") for here, subdirs, names in _walk(path): rel = [top] + list(Path(here).relative_to(path).parts) dirs.append(rel) for d in list(subdirs): if os.path.islink(long_path(Path(here) / d)): subdirs.remove(d) problems.append(f"{Path(here) / d} is a link to a folder; not " "followed") for n in names: p = Path(here) / n try: st_n = os.stat(long_path(p)) except OSError as e: problems.append(f"{p}: {e.strerror}") continue if stat.S_ISREG(st_n.st_mode): files.append(UploadFile(p, rel + [n], st_n.st_size)) subdirs.sort() else: problems.append(f"{src} is neither a file nor a folder") return dirs, files, problems def _walk(top: Path): """`os.walk` for a tree that may go past Windows' MAX_PATH: each folder is listed under its long form (`long_path`, which only prefixes a long path, so a short top alone would not do) and the paths given back stay plain.""" stack = [top] while stack: here = stack.pop() try: with os.scandir(long_path(here)) as it: entries = list(it) except OSError: continue subdirs, names = [], [] for e in entries: try: is_dir = e.is_dir() except OSError: is_dir = False (subdirs if is_dir else names).append(e.name) yield here, subdirs, names # the caller prunes and orders subdirs stack.extend(here / d for d in reversed(subdirs)) def taken_names(tree: Tree, folder: str, names: list[str]) -> set[str]: """The names among `names` already in `folder`, compared without case.""" there = [i.name for i in tree.children(folder, include_unavailable=True)] return {n for n in names if any(same_name(n, t) for t in there)} def refused_names(file_names: list[str], dirs: list[list[str]]) -> list[str]: """Names the node would refuse — the client's copy of its rule.""" out = [] for name in file_names: if (why := upload_name_problem(name)): out.append(f"{name}: {why}") for d in dirs: if (why := upload_name_problem(d[-1], is_dir=True)): out.append(f"{'/'.join(d)}/: {why}") return out