# 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))