summaryrefslogtreecommitdiff
path: root/slixmpp/plugins/xep_0012/last_activity.py
diff options
context:
space:
mode:
Diffstat (limited to 'slixmpp/plugins/xep_0012/last_activity.py')
-rw-r--r--slixmpp/plugins/xep_0012/last_activity.py312
1 files changed, 156 insertions, 156 deletions
diff --git a/slixmpp/plugins/xep_0012/last_activity.py b/slixmpp/plugins/xep_0012/last_activity.py
index 36d21b29..76a68c7a 100644
--- a/slixmpp/plugins/xep_0012/last_activity.py
+++ b/slixmpp/plugins/xep_0012/last_activity.py
@@ -1,156 +1,156 @@
-"""
- 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 datetime import datetime, timedelta
-
-from slixmpp.plugins import BasePlugin, register_plugin
-from slixmpp import Iq
-from slixmpp.exceptions import XMPPError
-from slixmpp.xmlstream import JID, register_stanza_plugin
-from slixmpp.xmlstream.handler import Callback
-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 = set(['xep_0030'])
- stanza = stanza
-
- def plugin_init(self):
- register_stanza_plugin(Iq, LastActivity)
-
- self._last_activities = {}
-
- self.xmpp.register_handler(
- Callback('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=None, status=None):
- self.set_last_activity(jid, 0, status)
-
- def end_idle(self, jid=None):
- self.del_last_activity(jid)
-
- def start_uptime(self, status=None):
- self.set_last_activity(jid, 0, status)
-
- def set_last_activity(self, jid=None, seconds=None, status=None):
- self.api['set_last_activity'](jid, args={
- 'seconds': seconds,
- 'status': status})
-
- def del_last_activity(self, jid):
- self.api['del_last_activity'](jid)
-
- def get_last_activity(self, jid, local=False, ifrom=None, timeout=None,
- callback=None, timeout_callback=None):
- 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.Iq()
- iq['from'] = ifrom
- iq['to'] = jid
- iq['type'] = 'get'
- iq.enable('last_activity')
- return iq.send(timeout=timeout, callback=callback,
- timeout_callback=timeout_callback)
-
- def _handle_get_last_activity(self, iq):
- log.debug("Received last activity query from " + \
- "<%s> to <%s>.", iq['from'], iq['to'])
- reply = 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, node, ifrom, data):
- 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, node, ifrom, data):
- if jid in self._last_activities:
- del self._last_activities[jid]
-
- def _default_get_last_activity(self, jid, node, ifrom, iq):
- if not isinstance(iq, Iq):
- reply = self.xmpp.Iq()
- else:
- iq.reply()
- reply = iq
-
- 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
+"""
+ 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 datetime import datetime, timedelta
+
+from slixmpp.plugins import BasePlugin, register_plugin
+from slixmpp import Iq
+from slixmpp.exceptions import XMPPError
+from slixmpp.xmlstream import JID, register_stanza_plugin
+from slixmpp.xmlstream.handler import Callback
+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 = set(['xep_0030'])
+ stanza = stanza
+
+ def plugin_init(self):
+ register_stanza_plugin(Iq, LastActivity)
+
+ self._last_activities = {}
+
+ self.xmpp.register_handler(
+ Callback('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=None, status=None):
+ self.set_last_activity(jid, 0, status)
+
+ def end_idle(self, jid=None):
+ self.del_last_activity(jid)
+
+ def start_uptime(self, status=None):
+ self.set_last_activity(jid, 0, status)
+
+ def set_last_activity(self, jid=None, seconds=None, status=None):
+ self.api['set_last_activity'](jid, args={
+ 'seconds': seconds,
+ 'status': status})
+
+ def del_last_activity(self, jid):
+ self.api['del_last_activity'](jid)
+
+ def get_last_activity(self, jid, local=False, ifrom=None, timeout=None,
+ callback=None, timeout_callback=None):
+ 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.Iq()
+ iq['from'] = ifrom
+ iq['to'] = jid
+ iq['type'] = 'get'
+ iq.enable('last_activity')
+ return iq.send(timeout=timeout, callback=callback,
+ timeout_callback=timeout_callback)
+
+ def _handle_get_last_activity(self, iq):
+ log.debug("Received last activity query from " + \
+ "<%s> to <%s>.", iq['from'], iq['to'])
+ reply = 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, node, ifrom, data):
+ 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, node, ifrom, data):
+ if jid in self._last_activities:
+ del self._last_activities[jid]
+
+ def _default_get_last_activity(self, jid, node, ifrom, iq):
+ if not isinstance(iq, Iq):
+ reply = self.xmpp.Iq()
+ else:
+ iq.reply()
+ reply = iq
+
+ 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
generated by cgit v1.2.3 (git 2.25.1) at 2024-05-04 00:52:18 +0000