104 lines
5.0 KiB
Python
104 lines
5.0 KiB
Python
#!/usr/bin/env python3.12
|
|
|
|
import os
|
|
import random
|
|
from typing import LiteralString
|
|
import aiosqlite as sqlite3
|
|
from fastapi import FastAPI
|
|
from .constructors import RandMsgRequest
|
|
|
|
class RandMsg(FastAPI):
|
|
"""Random Message Endpoint"""
|
|
def __init__(self, app: FastAPI, util, constants, glob_state): # pylint: disable=super-init-not-called
|
|
self.app = app
|
|
self.util = util
|
|
self.constants = constants
|
|
self.glob_state = glob_state
|
|
|
|
self.endpoint_name = "randmsg"
|
|
|
|
app.add_api_route(f"/{self.endpoint_name}", self.randmsg_handler, methods=["POST"])
|
|
|
|
async def randmsg_handler(self, data: RandMsgRequest = None):
|
|
"""
|
|
Get a randomly generated message
|
|
"""
|
|
random.seed()
|
|
short: bool = data.short if data else False
|
|
if not short:
|
|
db_rand_selected = random.choice([0, 1, 3])
|
|
else:
|
|
db_rand_selected = 9
|
|
title_attr = "Unknown"
|
|
|
|
match db_rand_selected:
|
|
case 0:
|
|
randmsg_db_path = os.path.join("/",
|
|
"usr", "local", "share",
|
|
"sqlite_dbs", "qajoke.db") # For qajoke db
|
|
db_query = "SELECT id, ('<b>Q:</b> ' || question || '<br/><b>A:</b> ' \
|
|
|| 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: str|LiteralString = os.path.join("/",
|
|
"usr", "local", "share",
|
|
"sqlite_dbs",
|
|
"philo.db") # For Philo DB
|
|
db_query: str = "SELECT id, (content || '<br> - ' || speaker) FROM quotes \
|
|
ORDER BY RANDOM() LIMIT 1" # For Philo DB
|
|
title_attr: str = "Philosophical Quotes DB"
|
|
case 4:
|
|
randmsg_db_path: str|LiteralString = os.path.join("/",
|
|
"usr", "local", "share",
|
|
"sqlite_dbs",
|
|
"hate.db") # For Hate DB
|
|
db_query: str = """SELECT id, ("<font color='#FF0000'>" || comment) FROM hate_speech \
|
|
WHERE length(comment) <= 180 ORDER BY RANDOM() LIMIT 1"""
|
|
title_attr: str = "Hate Speech DB"
|
|
case 5:
|
|
randmsg_db_path: str|LiteralString = os.path.join("/",
|
|
"usr", "local", "share",
|
|
"sqlite_dbs",
|
|
"rjokes.db") # r/jokes DB
|
|
db_query: str = """SELECT id, (title || "<br>" || body) FROM jokes \
|
|
WHERE score >= 10000 ORDER BY RANDOM() LIMIT 1"""
|
|
title_attr: str = "r/jokes DB"
|
|
case 6:
|
|
randmsg_db_path: str|LiteralString = os.path.join("/",
|
|
"usr", "local", "share",
|
|
"sqlite_dbs",
|
|
"donnies.db") # Donnies DB
|
|
random.seed()
|
|
twilight_or_mice: str = random.choice(["twilight", "mice"])
|
|
db_query: str = f"SELECT id, text FROM {twilight_or_mice} ORDER BY RANDOM() LIMIT 1"
|
|
title_attr: str = "Donnies DB"
|
|
|
|
async with sqlite3.connect(database=randmsg_db_path, timeout=1) as _db:
|
|
async with await _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
|
|
}
|
|
|