Mercurial > libervia-backend
comparison sat/plugins/plugin_xep_0447.py @ 3897:4b7106eede0c
plugin XEP-0447: Stateless File Sharing implementation:
rel 379
author | Goffi <goffi@goffi.org> |
---|---|
date | Wed, 21 Sep 2022 22:28:20 +0200 |
parents | |
children | 0ff265725489 |
comparison
equal
deleted
inserted
replaced
3896:dbf0c7faaf49 | 3897:4b7106eede0c |
---|---|
1 #!/usr/bin/env python3 | |
2 | |
3 # Copyright (C) 2009-2022 Jérôme Poisson (goffi@goffi.org) | |
4 | |
5 # This program is free software: you can redistribute it and/or modify | |
6 # it under the terms of the GNU Affero General Public License as published by | |
7 # the Free Software Foundation, either version 3 of the License, or | |
8 # (at your option) any later version. | |
9 | |
10 # This program is distributed in the hope that it will be useful, | |
11 # but WITHOUT ANY WARRANTY; without even the implied warranty of | |
12 # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the | |
13 # GNU Affero General Public License for more details. | |
14 | |
15 # You should have received a copy of the GNU Affero General Public License | |
16 # along with this program. If not, see <http://www.gnu.org/licenses/>. | |
17 | |
18 from typing import Optional, Dict, List, Tuple, Union, Any | |
19 | |
20 from twisted.words.xish import domish | |
21 | |
22 from sat.core.constants import Const as C | |
23 from sat.core.i18n import _ | |
24 from sat.core.log import getLogger | |
25 from sat.core import exceptions | |
26 | |
27 log = getLogger(__name__) | |
28 | |
29 | |
30 PLUGIN_INFO = { | |
31 C.PI_NAME: "Stateless File Sharing", | |
32 C.PI_IMPORT_NAME: "XEP-0447", | |
33 C.PI_TYPE: "XEP", | |
34 C.PI_MODES: C.PLUG_MODE_BOTH, | |
35 C.PI_PROTOCOLS: ["XEP-0447"], | |
36 C.PI_DEPENDENCIES: ["XEP-0103", "XEP-0446"], | |
37 C.PI_MAIN: "XEP_0447", | |
38 C.PI_HANDLER: "no", | |
39 C.PI_DESCRIPTION: _("""Implementation of XEP-0447 (Stateless File Sharing)"""), | |
40 } | |
41 | |
42 NS_SFS = "urn:xmpp:sfs:0" | |
43 | |
44 | |
45 class XEP_0447: | |
46 namespace = NS_SFS | |
47 | |
48 def __init__(self, host): | |
49 log.info(_("XEP-0447 (Stateless File Sharing) plugin initialization")) | |
50 host.registerNamespace("sfs", NS_SFS) | |
51 self._u = host.plugins["XEP-0103"] | |
52 self._m = host.plugins["XEP-0446"] | |
53 | |
54 def get_file_sharing_elt( | |
55 self, | |
56 sources: List[Dict[str, Any]], | |
57 disposition: Optional[str] = None, | |
58 name: Optional[str] = None, | |
59 media_type: Optional[str] = None, | |
60 desc: Optional[str] = None, | |
61 size: Optional[int] = None, | |
62 file_hash: Optional[Tuple[str, str]] = None, | |
63 date: Optional[Union[float, int]] = None, | |
64 width: Optional[int] = None, | |
65 height: Optional[int] = None, | |
66 length: Optional[int] = None, | |
67 thumbnail: Optional[str] = None, | |
68 **kwargs, | |
69 ) -> domish.Element: | |
70 """Generate the <file-sharing/> element | |
71 | |
72 @param extra: extra metadata describing how to access the URL | |
73 @return: ``<sfs/>`` element | |
74 """ | |
75 file_sharing_elt = domish.Element((NS_SFS, "file-sharing")) | |
76 if disposition is not None: | |
77 file_sharing_elt["disposition"] = disposition | |
78 file_sharing_elt.addChild( | |
79 self._m.get_file_metadata_elt( | |
80 name=name, | |
81 media_type=media_type, | |
82 desc=desc, | |
83 size=size, | |
84 file_hash=file_hash, | |
85 date=date, | |
86 width=width, | |
87 height=height, | |
88 length=length, | |
89 thumbnail=thumbnail, | |
90 ) | |
91 ) | |
92 sources_elt = file_sharing_elt.addElement("sources") | |
93 for source_data in sources: | |
94 if "url" in source_data: | |
95 sources_elt.addChild( | |
96 self._u.get_url_data_elt(**source_data) | |
97 ) | |
98 else: | |
99 raise NotImplementedError( | |
100 f"source data not implemented: {source_data}" | |
101 ) | |
102 | |
103 return file_sharing_elt | |
104 | |
105 def parse_file_sharing_elt( | |
106 self, | |
107 file_sharing_elt: domish.Element | |
108 ) -> Dict[str, Any]: | |
109 """Parse <file-sharing/> element and return file-sharing data | |
110 | |
111 @param file_sharing_elt: <file-sharing/> element | |
112 @return: file-sharing data. It a dict whose keys correspond to | |
113 [get_file_sharing_elt] parameters | |
114 """ | |
115 if file_sharing_elt.name != "file-sharing": | |
116 try: | |
117 file_sharing_elt = next( | |
118 file_sharing_elt.elements(NS_SFS, "file-sharing") | |
119 ) | |
120 except StopIteration: | |
121 raise exceptions.NotFound | |
122 try: | |
123 data = self._m.parse_file_metadata_elt(file_sharing_elt) | |
124 except exceptions.NotFound: | |
125 data = {} | |
126 disposition = file_sharing_elt.getAttribute("disposition") | |
127 if disposition is not None: | |
128 data["disposition"] = disposition | |
129 sources = data["sources"] = [] | |
130 try: | |
131 sources_elt = next(file_sharing_elt.elements(NS_SFS, "sources")) | |
132 except StopIteration: | |
133 raise ValueError(f"<sources/> element is missing: {file_sharing_elt.toXml()}") | |
134 for elt in sources_elt.elements(): | |
135 if elt.name == "url-data" and elt.uri == self._u.namespace: | |
136 source_data = self._u.parse_url_data_elt(elt) | |
137 else: | |
138 log.warning(f"unmanaged file sharing element: {elt.toXml}") | |
139 continue | |
140 sources.append(source_data) | |
141 | |
142 return data |