formatting/add meme endpoints

utils/meme_util.py

@@ -0,0 +1,77 @@
+import os
+import io
+from typing import Optional
+import aiosqlite as sqlite3
+from PIL import Image
+
+
+class MemeUtil:
+    """
+    Meme Utils
+    """
+
+    def __init__(self, constants) -> None:
+        self.constants = constants
+        self.meme_db_path = os.path.join("/usr/local/share", "sqlite_dbs", "meme.db")
+
+    def convert_to_png(self, in_buffer: io.BytesIO) -> bytes:
+        in_buffer.seek(0)
+        with Image.open(in_buffer) as im:
+            if im.format == "PNG":
+                raise ValueError("Already a PNG")
+            out_buffer = io.BytesIO()
+            im.save(out_buffer, format="PNG")
+            out_buffer.seek(0)
+        return out_buffer.read()
+
+    async def get_meme_by_id(self, meme_id: int) -> Optional[bytes]:
+        """
+        Get meme by id
+        Args:
+            meme_id (int)
+        Returns:
+            Optional[bytes]
+        """
+        ret_image: Optional[bytes] = None
+        buffer: Optional[io.BytesIO] = None
+        async with sqlite3.connect(self.meme_db_path, timeout=5) as db_conn:
+            db_conn.row_factory = sqlite3.Row
+            query: str = "SELECT image FROM memes WHERE id = ? LIMIT 1"
+            async with await db_conn.execute(query, (meme_id,)) as db_cursor:
+                result = await db_cursor.fetchone()
+                if not result:
+                    return None
+                buffer = io.BytesIO(result["image"])
+                with Image.open(buffer) as im:
+                    if im.format != "PNG":
+                        ret_image = self.convert_to_png(buffer)
+                    else:
+                        ret_image = result["image"]
+                return ret_image
+
+    async def list_memes(self, page: int) -> Optional[list]:
+        """
+        List memes (paginated)
+        Args:
+            page (id)
+        Returns:
+            list
+        """
+        out_result: list = []
+        async with sqlite3.connect(self.meme_db_path, timeout=5) as db_conn:
+            db_conn.row_factory = sqlite3.Row
+            rows_per_page: int = 10
+            offset: int = (page - 1) * rows_per_page
+            query: str = "SELECT id, timestamp FROM memes ORDER BY timestamp DESC LIMIT 10 OFFSET ?"
+            async with await db_conn.execute(query, (offset,)) as db_cursor:
+                results = await db_cursor.fetchall()
+                for result in results:
+                    result_id = result["id"]
+                    result_timestamp = result["timestamp"]
+                    out_result.append(
+                        {
+                            "id": result_id,
+                            "timestamp": result_timestamp,
+                        }
+                    )
+        return out_result

utils/radio_util.py

@@ -56,6 +56,7 @@ class RadioUtil:
             "deathcore",
             "edm",
             "electronic",
+            "hard rock",
         ]
         self.active_playlist: list[dict] = []
         self.playlist_loaded: bool = False
@@ -160,11 +161,9 @@ class RadioUtil:
         if not artistsong and (not artist or not song):
             raise RadioException("No query provided")
         try:
-            search_query: str = (
-                'SELECT id, artist, song, (artist || " - " || song) AS artistsong, album, file_path, duration FROM tracks\
+            search_query: str = 'SELECT id, artist, song, (artist || " - " || song) AS artistsong, album, file_path, duration FROM tracks\
                 WHERE editdist3((lower(artist) || " " || lower(song)), (? || " " || ?))\
                     <= 410 ORDER BY editdist3((lower(artist) || " " || lower(song)), ?) ASC LIMIT 1'
-            )
             if artistsong:
                 artistsong_split: list = artistsong.split(" - ", maxsplit=1)
                 (search_artist, search_song) = tuple(artistsong_split)
@@ -256,9 +255,7 @@ class RadioUtil:
                 for pair in pairs:
                     try:
                         artist, genre = pair
-                        query: str = (
-                            "INSERT OR IGNORE INTO artist_genre (artist, genre) VALUES(?, ?)"
-                        )
+                        query: str = "INSERT OR IGNORE INTO artist_genre (artist, genre) VALUES(?, ?)"
                         params: tuple[str, str] = (artist, genre)
                         res = _db.execute(query, params)
                         if isinstance(res.lastrowid, int):
@@ -376,7 +373,7 @@ class RadioUtil:
                     dedupe_processed.append(artistsongabc)
 
                 logging.info(
-                    "Duplicates removed." "New playlist size: %s",
+                    "Duplicates removed.New playlist size: %s",
                     len(self.active_playlist),
                 )