417 lines
17 KiB
Python
417 lines
17 KiB
Python
#!/usr/bin/env python3.12
|
|
# pylint: disable=bare-except, broad-exception-raised, broad-exception-caught
|
|
|
|
import importlib
|
|
import traceback
|
|
import logging
|
|
import os
|
|
import urllib.parse
|
|
from typing import Optional
|
|
import regex
|
|
import aiohttp
|
|
import aiosqlite as sqlite3
|
|
from fastapi import FastAPI, HTTPException
|
|
from pydantic import BaseModel
|
|
from lyric_search_new.sources import aggregate
|
|
|
|
|
|
class ValidLyricRequest(BaseModel):
|
|
"""
|
|
- **a**: artist
|
|
- **s**: song
|
|
- **t**: track (artist and song combined) [used only if a & s are not used]
|
|
- **extra**: include extra details in response [optional, default: false]
|
|
- **lrc**: Request LRCs?
|
|
- **sub**: text to search within lyrics, if found lyrics will begin at found verse [optional]
|
|
- **src**: the script/utility which initiated the request
|
|
- **excluded_sources**: sources to exclude (new only)
|
|
"""
|
|
|
|
a: str | None = None
|
|
s: str | None = None
|
|
t: str | None = None
|
|
sub: str | None = None
|
|
extra: bool | None = False
|
|
lrc: bool | None = False
|
|
src: str
|
|
excluded_sources: list | None = None
|
|
|
|
class Config: # pylint: disable=missing-class-docstring too-few-public-methods
|
|
schema_extra = {
|
|
"example": {
|
|
"a": "eminem",
|
|
"s": "rap god",
|
|
"src": "WEB",
|
|
"extra": True,
|
|
"lrc": False,
|
|
}
|
|
}
|
|
|
|
class ValidTypeAheadRequest(BaseModel):
|
|
"""
|
|
- **query**: query string
|
|
"""
|
|
pre_query: str|None = None
|
|
query: str
|
|
|
|
|
|
class ValidLyricSearchLogRequest(BaseModel):
|
|
"""
|
|
- **webradio**: whether or not to include requests generated automatically by the radio page on codey.lol, defaults to False
|
|
"""
|
|
|
|
webradio: bool = False
|
|
|
|
|
|
class CacheUtils:
|
|
"""Lyrics Cache DB Utils"""
|
|
def __init__(self):
|
|
self.lyrics_db_path = os.path.join("/", "var", "lib",
|
|
"singerdbs", "cached_lyrics.db")
|
|
|
|
async def check_typeahead(self, s: str, pre_query: str | None = None):
|
|
"""Check s against artists stored - for typeahead"""
|
|
async with sqlite3.connect(self.lyrics_db_path,
|
|
timeout=2) as db_conn:
|
|
db_conn.row_factory = lambda c, r: dict([(col[0], r[idx]) for idx, col in enumerate(c.description)])
|
|
if not pre_query:
|
|
query = "SELECT distinct(artist) FROM lyrics WHERE artist LIKE ? LIMIT 15"
|
|
query_params = (f"%{s}%",)
|
|
else:
|
|
query = "SELECT distinct(song) FROM lyrics WHERE artist LIKE ? AND song LIKE ? LIMIT 15"
|
|
query_params = (f"%{pre_query}%", f"%{s}%",)
|
|
async with db_conn.execute(query, query_params) as db_cursor:
|
|
return await db_cursor.fetchall()
|
|
|
|
|
|
class LyricSearch(FastAPI):
|
|
"""Lyric Search Endpoint"""
|
|
def __init__(self, app: FastAPI, util, constants, glob_state): # pylint: disable=super-init-not-called
|
|
self.app = app
|
|
self.util = util
|
|
self.constants = constants
|
|
self.glob_state = glob_state
|
|
self.cache_utils = CacheUtils()
|
|
self.lyrics_engine = importlib.import_module("lyrics_engine").LyricsEngine()
|
|
|
|
self.endpoint_name = "lyric_search"
|
|
self.endpoint2_name = "lyric_cache_list"
|
|
|
|
self.endpoints = {
|
|
"typeahead/artist": self.artist_typeahead_handler,
|
|
"typeahead/song": self.song_typeahead_handler,
|
|
# "lyric_search": self.lyric_search_handler,
|
|
"lyric_search": self.new_test, #test
|
|
"lyric_cache_list": self.lyric_cache_list_handler,
|
|
"lyric_search_history": self.lyric_search_log_handler,
|
|
"lyric_search_test": self.new_test,
|
|
}
|
|
|
|
self.acceptable_request_sources = [
|
|
"WEB",
|
|
"WEB-RADIO",
|
|
"IRC-MS",
|
|
"IRC-FS",
|
|
"IRC-KALI",
|
|
"DISC-ACES",
|
|
"DISC-HAVOC",
|
|
"IRC-SHARED"
|
|
]
|
|
|
|
self.lrc_regex = regex.compile(r'\[([0-9]{2}:[0-9]{2})\.[0-9]{1,3}\](\s(.*)){0,}')
|
|
|
|
for endpoint, handler in self.endpoints.items():
|
|
app.add_api_route(f"/{endpoint}/", handler, methods=["POST"])
|
|
|
|
async def lyric_cache_list_handler(self):
|
|
"""
|
|
Get currently cached lyrics entries
|
|
"""
|
|
return {
|
|
'err': False,
|
|
'data': await self.lyrics_engine.listCacheEntries()
|
|
}
|
|
|
|
async def artist_typeahead_handler(self, data: ValidTypeAheadRequest):
|
|
"""Artist Type Ahead Handler"""
|
|
if not isinstance(data.query, str) or len(data.query) < 2:
|
|
return {
|
|
'err': True,
|
|
'errorText': 'Invalid request',
|
|
}
|
|
query = data.query
|
|
typeahead_result = await self.cache_utils.check_typeahead(query)
|
|
typeahead_list = [str(r.get('artist')) for r in typeahead_result]
|
|
return typeahead_list
|
|
|
|
async def song_typeahead_handler(self, data: ValidTypeAheadRequest):
|
|
"""Song Type Ahead Handler"""
|
|
if not isinstance(data.pre_query, str) or len(data.pre_query) < 2\
|
|
or not isinstance(data.query, str) or len(data.query) < 2:
|
|
return {
|
|
'err': True,
|
|
'errorText': 'Invalid request',
|
|
}
|
|
pre_query = data.pre_query
|
|
query = data.query
|
|
typeahead_result = await self.cache_utils.check_typeahead(query, pre_query)
|
|
typeahead_list = [str(r.get('song')) for r in typeahead_result]
|
|
return typeahead_list
|
|
|
|
async def lyric_search_log_handler(self, data: ValidLyricSearchLogRequest):
|
|
"""Lyric Search Log Handler"""
|
|
include_radio = data.webradio
|
|
await self.glob_state.increment_counter('lyrichistory_requests')
|
|
last_10k_sings = await self.lyrics_engine.getHistory(limit=10000, webradio=include_radio)
|
|
return {
|
|
'err': False,
|
|
'history': last_10k_sings
|
|
}
|
|
|
|
async def new_test(self, data: ValidLyricRequest):
|
|
"""
|
|
Search for lyrics (testing)
|
|
|
|
- **a**: artist
|
|
- **s**: song
|
|
- **t**: track (artist and song combined) [used only if a & s are not used] [unused]
|
|
- **extra**: include extra details in response [optional, default: false] [unused]
|
|
- **lrc**: Request LRCs?
|
|
- **sub**: text to search within lyrics, if found lyrics will begin at found verse [optional, default: none]
|
|
- **src**: the script/utility which initiated the request
|
|
- **excluded_sources**: sources to exclude
|
|
"""
|
|
|
|
if not data.a or not data.s or not data.src:
|
|
raise HTTPException(detail="Invalid request", status_code=500)
|
|
|
|
if data.src.upper() not in self.acceptable_request_sources:
|
|
raise HTTPException(detail="Invalid request", status_code=500)
|
|
|
|
search_artist: Optional[str] = data.a
|
|
search_song: Optional[str] = data.s
|
|
|
|
|
|
|
|
if search_artist and search_song:
|
|
search_artist = self.constants.DOUBLE_SPACE_REGEX.sub(" ", search_artist.strip())
|
|
search_song = self.constants.DOUBLE_SPACE_REGEX.sub(" ", search_song.strip())
|
|
search_artist = urllib.parse.unquote(search_artist)
|
|
search_song = urllib.parse.unquote(search_song)
|
|
|
|
excluded_sources = data.excluded_sources
|
|
aggregate_search = aggregate.Aggregate(exclude_methods=excluded_sources)
|
|
plain_lyrics = not data.lrc
|
|
result = await aggregate_search.search(search_artist, search_song, plain_lyrics)
|
|
|
|
if not result:
|
|
return {
|
|
'err': True,
|
|
'errorText': 'Sources exhausted, lyrics not located.',
|
|
}
|
|
|
|
result = result.todict()
|
|
|
|
if data.sub and not data.lrc:
|
|
seeked_found_line = None
|
|
lyric_lines = result['lyrics'].strip().split(" / ")
|
|
for i, line in enumerate(lyric_lines):
|
|
line = regex.sub(r'\u2064', '', line.strip())
|
|
if data.sub.strip().lower() in line.strip().lower():
|
|
seeked_found_line = i
|
|
logging.debug("Found %s at %s, match for %s!",
|
|
line, seeked_found_line, data.sub) # REMOVEME: DEBUG
|
|
break
|
|
|
|
if not seeked_found_line:
|
|
return {
|
|
'failed_seek': True,
|
|
}
|
|
result['lyrics'] = " / ".join(lyric_lines[seeked_found_line:])
|
|
|
|
result['confidence'] = int(result.get('confidence', 0))
|
|
result['time'] = f'{float(result['time']):.4f}'
|
|
|
|
if plain_lyrics:
|
|
result['lyrics'] = regex.sub(r'(\s/\s|\n)', '<br>', result['lyrics']).strip()
|
|
else:
|
|
# Swap lyrics key for 'lrc'
|
|
result['lrc'] = result['lyrics']
|
|
result.pop('lyrics')
|
|
|
|
if "cache" in result['src']:
|
|
result['from_cache'] = True
|
|
|
|
"""
|
|
REMOVE BELOW AFTER TESTING IS DONE
|
|
"""
|
|
|
|
# if not data.extra:
|
|
# result.pop('src')
|
|
return result
|
|
|
|
|
|
|
|
async def lyric_search_handler(self, data: ValidLyricRequest):
|
|
"""
|
|
Search for lyrics
|
|
|
|
- **a**: artist
|
|
- **s**: song
|
|
- **t**: track (artist and song combined) [used only if a & s are not used]
|
|
- **extra**: include extra details in response [optional, default: false]
|
|
- **lrc**: Request LRCs?
|
|
- **sub**: text to search within lyrics, if found lyrics will begin at found verse [optional, default: none]
|
|
- **src**: the script/utility which initiated the request
|
|
"""
|
|
|
|
lrc = data.lrc
|
|
src = data.src.upper()
|
|
if not src in self.acceptable_request_sources:
|
|
raise HTTPException(detail="Invalid request source", status_code=403)
|
|
|
|
await self.glob_state.increment_counter('lyric_requests')
|
|
|
|
search_artist = data.a
|
|
search_song = data.s
|
|
search_text = data.t
|
|
add_extras = data.extra
|
|
sub_search = data.sub
|
|
search_object = None
|
|
|
|
random_song_requested = (search_artist == "!" and search_song == "!")
|
|
query_valid = (
|
|
not(search_artist is None) and
|
|
not(search_song is None) and
|
|
len(search_artist) >= 1 and
|
|
len(search_song) >= 1 and
|
|
len(search_artist) + len(search_song) >= 3
|
|
)
|
|
|
|
if not random_song_requested and (not search_text and not query_valid):
|
|
return {
|
|
"err": True,
|
|
"errorText": "Invalid parameters"
|
|
}
|
|
if search_artist and search_song:
|
|
search_artist = self.constants.DOUBLE_SPACE_REGEX.sub(" ", search_artist.strip())
|
|
search_song = self.constants.DOUBLE_SPACE_REGEX.sub(" ", search_song.strip())
|
|
search_artist = urllib.parse.unquote(search_artist)
|
|
search_song = urllib.parse.unquote(search_song)
|
|
|
|
if search_text is None:
|
|
# pylint: disable=consider-using-f-string
|
|
search_object = self.lyrics_engine.create_query_object("%s : %s" % (search_artist, search_song))
|
|
if sub_search:
|
|
sub_search = regex.sub(r'\s{2,}', ' ', sub_search.strip())
|
|
search_object = self.lyrics_engine.create_query_object("%s : %s : %s" % (search_artist, search_song, sub_search))
|
|
else:
|
|
search_object = self.lyrics_engine.create_query_object(str(search_text))
|
|
|
|
if lrc:
|
|
search_worker = await self.lyrics_engine.grabFromSpotify(searching=search_object,
|
|
lrc=True)
|
|
|
|
spotify_lyrics_unsynced = True
|
|
if search_worker and search_worker.get('l'):
|
|
for line in search_worker.get('l'):
|
|
if line.get('timeTag') and line.get('timeTag') != "00:00.00":
|
|
spotify_lyrics_unsynced = False
|
|
if not search_worker or spotify_lyrics_unsynced:
|
|
# Try LRCLib before failing out
|
|
try:
|
|
lrclib_api_url = "https://lrclib.net/api/get"
|
|
sane_artist = urllib.parse.quote_plus(search_artist)
|
|
sane_track = urllib.parse.quote_plus(search_song)
|
|
async with aiohttp.ClientSession() as session:
|
|
async with session.get(f"{lrclib_api_url}?artist_name={sane_artist}&track_name={sane_track}") as request:
|
|
request.raise_for_status()
|
|
response_json = await request.json()
|
|
if not "syncedLyrics" in response_json:
|
|
raise BaseException("LRCLib Fallback Failed")
|
|
lrc_content = response_json.get('syncedLyrics')
|
|
returned_artist = response_json.get('artistName')
|
|
returned_song = response_json.get('trackName')
|
|
logging.debug("Synced Lyrics [LRCLib]: %s",
|
|
lrc_content)
|
|
lrc_content_out = []
|
|
for line in lrc_content.split("\n"):
|
|
_timetag = None
|
|
_words = None
|
|
if not line.strip():
|
|
continue
|
|
reg_helper = regex.findall(self.lrc_regex, line.strip())
|
|
if not reg_helper:
|
|
continue
|
|
reg_helper = reg_helper[0]
|
|
_timetag = reg_helper[0]
|
|
if not reg_helper[1].strip():
|
|
_words = "♪"
|
|
else:
|
|
_words = reg_helper[1]
|
|
lrc_content_out.append({
|
|
"timeTag": _timetag,
|
|
"words": _words,
|
|
})
|
|
|
|
return {
|
|
'err': False,
|
|
'artist': returned_artist,
|
|
'song': returned_song,
|
|
'combo_lev': "N/A",
|
|
'lrc': lrc_content_out,
|
|
'from_cache': False,
|
|
'src': 'Alt LRC SRC',
|
|
'reqn': await self.glob_state.get_counter('lyric_requests'),
|
|
}
|
|
except:
|
|
traceback.print_exc()
|
|
return {
|
|
'err': True,
|
|
'errorText': 'Search failed!',
|
|
}
|
|
|
|
|
|
return {
|
|
'err': True,
|
|
'errorText': 'Search failed!',
|
|
}
|
|
if lrc:
|
|
return {
|
|
'err': False,
|
|
'artist': search_worker['artist'],
|
|
'song': search_worker['song'],
|
|
'combo_lev': search_worker['combo_lev'],
|
|
'lrc': search_worker['l'],
|
|
'from_cache': False,
|
|
'src': search_worker['method'],
|
|
'reqn': await self.glob_state.get_counter('lyric_requests'),
|
|
}
|
|
|
|
search_worker = await self.lyrics_engine.lyrics_worker(searching=search_object)
|
|
|
|
|
|
if not search_worker or not 'l' in search_worker.keys():
|
|
await self.glob_state.increment_counter('failedlyric_requests')
|
|
return {
|
|
'err': True,
|
|
'errorText': 'Sources exhausted, lyrics not located.'
|
|
}
|
|
|
|
|
|
await self.lyrics_engine.storeHistEntry(artist=search_worker.get('artist'),
|
|
song=search_worker.get('song'),
|
|
retr_method=search_worker.get('method'),
|
|
request_src=src.strip())
|
|
|
|
return {
|
|
'err': False,
|
|
'artist': search_worker['artist'],
|
|
'song': search_worker['song'],
|
|
'combo_lev': f'{search_worker['combo_lev']:.2f}',
|
|
'lyrics': regex.sub(r"\s/\s", "<br>", " ".join(search_worker['l'])),
|
|
'from_cache': search_worker['method'].strip().lower().startswith("local cache"),
|
|
'src': search_worker['method'] if add_extras else None,
|
|
'reqn': await self.glob_state.get_counter('lyric_requests')
|
|
}
|