Coverage for slidge/core/mixins/message_text.py: 88%
83 statements
« prev ^ index » next coverage.py v7.15.2, created at 2026-09-29 05:05 +0000
1import logging
2from collections.abc import Iterable
3from datetime import datetime
5from slixmpp import Message
7from ...util.types import (
8 LinkPreview,
9 MessageReference,
10 ProcessingHint,
11)
12from ...util.util import add_quote_prefix, remove_emoji_variation_selector_16
13from .message_maker import MessageMaker
16class TextMessageMixin(MessageMaker):
17 def __default_hints(
18 self, hints: Iterable[ProcessingHint] | None = None
19 ) -> Iterable[ProcessingHint]:
20 if hints is not None:
21 return hints
22 elif self.mtype == "chat":
23 return {"markable", "store"}
24 elif self.mtype == "groupchat":
25 return {"markable"}
26 else:
27 raise RuntimeError("Never")
29 def _replace_id(self, legacy_msg_id: str) -> str:
30 if (
31 self.mtype == "groupchat"
32 and (recipient_pk := self._recipient_pk()) is not None
33 ):
34 with self.xmpp.store.session() as orm:
35 ids = self.xmpp.store.id_map.get_origin(
36 orm, recipient_pk, str(legacy_msg_id)
37 )
38 if ids:
39 if len(ids) > 1:
40 log.warning(
41 "More than 1 origin msg ID for '%s': '%s'",
42 legacy_msg_id,
43 ids,
44 )
45 return ids[0]
46 return legacy_msg_id
47 else:
48 return self._legacy_to_xmpp(legacy_msg_id)[0]
50 def send_text(
51 self,
52 body: str,
53 legacy_msg_id: str | None = None,
54 *,
55 when: datetime | None = None,
56 reply_to: MessageReference | None = None,
57 thread: str | None = None,
58 hints: Iterable[ProcessingHint] | None = None,
59 carbon: bool = False,
60 archive_only: bool = False,
61 correction: bool = False,
62 correction_event_id: str | None = None,
63 link_previews: list[LinkPreview] | None = None,
64 **send_kwargs: object,
65 ) -> Message | None:
66 """
67 Send a text message from this :term:`XMPP Entity`.
69 :param body: Content of the message
70 :param legacy_msg_id: If you want to be able to transport read markers from the gateway
71 user to the legacy network, specify this
72 :param when: when the message was sent, for a "delay" tag (:xep:`0203`)
73 :param reply_to: Quote another message (:xep:`0461`)
74 :param hints:
75 :param thread:
76 :param carbon: (only used if called on a :class:`LegacyContact`)
77 Set this to ``True`` if this is actually a message sent **to** the
78 :class:`LegacyContact` by the :term:`User`.
79 Use this to synchronize outgoing history for legacy official apps.
80 :param correction: whether this message is a correction or not
81 :param correction_event_id: in the case where an ID is associated with the legacy
82 'correction event', specify it here to use it on the XMPP side. If not specified,
83 a random ID will be used.
84 :param link_previews: A little of sender (or server, or gateway)-generated
85 previews of URLs linked in the body.
86 :param archive_only: (only in groups) Do not send this message to user,
87 but store it in the archive. Meant to be used during ``MUC.backfill()``
88 """
89 if (
90 carbon
91 and not self.is_participant
92 and (recipient_pk := self._recipient_pk()) is not None
93 ):
94 with self.xmpp.store.session() as orm:
95 if not correction and self.xmpp.store.id_map.was_sent_by_user(
96 orm, recipient_pk, str(legacy_msg_id)
97 ):
98 log.warning(
99 "Carbon message for a message an XMPP has sent? This is a bug! %s",
100 legacy_msg_id,
101 )
102 return None
103 hints = self.__default_hints(hints)
104 msg = self._make_message(
105 mbody=body,
106 legacy_msg_id=correction_event_id if correction else legacy_msg_id,
107 when=when,
108 reply_to=reply_to,
109 hints=hints or (),
110 carbon=carbon,
111 thread=thread,
112 link_previews=link_previews,
113 )
114 if correction:
115 if not legacy_msg_id:
116 raise TypeError
117 msg["replace"]["id"] = self._replace_id(legacy_msg_id)
118 if legacy_msg_id is not None:
119 self._store_sent_msg(legacy_msg_id, when)
120 return self._send(
121 msg,
122 archive_only=archive_only,
123 carbon=carbon,
124 legacy_msg_id=legacy_msg_id,
125 **send_kwargs,
126 )
128 def _store_sent_msg(self, legacy_id: str, when: datetime | None) -> None:
129 pass
131 def correct(
132 self,
133 legacy_msg_id: str,
134 new_text: str,
135 *,
136 when: datetime | None = None,
137 reply_to: MessageReference | None = None,
138 thread: str | None = None,
139 hints: Iterable[ProcessingHint] | None = None,
140 carbon: bool = False,
141 archive_only: bool = False,
142 correction_event_id: str | None = None,
143 link_previews: list[LinkPreview] | None = None,
144 **send_kwargs: object,
145 ) -> None:
146 """
147 Modify a message that was previously sent by this :term:`XMPP Entity`.
149 Uses last message correction (:xep:`0308`)
151 :param new_text: New content of the message
152 :param legacy_msg_id: The legacy message ID of the message to correct
153 :param when: when the message was sent, for a "delay" tag (:xep:`0203`)
154 :param reply_to: Quote another message (:xep:`0461`)
155 :param hints:
156 :param thread:
157 :param carbon: (only in 1:1) Reflect a message sent to this ``Contact`` by the user.
158 Use this to synchronize outgoing history for legacy official apps.
159 :param archive_only: (only in groups) Do not send this message to user,
160 but store it in the archive. Meant to be used during ``MUC.backfill()``
161 :param correction_event_id: in the case where an ID is associated with the legacy
162 'correction event', specify it here to use it on the XMPP side. If not specified,
163 a random ID will be used.
164 :param link_previews: A little of sender (or server, or gateway)-generated
165 previews of URLs linked in the body.
166 """
167 self.send_text(
168 new_text,
169 legacy_msg_id,
170 when=when,
171 reply_to=reply_to,
172 hints=hints,
173 carbon=carbon,
174 thread=thread,
175 correction=True,
176 archive_only=archive_only,
177 correction_event_id=correction_event_id,
178 link_previews=link_previews,
179 **send_kwargs,
180 )
182 def react(
183 self,
184 legacy_msg_id: str,
185 emojis: Iterable[str] = (),
186 thread: str | None = None,
187 **kwargs: object,
188 ) -> None:
189 """
190 Send a reaction (:xep:`0444`) from this :term:`XMPP Entity`.
192 :param legacy_msg_id: The message which the reaction refers to.
193 :param emojis: An iterable of emojis used as reactions
194 :param thread:
195 """
196 xmpp_id = kwargs.pop("xmpp_id", None)
197 emojis = {remove_emoji_variation_selector_16(e) for e in emojis}
198 if xmpp_id:
199 assert isinstance(xmpp_id, str)
200 xmpp_ids = [xmpp_id]
201 else:
202 xmpp_ids = self._legacy_to_xmpp(legacy_msg_id)
203 for xmpp_id in xmpp_ids:
204 msg = self._make_message(
205 hints={"store"}, carbon=bool(kwargs.get("carbon")), thread=thread
206 )
207 self.xmpp.plugin["xep_0444"].set_reactions(
208 msg, to_id=xmpp_id, reactions=emojis
209 )
210 self.__add_reaction_fallback(msg, legacy_msg_id, emojis)
211 self._send(msg, **kwargs)
213 def __add_reaction_fallback(
214 self,
215 msg: Message,
216 legacy_msg_id: str,
217 emojis: Iterable[str] = (),
218 ) -> None:
219 session = self.session
220 if session is None or not session.user.preferences.get(
221 "reaction_fallback", False
222 ):
223 return
224 msg["fallback"]["for"] = self.xmpp.plugin["xep_0444"].namespace
225 msg["fallback"].enable("body")
226 msg["body"] = " ".join(emojis)
227 if not self.is_participant:
228 return
229 with self.xmpp.store.session() as orm:
230 body = self.xmpp.store.mam.get_body(
231 orm, self.muc.stored.id, str(legacy_msg_id)
232 )
233 if body is None:
234 return
235 msg["body"] = add_quote_prefix(body) + "\n" + msg["body"]
237 def retract(
238 self,
239 legacy_msg_id: str,
240 thread: str | None = None,
241 **kwargs: object,
242 ) -> None:
243 """
244 Send a message retraction (:XEP:`0424`) from this :term:`XMPP Entity`.
246 :param legacy_msg_id: Legacy ID of the message to delete
247 :param thread:
248 """
249 xmpp_ids = self._legacy_to_xmpp(legacy_msg_id)
250 replace_id = self._replace_id(legacy_msg_id)
251 if replace_id not in xmpp_ids:
252 xmpp_ids.append(replace_id)
253 for xmpp_id in xmpp_ids:
254 msg = self._make_message(
255 state=None,
256 hints={"store"},
257 mbody=f"/me retracted the message {legacy_msg_id}",
258 carbon=bool(kwargs.get("carbon")),
259 thread=thread,
260 )
261 msg.enable("fallback")
262 # namespace version mismatch between slidge and slixmpp, update me later
263 msg["fallback"]["for"] = self.xmpp.plugin["xep_0424"].namespace[:-1] + "1"
264 msg["retract"]["id"] = msg["replace"]["id"] = xmpp_id
265 self._send(msg, **kwargs)
268log = logging.getLogger(__name__)