diff options
Diffstat (limited to 'packages/meshbay-cli/src/meshbay_cli/commands/get.py')
| -rw-r--r-- | packages/meshbay-cli/src/meshbay_cli/commands/get.py | 235 |
1 files changed, 235 insertions, 0 deletions
diff --git a/packages/meshbay-cli/src/meshbay_cli/commands/get.py b/packages/meshbay-cli/src/meshbay_cli/commands/get.py new file mode 100644 index 0000000..eec73c4 --- /dev/null +++ b/packages/meshbay-cli/src/meshbay_cli/commands/get.py @@ -0,0 +1,235 @@ +""" +`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).") |