343 lines
9.8 KiB
Python
343 lines
9.8 KiB
Python
# 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
|
|
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
|
|
|
|
|
|
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'
|
|
#: 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._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: Union[str, datetime]):
|
|
self._setup_form()
|
|
if isinstance(value, datetime):
|
|
value = xep_0082.format_datetime(value)
|
|
self.set_custom_field('start', value)
|
|
|
|
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: Union[str, datetime]):
|
|
if isinstance(value, datetime):
|
|
value = xep_0082.format_datetime(value)
|
|
self.set_custom_field('end', value)
|
|
|
|
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: 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['value'] = str(value)
|
|
else:
|
|
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) -> List[Message]:
|
|
return self._results
|
|
|
|
def set_results(self, values: List[Message]):
|
|
self._results = values
|
|
|
|
def del_results(self):
|
|
self._results = []
|
|
|
|
|
|
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_fin'
|
|
interfaces = {'results', 'stable', 'complete'}
|
|
|
|
def setup(self, xml=None):
|
|
ElementBase.setup(self, xml)
|
|
self._results: List[Message] = []
|
|
|
|
# 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) -> List[Message]:
|
|
return self._results
|
|
|
|
def set_results(self, values: List[Message]):
|
|
self._results = values
|
|
|
|
def del_results(self):
|
|
self._results = []
|
|
|
|
|
|
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'}
|
|
|
|
|
|
class Metadata(ElementBase):
|
|
"""Element containing archive metadata
|
|
|
|
.. 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 = 'metadata'
|
|
namespace = 'urn:xmpp:mam:2'
|
|
plugin_attrib = 'mam_metadata'
|
|
|
|
|
|
class Start(ElementBase):
|
|
"""Metadata about the start 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 = 'start'
|
|
namespace = 'urn:xmpp:mam:2'
|
|
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 = 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
|