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