diff --git a/plugins/upload.py b/plugins/upload.py index e9fca74..db6ad84 100644 --- a/plugins/upload.py +++ b/plugins/upload.py @@ -16,7 +16,7 @@ Dependencies: - ircstyle Author: Your Name -Version: 1.1 +Version: 1.2 Date: 2025-02-12 """ @@ -43,6 +43,16 @@ class UploadPlugin: def __init__(self, bot): self.bot = bot + def _ensure_str(self, value): + """ + Ensure the value is a string. If it's bytes, decode it as UTF-8 with error replacement. + """ + if isinstance(value, bytes): + return value.decode('utf-8', errors='replace') + if value is None: + return '' + return str(value) + @command async def upload(self, mask, target, args): """ @@ -69,21 +79,21 @@ class UploadPlugin: try: await self.do_upload(url, target, mp3) except Exception as exc: + # Convert exception to a safe Unicode string. + exc_msg = self._ensure_str(exc) self.bot.privmsg( target, - ircstyle.style(f"Upload task error: {exc}", fg="red", bold=True, reset=True) + ircstyle.style(f"Upload task error: {exc_msg}", fg="red", bold=True, reset=True) ) async def do_upload(self, url, target, mp3): """ Download a file using yt-dlp and upload it to hardfiles.org. - Handles binary data properly to avoid UTF-8 decoding errors. + Handles binary data and non-UTF-8 strings to avoid decoding errors. """ max_size = 100 * 1024 * 1024 # 100MB limit - # Use a temporary directory for downloads (auto-cleanup) with tempfile.TemporaryDirectory() as tmp_dir: - # Determine whether to check headers for file size. parsed_url = urlparse(url) domain = parsed_url.netloc.lower() skip_check_domains = ("x.com", "instagram.com", "youtube.com", "youtu.be", "streamable.com") @@ -113,13 +123,13 @@ class UploadPlugin: ) return except Exception as e: + err_msg = self._ensure_str(e) self.bot.privmsg( target, - ircstyle.style(f"Error during header check: {e}", fg="red", bold=True, reset=True) + ircstyle.style(f"Error during header check: {err_msg}", fg="red", bold=True, reset=True) ) return - # Configure yt-dlp options. ydl_opts = { 'outtmpl': os.path.join(tmp_dir, '%(title)s.%(ext)s'), 'format': 'bestaudio/best' if mp3 else 'best[ext=mp4]/best', @@ -134,14 +144,14 @@ class UploadPlugin: }] if mp3 else [], } - # Extract info without downloading first. with yt_dlp.YoutubeDL(ydl_opts) as ydl: try: info = await asyncio.to_thread(ydl.extract_info, url, download=False) except DownloadError as e: + err_msg = self._ensure_str(e) self.bot.privmsg( target, - ircstyle.style(f"Info extraction failed: {e}", fg="red", bold=True, reset=True) + ircstyle.style(f"Info extraction failed: {err_msg}", fg="red", bold=True, reset=True) ) return except UnicodeDecodeError: @@ -151,7 +161,6 @@ class UploadPlugin: ) return - # Check estimated file size if available. estimated_size = info.get('filesize') or info.get('filesize_approx') if estimated_size and estimated_size > max_size: self.bot.privmsg( @@ -163,13 +172,13 @@ class UploadPlugin: ) return - # Download the file. try: info = await asyncio.to_thread(ydl.extract_info, url, download=True) except DownloadError as e: + err_msg = self._ensure_str(e) self.bot.privmsg( target, - ircstyle.style(f"Download failed: {e}", fg="red", bold=True, reset=True) + ircstyle.style(f"Download failed: {err_msg}", fg="red", bold=True, reset=True) ) return except UnicodeDecodeError: @@ -179,14 +188,14 @@ class UploadPlugin: ) return - # Prepare and send metadata (if available). + # Safely convert metadata to strings. metadata_parts = [] - title = info.get("title") - uploader = info.get("uploader") + title = self._ensure_str(info.get("title")) + uploader = self._ensure_str(info.get("uploader")) duration = info.get("duration") - upload_date = info.get("upload_date") + upload_date = self._ensure_str(info.get("upload_date")) view_count = info.get("view_count") - description = info.get("description") + description = self._ensure_str(info.get("description")) if title: metadata_parts.append(ircstyle.style(f"Title: {title}", fg="yellow", bold=True, reset=True)) @@ -206,7 +215,6 @@ class UploadPlugin: if metadata_parts: self.bot.privmsg(target, " | ".join(metadata_parts)) - # Retrieve the downloaded file path. downloaded_files = info.get('requested_downloads', []) if not downloaded_files: self.bot.privmsg( @@ -224,7 +232,6 @@ class UploadPlugin: ) return - # Extra safeguard: verify actual file size. file_size = os.path.getsize(downloaded_file) if file_size > max_size: self.bot.privmsg( @@ -236,13 +243,11 @@ class UploadPlugin: ) return - # Upload the file to hardfiles.org. try: async with aiohttp.ClientSession() as session: form = aiohttp.FormData() async with aiofiles.open(downloaded_file, 'rb') as f: file_content = await f.read() - # Ensure binary data is handled correctly. form.add_field( 'file', file_content, @@ -256,19 +261,21 @@ class UploadPlugin: ircstyle.style(f"Upload failed: HTTP {resp.status}", fg="red", bold=True, reset=True) ) return - # Read raw response bytes and decode with error replacement to avoid decode issues. 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') upload_url = self.extract_url_from_response(response_text) or "Unknown URL" + upload_url = self._ensure_str(upload_url) response_msg = ( ircstyle.style("Upload successful: ", fg="green", bold=True, reset=True) + ircstyle.style(upload_url, fg="blue", underline=True, reset=True) ) self.bot.privmsg(target, response_msg) except Exception as e: + err_msg = self._ensure_str(e) self.bot.privmsg( target, - ircstyle.style(f"Error during file upload: {e}", fg="red", bold=True, reset=True) + ircstyle.style(f"Error during file upload: {err_msg}", fg="red", bold=True, reset=True) ) return