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 []