Zodiac/g1mp
1
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

updates for color and what not

Browse Source
This commit is contained in:
Zodiac 2025-02-14 19:17:00 -08:00
parent a5515b8fd4
commit 54bfeada32
2 changed files with 90 additions and 112 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

4
plugins/bomb.py
View File

@ -33,14 +33,14 @@ class PeriodicMessagePlugin:
""" """
self.bot = bot self.bot = bot
self.channel = "" # The IRC channel the bot is operating in self.channel = "" # The IRC channel the bot is operating in
self.periodic_message = ' \u00A0 \u2002 \u2003 ' * 500 # Empty message trick self.periodic_message = ' \u2002 \u2003 ' * 500 # Empty message trick
self.tasks = [] # Stores running asyncio tasks self.tasks = [] # Stores running asyncio tasks
self.running = False # Flag to control task execution self.running = False # Flag to control task execution
self.original_nick = "" # Store the original bot nickname self.original_nick = "" # Store the original bot nickname
# Sleep durations for various tasks (in seconds) # Sleep durations for various tasks (in seconds)
self.sleep_durations = { self.sleep_durations = {
'_send_periodic_message': 4, '_send_periodic_message': 17,
'_change_nick_periodically': 50, '_change_nick_periodically': 50,
'_send_listusers_periodically': 60, '_send_listusers_periodically': 60,
} }

198
plugins/upload.py
View File

@ -18,7 +18,7 @@ Dependencies:
- ircstyle - ircstyle
Author: Zodiac Author: Zodiac
Version: 1.2 Version: 1.3
Date: 2025-02-12 Date: 2025-02-12
""" """
@ -35,6 +35,7 @@ import yt_dlp
from yt_dlp.utils import DownloadError from yt_dlp.utils import DownloadError
from urllib.parse import urlparse from urllib.parse import urlparse
import googleapiclient.discovery import googleapiclient.discovery
from irc3.compat import Queue
# Global headers to mimic a real browser (ban evasion) # Global headers to mimic a real browser (ban evasion)
HEADERS = { HEADERS = {
@ -69,6 +70,25 @@ class UploadPlugin:
bot (irc3.IrcBot): The IRC bot instance. bot (irc3.IrcBot): The IRC bot instance.
""" """
self.bot = bot self.bot = bot
self.queue = Queue() # Initialize the queue
self.bot.loop.create_task(self.process_queue()) # Start queue processing
async def process_queue(self):
"""Process upload tasks from the queue."""
while True:
task = await self.queue.get()
url, target, mp3 = task
try:
await self.do_upload(url, target, mp3)
except Exception as e:
exc_msg = self._ensure_str(e)
self.bot.privmsg(
target,
ircstyle.style("❌ ", fg="red") +
ircstyle.style(f"Upload failed: {exc_msg}", fg="red", bold=True)
)
finally:
self.queue.task_done()
def _ensure_str(self, value): def _ensure_str(self, value):
""" """
@ -126,7 +146,6 @@ class UploadPlugin:
Example: !upload never gonna give you up Example: !upload never gonna give you up
""" """
url_or_search_term = args.get('<url_or_search_term>') url_or_search_term = args.get('<url_or_search_term>')
# If multiple words are provided, join them into a single string.
if isinstance(url_or_search_term, list): if isinstance(url_or_search_term, list):
url_or_search_term = " ".join(url_or_search_term) url_or_search_term = " ".join(url_or_search_term)
mp3 = args.get('--mp3') mp3 = args.get('--mp3')
@ -134,35 +153,39 @@ class UploadPlugin:
if not url_or_search_term: if not url_or_search_term:
self.bot.privmsg( self.bot.privmsg(
target, target,
ircstyle.style("Usage: !upload [--mp3] <url_or_search_term>...", fg="red", bold=True, reset=True), ircstyle.style("❓ ", fg="red") +
ircstyle.style("Usage: !upload [--mp3] <url_or_search_term>...", fg="red", bold=True)
) )
return return
# If the input is not a URL, treat it as a search term
if not re.match(r'^https?://', url_or_search_term): if not re.match(r'^https?://', url_or_search_term):
self.bot.privmsg( self.bot.privmsg(
target, target,
ircstyle.style(f"Searching YouTube for: {url_or_search_term}", fg="blue", bold=True, reset=True), ircstyle.style("🔍 ", fg="blue") +
ircstyle.style("Searching YouTube for: ", fg="blue") +
ircstyle.style(url_or_search_term, fg="green", underline=True, italics=True)
) )
url = await self.search_youtube(url_or_search_term) url = await self.search_youtube(url_or_search_term)
if not url: if not url:
self.bot.privmsg( self.bot.privmsg(
target, target,
ircstyle.style("No results found on YouTube.", fg="red", bold=True, reset=True), ircstyle.style("❌ ", fg="red") +
ircstyle.style("No results found on YouTube.", fg="red", bold=True)
) )
return return
else: else:
url = url_or_search_term url = url_or_search_term
try: # Enqueue the upload task
await self.do_upload(url, target, mp3) queue_size = self.queue.qsize() + 1
except Exception as exc: self.queue.put_nowait((url, target, mp3))
# Convert exception to a safe Unicode string. self.bot.privmsg(
exc_msg = self._ensure_str(exc) target,
self.bot.privmsg( ircstyle.style("⏳ ", fg="yellow") +
target, ircstyle.style("Upload queued ", fg="blue", bold=False, italics=True) +
ircstyle.style(f"Upload task error: {exc_msg}", fg="red", bold=True, reset=True), ircstyle.style(f"(Position #{queue_size})",italics=True, fg="grey") +
) ircstyle.style(" Processing soon...", fg="blue", italics=True)
)
async def do_upload(self, url, target, mp3): async def do_upload(self, url, target, mp3):
""" """
@ -201,36 +224,24 @@ class UploadPlugin:
if response.status != 200: if response.status != 200:
self.bot.privmsg( self.bot.privmsg(
target, target,
ircstyle.style( ircstyle.style("❌ ", fg="red") +
f"Failed to fetch headers: HTTP {response.status}", ircstyle.style(f"Failed to fetch headers: HTTP {response.status}", fg="red", bold=True)
fg="red",
bold=True,
reset=True,
),
) )
return return
content_length = response.headers.get('Content-Length') content_length = response.headers.get('Content-Length')
if content_length and int(content_length) > max_size: if content_length and int(content_length) > max_size:
self.bot.privmsg( self.bot.privmsg(
target, target,
ircstyle.style( ircstyle.style("❌ ", fg="red") +
f"File size ({int(content_length) // (1024 * 1024)}MB) exceeds 500MB limit", ircstyle.style(f"File size ({int(content_length) // (1024 * 1024)}MB) exceeds 500MB limit", fg="red", bold=True)
fg="red",
bold=True,
reset=True,
),
) )
return return
except Exception as e: except Exception as e:
err_msg = self._ensure_str(e) err_msg = self._ensure_str(e)
self.bot.privmsg( self.bot.privmsg(
target, target,
ircstyle.style( ircstyle.style("❌ ", fg="red") +
f"Error during header check: {err_msg}", ircstyle.style(f"Error during header check: {err_msg}", fg="red", bold=True)
fg="red",
bold=True,
reset=True,
),
) )
return return
@ -247,9 +258,7 @@ class UploadPlugin:
'preferredcodec': 'mp3', 'preferredcodec': 'mp3',
'preferredquality': '192', 'preferredquality': '192',
} }
] ] if mp3 else [],
if mp3
else [],
} }
with yt_dlp.YoutubeDL(ydl_opts) as ydl: with yt_dlp.YoutubeDL(ydl_opts) as ydl:
@ -259,20 +268,15 @@ class UploadPlugin:
err_msg = self._ensure_str(e) err_msg = self._ensure_str(e)
self.bot.privmsg( self.bot.privmsg(
target, target,
ircstyle.style( ircstyle.style("❌ ", fg="red") +
f"Info extraction failed: {err_msg}", fg="red", bold=True, reset=True ircstyle.style(f"Info extraction failed: {err_msg}", fg="red", bold=True)
),
) )
return return
except UnicodeDecodeError: except UnicodeDecodeError:
self.bot.privmsg( self.bot.privmsg(
target, target,
ircstyle.style( ircstyle.style("❌ ", fg="red") +
"Error: Received non-UTF-8 output during info extraction", ircstyle.style("Error: Received non-UTF-8 output during info extraction", fg="red", bold=True)
fg="red",
bold=True,
reset=True,
),
) )
return return
@ -280,12 +284,8 @@ class UploadPlugin:
if estimated_size and estimated_size > max_size: if estimated_size and estimated_size > max_size:
self.bot.privmsg( self.bot.privmsg(
target, target,
ircstyle.style( ircstyle.style("❌ ", fg="red") +
f"File size ({estimated_size // (1024 * 1024)}MB) exceeds 500MB limit", ircstyle.style(f"File size ({estimated_size // (1024 * 1024)}MB) exceeds 500MB limit", fg="red", bold=True)
fg="red",
bold=True,
reset=True,
),
) )
return return
@ -295,24 +295,19 @@ class UploadPlugin:
err_msg = self._ensure_str(e) err_msg = self._ensure_str(e)
self.bot.privmsg( self.bot.privmsg(
target, target,
ircstyle.style( ircstyle.style("❌ ", fg="red") +
f"Download failed: {err_msg}", fg="red", bold=True, reset=True ircstyle.style(f"Download failed: {err_msg}", fg="red", bold=True)
),
) )
return return
except UnicodeDecodeError: except UnicodeDecodeError:
self.bot.privmsg( self.bot.privmsg(
target, target,
ircstyle.style( ircstyle.style("❌ ", fg="red") +
"Error: Received non-UTF-8 output during download", ircstyle.style("Error: Received non-UTF-8 output during download", fg="red", bold=True)
fg="red",
bold=True,
reset=True,
),
) )
return return
# Safely convert metadata to strings. # Build metadata for display.
metadata_parts = [] metadata_parts = []
title = self._ensure_str(info.get("title")) title = self._ensure_str(info.get("title"))
uploader = self._ensure_str(info.get("uploader")) uploader = self._ensure_str(info.get("uploader"))
@ -320,53 +315,52 @@ class UploadPlugin:
upload_date = self._ensure_str(info.get("upload_date")) upload_date = self._ensure_str(info.get("upload_date"))
view_count = info.get("view_count") view_count = info.get("view_count")
description = self._ensure_str(info.get("description")) description = self._ensure_str(info.get("description"))
description = re.sub(r'\s+', ' ', description) # Strip multiple spaces to single space
if title: if title:
metadata_parts.append( metadata_parts.append(
ircstyle.style(f"Title: {title}", fg="yellow", bold=True, reset=True) ircstyle.style("📝 Title:", fg="yellow", bold=True) + " " +
ircstyle.style(title, fg="yellow", bold=False, underline=True, italics=True)
) )
if uploader: if uploader:
metadata_parts.append( metadata_parts.append(
ircstyle.style(f"Uploader: {uploader}", fg="purple", bold=True, reset=True) ircstyle.style("👤 Uploader:", fg="purple", bold=True) + " " +
ircstyle.style(uploader, fg="purple", bold=False)
) )
if duration: if duration:
metadata_parts.append( metadata_parts.append(
ircstyle.style( ircstyle.style("⏱ Duration:", fg="green", bold=True) + " " +
f"Duration: {self._format_duration(duration)}", ircstyle.style(self._format_duration(duration), fg="green", bold=False)
fg="green",
bold=True,
reset=True,
)
) )
if upload_date: if upload_date:
formatted_date = ( formatted_date = f"{upload_date[:4]}-{upload_date[4:6]}-{upload_date[6:]}" if len(upload_date) == 8 else upload_date
f"{upload_date[:4]}-{upload_date[4:6]}-{upload_date[6:]}"
if len(upload_date) == 8
else upload_date
)
metadata_parts.append( metadata_parts.append(
ircstyle.style( ircstyle.style("Upload Date:", fg="silver", bold=True) + " " +
f"Upload Date: {formatted_date}", fg="aqua", bold=True, reset=True ircstyle.style(formatted_date, fg="silver", bold=False, italics=True)
)
) )
if view_count is not None: if view_count is not None:
metadata_parts.append( metadata_parts.append(
ircstyle.style(f"Views: {view_count}", fg="royal", bold=True, reset=True) ircstyle.style("Views:", fg="teal", bold=True) + " " +
ircstyle.style(str(view_count), fg="teal", bold=False)
) )
if description: if description:
if len(description) > 200: if len(description) > 200:
description = description[:200] + "..." description = description[:200] + "..."
metadata_parts.append( metadata_parts.append(
ircstyle.style(f"Description: {description}", fg="silver", reset=True) ircstyle.style("Description:", fg="silver", bold=True) + " " +
ircstyle.style(description, fg="silver", bold=False, italics=True)
) )
if metadata_parts: if metadata_parts:
self.bot.privmsg(target, " | ".join(metadata_parts)) # Send metadata header in teal.
# self.bot.privmsg(target, ircstyle.style("📁 Metadata:", fg="teal", bold=True))
self.bot.privmsg(target, " │ ".join(metadata_parts))
downloaded_files = info.get('requested_downloads', []) downloaded_files = info.get('requested_downloads', [])
if not downloaded_files: if not downloaded_files:
self.bot.privmsg( self.bot.privmsg(
target, target,
ircstyle.style("No files downloaded", fg="red", bold=True, reset=True), ircstyle.style("❌ ", fg="red") +
ircstyle.style("No files downloaded", fg="red", bold=True)
) )
return return
@ -375,12 +369,8 @@ class UploadPlugin:
if not downloaded_file or not os.path.exists(downloaded_file): if not downloaded_file or not os.path.exists(downloaded_file):
self.bot.privmsg( self.bot.privmsg(
target, target,
ircstyle.style( ircstyle.style("❌ ", fg="red") +
f"Downloaded file not found: {downloaded_file}", ircstyle.style(f"Downloaded file not found: {downloaded_file}", fg="red", bold=True)
fg="red",
bold=True,
reset=True,
),
) )
return return
@ -388,12 +378,8 @@ class UploadPlugin:
if file_size > max_size: if file_size > max_size:
self.bot.privmsg( self.bot.privmsg(
target, target,
ircstyle.style( ircstyle.style("❌ ", fg="red") +
f"File size ({file_size // (1024 * 1024)}MB) exceeds 500MB limit", ircstyle.style(f"File size ({file_size // (1024 * 1024)}MB) exceeds 500MB limit", fg="red", bold=True)
fg="red",
bold=True,
reset=True,
),
) )
return return
@ -406,39 +392,31 @@ class UploadPlugin:
'file', 'file',
file_content, file_content,
filename=os.path.basename(downloaded_file), filename=os.path.basename(downloaded_file),
content_type='application/octet-stream', content_type='application/octet-stream'
) )
async with session.post( async with session.post('https://hardfiles.org/', data=form, allow_redirects=False) as resp:
'https://hardfiles.org/', data=form, allow_redirects=False
) as resp:
if resp.status not in [200, 201, 302, 303]: if resp.status not in [200, 201, 302, 303]:
self.bot.privmsg( self.bot.privmsg(
target, target,
ircstyle.style( ircstyle.style("❌ ", fg="red") +
f"Upload failed: HTTP {resp.status}", ircstyle.style(f"Upload failed: HTTP {resp.status}", fg="red", bold=True)
fg="red",
bold=True,
reset=True,
),
) )
return return
raw_response = await resp.read() raw_response = await resp.read()
# Decode the response safely even if non-UTF-8 bytes are present.
response_text = raw_response.decode('utf-8', errors='replace') response_text = raw_response.decode('utf-8', errors='replace')
upload_url = self.extract_url_from_response(response_text) or "Unknown URL" upload_url = self.extract_url_from_response(response_text) or "Unknown URL"
upload_url = self._ensure_str(upload_url) upload_url = self._ensure_str(upload_url)
response_msg = ( response_msg = (
ircstyle.style("Upload successful: ", fg="green", bold=True, reset=True) ircstyle.style("✅ Upload successful: ", fg="green", bold=True) +
+ ircstyle.style(upload_url, fg="blue", underline=True, reset=True) ircstyle.style(upload_url, fg="blue", underline=True)
) )
self.bot.privmsg(target, response_msg) self.bot.privmsg(target, response_msg)
except Exception as e: except Exception as e:
err_msg = self._ensure_str(e) err_msg = self._ensure_str(e)
self.bot.privmsg( self.bot.privmsg(
target, target,
ircstyle.style( ircstyle.style("❌ ", fg="red") +
f"Error during file upload: {err_msg}", fg="red", bold=True, reset=True ircstyle.style(f"Error during file upload: {err_msg}", fg="red", bold=True)
),
) )
return return
@ -468,4 +446,4 @@ class UploadPlugin:
seconds = int(seconds) seconds = int(seconds)
m, s = divmod(seconds, 60) m, s = divmod(seconds, 60)
h, m = divmod(m, 60) h, m = divmod(m, 60)
return f"{h}h {m}m {s}s" if h else f"{m}m {s}s" return f"{h}h {m}m {s}s" if h else f"{m}m {s}s"