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/conversion.py | 245 ----------------------------------------------------- 1 file changed, 245 deletions(-) delete mode 100644 cogs/conversion.py (limited to 'cogs/conversion.py') diff --git a/cogs/conversion.py b/cogs/conversion.py deleted file mode 100644 index f9e4c84..0000000 --- a/cogs/conversion.py +++ /dev/null @@ -1,245 +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 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] - filename = urllib.parse.unquote(filename) - 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)) -- cgit v1.2.3