Mercurial > libervia-backend

comparison libervia/tui/chat.py @ 4076:b620a8e882e1

Find changesets by keywords (author, files, the commit message), revision number or hash, or revset expression.
refactoring: rename `libervia.frontends.primitivus` to `libervia.tui`
author Goffi <goffi@goffi.org>
date Fri, 02 Jun 2023 16:25:25 +0200
parents libervia/frontends/primitivus/chat.py@26b7ed2817da
children 0d7bb4df2343
comparison
equal deleted inserted replaced
4075:47401850dec6 4076:b620a8e882e1
1 #!/usr/bin/env python3
2
3
4 # Libervia TUI
5 # Copyright (C) 2009-2021 Jérôme Poisson (goffi@goffi.org)
6
7 # This program is free software: you can redistribute it and/or modify
8 # it under the terms of the GNU Affero General Public License as published by
9 # the Free Software Foundation, either version 3 of the License, or
10 # (at your option) any later version.
11
12 # This program is distributed in the hope that it will be useful,
13 # but WITHOUT ANY WARRANTY; without even the implied warranty of
14 # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
15 # GNU Affero General Public License for more details.
16
17 # You should have received a copy of the GNU Affero General Public License
18 # along with this program. If not, see <http://www.gnu.org/licenses/>.
19
20 from functools import total_ordering
21 from pathlib import Path
22 import bisect
23 import urwid
24 from urwid_satext import sat_widgets
25 from libervia.backend.core.i18n import _
26 from libervia.backend.core import log as logging
27 from libervia.frontends.quick_frontend import quick_widgets
28 from libervia.frontends.quick_frontend import quick_chat
29 from libervia.frontends.quick_frontend import quick_games
30 from libervia.tui import game_tarot
31 from libervia.tui.constants import Const as C
32 from libervia.tui.keys import action_key_map as a_key
33 from libervia.tui.widget import LiberviaTUIWidget
34 from libervia.tui.contact_list import ContactList
35
36
37 log = logging.getLogger(__name__)
38
39
40 OCCUPANTS_FOOTER = _("{} occupants")
41
42
43 class MessageWidget(urwid.WidgetWrap, quick_chat.MessageWidget):
44 def __init__(self, mess_data):
45 """
46 @param mess_data(quick_chat.Message, None): message data
47 None: used only for non text widgets (e.g.: focus separator)
48 """
49 self.mess_data = mess_data
50 mess_data.widgets.add(self)
51 super(MessageWidget, self).__init__(urwid.Text(self.markup))
52
53 @property
54 def markup(self):
55 return (
56 self._generate_info_markup()
57 if self.mess_data.type == C.MESS_TYPE_INFO
58 else self._generate_markup()
59 )
60
61 @property
62 def info_type(self):
63 return self.mess_data.info_type
64
65 @property
66 def parent(self):
67 return self.mess_data.parent
68
69 @property
70 def message(self):
71 """Return currently displayed message"""
72 return self.mess_data.main_message
73
74 @message.setter
75 def message(self, value):
76 self.mess_data.message = {"": value}
77 self.redraw()
78
79 @property
80 def type(self):
81 try:
82 return self.mess_data.type
83 except AttributeError:
84 return C.MESS_TYPE_INFO
85
86 def redraw(self):
87 self._w.set_text(self.markup)
88 self.mess_data.parent.host.redraw() # FIXME: should not be necessary
89
90 def selectable(self):
91 return True
92
93 def keypress(self, size, key):
94 return key
95
96 def get_cursor_coords(self, size):
97 return 0, 0
98
99 def render(self, size, focus=False):
100 # Text widget doesn't render cursor, but we want one
101 # so we add it here
102 canvas = urwid.CompositeCanvas(self._w.render(size, focus))
103 if focus:
104 canvas.set_cursor(self.get_cursor_coords(size))
105 return canvas
106
107 def _generate_info_markup(self):
108 return ("info_msg", self.message)
109
110 def _generate_markup(self):
111 """Generate text markup according to message data and Widget options"""
112 markup = []
113 d = self.mess_data
114 mention = d.mention
115
116 # message status
117 if d.status is None:
118 markup.append(" ")
119 elif d.status == "delivered":
120 markup.append(("msg_status_received", "✔"))
121 else:
122 log.warning("Unknown status: {}".format(d.status))
123
124 # timestamp
125 if self.parent.show_timestamp:
126 attr = "msg_mention" if mention else "date"
127 markup.append((attr, "[{}]".format(d.time_text)))
128 else:
129 if mention:
130 markup.append(("msg_mention", "[*]"))
131
132 # nickname
133 if self.parent.show_short_nick:
134 markup.append(
135 ("my_nick" if d.own_mess else "other_nick", "**" if d.own_mess else "*")
136 )
137 else:
138 markup.append(
139 ("my_nick" if d.own_mess else "other_nick", "[{}] ".format(d.nick or ""))
140 )
141
142 msg = self.message # needed to generate self.selected_lang
143
144 if d.selected_lang:
145 markup.append(("msg_lang", "[{}] ".format(d.selected_lang)))
146
147 # message body
148 markup.append(msg)
149
150 return markup
151
152 # events
153 def update(self, update_dict=None):
154 """update all the linked message widgets
155
156 @param update_dict(dict, None): key=attribute updated value=new_value
157 """
158 self.redraw()
159
160
161 @total_ordering
162 class OccupantWidget(urwid.WidgetWrap):
163 def __init__(self, occupant_data):
164 self.occupant_data = occupant_data
165 occupant_data.widgets.add(self)
166 markup = self._generate_markup()
167 text = sat_widgets.ClickableText(markup)
168 urwid.connect_signal(
169 text,
170 "click",
171 self.occupant_data.parent._occupants_clicked,
172 user_args=[self.occupant_data],
173 )
174 super(OccupantWidget, self).__init__(text)
175
176 def __hash__(self):
177 return id(self)
178
179 def __eq__(self, other):
180 if other is None:
181 return False
182 return self.occupant_data.nick == other.occupant_data.nick
183
184 def __lt__(self, other):
185 return self.occupant_data.nick.lower() < other.occupant_data.nick.lower()
186
187 @property
188 def markup(self):
189 return self._generate_markup()
190
191 @property
192 def parent(self):
193 return self.mess_data.parent
194
195 @property
196 def nick(self):
197 return self.occupant_data.nick
198
199 def redraw(self):
200 self._w.set_text(self.markup)
201 self.occupant_data.parent.host.redraw() # FIXME: should not be necessary
202
203 def selectable(self):
204 return True
205
206 def keypress(self, size, key):
207 return key
208
209 def get_cursor_coords(self, size):
210 return 0, 0
211
212 def render(self, size, focus=False):
213 # Text widget doesn't render cursor, but we want one
214 # so we add it here
215 canvas = urwid.CompositeCanvas(self._w.render(size, focus))
216 if focus:
217 canvas.set_cursor(self.get_cursor_coords(size))
218 return canvas
219
220 def _generate_markup(self):
221 # TODO: role and affiliation are shown in a Q&D way
222 # should be more intuitive and themable
223 o = self.occupant_data
224 markup = []
225 markup.append(
226 ("info_msg", "{}{} ".format(o.role[0].upper(), o.affiliation[0].upper()))
227 )
228 markup.append(o.nick)
229 if o.state is not None:
230 markup.append(" {}".format(C.CHAT_STATE_ICON[o.state]))
231 return markup
232
233 # events
234 def update(self, update_dict=None):
235 self.redraw()
236
237
238 class OccupantsWidget(urwid.WidgetWrap):
239 def __init__(self, parent):
240 self.parent = parent
241 self.occupants_walker = urwid.SimpleListWalker([])
242 self.occupants_footer = urwid.Text("", align="center")
243 self.update_footer()
244 occupants_widget = urwid.Frame(
245 urwid.ListBox(self.occupants_walker), footer=self.occupants_footer
246 )
247 super(OccupantsWidget, self).__init__(occupants_widget)
248 occupants_list = sorted(list(self.parent.occupants.keys()), key=lambda o: o.lower())
249 for occupant in occupants_list:
250 occupant_data = self.parent.occupants[occupant]
251 self.occupants_walker.append(OccupantWidget(occupant_data))
252
253 def clear(self):
254 del self.occupants_walker[:]
255
256 def update_footer(self):
257 """update footer widget"""
258 txt = OCCUPANTS_FOOTER.format(len(self.parent.occupants))
259 self.occupants_footer.set_text(txt)
260
261 def get_nicks(self, start=""):
262 """Return nicks of all occupants
263
264 @param start(unicode): only return nicknames which start with this text
265 """
266 return [
267 w.nick
268 for w in self.occupants_walker
269 if isinstance(w, OccupantWidget) and w.nick.startswith(start)
270 ]
271
272 def addUser(self, occupant_data):
273 """add a user to the list"""
274 bisect.insort(self.occupants_walker, OccupantWidget(occupant_data))
275 self.update_footer()
276 self.parent.host.redraw() # FIXME: should not be necessary
277
278 def removeUser(self, occupant_data):
279 """remove a user from the list"""
280 for widget in occupant_data.widgets:
281 self.occupants_walker.remove(widget)
282 self.update_footer()
283 self.parent.host.redraw() # FIXME: should not be necessary
284
285
286 class Chat(LiberviaTUIWidget, quick_chat.QuickChat):
287 def __init__(self, host, target, type_=C.CHAT_ONE2ONE, nick=None, occupants=None,
288 subject=None, statuses=None, profiles=None):
289 self.filters = [] # list of filter callbacks to apply
290 self.mess_walker = urwid.SimpleListWalker([])
291 self.mess_widgets = urwid.ListBox(self.mess_walker)
292 self.chat_widget = urwid.Frame(self.mess_widgets)
293 self.chat_colums = urwid.Columns([("weight", 8, self.chat_widget)])
294 self.pile = urwid.Pile([self.chat_colums])
295 LiberviaTUIWidget.__init__(self, self.pile, target)
296 quick_chat.QuickChat.__init__(
297 self, host, target, type_, nick, occupants, subject, statuses,
298 profiles=profiles
299 )
300
301 # we must adapt the behaviour with the type
302 if type_ == C.CHAT_GROUP:
303 if len(self.chat_colums.contents) == 1:
304 self.occupants_widget = OccupantsWidget(self)
305 self.occupants_panel = sat_widgets.VerticalSeparator(
306 self.occupants_widget
307 )
308 self._append_occupants_panel()
309 self.host.addListener("presence", self.presence_listener, [profiles])
310
311 # focus marker is a separator indicated last visible message before focus was lost
312 self.focus_marker = None # link to current marker
313 self.focus_marker_set = None # True if a new marker has been inserted
314 self.show_timestamp = True
315 self.show_short_nick = False
316 self.show_title = 1 # 0: clip title; 1: full title; 2: no title
317 self.post_init()
318
319 @property
320 def message_widgets_rev(self):
321 return reversed(self.mess_walker)
322
323 def keypress(self, size, key):
324 if key == a_key["OCCUPANTS_HIDE"]: # user wants to (un)hide the occupants panel
325 if self.type == C.CHAT_GROUP:
326 widgets = [widget for (widget, options) in self.chat_colums.contents]
327 if self.occupants_panel in widgets:
328 self._remove_occupants_panel()
329 else:
330 self._append_occupants_panel()
331 elif key == a_key["TIMESTAMP_HIDE"]: # user wants to (un)hide timestamp
332 self.show_timestamp = not self.show_timestamp
333 self.redraw()
334 elif key == a_key["SHORT_NICKNAME"]: # user wants to (not) use short nick
335 self.show_short_nick = not self.show_short_nick
336 self.redraw()
337 elif (key == a_key["SUBJECT_SWITCH"]):
338 # user wants to (un)hide group's subject or change its apperance
339 if self.subject:
340 self.show_title = (self.show_title + 1) % 3
341 if self.show_title == 0:
342 self.set_subject(self.subject, "clip")
343 elif self.show_title == 1:
344 self.set_subject(self.subject, "space")
345 elif self.show_title == 2:
346 self.chat_widget.header = None
347 self._invalidate()
348 elif key == a_key["GOTO_BOTTOM"]: # user wants to focus last message
349 self.mess_widgets.focus_position = len(self.mess_walker) - 1
350
351 return super(Chat, self).keypress(size, key)
352
353 def completion(self, text, completion_data):
354 """Completion method which complete nicknames in group chat
355
356 for params, see [sat_widgets.AdvancedEdit]
357 """
358 if self.type != C.CHAT_GROUP:
359 return text
360
361 space = text.rfind(" ")
362 start = text[space + 1 :]
363 words = self.occupants_widget.get_nicks(start)
364 if not words:
365 return text
366 try:
367 word_idx = words.index(completion_data["last_word"]) + 1
368 except (KeyError, ValueError):
369 word_idx = 0
370 else:
371 if word_idx == len(words):
372 word_idx = 0
373 word = completion_data["last_word"] = words[word_idx]
374 return "{}{}{}".format(text[: space + 1], word, ": " if space < 0 else "")
375
376 def get_menu(self):
377 """Return Menu bar"""
378 menu = sat_widgets.Menu(self.host.loop)
379 if self.type == C.CHAT_GROUP:
380 self.host.add_menus(menu, C.MENU_ROOM, {"room_jid": self.target.bare})
381 game = _("Game")
382 menu.add_menu(game, "Tarot", self.on_tarot_request)
383 elif self.type == C.CHAT_ONE2ONE:
384 # FIXME: self.target is a bare jid, we need to check that
385 contact_list = self.host.contact_lists[self.profile]
386 if not self.target.resource:
387 full_jid = contact_list.get_full_jid(self.target)
388 else:
389 full_jid = self.target
390 self.host.add_menus(menu, C.MENU_SINGLE, {"jid": full_jid})
391 return menu
392
393 def set_filter(self, args):
394 """set filtering of messages
395
396 @param args(list[unicode]): filters following syntax "[filter]=[value]"
397 empty list to clear all filters
398 only lang=XX is handled for now
399 """
400 del self.filters[:]
401 if args:
402 if args[0].startswith("lang="):
403 lang = args[0][5:].strip()
404 self.filters.append(lambda mess_data: lang in mess_data.message)
405
406 self.print_messages()
407
408 def presence_listener(self, entity, show, priority, statuses, profile):
409 """Update entity's presence status
410
411 @param entity (jid.JID): entity updated
412 @param show: availability
413 @param priority: resource's priority
414 @param statuses: dict of statuses
415 @param profile: %(doc_profile)s
416 """
417 # FIXME: disable for refactoring, need to be checked and re-enabled
418 return
419 # assert self.type == C.CHAT_GROUP
420 # if entity.bare != self.target:
421 # return
422 # self.update(entity)
423
424 def create_message(self, message):
425 self.appendMessage(message)
426
427 def _scrollDown(self):
428 """scroll down message only if we are already at the bottom (minus 1)"""
429 current_focus = self.mess_widgets.focus_position
430 bottom = len(self.mess_walker) - 1
431 if current_focus == bottom - 1:
432 self.mess_widgets.focus_position = bottom # scroll down
433 self.host.redraw() # FIXME: should not be necessary
434
435 def appendMessage(self, message, minor_notifs=True):
436 """Create a MessageWidget and append it
437
438 Can merge info messages together if desirable (e.g.: multiple joined/leave)
439 @param message(quick_chat.Message): message to add
440 @param minor_notifs(boolean): if True, basic notifications are allowed
441 If False, notification are not shown except if we have an important one
442 (like a mention).
443 False is generally used when printing history, when we don't want every
444 message to be notified.
445 """
446 if message.attachments:
447 # FIXME: Q&D way to see attachments in LiberviaTUI
448 # it should be done in a more user friendly way
449 for lang, body in message.message.items():
450 for attachment in message.attachments:
451 if 'url' in attachment:
452 body+=f"\n{attachment['url']}"
453 elif 'path' in attachment:
454 path = Path(attachment['path'])
455 body+=f"\n{path.as_uri()}"
456 else:
457 log.warning(f'No "url" nor "path" in attachment: {attachment}')
458 message.message[lang] = body
459
460 if self.filters:
461 if not all([f(message) for f in self.filters]):
462 return
463
464 if self.handle_user_moved(message):
465 return
466
467 if ((self.host.selected_widget != self or not self.host.x_notify.has_focus())
468 and self.focus_marker_set is not None):
469 if not self.focus_marker_set and not self._locked and self.mess_walker:
470 if self.focus_marker is not None:
471 try:
472 self.mess_walker.remove(self.focus_marker)
473 except ValueError:
474 # self.focus_marker may not be in mess_walker anymore if
475 # mess_walker has been cleared, e.g. when showing search
476 # result or using :history command
477 pass
478 self.focus_marker = urwid.Divider("—")
479 self.mess_walker.append(self.focus_marker)
480 self.focus_marker_set = True
481 self._scrollDown()
482 else:
483 if self.focus_marker_set:
484 self.focus_marker_set = False
485
486 wid = MessageWidget(message)
487 self.mess_walker.append(wid)
488 self._scrollDown()
489 if self.is_user_moved(message):
490 return # no notification for moved messages
491
492 # notifications
493
494 if self._locked:
495 # we don't want notifications when locked
496 # because that's history messages
497 return
498
499 if wid.mess_data.mention:
500 from_jid = wid.mess_data.from_jid
501 msg = _(
502 "You have been mentioned by {nick} in {room}".format(
503 nick=wid.mess_data.nick, room=self.target
504 )
505 )
506 self.host.notify(
507 C.NOTIFY_MENTION, from_jid, msg, widget=self, profile=self.profile
508 )
509 elif not minor_notifs:
510 return
511 elif self.type == C.CHAT_ONE2ONE:
512 from_jid = wid.mess_data.from_jid
513 msg = _("{entity} is talking to you".format(entity=from_jid))
514 self.host.notify(
515 C.NOTIFY_MESSAGE, from_jid, msg, widget=self, profile=self.profile
516 )
517 else:
518 self.host.notify(
519 C.NOTIFY_MESSAGE, self.target, widget=self, profile=self.profile
520 )
521
522 def addUser(self, nick):
523 occupant = super(Chat, self).addUser(nick)
524 self.occupants_widget.addUser(occupant)
525
526 def removeUser(self, occupant_data):
527 occupant = super(Chat, self).removeUser(occupant_data)
528 if occupant is not None:
529 self.occupants_widget.removeUser(occupant)
530
531 def occupants_clear(self):
532 super(Chat, self).occupants_clear()
533 self.occupants_widget.clear()
534
535 def _occupants_clicked(self, occupant, clicked_wid):
536 assert self.type == C.CHAT_GROUP
537 contact_list = self.host.contact_lists[self.profile]
538
539 # we have a click on a nick, we need to create the widget if it doesn't exists
540 self.get_or_create_private_widget(occupant.jid)
541
542 # now we select the new window
543 for contact_list in self.host.widgets.get_widgets(
544 ContactList, profiles=(self.profile,)
545 ):
546 contact_list.set_focus(occupant.jid, True)
547
548 def _append_occupants_panel(self):
549 self.chat_colums.contents.append((self.occupants_panel, ("weight", 2, False)))
550
551 def _remove_occupants_panel(self):
552 for widget, options in self.chat_colums.contents:
553 if widget is self.occupants_panel:
554 self.chat_colums.contents.remove((widget, options))
555 break
556
557 def add_game_panel(self, widget):
558 """Insert a game panel to this Chat dialog.
559
560 @param widget (Widget): the game panel
561 """
562 assert len(self.pile.contents) == 1
563 self.pile.contents.insert(0, (widget, ("weight", 1)))
564 self.pile.contents.insert(1, (urwid.Filler(urwid.Divider("-"), ("fixed", 1))))
565 self.host.redraw()
566
567 def remove_game_panel(self, widget):
568 """Remove the game panel from this Chat dialog.
569
570 @param widget (Widget): the game panel
571 """
572 assert len(self.pile.contents) == 3
573 del self.pile.contents[0]
574 self.host.redraw()
575
576 def set_subject(self, subject, wrap="space"):
577 """Set title for a group chat"""
578 quick_chat.QuickChat.set_subject(self, subject)
579 self.subj_wid = urwid.Text(
580 str(subject.replace("\n", "|") if wrap == "clip" else subject),
581 align="left" if wrap == "clip" else "center",
582 wrap=wrap,
583 )
584 self.chat_widget.header = urwid.AttrMap(self.subj_wid, "title")
585 self.host.redraw()
586
587 ## Messages
588
589 def print_messages(self, clear=True):
590 """generate message widgets
591
592 @param clear(bool): clear message before printing if true
593 """
594 if clear:
595 del self.mess_walker[:]
596 for message in self.messages.values():
597 self.appendMessage(message, minor_notifs=False)
598
599 def redraw(self):
600 """redraw all messages"""
601 for w in self.mess_walker:
602 try:
603 w.redraw()
604 except AttributeError:
605 pass
606
607 def update_history(self, size=C.HISTORY_LIMIT_DEFAULT, filters=None, profile="@NONE@"):
608 del self.mess_walker[:]
609 if filters and "search" in filters:
610 self.mess_walker.append(
611 urwid.Text(
612 _("Results for searching the globbing pattern: {}").format(
613 filters["search"]
614 )
615 )
616 )
617 self.mess_walker.append(
618 urwid.Text(_("Type ':history <lines>' to reset the chat history"))
619 )
620 super(Chat, self).update_history(size, filters, profile)
621
622 def _on_history_printed(self):
623 """Refresh or scroll down the focus after the history is printed"""
624 self.print_messages(clear=False)
625 super(Chat, self)._on_history_printed()
626
627 def on_private_created(self, widget):
628 self.host.contact_lists[widget.profile].set_special(
629 widget.target, C.CONTACT_SPECIAL_GROUP
630 )
631
632 def on_selected(self):
633 self.focus_marker_set = False
634
635 def notify(self, contact="somebody", msg=""):
636 """Notify the user of a new message if Libervia TUI doesn't have the focus.
637
638 @param contact (unicode): contact who wrote to the users
639 @param msg (unicode): the message that has been received
640 """
641 # FIXME: not called anymore after refactoring
642 if msg == "":
643 return
644 if self.mess_widgets.get_focus()[1] == len(self.mess_walker) - 2:
645 # we don't change focus if user is not at the bottom
646 # as that mean that he is probably watching discussion history
647 self.mess_widgets.focus_position = len(self.mess_walker) - 1
648 self.host.redraw()
649 if not self.host.x_notify.has_focus():
650 if self.type == C.CHAT_ONE2ONE:
651 self.host.x_notify.send_notification(
652 _("LiberviaTUI: %s is talking to you") % contact
653 )
654 elif self.nick is not None and self.nick.lower() in msg.lower():
655 self.host.x_notify.send_notification(
656 _("LiberviaTUI: %(user)s mentioned you in room '%(room)s'")
657 % {"user": contact, "room": self.target}
658 )
659
660 # MENU EVENTS #
661 def on_tarot_request(self, menu):
662 # TODO: move this to plugin_misc_tarot with dynamic menu
663 if len(self.occupants) != 4:
664 self.host.show_pop_up(
665 sat_widgets.Alert(
666 _("Can't start game"),
667 _(
668 "You need to be exactly 4 peoples in the room to start a Tarot game"
669 ),
670 ok_cb=self.host.remove_pop_up,
671 )
672 )
673 else:
674 self.host.bridge.tarot_game_create(
675 self.target, list(self.occupants), self.profile
676 )
677
678 # MISC EVENTS #
679
680 def on_delete(self):
681 # FIXME: to be checked after refactoring
682 super(Chat, self).on_delete()
683 if self.type == C.CHAT_GROUP:
684 self.host.removeListener("presence", self.presence_listener)
685
686 def on_chat_state(self, from_jid, state, profile):
687 super(Chat, self).on_chat_state(from_jid, state, profile)
688 if self.type == C.CHAT_ONE2ONE:
689 self.title_dynamic = C.CHAT_STATE_ICON[state]
690 self.host.redraw() # FIXME: should not be necessary
691
692 def _on_subject_dialog_cb(self, button, dialog):
693 self.change_subject(dialog.text)
694 self.host.remove_pop_up(dialog)
695
696 def on_subject_dialog(self, new_subject=None):
697 dialog = sat_widgets.InputDialog(
698 _("Change title"),
699 _("Enter the new title"),
700 default_txt=new_subject if new_subject is not None else self.subject,
701 )
702 dialog.set_callback("ok", self._on_subject_dialog_cb, dialog)
703 dialog.set_callback("cancel", lambda __: self.host.remove_pop_up(dialog))
704 self.host.show_pop_up(dialog)
705
706
707 quick_widgets.register(quick_chat.QuickChat, Chat)
708 quick_widgets.register(quick_games.Tarot, game_tarot.TarotGame)