""" `ls`: your groups, or what a folder of a group holds — as the Files tab shows it. With no path the hub answers alone (`/v1/groups/mine`) and no node is dialled. Each group is its own top-level folder and nothing is merged across groups (docs/MESHBAY_DESIGN.md §9.11). """ import asyncio import json import sys from ..client.groups import my_groups from ..client.paths import encode_name, handle, parse, resolve_group from ..client.pool import Pool from ..client.tree import Item, Tree from ..errors import NODE, NOT_FOUND, PARTIAL, CliError from ..output import columns, date, err, human_size, is_tty, safe, sentence from .context import Context HELP = "list your groups, or a folder of a group" def add_parser(sub, parent) -> None: p = sub.add_parser("ls", parents=[parent], help=HELP, description=sentence(HELP) + ".") p.add_argument("path", nargs="?", default="", help="group[@owner]/root/folder; none lists your groups") p.add_argument("-l", dest="long", action="store_true", help="long form: type, size, date added") p.add_argument("-R", dest="recursive", action="store_true", help="recurse into folders") p.add_argument("-a", dest="all", action="store_true", help="include the content of unavailable roots") p.add_argument("--sort", choices=("name", "size", "date"), default="name", help="the order within folders and files (default: name)") p.add_argument("-r", dest="reverse", action="store_true", help="reverse the order") def _sorted(items: list[Item], key: str, reverse: bool) -> list[Item]: """Folders first, as the Files tab; the order applies within each.""" def k(i: Item): if key == "size": return (i.size, i.name.lower()) if key == "date": return (i.added_at, i.name.lower()) return (i.name.lower(), i.name) dirs = sorted((i for i in items if i.is_dir), key=k, reverse=reverse) files = sorted((i for i in items if not i.is_dir), key=k, reverse=reverse) return dirs + files def _flags(item: Item) -> str: flags = [] if item.root is not None: flags.append("writable" if item.root.get("writable") else "read-only") if not item.available: flags.append("unavailable") return f" ({', '.join(flags)})" if flags else "" def _line(item: Item, long: bool) -> str: name = safe(item.name) + ("/" if item.is_dir else "") if not long: return name kind = "root" if item.root is not None else ("dir" if item.is_dir else "file") size = "" if item.is_dir else human_size(item.size) return f"{kind:<5}{size:>10} {date(item.added_at):<16} {name}{_flags(item)}" def _print_items(items: list[Item], long: bool) -> None: if long or not is_tty(): for i in items: print(_line(i, long)) elif items: print(columns([_line(i, False) for i in items])) def _json_item(group: dict, item: Item) -> dict: base = handle(group) out = {"group": group["name"], "owner": group.get("owner_username"), "path": f"{base}/{'/'.join(encode_name(p) for p in item.path.split('/'))}", "name": item.name, "available": item.available, # `kind`, not `type`: an index entry's `type` is its media type. "kind": "root" if item.root is not None else ("dir" if item.is_dir else "file")} if item.root is not None: out["writable"] = bool(item.root.get("writable")) if item.entry is not None: out.update({k: v for k, v in item.entry.items() if k not in ("name", "path")}) return out async def _list_groups(ctx: Context, account) -> int: groups = sorted(await my_groups(account), key=lambda g: (g["name"].lower(), g.get("owner_username") or "")) if ctx.args.json: if ctx.args.recursive: return await _every_group(ctx, account, groups) json.dump([{"group": g["name"], "owner": g.get("owner_username"), "id": g["id"], "path": handle(g), "online": bool(g.get("node_online")), "visibility": g.get("visibility"), "admin": bool(g.get("is_admin"))} for g in groups], sys.stdout, indent=1) print() return 0 if ctx.args.recursive: return await _every_group(ctx, account, groups) if ctx.args.long or not is_tty(): for g in groups: state = "up" if g.get("node_online") else "down" print(f"{state:<6}{safe(handle(g))}" if ctx.args.long else safe(handle(g))) else: print(columns([safe(handle(g)) for g in groups])) return 0 async def _every_group(ctx: Context, account, groups: list[dict]) -> int: """ Every group's tree. All are dialled at once through the pool, so a node that does not answer costs its own deadline and nobody else's; a group whose node the hub reports as down is not dialled at all. Printed in name order. """ pool = Pool(account) async def tree_of(group: dict): if not group.get("node_online"): raise CliError(NODE, "no node serving it is online") session = await pool.session(group) return Tree.from_index(await session.index()) try: results = await asyncio.gather(*(tree_of(g) for g in groups), return_exceptions=True) finally: await pool.close() failed = 0 listing_json = [] first = True for group, result in zip(groups, results): if isinstance(result, BaseException): if not isinstance(result, CliError): raise result failed += 1 err(f"meshbay-cli: {safe(handle(group))}: {result.message}") continue if ctx.args.json: listing_json += _json_listing(ctx, group, result, "") continue if not first: print() first = False _list_in_group(ctx, group, result, "") if ctx.args.json: json.dump(listing_json, sys.stdout, indent=1) print() return PARTIAL if failed else 0 async def run(ctx: Context) -> int: await ctx.hub.check_version() account = ctx.account() try: remote = parse(ctx.args.path) if remote is None: return await _list_groups(ctx, account) group = resolve_group(await my_groups(account), remote.group) pool = Pool(account) try: tree = Tree.from_index(await (await pool.session(group)).index()) finally: await pool.close() return _list_in_group(ctx, group, tree, remote.inner) finally: await account.close() def _list_in_group(ctx: Context, group: dict, tree: Tree, path: str) -> int: a = ctx.args if path and not tree.is_dir(path): entry = tree.file(path) if entry is None: raise CliError(NOT_FOUND, f"No file or folder {path!r} in {handle(group)}.") parent = path.rpartition("/")[0] listing = [(None, [Item(entry["name"], path, False, entry=entry, available=tree.root_available(parent))])] elif a.recursive: listing = list(tree.walk(path, include_unavailable=a.all)) else: listing = [(path, tree.children(path, include_unavailable=a.all))] if a.json: json.dump([_json_item(group, i) for _, items in listing for i in items], sys.stdout, indent=1) print() return 0 _print_listing(a, group, listing) return 0 def _json_listing(ctx: Context, group: dict, tree: Tree, path: str) -> list[dict]: return [_json_item(group, i) for _, items in tree.walk(path, include_unavailable=ctx.args.all) for i in items] def _print_listing(a, group: dict, listing) -> None: for n, (folder, items) in enumerate(listing): if a.recursive: if n: print() shown = "/".join(encode_name(p) for p in folder.split("/")) if folder else "" print(safe(f"{handle(group)}/{shown}".rstrip("/")) + ":") _print_items(_sorted(items, a.sort, a.reverse), a.long)