""" `get`: download files and folders from groups. The whole run is planned before the first byte: paths and patterns resolved against each group's index, local names made writable here, existing files found. A path that names nothing stops the run there (exit 3); a file that cannot be written is reported and the others go on (exit 4). """ import asyncio import os from dataclasses import dataclass from pathlib import Path from meshbay_common.paths import fold from ..client.groups import my_groups from ..client.names import UnsafeName, local_path from ..client.paths import handle, parse, resolve_group from ..client.pool import Pool from ..client.session import GroupSession from ..client.transfers import PART_SUFFIX, TransferError, download from ..client.tree import Tree from ..errors import NODE, NOT_FOUND, PARTIAL, USAGE, CliError from ..output import Progress, err, human_size, safe, sentence from .context import Context HELP = "download files, or folders with -R" PER_GROUP = 2 # the node's default lease cap per member and group (§5.5) def add_parser(sub, parent) -> None: p = sub.add_parser("get", parents=[parent], help=HELP, description=sentence(HELP) + ". Wildcards (* ? [...]) are matched against the group, " "so quote them.") p.add_argument("paths", nargs="+", metavar="PATH", help="group[@owner]/root/...") p.add_argument("-R", dest="recursive", action="store_true", help="folders, recursively") p.add_argument("-o", dest="dest", metavar="DEST", help="a folder (created when it ends with /), a file name for a " "single file, or - for stdout; default: the current folder") exist = p.add_mutually_exclusive_group() exist.add_argument("-f", dest="force", action="store_true", help="overwrite existing files") exist.add_argument("-n", dest="skip", action="store_true", help="skip existing files silently") @dataclass class Job: group: dict session: GroupSession entry: dict remote: str # as shown: handle/root/folders/name target: Path | None # None: stdout async def run(ctx: Context) -> int: await ctx.hub.check_version() account = ctx.account() pool = Pool(account) try: return await _run(ctx, account, pool) finally: await pool.close() await account.close() async def _run(ctx: Context, account, pool: Pool) -> int: a = ctx.args groups = await my_groups(account) trees: dict[str, Tree] = {} async def tree_of(group: dict) -> tuple[GroupSession, Tree]: session = await pool.session(group) if group["id"] not in trees: trees[group["id"]] = Tree.from_index(await session.index()) return session, trees[group["id"]] # ── What the paths name ────────────────────────────────────────────────── named: list[tuple[dict, GroupSession, Tree, str]] = [] # group, session, tree, path missing: list[str] = [] unreachable: list[str] = [] for text in a.paths: remote = parse(text) if remote is None: raise CliError(USAGE, "Name a group, and what to fetch in it.") group = resolve_group(groups, remote.group) try: session, tree = await tree_of(group) except CliError as e: # One group whose node is away does not cost the others their files. if e.code != NODE: raise unreachable.append(f"{text}: {e.message}") continue found = tree.expand(remote.parts) if not found and (remote.inner and (tree.is_dir(remote.inner) or tree.file(remote.inner))): found = [remote.inner] # a literal name with brackets in it if not found: missing.append(text) named += [(group, session, tree, p) for p in found] if missing: raise CliError(NOT_FOUND, "Nothing in your groups matches: " + ", ".join(safe(m) for m in missing)) to_stdout = a.dest == "-" files = [(g, s, t, p) for g, s, t, p in named if p and not t.is_dir(p)] folders = [(g, s, t, p) for g, s, t, p in named if not p or t.is_dir(p)] if unreachable and not named: raise CliError(NODE, "\n".join(safe(u) for u in unreachable)) if folders and not a.recursive: raise CliError(USAGE, "These are folders; use -R: " + ", ".join(safe(_shown(g, p)) for g, _, _, p in folders)) if to_stdout and (folders or len(files) != 1): raise CliError(USAGE, "-o - writes exactly one file to stdout.") # ── Where it goes ──────────────────────────────────────────────────────── single_file = not folders and len(files) == 1 and not to_stdout dest, file_name = _destination(a.dest, single_file) problems: list[str] = list(unreachable) jobs: list[Job] = [] dirs: list[Path] = [] planned: set[str] = set() renamed: list[tuple[str, Path]] = [] def plan_file(group, session, tree, path, parts): entry = tree.file(path) shown = _shown(group, path) if not tree.root_available(path): problems.append(f"{shown}: its folder on the node is unavailable") return if to_stdout: jobs.append(Job(group, session, entry, shown, None)) return try: target, changed = local_path(dest, parts) except UnsafeName as e: problems.append(f"{shown}: {e}") return key = fold(os.path.normcase(str(target))) if key in planned: problems.append(f"{shown}: another file of this run has the same local name " f"{target}") return planned.add(key) if changed: renamed.append((shown, target)) if target.exists() and not target.with_name(target.name + PART_SUFFIX).exists(): if a.skip: return if not a.force: problems.append(f"{target} already exists (-f to overwrite, -n to skip)") return jobs.append(Job(group, session, entry, shown, target)) for group, session, tree, path in files: name = path.rsplit("/", 1)[-1] plan_file(group, session, tree, path, [file_name or name]) for group, session, tree, path in folders: top = path.rsplit("/", 1)[-1] if path else group["name"] for folder, items in tree.walk(path): rel = [top] + (folder[len(path):].strip("/").split("/") if folder != path else []) rel = [r for r in rel if r] try: dirs.append(local_path(dest, rel)[0]) except UnsafeName as e: problems.append(f"{_shown(group, folder)}: {e}") continue for item in items: if not item.is_dir: plan_file(group, session, tree, item.path, rel + [item.name]) # ── Fetch ──────────────────────────────────────────────────────────────── progress = Progress("↓", quiet=a.quiet or to_stdout and not a.json, as_json=a.json) for shown, target in renamed: progress.note(f"meshbay-cli: {safe(shown)} is saved as {target.name} " "(its name cannot be written here as it is)") for problem in problems: progress.note(f"meshbay-cli: {safe(problem)}") for d in dirs: d.mkdir(parents=True, exist_ok=True) progress.plan(len(jobs), sum(int(j.entry["size"]) for j in jobs)) failed = len(problems) gates: dict[str, asyncio.Semaphore] = {} async def one(job: Job) -> None: nonlocal failed gate = gates.setdefault(job.group["id"], asyncio.Semaphore(PER_GROUP)) async with gate: progress.start(safe(job.entry["name"])) try: await download(job.session, job.entry, job.target, overwrite=a.force, progress=progress.advance, on_queued=lambda n: progress.queued( safe(job.entry["name"]), n)) progress.finish(safe(job.entry["name"]), path=job.remote, to=str(job.target or "-")) except TransferError as e: failed += 1 progress.note(f"meshbay-cli: {safe(job.remote)}: {e.message}") await asyncio.gather(*(one(j) for j in jobs)) progress.clear() if not a.quiet and not to_stdout and not a.json: done = len(jobs) - (failed - len(problems)) err(f"Downloaded {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 def _shown(group: dict, path: str) -> str: return f"{handle(group)}/{path}".rstrip("/") def _destination(dest: str | None, single_file: bool) -> tuple[Path, str | None]: """The folder files land in, and a file name when -o names one.""" if dest is None or dest == "-": return Path.cwd(), None path = Path(dest).expanduser() if dest.endswith(("/", os.sep)): path.mkdir(parents=True, exist_ok=True) return path, None if path.is_dir(): return path, None if single_file: if not path.parent.is_dir(): raise CliError(USAGE, f"{path.parent} is not a folder.") return path.parent, path.name raise CliError(USAGE, f"{dest} is not a folder (end it with / to create it).")