Skip to content
Back to top

# Source code for slixmpp.plugins.xep_0012.last_activity

    # 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__)

    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')

        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)

        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)

        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)

        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
            })

        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)

        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