reformat / resolves #32
This commit is contained in:
@ -11,9 +11,11 @@ from regex import Pattern
|
||||
import sqlite3
|
||||
import gpt
|
||||
import music_tag # type: ignore
|
||||
from rapidfuzz import fuzz
|
||||
from endpoints.constructors import RadioException
|
||||
|
||||
double_space: Pattern = regex.compile(r"\s{2,}")
|
||||
non_alnum: Pattern = regex.compile(r"[^a-zA-Z0-9]")
|
||||
|
||||
"""
|
||||
TODO:
|
||||
@ -110,7 +112,36 @@ class RadioUtil:
|
||||
out_result = [str(r["artistsong"]) for r in result]
|
||||
return out_result
|
||||
|
||||
async def search_playlist(
|
||||
def datatables_search(self, filter: str) -> Optional[list[dict]]:
|
||||
"""DataTables Search
|
||||
|
||||
Args:
|
||||
filter (str): The filter query to fuzzy match with
|
||||
|
||||
Returns:
|
||||
list[str]: List of matching playlist items (if any are found)
|
||||
"""
|
||||
filter = filter.strip().lower()
|
||||
matched: list[dict] = []
|
||||
for item in self.active_playlist:
|
||||
artist: str = item.get("artist", None)
|
||||
song: str = item.get("song", None)
|
||||
artistsong: str = item.get("artistsong", None)
|
||||
album: str = item.get("album", None)
|
||||
if not artist or not song or not artistsong:
|
||||
continue
|
||||
if non_alnum.sub("", filter) in non_alnum.sub("", artistsong).lower():
|
||||
matched.append(item)
|
||||
continue
|
||||
if (
|
||||
fuzz.ratio(filter, artist) >= 85
|
||||
or fuzz.ratio(filter, song) >= 85
|
||||
or fuzz.ratio(filter, album) >= 85
|
||||
):
|
||||
matched.append(item)
|
||||
return matched
|
||||
|
||||
def search_playlist(
|
||||
self,
|
||||
artistsong: Optional[str] = None,
|
||||
artist: Optional[str] = None,
|
||||
@ -130,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)
|
||||
@ -153,9 +182,7 @@ class RadioUtil:
|
||||
for ext in self.sqlite_exts:
|
||||
db_conn.load_extension(ext)
|
||||
db_conn.row_factory = sqlite3.Row
|
||||
db_cursor = db_conn.execute(
|
||||
search_query, search_params
|
||||
)
|
||||
db_cursor = db_conn.execute(search_query, search_params)
|
||||
result: Optional[sqlite3.Row | bool] = db_cursor.fetchone()
|
||||
if not result or not isinstance(result, sqlite3.Row):
|
||||
return False
|
||||
@ -178,7 +205,7 @@ class RadioUtil:
|
||||
traceback.print_exc()
|
||||
return False
|
||||
|
||||
async def add_genre(self, artist: str, genre: str) -> bool:
|
||||
def add_genre(self, artist: str, genre: str) -> bool:
|
||||
"""
|
||||
Add artist/genre pairing to DB
|
||||
Args:
|
||||
@ -213,7 +240,7 @@ class RadioUtil:
|
||||
traceback.print_exc()
|
||||
return False
|
||||
|
||||
async def add_genres(self, pairs: list[dict[str, str]]) -> bool:
|
||||
def add_genres(self, pairs: list[dict[str, str]]) -> bool:
|
||||
"""
|
||||
(BATCH) Add artist/genre pairings to DB
|
||||
Expects list of dicts comprised of artist name (key), genre (value)
|
||||
@ -228,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):
|
||||
@ -411,11 +436,13 @@ class RadioUtil:
|
||||
if isinstance(db_cursor.lastrowid, int):
|
||||
db_conn.commit()
|
||||
else:
|
||||
logging.debug("No row inserted for track_id: %s w/ file_path: %s", track_id,
|
||||
file_path)
|
||||
logging.debug(
|
||||
"No row inserted for track_id: %s w/ file_path: %s",
|
||||
track_id,
|
||||
file_path,
|
||||
)
|
||||
except Exception as e:
|
||||
logging.debug("cache_album_art Exception: %s",
|
||||
str(e))
|
||||
logging.debug("cache_album_art Exception: %s", str(e))
|
||||
traceback.print_exc()
|
||||
|
||||
async def get_album_art(self, track_id: int) -> Optional[bytes]:
|
||||
@ -433,15 +460,12 @@ class RadioUtil:
|
||||
query_params: tuple[int] = (track_id,)
|
||||
|
||||
db_cursor = db_conn.execute(query, query_params)
|
||||
result: Optional[Union[sqlite3.Row, bool]] = (
|
||||
db_cursor.fetchone()
|
||||
)
|
||||
result: Optional[Union[sqlite3.Row, bool]] = db_cursor.fetchone()
|
||||
if not result or not isinstance(result, sqlite3.Row):
|
||||
return None
|
||||
return result["album_art"]
|
||||
except Exception as e:
|
||||
logging.debug("get_album_art Exception: %s",
|
||||
str(e))
|
||||
logging.debug("get_album_art Exception: %s", str(e))
|
||||
traceback.print_exc()
|
||||
return None
|
||||
|
||||
|
||||
Reference in New Issue
Block a user