summaryrefslogtreecommitdiff
path: root/poezio/multiuserchat.py
diff options
context:
space:
mode:
Diffstat (limited to 'poezio/multiuserchat.py')
-rw-r--r--poezio/multiuserchat.py233
1 files changed, 61 insertions, 172 deletions
diff --git a/poezio/multiuserchat.py b/poezio/multiuserchat.py
index 30c36a77..3278e1bd 100644
--- a/poezio/multiuserchat.py
+++ b/poezio/multiuserchat.py
@@ -3,77 +3,51 @@
# This file is part of Poezio.
#
# Poezio is free software: you can redistribute it and/or modify
-# it under the terms of the zlib license. See the COPYING file.
+# it under the terms of the GPL-3.0+ license. See the COPYING file.
"""
Implementation of the XEP-0045: Multi-User Chat.
Add some facilities that are not available on the XEP_0045
slix plugin
"""
+from __future__ import annotations
+
+import asyncio
from xml.etree import ElementTree as ET
+from typing import (
+ Optional,
+ Union,
+ TYPE_CHECKING,
+)
+
+from slixmpp import (
+ JID,
+ ClientXMPP,
+ Iq,
+ Presence,
+)
-from poezio.common import safeJID
-from slixmpp import JID
-from slixmpp.exceptions import IqError, IqTimeout
import logging
log = logging.getLogger(__name__)
-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=''):
- """
- destroy a room
- """
- room = safeJID(room)
- if not room:
- return False
- iq = xmpp.make_iq_set()
- iq['to'] = room
- query = ET.Element('{%s}query' % NS_MUC_OWNER)
- destroy = ET.Element('{%s}destroy' % NS_MUC_OWNER)
- if altroom:
- destroy.attrib['jid'] = altroom
- if reason:
- xreason = ET.Element('{%s}reason' % NS_MUC_OWNER)
- xreason.text = reason
- destroy.append(xreason)
- query.append(destroy)
- iq.append(query)
-
- def callback(iq):
- if not iq or iq['type'] == 'error':
- xmpp.core.information('Unable to destroy room %s' % room, 'Info')
- else:
- xmpp.core.information('Room %s destroyed' % room, 'Info')
-
- iq.send(callback=callback)
- 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')
+if TYPE_CHECKING:
+ from poezio.core.core import Core
+ from poezio.tabs import MucTab
-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'
"""
- jid = safeJID(jid)
- pres = xmpp.make_presence(pto='%s/%s' % (jid, own_nick))
+ jid = JID(jid)
+ pres: Presence = xmpp.make_presence(pto='%s/%s' % (jid, own_nick))
if show: # if show is None, don't put a <show /> tag. It means "available"
pres['type'] = show
if status:
@@ -81,46 +55,45 @@ def change_show(xmpp, jid: JID, own_nick: str, show, status):
pres.send()
-def change_subject(xmpp, jid, subject):
- """
- Change the room subject
- """
- jid = safeJID(jid)
- msg = xmpp.make_message(jid)
- msg['type'] = 'groupchat'
- msg['subject'] = subject
- msg.send()
-
-
-def change_nick(core, jid, nick, status=None, show=None):
+def change_nick(
+ core: Core,
+ jid: Union[JID, str],
+ nick: str,
+ status: Optional[str] = None,
+ show: Optional[str] = None
+) -> None:
"""
Change our own nick in a room
"""
xmpp = core.xmpp
- presence = xmpp.make_presence(
- pshow=show, pstatus=status, pto=safeJID('%s/%s' % (jid, nick)))
+ presence: Presence = xmpp.make_presence(
+ pshow=show, pstatus=status, pto=JID('%s/%s' % (jid, nick)))
core.events.trigger('changing_nick', presence)
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['MucTab'] = None
+) -> None:
xmpp = core.xmpp
- stanza = xmpp.make_presence(
+ stanza: Presence = xmpp.make_presence(
pto='%s/%s' % (jid, nick), pstatus=status, pshow=show)
x = ET.Element('{http://jabber.org/protocol/muc}x')
if passwd:
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,17 +109,21 @@ def join_groupchat(core,
xmpp.plugin['xep_0045'].rooms[jid] = {}
xmpp.plugin['xep_0045'].our_nicks[jid] = to.resource
- try:
+ asyncio.create_task(
xmpp.plugin['xep_0030'].get_info(jid=jid, callback=on_disco)
- except (IqError, IqTimeout):
- return core.information('Failed to retrieve messages', 'Error')
+ )
-def leave_groupchat(xmpp, jid, own_nick, msg):
+def leave_groupchat(
+ xmpp: ClientXMPP,
+ jid: JID,
+ own_nick: str,
+ msg: str
+) -> None:
"""
Leave the groupchat
"""
- jid = safeJID(jid)
+ jid = JID(jid)
try:
xmpp.plugin['xep_0045'].leave_muc(jid, own_nick, msg)
except KeyError:
@@ -154,91 +131,3 @@ def leave_groupchat(xmpp, jid, own_nick, msg):
"muc.leave_groupchat: could not leave the room %s",
jid,
exc_info=True)
-
-
-def set_user_role(xmpp, jid, nick, reason, role, callback=None):
- """
- (try to) Set the role of a MUC user
- (role = 'none': eject user)
- """
- jid = safeJID(jid)
- iq = xmpp.make_iq_set()
- query = ET.Element('{%s}query' % NS_MUC_ADMIN)
- item = ET.Element('{%s}item' % NS_MUC_ADMIN, {'nick': nick, 'role': role})
- if reason:
- reason_el = ET.Element('{%s}reason' % NS_MUC_ADMIN)
- reason_el.text = reason
- item.append(reason_el)
- 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
-
-
-def set_user_affiliation(xmpp,
- muc_jid,
- affiliation,
- nick=None,
- jid=None,
- reason=None,
- callback=None):
- """
- (try to) Set the affiliation of a MUC user
- """
- muc_jid = safeJID(muc_jid)
- query = ET.Element('{http://jabber.org/protocol/muc#admin}query')
- if nick:
- item = ET.Element('{http://jabber.org/protocol/muc#admin}item', {
- 'affiliation': affiliation,
- 'nick': nick
- })
- else:
- item = ET.Element('{http://jabber.org/protocol/muc#admin}item', {
- 'affiliation': affiliation,
- 'jid': str(jid)
- })
-
- if reason:
- reason_item = ET.Element(
- '{http://jabber.org/protocol/muc#admin}reason')
- reason_item.text = reason
- item.append(reason_item)
-
- 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
-
-
-def cancel_config(xmpp, room):
- query = ET.Element('{http://jabber.org/protocol/muc#owner}query')
- x = ET.Element('{jabber:x:data}x', type='cancel')
- query.append(x)
- iq = xmpp.make_iq_set(query)
- iq['to'] = room
- iq.send()
-
-
-def configure_room(xmpp, room, form):
- if form is None:
- return
- iq = xmpp.make_iq_set()
- iq['to'] = room
- query = ET.Element('{http://jabber.org/protocol/muc#owner}query')
- form['type'] = 'submit'
- query.append(form.xml)
- iq.append(query)
- iq.send()
generated by cgit v1.2.3 (git 2.25.1) at 2024-05-03 06:07:31 +0000