# 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 mimetypes import os import asyncio from typing import List, Optional from os import PathLike import discord from discord.ext import commands import logging logger = logging.getLogger(__name__) def filetype(filename): mime_type, _ = mimetypes.guess_type(filename) if mime_type: if mime_type.startswith('video/'): return "Video" elif mime_type.startswith('audio/'): return "Audio" elif mime_type.startswith('image/'): return "Image" return "Unknown" class Compression(commands.Cog): def __init__(self, bot): self.bot = bot self.temp_dir = os.path.join(os.path.dirname(os.path.dirname(os.path.abspath(__file__))), "temp") if not os.path.exists(self.temp_dir): os.makedirs(self.temp_dir) def _process_video(self, file_path: str, output_path: str) -> str: """Process video compression""" command = f"ffmpeg -y -i {file_path} -c:v libx264 -crf 51 -b:v 100 -vf 'fps=10,scale=trunc(iw*0.5/2)*2:trunc(ih*0.5/2)*2' -preset ultrafast -c:a libopus -b:a 500 {output_path}" return command def _process_audio(self, file_path: str, output_path: str) -> str: """Process audio compression""" if output_path.lower().endswith('.mp3'): output_path = os.path.splitext(output_path)[0] + os.path.splitext(output_path)[1].replace('.mp3', '') if not output_path.lower().endswith('.ogg'): output_path = os.path.splitext(output_path)[0] + '.ogg' command = f"ffmpeg -y -i {file_path} -c:a libopus -b:a 500 -ar 12000 -ac 1 {output_path}" return command def _process_image(self, file_path: str, output_path: str) -> tuple[str, str]: """Process image compression""" ext = os.path.splitext(file_path)[1].lower() if ext in [".jpg", ".jpeg"]: command = f"magick {file_path} -strip -resize 30% -quality 5 {output_path}" elif ext == ".gif": command = f"magick {file_path} -coalesce -resize 30% -colors 32 -strip -layers Optimize {output_path}" else: output_path = os.path.splitext(output_path)[0] + ".jpg" command = f"magick {file_path} -strip -resize 30% -quality 5 {output_path}" return command, output_path async def _execute_compression(self, command: str) -> bool: """Execute compression command""" proc = await asyncio.create_subprocess_exec(command) await proc.wait() return proc.returncode == 0 async def _compress_attachment(self, attachment: discord.Attachment) -> Optional[discord.File]: """Compress a single attachment and return the compressed file""" file_path = PathLike[str] try: file_path: PathLike[str] | str = os.path.join(self.temp_dir, attachment.filename) output_path = os.path.join(self.temp_dir, f"better_{attachment.filename}") # Save the original file await attachment.save(file_path) # Process the file file_type = filetype(file_path) logger.info(f"Compression started for: {file_type}") command = self._get_compression_command(file_type, file_path, output_path) if command is None: logger.info(f"Unsupported file type: {file_type}") os.remove(file_path) return None # Update output_path if it was modified if file_type == "Image": command, output_path = command elif file_type == "Audio": # output_path is already modified in _process_audio pass # Run the compression success = await self._execute_compression(command) if success and os.path.exists(output_path): logger.info(f"Successfully compressed: {attachment.filename}") return discord.File(output_path, filename=f"compressed_{attachment.filename}") return None except Exception as e: logger.error(f"Error processing {attachment.filename}: {str(e)}") return None finally: # Clean up temporary files if 'file_path' in locals() and os.path.exists(file_path): os.remove(file_path) def _get_compression_command(self, file_type: str, file_path: str, output_path: str): """Get the appropriate compression command based on file type""" if file_type == "Video": return self._process_video(file_path, output_path) elif file_type == "Audio": return self._process_audio(file_path, output_path) elif file_type == "Image": return self._process_image(file_path, output_path) else: return None async def _compress_attachments(self, attachments: List[discord.Attachment]) -> List[discord.File]: """Compress multiple attachments and return list of compressed files""" compressed_files = [] for attachment in attachments: compressed_file = await self._compress_attachment(attachment) if compressed_file: compressed_files.append(compressed_file) return compressed_files @commands.message_command( name="kompresja", description="skompresuj wideło albo dźwienk :)", ) async def message_compression(self, ctx, message: discord.Message): if not message.attachments: await ctx.respond("Brak załączników w wiadomości", ephemeral=True) return await ctx.defer() compressed_files = await self._compress_attachments(message.attachments) # Send all compressed files in one message if compressed_files: await ctx.respond( f"Skompresowano {len(compressed_files)} z {len(message.attachments)} plików:", files=compressed_files ) # Clean up compressed files for file in compressed_files: if os.path.exists(file.fp.name): os.remove(file.fp.name) else: await ctx.respond("Nie udało się skompresować żadnych plików.", ephemeral=True) @commands.slash_command( name="kompresja", description="skompresuj wideło albo dźwienk :)", ) @discord.option( name="plik", description="plik do skompresowania", required=True, type=discord.Attachment ) async def compression(self, ctx, plik: discord.Attachment): await ctx.defer() compressed_file = await self._compress_attachment(plik) if compressed_file: await ctx.respond(file=compressed_file) if os.path.exists(compressed_file.fp.name): os.remove(compressed_file.fp.name) else: await ctx.respond("Nie udało się skompresować pliku.", ephemeral=True) def setup(bot): bot.add_cog(Compression(bot))