92 lines
4.4 KiB
Python
92 lines
4.4 KiB
Python
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, ('<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 = os.path.join("/usr/local/share",
|
|
"sqlite_dbs",
|
|
"philo.db") # For Philo DB
|
|
db_query = "SELECT id, (content || '<br> - ' || 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, ("<font color='#FF0000'>" || 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 || "<br>" || 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,
|
|
})
|
|
|