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
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
|
# 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))
|
