""" Global commands which are to be linked to the Core class """ import logging log = logging.getLogger(__name__) import os from datetime import datetime from xml.etree import cElementTree as ET from slixmpp.xmlstream.stanzabase import StanzaBase from slixmpp.xmlstream.handler import Callback from slixmpp.xmlstream.matcher import StanzaPath from poezio import common from poezio import fixes from poezio import pep from poezio import tabs from poezio.bookmarks import Bookmark from poezio.common import safeJID from poezio.config import config, DEFAULT_CONFIG, options as config_opts from poezio import multiuserchat as muc from poezio.plugin import PluginConfig from poezio.roster import roster from poezio.theming import dump_tuple, get_theme from poezio.decorators import command_args_parser from poezio.core.structs import Command, POSSIBLE_SHOW class CommandCore: def __init__(self, core): self.core = core @command_args_parser.quoted(0, 1) def help(self, args): """ /help [command_name] """ if not args: color = dump_tuple(get_theme().COLOR_HELP_COMMANDS) acc = [] buff = ['Global commands:'] for name, command in enumerate(self.core.commands): if isinstance(command, Command): acc.append(' \x19%s}%s\x19o - %s' % ( color, name, command.short_desc)) else: acc.append(' \x19%s}%s\x19o' % (color, name)) acc = sorted(acc) buff.extend(acc) acc = [] buff.append('Tab-specific commands:') tab_commands = self.core.current_tab().commands for name, command in enumerate(tab_commands): if isinstance(command, Command): acc.append(' \x19%s}%s\x19o - %s' % ( color, name, command.short_desc)) else: acc.append(' \x19%s}%s\x19o' % (color, name)) acc = sorted(acc) buff.extend(acc) msg = '\n'.join(buff) msg += "\nType /help to know what each command does" else: command = args[0].lstrip('/').strip() tab_commands = self.core.current_tab().commands if command in tab_commands: tup = tab_commands[command] elif command in self.core.commands: tup = self.core.commands[command] else: self.core.information('Unknown command: %s' % command, 'Error') return if isinstance(tup, Command): msg = 'Usage: /%s %s\n' % (command, tup.usage) msg += tup.desc else: msg = tup[1] self.core.information(msg, 'Help') @command_args_parser.quoted(1) def runkey(self, args): """ /runkey """ def replace_line_breaks(key): "replace ^J with \n" if key == '^J': return '\n' return key if args is None: return self.help('runkey') char = args[0] func = self.core.key_func.get(char, None) if func: func() else: res = self.core.do_command(replace_line_breaks(char), False) if res: self.core.refresh_window() @command_args_parser.quoted(1, 1, [None]) def status(self, args): """ /status [msg] """ if args is None: return self.help('status') if not args[0] in POSSIBLE_SHOW.keys(): return self.help('status') show = POSSIBLE_SHOW[args[0]] msg = args[1] pres = self.core.xmpp.make_presence() if msg: pres['status'] = msg pres['type'] = show self.core.events.trigger('send_normal_presence', pres) pres.send() current = self.core.current_tab() is_muctab = isinstance(current, tabs.MucTab) if is_muctab and current.joined and show in ('away', 'xa'): current.send_chat_state('inactive') for tab in self.core.tabs: if isinstance(tab, tabs.MucTab) and tab.joined: muc.change_show(self.core.xmpp, tab.name, tab.own_nick, show, msg) if hasattr(tab, 'directed_presence'): del tab.directed_presence self.core.set_status(show, msg) if is_muctab and current.joined and show not in ('away', 'xa'): current.send_chat_state('active') @command_args_parser.quoted(1, 2, [None, None]) def presence(self, args): """ /presence [type] [status] """ if args is None: return self.help('presence') jid, type, status = args[0], args[1], args[2] if jid == '.' and isinstance(self.core.current_tab(), tabs.ChatTab): jid = self.core.current_tab().name if type == 'available': type = None try: pres = self.core.xmpp.make_presence(pto=jid, ptype=type, pstatus=status) self.core.events.trigger('send_normal_presence', pres) pres.send() except: self.core.information('Could not send directed presence', 'Error') log.debug('Could not send directed presence to %s', jid, exc_info=True) return tab = self.core.get_tab_by_name(jid) if tab: if type in ('xa', 'away'): tab.directed_presence = False chatstate = 'inactive' else: tab.directed_presence = True chatstate = 'active' if tab == self.core.current_tab(): tab.send_chat_state(chatstate, True) if isinstance(tab, tabs.MucTab): for private in tab.privates: private.directed_presence = tab.directed_presence if self.core.current_tab() in tab.privates: self.core.current_tab().send_chat_state(chatstate, True) @command_args_parser.quoted(1) def theme(self, args=None): """/theme """ if args is None: return self.help('theme') self.set('theme %s' % (args[0],)) @command_args_parser.quoted(1) def win(self, args): """ /win """ if args is None: return self.help('win') nb = args[0] try: nb = int(nb) except ValueError: pass if self.core.current_tab_nb == nb: return self.core.previous_tab_nb = self.core.current_tab_nb old_tab = self.core.current_tab() if isinstance(nb, int): if 0 <= nb < len(self.core.tabs): if not self.core.tabs[nb]: return self.core.current_tab_nb = nb else: matchs = [] for tab in self.core.tabs: for name in tab.matching_names(): if nb.lower() in name[1].lower(): matchs.append((name[0], tab)) self.core.current_tab_nb = tab.nb if not matchs: return tab = min(matchs, key=lambda m: m[0])[1] self.core.current_tab_nb = tab.nb old_tab.on_lose_focus() self.core.current_tab().on_gain_focus() self.core.refresh_window() @command_args_parser.quoted(2) def move_tab(self, args): """ /move_tab old_pos new_pos """ if args is None: return self.help('move_tab') current_tab = self.core.current_tab() if args[0] == '.': args[0] = current_tab.nb if args[1] == '.': args[1] = current_tab.nb def get_nb_from_value(value): "parse the cmdline to guess the tab the users wants" ref = None try: ref = int(value) except ValueError: old_tab = None for tab in self.core.tabs: if not old_tab and value == tab.name: old_tab = tab if not old_tab: self.core.information("Tab %s does not exist" % args[0], "Error") return None ref = old_tab.nb return ref old = get_nb_from_value(args[0]) new = get_nb_from_value(args[1]) if new is None or old is None: return self.core.information('Unable to move the tab.', 'Info') result = self.core.insert_tab(old, new) if not result: self.core.information('Unable to move the tab.', 'Info') else: self.core.current_tab_nb = self.core.tabs.index(current_tab) self.core.refresh_window() @command_args_parser.quoted(0, 1) def list(self, args): """ /list [server] Opens a MucListTab containing the list of the room in the specified server """ if args is None: return self.help('list') elif args: jid = safeJID(args[0]) else: if not isinstance(self.core.current_tab(), tabs.MucTab): return self.core.information('Please provide a server', 'Error') jid = safeJID(self.core.current_tab().name).server list_tab = tabs.MucListTab(self.core, jid) self.core.add_tab(list_tab, True) cb = list_tab.on_muc_list_item_received self.core.xmpp.plugin['xep_0030'].get_items(jid=jid, callback=cb) @command_args_parser.quoted(1) def version(self, args): """ /version """ def callback(res): "Callback for /version" 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 is None: return self.help('version') jid = safeJID(args[0]) if jid.resource or jid not in roster: fixes.get_version(self.core.xmpp, jid, callback=callback) elif jid in roster: for resource in roster[jid].resources: fixes.get_version(self.core.xmpp, resource.jid, callback=callback) else: fixes.get_version(self.core.xmpp, jid, callback=callback) def _empty_join(self): tab = self.core.current_tab() if not isinstance(tab, (tabs.MucTab, tabs.PrivateTab)): return (None, None) room = safeJID(tab.name).bare nick = tab.own_nick return (room, nick) def _parse_join_jid(self, jid_string): # we try to join a server directly if jid_string.startswith('@'): server_root = True info = safeJID(jid_string[1:]) else: info = safeJID(jid_string) server_root = False set_nick = '' if len(jid_string) > 1 and jid_string.startswith('/'): set_nick = jid_string[1:] elif info.resource: set_nick = info.resource # happens with /join /nickname, which is OK if info.bare == '': tab = self.core.current_tab() if not isinstance(tab, tabs.MucTab): room, set_nick = (None, None) else: room = tab.name if not set_nick: set_nick = tab.own_nick else: room = info.bare # no server is provided, like "/join hello": # use the server of the current room if available # check if the current room's name has a server if room.find('@') == -1 and not server_root: tab = self.core.current_tab() if isinstance(tab, tabs.MucTab): if tab.name.find('@') != -1: domain = safeJID(tab.name).domain room += '@%s' % domain return (room, set_nick) @command_args_parser.quoted(0, 2) def join(self, args): """ /join [room][/nick] [password] """ if len(args) == 0: room, nick = self._empty_join() else: room, nick = self._parse_join_jid(args[0]) if not room and not nick: return # nothing was parsed room = room.lower() if nick == '': nick = self.core.own_nick # a password is provided if len(args) == 2: password = args[1] else: password = config.get_by_tabname('password', room, fallback=False) if room in self.core.pending_invites: del self.core.pending_invites[room] tab = self.core.get_tab_by_name(room, tabs.MucTab) # New tab if tab is None: tab = self.core.open_new_room(room, nick, password=password) tab.join() else: self.core.focus_tab_named(tab.name) if tab.own_nick == nick and tab.joined: self.core.information('/join: Nothing to do.', 'Info') else: tab.command_part('') tab.own_nick = nick tab.password = password tab.join() if tab == self.core.current_tab(): tab.refresh() self.core.doupdate() @command_args_parser.quoted(0, 2) def bookmark_local(self, args): """ /bookmark_local [room][/nick] [password] """ if not args and not isinstance(self.core.current_tab(), tabs.MucTab): return password = args[1] if len(args) > 1 else None jid = args[0] if args else None self._add_bookmark(jid, True, password, 'local') @command_args_parser.quoted(0, 3) def bookmark(self, args): """ /bookmark [room][/nick] [autojoin] [password] """ if not args and not isinstance(self.core.current_tab(), tabs.MucTab): return jid = args[0] if args else '' password = args[2] if len(args) > 2 else None if not config.get('use_remote_bookmarks'): return self._add_bookmark(jid, True, password, 'local') if len(args) > 1: autojoin = False if args[1].lower() != 'true' else True else: autojoin = True self._add_bookmark(jid, autojoin, password, 'remote') def _add_bookmark(self, jid, autojoin, password, method): nick = None if not jid: tab = self.core.current_tab() roomname = tab.name if tab.joined and tab.own_nick != self.core.own_nick: nick = tab.own_nick if password is None and tab.password is not None: password = tab.password elif jid == '*': return self._add_wildcard_bookmarks(method) else: info = safeJID(jid) roomname, nick = info.bare, info.resource if roomname == '': tab = self.core.current_tab() if not isinstance(tab, tabs.MucTab): return roomname = tab.name bookmark = self.core.bookmarks[roomname] if bookmark is None: bookmark = Bookmark(roomname) self.core.bookmarks.append(bookmark) bookmark.method = method bookmark.autojoin = autojoin if nick: bookmark.nick = nick if password: bookmark.password = password def callback(iq): if iq["type"] != "error": self.core.information('Bookmark added.', 'Info') else: self.core.information("Could not add the bookmarks.", "Info") self.core.bookmarks.save_local() self.core.bookmarks.save_remote(self.core.xmpp, callback) def _add_wildcard_bookmarks(self, method): new_bookmarks = [] for tab in self.core.get_tabs(tabs.MucTab): bookmark = self.core.bookmarks[tab.name] if not bookmark: bookmark = Bookmark(tab.name, autojoin=True, method=method) new_bookmarks.append(bookmark) else: bookmark.method = method new_bookmarks.append(bookmark) self.core.bookmarks.remove(bookmark) new_bookmarks.extend(self.core.bookmarks.bookmarks) self.core.bookmarks.set(new_bookmarks) def _cb(iq): if iq["type"] != "error": self.core.information("Bookmarks saved.", "Info") else: self.core.information("Could not save the remote bookmarks.", "Info") self.core.bookmarks.save_local() self.core.bookmarks.save_remote(self.core.xmpp, _cb) @command_args_parser.ignored def bookmarks(self): """/bookmarks""" tab = self.core.get_tab_by_name('Bookmarks', tabs.BookmarksTab) old_tab = self.core.current_tab() if tab: self.core.current_tab_nb = tab.nb else: tab = tabs.BookmarksTab(self.core, self.core.bookmarks) self.core.tabs.append(tab) self.core.current_tab_nb = tab.nb old_tab.on_lose_focus() tab.on_gain_focus() self.core.refresh_window() @command_args_parser.quoted(0, 1) def remove_bookmark(self, args): """/remove_bookmark [jid]""" def cb(success): if success: self.core.information('Bookmark deleted', 'Info') else: self.core.information('Error while deleting the bookmark', 'Error') if not args: tab = self.core.current_tab() if isinstance(tab, tabs.MucTab) and self.core.bookmarks[tab.name]: self.core.bookmarks.remove(tab.name) self.core.bookmarks.save(self.core.xmpp, callback=cb) else: self.core.information('No bookmark to remove', 'Info') else: if self.core.bookmarks[args[0]]: self.core.bookmarks.remove(args[0]) self.core.bookmarks.save(self.core.xmpp, callback=cb) else: self.core.information('No bookmark to remove', 'Info') @command_args_parser.quoted(0, 3) def set(self, args): """ /set [module|][section]