1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
|
"""
`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
|