# 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 permission.
import logging
from asyncio import Future
from datetime import datetime, timedelta
from typing import (
Dict,
Optional
)
from slixmpp.plugins import BasePlugin
from slixmpp import JID
from slixmpp.stanza import Iq
from slixmpp.exceptions import XMPPError
from slixmpp.xmlstream import register_stanza_plugin
from slixmpp.xmlstream.handler import CoroutineCallback
from slixmpp.xmlstream.matcher import StanzaPath
from slixmpp.plugins.xep_0012 import stanza, LastActivity
log = logging.getLogger(__name__)
[docs]
class XEP_0012(BasePlugin):
"""
XEP-0012 Last Activity
"""
name = 'xep_0012'
description = 'XEP-0012: Last Activity'
dependencies = {'xep_0030'}
stanza = stanza
def plugin_init(self):
register_stanza_plugin(Iq, LastActivity)
self._last_activities = {}
self.xmpp.register_handler(
CoroutineCallback('Last Activity',
StanzaPath('iq@type=get/last_activity'),
self._handle_get_last_activity))
self.api.register(self._default_get_last_activity,
'get_last_activity',
default=True)
self.api.register(self._default_set_last_activity,
'set_last_activity',
default=True)
self.api.register(self._default_del_last_activity,
'del_last_activity',
default=True)
def plugin_end(self):
self.xmpp.remove_handler('Last Activity')
self.xmpp['xep_0030'].del_feature(feature='jabber:iq:last')
def session_bind(self, jid):
self.xmpp['xep_0030'].add_feature('jabber:iq:last')
[docs]
def begin_idle(self, jid: Optional[JID] = None, status: Optional[str] = None) -> Future:
"""Reset the last activity for the given JID.
.. versionchanged:: 1.8.0
This function now returns a Future.
:param status: Optional status.
"""
return self.set_last_activity(jid, 0, status)
[docs]
def end_idle(self, jid: Optional[JID] = None) -> Future:
"""Remove the last activity of a JID.
.. versionchanged:: 1.8.0
This function now returns a Future.
"""
return self.del_last_activity(jid)
[docs]
def start_uptime(self, status: Optional[str] = None) -> Future:
"""
.. versionchanged:: 1.8.0
This function now returns a Future.
"""
return self.set_last_activity(None, 0, status)
[docs]
def set_last_activity(self, jid=None, seconds=None, status=None) -> Future:
"""Set last activity for a JID.
.. versionchanged:: 1.8.0
This function now returns a Future.
"""
return self.api['set_last_activity'](jid, args={
'seconds': seconds,
'status': status
})
[docs]
def del_last_activity(self, jid: JID) -> Future:
"""Remove the last activity of a JID.
.. versionchanged:: 1.8.0
This function now returns a Future.
"""
return self.api['del_last_activity'](jid)
[docs]
def get_last_activity(self, jid: JID, local: bool = False,
ifrom: Optional[JID] = None, **iqkwargs) -> Future:
"""Get last activity for a specific JID.
:param local: Fetch the value from the local cache.
"""
if jid is not None and not isinstance(jid, JID):
jid = JID(jid)
if self.xmpp.is_component:
if jid.domain == self.xmpp.boundjid.domain:
local = True
else:
if str(jid) == str(self.xmpp.boundjid):
local = True
jid = jid.full
if local or jid in (None, ''):
log.debug("Looking up local last activity data for %s", jid)
return self.api['get_last_activity'](jid, None, ifrom, None)
iq = self.xmpp.make_iq_get(ito=jid, ifrom=ifrom)
iq.enable('last_activity')
return iq.send(**iqkwargs)
async def _handle_get_last_activity(self, iq: Iq):
log.debug("Received last activity query from " + \
"<%s> to <%s>.", iq['from'], iq['to'])
reply = await self.api['get_last_activity'](iq['to'], None, iq['from'], iq)
reply.send()
# =================================================================
# Default in-memory implementations for storing last activity data.
# =================================================================
def _default_set_last_activity(self, jid: JID, node: str, ifrom: JID, data: Dict):
seconds = data.get('seconds', None)
if seconds is None:
seconds = 0
status = data.get('status', None)
if status is None:
status = ''
self._last_activities[jid] = {
'seconds': datetime.now() - timedelta(seconds=seconds),
'status': status}
def _default_del_last_activity(self, jid: JID, node: str, ifrom: JID, data: Dict):
if jid in self._last_activities:
del self._last_activities[jid]
def _default_get_last_activity(self, jid: JID, node: str, ifrom: JID, iq: Iq) -> Iq:
if not isinstance(iq, Iq):
reply = self.xmpp.Iq()
else:
reply = iq.reply()
if jid not in self._last_activities:
raise XMPPError('service-unavailable')
bare = JID(jid).bare
if bare != self.xmpp.boundjid.bare:
if bare in self.xmpp.roster[jid]:
sub = self.xmpp.roster[jid][bare]['subscription']
if sub not in ('from', 'both'):
raise XMPPError('forbidden')
td = datetime.now() - self._last_activities[jid]['seconds']
seconds = td.seconds + td.days * 24 * 3600
status = self._last_activities[jid]['status']
reply['last_activity']['seconds'] = seconds
reply['last_activity']['status'] = status
return reply