diff options
Diffstat (limited to 'slixmpp/plugins/xep_0313/stanza.py')
| -rw-r--r-- | slixmpp/plugins/xep_0313/stanza.py | 371 |
1 files changed, 277 insertions, 94 deletions
diff --git a/slixmpp/plugins/xep_0313/stanza.py b/slixmpp/plugins/xep_0313/stanza.py index 4e43eeba..6633717f 100644 --- a/slixmpp/plugins/xep_0313/stanza.py +++ b/slixmpp/plugins/xep_0313/stanza.py @@ -1,159 +1,342 @@ - # Slixmpp: The Slick XMPP Library # Copyright (C) 2012 Nathanael C. Fritz, Lance J.T. Stout # This file is part of Slixmpp. # See the file LICENSE for copying permissio -import datetime as dt - +from datetime import datetime +from typing import ( + Any, + Iterable, + List, + Optional, + Set, + Union, +) + +from slixmpp.stanza import Message from slixmpp.jid import JID from slixmpp.xmlstream import ElementBase, ET -from slixmpp.plugins import xep_0082, xep_0004 +from slixmpp.plugins import xep_0082 class MAM(ElementBase): + """A MAM Query element. + + .. code-block:: xml + + <iq type='set' id='juliet1'> + <query xmlns='urn:xmpp:mam:2'> + <x xmlns='jabber:x:data' type='submit'> + <field var='FORM_TYPE' type='hidden'> + <value>urn:xmpp:mam:2</value> + </field> + <field var='with'> + <value>juliet@capulet.lit</value> + </field> + </x> + </query> + </iq> + + """ name = 'query' namespace = 'urn:xmpp:mam:2' plugin_attrib = 'mam' - interfaces = {'queryid', 'start', 'end', 'with', 'results'} - sub_interfaces = {'start', 'end', 'with'} + #: Available interfaces: + #: + #: - ``queryid``: The MAM query id + #: - ``start`` and ``end``: Temporal boundaries of the query + #: - ``with``: JID of the other entity the conversation is with + #: - ``after_id``: Fetch stanzas after this specific ID + #: - ``before_id``: Fetch stanzas before this specific ID + #: - ``ids``: Fetch the stanzas matching those IDs + #: - ``results``: pseudo-interface used to accumulate MAM results during + #: fetch, not relevant for the stanza itself. + interfaces = { + 'queryid', 'start', 'end', 'with', 'results', + 'before_id', 'after_id', 'ids', + } + sub_interfaces = {'start', 'end', 'with', 'before_id', 'after_id', 'ids'} def setup(self, xml=None): ElementBase.setup(self, xml) - self._form = xep_0004.stanza.Form() - self._form['type'] = 'submit' - field = self._form.add_field(var='FORM_TYPE', ftype='hidden', - value='urn:xmpp:mam:2') - self.append(self._form) - self._results = [] - - def __get_fields(self): - return self._form.get_fields() - - def get_start(self): - fields = self.__get_fields() + self._results: List[Message] = [] + + def _setup_form(self): + found = self.xml.find( + '{jabber:x:data}x/' + '{jabber:x:data}field[@var="FORM_TYPE"]/' + "{jabber:x:data}value[.='urn:xmpp:mam:2']" + ) + if found is None: + self['form']['type'] = 'submit' + self['form'].add_field( + var='FORM_TYPE', ftype='hidden', value='urn:xmpp:mam:2' + ) + + def get_fields(self): + form = self.get_plugin('form', check=True) + if not form: + return {} + return form.get_fields() + + def get_start(self) -> Optional[datetime]: + fields = self.get_fields() field = fields.get('start') if field: return xep_0082.parse(field['value']) + return None - def set_start(self, value): - if isinstance(value, dt.datetime): + def set_start(self, value: Union[str, datetime]): + self._setup_form() + if isinstance(value, datetime): value = xep_0082.format_datetime(value) - fields = self.__get_fields() - field = fields.get('start') - if field: - field['value'] = value - else: - field = self._form.add_field(var='start') - field['value'] = value + self.set_custom_field('start', value) - def get_end(self): - fields = self.__get_fields() + def get_end(self) -> Optional[datetime]: + fields = self.get_fields() field = fields.get('end') if field: return xep_0082.parse(field['value']) + return None - def set_end(self, value): - if isinstance(value, dt.datetime): + def set_end(self, value: Union[str, datetime]): + if isinstance(value, datetime): value = xep_0082.format_datetime(value) - fields = self.__get_fields() - field = fields.get('end') - if field: - field['value'] = value - else: - field = self._form.add_field(var='end') - field['value'] = value + self.set_custom_field('end', value) - def get_with(self): - fields = self.__get_fields() + def get_with(self) -> Optional[JID]: + fields = self.get_fields() field = fields.get('with') if field: return JID(field['value']) + return None - def set_with(self, value): - fields = self.__get_fields() - field = fields.get('with') + def set_with(self, value: JID): + self.set_custom_field('with', value) + + def set_custom_field(self, fieldname: str, value: Any): + self._setup_form() + fields = self.get_fields() + field = fields.get(fieldname) if field: - field['with'] = str(value) + field['value'] = str(value) else: - field = self._form.add_field(var='with') + field = self['form'].add_field(var=fieldname) field['value'] = str(value) + + def get_custom_field(self, fieldname: str) -> Optional[str]: + fields = self.get_fields() + field = fields.get(fieldname) + if field: + return field['value'] + return None + + def set_before_id(self, value: str): + self.set_custom_field('before-id', value) + + def get_before_id(self): + self.get_custom_field('before-id') + + def set_after_id(self, value: str): + self.set_custom_field('after-id', value) + + def get_after_id(self): + self.get_custom_field('after-id') + + def set_ids(self, value: List[str]): + self._setup_form() + fields = self.get_fields() + field = fields.get('ids') + if field: + field['ids'] = value + else: + field = self['form'].add_field(var='ids') + field['value'] = value + + def get_ids(self): + self.get_custom_field('id') + # The results interface is meant only as an easy # way to access the set of collected message responses # from the query. - def get_results(self): + def get_results(self) -> List[Message]: return self._results - def set_results(self, values): + def set_results(self, values: List[Message]): self._results = values def del_results(self): self._results = [] -class Preferences(ElementBase): - name = 'prefs' +class Fin(ElementBase): + """A MAM fin element (end of query). + + .. code-block:: xml + + <iq type='result' id='juliet1'> + <fin xmlns='urn:xmpp:mam:2'> + <set xmlns='http://jabber.org/protocol/rsm'> + <first index='0'>28482-98726-73623</first> + <last>09af3-cc343-b409f</last> + </set> + </fin> + </iq> + + """ + name = 'fin' namespace = 'urn:xmpp:mam:2' - plugin_attrib = 'mam_prefs' - interfaces = {'default', 'always', 'never'} - sub_interfaces = {'always', 'never'} + plugin_attrib = 'mam_fin' + interfaces = {'results'} - def get_always(self): - results = set() + def setup(self, xml=None): + ElementBase.setup(self, xml) + self._results: List[Message] = [] - jids = self.xml.findall('{%s}always/{%s}jid' % ( - self.namespace, self.namespace)) + # The results interface is meant only as an easy + # way to access the set of collected message responses + # from the query. - for jid in jids: - results.add(JID(jid.text)) + def get_results(self) -> List[Message]: + return self._results - return results + def set_results(self, values: List[Message]): + self._results = values + + def del_results(self): + self._results = [] - def set_always(self, value): - self._set_sub_text('always', '', keep=True) - always = self.xml.find('{%s}always' % self.namespace) - always.clear() - if not isinstance(value, (list, set)): - value = [value] +class Result(ElementBase): + """A MAM result payload. + + .. code-block:: xml + + <message id='aeb213' to='juliet@capulet.lit/chamber'> + <result xmlns='urn:xmpp:mam:2' queryid='f27' id='28482-98726-73623'> + <forwarded xmlns='urn:xmpp:forward:0'> + <delay xmlns='urn:xmpp:delay' stamp='2010-07-10T23:08:25Z'/> + <message xmlns='jabber:client' from="witch@shakespeare.lit" + to="macbeth@shakespeare.lit"> + <body>Hail to thee</body> + </message> + </forwarded> + </result> + </message> + """ + name = 'result' + namespace = 'urn:xmpp:mam:2' + plugin_attrib = 'mam_result' + #: Available interfaces: + #: + #: - ``queryid``: MAM queryid + #: - ``id``: ID of the result + interfaces = {'queryid', 'id'} - 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): - results = set() +class Metadata(ElementBase): + """Element containing archive metadata - jids = self.xml.findall('{%s}never/{%s}jid' % ( - self.namespace, self.namespace)) + .. code-block:: xml - for jid in jids: - results.add(JID(jid.text)) + <iq type='result' id='jui8921rr9'> + <metadata xmlns='urn:xmpp:mam:2'> + <start id='YWxwaGEg' timestamp='2008-08-22T21:09:04Z' /> + <end id='b21lZ2Eg' timestamp='2020-04-20T14:34:21Z' /> + </metadata> + </iq> - return results + """ + name = 'metadata' + namespace = 'urn:xmpp:mam:2' + plugin_attrib = 'mam_metadata' - def set_never(self, value): - self._set_sub_text('never', '', keep=True) - never = self.xml.find('{%s}never' % self.namespace) - never.clear() - if not isinstance(value, (list, set)): - value = [value] +class Start(ElementBase): + """Metadata about the start of an archive. - for jid in value: - jid_xml = ET.Element('{%s}jid' % self.namespace) - jid_xml.text = str(jid) - never.append(jid_xml) + .. code-block:: xml + <iq type='result' id='jui8921rr9'> + <metadata xmlns='urn:xmpp:mam:2'> + <start id='YWxwaGEg' timestamp='2008-08-22T21:09:04Z' /> + <end id='b21lZ2Eg' timestamp='2020-04-20T14:34:21Z' /> + </metadata> + </iq> -class Fin(ElementBase): - name = 'fin' + """ + name = 'start' namespace = 'urn:xmpp:mam:2' - plugin_attrib = 'mam_fin' - -class Result(ElementBase): - name = 'result' + plugin_attrib = name + #: Available interfaces: + #: + #: - ``id``: ID of the first message of the archive + #: - ``timestamp`` (``datetime``): timestamp of the first message of the + #: archive + interfaces = {'id', 'timestamp'} + + def get_timestamp(self) -> Optional[datetime]: + """Get the timestamp. + + :returns: The timestamp. + """ + stamp = self.xml.attrib.get('timestamp', None) + if stamp is not None: + return xep_0082.parse(stamp) + return stamp + + def set_timestamp(self, value: Union[datetime, str]): + """Set the timestamp. + + :param value: Value of the timestamp (either a datetime or a + XEP-0082 timestamp string. + """ + if isinstance(value, str): + value = xep_0082.parse(value) + value = xep_0082.format_datetime(value) + self.xml.attrib['timestamp'] = value + + +class End(ElementBase): + """Metadata about the end of an archive. + + .. code-block:: xml + + <iq type='result' id='jui8921rr9'> + <metadata xmlns='urn:xmpp:mam:2'> + <start id='YWxwaGEg' timestamp='2008-08-22T21:09:04Z' /> + <end id='b21lZ2Eg' timestamp='2020-04-20T14:34:21Z' /> + </metadata> + </iq> + + """ + name = 'end' namespace = 'urn:xmpp:mam:2' - plugin_attrib = 'mam_result' - interfaces = {'queryid', 'id'} + plugin_attrib = name + #: Available interfaces: + #: + #: - ``id``: ID of the first message of the archive + #: - ``timestamp`` (``datetime``): timestamp of the first message of the + #: archive + interfaces = {'id', 'timestamp'} + + def get_timestamp(self) -> Optional[datetime]: + """Get the timestamp. + + :returns: The timestamp. + """ + stamp = self.xml.attrib.get('timestamp', None) + if stamp is not None: + return xep_0082.parse(stamp) + return stamp + + def set_timestamp(self, value: Union[datetime, str]): + """Set the timestamp. + + :param value: Value of the timestamp (either a datetime or a + XEP-0082 timestamp string. + """ + if isinstance(value, str): + value = xep_0082.parse(value) + value = xep_0082.format_datetime(value) + self.xml.attrib['timestamp'] = value |