diff --git a/base.py b/base.py index 8228e82..158298a 100644 --- a/base.py +++ b/base.py @@ -58,7 +58,8 @@ lyric_search_endpoint = importlib.import_module("endpoints.lyric_search").LyricS lastfm_endpoints = importlib.import_module("endpoints.lastfm").LastFM(app, util, constants) # Below: YT endpoint(s) yt_endpoints = importlib.import_module("endpoints.yt").YT(app, util, constants) - +# Below: Transcription endpoints +transcription_endpoints = importlib.import_module("endpoints.transcriptions").Transcriptions(app, util, constants) """ diff --git a/endpoints/transcriptions.py b/endpoints/transcriptions.py new file mode 100644 index 0000000..f47c02a --- /dev/null +++ b/endpoints/transcriptions.py @@ -0,0 +1,100 @@ +#!/usr/bin/env python3.12 + +import os +import aiosqlite as sqlite3 +from fastapi import FastAPI +from pydantic import BaseModel + +class ValidShowEpisodeListRequest(BaseModel): + """ + - **s**: show id + """ + + s: int + +class ValidShowEpisodeLineRequest(BaseModel): + """ + - **s**: show id + - **e**: episode id + """ + + s: str + e: str + +class Transcriptions(FastAPI): + """Transcription Endpoints""" + def __init__(self, app: FastAPI, util, constants): # pylint: disable=super-init-not-called + self.app = app + self.util = util + self.constants = constants + + self.endpoints = { + "get_episodes": self.get_episodes_handler, + "get_episode_lines": self.get_episode_lines_handler, + #tbd + } + + for endpoint, handler in self.endpoints.items(): + app.add_api_route(f"/{endpoint}/", handler, methods=["POST"]) + + async def get_episodes_handler(self, data: ValidShowEpisodeListRequest): + """ + /get_episodes/ + Get list of episodes by show id + """ + show_id = data.s + db_path = None + db_query = None + show_title = None + + if show_id is None: + return { + 'err': True, + 'errorText': 'Invalid request' + } + + show_id = int(show_id) + + if not(str(show_id).isnumeric()) or not(show_id in [0, 1, 2]): + return { + 'err': True, + 'errorText': 'Show not found.' + } + + match show_id: + case 0: + db_path = os.path.join("/", "var", "lib", "singerdbs", "sp.db") + db_query = """SELECT DISTINCT(("S" || Season || "E" || Episode || " " || Title)), ID FROM SP_DAT ORDER BY Season, Episode""" + show_title = "South Park" + case 1: + db_path = os.path.join("/", "var", "lib", "singerdbs", "futur.db") + db_query = """SELECT DISTINCT(("S" || EP_S || "E" || EP_EP || " " || EP_TITLE)), EP_ID FROM clean_dialog ORDER BY EP_S, EP_EP""" + show_title = "Futurama" + case 2: + db_path = os.path.join("/", "var", "lib", "singerdbs", "parks.db") + db_query = """SELECT DISTINCT(("S" || EP_S || "E" || EP_EP || " " || EP_TITLE)), EP_ID FROM clean_dialog ORDER BY EP_S, EP_EP""" + show_title = "Parks And Rec" + case _: + return { + 'err': True, + 'errorText': 'Unknown endpoint.' + } + async with sqlite3.connect(database=db_path, timeout=1) as _db: + async with _db.execute(db_query) as _cursor: + result = await _cursor.fetchall() + return { + "show_title": show_title, + "episodes": [ + { + 'id': item[1], + 'ep_friendly': item[0] + } for item in result] + } + + async def get_episode_lines_handler(self, data: ValidShowEpisodeLineRequest): + """/get_episode_lines/ + Get lines for a particular episode + """ + return { + 'hi': True + } \ No newline at end of file