codey/api
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

restore album art functionality, cleanup needed

Browse Source
This commit is contained in:
codey 2025-04-22 09:18:15 -04:00
parent e47478fb6d
commit 1a4c44e33b
2 changed files with 49 additions and 52 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

18
endpoints/radio.py
View File

@ -3,7 +3,6 @@ import traceback
import time import time
import random import random
import asyncio import asyncio
from utils import radio_util
from .constructors import ( from .constructors import (
ValidRadioNextRequest, ValidRadioNextRequest,
ValidRadioReshuffleRequest, ValidRadioReshuffleRequest,
@ -11,8 +10,8 @@ from .constructors import (
ValidRadioQueueRemovalRequest, ValidRadioQueueRemovalRequest,
ValidRadioSongRequest, ValidRadioSongRequest,
ValidRadioTypeaheadRequest, ValidRadioTypeaheadRequest,
RadioException,
) )
from utils import radio_util
from uuid import uuid4 as uuid from uuid import uuid4 as uuid
from typing import Optional from typing import Optional
@ -294,17 +293,14 @@ class Radio(FastAPI):
try: try:
background_tasks.add_task(self.radio_util.webhook_song_change, next) background_tasks.add_task(self.radio_util.webhook_song_change, next)
except Exception as e: except Exception as e:
logging.info("radio_get_next Exception: %s", str(e))
traceback.print_exc() traceback.print_exc()
try: try:
if not await self.radio_util.get_album_art(file_path=next["file_path"]): album_art = await self.radio_util.get_album_art(track_id=next["id"])
album_art = await self.radio_util.get_album_art( if not album_art:
file_path=next["file_path"] await self.radio_util.cache_album_art(next["id"], next["file_path"])
) except Exception as e:
if album_art: logging.info("radio_get_next Exception: %s", str(e))
await self.radio_util.cache_album_art(next["id"], album_art)
else:
logging.debug("Could not read album art for %s", next["file_path"])
except:
traceback.print_exc() traceback.print_exc()
return JSONResponse(content=next) return JSONResponse(content=next)

83
utils/radio_util.py
View File

@ -10,6 +10,7 @@ import regex
from regex import Pattern from regex import Pattern
import aiosqlite as sqlite3 import aiosqlite as sqlite3
import gpt import gpt
import music_tag # type: ignore
from endpoints.constructors import RadioException from endpoints.constructors import RadioException
double_space: Pattern = regex.compile(r"\s{2,}") double_space: Pattern = regex.compile(r"\s{2,}")
@ -44,6 +45,9 @@ class RadioUtil:
self.artist_genre_db_path: str = os.path.join( self.artist_genre_db_path: str = os.path.join(
"/usr/local/share", "sqlite_dbs", "artist_genre_map.db" "/usr/local/share", "sqlite_dbs", "artist_genre_map.db"
) )
self.album_art_db_path: str = os.path.join(
"/usr/local/share", "sqlite_dbs", "track_album_art.db"
)
self.active_playlist_name = "default" # not used self.active_playlist_name = "default" # not used
self.active_playlist: list[dict] = [] self.active_playlist: list[dict] = []
self.now_playing: dict = { self.now_playing: dict = {
@ -261,61 +265,58 @@ class RadioUtil:
logging.info("Playlist load failed: %s", str(e)) logging.info("Playlist load failed: %s", str(e))
traceback.print_exc() traceback.print_exc()
async def cache_album_art(self, track_id: int, album_art: bytes) -> None: async def cache_album_art(self, track_id: int, file_path: str) -> None:
""" """
Cache Album Art to SQLite DB Cache Album Art to SQLite DB
Args: Args:
track_id (int): Track ID to update track_id (int): Track ID to update
album_art (bytes): Album art data file_path (str): Path to file, for artwork extraction
Returns: Returns:
None None
""" """
return None # TODO: Album art is being reworked, temporarily return None try:
# try: logging.info(
# async with sqlite3.connect(self.active_playlist_path, timeout=2) as db_conn: "cache_album_art: Attempting to store album art for track_id: %s",
# async with await db_conn.execute( track_id,
# "UPDATE tracks SET album_art = ? WHERE id = ?", )
# ( tagger = music_tag.load_file(file_path)
# album_art, album_art = tagger["artwork"].first.data
# track_id, async with sqlite3.connect(self.album_art_db_path, timeout=2) as db_conn:
# ), async with await db_conn.execute(
# ) as db_cursor: "INSERT OR IGNORE INTO album_art (track_id, album_art) VALUES(?, ?)",
# await db_conn.commit() (
# except: track_id,
# traceback.print_exc() album_art,
),
) as db_cursor:
await db_conn.commit()
except:
traceback.print_exc()
async def get_album_art( async def get_album_art(self, track_id: int) -> Optional[bytes]:
self, track_id: Optional[int] = None, file_path: Optional[str] = None
) -> Optional[bytes]:
""" """
Get Album Art Get Album Art
Args: Args:
track_id (Optional[int]): Track ID to query (ignored if file_path is specified) track_id (int): Track ID to query
file_path (Optional[str]): file_path to query (ignored if track_id is specified)
Returns: Returns:
bytes Optional[bytes]
""" """
return None # TODO: Album art is being reworked, temporarily return None try:
# try: async with sqlite3.connect(self.album_art_db_path, timeout=2) as db_conn:
# async with sqlite3.connect(self.active_playlist_path, timeout=2) as db_conn: db_conn.row_factory = sqlite3.Row
# db_conn.row_factory = sqlite3.Row query: str = "SELECT album_art FROM album_art WHERE track_id = ?"
# query: str = "SELECT album_art FROM tracks WHERE id = ?" query_params: tuple = (track_id,)
# query_params: tuple = (track_id,)
# if file_path and not track_id: async with await db_conn.execute(query, query_params) as db_cursor:
# query = "SELECT album_art FROM tracks WHERE file_path = ?" result: Optional[Union[sqlite3.Row, bool]] = (
# query_params = (file_path,) await db_cursor.fetchone()
)
# async with await db_conn.execute(query, query_params) as db_cursor: if not result or not isinstance(result, sqlite3.Row):
# result: Optional[Union[sqlite3.Row, bool]] = ( return None
# await db_cursor.fetchone() return result["album_art"]
# ) except:
# if not result or not isinstance(result, sqlite3.Row): traceback.print_exc()
# return None return None
# return result["album_art"]
# except:
# traceback.print_exc()
# return None
def get_queue_item_by_uuid(self, _uuid: str) -> Optional[tuple[int, dict]]: def get_queue_item_by_uuid(self, _uuid: str) -> Optional[tuple[int, dict]]:
""" """