# This file is part of Gajim.
#
# Gajim is free software; you can redistribute it and/or modify
# it under the terms of the GNU General Public License as published
# by the Free Software Foundation; version 3 only.
#
# Gajim is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
# GNU General Public License for more details.
#
# You should have received a copy of the GNU General Public License
# along with Gajim. If not, see <http://www.gnu.org/licenses/>.
# XEP-0085: Chat State Notifications
from typing import Any
from typing import Dict # pylint: disable=unused-import
from typing import Optional
from typing import Tuple
import time
import logging
import nbxmpp
from gi.repository import GLib
from gajim.common import app
from gajim.common.nec import NetworkEvent
from gajim.common.const import Chatstate as State
from gajim.common.modules.misc import parse_delay
from gajim.common.connection_handlers_events import MessageOutgoingEvent
from gajim.common.connection_handlers_events import GcMessageOutgoingEvent
from gajim.common.types import ContactT
from gajim.common.types import ConnectionT
log = logging.getLogger('gajim.c.m.chatstates')
INACTIVE_AFTER = 60
PAUSED_AFTER = 5
def parse_chatstate(stanza: nbxmpp.Message) -> Optional[str]:
if parse_delay(stanza) is not None:
return None
children = stanza.getChildren()
for child in children:
if child.getNamespace() == nbxmpp.NS_CHATSTATES:
return child.getName()
return None
class Chatstate:
def __init__(self, con: ConnectionT) -> None:
self._con = con
self._account = con.name
self.handlers = [
('presence', self._presence_received),
]
self._chatstates = {} # type: Dict[str, State]
self._last_keyboard_activity = {} # type: Dict[str, float]
self._last_mouse_activity = {} # type: Dict[str, float]
self._timeout_id = GLib.timeout_add_seconds(
2, self._check_last_interaction)
def _presence_received(self,
_con: ConnectionT,
stanza: nbxmpp.Presence) -> None:
if stanza.getType() not in ('unavailable', 'error'):
return
full_jid = stanza.getFrom()
if full_jid is None or self._con.get_own_jid().bareMatch(full_jid):
# Presence from ourself
return
contact = app.contacts.get_gc_contact(
self._account, full_jid.getStripped(), full_jid.getResource())
if contact is None:
contact = app.contacts.get_contact_from_full_jid(
self._account, str(full_jid))
if contact is None:
return
if contact.chatstate is None:
return
if contact.is_gc_contact:
jid = contact.get_full_jid()
else:
jid = contact.jid
contact.chatstate = None
self._chatstates.pop(jid, None)
self._last_mouse_activity.pop(jid, None)
self._last_keyboard_activity.pop(jid, None)
log.info('Reset chatstate for %s', jid)
app.nec.push_outgoing_event(
NetworkEvent('chatstate-received',
account=self._account,
contact=contact))
def delegate(self, event: Any) -> None:
if self._con.get_own_jid().bareMatch(event.jid) or event.sent:
# Dont show chatstates from our own resources
return
if event.mtype == 'groupchat':
# Not implemented yet
return
chatstate = parse_chatstate(event.stanza)
if chatstate is None:
return
if event.muc_pm:
contact = app.contacts.get_gc_contact(
self._account, event.jid, event.resource)
else:
contact = app.contacts.get_contact_from_full_jid(
self._account, event.fjid)
if contact is None:
return
contact.chatstate = chatstate
log.info('Recv: %-10s - %s', chatstate, event.fjid)
app.nec.push_outgoing_event(
NetworkEvent('chatstate-received',
account=self._account,
contact=contact))
def _check_last_interaction(self) -> GLib.SOURCE_CONTINUE:
setting = app.config.get('outgoing_chat_state_notifications')
if setting in ('composing_only', 'disabled'):
return GLib.SOURCE_CONTINUE
now = time.time()
for jid in list(self._last_mouse_activity.keys()):
time_ = self._last_mouse_activity[jid]
current_state = self._chatstates.get(jid)
if current_state is None:
self._last_mouse_activity.pop(jid, None)
self._last_keyboard_activity.pop(jid, None)
continue
if current_state in (State.GONE, State.INACTIVE):
continue
new_chatstate = None
if now - time_ > INACTIVE_AFTER:
new_chatstate = State.INACTIVE
elif current_state == State.COMPOSING:
key_time = self._last_keyboard_activity[jid]
if now - key_time > PAUSED_AFTER:
new_chatstate = State.PAUSED
if new_chatstate is not None:
if self._chatstates.get(jid) != new_chatstate:
contact = app.contacts.get_contact(self._account, jid)
if contact is None:
room, nick = app.get_room_and_nick_from_fjid(jid)
contact = app.contacts.get_gc_contact(
self._account, room, nick)
if contact is not None:
contact = contact.as_contact()
else:
# Contact not found, maybe we left the group chat
# or the contact was removed from the roster
log.info(
'Contact %s not found, reset chatstate', jid)
self._chatstates.pop(jid, None)
self._last_mouse_activity.pop(jid, None)
self._last_keyboard_activity.pop(jid, None)
continue
self.set_chatstate(contact, new_chatstate)
return GLib.SOURCE_CONTINUE
def set_active(self, jid: str) -> None:
self._last_mouse_activity[jid] = time.time()
setting = app.config.get('outgoing_chat_state_notifications')
if setting == 'disabled':
return
self._chatstates[jid] = State.ACTIVE
def get_active_chatstate(self, contact: ContactT) -> Optional[str]:
# determines if we add 'active' on outgoing messages
setting = app.config.get('outgoing_chat_state_notifications')
if setting == 'disabled':
return None
if not contact.is_groupchat():
# Dont send chatstates to ourself
if self._con.get_own_jid().bareMatch(contact.jid):
return None
if not contact.supports(nbxmpp.NS_CHATSTATES):
return None
self.set_active(contact.jid)
return 'active'
def set_chatstate(self, contact: ContactT, state: State) -> None:
current_state = self._chatstates.get(contact.jid)
setting = app.config.get('outgoing_chat_state_notifications')
if setting == 'disabled':
# Send a last 'active' state after user disabled chatstates
if current_state is not None:
log.info('Send: %-10s - %s', State.ACTIVE, contact.jid)
event_attrs = {'account': self._account,
'jid': contact.jid,
'chatstate': str(State.ACTIVE)}
if contact.is_groupchat():
app.nec.push_outgoing_event(
GcMessageOutgoingEvent(None, **event_attrs))
else:
app.nec.push_outgoing_event(
MessageOutgoingEvent(None, **event_attrs))
self._chatstates.pop(contact.jid, None)
self._last_mouse_activity.pop(contact.jid, None)
self._last_keyboard_activity.pop(contact.jid, None)
return
if not contact.is_groupchat():
# Dont leak presence to contacts
# which are not allowed to see our status
if not contact.is_pm_contact:
if contact and contact.sub in ('to', 'none'):
return
if contact.show == 'offline':
return
if not contact.supports(nbxmpp.NS_CHATSTATES):
return
if state in (State.ACTIVE, State.COMPOSING):
self._last_mouse_activity[contact.jid] = time.time()
if setting == 'composing_only':
if state in (State.INACTIVE, State.GONE, State.PAUSED):
state = State.ACTIVE
if current_state == state:
return
# Dont send chatstates to ourself
if self._con.get_own_jid().bareMatch(contact.jid):
return
log.info('Send: %-10s - %s', state, contact.jid)
event_attrs = {'account': self._account,
'jid': contact.jid,
'chatstate': str(state)}
if contact.is_groupchat():
app.nec.push_outgoing_event(
GcMessageOutgoingEvent(None, **event_attrs))
else:
app.nec.push_outgoing_event(
MessageOutgoingEvent(None, **event_attrs))
self._chatstates[contact.jid] = state
def set_mouse_activity(self, contact: ContactT) -> None:
self._last_mouse_activity[contact.jid] = time.time()
setting = app.config.get('outgoing_chat_state_notifications')
if setting == 'disabled':
return
if self._chatstates.get(contact.jid) == State.INACTIVE:
self.set_chatstate(contact, State.ACTIVE)
def set_keyboard_activity(self, contact: ContactT) -> None:
self._last_keyboard_activity[contact.jid] = time.time()
def cleanup(self):
GLib.source_remove(self._timeout_id)
def get_instance(*args: Any, **kwargs: Any) -> Tuple[Chatstate, str]:
return Chatstate(*args, **kwargs), 'Chatstate'