Skip to content
Back to top

# Source code for slixmpp.plugins.xep_0441.stanza

    # Slixmpp: The Slick XMPP Library
    # Copyright (C) 2021 Mathieu Pasquet
    # This file is part of Slixmpp.
    # See the file LICENSE for copying permissio
    from collections import abc
    import logging
    from typing import (
        Iterable,
    )

    from slixmpp.jid import JID
    from slixmpp.xmlstream import ElementBase, ET

    log = logging.getLogger(__name__)

    class Preferences(ElementBase):
        """MAM Preferences payload.

        .. code-block:: xml

            <iq type='set' id='juliet3'>
              <prefs xmlns='urn:xmpp:mam:2' default='roster'>
                <always>
                  <jid>romeo@montague.lit</jid>
                </always>
                <never>
                  <jid>montague@montague.lit</jid>
                </never>
              </prefs>
            </iq>

        """
        name = 'prefs'
        namespace = 'urn:xmpp:mam:2'
        plugin_attrib = 'mam_prefs'
        #: Available interfaces:
        #:
        #: - ``default``: Default MAM policy (must be one of 'roster', 'always',
        #:   'never'
        #: - ``always``  (``list[JID]``): list of JIDs to always store
        #:   conversations with.
        #: - ``never``  (``list[JID]``): list of JIDs to never store
        #:   conversations with.
        interfaces = {'default', 'always', 'never'}
        sub_interfaces = {'always', 'never'}

        def get_always(self) -> set[JID]:
            """
            Get a usable set of JIDs the server always stores conversations for.
            """
            results = set()

            jids = self.xml.findall('{%s}always/{%s}jid' % (
                self.namespace, self.namespace))

            for jid in jids:
                results.add(JID(jid.text))

            return results

        def set_always(self, value: Iterable[JID]):
            """
            Set the MAM policy to always archive for the specified JIDs.

            :param value: List of JIDs.
            """
            if not isinstance(value, abc.Iterable):
                if isinstance(value, JID):
                    log.warning("Wrong type provided to set_always(): %s", value)
                    value = [value]
                else:
                    raise ValueError(f'Wrong type for "value" ({value}).')
            elif isinstance(value, str):
                raise ValueError('Wrong type for "value" (str).')

            self._set_sub_text('always', '', keep=True)
            always = self.xml.find('{%s}always' % self.namespace)
            if always is None:
                raise ValueError("<always/> was not found, it should not happen")
            always.clear()

            for jid in value:
                jid_xml = ET.Element('{%s}jid' % self.namespace)
                jid_xml.text = str(jid)
                always.append(jid_xml)

        def get_never(self) -> set[JID]:
            """
            Get a usable set of JIDs the server never stores conversations for.
            """
            results = set()

            jids = self.xml.findall('{%s}never/{%s}jid' % (
                self.namespace, self.namespace))

            for jid in jids:
                results.add(JID(jid.text))

            return results

        def set_never(self, value: Iterable[JID]):
            """
            Set the MAM policy to never archive for the specified JIDs.

            :param value: List of JIDs.
            """
            if not isinstance(value, abc.Iterable):
                if isinstance(value, JID):
                    log.warning("Wrong type provided to set_never(): %s", value)
                    value = [value]
                else:
                    raise ValueError(f'Wrong type for "value" ({value}).')
            elif isinstance(value, str):
                raise ValueError('Wrong type for "value" (str).')

            self._set_sub_text('never', '', keep=True)
            never = self.xml.find('{%s}never' % self.namespace)
            never.clear()
            for jid in value:
                jid_xml = ET.Element('{%s}jid' % self.namespace)
                jid_xml.text = str(jid)
                never.append(jid_xml)