diff --git a/endpoints/RandMsg.py b/endpoints/RandMsg.py deleted file mode 100644 index 130c6d9..0000000 --- a/endpoints/RandMsg.py +++ /dev/null @@ -1,78 +0,0 @@ -#!/usr/bin/env python3.12 - -import importlib -import urllib.parse -import regex -import logging -import json -import os -import random -import aiosqlite as sqlite3 - -from typing import Any, Annotated -from fastapi import FastAPI, Form, HTTPException -from pydantic import BaseModel - - -class RandMsg(FastAPI): - def __init__(self, app: FastAPI, util, constants): - self.app = app - self.util = util - self.constants = constants - - self.endpoint_name = "randmsg" - - app.add_api_route("/%s/" % self.endpoint_name, self.randmsg_handler, methods=["POST"]) - - async def randmsg_handler(self): - """ - Get a randomly generated message - """ - - random.seed() - db_rand_selected = random.choice([0, 1, 3]) - title_attr = "Unknown" - - match db_rand_selected: - case 0: - randmsg_db_path = os.path.join("/", "var", "lib", "singerdbs", "qajoke.db") # For qajoke db - db_query = "SELECT id, ('Q: ' || question || '
A: ' || answer) FROM jokes ORDER BY RANDOM() LIMIT 1" # For qajoke db - title_attr = "QA Joke DB" - case 1: - randmsg_db_path = os.path.join("/", "var", "lib", "singerdbs", "randmsg.db") # For randmsg db - db_query = "SELECT id, msg FROM msgs WHERE LENGTH(msg) <= 180 ORDER BY RANDOM() LIMIT 1" # For randmsg db - title_attr = "Random Msg DB" - case 2: - randmsg_db_path = os.path.join("/", "var", "lib", "singerdbs", "trump.db") # For Trump Tweet DB - db_query = "SELECT id, content FROM tweets ORDER BY RANDOM() LIMIT 1" # For Trump Tweet DB - title_attr = "Trump Tweet DB" - case 3: - randmsg_db_path = os.path.join("/", "var", "lib", "singerdbs", "philo.db") # For Philo DB - db_query = "SELECT id, (content || '
- ' || speaker) FROM quotes ORDER BY RANDOM() LIMIT 1" # For Philo DB - title_attr = "Philosophical Quotes DB" - case 4: - randmsg_db_path = os.path.join("/", "var", "lib", "singerdbs", "hate.db") # For Hate DB - db_query = """SELECT id, ("" || comment) FROM hate_speech WHERE length(comment) <= 180 ORDER BY RANDOM() LIMIT 1""" - title_attr = "Hate Speech DB" - case 5: - randmsg_db_path = os.path.join("/", "var", "lib", "singerdbs", "rjokes.db") # r/jokes DB - db_query = """SELECT id, (title || "
" || body) FROM jokes WHERE score >= 10000 ORDER BY RANDOM() LIMIT 1""" - title_attr = "r/jokes DB" - case 6: - randmsg_db_path = os.path.join("/", "var", "lib", "singerdbs", "donnies.db") # Donnies DB - random.seed() - twilight_or_mice = random.choice(["twilight", "mice"]) - db_query = f"SELECT id, text FROM {twilight_or_mice} ORDER BY RANDOM() LIMIT 1" - title_attr = "Donnies DB" - - - async with sqlite3.connect(database=randmsg_db_path, timeout=1) as _db: - async with _db.execute(db_query) as _cursor: - result = await _cursor.fetchone() - (result_id, result_msg) = result - result_msg = result_msg.strip() - return { - "id": result_id, - "msg": result_msg, - 'title': title_attr - } \ No newline at end of file diff --git a/endpoints/LyricSearch.py b/endpoints/lyric_search.py similarity index 51% rename from endpoints/LyricSearch.py rename to endpoints/lyric_search.py index 117957f..7e02f2b 100644 --- a/endpoints/LyricSearch.py +++ b/endpoints/lyric_search.py @@ -3,11 +3,8 @@ import importlib import urllib.parse import regex -import logging -import json -from typing import Any, Annotated -from fastapi import FastAPI, Form, HTTPException +from fastapi import FastAPI, HTTPException from pydantic import BaseModel @@ -17,7 +14,7 @@ class ValidLyricRequest(BaseModel): - **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] - - **sub**: text to search within lyrics, if found lyrics will begin at found verse [optional, default: none] + - **sub**: text to search within lyrics, if found lyrics will begin at found verse [optional] - **src**: the script/utility which initiated the request """ @@ -28,7 +25,7 @@ class ValidLyricRequest(BaseModel): extra: bool | None = False src: str - class Config: + class Config: # pylint: disable=missing-class-docstring too-few-public-methods schema_extra = { "example": { "a": "eminem", @@ -39,7 +36,8 @@ class ValidLyricRequest(BaseModel): } class LyricSearch(FastAPI): - def __init__(self, app: FastAPI, util, constants): + """Lyric Search Endpoint""" + def __init__(self, app: FastAPI, util, constants): # pylint: disable=super-init-not-called self.app = app self.util = util self.constants = constants @@ -56,8 +54,8 @@ class LyricSearch(FastAPI): "LIMNORIA-SHARED" ] - app.add_api_route("/%s/" % self.endpoint_name, self.lyric_search_handler, methods=["POST"]) - + app.add_api_route("/%s/" % self.endpoint_name, self.lyric_search_handler, methods=["POST"]) # pylint: disable=consider-using-f-string trailing-whitespace + async def lyric_search_handler(self, data: ValidLyricRequest): """ Search for lyrics @@ -71,49 +69,48 @@ class LyricSearch(FastAPI): """ src = data.src.upper() - if not(src in self.acceptable_request_sources): - raise HTTPException(detail="Invalid request source", status_code=403) - + 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 + 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 = (searchArtist == "!" and searchSong == "!") + random_song_requested = (search_artist == "!" and search_song == "!") 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 + 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(searchText) and not(query_valid): + if not random_song_requested and (not search_text and not query_valid): 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 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 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)) + 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: + search_object = self.lyrics_engine.create_query_object("%s : %s : %s" % (search_artist, search_song, sub_search)) else: - searchObject = self.lyrics_engine.create_query_object(str(searchText)) - - searchWorker = await self.lyrics_engine.lyrics_worker(searching=searchObject, + search_object = self.lyrics_engine.create_query_object(str(search_text)) + + search_worker = await self.lyrics_engine.lyrics_worker(searching=search_object, recipient='anyone') - if not(searchWorker) or not('l') in searchWorker.keys(): + if not search_worker or not 'l' in search_worker.keys(): return { 'err': True, 'errorText': 'Sources exhausted, lyrics not located.' @@ -121,12 +118,10 @@ class LyricSearch(FastAPI): return { 'err': False, - 'artist': searchWorker['artist'], - 'song': searchWorker['song'], - 'combo_lev': f'{searchWorker['combo_lev']:.2f}', - 'lyrics': regex.sub(r"\s/\s", "
", " ".join(searchWorker['l'])), - 'from_cache': searchWorker['method'].strip().lower().startswith("local cache"), - 'src': searchWorker['method'] if addExtras else None, + 'artist': search_worker['artist'], + 'song': search_worker['song'], + 'combo_lev': f'{search_worker['combo_lev']:.2f}', + 'lyrics': regex.sub(r"\s/\s", "
", " ".join(search_worker['l'])), + 'from_cache': search_worker['method'].strip().lower().startswith("local cache"), + 'src': search_worker['method'] if add_extras else None, } - - diff --git a/endpoints/rand_msg.py b/endpoints/rand_msg.py new file mode 100644 index 0000000..3e423ac --- /dev/null +++ b/endpoints/rand_msg.py @@ -0,0 +1,105 @@ +#!/usr/bin/env python3.12 + +import os +import random +import aiosqlite as sqlite3 + +from fastapi import FastAPI + + +class RandMsg(FastAPI): + """Random Message Endpoint""" + def __init__(self, app: FastAPI, util, constants): # pylint: disable=super-init-not-called + self.app = app + self.util = util + self.constants = constants + + self.endpoint_name = "randmsg" + + app.add_api_route("/%s/" % self.endpoint_name, self.randmsg_handler, methods=["POST"]) + + async def randmsg_handler(self): + """ + Get a randomly generated message + """ + random.seed() + db_rand_selected = random.choice([0, 1, 3]) + title_attr = "Unknown" + + match db_rand_selected: + case 0: + randmsg_db_path = os.path.join("/", + "var", + "lib", + "singerdbs", + "qajoke.db") # For qajoke db + db_query = "SELECT id, ('Q: ' || question || '
A: '" \ + "|| answer) FROM jokes ORDER BY RANDOM() LIMIT 1" # For qajoke db + title_attr = "QA Joke DB" + case 1: + randmsg_db_path = os.path.join("/", + "var", + "lib", + "singerdbs", + "randmsg.db") # For randmsg db + db_query = "SELECT id, msg FROM msgs WHERE" \ + "LENGTH(msg) <= 180 ORDER BY RANDOM() LIMIT 1" # For randmsg db + title_attr = "Random Msg DB" + case 2: + randmsg_db_path = os.path.join("/", + "var", + "lib", + "singerdbs", + "trump.db") # For Trump Tweet DB + db_query = "SELECT id, content FROM tweets" \ + "ORDER BY RANDOM() LIMIT 1" # For Trump Tweet DB + title_attr = "Trump Tweet DB" + case 3: + randmsg_db_path = os.path.join("/", + "var", + "lib", + "singerdbs", + "philo.db") # For Philo DB + db_query = "SELECT id, (content || '
- ' || speaker) FROM quotes" \ + "ORDER BY RANDOM() LIMIT 1" # For Philo DB + title_attr = "Philosophical Quotes DB" + case 4: + randmsg_db_path = os.path.join("/", + "var", + "lib", + "singerdbs", + "hate.db") # For Hate DB + db_query = """SELECT id, ("" || comment) FROM hate_speech" \ + "WHERE length(comment) <= 180 ORDER BY RANDOM() LIMIT 1""" + title_attr = "Hate Speech DB" + case 5: + randmsg_db_path = os.path.join("/", + "var", + "lib", + "singerdbs", + "rjokes.db") # r/jokes DB + db_query = """SELECT id, (title || "
" || body) FROM jokes""" \ + """WHERE score >= 10000 ORDER BY RANDOM() LIMIT 1""" + title_attr = "r/jokes DB" + case 6: + randmsg_db_path = os.path.join("/", + "var", + "lib", + "singerdbs", + "donnies.db") # Donnies DB + random.seed() + twilight_or_mice = random.choice(["twilight", "mice"]) + db_query = f"SELECT id, text FROM {twilight_or_mice} ORDER BY RANDOM() LIMIT 1" + title_attr = "Donnies DB" + + async with sqlite3.connect(database=randmsg_db_path, timeout=1) as _db: + async with _db.execute(db_query) as _cursor: + result = await _cursor.fetchone() + (result_id, result_msg) = result + result_msg = result_msg.strip() + return { + "id": result_id, + "msg": result_msg, + 'title': title_attr + } + \ No newline at end of file diff --git a/main.py b/main.py index bafb4cc..b1d680d 100644 --- a/main.py +++ b/main.py @@ -55,8 +55,8 @@ End Blacklisted Routes Actionable Routes """ -randmsg_endpoint = importlib.import_module("endpoints.RandMsg").RandMsg(app, util, constants) -lyric_search_endpoint = importlib.import_module("endpoints.LyricSearch").LyricSearch(app, util, constants) +randmsg_endpoint = importlib.import_module("endpoints.rand_msg").RandMsg(app, util, constants) +lyric_search_endpoint = importlib.import_module("endpoints.lyric_search").LyricSearch(app, util, constants)