Mercurial > libervia-backend

comparison libervia/backend/plugins/plugin_xep_0264.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_0264.py@524856bd7b19
children 0d7bb4df2343
comparison
equal deleted inserted replaced
4070:d10748475025 4071:4b842c1fb686
1 #!/usr/bin/env python3
2
3 # SàT plugin for managing xep-0264
4 # Copyright (C) 2009-2021 Jérôme Poisson (goffi@goffi.org)
5 # Copyright (C) 2014 Emmanuel Gil Peyrot (linkmauve@linkmauve.fr)
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 libervia.backend.core.i18n import _
21 from libervia.backend.core.constants import Const as C
22 from libervia.backend.core.log import getLogger
23
24 log = getLogger(__name__)
25 from twisted.internet import threads
26 from twisted.python.failure import Failure
27
28 from zope.interface import implementer
29
30 from wokkel import disco, iwokkel
31
32 from libervia.backend.core import exceptions
33 import hashlib
34
35 try:
36 from PIL import Image, ImageOps
37 except:
38 raise exceptions.MissingModule(
39 "Missing module pillow, please download/install it from https://python-pillow.github.io"
40 )
41
42 #  cf. https://stackoverflow.com/a/23575424
43 from PIL import ImageFile
44
45 ImageFile.LOAD_TRUNCATED_IMAGES = True
46
47 try:
48 from twisted.words.protocols.xmlstream import XMPPHandler
49 except ImportError:
50 from wokkel.subprotocols import XMPPHandler
51
52
53 MIME_TYPE = "image/jpeg"
54 SAVE_FORMAT = "JPEG" # (cf. Pillow documentation)
55
56 NS_THUMBS = "urn:xmpp:thumbs:1"
57
58 PLUGIN_INFO = {
59 C.PI_NAME: "XEP-0264",
60 C.PI_IMPORT_NAME: "XEP-0264",
61 C.PI_TYPE: "XEP",
62 C.PI_MODES: C.PLUG_MODE_BOTH,
63 C.PI_PROTOCOLS: ["XEP-0264"],
64 C.PI_DEPENDENCIES: ["XEP-0234"],
65 C.PI_MAIN: "XEP_0264",
66 C.PI_HANDLER: "yes",
67 C.PI_DESCRIPTION: _("""Thumbnails handling"""),
68 }
69
70
71 class XEP_0264(object):
72 SIZE_SMALL = (320, 320)
73 SIZE_MEDIUM = (640, 640)
74 SIZE_BIG = (1280, 1280)
75 SIZE_FULL_SCREEN = (2560, 2560)
76 # FIXME: SIZE_FULL_SCREEN is currently discarded as the resulting files are too big
77 # for BoB
78 # TODO: use an other mechanism than BoB for bigger files
79 SIZES = (SIZE_SMALL, SIZE_MEDIUM, SIZE_BIG)
80
81 def __init__(self, host):
82 log.info(_("Plugin XEP_0264 initialization"))
83 self.host = host
84 host.trigger.add("XEP-0234_buildFileElement", self._add_file_thumbnails)
85 host.trigger.add("XEP-0234_parseFileElement", self._get_file_thumbnails)
86
87 def get_handler(self, client):
88 return XEP_0264_handler()
89
90 ## triggers ##
91
92 def _add_file_thumbnails(self, client, file_elt, extra_args):
93 try:
94 thumbnails = extra_args["extra"][C.KEY_THUMBNAILS]
95 except KeyError:
96 return
97 for thumbnail in thumbnails:
98 thumbnail_elt = file_elt.addElement((NS_THUMBS, "thumbnail"))
99 thumbnail_elt["uri"] = "cid:" + thumbnail["id"]
100 thumbnail_elt["media-type"] = MIME_TYPE
101 width, height = thumbnail["size"]
102 thumbnail_elt["width"] = str(width)
103 thumbnail_elt["height"] = str(height)
104 return True
105
106 def _get_file_thumbnails(self, client, file_elt, file_data):
107 thumbnails = []
108 for thumbnail_elt in file_elt.elements(NS_THUMBS, "thumbnail"):
109 uri = thumbnail_elt["uri"]
110 if uri.startswith("cid:"):
111 thumbnail = {"id": uri[4:]}
112 width = thumbnail_elt.getAttribute("width")
113 height = thumbnail_elt.getAttribute("height")
114 if width and height:
115 try:
116 thumbnail["size"] = (int(width), int(height))
117 except ValueError:
118 pass
119 try:
120 thumbnail["mime_type"] = thumbnail_elt["media-type"]
121 except KeyError:
122 pass
123 thumbnails.append(thumbnail)
124
125 if thumbnails:
126 # we want thumbnails ordered from smallest to biggest
127 thumbnails.sort(key=lambda t: t.get('size', (0, 0)))
128 file_data.setdefault("extra", {})[C.KEY_THUMBNAILS] = thumbnails
129 return True
130
131 ## thumbnails generation ##
132
133 def get_thumb_id(self, image_uid, size):
134 """return an ID unique for image/size combination
135
136 @param image_uid(unicode): unique id of the image
137 can be a hash
138 @param size(tuple(int)): requested size of thumbnail
139 @return (unicode): unique id for this image/size
140 """
141 return hashlib.sha256(repr((image_uid, size)).encode()).hexdigest()
142
143 def _blocking_gen_thumb(
144 self, source_path, size=None, max_age=None, image_uid=None,
145 fix_orientation=True):
146 """Generate a thumbnail for image
147
148 This is a blocking method and must be executed in a thread
149 params are the same as for [generate_thumbnail]
150 """
151 if size is None:
152 size = self.SIZE_SMALL
153 try:
154 img = Image.open(source_path)
155 except IOError:
156 return Failure(exceptions.DataError("Can't open image"))
157
158 img.thumbnail(size)
159 if fix_orientation:
160 img = ImageOps.exif_transpose(img)
161
162 uid = self.get_thumb_id(image_uid or source_path, size)
163
164 with self.host.common_cache.cache_data(
165 PLUGIN_INFO[C.PI_IMPORT_NAME], uid, MIME_TYPE, max_age
166 ) as f:
167 img.save(f, SAVE_FORMAT)
168 if fix_orientation:
169 log.debug(f"fixed orientation for {f.name}")
170
171 return img.size, uid
172
173 def generate_thumbnail(
174 self, source_path, size=None, max_age=None, image_uid=None, fix_orientation=True):
175 """Generate a thumbnail of image
176
177 @param source_path(unicode): absolute path to source image
178 @param size(int, None): max size of the thumbnail
179 can be one of self.SIZE_*
180 None to use default value (i.e. self.SIZE_SMALL)
181 @param max_age(int, None): same as for [memory.cache.Cache.cache_data])
182 @param image_uid(unicode, None): unique ID to identify the image
183 use hash whenever possible
184 if None, source_path will be used
185 @param fix_orientation(bool): if True, fix orientation using EXIF data
186 @return D(tuple[tuple[int,int], unicode]): tuple with:
187 - size of the thumbnail
188 - unique Id of the thumbnail
189 """
190 d = threads.deferToThread(
191 self._blocking_gen_thumb, source_path, size, max_age, image_uid=image_uid,
192 fix_orientation=fix_orientation
193 )
194 d.addErrback(
195 lambda failure_: log.error("thumbnail generation error: {}".format(failure_))
196 )
197 return d
198
199
200 @implementer(iwokkel.IDisco)
201 class XEP_0264_handler(XMPPHandler):
202
203 def getDiscoInfo(self, requestor, target, nodeIdentifier=""):
204 return [disco.DiscoFeature(NS_THUMBS)]
205
206 def getDiscoItems(self, requestor, target, nodeIdentifier=""):
207 return []