""" `put`: upload files, or folders with -R, into a folder of a group. **Decided whole before anything is sent**, as a drop on the Files tab is (docs/MESHBAY_DESIGN.md ยง6.4): a name already in the destination โ€” compared without case โ€” or one the node does not accept refuses the run and lists every offender, so a folder never arrives with holes in it. `-n` skips the names already there instead. There is no overwrite: the node never replaces a file, and if one appears while an upload runs, the upload takes a free name and the CLI says which. """ import asyncio import contextlib import os import sys import tempfile import threading from pathlib import Path from meshbay_common.paths import long_path from ..client.groups import my_groups from ..client.paths import handle, parse, resolve_group from ..client.pool import Pool from ..client.transfers import TransferError, create_dir, upload from ..client.tree import Tree from ..client.upload_plan import UploadFile, plan_upload, refused_names, taken_names from ..errors import NOT_FOUND, PARTIAL, REFUSED, USAGE, CliError from ..output import Progress, err, human_size, safe, sentence from .context import Context HELP = "upload files, or folders with -R, into a folder of a group" PER_GROUP = 2 def add_parser(sub, parent) -> None: p = sub.add_parser("put", parents=[parent], help=HELP, description=sentence(HELP) + ". With - as the source, stdin is sent and DEST names the file.") p.add_argument("sources", nargs="+", metavar="SOURCE", help="local files or folders, or - for stdin") p.add_argument("dest", metavar="DEST", help="the destination folder: group[@owner]/root/folder; " "with - as the source, the file to write") p.add_argument("-R", dest="recursive", action="store_true", help="folders, recursively") p.add_argument("-n", dest="skip", action="store_true", help="skip names already in the destination instead of refusing") async def run(ctx: Context) -> int: a = ctx.args sources, dest_text = a.sources, a.dest remote = parse(dest_text) if remote is None or not remote.parts: raise CliError(USAGE, "The destination is a folder of a group: " "group[@owner]/root/folder") from_stdin = sources == ["-"] if "-" in sources and not from_stdin: raise CliError(USAGE, "- (stdin) is the only source when it is used.") await ctx.hub.check_version() account = ctx.account() pool = Pool(account) spool = None try: group = resolve_group(await my_groups(account), remote.group) session = await pool.session(group) tree = Tree.from_index(await session.index()) parts = remote.parts[:-1] if from_stdin else remote.parts folder = "/".join(parts) _check_folder(tree, group, folder) if from_stdin: spool = await asyncio.to_thread(_spool_stdin) name = remote.parts[-1] plan_dirs: list[list[str]] = [] files = [UploadFile(Path(spool.name), [name], os.path.getsize(spool.name))] problems = refused_names([name], []) problems += [f"{folder}/{t} is already there" for t in sorted(taken_names(tree, folder, [name]))] else: plan_dirs, files, problems = plan_upload(sources, a.recursive) tops = sorted({tuple(d[:1]) for d in plan_dirs} | {tuple(f.rel[:1]) for f in files}) taken = taken_names(tree, folder, [t[0] for t in tops]) if taken and a.skip: plan_dirs = [d for d in plan_dirs if d[0] not in taken] files = [f for f in files if f.rel[0] not in taken] elif taken: problems += [f"{folder}/{t} is already there (-n to skip it)" for t in sorted(taken)] problems += refused_names([f.rel[-1] for f in files], plan_dirs) if problems: raise CliError(REFUSED, "Nothing was sent:\n " + "\n ".join(safe(p) for p in problems)) return await _send(ctx, session, group, folder, plan_dirs, files, display_names=[remote.parts[-1]] if from_stdin else None) finally: if spool is not None: with contextlib.suppress(OSError): os.unlink(spool.name) await pool.close() await account.close() def _check_folder(tree: Tree, group: dict, folder: str) -> None: shown = f"{handle(group)}/{folder}" if not tree.is_dir(folder): raise CliError(NOT_FOUND, f"No folder {safe(shown)}.") root = tree.roots.get(folder.split("/", 1)[0], {}) if not root.get("available", True): raise CliError(REFUSED, f"{safe(shown)} is unavailable on the node " "(its drive is disconnected).") if not root.get("writable"): raise CliError(REFUSED, f"{safe(shown)} is read-only: its operator does not " "accept uploads there.") def _spool_stdin(): """stdin to a file first: every chunk names the total, so the size comes first.""" f = tempfile.NamedTemporaryFile(prefix="meshbay-cli-", delete=False) with f: while block := sys.stdin.buffer.read(1024 * 1024): f.write(block) return f async def _send(ctx: Context, session, group: dict, folder: str, dirs: list[list[str]], files: list[UploadFile], display_names: list[str] | None) -> int: a = ctx.args progress = Progress("โ†‘", quiet=a.quiet, as_json=a.json) progress.plan(len(files), sum(f.size for f in files)) # Folders first, parents before children, one at a time. remote_dir: dict[tuple, str] = {(): folder} try: for d in dirs: remote_dir[tuple(d)] = await create_dir(session, remote_dir[tuple(d[:-1])], d[-1]) except TransferError as e: progress.clear() created = len(remote_dir) - 1 raise CliError(PARTIAL if created else e.code, f"Stopped: {e.message}" + (f" ({created} folder{'s' if created != 1 else ''} created)" if created else "")) from e failed = 0 stop: TransferError | None = None gate = asyncio.Semaphore(PER_GROUP) async def one(job: UploadFile) -> None: nonlocal failed, stop async with gate: if stop is not None: return name = job.rel[-1] shown = f"{handle(group)}/{remote_dir[tuple(job.rel[:-1])]}/{name}" progress.start(safe(name)) lock = threading.Lock() try: with open(long_path(job.local), "rb") as fh: def read(offset: int, length: int) -> bytes: with lock: fh.seek(offset) return fh.read(length) stored = await upload( session, read, job.size, name, remote_dir[tuple(job.rel[:-1])], progress=progress.advance, on_queued=lambda n: progress.queued(safe(name), n)) except OSError as e: failed += 1 progress.note(f"meshbay-cli: {job.local}: {e.strerror}") return except TransferError as e: failed += 1 progress.note(f"meshbay-cli: {safe(shown)}: {e.message}") if e.fatal: stop = e return if stored != name: progress.note(f"meshbay-cli: {safe(name)} was stored as {safe(stored)} " "(the name was taken meanwhile)") progress.finish(safe(name), path=shown.rsplit("/", 1)[0] + "/" + stored, size=job.size) await asyncio.gather(*(one(f) for f in files)) progress.clear() if stop is not None: skipped = len(files) - progress.files_done - failed err(f"meshbay-cli: stopped: {stop.message}" + (f" ({skipped} file{'s' if skipped != 1 else ''} not sent)" if skipped else "")) failed += skipped if not a.quiet and not a.json: done = progress.files_done err(f"Uploaded {done} file{'s' if done != 1 else ''} " f"({human_size(progress.bytes_done)})" + (f"; {failed} failed." if failed else ".")) return PARTIAL if failed else 0