refactor: move image upload to a unified function

matemat/util/thumbnails.py

@@ -0,0 +1,31 @@
+
+import os
+from io import BytesIO
+from shutil import copyfile
+
+import magic
+from PIL import Image
+
+
+def upload_thumbnail(image_data: bytes, filename: str, size: int = 150) -> bool:
+    # Only process the image, if its size is more than zero.  Zero size means no new image was uploaded
+    if image_data is None or len(image_data) == 0:
+        return False
+    # Detect the MIME type
+    filemagic: magic.FileMagic = magic.detect_from_content(image_data)
+    if not filemagic.mime_type.startswith('image/'):
+        raise ValueError(f'Unsupported file type: {filemagic.mime_type}')
+    # Create the absolute path of the upload directory
+    dirname: str = os.path.abspath(os.path.dirname(filename))
+    os.makedirs(dirname, exist_ok=True)
+
+    # Parse the image data
+    try:
+        image: Image = Image.open(BytesIO(image_data))
+    except IOError:
+        raise ValueError(f'Unsupported file type: {filemagic.mime_type}')
+    # Resize the image to 150x150
+    image.thumbnail((150, 150), Image.LANCZOS)
+    # Write the image to the file
+    image.save(filename, 'PNG')
+    return True

matemat/webserver/pagelets/admin.py

@@ -1,16 +1,13 @@
 import os
 from datetime import datetime, UTC
-from io import BytesIO
-from shutil import copyfile
 
-import magic
-from PIL import Image
 from bottle import get, post, abort, redirect, request, FormsDict
 
 from matemat.db import MatematDatabase
 from matemat.db.primitives import User, ReceiptPreference
 from matemat.exceptions import AuthenticationError, DatabaseConsistencyError
 from matemat.util.currency_format import parse_chf
+from matemat.util.thumbnails import upload_thumbnail
 from matemat.webserver import session, template
 from matemat.webserver.config import get_app_config, get_stock_provider
 from matemat.webserver.template import Notification
@@ -101,22 +98,11 @@ def handle_change(args: FormsDict, files: FormsDict, db: MatematDatabase):
             newproduct = db.create_product(name, price_member, price_non_member, custom_price, stockable, barcode)
             # If a new product image was uploaded, process it
             image = files.image.file.read() if 'image' in files else None
+            filename = os.path.join(os.path.abspath(config['UploadDir']), f'thumbnails/products/{newproduct.id}.png')
             if image is not None and len(image) > 0:
-                # Detect the MIME type
-                filemagic: magic.FileMagic = magic.detect_from_content(image)
-                if not filemagic.mime_type.startswith('image/'):
-                    return
-                # Create the absolute path of the upload directory
-                abspath: str = os.path.join(os.path.abspath(config['UploadDir']), 'thumbnails/products/')
-                os.makedirs(abspath, exist_ok=True)
                 try:
-                    # Parse the image data
+                    upload_thumbnail(image, filename)
-                    image: Image = Image.open(BytesIO(image))
+                except Exception as e:
-                    # Resize the image to 150x150
-                    image.thumbnail((150, 150), Image.LANCZOS)
-                    # Write the image to the file
-                    image.save(os.path.join(abspath, f'{newproduct.id}.png'), 'PNG')
-                except OSError as e:
                     Notification.error(str(e), decay=True)
                     return
 
@@ -145,24 +131,11 @@ def handle_change(args: FormsDict, files: FormsDict, db: MatematDatabase):
                     continue
                 # Read the raw image data from the request
                 default: bytes = files[category].file.read()
-                # Only process the image, if its size is more than zero.  Zero size means no new image was uploaded
+                filename: str = os.path.join(os.path.abspath(config['UploadDir']), f'thumbnails/{category}/default.png')
-                if len(default) == 0:
-                    continue
-                # Detect the MIME type
-                filemagic: magic.FileMagic = magic.detect_from_content(default)
-                if not filemagic.mime_type.startswith('image/'):
-                    continue
-                # Create the absolute path of the upload directory
-                abspath: str = os.path.join(os.path.abspath(config['UploadDir']), f'thumbnails/{category}/')
-                os.makedirs(abspath, exist_ok=True)
                 try:
-                    # Parse the image data
+                    if upload_thumbnail(default, filename):
-                    image: Image = Image.open(BytesIO(default))
+                        Notification.success(f'{category} default image updated successfully.', decay=True)
-                    # Resize the image to 150x150
+                except Exception as e:
-                    image.thumbnail((150, 150), Image.LANCZOS)
-                    # Write the image to the file
-                    image.save(os.path.join(abspath, f'default.png'), 'PNG')
-                except OSError as e:
                     Notification.error(str(e), decay=True)
                     return
 

matemat/webserver/pagelets/modproduct.py

@@ -1,18 +1,17 @@
 import os
-from io import BytesIO
 from datetime import datetime, UTC
 from typing import Dict
 
-import magic
-from PIL import Image
 from bottle import get, post, redirect, abort, request, FormsDict
 
 from matemat.db import MatematDatabase
 from matemat.db.primitives import Product
 from matemat.exceptions import DatabaseConsistencyError
 from matemat.util.currency_format import parse_chf
+from matemat.util.thumbnails import upload_thumbnail
 from matemat.webserver import template, session
 from matemat.webserver.config import get_app_config, get_stock_provider
+from matemat.webserver.template import Notification
 
 
 @get('/modproduct')
@@ -136,23 +135,9 @@ def handle_change(args: FormsDict, files: FormsDict, product: Product, db: Matem
         # If a new product image was uploaded, process it
         if 'image' in files:
             # Read the raw image data from the request
-            avatar = files.image.file.read()
+            image = files.image.file.read()
-            # Only process the image, if its size is more than zero.  Zero size means no new image was uploaded
+            filename: str = os.path.join(os.path.abspath(config['UploadDir']), f'thumbnails/products/{product.id}.png')
-            if len(avatar) == 0:
-                return
-            # Detect the MIME type
-            filemagic: magic.FileMagic = magic.detect_from_content(avatar)
-            if not filemagic.mime_type.startswith('image/'):
-                return
-            # Create the absolute path of the upload directory
-            abspath: str = os.path.join(os.path.abspath(config['UploadDir']), 'thumbnails/products/')
-            os.makedirs(abspath, exist_ok=True)
             try:
-                # Parse the image data
+                upload_thumbnail(image, filename)
-                image: Image = Image.open(BytesIO(avatar))
+            except Exception as e:
-                # Resize the image to 150x150
+                Notification.error(str(e), decay=True)
-                image.thumbnail((150, 150), Image.LANCZOS)
-                # Write the image to the file
-                image.save(os.path.join(abspath, f'{product.id}.png'), 'PNG')
-            except OSError:
-                return

matemat/webserver/pagelets/moduser.py

@@ -1,18 +1,17 @@
 import os
 from datetime import datetime, UTC
-from io import BytesIO
 from typing import Dict, Optional
 
-import magic
-from PIL import Image
 from bottle import get, post, redirect, abort, request, FormsDict
 
 from matemat.db import MatematDatabase
 from matemat.db.primitives import User, ReceiptPreference
 from matemat.exceptions import DatabaseConsistencyError
 from matemat.util.currency_format import parse_chf
+from matemat.util.thumbnails import upload_thumbnail
 from matemat.webserver import template, session
 from matemat.webserver.config import get_app_config
+from matemat.webserver.template import Notification
 
 
 @get('/moduser')
@@ -144,23 +143,9 @@ def handle_change(args: FormsDict, files: FormsDict, user: User, authuser: User,
         # If a new avatar was uploaded, process it
         if 'avatar' in files:
             # Read the raw image data from the request
-            avatar = files.avatar.file.read()
+            image = files.avatar.file.read()
-            # Only process the image, if its size is more than zero.  Zero size means no new image was uploaded
+            filename: str = os.path.join(os.path.abspath(config['UploadDir']), f'thumbnails/users/{user.id}.png')
-            if len(avatar) == 0:
-                return
-            # Detect the MIME type
-            filemagic: magic.FileMagic = magic.detect_from_content(avatar)
-            if not filemagic.mime_type.startswith('image/'):
-                return
-            # Create the absolute path of the upload directory
-            abspath: str = os.path.join(os.path.abspath(config['UploadDir']), 'thumbnails/users/')
-            os.makedirs(abspath, exist_ok=True)
             try:
-                # Parse the image data
+                upload_thumbnail(image, filename)
-                image: Image = Image.open(BytesIO(avatar))
+            except Exception as e:
-                # Resize the image to 150x150
+                Notification.error(str(e), decay=True)
-                image.thumbnail((150, 150), Image.LANCZOS)
-                # Write the image to the file
-                image.save(os.path.join(abspath, f'{user.id}.png'), 'PNG')
-            except OSError:
-                return

matemat/webserver/pagelets/settings.py

@@ -1,16 +1,13 @@
 import os
 from datetime import datetime, UTC
-from io import BytesIO
-from shutil import copyfile
 
-import magic
-from PIL import Image
 from bottle import get, post, abort, redirect, request, FormsDict
 
 from matemat.db import MatematDatabase
 from matemat.db.primitives import User, ReceiptPreference
 from matemat.exceptions import AuthenticationError, DatabaseConsistencyError
 from matemat.util.currency_format import parse_chf
+from matemat.util.thumbnails import upload_thumbnail
 from matemat.webserver import session, template
 from matemat.webserver.config import get_app_config, get_stock_provider
 from matemat.webserver.template import Notification
@@ -149,31 +146,15 @@ def handle_change(args: FormsDict, files: FormsDict, user: User, db: MatematData
             Notification.success(f'Token {token.name} removed', decay=True)
 
         # The user requested an avatar change
-        elif change == 'avatar':
+        elif change == 'avatar' and 'avatar' in files:
-            # The new avatar field must be present
-            if 'avatar' not in files:
-                return
             # Read the raw image data from the request
-            avatar = files.avatar.file.read()
+            image = files.avatar.file.read()
-            # Only process the image, if its size is more than zero.  Zero size means no new image was uploaded
+            filename: str = os.path.join(os.path.abspath(config['UploadDir']), f'thumbnails/users/{user.id}.png')
-            if len(avatar) == 0:
-                return
-            # Detect the MIME type
-            filemagic: magic.FileMagic = magic.detect_from_content(avatar)
-            if not filemagic.mime_type.startswith('image/'):
-                return
-            # Create the absolute path of the upload directory
-            abspath: str = os.path.join(os.path.abspath(config['UploadDir']), 'thumbnails/users/')
-            os.makedirs(abspath, exist_ok=True)
             try:
-                # Parse the image data
+                if upload_thumbnail(image, filename):
-                image: Image = Image.open(BytesIO(avatar))
+                    Notification.success('Avatar changed')
-                # Resize the image to 150x150
+            except Exception as e:
-                image.thumbnail((150, 150), Image.LANCZOS)
+                Notification.error(str(e), decay=True)
-                # Write the image to the file
-                image.save(os.path.join(abspath, f'{user.id}.png'), 'PNG')
-            except OSError:
-                return
 
     except UnicodeDecodeError:
         raise ValueError('an argument not a string')

matemat/webserver/pagelets/signup.py

@@ -1,15 +1,12 @@
 
 import os
-from shutil import copyfile
-from io import BytesIO
 
-import magic
 from bottle import get, post, redirect, abort, request, FormsDict
-from PIL import Image
 import netaddr
 
 from matemat.db import MatematDatabase
 from matemat.db.primitives import User
+from matemat.util.thumbnails import upload_thumbnail
 from matemat.webserver import template, session
 from matemat.webserver.config import get_app_config
 
@@ -33,21 +30,9 @@ def signup_user(args: FormsDict, files: FormsDict, db: MatematDatabase) -> User:
         touchkey = str(args.touchkey)
         db.change_touchkey(new_user, password, touchkey, verify_password=False)
     # Finally, set the avatar, if provided
-    if 'avatar' in files:
+    image = files.avatar.file.read() if 'avatar' in files else None
-        avatar = files.avatar.file.read()
+    filename: str = os.path.join(os.path.abspath(config['UploadDir']), f'thumbnails/users/{new_user.id}.png')
-    else:
+    upload_thumbnail(image, filename)
-        avatar = None
-    if avatar:
-        filemagic: magic.FileMagic = magic.detect_from_content(avatar)
-        if filemagic.mime_type.startswith('image/'):
-            abspath: str = os.path.join(os.path.abspath(config['UploadDir']), 'thumbnails/users/')
-            os.makedirs(abspath, exist_ok=True)
-            # Parse the image data
-            image: Image = Image.open(BytesIO(avatar))
-            # Resize the image to 150x150
-            image.thumbnail((150, 150), Image.LANCZOS)
-            # Write the image to the file
-            image.save(os.path.join(abspath, f'{new_user.id}.png'), 'PNG')
 
     return new_user
 
@@ -73,7 +58,8 @@ def signup():
         with MatematDatabase(config['DatabaseFile']) as db:
             try:
                 user = signup_user(request.params, request.files, db)
-            except ValueError as e:
+            except Exception as e:
+                Notification.error(str(e), decay=True)
                 redirect('/signup')
             # Set the user ID session variable
             session.put(session_id, 'authenticated_user', user.id)