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

misc

Browse Source
This commit is contained in:
codey 2025-02-18 06:55:47 -05:00
parent 36e191e27f
commit ed1bc29cc8
8 changed files with 54 additions and 31 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

10
base.py
View File

@ -1,5 +1,3 @@
#!/usr/bin/env python3.12
import importlib
import sys
sys.path.insert(0, ".")
@ -11,6 +9,7 @@ from fastapi.middleware.cors import CORSMiddleware
from lyric_search.sources import redis_cache
logger = logging.getLogger()
logger.setLevel(logging.INFO)
@ -23,6 +22,7 @@ app = FastAPI(title="codey.lol API",
redirect_slashes=False,
loop=loop)
constants = importlib.import_module("constants").Constants()
util = importlib.import_module("util").Utilities(app, constants)
@ -31,11 +31,11 @@ origins = [
"https://api.codey.lol"
]
app.add_middleware(CORSMiddleware,
app.add_middleware(CORSMiddleware, # type: ignore
allow_origins=origins,
allow_credentials=True,
allow_methods=["POST", "GET", "HEAD"],
allow_headers=["*"])
allow_headers=["*"]) # type: ignore
"""
Blacklisted routes
@ -91,6 +91,8 @@ karma_endpoints = importlib.import_module("endpoints.karma").Karma(app, util, co
radio_endpoints = importlib.import_module("endpoints.radio").Radio(app, util, constants)
# Below: Misc endpoints
misc_endpoints = importlib.import_module("endpoints.misc").Misc(app, util, constants, radio_endpoints)
# Below: Mgr Endpoints
mgr_endpoints = importlib.import_module("endpoints.mgr.mgr_test").Mgr(app, util, constants)
"""
End Actionable Routes

7
endpoints/lastfm.py
View File

@ -157,8 +157,13 @@ class LastFM(FastAPI):
'errorText': 'Invalid request'
})
track_info_result: dict = await self.lastfm.get_track_info(artist=artist,
track_info_result: Optional[dict] = await self.lastfm.get_track_info(artist=artist,
track=track)
if not track_info_result:
return JSONResponse(status_code=500, content={
'err': True,
'errorText': 'Not found.',
})
if "err" in track_info_result:
raise LastFMException("Unknown error occurred: %s",
track_info_result.get('errorText', '??'))

13
endpoints/radio_util.py
View File

@ -72,8 +72,9 @@ class RadioUtil:
async with sqlite3.connect(self.active_playlist_path,
timeout=1) as _db:
_db.row_factory = sqlite3.Row
db_query: str = 'SELECT DISTINCT(LOWER(TRIM(artist) || " - " || TRIM(song))) as artistsong FROM tracks WHERE\
artistsong LIKE ? LIMIT 30'
db_query: str = """SELECT DISTINCT(LOWER(TRIM(artist) || " - " || TRIM(song))),\
(TRIM(artist) || " - " || TRIM(song)) as artistsong FROM tracks WHERE\
artistsong LIKE ? LIMIT 30"""
db_params: tuple[str] = (f"%{query}%",)
async with _db.execute(db_query, db_params) as _cursor:
result: Iterable[sqlite3.Row] = await _cursor.fetchall()
@ -150,7 +151,7 @@ class RadioUtil:
LIMITED GENRES
"""
db_query: str = """SELECT distinct(LOWER(TRIM(artist)) || " - " || LOWER(TRIM(song))) AS artistdashsong, id, artist, song, genre, file_path, duration FROM tracks\
db_query: str = """SELECT distinct(LOWER(TRIM(artist)) || " - " || LOWER(TRIM(song))), (TRIM(artist) || " - " || TRIM(song)) AS artistdashsong, id, artist, song, genre, file_path, duration FROM tracks\
WHERE genre LIKE "%metalcore%"\
OR genre LIKE "%rock%"\
OR genre LIKE "%pop punk%"\
@ -182,7 +183,9 @@ class RadioUtil:
OR genre LIKE "%hardcore punk%"\
OR genre LIKE "%synthwave%"\
OR genre LIKE "%trap%"\
OR genre LIKE "%indie pop%" GROUP BY artistdashsong ORDER BY RANDOM()"""
OR genre LIKE "%indie pop%"\
OR genre LIKE "untagged"\
GROUP BY artistdashsong ORDER BY RANDOM()"""
"""
LIMITED TO ONE ARTIST...
@ -333,7 +336,7 @@ class RadioUtil:
"description": f'## {track['song']}\nby\n## {track['artist']}',
"color": 0x30c56f,
"thumbnail": {
"url": f"https://api.codey.lol/radio/album_art?track_id={track['id']}&{time.time()}",
"url": f"https://api.codey.lol/radio/album_art?track_id={track['id']}&{int(time.time())}",
},
"fields": [
{

4
lastfm_wrapper.py
View File

@ -44,7 +44,7 @@ class LastFM:
}
async def get_track_info(self, artist: Optional[str] = None,
track: Optional[str] = None) -> dict:
track: Optional[str] = None) -> Optional[dict]:
"""
Get Track Info from LastFM
Args:
@ -66,6 +66,8 @@ class LastFM:
request.raise_for_status()
data: dict = await request.json()
data = data.get('track', None)
if not isinstance(data.get('artist'), dict):
return None
ret_obj: dict = {
'artist_mbid': data.get('artist', None).get('mbid'),
'album': data.get('album', None).get('title'),

44
lyric_search/sources/redis_cache.py
View File

@ -13,10 +13,10 @@ sys.path.insert(1,'..')
from lyric_search import notifier
from lyric_search.constructors import LyricsResult
import redis.asyncio as redis
from redis.commands.search.query import Query
from redis.commands.search.indexDefinition import IndexDefinition, IndexType
from redis.commands.search.field import TextField, TagField
from redis.commands.json.path import Path
from redis.commands.search.query import Query # type: ignore
from redis.commands.search.indexDefinition import IndexDefinition, IndexType # type: ignore
from redis.commands.search.field import TextField, TagField # type: ignore
from redis.commands.json.path import Path # type: ignore
from . import private
logger = logging.getLogger()
@ -33,7 +33,7 @@ class RedisCache:
"""
def __init__(self) -> None:
self.redis_client = redis.Redis(password=private.REDIS_PW)
self.redis_client: redis.Redis = redis.Redis(password=private.REDIS_PW)
self.notifier = notifier.DiscordNotifier()
self.notify_warnings = True
self.regexes: list[Pattern] = [
@ -95,10 +95,11 @@ class RedisCache:
src = src.strip().lower()
await self.redis_client.incr(f"returned:{src}")
except Exception as e:
await self.notifier.send(f"ERROR @ {__file__.rsplit("/", maxsplit=1)[-1]}", f"{str(e)}")
file = __file__.rsplit("/", maxsplit=1)[-1]
await self.notifier.send(f"ERROR @ {file}", str(e))
traceback.print_exc()
async def get_found_counts(self) -> dict:
async def get_found_counts(self) -> Optional[dict]:
"""
Get found counts for all sources (and failed count)
Args:
@ -111,16 +112,20 @@ class RedisCache:
counts: dict[str, int] = {}
for src in sources:
src_found_count = await self.redis_client.get(f"returned:{src}")
if not isinstance(src_found_count, int):
return None
counts[src] = int(src_found_count) # Redis returns bytes
return counts
except Exception as e:
await self.notifier.send(f"ERROR @ {__file__.rsplit("/", maxsplit=1)[-1]}", f"{str(e)}")
file = __file__.rsplit("/", maxsplit=1)[-1]
await self.notifier.send(f"ERROR @ {file}", str(e))
traceback.print_exc()
return None
async def search(self, artist: Optional[str] = None,
song: Optional[str] = None,
lyrics: Optional[str] = None) -> list[tuple]:
lyrics: Optional[str] = None) -> Optional[list[tuple]]:
"""
Search Redis Cache
Args:
@ -142,25 +147,30 @@ class RedisCache:
if not is_random_search:
logging.debug("Redis: Searching normally first")
if not artist or not song:
logging.info("redis_cache:: search failed: No artist or song provided.")
return None
(artist, song) = self.sanitize_input(artist, song)
logging.debug("Seeking: %s - %s", artist, song)
search_res: Union[dict, list] = await self.redis_client.ft().search(Query(
search_res: Union[dict, list] = await self.redis_client.ft().search(Query( # type: ignore
f"@artist:{artist} @song:{song}"
))
search_res_out: list[tuple] = [(result['id'].split(":",
maxsplit=1)[1], dict(json.loads(result['json'])))
for result in search_res.docs]
for result in search_res.docs] # type: ignore
if not search_res_out:
logging.debug("Redis: Normal search failed, trying with fuzzy search")
(fuzzy_artist, fuzzy_song) = self.sanitize_input(artist=artist.split(" ")[0:5],
song=song.split(" ")[0:6], fuzzy=True)
search_res = await self.redis_client.ft().search(Query(
short_artist = " ".join(artist.split(" ")[0:5])
short_song = " ".join(song.split(" ")[0:5])
(fuzzy_artist, fuzzy_song) = self.sanitize_input(artist=short_artist.strip(),
song=short_song.strip(), fuzzy=True)
search_res = await self.redis_client.ft().search(Query( # type: ignore
f"@artist:{fuzzy_artist} @song:{fuzzy_song}"
))
search_res_out = [(result['id'].split(":",
maxsplit=1)[1], dict(json.loads(result['json'])))
for result in search_res.docs]
for result in search_res.docs] # type: ignore
else:
random_redis_key: str = await self.redis_client.randomkey()
@ -175,6 +185,7 @@ class RedisCache:
except Exception as e:
traceback.print_exc()
# await self.notifier.send(f"ERROR @ {__file__.rsplit("/", maxsplit=1)[-1]}", f"{str(e)}\nSearch was: {artist} - {song}; fuzzy: {fuzzy_artist} - {fuzzy_song}")
return None
async def redis_store(self, sqlite_id: int,
lyr_result: LyricsResult) -> None:
"""
@ -213,6 +224,7 @@ class RedisCache:
await self.notifier.send("INFO",
f"Stored {lyr_result.artist} - {lyr_result.song} (related SQLite Row ID: {sqlite_id}) to redis: {newkey}")
except Exception as e:
await self.notifier.send(f"ERROR @ {__file__.rsplit("/", maxsplit=1)[-1]}",
file = __file__.rsplit("/", maxsplit=1)[-1]
await self.notifier.send(f"ERROR @ {file}",
f"Failed to store {lyr_result.artist} - {lyr_result.song}\
(SQLite id: {sqlite_id}) to Redis:\n{str(e)}")

1
mypy.ini Normal file
View File

@ -0,0 +1 @@
[mypy]

0
py.typed Normal file
View File

2
pyproject.toml
View File

@ -23,5 +23,3 @@ dependencies = [
"typing-inspect>=0.9.0",
"http3>=0.6.7",
]
[tool.mypy]
disable_error_code = ["import-untyped"]