111 lines
3.5 KiB
Python
111 lines
3.5 KiB
Python
#!/usr/bin/env python3.12
|
|
|
|
import importlib
|
|
import urllib.parse
|
|
import regex
|
|
import logging
|
|
|
|
from typing import Any, Annotated
|
|
from fastapi import FastAPI, Form, HTTPException
|
|
from pydantic import BaseModel
|
|
|
|
|
|
class ValidRequest(BaseModel):
|
|
a: str | None = None
|
|
s: str | None = None
|
|
t: str | None = None
|
|
sub: str | None = None
|
|
extra: bool | None = False
|
|
src: str | None = None
|
|
|
|
|
|
class LyricSearch(FastAPI):
|
|
def __init__(self, app: FastAPI, util, constants):
|
|
self.app = app
|
|
self.util = util
|
|
self.constants = constants
|
|
self.lyrics_engine = importlib.import_module("lyrics_engine").LyricsEngine()
|
|
|
|
self.endpoint_name = "/search/"
|
|
self.search_method_order = [
|
|
666,
|
|
1,
|
|
8,
|
|
2,
|
|
3
|
|
]
|
|
self.acceptable_request_sources = [
|
|
"WEB",
|
|
"IRC-MS",
|
|
"IRC-FS",
|
|
"IRC-KALI",
|
|
"DISC-ACES",
|
|
"DISC-HAVOC",
|
|
"LIMNORIA-SHARED"
|
|
]
|
|
|
|
app.add_api_route(self.endpoint_name, self.search_handler, methods=["POST"])
|
|
|
|
async def search_handler(self, data: ValidRequest):
|
|
print(f"HI, DATA:\n{data}")
|
|
src = data.src.upper()
|
|
if not(src in self.acceptable_request_sources):
|
|
raise HTTPException(detail="Invalid request source", status_code=403)
|
|
|
|
|
|
searchArtist = data.a
|
|
searchSong = data.s
|
|
searchText = data.t
|
|
addExtras = data.extra
|
|
subSearch = data.sub
|
|
searchObject = None
|
|
|
|
random_song_requested = (searchArtist == "!" and searchSong == "!")
|
|
query_valid = (
|
|
not(searchArtist is None) and
|
|
not(searchSong is None) and
|
|
len(searchArtist) >= 1 and
|
|
len(searchSong) >= 1 and
|
|
len(searchArtist) + len(searchSong) >= 3
|
|
)
|
|
|
|
if not(random_song_requested) and not(query_valid):
|
|
logging.debug(f"DEBUG!\nsearchSong: {searchSong}\nsearchArtist: {searchArtist}")
|
|
return {
|
|
"err": True,
|
|
"errorText": "Invalid parameters"
|
|
}
|
|
|
|
if searchArtist and searchSong:
|
|
searchArtist = self.constants.DOUBLE_SPACE_REGEX.sub(" ", searchArtist.strip())
|
|
searchSong = self.constants.DOUBLE_SPACE_REGEX.sub(" ", searchSong.strip())
|
|
searchArtist = urllib.parse.unquote(searchArtist)
|
|
searchSong = urllib.parse.unquote(searchSong)
|
|
|
|
if searchText is None:
|
|
searchObject = self.lyrics_engine.create_query_object("%s : %s" % (searchArtist, searchSong))
|
|
if subSearch:
|
|
searchObject = self.lyrics_engine.create_query_object("%s : %s : %s" % (searchArtist, searchSong, subSearch))
|
|
|
|
searchWorker = await self.lyrics_engine.lyrics_worker(searching=searchObject,
|
|
method_order=self.search_method_order,
|
|
recipient='anyone')
|
|
|
|
if not(searchWorker) or not('l') in searchWorker.keys():
|
|
return {
|
|
'err': True,
|
|
'errorText': 'Sources exhausted, lyrics not located.'
|
|
}
|
|
|
|
return {
|
|
'err': False,
|
|
'artist': searchWorker['artist'],
|
|
'song': searchWorker['song'],
|
|
'combo_lev': f'{searchWorker['combo_lev']:.2f}',
|
|
'lyrics': regex.sub(r"\s/\s", "<br>", " ".join(searchWorker['l'])),
|
|
'from_cache': searchWorker['method'].strip().lower().startswith("local cache"),
|
|
'src': searchWorker['method'] if addExtras else None,
|
|
}
|
|
|
|
|