#!/usr/bin/env python3.12 import os import random from typing import Union, LiteralString import aiosqlite as sqlite3 from fastapi import FastAPI from fastapi.responses import JSONResponse from .constructors import RandMsgRequest class RandMsg(FastAPI): """ Random Message Endpoint """ def __init__(self, app: FastAPI, util, constants) -> None: # pylint: disable=super-init-not-called self.app: FastAPI = app self.util = util self.constants = constants self.endpoint_name = "randmsg" app.add_api_route(f"/{self.endpoint_name}", self.randmsg_handler, methods=["POST"]) async def randmsg_handler(self, data: RandMsgRequest) -> JSONResponse: """ Get a randomly generated message """ random.seed() short: bool = data.short if data.short else False if not short: db_rand_selected: int = random.choice([0, 1, 3]) else: db_rand_selected = 9 title_attr: str = "Unknown" match db_rand_selected: case 0: randmsg_db_path: Union[str, LiteralString] = os.path.join("/usr/local/share", "sqlite_dbs", "qajoke.db") # For qajoke db db_query: str = "SELECT id, ('Q: ' || question || '
A: ' \ || answer) FROM jokes ORDER BY RANDOM() LIMIT 1" # For qajoke db title_attr = "QA Joke DB" case 1 | 9: randmsg_db_path = os.path.join("/usr/local/share", "sqlite_dbs", "randmsg.db") # For randmsg db db_query = "SELECT id, msg FROM msgs WHERE \ LENGTH(msg) <= 180 ORDER BY RANDOM() LIMIT 1" # For randmsg db if db_rand_selected == 9: db_query = db_query.replace("<= 180", "<= 126") title_attr = "Random Msg DB" case 2: randmsg_db_path = os.path.join("/usr/local/share", "sqlite_dbs", "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("/usr/local/share", "sqlite_dbs", "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("/usr/local/share", "sqlite_dbs", "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("/usr/local/share", "sqlite_dbs", "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" async with sqlite3.connect(database=randmsg_db_path, timeout=1) as _db: async with await _db.execute(db_query) as _cursor: result: sqlite3.Row = await _cursor.fetchone() (result_id, result_msg) = result result_msg = result_msg.strip() return JSONResponse(content= { "id": result_id, "msg": result_msg, "title": title_attr, })