Mercurial > libervia-backend

comparison libervia/backend/plugins/plugin_xep_0163.py @ 4071:4b842c1fb686

Find changesets by keywords (author, files, the commit message), revision number or hash, or revset expression.
refactoring: renamed `sat` package to `libervia.backend`
author Goffi <goffi@goffi.org>
date Fri, 02 Jun 2023 11:49:51 +0200
parents sat/plugins/plugin_xep_0163.py@524856bd7b19
children
comparison
equal deleted inserted replaced
4070:d10748475025 4071:4b842c1fb686
1 #!/usr/bin/env python3
2
3
4 # SAT plugin for Personal Eventing Protocol (xep-0163)
5 # Copyright (C) 2009-2021 Jérôme Poisson (goffi@goffi.org)
6
7 # This program is free software: you can redistribute it and/or modify
8 # it under the terms of the GNU Affero General Public License as published by
9 # the Free Software Foundation, either version 3 of the License, or
10 # (at your option) any later version.
11
12 # This program is distributed in the hope that it will be useful,
13 # but WITHOUT ANY WARRANTY; without even the implied warranty of
14 # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
15 # GNU Affero General Public License for more details.
16
17 # You should have received a copy of the GNU Affero General Public License
18 # along with this program. If not, see <http://www.gnu.org/licenses/>.
19
20 from typing import Optional, Callable
21 from libervia.backend.core.i18n import _
22 from libervia.backend.core import exceptions
23 from libervia.backend.core.constants import Const as C
24 from libervia.backend.core.log import getLogger
25
26 from twisted.words.xish import domish
27
28 from wokkel import disco, pubsub
29 from wokkel.formats import Mood
30 from libervia.backend.tools.common import data_format
31
32
33 log = getLogger(__name__)
34
35 NS_USER_MOOD = "http://jabber.org/protocol/mood"
36
37 PLUGIN_INFO = {
38 C.PI_NAME: "Personal Eventing Protocol Plugin",
39 C.PI_IMPORT_NAME: "XEP-0163",
40 C.PI_TYPE: "XEP",
41 C.PI_MODES: C.PLUG_MODE_BOTH,
42 C.PI_PROTOCOLS: ["XEP-0163", "XEP-0107"],
43 C.PI_DEPENDENCIES: ["XEP-0060"],
44 C.PI_MAIN: "XEP_0163",
45 C.PI_HANDLER: "no",
46 C.PI_DESCRIPTION: _("""Implementation of Personal Eventing Protocol"""),
47 }
48
49
50 class XEP_0163(object):
51 def __init__(self, host):
52 log.info(_("PEP plugin initialization"))
53 self.host = host
54 self.pep_events = set()
55 self.pep_out_cb = {}
56 host.trigger.add("PubSub Disco Info", self.diso_info_trigger)
57 host.bridge.add_method(
58 "pep_send",
59 ".plugin",
60 in_sign="sa{ss}s",
61 out_sign="",
62 method=self.pep_send,
63 async_=True,
64 ) # args: type(MOOD, TUNE, etc), data, profile_key;
65 self.add_pep_event("MOOD", NS_USER_MOOD, self.user_mood_cb, self.send_mood)
66
67 def diso_info_trigger(self, disco_info, profile):
68 """Add info from managed PEP
69
70 @param disco_info: list of disco feature as returned by PubSub,
71 will be filled with PEP features
72 @param profile: profile we are handling
73 """
74 disco_info.extend(list(map(disco.DiscoFeature, self.pep_events)))
75 return True
76
77 def add_pep_event(
78 self,
79 event_type: Optional[str],
80 node: str,
81 in_callback: Callable,
82 out_callback: Optional[Callable] = None,
83 notify: bool = True
84 ) -> None:
85 """Add a Personal Eventing Protocol event manager
86
87 @param event_type: type of the event (stored uppercase),
88 only used when out_callback is set.
89 Can be MOOD, TUNE, etc.
90 @param node: namespace of the node (e.g. http://jabber.org/protocol/mood
91 for User Mood)
92 @param in_callback: method to call when this event occur
93 the callable will be called with (itemsEvent, profile) as arguments
94 @param out_callback: method to call when we want to publish this
95 event (must return a deferred)
96 the callable will be called when send_pep_event is called
97 @param notify: add autosubscribe (+notify) if True
98 """
99 if event_type and out_callback:
100 event_type = event_type.upper()
101 if event_type in self.pep_out_cb:
102 raise exceptions.ConflictError(
103 f"event_type {event_type!r} already exists"
104 )
105 self.pep_out_cb[event_type] = out_callback
106 self.pep_events.add(node)
107 if notify:
108 self.pep_events.add(node + "+notify")
109
110 def filter_pep_event(client, itemsEvent):
111 """Ignore messages which are not coming from PEP (i.e. a bare jid)
112
113 @param itemsEvent(pubsub.ItemsEvent): pubsub event
114 """
115 if not itemsEvent.sender.user or itemsEvent.sender.resource:
116 log.debug(
117 "ignoring non PEP event from {} (profile={})".format(
118 itemsEvent.sender.full(), client.profile
119 )
120 )
121 return
122 in_callback(itemsEvent, client.profile)
123
124 self.host.plugins["XEP-0060"].add_managed_node(node, items_cb=filter_pep_event)
125
126 def send_pep_event(self, node, data, profile):
127 """Publish the event data
128
129 @param node(unicode): node namespace
130 @param data: domish.Element to use as payload
131 @param profile: profile which send the data
132 """
133 client = self.host.get_client(profile)
134 item = pubsub.Item(payload=data)
135 return self.host.plugins["XEP-0060"].publish(client, None, node, [item])
136
137 def pep_send(self, event_type, data, profile_key=C.PROF_KEY_NONE):
138 """Send personal event after checking the data is alright
139
140 @param event_type: type of event (eg: MOOD, TUNE),
141 must be in self.pep_out_cb.keys()
142 @param data: dict of {string:string} of event_type dependant data
143 @param profile_key: profile who send the event
144 """
145 profile = self.host.memory.get_profile_name(profile_key)
146 if not profile:
147 log.error(
148 _("Trying to send personal event with an unknown profile key [%s]")
149 % profile_key
150 )
151 raise exceptions.ProfileUnknownError
152 if not event_type in list(self.pep_out_cb.keys()):
153 log.error(_("Trying to send personal event for an unknown type"))
154 raise exceptions.DataError("Type unknown")
155 return self.pep_out_cb[event_type](data, profile)
156
157 def user_mood_cb(self, itemsEvent, profile):
158 if not itemsEvent.items:
159 log.debug(_("No item found"))
160 return
161 try:
162 mood_elt = [
163 child for child in itemsEvent.items[0].elements() if child.name == "mood"
164 ][0]
165 except IndexError:
166 log.error(_("Can't find mood element in mood event"))
167 return
168 mood = Mood.fromXml(mood_elt)
169 if not mood:
170 log.debug(_("No mood found"))
171 return
172 self.host.bridge.ps_event(
173 C.PS_PEP,
174 itemsEvent.sender.full(),
175 itemsEvent.nodeIdentifier,
176 "MOOD",
177 data_format.serialise({"mood": mood.value or "", "text": mood.text or ""}),
178 profile,
179 )
180
181 def send_mood(self, data, profile):
182 """Send XEP-0107's User Mood
183
184 @param data: must include mood and text
185 @param profile: profile which send the mood"""
186 try:
187 value = data["mood"].lower()
188 text = data["text"] if "text" in data else ""
189 except KeyError:
190 raise exceptions.DataError("Mood data must contain at least 'mood' key")
191 mood = UserMood(value, text)
192 return self.send_pep_event(NS_USER_MOOD, mood, profile)
193
194
195 class UserMood(Mood, domish.Element):
196 """Improved wokkel Mood which is also a domish.Element"""
197
198 def __init__(self, value, text=None):
199 Mood.__init__(self, value, text)
200 domish.Element.__init__(self, (NS_USER_MOOD, "mood"))
201 self.addElement(value)
202 if text:
203 self.addElement("text", content=text)