Coverage for slidge/contact/contact.py: 87%
279 statements
« prev ^ index » next coverage.py v7.15.2, created at 2026-08-18 04:30 +0000
1import datetime
2import logging
3import warnings
4from collections.abc import Iterable, Iterator, Sequence
5from datetime import date
6from typing import TYPE_CHECKING, Any, ClassVar, Literal, Self
7from xml.etree import ElementTree as ET
9from slixmpp import JID, Message, Presence
10from slixmpp.exceptions import IqError, IqTimeout
11from slixmpp.plugins.xep_0292.stanza import VCard4
12from slixmpp.types import MessageTypes
14from slidge.db.avatar import CachedAvatar
16from ..core.mixins import AvatarMixin, FullCarbonMixin
17from ..core.mixins.disco import ContactAccountDiscoMixin
18from ..core.mixins.recipient import RecipientMixin
19from ..db.models import Contact, ContactSent
20from ..util.types import (
21 AnySession,
22 ClientType,
23 ContactMessage,
24 ContactSticker,
25 MessageOrPresenceTypeVar,
26)
28if TYPE_CHECKING:
29 from ..command.base import ContactCommand
30 from ..group.participant import LegacyParticipant
33class LegacyContact(
34 AvatarMixin,
35 ContactAccountDiscoMixin,
36 FullCarbonMixin,
37 RecipientMixin,
38):
39 """
40 This class centralizes actions in relation to a specific legacy contact.
42 You shouldn't create instances of contacts manually, but rather rely on
43 :meth:`.LegacyRoster.by_legacy_id` to ensure that contact instances are
44 singletons. The :class:`.LegacyRoster` instance of a session is accessible
45 through the :attr:`.BaseSession.contacts` attribute.
47 Typically, your plugin should have methods hook to the legacy events and
48 call appropriate methods here to transmit the "legacy action" to the xmpp
49 user. This should look like this:
51 .. code-block:python
53 class Session(BaseSession):
54 ...
56 async def on_cool_chat_network_new_text_message(self, legacy_msg_event):
57 contact = self.contacts.by_legacy_id(legacy_msg_event.from)
58 contact.send_text(legacy_msg_event.text)
60 async def on_cool_chat_network_new_typing_event(self, legacy_typing_event):
61 contact = self.contacts.by_legacy_id(legacy_msg_event.from)
62 contact.composing()
63 ...
65 Use ``carbon=True`` as a keyword arg for methods to represent an action FROM
66 the user TO the contact, typically when the user uses an official client to
67 do an action such as sending a message or marking as message as read.
68 This will use :xep:`0363` to impersonate the XMPP user in order.
69 """
71 RESOURCE: str = "slidge"
72 """
73 A full JID, including a resource part is required for chat states (and maybe other stuff)
74 to work properly. This is the name of the resource the contacts will use.
75 """
76 PROPAGATE_PRESENCE_TO_GROUPS = True
78 mtype: MessageTypes = "chat"
79 _can_send_carbon = True
80 is_participant: Literal[False] = False
81 is_group: Literal[False] = False
83 _ONLY_SEND_PRESENCE_CHANGES = True
85 STRIP_SHORT_DELAY = True
86 _NON_FRIEND_PRESENCES_FILTER: ClassVar[set[str]] = {"subscribe", "unsubscribed"}
88 INVITATION_RECIPIENT = True
90 commands: ClassVar[dict[str, "type[ContactCommand[LegacyContact]]"]] = {}
91 commands_chat: ClassVar[dict[str, "type[ContactCommand[LegacyContact]]"]] = {}
93 stored: Contact
94 model: Contact
96 def __init__(self, session: AnySession, stored: Contact) -> None:
97 self.session = session
98 self.xmpp = session.xmpp
99 self.stored = stored
100 self._set_logger()
101 super().__init__()
103 def _recipient_pk(self) -> int:
104 return self.stored.id
106 async def on_message(self, message: ContactMessage) -> str | None:
107 """
108 Triggered when the user sends a message to this :term:`Contact`.
110 :return: A message ID of that can be used later to further reference
111 this message (reactions, read marks, etc.).
112 """
113 raise NotImplementedError
115 async def on_sticker(self, sticker: ContactSticker) -> str | None:
116 """
117 Triggered when the user sends a sticker to this :term:`Contact`.
119 :param sticker: The sticker sent by the user.
121 :return: A message ID of that can be used later to further reference
122 this message (reactions, read marks, etc.).
123 """
124 raise NotImplementedError
126 @property
127 def jid(self) -> JID:
128 jid = JID(self.stored.jid)
129 jid.resource = self.RESOURCE
130 return jid
132 @jid.setter
133 def jid(self, _jid: JID) -> None:
134 raise RuntimeError
136 @property
137 def legacy_id(self) -> str:
138 return self.stored.legacy_id
140 async def get_vcard(self, fetch: bool = True) -> VCard4 | None:
141 if fetch and not self.stored.vcard_fetched:
142 await self.fetch_vcard()
143 if self.stored.vcard is None:
144 return None
146 return VCard4(xml=ET.fromstring(self.stored.vcard))
148 @property
149 def is_friend(self) -> bool:
150 return self.stored.is_friend
152 @is_friend.setter
153 def is_friend(self, value: bool) -> None:
154 if value == self.is_friend:
155 return
156 self.update_stored_attribute(is_friend=value)
158 @property
159 def added_to_roster(self) -> bool:
160 return self.stored.added_to_roster
162 @added_to_roster.setter
163 def added_to_roster(self, value: bool) -> None:
164 if value == self.added_to_roster:
165 return
166 self.update_stored_attribute(added_to_roster=value)
168 @property
169 def participants(self) -> Iterator["LegacyParticipant[Self]"]:
170 with self.xmpp.store.session() as orm:
171 self.stored = orm.merge(self.stored)
172 participants = self.stored.participants
173 for p in participants:
174 with self.xmpp.store.session() as orm:
175 p = orm.merge(p)
176 muc = self.session.bookmarks.from_store(p.room)
177 part = muc.participant_from_store(p, contact=self)
178 yield part
180 @property # type:ignore
181 def DISCO_TYPE(self) -> ClientType:
182 return self.client_type
184 @DISCO_TYPE.setter
185 def DISCO_TYPE(self, value: ClientType) -> None:
186 self.client_type = value
188 @property
189 def client_type(self) -> ClientType:
190 """
191 The client type of this contact, cf https://xmpp.org/registrar/disco-categories.html#client
193 Default is "pc".
194 """
195 return self.stored.client_type
197 @client_type.setter
198 def client_type(self, value: ClientType) -> None:
199 if self.stored.client_type == value:
200 return
201 self.update_stored_attribute(client_type=value)
203 def _set_logger(self) -> None:
204 self.log = logging.getLogger(f"{self.user_jid.bare}:contact:{self}")
206 def __repr__(self) -> str:
207 return f"<Contact #{self.stored.id} '{self.name}' ({self.legacy_id} - {self.jid.user})'>"
209 def __get_subscription_string(self) -> str:
210 if self.is_friend:
211 return "both"
212 return "none"
214 def __propagate_to_participants(self, stanza: Presence) -> None:
215 if not self.PROPAGATE_PRESENCE_TO_GROUPS:
216 return
218 ptype = stanza["type"]
219 if ptype in ("available", "chat"):
220 func_name = "online"
221 elif ptype in ("xa", "unavailable"):
222 # we map unavailable to extended_away, because offline is
223 # "participant leaves the MUC"
224 # TODO: improve this with a clear distinction between participant
225 # and member list
226 func_name = "extended_away"
227 elif ptype == "busy":
228 func_name = "busy"
229 elif ptype == "away":
230 func_name = "away"
231 else:
232 return
234 last_seen: datetime.datetime | None = (
235 stanza["idle"]["since"] if "idle" in stanza else None
236 )
238 kw = {"status": stanza["status"], "last_seen": last_seen}
240 for part in self.participants:
241 func = getattr(part, func_name)
242 func(**kw)
244 def _send(
245 self,
246 stanza: MessageOrPresenceTypeVar,
247 carbon: bool = False,
248 nick: bool = False,
249 **send_kwargs: Any, # noqa:ANN401
250 ) -> MessageOrPresenceTypeVar:
251 if carbon and isinstance(stanza, Message):
252 stanza["to"] = self.jid.bare
253 stanza["from"] = self.user_jid
254 self._privileged_send(stanza)
255 return stanza
257 if isinstance(stanza, Presence):
258 if not self._updating_info:
259 self.__propagate_to_participants(stanza)
260 if (
261 not self.is_friend
262 and stanza["type"] not in self._NON_FRIEND_PRESENCES_FILTER
263 ):
264 return stanza
265 if self.name and (nick or not self.is_friend):
266 n = self.xmpp.plugin["xep_0172"].stanza.UserNick()
267 n["nick"] = self.name
268 stanza.append(n)
269 if (
270 not self._updating_info
271 and self.xmpp.MARK_ALL_MESSAGES
272 and is_markable(stanza)
273 ):
274 with self.xmpp.store.session(expire_on_commit=False) as orm:
275 self.stored = orm.merge(self.stored)
276 exists = (
277 orm.query(ContactSent)
278 .filter_by(contact_id=self.stored.id, msg_id=stanza["id"])
279 .first()
280 )
281 if exists:
282 self.log.warning(
283 "Contact has already sent message %s", stanza["id"]
284 )
285 else:
286 new = ContactSent(contact=self.stored, msg_id=stanza["id"])
287 orm.add(new)
288 self.stored.sent_order.append(new)
289 orm.commit()
290 stanza["to"] = self.user_jid
291 stanza.send()
292 return stanza
294 def pop_unread_xmpp_ids_up_to(self, horizon_xmpp_id: str) -> list[str]:
295 """
296 Return XMPP msg ids sent by this contact up to a given XMPP msg id.
298 Legacy modules have no reason to use this, but it is used by slidge core
299 for legacy networks that need to mark all messages as read (most XMPP
300 clients only send a read marker for the latest message).
302 This has side effects, if the horizon XMPP id is found, messages up to
303 this horizon are cleared, to avoid sending the same read mark twice.
305 :param horizon_xmpp_id: The latest message
306 :return: A list of XMPP ids up to horizon_xmpp_id, included
307 """
308 with self.xmpp.store.session() as orm:
309 assert self.stored.id is not None
310 ids = self.xmpp.store.contacts.pop_sent_up_to(
311 orm, self.stored.id, horizon_xmpp_id
312 )
313 orm.commit()
314 return ids
316 @property
317 def name(self) -> str:
318 """
319 Friendly name of the contact, as it should appear in the user's roster
320 """
321 return self.stored.nick or ""
323 @name.setter
324 def name(self, n: str | None) -> None:
325 if self.stored.nick == n:
326 return
327 self.update_stored_attribute(nick=n)
328 self._set_logger()
329 if self.is_friend and self.added_to_roster:
330 self.xmpp.pubsub.broadcast_nick(
331 user_jid=self.user_jid, jid=self.jid.bare, nick=n
332 )
333 for p in self.participants:
334 p.nickname = n or str(self.legacy_id)
336 def _post_avatar_update(self, cached_avatar: CachedAvatar | None) -> None:
337 if self.is_friend and self.added_to_roster:
338 self.session.create_task(
339 self.session.xmpp.pubsub.broadcast_avatar(
340 self.jid.bare, self.session.user_jid, cached_avatar
341 )
342 )
343 for p in self.participants:
344 self.log.debug("Propagating new avatar to %s", p.muc)
345 p.send_last_presence(force=True, no_cache_online=True)
347 def set_vcard(
348 self,
349 /,
350 full_name: str | None = None,
351 given: str | None = None,
352 surname: str | None = None,
353 birthday: date | None = None,
354 phone: str | None = None,
355 phones: Iterable[str] = (),
356 note: str | None = None,
357 url: str | None = None,
358 email: str | None = None,
359 country: str | None = None,
360 locality: str | None = None,
361 pronouns: str | None = None,
362 ) -> None:
363 """
364 Update xep:`0292` data for this contact.
366 Use this for additional metadata about this contact to be available to XMPP
367 clients. The "note" argument is a text of arbitrary size and can be useful when
368 no other field is a good fit.
369 """
370 vcard = VCard4()
371 vcard.add_impp(f"xmpp:{self.jid.bare}")
373 if n := self.name:
374 vcard.add_nickname(n)
375 if full_name:
376 vcard["full_name"] = full_name
377 elif n:
378 vcard["full_name"] = n
380 if given:
381 vcard["given"] = given
382 if surname:
383 vcard["surname"] = surname
384 if birthday:
385 vcard["birthday"] = birthday
387 if note:
388 vcard.add_note(note)
389 if url:
390 vcard.add_url(url)
391 if email:
392 vcard.add_email(email)
393 if phone:
394 vcard.add_tel(phone)
395 for p in phones:
396 vcard.add_tel(p)
397 if (country and locality) or country:
398 vcard.add_address(country, locality)
399 if pronouns:
400 vcard["pronouns"]["text"] = pronouns
402 self.update_stored_attribute(vcard=str(vcard), vcard_fetched=True)
403 self.session.create_task(
404 self.xmpp.pubsub.broadcast_vcard_event(self.jid, self.user_jid, vcard)
405 )
407 def get_roster_item(self) -> dict[str, dict[str, str | Sequence[str]]]:
408 item = {
409 "subscription": self.__get_subscription_string(),
410 "groups": [self.xmpp.ROSTER_GROUP],
411 }
412 if (n := self.name) is not None:
413 item["name"] = n
414 return {self.jid.bare: item}
416 async def add_to_roster(self, force: bool = False) -> None:
417 """
418 Add this contact to the user roster using :xep:`0356`
420 :param force: add even if the contact was already added successfully
421 """
422 if self.added_to_roster and not force:
423 return
424 if not self.session.user.preferences.get("roster_push", True):
425 log.debug("Roster push request by plugin ignored (--no-roster-push)")
426 return
427 try:
428 await self.xmpp.plugin["xep_0356"].set_roster(
429 jid=self.user_jid, roster_items=self.get_roster_item()
430 )
431 except PermissionError:
432 warnings.warn(
433 f"Slidge does not have the privilege (XEP-0356) to manage the roster of {self.user_jid}. "
434 "If this is a local user, consider configuring your XMPP server for that."
435 )
436 self.send_friend_request(
437 f"I'm already your friend on {self.xmpp.COMPONENT_TYPE}, but "
438 "slidge is not allowed to manage your roster."
439 )
440 return
441 except (IqError, IqTimeout) as e:
442 self.log.warning("Could not add to roster", exc_info=e)
443 else:
444 # we only broadcast pubsub events for contacts added to the roster
445 # so if something was set before, we need to push it now
446 self.added_to_roster = True
447 self.send_last_presence(force=True)
449 async def __broadcast_pubsub_items(self) -> None:
450 if not self.is_friend:
451 return
452 if not self.added_to_roster:
453 return
454 cached_avatar = self.get_cached_avatar()
455 if cached_avatar is not None:
456 await self.xmpp.pubsub.broadcast_avatar(
457 self.jid.bare, self.session.user_jid, cached_avatar
458 )
459 nick = self.name
461 if nick is not None:
462 self.xmpp.pubsub.broadcast_nick(
463 self.session.user_jid,
464 self.jid.bare,
465 nick,
466 )
468 def send_friend_request(self, text: str | None = None) -> None:
469 presence = self._make_presence(ptype="subscribe", pstatus=text, bare=True)
470 self._send(presence, nick=True)
472 async def accept_friend_request(self, text: str | None = None) -> None:
473 """
474 Call this to signify that this Contact has accepted to be a friend
475 of the user.
477 :param text: Optional message from the friend to the user
478 """
479 self.is_friend = True
480 self.added_to_roster = True
481 self.log.debug("Accepting friend request")
482 presence = self._make_presence(ptype="subscribed", pstatus=text, bare=True)
483 self._send(presence, nick=True)
484 self.send_last_presence()
485 await self.__broadcast_pubsub_items()
486 self.log.debug("Accepted friend request")
488 def reject_friend_request(self, text: str | None = None) -> None:
489 """
490 Call this to signify that this Contact has refused to be a contact
491 of the user (or that they don't want to be friends anymore)
493 :param text: Optional message from the non-friend to the user
494 """
495 presence = self._make_presence(ptype="unsubscribed", pstatus=text, bare=True)
496 self.offline()
497 self._send(presence, nick=True)
498 self.is_friend = False
500 async def on_friend_request(self, text: str = "") -> None:
501 """
502 Called when receiving a "subscribe" presence, ie, "I would like to add
503 you to my contacts/friends", from the user to this contact.
505 In XMPP terms: "I would like to receive your presence updates"
507 This is only called if self.is_friend = False. If self.is_friend = True,
508 slidge will automatically "accept the friend request", ie, reply with
509 a "subscribed" presence.
511 When called, a 'friend request event' should be sent to the legacy
512 service, and when the contact responds, you should either call
513 self.accept_subscription() or self.reject_subscription()
514 """
516 async def on_friend_delete(self, text: str = "") -> None:
517 """
518 Called when receiving an "unsubscribed" presence, ie, "I would like to
519 remove you to my contacts/friends" or "I refuse your friend request"
520 from the user to this contact.
522 In XMPP terms: "You won't receive my presence updates anymore (or you
523 never have)".
524 """
526 async def on_friend_accept(self) -> None:
527 """
528 Called when receiving a "subscribed" presence, ie, "I accept to be
529 your/confirm that you are my friend" from the user to this contact.
531 In XMPP terms: "You will receive my presence updates".
532 """
534 def unsubscribe(self) -> None:
535 """
536 (internal use by slidge)
538 Send an "unsubscribe", "unsubscribed", "unavailable" presence sequence
539 from this contact to the user, ie, "this contact has removed you from
540 their 'friends'".
541 """
542 for ptype in "unsubscribe", "unsubscribed", "unavailable":
543 self.xmpp.send_presence(pfrom=self.jid, pto=self.user_jid.bare, ptype=ptype)
545 async def update_info(self) -> None:
546 """
547 Fetch information about this contact from the legacy network
549 This is awaited on Contact instantiation, and should be overridden to
550 update the nickname, avatar, vcard [...] of this contact, by making
551 "legacy API calls".
553 To take advantage of the slidge avatar cache, you can check the .avatar
554 property to retrieve the "legacy file ID" of the cached avatar. If there
555 is no change, you should not call
556 :py:meth:`slidge.core.mixins.avatar.AvatarMixin.set_avatar` or attempt
557 to modify the ``.avatar`` property.
558 """
560 async def fetch_vcard(self) -> None:
561 """
562 It the legacy network doesn't like that you fetch too many profiles on startup,
563 it's also possible to fetch it here, which will be called when XMPP clients
564 of the user request the vcard, if it hasn't been fetched before
565 :return:
566 """
568 def _make_presence(
569 self,
570 *,
571 last_seen: datetime.datetime | None = None,
572 status_codes: set[int] | None = None,
573 user_full_jid: JID | None = None,
574 **presence_kwargs: Any, # noqa:ANN401
575 ) -> Presence:
576 p = super()._make_presence(last_seen=last_seen, **presence_kwargs)
577 caps = self.xmpp.plugin["xep_0115"]
578 if p.get_from().resource and self.stored.caps_ver:
579 p["caps"]["node"] = caps.caps_node
580 p["caps"]["hash"] = caps.hash
581 p["caps"]["ver"] = self.stored.caps_ver
582 return p
585def is_markable(stanza: Message | Presence) -> bool:
586 if isinstance(stanza, Presence):
587 return False
588 return bool(stanza["body"])
591log = logging.getLogger(__name__)