import os import tempfile import logging import requests import discord import re import yt_dlp import json from discord.ext import commands from dotenv import load_dotenv logger = logging.getLogger(__name__) # Load environment variables load_dotenv() 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: 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): try: logger.info(f"Downloading {url} as {fformat}") with tempfile.TemporaryDirectory(delete=False) as temp_dir: # Configure yt-dlp to use the original filename ydl_opts = { 'outtmpl': os.path.join(temp_dir, '%(title)s.%(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:/usr/bin/deno', 'o': '"%(title)s.%(ext)s"' } # Add post-processing for audio formats if fformat in ['mp3', 'wav', 'm4a']: ydl_opts['postprocessors'].append({ 'key': 'FFmpegExtractAudio', 'preferredcodec': fformat, 'preferredquality': '192', }) elif fformat != 'mp4': # For other video formats ydl_opts['postprocessors'].append({ 'key': 'FFmpegVideoConvertor', 'preferedformat': fformat, }) with yt_dlp.YoutubeDL(ydl_opts) as ydl: ydl.download([url]) # Find the downloaded file (should be the only file in the temp directory) files = [f for f in os.listdir(temp_dir) if os.path.isfile(os.path.join(temp_dir, f))] if not files: raise Exception(f"No file was downloaded") actual_file = os.path.join(temp_dir, files[0]) extension = actual_file.split('.')[-1] if not filename_user: filename = os.path.basename(actual_file) else: filename = filename_user + '.' + extension actual_file = os.path.join(temp_dir, filename) if not os.path.exists(actual_file): raise Exception(f"Downloaded file not found: {actual_file}") if os.path.getsize(actual_file) > 1024*1024*10 or catbox: try: # Upload to Copyparty with open(actual_file, 'rb') as f: files = { 'f': (filename, f, 'application/octet-stream') } headers = { 'Cookie': f'cppwd={os.getenv("COPYPARTY_PASSWORD")}' } params = { 'j': '1', # Return JSON response 'ck': 'b2' # Use blake2b checksum } # Ensure the URL ends with a single slash base_url = "https://catbox.2137697.xyz/sperma/" response = requests.post( base_url, files=files, headers=headers, params=params ) if response.status_code == 200 or response.status_code == 201: try: result = response.json() if 'fileurl' in result: dl_url = result['fileurl'] return { "status": "success", "type": "url", "content": dl_url, "filename": filename, "temp_dir": temp_dir } else: return { "status": "error", "type": "error", "content": "Invalid response from file server", "details": str(result), "temp_dir": temp_dir } except ValueError: return { "status": "error", "type": "error", "content": "Failed to parse server response", "details": response.text[:200], # First 200 chars of response "temp_dir": temp_dir } else: error_msg = response.text if response.text else 'Unknown error' return { "status": "error", "type": "error", "content": f"Server returned {response.status_code}", "details": error_msg, "temp_dir": temp_dir } except Exception as upload_error: return { "status": "error", "type": "error", "content": "Upload failed", "details": str(upload_error), "temp_dir": temp_dir } else: # If we have a file to return directly return { "status": "success", "type": "file", "content": actual_file, "filename": filename, "temp_dir": temp_dir } except Exception as e: error_msg = f"Error in download command: {e}" logger.error(error_msg, exc_info=True) return { "status": "error", "type": "error", "content": "Download failed", "details": str(e), "temp_dir": temp_dir } def setup(bot): bot.add_cog(Downloader(bot))