From a8014f286a38276260b5df7cff5926563d87db28 Mon Sep 17 00:00:00 2001 From: codey Date: Sun, 16 Feb 2025 13:16:53 -0500 Subject: [PATCH] misc --- catbox.py | 88 ++++++++++++++++++++++++++++++++++++++ cogs/radio.py | 10 ++--- jesusmemes.py | 11 +++-- litterbox.py | 104 +++++++++++++++++++++++++++++++++++++++++++++ litterbox_test.py | 25 +++++++++++ util/radio_util.py | 6 ++- 6 files changed, 233 insertions(+), 11 deletions(-) create mode 100644 catbox.py create mode 100644 litterbox.py create mode 100644 litterbox_test.py diff --git a/catbox.py b/catbox.py new file mode 100644 index 0000000..5ffb742 --- /dev/null +++ b/catbox.py @@ -0,0 +1,88 @@ +#!/usr/bin/env python3.12 + +from typing import Optional +from aiohttp import ClientSession, ClientTimeout, FormData +import traceback +import logging +import random +import os +import magic + +""" +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' + } + +class CatboxAsync: + 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' + return f"{random.getrandbits(32)}.{fileExt}" + + def __init__(self): + self.catbox_api_url = catbox_api_url + self.headers = http_headers + + async def upload(self, file: str) -> Optional[str]: + """ + Upload file to catbox + Args: + file (str): Path of file to be uploaded + Returns: + str + """ + try: + if not file: + return None + 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: + post_data: FormData = FormData() + 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 + ) + 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: + request.raise_for_status() + return await request.text() + except: + traceback.print_exc() + return None + finally: + try: + fileContents.close() + except: + return None \ No newline at end of file diff --git a/cogs/radio.py b/cogs/radio.py index ed600c8..fe628f5 100644 --- a/cogs/radio.py +++ b/cogs/radio.py @@ -106,13 +106,13 @@ class Radio(commands.Cog): but they exist. """ logging.info("Detected VC not playing... playing!") - source = discord.FFmpegOpusAudio(self.STREAM_URL, + source: discord.FFmpegAudio = discord.FFmpegOpusAudio(self.STREAM_URL, before_options="-timeout 3000000") - - vc.play(source, after=lambda e: logging.info("Error: %s", e) if e else None) # type: ignore - + vc.play(source, # type: ignore + after=lambda e: logging.info("Error: %s", e) if e\ + else None) # Get Now Playing - np_track = await get_now_playing() + np_track: Optional[str] = await get_now_playing() if np_track and not self.LAST_NP_TRACK == np_track: self.LAST_NP_TRACK = np_track if isinstance(vc.channel, discord.VoiceChannel): diff --git a/jesusmemes.py b/jesusmemes.py index 22375c2..35b5dda 100644 --- a/jesusmemes.py +++ b/jesusmemes.py @@ -6,9 +6,9 @@ import regex from regex import Pattern import os import random +import logging import traceback -from urllib.parse import quote as urlquote -from catbox import Catbox +from catbox import CatboxAsync """ Jesus Meme Generator @@ -65,8 +65,11 @@ class JesusMemeGenerator(): f.write(await response.read()) if not out_fname: - uploader = Catbox() - meme_link: str = uploader.upload(f"{self.MEMESTORAGEDIR}/{out_fname}") + uploader = CatboxAsync() + meme_link: Optional[str] = await uploader.upload(f"{self.MEMESTORAGEDIR}/{out_fname}") + if not meme_link: + logging.info("Meme upload failed!") + return None return meme_link except: print(traceback.format_exc()) diff --git a/litterbox.py b/litterbox.py new file mode 100644 index 0000000..7f091ce --- /dev/null +++ b/litterbox.py @@ -0,0 +1,104 @@ +#!/usr/bin/env python3.12 + +""" +Litterbox (Catbox) Uploader (Async) +""" + +import traceback +from typing import Optional, Union +from io import BufferedReader +from aiohttp import ClientSession, ClientTimeout, FormData +import magic +import logging +import random +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' + } + +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' + return f"{random.getrandbits(32)}.{fileExt}" + + 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 + Returns: + Optional[str] + """ + try: + if not file: + return None + if isinstance(file, str): + if not os.path.exists(file): + logging.critical("Could not find %s", file) + return None + if isinstance(file, BufferedReader): + file = file.read() + + fileExt: str = 'png' + if isinstance(file, str): + if file.find(".") > 0: + fileExt = "".join(file.split(".")[-1:]) + + 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)) + + 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 + ) + 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: + request.raise_for_status() + return await request.text() + except: + traceback.print_exc() + return None + finally: + if isinstance(file, BufferedReader): + try: + file.close() + except: + return None diff --git a/litterbox_test.py b/litterbox_test.py new file mode 100644 index 0000000..2f6779a --- /dev/null +++ b/litterbox_test.py @@ -0,0 +1,25 @@ +#!/usr/bin/env python3.12 + +"""Tests for both Catbox & Litterbox""" + +import os +import logging +import asyncio +from typing import Optional +from litterbox import LitterboxAsync +from catbox import CatboxAsync + +logger = logging.getLogger() +logger.setLevel(logging.INFO) + +async def test() -> None: + f = os.path.join(os.path.expanduser("~"), "qu.png") + box1: LitterboxAsync = LitterboxAsync() + box2: CatboxAsync = CatboxAsync() + url1: Optional[str] = await box1.upload(f) + url2: Optional[str] = await box2.upload(f) + logging.info("""Uploaded URLs: + Litter - %s\n + Cat - %s""", url1, url2) + +asyncio.run(test()) \ No newline at end of file diff --git a/util/radio_util.py b/util/radio_util.py index 5e108ff..b794df9 100644 --- a/util/radio_util.py +++ b/util/radio_util.py @@ -1,11 +1,12 @@ #!/usr/bin/env python3.12 import logging +from typing import Optional from aiohttp import ClientSession, ClientTimeout """Radio Utils""" -async def get_now_playing() -> str: +async def get_now_playing() -> Optional[str]: np_url: str = "https://api.codey.lol/radio/np" try: async with ClientSession() as session: @@ -18,4 +19,5 @@ async def get_now_playing() -> str: return artistsong except Exception as e: logging.critical("Now playing retrieval failed: %s", - str(e)) \ No newline at end of file + str(e)) + return None \ No newline at end of file