Coverage for slidge/core/mixins/presence.py: 91%
130 statements
« prev ^ index » next coverage.py v7.15.2, created at 2026-07-28 18:29 +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
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):
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 )
163 _UPDATE_LAST_SEEN_FALLBACK_TASKS[pk] = task
164 task.add_done_callback(partial(_clear_last_seen_task, pk))
165 p["idle"]["since"] = last_seen
166 return p
168 def send_last_presence(
169 self, force: bool = False, no_cache_online: bool = False
170 ) -> None:
171 if (cache := self._get_last_presence()) is None:
172 if force:
173 if no_cache_online:
174 self.online()
175 else:
176 self.offline()
177 return
178 self._send(
179 self._make_presence(
180 last_seen=cache.last_seen,
181 force=True,
182 ptype=cache.ptype,
183 pshow=cache.pshow,
184 pstatus=cache.pstatus,
185 )
186 )
188 def online(
189 self,
190 status: str | None = None,
191 last_seen: datetime | None = None,
192 ) -> None:
193 """
194 Send an "online" presence from this contact to the user.
196 :param status: Arbitrary text, details of the status, eg: "Listening to Britney Spears"
197 :param last_seen: For :xep:`0319`
198 """
199 with contextlib.suppress(_NoChange):
200 self._send(self._make_presence(pstatus=status, last_seen=last_seen))
202 def away(
203 self,
204 status: str | None = None,
205 last_seen: datetime | None = None,
206 ) -> None:
207 """
208 Send an "away" presence from this contact to the user.
210 This is a global status, as opposed to :meth:`.LegacyContact.inactive`
211 which concerns a specific conversation, ie a specific "chat window"
213 :param status: Arbitrary text, details of the status, eg: "Gone to fight capitalism"
214 :param last_seen: For :xep:`0319`
215 """
216 with contextlib.suppress(_NoChange):
217 self._send(
218 self._make_presence(pstatus=status, pshow="away", last_seen=last_seen)
219 )
221 def extended_away(
222 self,
223 status: str | None = None,
224 last_seen: datetime | None = None,
225 ) -> None:
226 """
227 Send an "extended away" presence from this contact to the user.
229 This is a global status, as opposed to :meth:`.LegacyContact.inactive`
230 which concerns a specific conversation, ie a specific "chat window"
232 :param status: Arbitrary text, details of the status, eg: "Gone to fight capitalism"
233 :param last_seen: For :xep:`0319`
234 """
235 with contextlib.suppress(_NoChange):
236 self._send(
237 self._make_presence(pstatus=status, pshow="xa", last_seen=last_seen)
238 )
240 def busy(
241 self,
242 status: str | None = None,
243 last_seen: datetime | None = None,
244 ) -> None:
245 """
246 Send a "busy" (ie, "dnd") presence from this contact to the user,
248 :param status: eg: "Trying to make sense of XEP-0100"
249 :param last_seen: For :xep:`0319`
250 """
251 with contextlib.suppress(_NoChange):
252 self._send(
253 self._make_presence(pstatus=status, pshow="dnd", last_seen=last_seen)
254 )
256 def offline(
257 self,
258 status: str | None = None,
259 last_seen: datetime | None = None,
260 ) -> None:
261 """
262 Send an "offline" presence from this contact to the user.
264 :param status: eg: "Trying to make sense of XEP-0100"
265 :param last_seen: For :xep:`0319`
266 """
267 with contextlib.suppress(_NoChange):
268 self._send(
269 self._make_presence(
270 pstatus=status, ptype="unavailable", last_seen=last_seen
271 )
272 )
275def get_last_seen_fallback(last_seen: datetime) -> tuple[str, bool]:
276 now = datetime.now(tz=UTC)
277 if now - last_seen < timedelta(days=7):
278 return f"Last seen {last_seen:%A %H:%M %p GMT}", True
279 else:
280 return f"Last seen {last_seen:%b %-d %Y}", False