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
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
|
"""
`chat`: a group's conversation — the last messages, or one more posted.
`chat` alone opens the interactive chat (`tui/chat.py`): the groups, one
group's conversation IRC-style, back to the groups. `chat GROUP` prints the
last ten, oldest first; `-N` (as `head -N`) or `-n N`
asks for another count. `chat GROUP -m TEXT` posts TEXT (`-m -` reads it from
stdin), sealed and signed as the application seals and signs it. Each message
is shown with its time and the name its sender sealed in it; one that cannot
be read is shown as such, never left out.
"""
import json
import re
import sys
from ..client.chat import chat_keys, history, open_row, post
from ..client.groups import my_groups, open_group
from ..client.paths import group_arg, handle, resolve_group
from ..client.roster import RosterCheck
from ..errors import USAGE, CliError
from ..output import date, err, safe, sentence
from .context import Context
HELP = ("chat: alone, interactively; with a group, its last 10 messages (-N for N), "
"or post with -m")
_COUNT = re.compile(r"^-(\d+)$")
# The one verdict worth a line (roster.py): the application's words.
KEY_CHANGED = ("! This account is using a key you have not seen before: a new device "
"added elsewhere, or the node naming the wrong person.")
def add_parser(sub, parent) -> None:
p = sub.add_parser("chat", parents=[parent], help=HELP, description=sentence(HELP)
+ ". Messages are end-to-end encrypted and signed; the node "
"stores what it cannot read.")
p.add_argument("group", metavar="GROUP", nargs="?",
help="the group, as `groups` lists it; without it, the interactive "
"chat: choose a group, talk, leave it, choose another")
p.add_argument("-n", "--last", type=int, default=10, metavar="N",
help="how many of the last messages to show (also -N, as -100)")
p.add_argument("-m", "--message", metavar="TEXT",
help="post TEXT instead of reading; - reads it from stdin")
def rewrite_count(argv: list[str], actions) -> list[str]:
"""`chat -100 GROUP` as `chat --last 100 GROUP`: argparse reads -100 as a
negative number, not as an option. Only when the action is `chat`."""
action = next((i for i, a in enumerate(argv) if a in actions), None)
if action is None or argv[action] != "chat":
return argv
out = argv[:action + 1]
for a in argv[action + 1:]:
m = _COUNT.match(a)
out += ["--last", m.group(1)] if m else [a]
return out
def _text(message: str) -> str:
text = sys.stdin.read() if message == "-" else message
text = text.rstrip("\n")
if not text.strip():
raise CliError(USAGE, "Nothing to post.")
return text
async def run(ctx: Context) -> int:
a = ctx.args
if a.last < 1:
raise CliError(USAGE, "Ask for one message at least.")
if a.group is None:
if a.message is not None:
raise CliError(USAGE, "Name the group to post to.")
if not (sys.stdin.isatty() and sys.stdout.isatty()):
raise CliError(USAGE, "The interactive chat needs a terminal; name a group "
"to read it, or to post with -m.")
from ..tui.app import run_interactive
from ..tui.chat import ChatGroupsScreen
# As with no argument at all: the sign-in screen checks the hub.
return await run_interactive(ctx, ChatGroupsScreen)
text = _text(a.message) if a.message is not None else None
await ctx.hub.check_version()
account = ctx.account()
session = None
try:
group = resolve_group(await my_groups(account), group_arg(a.group))
session = await open_group(account, group)
keys = await chat_keys(session)
if text is not None:
await post(session, keys, text, account.username)
if a.json:
print(json.dumps({"posted": True, "group": handle(group)}))
elif not a.quiet:
err(f"Posted to {safe(handle(group))}.")
return 0
check = RosterCheck(account, session)
messages = [await open_row(r, keys, group["id"], check)
for r in await history(session, a.last)]
finally:
if session is not None:
await session.close()
await account.close()
if a.json:
json.dump([{"id": m.id, "timestamp": m.timestamp, "sender": m.sender,
"text": m.text, "trust": m.trust,
**({"unreadable": m.unreadable} if m.unreadable else {})}
for m in messages],
sys.stdout, indent=1, ensure_ascii=False)
print()
return 0
if not messages and not a.quiet:
err(f"No message in {safe(handle(group))} yet.")
for m in messages:
head = f"{date(int(m.timestamp))} {safe(m.sender)}: "
if m.unreadable:
print(f"{head}[this message cannot be read: {m.unreadable}]")
continue
lines = safe_lines(m.text)
print(head + lines[0])
for line in lines[1:]:
print(" " * len(head) + line)
if m.trust == "changed":
print(" " * len(head) + KEY_CHANGED)
return 0
def safe_lines(text: str) -> list[str]:
"""A message's lines, each escaped: a line break stays a line break, every
other control character is shown, never sent to the terminal."""
return [safe(line) for line in text.split("\n")] or [""]
|