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