From cd8ed8bc5793985df2c1692072a932ceee517b95 Mon Sep 17 00:00:00 2001 From: salmonik Date: Wed, 8 Jul 2026 02:13:35 +0200 Subject: OPEN THE RUST GATES Signed-off-by: salmonik --- cogs/downloader.py | 234 ----------------------------------------------------- 1 file changed, 234 deletions(-) delete mode 100644 cogs/downloader.py (limited to 'cogs/downloader.py') diff --git a/cogs/downloader.py b/cogs/downloader.py deleted file mode 100644 index 900bafb..0000000 --- a/cogs/downloader.py +++ /dev/null @@ -1,234 +0,0 @@ -# This file is part of Senator Sperma -# Licensed under the GPL-3.0 License -# Copyright (C) 2026 Salmonik @ salmonik@tuta.io -# https://codeberg.org/salmonik/senator-sperma - -import os -import tempfile -import logging -import requests -import discord -import re -import yt_dlp -from discord.ext import commands -import aiofiles -from cogs.utils.url_helper import clean_url_for_ytdlp - -logger = logging.getLogger(__name__) - - -class Downloader(commands.Cog): - def __init__(self, bot): - self.bot = bot - - URL_PATTERN = re.compile( - r'https://(?:[A-Za-z0-9-]+\.)*[A-Za-z0-9-]{1,63}\.[A-Za-z]{2,63}/\S*' - ) - - @commands.message_command( - name="download", - description="Download and convert video/audio from internet" - ) - async def message_download(self, ctx, message: discord.Message): - # check if message has a valid url - url = [m.group(0) for m in self.URL_PATTERN.finditer(message.content)] - if url: - for url in url: - await self.download(ctx=ctx, url=url, fformat="mp4") - else: - await ctx.respond("Brak załączników w wiadomości", ephemeral=True) - - @commands.slash_command( - name="download", - description="Download and convert video/audio from internet" - ) - async def download( - self, - ctx: discord.ApplicationContext, - url: str = discord.Option(str, description="Link media", required=True), - catbox: bool = discord.Option(bool, description="wysłać na serwer??", required=False, default=False), - fformat: str = discord.Option(str, description="jaki format? (mp4, mp3, wav, etc.)", required=False, default="mp4"), - filename_user: str = discord.Option(str, description="jaka nazwa pliku?", required=False) - ): - """Download and convert media from a URL to the specified format""" - await ctx.defer() - # Call the function and get the response - response = await self.download_standalone(url=url, catbox=catbox, fformat=fformat, filename_user=filename_user) - if isinstance(response, dict): - if response["status"] == "success": - if response["type"] == "url": - await ctx.respond(f"pobioron i wysyłon: {response['content']}") - elif response["type"] == "file": - file_data = response["content"] - await ctx.respond(file=discord.File(file_data)) - else: - error_msg = response["content"] - if 'details' in response: - error_msg += f"\n{response['details']}" - await ctx.respond(f"Błąd: {error_msg}", ephemeral=True) - logger.error(f"Error: {response['content']}") - if 'details' in response: - logger.error(f"Details: {response['details']}") - - @staticmethod - async def download_standalone(url: str, catbox: bool = False, fformat: str = "mp4", filename_user: str | None = None): - try: - url = clean_url_for_ytdlp(url) - logger.info(f"Downloading {url} as {fformat}") - - with tempfile.TemporaryDirectory(delete=False) as temp_dir: - ydl_opts = Downloader._get_ydl_options(filename_user, fformat, temp_dir) - - with yt_dlp.YoutubeDL(ydl_opts) as ydl: - ydl.download([url]) - - actual_file, filename = Downloader._find_and_prepare_file(temp_dir, filename_user) - - if Downloader._should_upload_to_server(actual_file, catbox): - return await Downloader._upload_to_server(actual_file, filename, temp_dir) - else: - return Downloader._return_file_response(actual_file, filename, temp_dir) - - except Exception as e: - return Downloader._create_error_response("Download failed", str(e), None) - - @staticmethod - def _sanitize_filename(name: str) -> str: - invalid = '<>:"/\\|?*' - for c in invalid: - name = name.replace(c, '') - return name.strip() - - @staticmethod - def _get_ydl_options(filename: str = None, fformat: str = None, temp_dir: str = None) -> dict: - if filename: - filename = Downloader._sanitize_filename(filename)[:100] - ydl_opts = { - 'outtmpl': os.path.join(temp_dir, filename if filename is not None else '%(title).100s.%(ext)s'), - 'format': 'bestaudio/best' if fformat in ['mp3', 'wav', 'm4a'] else 'bestvideo[ext=mp4]+bestaudio[ext=m4a]/best[ext=mp4]/best', - 'merge_output_format': 'mp4', - 'postprocessors': [], - 'remote-components': 'ejs:github', - 'js-runtimes': {'deno': {'path': '/usr/bin/deno'}}, - } - - if fformat in ['mp3', 'wav', 'm4a']: - ydl_opts['postprocessors'].append({ - 'key': 'FFmpegExtractAudio', - 'preferredcodec': fformat, - 'preferredquality': '192', - }) - elif fformat != 'mp4': - ydl_opts['postprocessors'].append({ - 'key': 'FFmpegVideoConvertor', - 'preferedformat': fformat, - }) - - return ydl_opts - - @staticmethod - def _find_and_prepare_file(temp_dir: str, filename_user: str) -> tuple: - files = [f for f in os.listdir(temp_dir) if os.path.isfile(os.path.join(temp_dir, f))] - if not files: - raise FileNotFoundError("No file was downloaded") - - actual_file = os.path.join(temp_dir, files[0]) - extension = actual_file.split('.')[-1] - extension = Downloader._sanitize_filename(extension) - - if not filename_user: - filename = os.path.basename(actual_file) - filename = Downloader._sanitize_filename(filename) - if len(filename) > 100: - filename = filename[:100] - else: - filename_user = Downloader._sanitize_filename(filename_user)[:100] - filename = filename_user + '.' + extension - actual_file = os.path.join(temp_dir, filename) - - if not os.path.exists(actual_file): - raise FileNotFoundError(f"Downloaded file not found: {actual_file}") - - return actual_file, filename - - @staticmethod - def _should_upload_to_server(actual_file: str, catbox: bool) -> bool: - return os.path.getsize(actual_file) > 1024*1024*10 or catbox - - @staticmethod - async def _upload_to_server(actual_file: str, filename: str, temp_dir: str) -> dict: - try: - async with aiofiles.open(actual_file, 'rb') as f: - content = await f.read() - files = {'f': (filename, content, 'application/octet-stream')} - headers = {'Cookie': f'cppwd={os.getenv("COPYPARTY_PASSWORD")}'} - params = {'j': '1', 'ck': 'b2'} - base_url = os.getenv("COPYPARTY_URL_UPLOAD") - - response = requests.post(base_url, files=files, headers=headers, params=params) - - if response.status_code in [200, 201]: - return Downloader._handle_upload_success(response, filename, temp_dir) - elif response.status_code == 413: - return Downloader._create_error_response( - "File too large for upload", - "Plik jest za duży, by wysłać go na serwer (limit to ok. 100MB)", - temp_dir - ) - else: - return Downloader._create_error_response( - f"Server returned {response.status_code}", - response.text or 'Unknown error', - temp_dir - ) - except Exception as upload_error: - return Downloader._create_error_response("Upload failed", str(upload_error), temp_dir) - - @staticmethod - def _handle_upload_success(response: requests.Response, filename: str, temp_dir: str) -> dict: - try: - result = response.json() - if 'fileurl' in result: - return { - "status": "success", - "type": "url", - "content": result['fileurl'], - "filename": filename, - "temp_dir": temp_dir - } - else: - return Downloader._create_error_response( - "Invalid response from file server", - str(result), - temp_dir - ) - except ValueError: - return Downloader._create_error_response( - "Failed to parse server response", - response.text[:200], - temp_dir - ) - - @staticmethod - def _return_file_response(actual_file: str, filename: str, temp_dir: str) -> dict: - return { - "status": "success", - "type": "file", - "content": actual_file, - "filename": filename, - "temp_dir": temp_dir - } - - @staticmethod - def _create_error_response(content: str, details: str, temp_dir: str | None) -> dict: - return { - "status": "error", - "type": "error", - "content": content, - "details": details, - "temp_dir": temp_dir - } - - -def setup(bot): - bot.add_cog(Downloader(bot)) -- cgit v1.2.3