memes - change method used to check if images are already stored as PNG

utils/meme_util.py

@@ -1,4 +1,5 @@
 import os
+import logging
 import io
 from typing import Optional
 import aiosqlite as sqlite3
@@ -14,7 +15,33 @@ class MemeUtil:
         self.constants = constants
         self.meme_db_path = os.path.join("/usr/local/share", "sqlite_dbs", "meme.db")
 
+    def is_png(self, buffer: bytes | io.BytesIO) -> bool:
+        """
+        Check if image (in-memory buffer, or bytes) is a PNG
+        Args:
+            buffer (bytes|io.BytesIO)
+        Returns:
+            bool
+        """
+        # Accepts either bytes or a BytesIO-like object
+        if isinstance(buffer, io.BytesIO):
+            if hasattr(buffer, "read") and hasattr(buffer, "seek"):
+                pos = buffer.tell()
+                buffer.seek(0)
+                signature = buffer.read(8)
+                buffer.seek(pos)
+        else:
+            signature = buffer[:8]
+        return signature == b"\x89PNG\r\n\x1a\n"
+
     def convert_to_png(self, in_buffer: io.BytesIO) -> bytes:
+        """
+        Convert an in-memory buffer to PNG
+        Args:
+            in_buffer (io.BytesIO)
+        Returns:
+            bytes
+        """
         in_buffer.seek(0)
         with Image.open(in_buffer) as im:
             if im.format == "PNG":
@@ -42,11 +69,12 @@ class MemeUtil:
                 if not result:
                     return None
                 buffer = io.BytesIO(result["image"])
-                with Image.open(buffer) as im:
+                is_png = self.is_png(buffer)
-                    if im.format != "PNG":
+                if not is_png:
-                        ret_image = self.convert_to_png(buffer)
+                    logging.debug("Converting %s, not detected as PNG", meme_id)
-                    else:
+                    ret_image = self.convert_to_png(buffer)
-                        ret_image = result["image"]
+                else:
+                    ret_image = result["image"]
                 return ret_image
 
     async def list_memes(self, page: int) -> Optional[list]: