cleanup
This commit is contained in:
parent
cb4e51ff9c
commit
0d48b754de
@ -1,78 +0,0 @@
|
||||
#!/usr/bin/env python3.12
|
||||
|
||||
import importlib
|
||||
import urllib.parse
|
||||
import regex
|
||||
import logging
|
||||
import json
|
||||
import os
|
||||
import random
|
||||
import aiosqlite as sqlite3
|
||||
|
||||
from typing import Any, Annotated
|
||||
from fastapi import FastAPI, Form, HTTPException
|
||||
from pydantic import BaseModel
|
||||
|
||||
|
||||
class RandMsg(FastAPI):
|
||||
def __init__(self, app: FastAPI, util, constants):
|
||||
self.app = app
|
||||
self.util = util
|
||||
self.constants = constants
|
||||
|
||||
self.endpoint_name = "randmsg"
|
||||
|
||||
app.add_api_route("/%s/" % self.endpoint_name, self.randmsg_handler, methods=["POST"])
|
||||
|
||||
async def randmsg_handler(self):
|
||||
"""
|
||||
Get a randomly generated message
|
||||
"""
|
||||
|
||||
random.seed()
|
||||
db_rand_selected = random.choice([0, 1, 3])
|
||||
title_attr = "Unknown"
|
||||
|
||||
match db_rand_selected:
|
||||
case 0:
|
||||
randmsg_db_path = os.path.join("/", "var", "lib", "singerdbs", "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:
|
||||
randmsg_db_path = os.path.join("/", "var", "lib", "singerdbs", "randmsg.db") # For randmsg db
|
||||
db_query = "SELECT id, msg FROM msgs WHERE LENGTH(msg) <= 180 ORDER BY RANDOM() LIMIT 1" # For randmsg db
|
||||
title_attr = "Random Msg DB"
|
||||
case 2:
|
||||
randmsg_db_path = os.path.join("/", "var", "lib", "singerdbs", "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("/", "var", "lib", "singerdbs", "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("/", "var", "lib", "singerdbs", "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("/", "var", "lib", "singerdbs", "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"
|
||||
case 6:
|
||||
randmsg_db_path = os.path.join("/", "var", "lib", "singerdbs", "donnies.db") # Donnies DB
|
||||
random.seed()
|
||||
twilight_or_mice = random.choice(["twilight", "mice"])
|
||||
db_query = f"SELECT id, text FROM {twilight_or_mice} ORDER BY RANDOM() LIMIT 1"
|
||||
title_attr = "Donnies DB"
|
||||
|
||||
|
||||
async with sqlite3.connect(database=randmsg_db_path, timeout=1) as _db:
|
||||
async with _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
|
||||
}
|
||||
@ -3,11 +3,8 @@
|
||||
import importlib
|
||||
import urllib.parse
|
||||
import regex
|
||||
import logging
|
||||
import json
|
||||
|
||||
from typing import Any, Annotated
|
||||
from fastapi import FastAPI, Form, HTTPException
|
||||
from fastapi import FastAPI, HTTPException
|
||||
from pydantic import BaseModel
|
||||
|
||||
|
||||
@ -17,7 +14,7 @@ class ValidLyricRequest(BaseModel):
|
||||
- **s**: song
|
||||
- **t**: track (artist and song combined) [used only if a & s are not used]
|
||||
- **extra**: include extra details in response [optional, default: false]
|
||||
- **sub**: text to search within lyrics, if found lyrics will begin at found verse [optional, default: none]
|
||||
- **sub**: text to search within lyrics, if found lyrics will begin at found verse [optional]
|
||||
- **src**: the script/utility which initiated the request
|
||||
"""
|
||||
|
||||
@ -28,7 +25,7 @@ class ValidLyricRequest(BaseModel):
|
||||
extra: bool | None = False
|
||||
src: str
|
||||
|
||||
class Config:
|
||||
class Config: # pylint: disable=missing-class-docstring too-few-public-methods
|
||||
schema_extra = {
|
||||
"example": {
|
||||
"a": "eminem",
|
||||
@ -39,7 +36,8 @@ class ValidLyricRequest(BaseModel):
|
||||
}
|
||||
|
||||
class LyricSearch(FastAPI):
|
||||
def __init__(self, app: FastAPI, util, constants):
|
||||
"""Lyric Search Endpoint"""
|
||||
def __init__(self, app: FastAPI, util, constants): # pylint: disable=super-init-not-called
|
||||
self.app = app
|
||||
self.util = util
|
||||
self.constants = constants
|
||||
@ -56,7 +54,7 @@ class LyricSearch(FastAPI):
|
||||
"LIMNORIA-SHARED"
|
||||
]
|
||||
|
||||
app.add_api_route("/%s/" % self.endpoint_name, self.lyric_search_handler, methods=["POST"])
|
||||
app.add_api_route("/%s/" % self.endpoint_name, self.lyric_search_handler, methods=["POST"]) # pylint: disable=consider-using-f-string trailing-whitespace
|
||||
|
||||
async def lyric_search_handler(self, data: ValidLyricRequest):
|
||||
"""
|
||||
@ -71,49 +69,48 @@ class LyricSearch(FastAPI):
|
||||
"""
|
||||
|
||||
src = data.src.upper()
|
||||
if not(src in self.acceptable_request_sources):
|
||||
if not src in self.acceptable_request_sources:
|
||||
raise HTTPException(detail="Invalid request source", status_code=403)
|
||||
|
||||
search_artist = data.a
|
||||
search_song = data.s
|
||||
search_text = data.t
|
||||
add_extras = data.extra
|
||||
sub_search = data.sub
|
||||
search_object = None
|
||||
|
||||
searchArtist = data.a
|
||||
searchSong = data.s
|
||||
searchText = data.t
|
||||
addExtras = data.extra
|
||||
subSearch = data.sub
|
||||
searchObject = None
|
||||
|
||||
random_song_requested = (searchArtist == "!" and searchSong == "!")
|
||||
random_song_requested = (search_artist == "!" and search_song == "!")
|
||||
query_valid = (
|
||||
not(searchArtist is None) and
|
||||
not(searchSong is None) and
|
||||
len(searchArtist) >= 1 and
|
||||
len(searchSong) >= 1 and
|
||||
len(searchArtist) + len(searchSong) >= 3
|
||||
not(search_artist is None) and
|
||||
not(search_song is None) and
|
||||
len(search_artist) >= 1 and
|
||||
len(search_song) >= 1 and
|
||||
len(search_artist) + len(search_song) >= 3
|
||||
)
|
||||
|
||||
if not(random_song_requested) and not(searchText) and not(query_valid):
|
||||
if not random_song_requested and (not search_text and not query_valid):
|
||||
return {
|
||||
"err": True,
|
||||
"errorText": "Invalid parameters"
|
||||
}
|
||||
if search_artist and search_song:
|
||||
search_artist = self.constants.DOUBLE_SPACE_REGEX.sub(" ", search_artist.strip())
|
||||
search_song = self.constants.DOUBLE_SPACE_REGEX.sub(" ", search_song.strip())
|
||||
search_artist = urllib.parse.unquote(search_artist)
|
||||
search_song = urllib.parse.unquote(search_song)
|
||||
|
||||
if searchArtist and searchSong:
|
||||
searchArtist = self.constants.DOUBLE_SPACE_REGEX.sub(" ", searchArtist.strip())
|
||||
searchSong = self.constants.DOUBLE_SPACE_REGEX.sub(" ", searchSong.strip())
|
||||
searchArtist = urllib.parse.unquote(searchArtist)
|
||||
searchSong = urllib.parse.unquote(searchSong)
|
||||
|
||||
if searchText is None:
|
||||
searchObject = self.lyrics_engine.create_query_object("%s : %s" % (searchArtist, searchSong))
|
||||
if subSearch:
|
||||
searchObject = self.lyrics_engine.create_query_object("%s : %s : %s" % (searchArtist, searchSong, subSearch))
|
||||
if search_text is None:
|
||||
# pylint: disable=consider-using-f-string
|
||||
search_object = self.lyrics_engine.create_query_object("%s : %s" % (search_artist, search_song))
|
||||
if sub_search:
|
||||
search_object = self.lyrics_engine.create_query_object("%s : %s : %s" % (search_artist, search_song, sub_search))
|
||||
else:
|
||||
searchObject = self.lyrics_engine.create_query_object(str(searchText))
|
||||
search_object = self.lyrics_engine.create_query_object(str(search_text))
|
||||
|
||||
searchWorker = await self.lyrics_engine.lyrics_worker(searching=searchObject,
|
||||
search_worker = await self.lyrics_engine.lyrics_worker(searching=search_object,
|
||||
recipient='anyone')
|
||||
|
||||
if not(searchWorker) or not('l') in searchWorker.keys():
|
||||
if not search_worker or not 'l' in search_worker.keys():
|
||||
return {
|
||||
'err': True,
|
||||
'errorText': 'Sources exhausted, lyrics not located.'
|
||||
@ -121,12 +118,10 @@ class LyricSearch(FastAPI):
|
||||
|
||||
return {
|
||||
'err': False,
|
||||
'artist': searchWorker['artist'],
|
||||
'song': searchWorker['song'],
|
||||
'combo_lev': f'{searchWorker['combo_lev']:.2f}',
|
||||
'lyrics': regex.sub(r"\s/\s", "<br>", " ".join(searchWorker['l'])),
|
||||
'from_cache': searchWorker['method'].strip().lower().startswith("local cache"),
|
||||
'src': searchWorker['method'] if addExtras else None,
|
||||
'artist': search_worker['artist'],
|
||||
'song': search_worker['song'],
|
||||
'combo_lev': f'{search_worker['combo_lev']:.2f}',
|
||||
'lyrics': regex.sub(r"\s/\s", "<br>", " ".join(search_worker['l'])),
|
||||
'from_cache': search_worker['method'].strip().lower().startswith("local cache"),
|
||||
'src': search_worker['method'] if add_extras else None,
|
||||
}
|
||||
|
||||
|
||||
105
endpoints/rand_msg.py
Normal file
105
endpoints/rand_msg.py
Normal file
@ -0,0 +1,105 @@
|
||||
#!/usr/bin/env python3.12
|
||||
|
||||
import os
|
||||
import random
|
||||
import aiosqlite as sqlite3
|
||||
|
||||
from fastapi import FastAPI
|
||||
|
||||
|
||||
class RandMsg(FastAPI):
|
||||
"""Random Message Endpoint"""
|
||||
def __init__(self, app: FastAPI, util, constants): # pylint: disable=super-init-not-called
|
||||
self.app = app
|
||||
self.util = util
|
||||
self.constants = constants
|
||||
|
||||
self.endpoint_name = "randmsg"
|
||||
|
||||
app.add_api_route("/%s/" % self.endpoint_name, self.randmsg_handler, methods=["POST"])
|
||||
|
||||
async def randmsg_handler(self):
|
||||
"""
|
||||
Get a randomly generated message
|
||||
"""
|
||||
random.seed()
|
||||
db_rand_selected = random.choice([0, 1, 3])
|
||||
title_attr = "Unknown"
|
||||
|
||||
match db_rand_selected:
|
||||
case 0:
|
||||
randmsg_db_path = os.path.join("/",
|
||||
"var",
|
||||
"lib",
|
||||
"singerdbs",
|
||||
"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:
|
||||
randmsg_db_path = os.path.join("/",
|
||||
"var",
|
||||
"lib",
|
||||
"singerdbs",
|
||||
"randmsg.db") # For randmsg db
|
||||
db_query = "SELECT id, msg FROM msgs WHERE" \
|
||||
"LENGTH(msg) <= 180 ORDER BY RANDOM() LIMIT 1" # For randmsg db
|
||||
title_attr = "Random Msg DB"
|
||||
case 2:
|
||||
randmsg_db_path = os.path.join("/",
|
||||
"var",
|
||||
"lib",
|
||||
"singerdbs",
|
||||
"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("/",
|
||||
"var",
|
||||
"lib",
|
||||
"singerdbs",
|
||||
"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("/",
|
||||
"var",
|
||||
"lib",
|
||||
"singerdbs",
|
||||
"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("/",
|
||||
"var",
|
||||
"lib",
|
||||
"singerdbs",
|
||||
"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"
|
||||
case 6:
|
||||
randmsg_db_path = os.path.join("/",
|
||||
"var",
|
||||
"lib",
|
||||
"singerdbs",
|
||||
"donnies.db") # Donnies DB
|
||||
random.seed()
|
||||
twilight_or_mice = random.choice(["twilight", "mice"])
|
||||
db_query = f"SELECT id, text FROM {twilight_or_mice} ORDER BY RANDOM() LIMIT 1"
|
||||
title_attr = "Donnies DB"
|
||||
|
||||
async with sqlite3.connect(database=randmsg_db_path, timeout=1) as _db:
|
||||
async with _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
|
||||
}
|
||||
|
||||
4
main.py
4
main.py
@ -55,8 +55,8 @@ End Blacklisted Routes
|
||||
Actionable Routes
|
||||
"""
|
||||
|
||||
randmsg_endpoint = importlib.import_module("endpoints.RandMsg").RandMsg(app, util, constants)
|
||||
lyric_search_endpoint = importlib.import_module("endpoints.LyricSearch").LyricSearch(app, util, constants)
|
||||
randmsg_endpoint = importlib.import_module("endpoints.rand_msg").RandMsg(app, util, constants)
|
||||
lyric_search_endpoint = importlib.import_module("endpoints.lyric_search").LyricSearch(app, util, constants)
|
||||
|
||||
|
||||
|
||||
|
||||
Loading…
x
Reference in New Issue
Block a user