codey/discord-havoc
1
0
Fork 0
Code Issues Pull Requests Wiki Activity

reformat (black)

Browse Source
This commit is contained in:
codey
2025-04-17 14:35:56 -04:00
parent d12b066c8e
commit 1bb482315e
20 changed files with 1928 additions and 1326 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
util/__init__.py
View File

@@ -1,4 +1,4 @@
import importlib
from . import discord_helpers
importlib.reload(discord_helpers)
importlib.reload(discord_helpers)

63
util/catbox.py
View File

@@ -12,35 +12,36 @@ Catbox Uploader (Async)
catbox_api_url: str = "https://catbox.moe/user/api.php"
http_headers: dict[str, str] = {
'accept': '*/*',
'User-Agent': 'Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/101.0.4951.64 Safari/537.36 Edg/101.0.1210.53',
'Accept-Language': 'en-US,en;q=0.9,it;q=0.8,es;q=0.7',
'referer': 'https://www.google.com/',
'cookie': 'DSID=AAO-7r4OSkS76zbHUkiOpnI0kk-X19BLDFF53G8gbnd21VZV2iehu-w_2v14cxvRvrkd_NjIdBWX7wUiQ66f-D8kOkTKD1BhLVlqrFAaqDP3LodRK2I0NfrObmhV9HsedGE7-mQeJpwJifSxdchqf524IMh9piBflGqP0Lg0_xjGmLKEQ0F4Na6THgC06VhtUG5infEdqMQ9otlJENe3PmOQTC_UeTH5DnENYwWC8KXs-M4fWmDADmG414V0_X0TfjrYu01nDH2Dcf3TIOFbRDb993g8nOCswLMi92LwjoqhYnFdf1jzgK0'
}
"accept": "*/*",
"User-Agent": "Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/101.0.4951.64 Safari/537.36 Edg/101.0.1210.53",
"Accept-Language": "en-US,en;q=0.9,it;q=0.8,es;q=0.7",
"referer": "https://www.google.com/",
"cookie": "DSID=AAO-7r4OSkS76zbHUkiOpnI0kk-X19BLDFF53G8gbnd21VZV2iehu-w_2v14cxvRvrkd_NjIdBWX7wUiQ66f-D8kOkTKD1BhLVlqrFAaqDP3LodRK2I0NfrObmhV9HsedGE7-mQeJpwJifSxdchqf524IMh9piBflGqP0Lg0_xjGmLKEQ0F4Na6THgC06VhtUG5infEdqMQ9otlJENe3PmOQTC_UeTH5DnENYwWC8KXs-M4fWmDADmG414V0_X0TfjrYu01nDH2Dcf3TIOFbRDb993g8nOCswLMi92LwjoqhYnFdf1jzgK0",
}
class CatboxAsync:
def __init__(self) -> None:
self.catbox_api_url = catbox_api_url
self.headers = http_headers
def generateRandomFileName(self, fileExt: Optional[str] = None) -> str:
"""
Generate random file name
Args:
fileExt (Optional[str]): File extension to use for naming
Returns:
str
"""
if not fileExt:
fileExt = 'png'
fileExt = "png"
return f"{random.getrandbits(32)}.{fileExt}"
async def upload(self, file: str) -> Optional[str]:
"""
Upload file to catbox
Args:
file (str): Path of file to be uploaded
Returns:
@@ -49,33 +50,33 @@ class CatboxAsync:
try:
if not file:
return None
if not(os.path.exists(file)):
logging.critical("Could not find %s",
file)
if not (os.path.exists(file)):
logging.critical("Could not find %s", file)
return None
fileExt: Optional[str] = None
if file.find(".") > 0:
fileExt = "".join(file.split(".")[-1:])
with open(file, 'rb') as fileContents:
with open(file, "rb") as fileContents:
post_data: FormData = FormData()
post_data.add_field(name="reqtype",
value="fileupload")
post_data.add_field(name="userhash",
value="")
post_data.add_field(name="reqtype", value="fileupload")
post_data.add_field(name="userhash", value="")
with magic.Magic(flags=magic.MAGIC_MIME) as m:
content_type = m.id_filename(file)
post_data.add_field(name="fileToUpload",
value=fileContents,
filename=self.generateRandomFileName(fileExt),
content_type=content_type
)
post_data.add_field(
name="fileToUpload",
value=fileContents,
filename=self.generateRandomFileName(fileExt),
content_type=content_type,
)
async with ClientSession() as session:
async with await session.post(self.catbox_api_url,
headers=self.headers,
data=post_data,
timeout=ClientTimeout(connect=10, sock_read=10)) as request:
async with await session.post(
self.catbox_api_url,
headers=self.headers,
data=post_data,
timeout=ClientTimeout(connect=10, sock_read=10),
) as request:
request.raise_for_status()
return await request.text()
except:
@@ -85,4 +86,4 @@ class CatboxAsync:
try:
fileContents.close()
except:
return None
return None

26
util/discord_helpers.py
View File

@@ -6,11 +6,13 @@ from typing import Optional, Any
Discord Helper Methods
"""
async def get_channel_by_name(bot: discord.Bot, channel: str,
guild: int | None = None) -> Optional[Any]: # Optional[Any] used as pycord channel types can be ambigious
async def get_channel_by_name(
bot: discord.Bot, channel: str, guild: int | None = None
) -> Optional[Any]: # Optional[Any] used as pycord channel types can be ambigious
"""
Get Channel by Name
Args:
bot (discord.Bot)
channel (str)
@@ -18,10 +20,9 @@ async def get_channel_by_name(bot: discord.Bot, channel: str,
Returns:
Optional[Any]
"""
channel = re.sub(r'^#', '', channel.strip())
channel = re.sub(r"^#", "", channel.strip())
if not guild:
return discord.utils.get(bot.get_all_channels(),
name=channel)
return discord.utils.get(bot.get_all_channels(), name=channel)
else:
_guild: Optional[discord.Guild] = bot.get_guild(guild)
if not _guild:
@@ -32,12 +33,14 @@ async def get_channel_by_name(bot: discord.Bot, channel: str,
return _channel
return None
async def send_message(bot: discord.Bot, channel: str,
message: str, guild: int | None = None) -> None:
async def send_message(
bot: discord.Bot, channel: str, message: str, guild: int | None = None
) -> None:
"""
Send Message to the provided channel. If guild is provided, will limit to channels within that guild to ensure the correct
channel is selected. Useful in the event a channel exists in more than one guild that the bot resides in.
Args:
bot (discord.Bot)
channel (str)
@@ -50,9 +53,8 @@ async def send_message(bot: discord.Bot, channel: str,
channel_int: int = int(channel)
_channel = bot.get_channel(channel_int)
else:
channel = re.sub(r'^#', '', channel.strip())
_channel = await get_channel_by_name(bot=bot,
channel=channel, guild=guild)
channel = re.sub(r"^#", "", channel.strip())
_channel = await get_channel_by_name(bot=bot, channel=channel, guild=guild)
if not isinstance(_channel, discord.TextChannel):
return None
await _channel.send(message)

67
util/jesusmemes.py
View File

@@ -1,9 +1,9 @@
import aiohttp
import aiohttp
from typing import Optional
import regex
import regex
from regex import Pattern
import os
import random
import os
import random
import logging
import traceback
from util.catbox import CatboxAsync
@@ -13,61 +13,64 @@ Jesus Meme Generator
(requires Catbox uploader)
"""
class JesusMemeGenerator():
class JesusMemeGenerator:
def __init__(self) -> None:
self.MEMEAPIURL = "https://apimeme.com/meme?meme="
self.MEMESTORAGEDIR = os.path.join(os.path.expanduser("~"),
"memes")
self.MEMESTORAGEDIR = os.path.join(os.path.expanduser("~"), "memes")
self.top_line_regex: Pattern = regex.compile(
r'[^\p{Letter}\p{Number}\p{Punctuation}\p{Horiz_Space}\p{Currency_Symbol}]'
)
r"[^\p{Letter}\p{Number}\p{Punctuation}\p{Horiz_Space}\p{Currency_Symbol}]"
)
self.bottom_line_regex: Pattern = regex.compile(
r'[^\p{Letter}\p{Number}\p{Punctuation}\p{Horiz_Space}\p{Currency_Symbol}]'
)
self.url_regex_1: Pattern = regex.compile(
r'\p{Horiz_Space}')
self.url_regex_2: Pattern = regex.compile(
r'#')
async def create_meme(self, top_line: str, bottom_line: str,
meme="Jesus-Talking-To-Cool-Dude") -> Optional[str]:
r"[^\p{Letter}\p{Number}\p{Punctuation}\p{Horiz_Space}\p{Currency_Symbol}]"
)
self.url_regex_1: Pattern = regex.compile(r"\p{Horiz_Space}")
self.url_regex_2: Pattern = regex.compile(r"#")
async def create_meme(
self, top_line: str, bottom_line: str, meme="Jesus-Talking-To-Cool-Dude"
) -> Optional[str]:
"""
Create Meme
Args:
top_line (str): Top line of meme
bottom_line (str): Bottom line of meme
meme (str): The meme to use, defaults to Jesus-Talking-To-Cool-Dude
Returns:
Optional[str]
"""
try:
if not top_line or not bottom_line:
return None
top_line = self.top_line_regex.sub('',
top_line.strip())
bottom_line = self.bottom_line_regex.sub('',
bottom_line.strip())
top_line = self.top_line_regex.sub("", top_line.strip())
bottom_line = self.bottom_line_regex.sub("", bottom_line.strip())
out_fname: Optional[str] = None
if len(top_line) < 1 or len(bottom_line) < 1:
return None
formed_url: str = f"{self.MEMEAPIURL}{meme}&top={top_line.strip()}&bottom={bottom_line.strip()}"
formed_url = self.url_regex_1.sub('+', self.url_regex_2.sub('%23', formed_url.strip()))
formed_url: str = (
f"{self.MEMEAPIURL}{meme}&top={top_line.strip()}&bottom={bottom_line.strip()}"
)
formed_url = self.url_regex_1.sub(
"+", self.url_regex_2.sub("%23", formed_url.strip())
)
timeout = aiohttp.ClientTimeout(total=15)
async with aiohttp.ClientSession(timeout=timeout) as session:
async with session.get(formed_url) as response:
UUID = f"{random.getrandbits(8)}-{random.getrandbits(8)}"
out_fname = f"{UUID}.jpg"
with open(f"{self.MEMESTORAGEDIR}/{out_fname}", 'wb') as f:
with open(f"{self.MEMESTORAGEDIR}/{out_fname}", "wb") as f:
f.write(await response.read())
if not out_fname:
uploader = CatboxAsync()
meme_link: Optional[str] = await uploader.upload(f"{self.MEMESTORAGEDIR}/{out_fname}")
meme_link: Optional[str] = await uploader.upload(
f"{self.MEMESTORAGEDIR}/{out_fname}"
)
if not meme_link:
logging.info("Meme upload failed!")
return None

69
util/litterbox.py
View File

@@ -13,37 +13,38 @@ import os
litterbox_api_url: str = "https://litterbox.catbox.moe/resources/internals/api.php"
http_headers: dict[str, str] = {
'accept': '*/*',
'User-Agent': 'Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/101.0.4951.64 Safari/537.36 Edg/101.0.1210.53',
'Accept-Language': 'en-US,en;q=0.9,it;q=0.8,es;q=0.7',
'referer': 'https://www.google.com/',
'cookie': 'DSID=AAO-7r4OSkS76zbHUkiOpnI0kk-X19BLDFF53G8gbnd21VZV2iehu-w_2v14cxvRvrkd_NjIdBWX7wUiQ66f-D8kOkTKD1BhLVlqrFAaqDP3LodRK2I0NfrObmhV9HsedGE7-mQeJpwJifSxdchqf524IMh9piBflGqP0Lg0_xjGmLKEQ0F4Na6THgC06VhtUG5infEdqMQ9otlJENe3PmOQTC_UeTH5DnENYwWC8KXs-M4fWmDADmG414V0_X0TfjrYu01nDH2Dcf3TIOFbRDb993g8nOCswLMi92LwjoqhYnFdf1jzgK0'
}
"accept": "*/*",
"User-Agent": "Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/101.0.4951.64 Safari/537.36 Edg/101.0.1210.53",
"Accept-Language": "en-US,en;q=0.9,it;q=0.8,es;q=0.7",
"referer": "https://www.google.com/",
"cookie": "DSID=AAO-7r4OSkS76zbHUkiOpnI0kk-X19BLDFF53G8gbnd21VZV2iehu-w_2v14cxvRvrkd_NjIdBWX7wUiQ66f-D8kOkTKD1BhLVlqrFAaqDP3LodRK2I0NfrObmhV9HsedGE7-mQeJpwJifSxdchqf524IMh9piBflGqP0Lg0_xjGmLKEQ0F4Na6THgC06VhtUG5infEdqMQ9otlJENe3PmOQTC_UeTH5DnENYwWC8KXs-M4fWmDADmG414V0_X0TfjrYu01nDH2Dcf3TIOFbRDb993g8nOCswLMi92LwjoqhYnFdf1jzgK0",
}
class LitterboxAsync:
def __init__(self) -> None:
self.headers: dict[str, str] = http_headers
self.api_path = litterbox_api_url
def generateRandomFileName(self, fileExt: Optional[str] = None) -> str:
"""
Generate Random Filename
Args:
fileExt (Optional[str]): File extension to use for naming
Returns:
str
"""
if not fileExt:
fileExt = 'png'
fileExt = "png"
return f"{random.getrandbits(32)}.{fileExt}"
async def upload(self,
file: Union[str, bytes, BufferedReader],
time='1h') -> Optional[str]:
async def upload(
self, file: Union[str, bytes, BufferedReader], time="1h"
) -> Optional[str]:
"""
Upload File to Litterbox
Args:
file (Union[str, bytes, BufferedReader]): File to upload (accepts either filepath or io.BufferedReader)
time (str): Expiration time, default: 1h
@@ -59,38 +60,38 @@ class LitterboxAsync:
return None
if isinstance(file, BufferedReader):
file = file.read()
fileExt: str = 'png'
fileExt: str = "png"
if isinstance(file, str):
if file.find(".") > 0:
fileExt = "".join(file.split(".")[-1:])
file = open(file, 'rb').read()
file = open(file, "rb").read()
with magic.Magic(flags=magic.MAGIC_MIME) as m:
if isinstance(file, BufferedReader):
content_type = str(m.id_buffer(file))
else:
content_type = str(m.id_filename(file))
content_type = str(m.id_filename(file))
post_data: FormData = FormData()
post_data.add_field(name="reqtype",
value="fileupload")
post_data.add_field(name="userhash",
value="")
post_data.add_field(name="time",
value=time)
post_data.add_field(name="fileToUpload",
value=file,
filename=self.generateRandomFileName(fileExt),
content_type=content_type
post_data.add_field(name="reqtype", value="fileupload")
post_data.add_field(name="userhash", value="")
post_data.add_field(name="time", value=time)
post_data.add_field(
name="fileToUpload",
value=file,
filename=self.generateRandomFileName(fileExt),
content_type=content_type,
)
async with ClientSession() as session:
async with await session.post(self.api_path,
headers=self.headers,
data=post_data,
timeout=ClientTimeout(connect=5, sock_read=70)) as request:
async with await session.post(
self.api_path,
headers=self.headers,
data=post_data,
timeout=ClientTimeout(connect=5, sock_read=70),
) as request:
request.raise_for_status()
return await request.text()
except:

107
util/lovehate_db.py
View File

@@ -4,24 +4,28 @@ from typing import LiteralString, Optional, Union
import aiosqlite as sqlite3
from constructors import LoveHateException
class DB:
"""LoveHate DB Utility Class"""
def __init__(self, bot) -> None:
self.db_path: str|LiteralString = os.path.join("/usr/local/share",
"sqlite_dbs", "lovehate.db")
async def get_wholovehates(self, thing: str, loves: bool = False,
hates: bool = False) -> Union[list[tuple], bool]:
self.db_path: str | LiteralString = os.path.join(
"/usr/local/share", "sqlite_dbs", "lovehate.db"
)
async def get_wholovehates(
self, thing: str, loves: bool = False, hates: bool = False
) -> Union[list[tuple], bool]:
"""
Get a list of users who have professed their love OR hatred for <thing>
Args:
thing (str): The <thing> to check
loves (bool): Are we looking for loves?
hates (bool): ...or are we looking for hates?
Returns:
Union[list[tuple], bool]
"""
"""
query: str = "SELECT display_name FROM lovehate WHERE thing LIKE ? AND flag = ?"
params: tuple = tuple()
@@ -35,8 +39,11 @@ class DB:
flag = 1
elif not hates and not loves:
raise LoveHateException("Neither loves nor hates were requested")
params = (thing, flag,)
params = (
thing,
flag,
)
async with sqlite3.connect(self.db_path, timeout=2) as db_conn:
async with await db_conn.execute(query, params) as db_cursor:
result: list[tuple] = await db_cursor.fetchall()
@@ -44,28 +51,35 @@ class DB:
if not result:
return False
return result
async def get_lovehates(self, loves: bool = False, hates: bool = False,
user: Optional[str] = None, thing: Optional[str] = None) -> Union[list[tuple], bool]:
async def get_lovehates(
self,
loves: bool = False,
hates: bool = False,
user: Optional[str] = None,
thing: Optional[str] = None,
) -> Union[list[tuple], bool]:
"""
Get a list of either 1) what {user} loves/hates, or who loves/hates {thing}, depending on bools loves, hates
Args:
loves (bool): Are we looking for loves?
hates (bool): ...OR are we looking for hates?
user (Optional[str]): the user to query against
thing (Optional[str]): ... OR the thing to query against
Returns:
Union[list[tuple], bool]
"""
query: str = ""
params: tuple = tuple()
if not user and not thing:
raise LoveHateException("Neither a <user> or <thing> was specified to query against")
raise LoveHateException(
"Neither a <user> or <thing> was specified to query against"
)
flag: Optional[int] = None
if hates and loves:
@@ -79,10 +93,16 @@ class DB:
if user:
query = "SELECT thing FROM lovehate WHERE display_name LIKE ? AND flag == ?"
params = (user, flag,)
params = (
user,
flag,
)
elif thing:
query = "SELECT display_name FROM lovehate WHERE thing LIKE ? AND flag == ?"
params = (thing, flag,)
params = (
thing,
flag,
)
async with sqlite3.connect(self.db_path, timeout=2) as db_conn:
async with await db_conn.execute(query, params) as db_cursor:
@@ -91,22 +111,27 @@ class DB:
return False
return result
async def check_existence(self, user: str, thing: str) -> Optional[int]:
"""
Determine whether a user is opinionated on a <thing>
Args:
user (str): The user to check
thing (str): The thing to check if the user has an opinion on
Returns:
Optional[int]
"""
params = (user, thing,)
params = (
user,
thing,
)
async with sqlite3.connect(self.db_path, timeout=2) as db_conn:
async with await db_conn.execute("SELECT id, flag FROM lovehate WHERE display_name LIKE ? AND thing LIKE ?", params) as db_cursor:
async with await db_conn.execute(
"SELECT id, flag FROM lovehate WHERE display_name LIKE ? AND thing LIKE ?",
params,
) as db_cursor:
result = await db_cursor.fetchone()
if not result:
return None
@@ -116,29 +141,38 @@ class DB:
async def update(self, user: str, thing: str, flag: int) -> str:
"""
Updates the lovehate database, and returns an appropriate response
Args:
user (str): The user to update
thing (str): The thing the user loves/hates/doesn't care about anymore
flag (int): int representation of love (1), hate (-1), and dontcare (0)
Returns:
Returns:
str
"""
if not flag in range(-1, 2):
raise LoveHateException(f"Invalid flag {flag} specified, is this love (1), hate (-1), or dontcare? (0)")
raise LoveHateException(
f"Invalid flag {flag} specified, is this love (1), hate (-1), or dontcare? (0)"
)
db_query: str = ""
params: tuple = (user, thing,)
params: tuple = (
user,
thing,
)
already_opinionated: Optional[int] = await self.check_existence(user, thing)
if already_opinionated:
if already_opinionated:
if flag == 0:
db_query = "DELETE FROM lovehate WHERE display_name LIKE ? AND thing LIKE ?"
db_query = (
"DELETE FROM lovehate WHERE display_name LIKE ? AND thing LIKE ?"
)
else:
loves_or_hates: str = "loves"
if already_opinionated == -1:
loves_or_hates = "hates"
raise LoveHateException(f"But {user} already {loves_or_hates} {thing}...")
raise LoveHateException(
f"But {user} already {loves_or_hates} {thing}..."
)
else:
match flag:
case -1:
@@ -147,19 +181,22 @@ class DB:
db_query = "INSERT INTO lovehate(display_name, flag, thing) VALUES(?, 1, ?)"
case _:
raise LoveHateException("Unknown error, default case matched")
async with sqlite3.connect(self.db_path, timeout=2) as db_conn:
async with await db_conn.execute(db_query, params) as db_cursor:
await db_conn.commit()
if db_cursor.rowcount != 1:
raise LoveHateException(f"DB Error - RowCount: {db_cursor.rowcount} for INSERT query")
raise LoveHateException(
f"DB Error - RowCount: {db_cursor.rowcount} for INSERT query"
)
match flag:
case -1:
return f"We're done here, {user} hates {thing}."
case 0:
case 0:
return f"We're done here, {user} no longer cares one way or the other about {thing}."
case 1:
return f"We're done here, {user} loves {thing}."
case _:
raise LoveHateException("Unknown error, default case matched [2]")
raise LoveHateException(
"Unknown error, default case matched [2]"
)

464
util/misc_util.py
View File

@@ -11,144 +11,205 @@ from aiohttp import ClientSession, ClientTimeout
from bohancompliment import ComplimentGenerator
from discord import Embed
class Util:
"""Misc Utility"""
def __init__(self) -> None:
self.URL_URBANDICTIONARY: str = "http://api.urbandictionary.com/v0/define"
self.URL_INSULTAPI: str = "https://insult.mattbas.org/api/insult"
self.COMPLIMENT_GENERATOR = ComplimentGenerator()
self.dbs: dict[str, str|LiteralString] = {
'whisky': os.path.join("/usr/local/share",
"sqlite_dbs", "whiskey.db"),
'drinks': os.path.join("/usr/local/share",
"sqlite_dbs", "cocktails.db"),
'strains': os.path.join("/usr/local/share",
"sqlite_dbs", "strains.db"),
'qajoke': os.path.join("/usr/local/share",
"sqlite_dbs", "qajoke.db"),
'rjokes': os.path.join("/usr/local/share",
"sqlite_dbs", "rjokes.db"),
'randmsg': os.path.join("/usr/local/share",
"sqlite_dbs", "randmsg.db"),
'stats': os.path.join("/usr/local/share",
"sqlite_dbs", "havoc_stats.db"),
'cookies': os.path.join("/usr/local/share",
"sqlite_dbs", "cookies.db"),
self.dbs: dict[str, str | LiteralString] = {
"whisky": os.path.join("/usr/local/share", "sqlite_dbs", "whiskey.db"),
"drinks": os.path.join("/usr/local/share", "sqlite_dbs", "cocktails.db"),
"strains": os.path.join("/usr/local/share", "sqlite_dbs", "strains.db"),
"qajoke": os.path.join("/usr/local/share", "sqlite_dbs", "qajoke.db"),
"rjokes": os.path.join("/usr/local/share", "sqlite_dbs", "rjokes.db"),
"randmsg": os.path.join("/usr/local/share", "sqlite_dbs", "randmsg.db"),
"stats": os.path.join("/usr/local/share", "sqlite_dbs", "havoc_stats.db"),
"cookies": os.path.join("/usr/local/share", "sqlite_dbs", "cookies.db"),
}
self.COFFEES: list = ['a cup of french-pressed coffee', 'a cup of cold brew', 'a cup of flash brew',
'a cup of Turkish coffee', 'a cup of Moka', 'an espresso',
'a cup of Nescafe coffee',
'an iced coffee', 'a Frappé', 'a freddo cappuccino',
'a cup of Chock full o\'Nuts', 'a cup of Folgers', 'a cup of Lavazza',
'a cup of Maxwell House', 'a cup of Moccona', 'a cup of Mr. Brown Coffee',
'a cup of affogato al caffè',
'a cup of Caffè Medici', 'a cup of Café Touba',
'a double-double', 'an indian filter coffee', 'a cup of pocillo',
'a cup of caffè americano', 'a cup of caffè lungo', 'a latte', 'a manilo',
'a flat white', 'a cup of café cubano', 'a cup of caffè crema',
'a cup of cafe zorro', 'an espresso roberto', 'an espresso romano',
'an espresso sara', 'a guillermo', 'a ristretto', 'a cup of melya',
'a cup of caffè marocchino', 'a cup of café miel', 'a cup of café de olla',
'a Mazagran', 'a Palazzo', 'an ice shot', 'a macchiato',
'a cortado', 'a red eye', 'a cappuccino',
'a mocha', 'a café au lait', 'a bicerin',
'a caffè corretto', 'a ca phe trung', 'a café bombón', 'a Vienna coffee',
'a flat black', 'a lungo', 'a doppio', 'a ristretto bianco', 'a piccolo latte',
'a gibraltar', 'a breve', 'a café con leche', 'a su café', 'a café del tiempo',
'a java chip frappuccino', 'a pumpkin spice latte', 'a caramel macchiato',
'a white chocolate mocha', 'a hazelnut coffee', 'a toffee nut latte',
'a peppermint mocha', 'a cinnamon dolce latte', 'a coconut milk latte',
'an almond milk cappuccino', 'an oat milk latte', 'a caramel frappuccino',
'a chocolate frappuccino', 'a butter pecan coffee', 'a maple pecan latte',
'a sea salt caramel mocha', 'a nitro cold brew', 'a pumpkin cold brew',
'a honey almond flat white', 'a sweet cream cold brew', 'a matcha latte',
'a golden latte', 'a turmeric latte', 'a beetroot latte', 'a kopi luwak']
self.COFFEES: list = [
"a cup of french-pressed coffee",
"a cup of cold brew",
"a cup of flash brew",
"a cup of Turkish coffee",
"a cup of Moka",
"an espresso",
"a cup of Nescafe coffee",
"an iced coffee",
"a Frappé",
"a freddo cappuccino",
"a cup of Chock full o'Nuts",
"a cup of Folgers",
"a cup of Lavazza",
"a cup of Maxwell House",
"a cup of Moccona",
"a cup of Mr. Brown Coffee",
"a cup of affogato al caffè",
"a cup of Caffè Medici",
"a cup of Café Touba",
"a double-double",
"an indian filter coffee",
"a cup of pocillo",
"a cup of caffè americano",
"a cup of caffè lungo",
"a latte",
"a manilo",
"a flat white",
"a cup of café cubano",
"a cup of caffè crema",
"a cup of cafe zorro",
"an espresso roberto",
"an espresso romano",
"an espresso sara",
"a guillermo",
"a ristretto",
"a cup of melya",
"a cup of caffè marocchino",
"a cup of café miel",
"a cup of café de olla",
"a Mazagran",
"a Palazzo",
"an ice shot",
"a macchiato",
"a cortado",
"a red eye",
"a cappuccino",
"a mocha",
"a café au lait",
"a bicerin",
"a caffè corretto",
"a ca phe trung",
"a café bombón",
"a Vienna coffee",
"a flat black",
"a lungo",
"a doppio",
"a ristretto bianco",
"a piccolo latte",
"a gibraltar",
"a breve",
"a café con leche",
"a su café",
"a café del tiempo",
"a java chip frappuccino",
"a pumpkin spice latte",
"a caramel macchiato",
"a white chocolate mocha",
"a hazelnut coffee",
"a toffee nut latte",
"a peppermint mocha",
"a cinnamon dolce latte",
"a coconut milk latte",
"an almond milk cappuccino",
"an oat milk latte",
"a caramel frappuccino",
"a chocolate frappuccino",
"a butter pecan coffee",
"a maple pecan latte",
"a sea salt caramel mocha",
"a nitro cold brew",
"a pumpkin cold brew",
"a honey almond flat white",
"a sweet cream cold brew",
"a matcha latte",
"a golden latte",
"a turmeric latte",
"a beetroot latte",
"a kopi luwak",
]
self.LAST_5_COFFEES: list = []
def tdTuple(self, td:datetime.timedelta) -> tuple:
def tdTuple(self, td: datetime.timedelta) -> tuple:
"""
Create TimeDelta Tuple
Args:
td (datetime.timedelta)
Returns:
tuple
"""
def _t(t, n):
if t < n:
if t < n:
return (t, 0)
v = t//n
return (t - (v * n), v)
v = t // n
return (t - (v * n), v)
(s, h) = _t(td.seconds, 3600)
(s, m) = _t(s, 60)
(s, m) = _t(s, 60)
(mics, mils) = _t(td.microseconds, 1000)
return (td.days, h, m, s, mics, mils)
async def get_counter(self, counter: Optional[str] = None) -> Optional[dict]:
"""
Get Counter
Args:
counter (Optional[str])
Returns:
Optional[dict]
"""
stats_db: str|LiteralString = self.dbs.get('stats', '')
stats_db: str | LiteralString = self.dbs.get("stats", "")
if not stats_db:
return None
async with sqlite3.connect(stats_db,
timeout=3) as db_conn:
async with sqlite3.connect(stats_db, timeout=3) as db_conn:
db_conn.row_factory = sqlite3.Row
query: str = "SELECT ? FROM stats LIMIT 1"
if not counter:
query = "SELECT * FROM stats LIMIT 1"
async with await db_conn.execute(query, (counter,) if counter else None) as db_cursor:
async with await db_conn.execute(
query, (counter,) if counter else None
) as db_cursor:
result = await db_cursor.fetchone()
return result
async def get_stats_embed(self) -> Optional[Embed]:
"""
Get Stats Embed
Returns:
Optional[Embed]
"""
counters: Optional[dict] = await self.get_counter()
if not counters:
return None
embed: Embed = Embed(title="Stats")
counter_message: str = ""
counters_sorted: dict = dict(sorted(counters.items(),
key=lambda item: item[1], reverse=True))
counters_sorted: dict = dict(
sorted(counters.items(), key=lambda item: item[1], reverse=True)
)
for counter, value in counters_sorted.items():
counter = regex.sub(r'_', ' ',
counter.strip()).title()
counter = regex.sub(r"_", " ", counter.strip()).title()
counter_message += f"- {value} {counter}\n"
embed.description = counter_message.strip()
return embed
async def increment_counter(self, counter: str) -> bool:
"""
Increment Counter
Args:
counter (str)
Returns:
bool
"""
stats_db: str|LiteralString = self.dbs.get('stats', '')
stats_db: str | LiteralString = self.dbs.get("stats", "")
if not stats_db:
return False
async with sqlite3.connect(stats_db,
timeout=3) as db_conn:
async with await db_conn.execute(f"UPDATE stats SET {counter} = {counter} + 1") as db_cursor:
async with sqlite3.connect(stats_db, timeout=3) as db_conn:
async with await db_conn.execute(
f"UPDATE stats SET {counter} = {counter} + 1"
) as db_cursor:
if db_cursor.rowcount < 0:
logging.critical("[karma::increment_counter] Fail! %s", db_cursor.rowcount)
logging.critical(
"[karma::increment_counter] Fail! %s", db_cursor.rowcount
)
return False
await db_conn.commit()
return True
@@ -156,32 +217,39 @@ class Util:
async def get_ud_def(self, term: str) -> tuple[str, str]:
"""
Get Definition from UD
Args:
term (str)
Returns:
tuple[str, str]
"""
try:
async with ClientSession() as session:
async with await session.get(self.URL_URBANDICTIONARY,
params={
"term": term,
},
headers = {
'content-type': 'application/json; charset=utf-8',
}, timeout=ClientTimeout(connect=5, sock_read=5)) as request:
logging.debug("UD returned: %s",
await request.text())
async with await session.get(
self.URL_URBANDICTIONARY,
params={
"term": term,
},
headers={
"content-type": "application/json; charset=utf-8",
},
timeout=ClientTimeout(connect=5, sock_read=5),
) as request:
logging.debug("UD returned: %s", await request.text())
data: dict = await request.json()
if "list" in data:
definitions: list[dict] = data["list"]
if definitions:
definition: dict = definitions[0]
definition_word: str = definition.get("word", "N/A")
definition_text: str = regex.sub(r'(\r|\n|\r\n)', ' ', definition["definition"].strip())
return (definition_word, definition_text) # Tuple: Returned word, returned definition
definition_text: str = regex.sub(
r"(\r|\n|\r\n)", " ", definition["definition"].strip()
)
return (
definition_word,
definition_text,
) # Tuple: Returned word, returned definition
else:
return (term, "Not found!")
else:
@@ -193,24 +261,24 @@ class Util:
async def get_insult(self, recipient: str) -> str:
"""
Get Insult
Args:
recipient (str)
Returns:
str
"""
async with ClientSession() as session:
async with await session.get(f"{self.URL_INSULTAPI}?who={recipient}") as request:
async with await session.get(
f"{self.URL_INSULTAPI}?who={recipient}"
) as request:
request.raise_for_status()
return await request.text()
async def get_compliment(self, subject: str,
language: Optional[str] = 'en') -> str:
async def get_compliment(self, subject: str, language: Optional[str] = "en") -> str:
"""
Get Compliment
Args:
subject (str)
language (Optional[str]) (default: 'en')
@@ -223,95 +291,117 @@ class Util:
async def get_whisky(self) -> Optional[tuple]:
"""
Get Whisky
Returns:
Optional[tuple]
"""
whisky_db: str|LiteralString = self.dbs.get('whisky', '')
whisky_db: str | LiteralString = self.dbs.get("whisky", "")
if not whisky_db:
return None
async with sqlite3.connect(database=whisky_db,
timeout=2) as db_conn:
db_query: str = "SELECT name, category, description FROM whiskeys ORDER BY random() LIMIT 1"
async with sqlite3.connect(database=whisky_db, timeout=2) as db_conn:
db_query: str = (
"SELECT name, category, description FROM whiskeys ORDER BY random() LIMIT 1"
)
async with await db_conn.execute(db_query) as db_cursor:
db_result: Optional[Union[sqlite3.Row, tuple]] = await db_cursor.fetchone()
db_result: Optional[Union[sqlite3.Row, tuple]] = (
await db_cursor.fetchone()
)
if not db_result:
return None
(name, category, description) = db_result
name = regex.sub(r'(^\p{White_Space}|\r|\n)', '',
regex.sub(r'\p{White_Space}{2,}', ' ',
name.strip()))
category = regex.sub(r'(^\p{White_Space}|\r|\n)', '',
regex.sub(r'\p{White_Space}{2,}', ' ',
category.strip()))
description = regex.sub(r'(^\p{White_Space}|\r|\n)', '',
regex.sub(r'\p{White_Space}{2,}', ' ',
description.strip()))
name = regex.sub(
r"(^\p{White_Space}|\r|\n)",
"",
regex.sub(r"\p{White_Space}{2,}", " ", name.strip()),
)
category = regex.sub(
r"(^\p{White_Space}|\r|\n)",
"",
regex.sub(r"\p{White_Space}{2,}", " ", category.strip()),
)
description = regex.sub(
r"(^\p{White_Space}|\r|\n)",
"",
regex.sub(r"\p{White_Space}{2,}", " ", description.strip()),
)
return (name, category, description)
async def get_drink(self) -> Optional[tuple]:
"""
Get Drink
Returns:
Optional[tuple]
"""
drinks_db: str|LiteralString = self.dbs.get('drinks', '')
drinks_db: str | LiteralString = self.dbs.get("drinks", "")
if not drinks_db:
return None
async with sqlite3.connect(database=drinks_db,
timeout=2) as db_conn:
db_query: str = "SELECT name, ingredients FROM cocktails ORDER BY random() LIMIT 1"
async with sqlite3.connect(database=drinks_db, timeout=2) as db_conn:
db_query: str = (
"SELECT name, ingredients FROM cocktails ORDER BY random() LIMIT 1"
)
async with await db_conn.execute(db_query) as db_cursor:
db_result: tuple = await db_cursor.fetchone()
(name, ingredients) = db_result
name = regex.sub(r'(^\p{White_Space}|\r|\n)', '', regex.sub(r'\p{White_Space}{2,}', ' ', name.strip()))
ingredients = regex.sub(r'(^\p{White_Space}|\r|\n)', '', regex.sub(r'\p{White_Space}{2,}', ' ', ingredients.strip()))
ingredients = regex.sub(r'\*', '\u2731', ingredients.strip())
name = regex.sub(
r"(^\p{White_Space}|\r|\n)",
"",
regex.sub(r"\p{White_Space}{2,}", " ", name.strip()),
)
ingredients = regex.sub(
r"(^\p{White_Space}|\r|\n)",
"",
regex.sub(r"\p{White_Space}{2,}", " ", ingredients.strip()),
)
ingredients = regex.sub(r"\*", "\u2731", ingredients.strip())
return (name, ingredients)
async def get_strain(self, strain: Optional[str] = None) -> Optional[tuple]:
"""
Get Strain
Args:
strain (Optional[str])
Returns:
Optional[tuple]
"""
strains_db: str|LiteralString = self.dbs.get('strains', '')
strains_db: str | LiteralString = self.dbs.get("strains", "")
if not strains_db:
return None
async with sqlite3.connect(database=strains_db,
timeout=2) as db_conn:
async with sqlite3.connect(database=strains_db, timeout=2) as db_conn:
db_params: Optional[tuple] = None
if not strain:
db_query: str = "SELECT name, description FROM strains_w_desc ORDER BY random() LIMIT 1"
db_query: str = (
"SELECT name, description FROM strains_w_desc ORDER BY random() LIMIT 1"
)
else:
db_query = "SELECT name, description FROM strains_w_desc WHERE name LIKE ?"
db_params = (f"%{strain.strip()}%",)
db_query = (
"SELECT name, description FROM strains_w_desc WHERE name LIKE ?"
)
db_params = (f"%{strain.strip()}%",)
async with await db_conn.execute(db_query, db_params) as db_cursor:
db_result: Optional[tuple] = await db_cursor.fetchone()
return db_result
async def get_qajoke(self) -> Optional[tuple]:
"""
Get QA Joke
Returns:
Optional[tuple]
"""
qajoke_db: str|LiteralString = self.dbs.get('qajoke', '')
qajoke_db: str | LiteralString = self.dbs.get("qajoke", "")
if not qajoke_db:
return None
async with sqlite3.connect(database=qajoke_db,
timeout=2) as db_conn:
db_query: str = "SELECT question, answer FROM jokes ORDER BY RANDOM() LIMIT 1"
async with sqlite3.connect(database=qajoke_db, timeout=2) as db_conn:
db_query: str = (
"SELECT question, answer FROM jokes ORDER BY RANDOM() LIMIT 1"
)
async with await db_conn.execute(db_query) as cursor:
(question, answer) = await cursor.fetchone()
return (question, answer)
@@ -320,107 +410,110 @@ class Util:
async def get_rjoke(self) -> Optional[tuple]:
"""
Get r/joke Joke
Returns:
Optional[tuple]
"""
rjokes_db: str|LiteralString = self.dbs.get('rjokes', '')
rjokes_db: str | LiteralString = self.dbs.get("rjokes", "")
if not rjokes_db:
return None
async with sqlite3.connect(database=rjokes_db, timeout=2) as db_conn:
db_query: str = "SELECT title, body, score FROM jokes WHERE score >= 100 ORDER BY RANDOM() LIMIT 1'"
db_query: str = (
"SELECT title, body, score FROM jokes WHERE score >= 100 ORDER BY RANDOM() LIMIT 1'"
)
async with await db_conn.execute(db_query) as cursor:
(title, body, score) = await cursor.fetchone()
return (title, body, score)
return None
return None
async def get_random_fact(self) -> str:
"""
Get Random Fact
Returns:
str
"""
try:
facts_api_url: str = "https://uselessfacts.jsph.pl/api/v2/facts/random"
facts_backup_url: str = "https://cnichols1734.pythonanywhere.com/facts/random"
facts_backup_url: str = (
"https://cnichols1734.pythonanywhere.com/facts/random"
)
async with ClientSession() as client:
try:
async with await client.get(facts_api_url,
timeout=ClientTimeout(connect=5, sock_read=5)) as request:
async with await client.get(
facts_api_url, timeout=ClientTimeout(connect=5, sock_read=5)
) as request:
_json: dict = await request.json()
fact: str = _json.get('text', None)
fact: str = _json.get("text", None)
if not fact:
raise BaseException("RandFact Src 1 Failed")
return fact
except:
async with await client.get(facts_backup_url,
timeout=ClientTimeout(connect=5, sock_read=5)) as request:
async with await client.get(
facts_backup_url, timeout=ClientTimeout(connect=5, sock_read=5)
) as request:
_json = await request.json()
fact = _json.get('fact', None)
fact = _json.get("fact", None)
return fact
except Exception as e:
traceback.print_exc()
return f"Failed to get a random fact :( [{str(e)}]"
async def get_cookie(self) -> Optional[dict]:
"""
Get Cookie
Returns:
Optional[dict]
"""
cookies_db = self.dbs.get('cookies', '')
cookies_db = self.dbs.get("cookies", "")
if not cookies_db:
return None
async with sqlite3.connect(cookies_db,
timeout=2) as db_conn:
db_query: str = "SELECT name, origin, image_url FROM cookies ORDER BY RANDOM() LIMIT 1"
async with sqlite3.connect(cookies_db, timeout=2) as db_conn:
db_query: str = (
"SELECT name, origin, image_url FROM cookies ORDER BY RANDOM() LIMIT 1"
)
async with await db_conn.execute(db_query) as db_cursor:
(name, origin, image_url) = await db_cursor.fetchone()
return {
'name': name,
'origin': origin,
'image_url': image_url
}
return {"name": name, "origin": origin, "image_url": image_url}
def get_coffee(self,
recipient_allergic: Optional[bool] = False) -> Optional[str]:
def get_coffee(self, recipient_allergic: Optional[bool] = False) -> Optional[str]:
"""
Get Coffee
Args:
recipient_allergic (bool): Is the recipient allergic? (so we know when to keep our nuts out of it)
Returns:
str
"""
try:
randomCoffee: str = random.choice(self.COFFEES)
if self.LAST_5_COFFEES and randomCoffee in self.LAST_5_COFFEES\
or (recipient_allergic and "nut" in randomCoffee.lower()):
return self.get_coffee() # Recurse
if (
self.LAST_5_COFFEES
and randomCoffee in self.LAST_5_COFFEES
or (recipient_allergic and "nut" in randomCoffee.lower())
):
return self.get_coffee() # Recurse
if len(self.LAST_5_COFFEES) >= 5:
self.LAST_5_COFFEES.pop() # Store no more than 5 of the last served coffees
self.LAST_5_COFFEES.pop() # Store no more than 5 of the last served coffees
self.LAST_5_COFFEES.append(randomCoffee)
return randomCoffee
except:
traceback.print_exc()
return None
def get_days_to_xmas(self) -> Optional[tuple]:
"""
Get # of Days until Xmas
Returns:
Optional[tuple]
"""
today: datetime.datetime = datetime.datetime.now(tz=pytz.UTC)
xmas: datetime.datetime = datetime.datetime(
@@ -429,25 +522,24 @@ class Util:
day=25,
tzinfo=pytz.UTC,
)
td: datetime.timedelta = (xmas - today)
td: datetime.timedelta = xmas - today
days, hours, minutes, seconds, us, ms = self.tdTuple(td)
return (days, hours, minutes, seconds, ms, us)
async def get_randmsg(self) -> Optional[str]:
"""
Get Random Message from randmsg.db
Returns:
Optional[str]
"""
randmsg_db: str|LiteralString = self.dbs.get('randmsg', '')
randmsg_db: str | LiteralString = self.dbs.get("randmsg", "")
if not randmsg_db:
return None
async with sqlite3.connect(database=randmsg_db,
timeout=2) as db_conn:
async with sqlite3.connect(database=randmsg_db, timeout=2) as db_conn:
db_query: str = "SELECT msg FROM msgs ORDER BY RANDOM() LIMIT 1"
async with await db_conn.execute(db_query) as db_cursor:
(result,) = await db_cursor.fetchone()
return result
return result

37
util/radio_util.py
View File

@@ -4,6 +4,7 @@ from aiohttp import ClientSession, ClientTimeout
"""Radio Utils"""
async def get_now_playing() -> Optional[str]:
"""
Get radio now playing
@@ -15,17 +16,21 @@ async def get_now_playing() -> Optional[str]:
np_url: str = "https://api.codey.lol/radio/np"
try:
async with ClientSession() as session:
async with await session.post(np_url, headers={
'content-type': 'application/json; charset=utf-8',
}, timeout=ClientTimeout(connect=1.5, sock_read=1.5)) as request:
async with await session.post(
np_url,
headers={
"content-type": "application/json; charset=utf-8",
},
timeout=ClientTimeout(connect=1.5, sock_read=1.5),
) as request:
request.raise_for_status()
response_json = await request.json()
artistsong = response_json.get('artistsong')
artistsong = response_json.get("artistsong")
return artistsong
except Exception as e:
logging.critical("Now playing retrieval failed: %s",
str(e))
return None
logging.critical("Now playing retrieval failed: %s", str(e))
return None
async def skip() -> bool:
"""
@@ -38,13 +43,13 @@ async def skip() -> bool:
try:
ls_uri: str = "http://127.0.0.1:29000"
async with ClientSession() as session:
async with session.get(f"{ls_uri}/next",
timeout=ClientTimeout(connect=2, sock_read=2)) as request:
request.raise_for_status()
text: Optional[str] = await request.text()
return text == "OK"
async with session.get(
f"{ls_uri}/next", timeout=ClientTimeout(connect=2, sock_read=2)
) as request:
request.raise_for_status()
text: Optional[str] = await request.text()
return text == "OK"
except Exception as e:
logging.debug("Skip failed: %s", str(e))
return False # failsafe
logging.debug("Skip failed: %s", str(e))
return False # failsafe

166
util/sing_util.py
View File

@@ -6,21 +6,24 @@ import traceback
from discord import Activity
from typing import Optional, Union
class Utility:
"""Sing Utility"""
def __init__(self) -> None:
self.api_url: str = "http://127.0.0.1:52111/lyric/search"
self.api_src: str = "DISC-HAVOC"
def parse_song_input(self, song: Optional[str] = None,
activity: Optional[Activity] = None) -> Union[bool, tuple]:
def parse_song_input(
self, song: Optional[str] = None, activity: Optional[Activity] = None
) -> Union[bool, tuple]:
"""
Parse Song (Sing Command) Input
Args:
song (Optional[str]): Song to search
activity (Optional[discord.Activity]): Discord activity, used to attempt lookup if no song is provided
Returns:
Union[bool, tuple]
"""
@@ -29,13 +32,18 @@ class Utility:
return False
if not song and activity:
if not activity.name:
return False # No valid activity found
return False # No valid activity found
match activity.name.lower():
case "codey toons" | "cider" | "sonixd":
search_artist: str = " ".join(str(activity.state)\
.strip().split(" ")[1:])
search_artist = regex.sub(r"(\s{0,})(\[(spotify|tidal|sonixd|browser|yt music)])$", "",
search_artist.strip(), flags=regex.IGNORECASE)
search_artist: str = " ".join(
str(activity.state).strip().split(" ")[1:]
)
search_artist = regex.sub(
r"(\s{0,})(\[(spotify|tidal|sonixd|browser|yt music)])$",
"",
search_artist.strip(),
flags=regex.IGNORECASE,
)
search_song = str(activity.details)
song = f"{search_artist} : {search_song}"
case "tidal hi-fi":
@@ -43,110 +51,136 @@ class Utility:
search_song = str(activity.details)
song = f"{search_artist} : {search_song}"
case "spotify":
if not activity.title or not activity.artist: # type: ignore
if not activity.title or not activity.artist: # type: ignore
"""
Attributes exist, but mypy does not recognize them. Ignored.
"""
return False
search_artist = str(activity.title) # type: ignore
search_song = str(activity.artist) # type: ignore
search_artist = str(activity.title) # type: ignore
search_song = str(activity.artist) # type: ignore
song = f"{search_artist} : {search_song}"
case "serious.fm" | "cocks.fm" | "something":
if not activity.details:
song = str(activity.state)
else:
search_artist = str(activity.state).rsplit("[", maxsplit=1)[0] # Strip genre
search_artist = str(activity.state).rsplit("[", maxsplit=1)[
0
] # Strip genre
search_song = str(activity.details)
song = f"{search_artist} : {search_song}"
case _:
return False # Unsupported activity detected
search_split_by: str = ":" if not(song) or len(song.split(":")) > 1\
else "-" # Support either : or - to separate artist/track
return False # Unsupported activity detected
search_split_by: str = (
":" if not (song) or len(song.split(":")) > 1 else "-"
) # Support either : or - to separate artist/track
if not song:
return False
search_artist = song.split(search_split_by)[0].strip()
search_song = "".join(song.split(search_split_by)[1:]).strip()
search_subsearch: Optional[str] = None
if search_split_by == ":" and len(song.split(":")) > 2: # Support sub-search if : is used (per instructions)
search_song = song.split(search_split_by)[1].strip() # Reduce search_song to only the 2nd split of : [the rest is meant to be lyric text]
search_subsearch = "".join(song.split(search_split_by)[2:]) # Lyric text from split index 2 and beyond
if (
search_split_by == ":" and len(song.split(":")) > 2
): # Support sub-search if : is used (per instructions)
search_song = song.split(search_split_by)[
1
].strip() # Reduce search_song to only the 2nd split of : [the rest is meant to be lyric text]
search_subsearch = "".join(
song.split(search_split_by)[2:]
) # Lyric text from split index 2 and beyond
return (search_artist, search_song, search_subsearch)
except:
traceback.print_exc()
return False
async def lyric_search(self, artist: str, song: str,
sub: Optional[str] = None) -> Optional[list]:
async def lyric_search(
self, artist: str, song: str, sub: Optional[str] = None
) -> Optional[list]:
"""
Lyric Search
Args:
artist (str): Artist to search
song (str): Song to search
sub (Optional[str]): Lyrics for subsearch
sub (Optional[str]): Lyrics for subsearch
Returns:
Optional[list]
"""
try:
if not artist or not song:
return [("FAIL! Artist/Song not provided",)]
search_obj: dict = {
'a': artist.strip(),
's': song.strip(),
'extra': True,
'src': self.api_src,
"a": artist.strip(),
"s": song.strip(),
"extra": True,
"src": self.api_src,
}
if len(song.strip()) < 1:
search_obj.pop('a')
search_obj.pop('s')
search_obj['t'] = artist.strip() # Parse failed, try title without sep
search_obj.pop("a")
search_obj.pop("s")
search_obj["t"] = artist.strip() # Parse failed, try title without sep
if sub and len(sub) >= 2:
search_obj['sub'] = sub.strip()
search_obj["sub"] = sub.strip()
async with aiohttp.ClientSession() as session:
async with await session.post(self.api_url,
json=search_obj,
timeout=aiohttp.ClientTimeout(connect=5, sock_read=10)) as request:
async with await session.post(
self.api_url,
json=search_obj,
timeout=aiohttp.ClientTimeout(connect=5, sock_read=10),
) as request:
request.raise_for_status()
response: dict = await request.json()
if response.get('err'):
if response.get("err"):
return [(f"ERR: {response.get('errorText')}",)]
out_lyrics = regex.sub(r'<br>', '\u200B\n', response.get('lyrics', ''))
out_lyrics = regex.sub(
r"<br>", "\u200b\n", response.get("lyrics", "")
)
response_obj: dict = {
'artist': response.get('artist'),
'song': response.get('song'),
'lyrics': out_lyrics,
'src': response.get('src'),
'confidence': float(response.get('confidence', 0.0)),
'time': float(response.get('time', -1.0)),
"artist": response.get("artist"),
"song": response.get("song"),
"lyrics": out_lyrics,
"src": response.get("src"),
"confidence": float(response.get("confidence", 0.0)),
"time": float(response.get("time", -1.0)),
}
lyrics = response_obj.get('lyrics')
lyrics = response_obj.get("lyrics")
if not lyrics:
return None
response_obj['lyrics'] = textwrap.wrap(text=lyrics.strip(),
width=1500, drop_whitespace=False,
replace_whitespace=False, break_long_words=True,
break_on_hyphens=True, max_lines=8)
response_obj['lyrics_short'] = textwrap.wrap(text=lyrics.strip(),
width=750, drop_whitespace=False,
replace_whitespace=False, break_long_words=True,
break_on_hyphens=True, max_lines=1)
response_obj["lyrics"] = textwrap.wrap(
text=lyrics.strip(),
width=1500,
drop_whitespace=False,
replace_whitespace=False,
break_long_words=True,
break_on_hyphens=True,
max_lines=8,
)
response_obj["lyrics_short"] = textwrap.wrap(
text=lyrics.strip(),
width=750,
drop_whitespace=False,
replace_whitespace=False,
break_long_words=True,
break_on_hyphens=True,
max_lines=1,
)
return [
(
response_obj.get('artist'), response_obj.get('song'), response_obj.get('src'),
response_obj.get("artist"),
response_obj.get("song"),
response_obj.get("src"),
f"{int(response_obj.get('confidence', -1.0))}%",
f"{response_obj.get('time', -666.0):.4f}s",
),
response_obj.get('lyrics'),
response_obj.get('lyrics_short'),
]
response_obj.get("lyrics"),
response_obj.get("lyrics_short"),
]
except Exception as e:
traceback.print_exc()
return [f"Retrieval failed: {str(e)}"]
return [f"Retrieval failed: {str(e)}"]