# 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 discord import requests from discord.ext import commands import logging import os import asyncio import aiofiles import mimetypes import subprocess import urllib.parse import shlex from typing import Optional from discord import AutocompleteContext from cogs.utils import file_search as search_utils 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 Conversion(commands.Cog): def __init__(self, bot): self.bot = bot from cogs.utils.temp import get_temp_dir from cogs.utils.settings import get_baseurl, get_uploadurl self.temp_dir = get_temp_dir() self._cached_urls = {} self._baseurl = get_baseurl() self._uploadurl = get_uploadurl() async def _file_autocomplete(self, ctx: AutocompleteContext): query = ctx.value.lower() if ctx.value else "" results = await search_utils.UniversalSearcher.search_copyparty_autocomplete(query, max_results=25) user_id = ctx.interaction.user.id self._cached_urls[user_id] = {urllib.parse.unquote(url.split('/')[-1]): url for url in results} choices = [] for url in results[:25]: filename = urllib.parse.unquote(url.split('/')[-1]) display_name = filename[:99] + "…" if len(filename) > 100 else filename choices.append(discord.OptionChoice(name=display_name, value=filename)) return choices async def _download_from_url(self, url: str) -> Optional[str]: """Download file from copyparty URL and return local path""" try: response = requests.get(url) if response.status_code != 200: return None filename = url.split('/')[-1].split('?')[0] file_path = os.path.join(self.temp_dir, filename) async with aiofiles.open(file_path, 'wb') as f: await f.write(response.content) return file_path except Exception as e: logger.error(f"Error downloading from {url}: {e}") return None async def _setup_file_paths(self, plik, output_format): """Setup input and output file paths""" from cogs.utils.temp import get_temp_dir temp_dir = get_temp_dir() file_path = os.path.join(temp_dir, plik.filename) filename_noext = os.path.splitext(plik.filename)[0] await plik.save(file_path) input_path = file_path output_path = os.path.join(temp_dir, f"{filename_noext}.{output_format}") return input_path, output_path def _build_conversion_command(self, file_type, input_path, output_path, args): """Build the appropriate conversion command based on file type""" if file_type in ["Video", "Audio"]: return f"ffmpeg -i {shlex.quote(input_path)} {args} {shlex.quote(output_path)}" elif file_type == "Image": return f"magick {shlex.quote(input_path)} {args} {shlex.quote(output_path)}" else: return None async def _upload_to_copyparty(self, output_path): """Upload file to copyparty and return download URL""" try: async with aiofiles.open(output_path, 'rb') as f: file_content = await f.read() files = { 'f': (os.path.basename(output_path), file_content, 'application/octet-stream') } headers = { 'Cookie': f'cppwd={os.getenv("COPYPARTY_PASSWORD")}' } params = { 'j': '1', 'ck': 'b2' } response = requests.post(self._uploadurl, files=files, headers=headers, params=params) if response.status_code in [200, 201]: result = response.json() if 'fileurl' in result: return result['fileurl'] else: raise ValueError(f"Server response missing fileurl: {result}") else: error_msg = response.text if response.text else 'Unknown error' raise requests.RequestException(f"HTTP {response.status_code}: {error_msg}") except Exception as e: logger.error(f"copyparty upload error: {e}", exc_info=True) raise async def _handle_file_output(self, ctx, output_path): """Handle file output based on size""" if os.path.getsize(output_path) > 1024*1024*10: try: dl_url = await self._upload_to_copyparty(output_path) await ctx.followup.send(f"wysyłon\n{dl_url}\n") except Exception as upload_error: await ctx.followup.send(f"copyparty mnie nienawidzi :cccc\n {str(upload_error)}") else: await ctx.respond(file=discord.File(output_path)) def _cleanup_files(self, *file_paths): """Clean up temporary files""" for path in file_paths: if os.path.exists(path): os.remove(path) @commands.slash_command(name="konwersja", description="konwertuj plik na inny format") @discord.option( name="plik", description="plik z copyparty", required=False, autocomplete=_file_autocomplete ) @discord.option( name="zalacznik", description="plik z discorda", required=False, type=discord.Attachment ) @discord.option( name="serwer?", description="wysłać plik na serwer?", required=False, type=bool, default=False ) @discord.option( name="format", description="format konwersji", required=True, type=str ) async def convert(self, ctx, plik: str = None, zalacznik: discord.Attachment = None, server: bool = False, output_format: str = None): await ctx.defer() input_path = None original_filename = None try: if plik: url = self._cached_urls.get(ctx.author.id, {}).get(plik) if not url: await ctx.respond("Wybierz plik z podpowiedzi", ephemeral=True) return input_path = await self._download_from_url(url) original_filename = plik elif zalacznik: input_path = os.path.join(self.temp_dir, zalacznik.filename) original_filename = zalacznik.filename await zalacznik.save(input_path) else: await ctx.respond("Podaj plik z copyparty lub załącznik", ephemeral=True) return if not input_path or not os.path.exists(input_path): await ctx.respond("Nie udało się pobrać pliku", ephemeral=True) return output_path = os.path.join(self.temp_dir, f"{os.path.splitext(original_filename)[0]}.{output_format}") file_type = filetype(input_path) logger.info(f"Conversion started for: {file_type}") command = self._build_conversion_command(file_type, input_path, output_path, "") logger.info(f"Running: {command}") if not command: await ctx.respond("Plik jest nieobsługiwany", ephemeral=True) return proc = await asyncio.create_subprocess_exec( *shlex.split(command), stdout=asyncio.subprocess.PIPE, stderr=asyncio.subprocess.PIPE ) stdout, stderr = await proc.communicate() if proc.returncode != 0: logger.error(f"FFmpeg error: {stderr.decode() if stderr else 'unknown'}") await ctx.respond(f"Błąd konwersji: {stderr.decode() if stderr else 'unknown'}", ephemeral=True) return if os.path.getsize(output_path) > 1024 * 1024 * 10 or server: dl_url = await self._upload_to_copyparty(output_path) if dl_url: await ctx.respond(f"Skonwertowano\n{dl_url}") else: await ctx.respond("Nie udało się wysłać na serwer", ephemeral=True) else: await ctx.respond(file=discord.File(output_path, filename=f"{original_filename}")) logger.info(f"Conversion completed for: {file_type}") except subprocess.CalledProcessError as e: await ctx.respond(f"Conversion failed: {str(e)}", ephemeral=True) except Exception as e: await ctx.respond(f"An error occurred: {str(e)}", ephemeral=True) finally: if input_path and os.path.exists(input_path): os.remove(input_path) if 'output_path' in locals() and os.path.exists(output_path): os.remove(output_path) def setup(bot): bot.add_cog(Conversion(bot))