Mercurial > libervia-backend

comparison sat/plugins/plugin_xep_0372.py @ 3830:68a11b95a7d3

Find changesets by keywords (author, files, the commit message), revision number or hash, or revset expression.
plugin XEP-0372: References implementation: rel 369
author Goffi <goffi@goffi.org>
date Sun, 10 Jul 2022 15:16:05 +0200
parents
children 524856bd7b19
comparison
equal deleted inserted replaced
3829:67fc066ed2cd 3830:68a11b95a7d3
1 #!/usr/bin/env python3
2
3 # Libervia plugin for XEP-0372
4 # Copyright (C) 2009-2022 Jérôme Poisson (goffi@goffi.org)
5
6 # This program is free software: you can redistribute it and/or modify
7 # it under the terms of the GNU Affero General Public License as published by
8 # the Free Software Foundation, either version 3 of the License, or
9 # (at your option) any later version.
10
11 # This program is distributed in the hope that it will be useful,
12 # but WITHOUT ANY WARRANTY; without even the implied warranty of
13 # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
14 # GNU Affero General Public License for more details.
15
16 # You should have received a copy of the GNU Affero General Public License
17 # along with this program. If not, see <http://www.gnu.org/licenses/>.
18
19 from typing import Optional, Dict, Union
20 from textwrap import dedent
21 from sat.core import exceptions
22 from sat.tools.common import uri as xmpp_uri
23
24 from twisted.internet import defer
25 from twisted.words.protocols.jabber import jid
26 from twisted.words.protocols.jabber.xmlstream import XMPPHandler
27 from twisted.words.xish import domish
28 from zope.interface import implementer
29 from wokkel import disco, iwokkel
30
31 from sat.core.constants import Const as C
32 from sat.core.i18n import _
33 from sat.core.log import getLogger
34 from sat.core.core_types import SatXMPPEntity
35 from sat.tools.common import data_format
36
37
38 log = getLogger(__name__)
39
40 PLUGIN_INFO = {
41 C.PI_NAME: "References",
42 C.PI_IMPORT_NAME: "XEP-0372",
43 C.PI_TYPE: C.PLUG_TYPE_XEP,
44 C.PI_MODES: C.PLUG_MODE_BOTH,
45 C.PI_PROTOCOLS: ["XEP-0372"],
46 C.PI_DEPENDENCIES: ["XEP-0334"],
47 C.PI_MAIN: "XEP_0372",
48 C.PI_HANDLER: "yes",
49 C.PI_DESCRIPTION: _(dedent("""\
50 XEP-0372 (References) implementation
51
52 This plugin implement generic references and mentions.
53 """)),
54 }
55
56 NS_REFS = "urn:xmpp:reference:0"
57 ALLOWED_TYPES = ("mention", "data")
58
59
60 class XEP_0372:
61 namespace = NS_REFS
62
63 def __init__(self, host):
64 log.info(_("References plugin initialization"))
65 host.registerNamespace("refs", NS_REFS)
66 self.host = host
67 self._h = host.plugins["XEP-0334"]
68 host.trigger.add("messageReceived", self._messageReceivedTrigger)
69 host.bridge.addMethod(
70 "referenceSend",
71 ".plugin",
72 in_sign="sssss",
73 out_sign="",
74 method=self._sendReference,
75 async_=False,
76 )
77
78 def getHandler(self, client):
79 return XEP_0372_Handler()
80
81 def refElementToRefData(
82 self,
83 reference_elt: domish.Element
84 ) -> Dict[str, Union[str, int, dict]]:
85 ref_data: Dict[str, Union[str, int, dict]] = {
86 "uri": reference_elt["uri"],
87 "type": reference_elt["type"]
88 }
89
90 if ref_data["uri"].startswith("xmpp:"):
91 ref_data["parsed_uri"] = xmpp_uri.parseXMPPUri(ref_data["uri"])
92
93 for attr in ("begin", "end"):
94 try:
95 ref_data[attr] = int(reference_elt[attr])
96 except (KeyError, ValueError, TypeError):
97 continue
98
99 anchor = reference_elt.getAttribute("anchor")
100 if anchor is not None:
101 ref_data["anchor"] = anchor
102 if anchor.startswith("xmpp:"):
103 ref_data["parsed_anchor"] = xmpp_uri.parseXMPPUri(anchor)
104 return ref_data
105
106 async def _messageReceivedTrigger(
107 self,
108 client: SatXMPPEntity,
109 message_elt: domish.Element,
110 post_treat: defer.Deferred
111 ) -> bool:
112 """Check if a direct invitation is in the message, and handle it"""
113 reference_elt = next(message_elt.elements(NS_REFS, "reference"), None)
114 if reference_elt is None:
115 return True
116 try:
117 ref_data = self.refElementToRefData(reference_elt)
118 except KeyError:
119 log.warning("invalid <reference> element: {reference_elt.toXml}")
120 return True
121
122 if not await self.host.trigger.asyncPoint(
123 "XEP-0372_ref_received", client, message_elt, ref_data
124 ):
125 return False
126 return True
127
128 def buildRefElement(
129 self,
130 uri: str,
131 type_: str = "mention",
132 begin: Optional[int] = None,
133 end: Optional[int] = None,
134 anchor: Optional[str] = None,
135 ) -> domish.Element:
136 """Build and return the <reference> element"""
137 if type_ not in ALLOWED_TYPES:
138 raise ValueError(f"Unknown type: {type_!r}")
139 reference_elt = domish.Element(
140 (NS_REFS, "reference"),
141 attribs={"uri": uri, "type": type_}
142 )
143 if begin is not None:
144 reference_elt["begin"] = str(begin)
145 if end is not None:
146 reference_elt["end"] = str(end)
147 if anchor is not None:
148 reference_elt["anchor"] = anchor
149 return reference_elt
150
151 def _sendReference(
152 self,
153 recipient: str,
154 anchor: str,
155 type_: str,
156 extra_s: str,
157 profile_key: str
158 ) -> defer.Deferred:
159 recipient_jid = jid.JID(recipient)
160 client = self.host.getClient(profile_key)
161 extra: dict = data_format.deserialise(extra_s, default={})
162 self.sendReference(
163 client,
164 uri=extra.get("uri"),
165 type_=type_,
166 anchor=anchor,
167 to_jid=recipient_jid
168 )
169
170 def sendReference(
171 self,
172 client: "SatXMPPEntity",
173 uri: Optional[str] = None,
174 type_: str = "mention",
175 begin: Optional[int] = None,
176 end: Optional[int] = None,
177 anchor: Optional[str] = None,
178 message_elt: Optional[domish.Element] = None,
179 to_jid: Optional[jid.JID] = None
180 ) -> None:
181 """Build and send a reference_elt
182
183 @param uri: URI pointing to referenced object (XMPP entity, Pubsub Item, etc)
184 if not set, "to_jid" will be used to build an URI to the entity
185 @param type_: type of reference
186 one of [ALLOWED_TYPES]
187 @param begin: optional begin index
188 @param end: optional end index
189 @param anchor: URI of refering object (message, pubsub item), when the refence
190 is not already in the wrapping message element. In other words, it's the
191 object where the reference appears.
192 @param message_elt: wrapping <message> element, if not set a new one will be
193 generated
194 @param to_jid: destinee of the reference. If not specified, "to" attribute of
195 message_elt will be used.
196
197 """
198 if uri is None:
199 if to_jid is None:
200 raise exceptions.InternalError(
201 '"to_jid" must be set if "uri is None"'
202 )
203 uri = xmpp_uri.buildXMPPUri(path=to_jid.full())
204 if message_elt is None:
205 message_elt = domish.Element((None, "message"))
206
207 if to_jid is not None:
208 message_elt["to"] = to_jid.full()
209 else:
210 try:
211 to_jid = jid.JID(message_elt["to"])
212 except (KeyError, RuntimeError):
213 raise exceptions.InternalError(
214 'invalid "to" attribute in given message element: '
215 '{message_elt.toXml()}'
216 )
217
218 message_elt.addChild(self.buildRefElement(uri, type_, begin, end, anchor))
219 self._h.addHintElements(message_elt, [self._h.HINT_STORE])
220 client.send(message_elt)
221
222
223 @implementer(iwokkel.IDisco)
224 class XEP_0372_Handler(XMPPHandler):
225
226 def getDiscoInfo(self, requestor, service, nodeIdentifier=""):
227 return [disco.DiscoFeature(NS_REFS)]
228
229 def getDiscoItems(self, requestor, service, nodeIdentifier=""):
230 return []