1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
|
# 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
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
async def _setup_file_paths(self, plik, output_format):
"""Setup input and output file paths"""
bot_dir = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
temp_dir = os.path.join(bot_dir, "temp")
if not os.path.exists(temp_dir):
os.makedirs(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}k.{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 {input_path} {args} {output_path}"
elif file_type == "Image":
return f"magick {input_path} {args} {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:
files = {
'f': (output_path, f, 'application/octet-stream')
}
headers = {
'Cookie': f'cppwd={os.getenv("COPYPARTY_PASSWORD")}'
}
params = {
'j': '1',
'ck': 'b2'
}
base_url = "https://catbox.2137697.xyz/sperma/"
response = requests.post(base_url, 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(
"plik",
description="plik do konwersji",
required=True,
type=discord.Attachment
)
@discord.option(
"format",
description="format konwersji",
required=True,
type=str
)
async def convert(self, ctx, plik, output_format, args):
await ctx.defer()
try:
input_path, output_path = await self._setup_file_paths(plik, 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, args)
if not command:
await ctx.respond("Plik jest nieobsługiwany", ephemeral=True)
logger.info("Unsupported file type")
return
proc = await asyncio.create_subprocess_exec(*command.split())
await proc.wait()
await self._handle_file_output(ctx, output_path)
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' in locals():
self._cleanup_files(input_path, output_path)
def setup(bot):
bot.add_cog(Conversion(bot))
|
