diff --git a/poezio/multiuserchat.py b/poezio/multiuserchat.py index 30c36a77..12f97661 100644 --- a/poezio/multiuserchat.py +++ b/poezio/multiuserchat.py @@ -11,18 +11,39 @@ slix plugin """ from xml.etree import ElementTree as ET +from typing import ( + Callable, + Optional, + TYPE_CHECKING, +) from poezio.common import safeJID -from slixmpp import JID -from slixmpp.exceptions import IqError, IqTimeout +from slixmpp import ( + JID, + ClientXMPP, + Iq, +) + import logging log = logging.getLogger(__name__) + +if TYPE_CHECKING: + from poezio.core import Core + from poezio.tabs import Tab + from slixmpp.plugins.xep_0004 import Form + + NS_MUC_ADMIN = 'http://jabber.org/protocol/muc#admin' NS_MUC_OWNER = 'http://jabber.org/protocol/muc#owner' -def destroy_room(xmpp, room, reason='', altroom=''): +def destroy_room( + xmpp: ClientXMPP, + room: str, + reason: str = '', + altroom: str = '' +) -> bool: """ destroy a room """ @@ -42,7 +63,7 @@ def destroy_room(xmpp, room, reason='', altroom=''): query.append(destroy) iq.append(query) - def callback(iq): + def callback(iq: Iq) -> None: if not iq or iq['type'] == 'error': xmpp.core.information('Unable to destroy room %s' % room, 'Info') else: @@ -52,23 +73,13 @@ def destroy_room(xmpp, room, reason='', altroom=''): return True -def send_private_message(xmpp, jid, line): - """ - Send a private message - """ - jid = safeJID(jid) - xmpp.send_message(mto=jid, mbody=line, mtype='chat') - - -def send_groupchat_message(xmpp, jid, line): - """ - Send a message to the groupchat - """ - jid = safeJID(jid) - xmpp.send_message(mto=jid, mbody=line, mtype='groupchat') - - -def change_show(xmpp, jid: JID, own_nick: str, show, status): +def change_show( + xmpp: ClientXMPP, + jid: JID, + own_nick: str, + show: str, + status: Optional[str] +) -> None: """ Change our 'Show' """ @@ -81,7 +92,7 @@ def change_show(xmpp, jid: JID, own_nick: str, show, status): pres.send() -def change_subject(xmpp, jid, subject): +def change_subject(xmpp: ClientXMPP, jid: JID, subject: str) -> None: """ Change the room subject """ @@ -92,7 +103,13 @@ def change_subject(xmpp, jid, subject): msg.send() -def change_nick(core, jid, nick, status=None, show=None): +def change_nick( + core: 'Core', + jid: JID, + nick: str, + status: Optional[str] = None, + show: Optional[str] = None +) -> None: """ Change our own nick in a room """ @@ -103,14 +120,16 @@ def change_nick(core, jid, nick, status=None, show=None): presence.send() -def join_groupchat(core, - jid, - nick, - passwd='', - status=None, - show=None, - seconds=None, - tab=None): +def join_groupchat( + core: 'Core', + jid: JID, + nick: str, + passwd: str = '', + status: Optional[str] = None, + show: Optional[str] = None, + seconds: Optional[int] = None, + tab: Optional['Tab'] = None +) -> None: xmpp = core.xmpp stanza = xmpp.make_presence( pto='%s/%s' % (jid, nick), pstatus=status, pshow=show) @@ -119,8 +138,10 @@ def join_groupchat(core, passelement = ET.Element('password') passelement.text = passwd x.append(passelement) - def on_disco(iq): - if 'urn:xmpp:mam:2' in iq['disco_info'].get_features() or (tab and tab._text_buffer.last_message): + + def on_disco(iq: Iq) -> None: + if ('urn:xmpp:mam:2' in iq['disco_info'].get_features() + or (tab and tab._text_buffer.last_message)): history = ET.Element('{http://jabber.org/protocol/muc}history') history.attrib['seconds'] = str(0) x.append(history) @@ -136,13 +157,15 @@ def join_groupchat(core, xmpp.plugin['xep_0045'].rooms[jid] = {} xmpp.plugin['xep_0045'].our_nicks[jid] = to.resource - try: - xmpp.plugin['xep_0030'].get_info(jid=jid, callback=on_disco) - except (IqError, IqTimeout): - return core.information('Failed to retrieve messages', 'Error') + xmpp.plugin['xep_0030'].get_info(jid=jid, callback=on_disco) -def leave_groupchat(xmpp, jid, own_nick, msg): +def leave_groupchat( + xmpp: ClientXMPP, + jid: JID, + own_nick: str, + msg: str +) -> None: """ Leave the groupchat """ @@ -156,7 +179,14 @@ def leave_groupchat(xmpp, jid, own_nick, msg): exc_info=True) -def set_user_role(xmpp, jid, nick, reason, role, callback=None): +def set_user_role( + xmpp: ClientXMPP, + jid: JID, + nick: str, + reason: str, + role: str, + callback: Callable[[Iq], None] +) -> None: """ (try to) Set the role of a MUC user (role = 'none': eject user) @@ -172,21 +202,18 @@ def set_user_role(xmpp, jid, nick, reason, role, callback=None): query.append(item) iq.append(query) iq['to'] = jid - if callback: - return iq.send(callback=callback) - try: - return iq.send() - except (IqError, IqTimeout) as e: - return e.iq + iq.send(callback=callback) -def set_user_affiliation(xmpp, - muc_jid, - affiliation, - nick=None, - jid=None, - reason=None, - callback=None): +def set_user_affiliation( + xmpp: ClientXMPP, + muc_jid: JID, + affiliation: str, + callback: Callable[[Iq], None], + nick: Optional[str] = None, + jid: Optional[JID] = None, + reason: Optional[str] = None +) -> None: """ (try to) Set the affiliation of a MUC user """ @@ -212,18 +239,10 @@ def set_user_affiliation(xmpp, query.append(item) iq = xmpp.make_iq_set(query) iq['to'] = muc_jid - if callback: - return iq.send(callback=callback) - try: - return xmpp.plugin['xep_0045'].set_affiliation( - str(muc_jid), - str(jid) if jid else None, nick, affiliation) - except: - log.debug('Error setting the affiliation: %s', exc_info=True) - return False + iq.send(callback=callback) -def cancel_config(xmpp, room): +def cancel_config(xmpp: ClientXMPP, room: str) -> None: query = ET.Element('{http://jabber.org/protocol/muc#owner}query') x = ET.Element('{jabber:x:data}x', type='cancel') query.append(x) @@ -232,7 +251,7 @@ def cancel_config(xmpp, room): iq.send() -def configure_room(xmpp, room, form): +def configure_room(xmpp: ClientXMPP, room: str, form: 'Form') -> None: if form is None: return iq = xmpp.make_iq_set()