Coverage for slidge/core/mixins/message.py: 83%

1import logging 

2import uuid 

3import warnings 

4 

5from slixmpp import JID, Message 

6 

7from ...util.types import AnyRoom, AnySession, ChatState, Marker 

8from .attachment import AttachmentMixin 

9from .base import SessionBound 

10from .message_maker import MessageMaker 

11from .message_text import TextMessageMixin 

12 

13 

14class ChatStateMixin(MessageMaker): 

15 def __init__(self) -> None: 

16 super().__init__() 

17 self.__last_chat_state: ChatState | None = None 

18 

19 def _chat_state( 

20 self, state: ChatState, forced: bool = False, **kwargs: object 

21 ) -> None: 

22 carbon = kwargs.get("carbon", False) 

23 if carbon or (state == self.__last_chat_state and not forced): 

24 return 

25 self.__last_chat_state = state 

26 msg = self._make_message(state=state, hints={"no-store"}) 

27 self._send(msg, **kwargs) 

28 

29 def active(self, **kwargs: object) -> None: 

30 """ 

31 Send an "active" chat state (:xep:`0085`) from this 

32 :term:`XMPP Entity`. 

33 """ 

34 self._chat_state("active", forced=False, **kwargs) 

35 

36 def composing(self, **kwargs: object) -> None: 

37 """ 

38 Send a "composing" (ie "typing notification") chat state (:xep:`0085`) 

39 from this :term:`XMPP Entity`. 

40 """ 

41 self._chat_state("composing", forced=True, **kwargs) 

42 

43 def paused(self, **kwargs: object) -> None: 

44 """ 

45 Send a "paused" (ie "typing paused notification") chat state 

46 (:xep:`0085`) from this :term:`XMPP Entity`. 

47 """ 

48 self._chat_state("paused", forced=False, **kwargs) 

49 

50 def inactive(self, **kwargs: object) -> None: 

51 """ 

52 Send an "inactive" (ie "contact has not interacted with the chat session 

53 interface for an intermediate period of time") chat state (:xep:`0085`) 

54 from this :term:`XMPP Entity`. 

55 """ 

56 self._chat_state("inactive", forced=False, **kwargs) 

57 

58 def gone(self, **kwargs: object) -> None: 

59 """ 

60 Send a "gone" (ie "contact has not interacted with the chat session interface, 

61 system, or device for a relatively long period of time") chat state 

62 (:xep:`0085`) from this :term:`XMPP Entity`. 

63 """ 

64 self._chat_state("gone", forced=False, **kwargs) 

65 

66 

67class MarkerMixin(MessageMaker, SessionBound): 

68 def _make_marker( 

69 self, legacy_msg_id: str, marker: Marker, carbon: bool = False 

70 ) -> Message: 

71 msg = self._make_message(carbon=carbon) 

72 msg[marker]["id"] = self._legacy_to_xmpp(legacy_msg_id)[-1] 

73 return msg 

74 

75 def ack(self, legacy_msg_id: str, **kwargs: object) -> None: 

76 """ 

77 Send an "acknowledged" message marker (:xep:`0333`) from this :term:`XMPP Entity`. 

78 

79 :param legacy_msg_id: The message this marker refers to 

80 """ 

81 self._send( 

82 self._make_marker( 

83 legacy_msg_id, "acknowledged", carbon=bool(kwargs.get("carbon")) 

84 ), 

85 **kwargs, 

86 ) 

87 

88 def received(self, legacy_msg_id: str, **kwargs: object) -> None: 

89 """ 

90 Send a "received" message marker (:xep:`0333`) from this :term:`XMPP Entity`. 

91 If called on a :class:`LegacyContact`, also send a delivery receipt 

92 marker (:xep:`0184`). 

93 

94 :param legacy_msg_id: The message this marker refers to 

95 """ 

96 carbon = bool(kwargs.get("carbon")) 

97 if self._mtype == "chat": 

98 for msg_id in self._legacy_to_xmpp(legacy_msg_id): 

99 self._send( 

100 self.xmpp.delivery_receipt.make_ack( 

101 msg_id, 

102 mfrom=self.jid, 

103 mto=self.user_jid, 

104 ) 

105 ) 

106 self._send( 

107 self._make_marker(legacy_msg_id, "received", carbon=carbon), **kwargs 

108 ) 

109 

110 def displayed(self, legacy_msg_id: str, **kwargs: object) -> None: 

111 """ 

112 Send a "displayed" message marker (:xep:`0333`) from this :term:`XMPP Entity`. 

113 

114 :param legacy_msg_id: The message this marker refers to 

115 """ 

116 self._send( 

117 self._make_marker( 

118 legacy_msg_id, "displayed", carbon=bool(kwargs.get("carbon")) 

119 ), 

120 **kwargs, 

121 ) 

122 

123 

124class ContentMessageMixin(AttachmentMixin, TextMessageMixin): 

125 pass 

126 

127 

128class CarbonMessageMixin(ContentMessageMixin, MarkerMixin): 

129 def _privileged_send(self, msg: Message) -> None: 

130 i = msg.get_id() 

131 if i: 

132 self._session.ignore_messages.add(i) 

133 else: 

134 i = "slidge-carbon-" + str(uuid.uuid4()) 

135 msg.set_id(i) 

136 msg.del_origin_id() 

137 try: 

138 self.xmpp.plugin["xep_0356"].send_privileged_message(msg) 

139 except PermissionError: 

140 warnings.warn( 

141 f"Slidge does not have the privilege (XEP-0356) to send messages on behalf of {self.user_jid}. " 

142 "If this is a local user, consider configuring your XMPP server for that." 

143 ) 

144 

145 

146class InviteMixin(MessageMaker): 

147 def invite_to( 

148 self, 

149 muc: AnyRoom | JID | str, 

150 reason: str | None = None, 

151 password: str | None = None, 

152 *, 

153 session: AnySession | None = None, 

154 **send_kwargs: object, 

155 ) -> None: 

156 """ 

157 Send an invitation to join a group (:xep:`0249`) from this :term:`XMPP Entity`. 

158 

159 :param muc: the muc the user is invited to 

160 :param reason: a text explaining why the user should join this muc 

161 :param password: maybe this will make sense later? not sure 

162 :param session: session for the user being invited. Defaults to 

163 ``self._session`` and MUST be provided if that is ``None``. 

164 :param send_kwargs: additional kwargs to be passed to _send() 

165 (internal use by slidge) 

166 """ 

167 msg = self._make_message(mtype="normal") 

168 jid = muc if isinstance(muc, str | JID) else muc.jid 

169 msg["groupchat_invite"]["jid"] = jid 

170 if reason: 

171 msg["groupchat_invite"]["reason"] = reason 

172 if password: 

173 msg["groupchat_invite"]["password"] = password 

174 if session is None: 

175 session = self._session 

176 if session is None: 

177 raise ValueError( 

178 "invite_to() requires an explicit session when self._session == None" 

179 ) 

180 session.create_task( 

181 self.__invite_when_ready(msg, session, **send_kwargs), 

182 name=f"Invite from {self} to {jid}", 

183 ) 

184 

185 async def __invite_when_ready( 

186 self, 

187 msg: Message, 

188 session: AnySession, 

189 **send_kwargs: object, 

190 ) -> None: 

191 await session.rooms.ready 

192 self._send(msg, **send_kwargs) 

193 

194 

195class MessageMixin(InviteMixin, ChatStateMixin, MarkerMixin, ContentMessageMixin): 

196 pass 

197 

198 

199class MessageCarbonMixin(InviteMixin, ChatStateMixin, CarbonMessageMixin): 

200 pass 

201 

202 

203log = logging.getLogger(__name__)