Coverage for slidge/core/mixins/message_text.py: 88%
85 statements
« prev ^ index » next coverage.py v7.15.2, created at 2026-09-10 04:45 +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 if legacy_msg_id is not None:
120 self._store_sent_msg(legacy_msg_id, when)
121 return self._send(
122 msg,
123 archive_only=archive_only,
124 carbon=carbon,
125 legacy_msg_id=legacy_msg_id,
126 **send_kwargs,
127 )
129 def _store_sent_msg(self, legacy_id: str, when: datetime | None) -> None:
130 pass
132 def correct(
133 self,
134 legacy_msg_id: str,
135 new_text: str,
136 *,
137 when: datetime | None = None,
138 reply_to: MessageReference | None = None,
139 thread: str | None = None,
140 hints: Iterable[ProcessingHint] | None = None,
141 carbon: bool = False,
142 archive_only: bool = False,
143 correction_event_id: str | None = None,
144 link_previews: list[LinkPreview] | None = None,
145 **send_kwargs: object,
146 ) -> None:
147 """
148 Modify a message that was previously sent by this :term:`XMPP Entity`.
150 Uses last message correction (:xep:`0308`)
152 :param new_text: New content of the message
153 :param legacy_msg_id: The legacy message ID of the message to correct
154 :param when: when the message was sent, for a "delay" tag (:xep:`0203`)
155 :param reply_to: Quote another message (:xep:`0461`)
156 :param hints:
157 :param thread:
158 :param carbon: (only in 1:1) Reflect a message sent to this ``Contact`` by the user.
159 Use this to synchronize outgoing history for legacy official apps.
160 :param archive_only: (only in groups) Do not send this message to user,
161 but store it in the archive. Meant to be used during ``MUC.backfill()``
162 :param correction_event_id: in the case where an ID is associated with the legacy
163 'correction event', specify it here to use it on the XMPP side. If not specified,
164 a random ID will be used.
165 :param link_previews: A little of sender (or server, or gateway)-generated
166 previews of URLs linked in the body.
167 """
168 self.send_text(
169 new_text,
170 legacy_msg_id,
171 when=when,
172 reply_to=reply_to,
173 hints=hints,
174 carbon=carbon,
175 thread=thread,
176 correction=True,
177 archive_only=archive_only,
178 correction_event_id=correction_event_id,
179 link_previews=link_previews,
180 **send_kwargs,
181 )
183 def react(
184 self,
185 legacy_msg_id: str,
186 emojis: Iterable[str] = (),
187 thread: str | None = None,
188 **kwargs: object,
189 ) -> None:
190 """
191 Send a reaction (:xep:`0444`) from this :term:`XMPP Entity`.
193 :param legacy_msg_id: The message which the reaction refers to.
194 :param emojis: An iterable of emojis used as reactions
195 :param thread:
196 """
197 xmpp_id = kwargs.pop("xmpp_id", None)
198 emojis = {remove_emoji_variation_selector_16(e) for e in emojis}
199 if xmpp_id:
200 assert isinstance(xmpp_id, str)
201 xmpp_ids = [xmpp_id]
202 else:
203 xmpp_ids = self._legacy_to_xmpp(legacy_msg_id)
204 for xmpp_id in xmpp_ids:
205 msg = self._make_message(
206 hints={"store"}, carbon=bool(kwargs.get("carbon")), thread=thread
207 )
208 self.xmpp.plugin["xep_0444"].set_reactions(
209 msg, to_id=xmpp_id, reactions=emojis
210 )
211 self.__add_reaction_fallback(msg, legacy_msg_id, emojis)
212 self._send(msg, **kwargs)
214 def __add_reaction_fallback(
215 self,
216 msg: Message,
217 legacy_msg_id: str,
218 emojis: Iterable[str] = (),
219 ) -> None:
220 session = self.session
221 if session is None or not session.user.preferences.get(
222 "reaction_fallback", False
223 ):
224 return
225 msg["fallback"]["for"] = self.xmpp.plugin["xep_0444"].namespace
226 msg["fallback"].enable("body")
227 msg["body"] = " ".join(emojis)
228 if not self.is_participant:
229 return
230 with self.xmpp.store.session() as orm:
231 archived = self.xmpp.store.mam.get_by_legacy_id(
232 orm, self.muc.stored.id, str(legacy_msg_id)
233 )
234 if archived is None:
235 return
236 history_msg = HistoryMessage(archived.stanza)
237 msg["body"] = (
238 add_quote_prefix(history_msg.stanza["body"]) + "\n" + msg["body"]
239 )
241 def retract(
242 self,
243 legacy_msg_id: str,
244 thread: str | None = None,
245 **kwargs: object,
246 ) -> None:
247 """
248 Send a message retraction (:XEP:`0424`) from this :term:`XMPP Entity`.
250 :param legacy_msg_id: Legacy ID of the message to delete
251 :param thread:
252 """
253 xmpp_ids = self._legacy_to_xmpp(legacy_msg_id)
254 replace_id = self._replace_id(legacy_msg_id)
255 if replace_id not in xmpp_ids:
256 xmpp_ids.append(replace_id)
257 for xmpp_id in xmpp_ids:
258 msg = self._make_message(
259 state=None,
260 hints={"store"},
261 mbody=f"/me retracted the message {legacy_msg_id}",
262 carbon=bool(kwargs.get("carbon")),
263 thread=thread,
264 )
265 msg.enable("fallback")
266 # namespace version mismatch between slidge and slixmpp, update me later
267 msg["fallback"]["for"] = self.xmpp.plugin["xep_0424"].namespace[:-1] + "1"
268 msg["retract"]["id"] = msg["replace"]["id"] = xmpp_id
269 self._send(msg, **kwargs)
272log = logging.getLogger(__name__)