Coverage for slidge/core/mixins/presence.py: 88%
123 statements
« prev ^ index » next coverage.py v7.15.2, created at 2026-09-29 05:05 +0000
1import contextlib
2import re
3from asyncio import Task, sleep
4from datetime import UTC, datetime, timedelta
5from functools import partial
7from slixmpp import Presence
8from slixmpp.types import PresenceShows, PresenceTypes
9from sqlalchemy.orm.exc import DetachedInstanceError
11from ...db.models import Contact, Participant
12from ...util.types import AnySession, CachedPresence
13from .base import BaseSender, SessionBound
14from .db import DBMixin
17class _NoChange(Exception):
18 pass
21_FRIEND_REQUEST_PRESENCES = {"subscribe", "unsubscribe", "subscribed", "unsubscribed"}
22_UPDATE_LAST_SEEN_FALLBACK_TASKS = dict[int, Task[None]]()
23_ONE_WEEK_SECONDS = 3600 * 24 * 7
26async def _update_last_seen_fallback(session: AnySession, contact_pk: int) -> None:
27 await sleep(_ONE_WEEK_SECONDS)
28 with session.xmpp.store.session() as orm:
29 stored = orm.get(Contact, contact_pk)
30 if stored is None:
31 return
32 contact = session.contacts.from_store(stored)
33 contact.send_last_presence(force=True, no_cache_online=False)
36def _clear_last_seen_task(contact_pk: int, _task: Task[None]) -> None:
37 with contextlib.suppress(KeyError):
38 del _UPDATE_LAST_SEEN_FALLBACK_TASKS[contact_pk]
41class PresenceMixin(BaseSender, DBMixin, SessionBound):
42 _ONLY_SEND_PRESENCE_CHANGES = False
44 # this attribute actually only exists for contacts and not participants
45 _updating_info: bool
46 stored: Contact | Participant
47 is_participant: bool
49 def __init__(self, *a: object, **k: object) -> None:
50 super().__init__(*a, **k)
51 # this is only used when a presence is set during Contact.update_info(),
52 # when the contact does not have a DB primary key yet, and is written
53 # to DB at the end of update_info()
54 self.cached_presence: CachedPresence | None = None
56 def __is_contact(self) -> bool:
57 return isinstance(self.stored, Contact)
59 def __stored(self) -> Contact | None:
60 if self.__is_contact():
61 assert isinstance(self.stored, Contact)
62 return self.stored
63 else:
64 assert isinstance(self.stored, Participant)
65 if self.stored.contact_id is None:
66 return None
67 try:
68 return self.stored.contact
69 except DetachedInstanceError:
70 with self.xmpp.store.session() as orm, orm.no_autoflush:
71 orm.add(self.stored)
72 return self.stored.contact
74 @property
75 def __contact_pk(self) -> int | None:
76 stored = self.__stored()
77 return None if stored is None else stored.id
79 def _get_last_presence(self) -> CachedPresence | None:
80 stored = self.__stored()
81 if stored is None:
82 return None
83 stored.ensure_loaded("presence", self.xmpp.store.session)
84 return (
85 CachedPresence(
86 None
87 if stored.last_seen is None
88 else stored.last_seen.replace(tzinfo=UTC),
89 stored.ptype, # type:ignore
90 stored.pstatus,
91 stored.pshow, # type:ignore
92 )
93 if stored.cached_presence
94 else None
95 )
97 def _store_last_presence(self, new: CachedPresence) -> None:
98 if self.__is_contact():
99 contact = self
100 elif (contact := getattr(self, "contact", None)) is None: # type:ignore[assignment]
101 return
102 contact.update_stored_attribute( # type:ignore[attr-defined]
103 cached_presence=True,
104 **new._asdict(),
105 )
107 def _make_presence(
108 self,
109 *,
110 last_seen: datetime | None = None,
111 force: bool = False,
112 bare: bool = False,
113 ptype: PresenceTypes | None = None,
114 pstatus: str | None = None,
115 pshow: PresenceShows | None = None,
116 ) -> Presence:
117 if last_seen and last_seen.tzinfo is None:
118 last_seen = last_seen.astimezone(UTC)
120 if self.is_participant or ptype in _FRIEND_REQUEST_PRESENCES:
121 self.log.debug("Not storing %s", ptype)
122 else:
123 old = self._get_last_presence()
124 new = CachedPresence(
125 last_seen=last_seen, ptype=ptype, pstatus=pstatus, pshow=pshow
126 )
127 if old == new:
128 if force:
129 self.log.debug("Force sending a presence")
130 else:
131 self.log.debug("Presence is the same as cached")
132 raise _NoChange
133 else:
134 self.log.debug("Presence is not the same as cached: %s vs %s", old, new)
135 self._store_last_presence(new)
137 p = self.xmpp.make_presence(
138 pfrom=self.jid.bare if bare else self.jid,
139 ptype=ptype,
140 pshow=pshow,
141 pstatus=pstatus,
142 )
143 if last_seen:
144 # it's ugly to check for the presence of this string, but a better fix is more work
145 if not re.match(
146 ".*Last seen .*", p["status"]
147 ) and self.session.user.preferences.get("last_seen_fallback", True):
148 last_seen_fallback, recent = get_last_seen_fallback(last_seen)
149 if p["status"]:
150 p["status"] = p["status"] + " -- " + last_seen_fallback
151 else:
152 p["status"] = last_seen_fallback
153 pk = self.__contact_pk
154 if recent and pk is not None:
155 # if less than a week, we use sth like 'Last seen: Monday, 8:05",
156 # but if lasts more than a week, this is not very informative, so
157 # we need to force resend an updated presence status
158 task = _UPDATE_LAST_SEEN_FALLBACK_TASKS.get(pk)
159 if task is not None:
160 task.cancel()
161 task = self.session.create_task(
162 _update_last_seen_fallback(self.session, pk),
163 name=f"update last seen fallback of {self}",
164 )
165 _UPDATE_LAST_SEEN_FALLBACK_TASKS[pk] = task
166 task.add_done_callback(partial(_clear_last_seen_task, pk))
167 p["idle"]["since"] = last_seen
168 return p
170 def send_last_presence(
171 self, force: bool = False, no_cache_online: bool = False
172 ) -> None:
173 if (cache := self._get_last_presence()) is None:
174 if force:
175 if no_cache_online:
176 self.online()
177 else:
178 self.offline()
179 return
180 self._send(
181 self._make_presence(
182 last_seen=cache.last_seen,
183 force=True,
184 ptype=cache.ptype,
185 pshow=cache.pshow,
186 pstatus=cache.pstatus,
187 )
188 )
190 def online(
191 self,
192 status: str | None = None,
193 last_seen: datetime | None = None,
194 ) -> None:
195 """
196 Send an "online" presence from this contact to the user.
198 :param status: Arbitrary text, details of the status, eg: "Listening to Britney Spears"
199 :param last_seen: For :xep:`0319`
200 """
201 with contextlib.suppress(_NoChange):
202 self._send(self._make_presence(pstatus=status, last_seen=last_seen))
204 def away(
205 self,
206 status: str | None = None,
207 last_seen: datetime | None = None,
208 ) -> None:
209 """
210 Send an "away" presence from this contact to the user.
212 This is a global status, as opposed to :meth:`.LegacyContact.inactive`
213 which concerns a specific conversation, ie a specific "chat window"
215 :param status: Arbitrary text, details of the status, eg: "Gone to fight capitalism"
216 :param last_seen: For :xep:`0319`
217 """
218 with contextlib.suppress(_NoChange):
219 self._send(
220 self._make_presence(pstatus=status, pshow="away", last_seen=last_seen)
221 )
223 def extended_away(
224 self,
225 status: str | None = None,
226 last_seen: datetime | None = None,
227 ) -> None:
228 """
229 Send an "extended away" presence from this contact to the user.
231 This is a global status, as opposed to :meth:`.LegacyContact.inactive`
232 which concerns a specific conversation, ie a specific "chat window"
234 :param status: Arbitrary text, details of the status, eg: "Gone to fight capitalism"
235 :param last_seen: For :xep:`0319`
236 """
237 with contextlib.suppress(_NoChange):
238 self._send(
239 self._make_presence(pstatus=status, pshow="xa", last_seen=last_seen)
240 )
242 def busy(
243 self,
244 status: str | None = None,
245 last_seen: datetime | None = None,
246 ) -> None:
247 """
248 Send a "busy" (ie, "dnd") presence from this contact to the user,
250 :param status: eg: "Trying to make sense of XEP-0100"
251 :param last_seen: For :xep:`0319`
252 """
253 with contextlib.suppress(_NoChange):
254 self._send(
255 self._make_presence(pstatus=status, pshow="dnd", last_seen=last_seen)
256 )
258 def offline(
259 self,
260 status: str | None = None,
261 last_seen: datetime | None = None,
262 ) -> None:
263 """
264 Send an "offline" presence from this contact to the user.
266 :param status: eg: "Trying to make sense of XEP-0100"
267 :param last_seen: For :xep:`0319`
268 """
269 with contextlib.suppress(_NoChange):
270 self._send(
271 self._make_presence(
272 pstatus=status, ptype="unavailable", last_seen=last_seen
273 )
274 )
277def get_last_seen_fallback(last_seen: datetime) -> tuple[str, bool]:
278 now = datetime.now(tz=UTC)
279 if now - last_seen < timedelta(days=7):
280 return f"Last seen {last_seen:%A %H:%M %p GMT}", True
281 else:
282 return f"Last seen {last_seen:%b %-d %Y}", False