""" `groups`: the groups this account belongs to, one line each. The hub answers alone (`/v1/groups/mine`); no node is dialled. Whether a node is up is what the hub sees of its socket, not a promise this machine can reach it. "owner" is the account that created the group on the hub — a node's operator is another notion. """ import contextlib import json import sys from datetime import UTC, datetime from ..client.groups import my_groups from ..client.hub import HubError from ..client.paths import handle from ..errors import CliError from ..output import safe, sentence from .context import Context HELP = "list the groups you belong to" def add_parser(sub, parent) -> None: sub.add_parser("groups", parents=[parent], help=HELP, description=sentence(HELP) + ": the node's state, your role, the visibility, the last activity. " "The GROUP column is the path `ls`, `get` and `put` take.") def node_state(g: dict) -> str: if not g.get("hosted", True): return "none" return "up" if g.get("node_online") else "down" def local_time(iso: str | None) -> str: if not iso: return "" try: moment = datetime.fromisoformat(iso) except ValueError: return "" if moment.tzinfo is None: # a hub on SQLite sends UTC without saying so moment = moment.replace(tzinfo=UTC) return moment.astimezone().strftime("%Y-%m-%d %H:%M") async def run(ctx: Context) -> int: await ctx.hub.check_version() account = ctx.account() try: groups = await my_groups(account) waiting = 0 if not ctx.args.json and not ctx.args.quiet: # A hub without invitations answers nothing useful; the list stands. with contextlib.suppress(HubError, CliError): waiting = len(await account.call(account.hub.invitations)) finally: await account.close() groups.sort(key=lambda g: (g["name"].lower(), g.get("owner_username") or "")) if ctx.args.json: json.dump([{"group": g["name"], "owner": g.get("owner_username"), "id": g["id"], "path": handle(g), "node": node_state(g), "role": "owner" if g.get("is_admin") else "member", "visibility": g.get("visibility"), "join_policy": g.get("join_policy"), "muted": bool(g.get("muted")), "description": g.get("description", ""), "last_activity_at": g.get("last_activity_at")} for g in groups], sys.stdout, indent=1) print() return 0 if waiting: print(f"{waiting} invitation{'s' if waiting != 1 else ''} waiting for your answer: " "meshbay-cli invitations", file=sys.stderr) if not groups: print("You belong to no group yet.", file=sys.stderr) return 0 print(f"{'NODE':<6}{'ROLE':<8}{'VISIBILITY':<12}{'LAST ACTIVITY':<18}GROUP") for g in groups: role = "owner" if g.get("is_admin") else "member" print(f"{node_state(g):<6}{role:<8}{safe(g.get('visibility') or ''):<12}" f"{local_time(g.get('last_activity_at')):<18}{safe(handle(g))}") return 0