import os import random from typing import LiteralString, Optional, Union 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: 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: Optional[RandMsgRequest] = None ) -> JSONResponse: """ Get a randomly generated message - **short**: Optional, if True, will limit length of returned random messages to <=126 characters (Discord restriction related) """ random.seed() short: Optional[bool] = False if isinstance(data, RandMsgRequest): short = data.short if short: db_rand_selected: int = 9 db_rand_selected = random.choice([0, 1, 3]) 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, } )