Coverage for slidge/core/mixins/presence.py: 91%
128 statements
« prev ^ index » next coverage.py v7.15.2, created at 2026-09-10 04:45 +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
48 def __init__(self, *a: object, **k: object) -> None:
49 super().__init__(*a, **k)
50 # this is only used when a presence is set during Contact.update_info(),
51 # when the contact does not have a DB primary key yet, and is written
52 # to DB at the end of update_info()
53 self.cached_presence: CachedPresence | None = None
55 def __is_contact(self) -> bool:
56 return isinstance(self.stored, Contact)
58 def __stored(self) -> Contact | None:
59 if self.__is_contact():
60 assert isinstance(self.stored, Contact)
61 return self.stored
62 else:
63 assert isinstance(self.stored, Participant)
64 try:
65 return self.stored.contact
66 except DetachedInstanceError:
67 with self.xmpp.store.session() as orm:
68 orm.add(self.stored)
69 if self.stored.contact is None:
70 return None
71 orm.refresh(self.stored.contact)
72 orm.merge(self.stored)
73 return self.stored.contact
75 @property
76 def __contact_pk(self) -> int | None:
77 stored = self.__stored()
78 return None if stored is None else stored.id
80 def _get_last_presence(self) -> CachedPresence | None:
81 stored = self.__stored()
82 if stored is None or not stored.cached_presence:
83 return None
84 return CachedPresence(
85 None if stored.last_seen is None else stored.last_seen.replace(tzinfo=UTC),
86 stored.ptype, # type:ignore
87 stored.pstatus,
88 stored.pshow, # type:ignore
89 )
91 def _store_last_presence(self, new: CachedPresence) -> None:
92 if self.__is_contact():
93 contact = self
94 elif (contact := getattr(self, "contact", None)) is None: # type:ignore[assignment]
95 return
96 contact.update_stored_attribute( # type:ignore[attr-defined]
97 cached_presence=True,
98 **new._asdict(),
99 )
101 def _make_presence(
102 self,
103 *,
104 last_seen: datetime | None = None,
105 force: bool = False,
106 bare: bool = False,
107 ptype: PresenceTypes | None = None,
108 pstatus: str | None = None,
109 pshow: PresenceShows | None = None,
110 ) -> Presence:
111 if last_seen and last_seen.tzinfo is None:
112 last_seen = last_seen.astimezone(UTC)
114 old = self._get_last_presence()
116 if ptype not in _FRIEND_REQUEST_PRESENCES:
117 new = CachedPresence(
118 last_seen=last_seen, ptype=ptype, pstatus=pstatus, pshow=pshow
119 )
120 if old != new:
121 if hasattr(self, "muc") and ptype == "unavailable":
122 stored = self.__stored()
123 if stored is not None:
124 stored.cached_presence = False
125 self.commit()
126 else:
127 self._store_last_presence(new)
128 if old and not force and self._ONLY_SEND_PRESENCE_CHANGES:
129 if old == new:
130 self.session.log.debug("Presence is the same as cached")
131 raise _NoChange
132 self.session.log.debug(
133 "Presence is not the same as cached: %s vs %s", old, new
134 )
136 p = self.xmpp.make_presence(
137 pfrom=self.jid.bare if bare else self.jid,
138 ptype=ptype,
139 pshow=pshow,
140 pstatus=pstatus,
141 )
142 if last_seen:
143 # it's ugly to check for the presence of this string, but a better fix is more work
144 if not re.match(
145 ".*Last seen .*", p["status"]
146 ) and self.session.user.preferences.get("last_seen_fallback", True):
147 last_seen_fallback, recent = get_last_seen_fallback(last_seen)
148 if p["status"]:
149 p["status"] = p["status"] + " -- " + last_seen_fallback
150 else:
151 p["status"] = last_seen_fallback
152 pk = self.__contact_pk
153 if recent and pk is not None:
154 # if less than a week, we use sth like 'Last seen: Monday, 8:05",
155 # but if lasts more than a week, this is not very informative, so
156 # we need to force resend an updated presence status
157 task = _UPDATE_LAST_SEEN_FALLBACK_TASKS.get(pk)
158 if task is not None:
159 task.cancel()
160 task = self.session.create_task(
161 _update_last_seen_fallback(self.session, pk),
162 name=f"update last seen fallback of {self}",
163 )
164 _UPDATE_LAST_SEEN_FALLBACK_TASKS[pk] = task
165 task.add_done_callback(partial(_clear_last_seen_task, pk))
166 p["idle"]["since"] = last_seen
167 return p
169 def send_last_presence(
170 self, force: bool = False, no_cache_online: bool = False
171 ) -> None:
172 if (cache := self._get_last_presence()) is None:
173 if force:
174 if no_cache_online:
175 self.online()
176 else:
177 self.offline()
178 return
179 self._send(
180 self._make_presence(
181 last_seen=cache.last_seen,
182 force=True,
183 ptype=cache.ptype,
184 pshow=cache.pshow,
185 pstatus=cache.pstatus,
186 )
187 )
189 def online(
190 self,
191 status: str | None = None,
192 last_seen: datetime | None = None,
193 ) -> None:
194 """
195 Send an "online" presence from this contact to the user.
197 :param status: Arbitrary text, details of the status, eg: "Listening to Britney Spears"
198 :param last_seen: For :xep:`0319`
199 """
200 with contextlib.suppress(_NoChange):
201 self._send(self._make_presence(pstatus=status, last_seen=last_seen))
203 def away(
204 self,
205 status: str | None = None,
206 last_seen: datetime | None = None,
207 ) -> None:
208 """
209 Send an "away" presence from this contact to the user.
211 This is a global status, as opposed to :meth:`.LegacyContact.inactive`
212 which concerns a specific conversation, ie a specific "chat window"
214 :param status: Arbitrary text, details of the status, eg: "Gone to fight capitalism"
215 :param last_seen: For :xep:`0319`
216 """
217 with contextlib.suppress(_NoChange):
218 self._send(
219 self._make_presence(pstatus=status, pshow="away", last_seen=last_seen)
220 )
222 def extended_away(
223 self,
224 status: str | None = None,
225 last_seen: datetime | None = None,
226 ) -> None:
227 """
228 Send an "extended away" presence from this contact to the user.
230 This is a global status, as opposed to :meth:`.LegacyContact.inactive`
231 which concerns a specific conversation, ie a specific "chat window"
233 :param status: Arbitrary text, details of the status, eg: "Gone to fight capitalism"
234 :param last_seen: For :xep:`0319`
235 """
236 with contextlib.suppress(_NoChange):
237 self._send(
238 self._make_presence(pstatus=status, pshow="xa", last_seen=last_seen)
239 )
241 def busy(
242 self,
243 status: str | None = None,
244 last_seen: datetime | None = None,
245 ) -> None:
246 """
247 Send a "busy" (ie, "dnd") presence from this contact to the user,
249 :param status: eg: "Trying to make sense of XEP-0100"
250 :param last_seen: For :xep:`0319`
251 """
252 with contextlib.suppress(_NoChange):
253 self._send(
254 self._make_presence(pstatus=status, pshow="dnd", last_seen=last_seen)
255 )
257 def offline(
258 self,
259 status: str | None = None,
260 last_seen: datetime | None = None,
261 ) -> None:
262 """
263 Send an "offline" presence from this contact to the user.
265 :param status: eg: "Trying to make sense of XEP-0100"
266 :param last_seen: For :xep:`0319`
267 """
268 with contextlib.suppress(_NoChange):
269 self._send(
270 self._make_presence(
271 pstatus=status, ptype="unavailable", last_seen=last_seen
272 )
273 )
276def get_last_seen_fallback(last_seen: datetime) -> tuple[str, bool]:
277 now = datetime.now(tz=UTC)
278 if now - last_seen < timedelta(days=7):
279 return f"Last seen {last_seen:%A %H:%M %p GMT}", True
280 else:
281 return f"Last seen {last_seen:%b %-d %Y}", False