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/music/media_processing.py | 135 ----------------------------------------- 1 file changed, 135 deletions(-) delete mode 100644 cogs/music/media_processing.py (limited to 'cogs/music/media_processing.py') diff --git a/cogs/music/media_processing.py b/cogs/music/media_processing.py deleted file mode 100644 index 91bd406..0000000 --- a/cogs/music/media_processing.py +++ /dev/null @@ -1,135 +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 asyncio -import logging -import os - -import yt_dlp -from youtubesearchpython import VideosSearch -from cogs.utils.url_helper import clean_url_for_ytdlp - -logger = logging.getLogger(__name__) - - -class MediaDownloader: - """Handles downloading media from various sources.""" - - @staticmethod - async def download(ctx, url): - """Download audio from YouTube URL.""" - from cogs.utils.temp import get_temp_dir - - temp_dir = get_temp_dir("senator_sperma_music") - - ytdl_opts = { - "outtmpl": os.path.join(temp_dir, "%(title)s.%(ext)s"), - "postprocessors": [ - { - "key": "FFmpegExtractAudio", - "preferredcodec": "mp3", - "preferredquality": "192", - } - ], - "force_ipv4": True, - } - - try: - url = clean_url_for_ytdlp(url) - with yt_dlp.YoutubeDL(ytdl_opts) as ydl: - info_dict = ydl.extract_info(url, download=True) - # Get the base filename without extension - base_filename = os.path.splitext(ydl.prepare_filename(info_dict))[0] - # Add .mp3 extension since we're extracting audio - filename = f"{base_filename}.mp3" - logger.info(f"Downloaded file: {filename}") - return filename - except Exception as e: - logger.error(f"Error downloading video: {e}") - await ctx.respond("eror pobirajync wideło", ephemeral=True) - - @staticmethod - async def get_url(ctx, query): - """Search YouTube and get URL for a query.""" - search = VideosSearch(query, limit=1) - if search.result()["result"]: - title = search.result()["result"][0]["title"] - url = search.result()["result"][0]["link"] - logger.info(f"Found video: {title}") - return url, title - else: - logger.warning("No results found for the query.") - await ctx.respond("nic ni ma", ephemeral=True) - return None - - -class AudioCompressor: - """Handles audio compression using FFmpeg.""" - - @staticmethod - async def compress(ctx, filename): - """Compress audio file to opus format.""" - from cogs.utils.temp import get_temp_dir - - if not filename: - await ctx.respond("dzie wideło", ephemeral=True) - return None - - temp_dir = get_temp_dir("senator_sperma_compressed") - - # Create a temporary output filename in the temp directory - base_name = os.path.basename(filename) - output_filename = os.path.join( - temp_dir, f"{os.path.splitext(base_name)[0]}_compressed.opus" - ) - - # Ensure the output directory exists - os.makedirs(os.path.dirname(output_filename) or ".", exist_ok=True) - - # Use proper quoting for filenames with spaces - ffmpeg_command = [ - "ffmpeg", - "-y", # Overwrite output file if it exists - "-i", - filename, - "-ac", - "1", # Mono audio - "-b:a", - "500", # 500k bitrate - "-acodec", - "libopus", - output_filename, - ] - - try: - logger.info(f"Running FFmpeg command: {' '.join(ffmpeg_command)}") - process = await asyncio.create_subprocess_exec( - *ffmpeg_command, - stdout=asyncio.subprocess.PIPE, - stderr=asyncio.subprocess.PIPE, - ) - - _, stderr = await process.communicate() - - if process.returncode != 0: - error_msg = stderr.decode() - logger.error(f"FFmpeg error ({process.returncode}): {error_msg}") - await ctx.respond("eror kompresujync pilk", ephemeral=True) - return None - - if not os.path.exists(output_filename): - logger.error(f"Output file was not created: {output_filename}") - await ctx.respond( - "eror kompresujync pilk - output file not created", ephemeral=True - ) - return None - - logger.info(f"Successfully compressed file: {output_filename}") - return output_filename - - except Exception as e: - logger.error(f"Error compressing file: {str(e)}", exc_info=True) - await ctx.respond(f"eror kompresujync pilk: {str(e)}", ephemeral=True) - return None -- cgit v1.2.3