""" Module for the PrivateTab A PrivateTab is a private conversation opened with someone from a MUC (see muctab.py). The conversation happens with both JID being relative to the MUC (room@server/nick1 and room@server/nick2). This tab references his parent room, and is modified to keep track of both participant’s nicks. It also has slightly different features than the ConversationTab (such as tab-completion on nicks from the room). """ import logging log = logging.getLogger(__name__) import curses from poezio.tabs import OneToOneTab, MucTab, Tab from poezio import fixes from poezio import windows from poezio import xhtml from poezio.common import safeJID from poezio.config import config from poezio.decorators import refresh_wrapper from poezio.logger import logger from poezio.theming import get_theme, dump_tuple from poezio.decorators import command_args_parser class PrivateTab(OneToOneTab): """ The tab containg a private conversation (someone from a MUC) """ message_type = 'chat' plugin_commands = {} additional_information = {} plugin_keys = {} def __init__(self, core, name, nick): OneToOneTab.__init__(self, core, name) self.own_nick = nick self.name = name self.text_win = windows.TextWin() self._text_buffer.add_window(self.text_win) self.info_header = windows.PrivateInfoWin() self.input = windows.MessageInput() # keys self.key_func['^I'] = self.completion # commands self.register_command('info', self.command_info, desc='Display some information about the user in the MUC: its/his/her role, affiliation, status and status message.', shortdesc='Info about the user.') self.register_command('unquery', self.command_unquery, shortdesc='Close the tab.') self.register_command('close', self.command_unquery, shortdesc='Close the tab.') self.register_command('version', self.command_version, desc='Get the software version of the current interlocutor (usually its XMPP client and Operating System).', shortdesc='Get the software version of a jid.') self.resize() self.parent_muc = self.core.get_tab_by_name(safeJID(name).bare, MucTab) self.on = True self.update_commands() self.update_keys() def remote_user_color(self): user = self.parent_muc.get_user_by_name(safeJID(self.name).resource) if user: return dump_tuple(user.color); return super().remote_user_color() @property def general_jid(self): return self.name def get_dest_jid(self): return self.name @property def nick(self): return self.get_nick() @staticmethod def add_information_element(plugin_name, callback): """ Lets a plugin add its own information to the PrivateInfoWin """ PrivateTab.additional_information[plugin_name] = callback @staticmethod def remove_information_element(plugin_name): del PrivateTab.additional_information[plugin_name] def load_logs(self, log_nb): logs = logger.get_logs(safeJID(self.name).full.replace('/', '\\'), log_nb) return logs def log_message(self, txt, nickname, time=None, typ=1): """ Log the messages in the archives. """ if not logger.log_message(self.name, nickname, txt, date=time, typ=typ): self.core.information('Unable to write in the log file', 'Error') def on_close(self): super().on_close() self.parent_muc.privates.remove(self) def completion(self): """ Called when Tab is pressed, complete the nickname in the input """ if self.complete_commands(self.input): return # If we are not completing a command or a command's argument, complete a nick compare_users = lambda x: x.last_talked word_list = [user.nick for user in sorted(self.parent_muc.users, key=compare_users, reverse=True)\ if user.nick != self.own_nick] after = config.get('after_completion') + ' ' input_pos = self.input.pos if ' ' not in self.input.get_text()[:input_pos] or (self.input.last_completion and\ self.input.get_text()[:input_pos] == self.input.last_completion + after): add_after = after else: add_after = '' self.input.auto_completion(word_list, add_after, quotify=False) empty_after = self.input.get_text() == '' or (self.input.get_text().startswith('/') and not self.input.get_text().startswith('//')) self.send_composing_chat_state(empty_after) @command_args_parser.raw def command_say(self, line, attention=False, correct=False): if not self.on: return msg = self.core.xmpp.make_message(self.name) msg['type'] = 'chat' msg['body'] = line # trigger the event BEFORE looking for colors. # This lets a plugin insert \x19xxx} colors, that will # be converted in xhtml. self.core.events.trigger('private_say', msg, self) if not msg['body']: self.cancel_paused_delay() self.text_win.refresh() self.input.refresh() return user = self.parent_muc.get_user_by_name(self.own_nick) replaced = False if correct or msg['replace']['id']: msg['replace']['id'] = self.last_sent_message['id'] if config.get_by_tabname('group_corrections', self.name): try: self.modify_message(msg['body'], self.last_sent_message['id'], msg['id'], user=user, jid=self.core.xmpp.boundjid, nickname=self.own_nick) replaced = True except: log.error('Unable to correct a message', exc_info=True) else: del msg['replace'] if msg['body'].find('\x19') != -1: msg.enable('html') msg['html']['body'] = xhtml.poezio_colors_to_html(msg['body']) msg['body'] = xhtml.clean_text(msg['body']) if (config.get_by_tabname('send_chat_states', self.general_jid) and self.remote_wants_chatstates is not False): needed = 'inactive' if self.inactive else 'active' msg['chat_state'] = needed if attention and self.remote_supports_attention: msg['attention'] = True self.core.events.trigger('private_say_after', msg, self) if not msg['body']: self.cancel_paused_delay() self.text_win.refresh() self.input.refresh() return if not replaced: self.add_message(msg['body'], nickname=self.own_nick or self.core.own_nick, forced_user=user, nick_color=get_theme().COLOR_OWN_NICK, identifier=msg['id'], jid=self.core.xmpp.boundjid, typ=1) self.last_sent_message = msg if self.remote_supports_receipts: msg._add_receipt = True msg.send() self.cancel_paused_delay() self.text_win.refresh() self.input.refresh() @command_args_parser.ignored def command_unquery(self): """ /unquery """ self.core.close_tab(self) @command_args_parser.quoted(0, 1) def command_version(self, args): """ /version """ def callback(res): if not res: return self.core.information('Could not get the software version from %s' % (jid,), 'Warning') version = '%s is running %s version %s on %s' % (jid, res.get('name') or 'an unknown software', res.get('version') or 'unknown', res.get('os') or 'an unknown platform') self.core.information(version, 'Info') if args: return self.core.command.version(args[0]) jid = safeJID(self.name) fixes.get_version(self.core.xmpp, jid, callback=callback) @command_args_parser.quoted(0, 1) def command_info(self, arg): """ /info """ if arg and arg[0]: self.parent_muc.command_info(arg[0]) else: user = safeJID(self.name).resource self.parent_muc.command_info(user) def resize(self): self.need_resize = False if self.size.tab_degrade_y: info_win_height = 0 tab_win_height = 0 else: info_win_height = self.core.information_win_size tab_win_height = Tab.tab_win_height() self.text_win.resize(self.height - 2 - info_win_height - tab_win_height, self.width, 0, 0) self.text_win.rebuild_everything(self._text_buffer) self.info_header.resize(1, self.width, self.height - 2 - info_win_height - tab_win_height, 0) self.input.resize(1, self.width, self.height-1, 0) def refresh(self): if self.need_resize: self.resize() log.debug(' TAB Refresh: %s', self.__class__.__name__) display_info_win = not self.size.tab_degrade_y self.text_win.refresh() self.info_header.refresh(self.name, self.text_win, self.chatstate, PrivateTab.additional_information) if display_info_win: self.info_win.refresh() self.refresh_tab_win() self.input.refresh() def refresh_info_header(self): self.info_header.refresh(self.name, self.text_win, self.chatstate, PrivateTab.additional_information) self.input.refresh() def get_nick(self): return safeJID(self.name).resource def on_input(self, key, raw): if not raw and key in self.key_func: self.key_func[key]() return False self.input.do_command(key, raw=raw) if not self.on: return False empty_after = self.input.get_text() == '' or (self.input.get_text().startswith('/') and not self.input.get_text().startswith('//')) tab = self.core.get_tab_by_name(safeJID(self.name).bare, MucTab) if tab and tab.joined: self.send_composing_chat_state(empty_after) return False def on_lose_focus(self): if self.input.text: self.state = 'nonempty' else: self.state = 'normal' self.text_win.remove_line_separator() self.text_win.add_line_separator(self._text_buffer) tab = self.core.get_tab_by_name(safeJID(self.name).bare, MucTab) if tab and tab.joined and config.get_by_tabname('send_chat_states', self.general_jid) and not self.input.get_text() and self.on: self.send_chat_state('inactive') self.check_scrolled() def on_gain_focus(self): self.state = 'current' curses.curs_set(1) tab = self.core.get_tab_by_name(safeJID(self.name).bare, MucTab) if tab and tab.joined and config.get_by_tabname('send_chat_states', self.general_jid,) and not self.input.get_text() and self.on: self.send_chat_state('active') def on_info_win_size_changed(self): if self.core.information_win_size >= self.height-3: return self.text_win.resize(self.height-2-self.core.information_win_size - Tab.tab_win_height(), self.width, 0, 0) self.info_header.resize(1, self.width, self.height-2-self.core.information_win_size - Tab.tab_win_height(), 0) def get_text_window(self): return self.text_win @refresh_wrapper.conditional def rename_user(self, old_nick, user): """ The user changed her nick in the corresponding muc: update the tab’s name and display a message. """ self.add_message('\x19%(nick_col)s}%(old)s\x19%(info_col)s} is now ' 'known as \x19%(nick_col)s}%(new)s' % { 'old':old_nick, 'new': user.nick, 'nick_col': dump_tuple(user.color), 'info_col': dump_tuple(get_theme().COLOR_INFORMATION_TEXT)}, typ=2) new_jid = safeJID(self.name).bare+'/'+user.nick self.name = new_jid return self.core.current_tab() is self @refresh_wrapper.conditional def user_left(self, status_message, user): """ The user left the associated MUC """ self.deactivate() if config.get_by_tabname('display_user_color_in_join_part', self.general_jid): color = dump_tuple(user.color) else: color = dump_tuple(get_theme().COLOR_REMOTE_USER) if not status_message: self.add_message('\x19%(quit_col)s}%(spec)s \x19%(nick_col)s}' '%(nick)s\x19%(info_col)s} has left the room' % { 'nick': user.nick, 'spec': get_theme().CHAR_QUIT, 'nick_col': color, 'quit_col': dump_tuple(get_theme().COLOR_QUIT_CHAR), 'info_col': dump_tuple(get_theme().COLOR_INFORMATION_TEXT)}, typ=2) else: self.add_message('\x19%(quit_col)s}%(spec)s \x19%(nick_col)s}' '%(nick)s\x19%(info_col)s} has left the room' ' (%(status)s)' % { 'status': status_message, 'nick': user.nick, 'spec': get_theme().CHAR_QUIT, 'nick_col': color, 'quit_col': dump_tuple(get_theme().COLOR_QUIT_CHAR), 'info_col': dump_tuple(get_theme().COLOR_INFORMATION_TEXT)}, typ=2) return self.core.current_tab() is self @refresh_wrapper.conditional def user_rejoined(self, nick): """ The user (or at least someone with the same nick) came back in the MUC """ self.activate() self.check_features() tab = self.parent_muc color = dump_tuple(get_theme().COLOR_REMOTE_USER) if tab and config.get_by_tabname('display_user_color_in_join_part', self.general_jid): user = tab.get_user_by_name(nick) if user: color = dump_tuple(user.color) self.add_message('\x19%(join_col)s}%(spec)s \x19%(color)s}%(nick)s\x19' '%(info_col)s} joined the room' % {'nick':nick, 'color': color, 'spec':get_theme().CHAR_JOIN, 'join_col': dump_tuple(get_theme().COLOR_JOIN_CHAR), 'info_col': dump_tuple(get_theme().COLOR_INFORMATION_TEXT)}, typ=2) return self.core.current_tab() is self def activate(self, reason=None): self.on = True if reason: self.add_message(txt=reason, typ=2) def deactivate(self, reason=None): self.on = False self.remote_wants_chatstates = None if reason: self.add_message(txt=reason, typ=2) def matching_names(self): return [(3, safeJID(self.name).resource), (4, self.name)] def add_error(self, error_message): error = '\x19%s}%s\x19o' % (dump_tuple(get_theme().COLOR_CHAR_NACK), error_message) self.add_message(error, highlight=True, nickname='Error', nick_color=get_theme().COLOR_ERROR_MSG, typ=2) self.core.refresh_window()