Mercurial > libervia-backend
comparison libervia/backend/plugins/plugin_xep_0358.py @ 4335:430d5d99a740
plugin XEP-0358: "Publishing Available Jingle Sessions" implementation:
rel 453
author | Goffi <goffi@goffi.org> |
---|---|
date | Tue, 03 Dec 2024 00:13:07 +0100 |
parents | |
children |
comparison
equal
deleted
inserted
replaced
4334:111dce64dcb5 | 4335:430d5d99a740 |
---|---|
1 #!/usr/bin/env python3 | |
2 | |
3 # Libervia plugin to jingle session publishing. | |
4 # Copyright (C) 2009-2024 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 TYPE_CHECKING, Self | |
20 | |
21 from pydantic import BaseModel, Field | |
22 from twisted.words.protocols.jabber import jid | |
23 from twisted.words.protocols.jabber.xmlstream import XMPPHandler | |
24 from twisted.words.xish import domish | |
25 from wokkel import disco, iwokkel | |
26 from zope.interface import implementer | |
27 | |
28 from libervia.backend.core import exceptions | |
29 from libervia.backend.core.constants import Const as C | |
30 from libervia.backend.core.core_types import SatXMPPEntity | |
31 from libervia.backend.core.i18n import _ | |
32 from libervia.backend.core.log import getLogger | |
33 from libervia.backend.models.types import DomishElementType, JIDType | |
34 | |
35 if TYPE_CHECKING: | |
36 from libervia.backend.core.main import LiberviaBackend | |
37 | |
38 log = getLogger(__name__) | |
39 | |
40 | |
41 PLUGIN_INFO = { | |
42 C.PI_NAME: "Publishing Available Jingle Sessions", | |
43 C.PI_IMPORT_NAME: "XEP-0358", | |
44 C.PI_TYPE: "XEP", | |
45 C.PI_MODES: C.PLUG_MODE_BOTH, | |
46 C.PI_PROTOCOLS: [], | |
47 C.PI_DEPENDENCIES: [ | |
48 "XEP-0234", | |
49 ], | |
50 C.PI_RECOMMENDATIONS: [], | |
51 C.PI_MAIN: "XEP_0358", | |
52 C.PI_HANDLER: "yes", | |
53 C.PI_DESCRIPTION: _("""Indicate available Jingle sessions."""), | |
54 } | |
55 | |
56 NS_JINGLEPUB = "urn:xmpp:jinglepub:1" | |
57 | |
58 | |
59 class Meta(BaseModel): | |
60 """Model for meta element in JinglePub element.""" | |
61 | |
62 lang: str | None = Field(None, description="The language of the meta data.") | |
63 title: str = Field(description="The title of the meta data.") | |
64 summary: str | None = Field(None, description="A summary of the meta data.") | |
65 | |
66 def to_element(self) -> domish.Element: | |
67 """Build the <meta> element from this instance's data. | |
68 | |
69 @return: <meta> element. | |
70 """ | |
71 meta_elt = domish.Element((NS_JINGLEPUB, "meta")) | |
72 if self.lang: | |
73 meta_elt["xml:lang"] = self.lang | |
74 meta_elt["title"] = self.title | |
75 if self.summary: | |
76 meta_elt["summary"] = self.summary | |
77 return meta_elt | |
78 | |
79 @classmethod | |
80 def from_element(cls, meta_elt: domish.Element) -> Self: | |
81 """Create a Meta instance from a <meta> element. | |
82 | |
83 @param meta_elt: The <meta> element. | |
84 @return: Meta instance. | |
85 """ | |
86 assert meta_elt.name == "meta" and meta_elt.uri == NS_JINGLEPUB | |
87 kwargs = {} | |
88 if meta_elt.hasAttribute("xml:lang"): | |
89 kwargs["lang"] = meta_elt["xml:lang"] | |
90 if meta_elt.hasAttribute("title"): | |
91 kwargs["title"] = meta_elt["title"] | |
92 if meta_elt.hasAttribute("summary"): | |
93 kwargs["summary"] = meta_elt["summary"] | |
94 return cls(**kwargs) | |
95 | |
96 | |
97 class JinglePub(BaseModel): | |
98 """Model for JinglePub element.""" | |
99 | |
100 from_jid: JIDType = Field(description="The JID of the session owner.") | |
101 id: str = Field(description="The jinglepub identifier.") | |
102 uri: str | None = Field( | |
103 default=None, description="An alternate or informational URI of the content." | |
104 ) | |
105 meta: list[Meta] = Field( | |
106 default_factory=list, description="Human-friendly information about the session." | |
107 ) | |
108 descriptions: list[DomishElementType] = Field( | |
109 default_factory=list, description="Application format(s) for the Jingle session." | |
110 ) | |
111 | |
112 def set_attributes(self, jinglepub_elt: domish.Element) -> None: | |
113 """Set <jinglepub> element attributes from this instance's data.""" | |
114 jinglepub_elt["from"] = str(self.from_jid) | |
115 jinglepub_elt["id"] = self.id | |
116 | |
117 def set_children(self, jinglepub_elt: domish.Element) -> None: | |
118 """Set <jinglepub> element children from this instance's data.""" | |
119 if self.uri: | |
120 uri_elt = jinglepub_elt.addElement((NS_JINGLEPUB, "uri")) | |
121 uri_elt.addContent(self.uri) | |
122 for meta in self.meta: | |
123 jinglepub_elt.addChild(meta.to_element()) | |
124 for description in self.descriptions: | |
125 jinglepub_elt.addChild(description) | |
126 | |
127 def to_element(self) -> domish.Element: | |
128 """Build the <jinglepub> element from this instance's data. | |
129 | |
130 @return: <jinglepub> element. | |
131 """ | |
132 jinglepub_elt = domish.Element((NS_JINGLEPUB, "jinglepub")) | |
133 self.set_attributes(jinglepub_elt) | |
134 self.set_children(jinglepub_elt) | |
135 return jinglepub_elt | |
136 | |
137 @classmethod | |
138 def from_element(cls, element: domish.Element) -> Self: | |
139 """Create a JinglePub instance from a <jinglepub> element. | |
140 | |
141 @param jinglepub_elt: The <jinglepub> element. | |
142 @return: JinglePub instance. | |
143 @raise exceptions.NotFound: If the <jinglepub> element is not found. | |
144 """ | |
145 if element.uri != NS_JINGLEPUB or element.name != "jinglepub": | |
146 raise exceptions.NotFound("<jinglepub> element not found") | |
147 | |
148 kwargs = {} | |
149 try: | |
150 kwargs["from_jid"] = jid.JID(element["from"]) | |
151 except RuntimeError as e: | |
152 log.warning(f"Can't parse from_jid {e}: {element.toXml()}") | |
153 raise exceptions.NotFound("<jinglepub> element has invalid 'from' attribute") | |
154 | |
155 if element.hasAttribute("id"): | |
156 kwargs["id"] = element["id"] | |
157 | |
158 uri_elt = next(element.elements(NS_JINGLEPUB, "uri"), None) | |
159 if uri_elt: | |
160 kwargs["uri"] = str(uri_elt) | |
161 | |
162 kwargs["meta"] = [ | |
163 Meta.from_element(meta_elt) | |
164 for meta_elt in element.elements(NS_JINGLEPUB, "meta") | |
165 ] | |
166 | |
167 kwargs["descriptions"] = [ | |
168 child | |
169 for child in element.elements() | |
170 if child.uri != NS_JINGLEPUB and child.name == "description" | |
171 ] | |
172 | |
173 return cls(**kwargs) | |
174 | |
175 | |
176 class XEP_0358: | |
177 namespace = NS_JINGLEPUB | |
178 | |
179 def __init__(self, host: "LiberviaBackend") -> None: | |
180 log.info(f"plugin {PLUGIN_INFO[C.PI_NAME]!r} initialization") | |
181 self.host = host | |
182 host.register_namespace("jinglepub", NS_JINGLEPUB) | |
183 | |
184 def get_handler(self, client: SatXMPPEntity) -> XMPPHandler: | |
185 return JinglePubHandler(self) | |
186 | |
187 | |
188 @implementer(iwokkel.IDisco) | |
189 class JinglePubHandler(XMPPHandler): | |
190 | |
191 def __init__(self, plugin_parent): | |
192 self.plugin_parent = plugin_parent | |
193 | |
194 def getDiscoInfo( | |
195 self, requestor: jid.JID, target: jid.JID, nodeIdentifier: str = "" | |
196 ) -> list[disco.DiscoFeature]: | |
197 return [ | |
198 disco.DiscoFeature(NS_JINGLEPUB), | |
199 ] | |
200 | |
201 def getDiscoItems( | |
202 self, requestor: jid.JID, target: jid.JID, nodeIdentifier: str = "" | |
203 ) -> list[disco.DiscoItems]: | |
204 return [] |