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
|
import mimetypes
import os
import subprocess
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
@commands.message_command(
name="kompresja",
description="skompresuj wideło albo dźwienk :)",
)
async def message_compression(self, ctx, message: discord.Message):
if message.attachments:
for attachment in message.attachments:
await self.compression(ctx, attachment)
else:
await ctx.respond("Brak załączników w wiadomości", 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):
await ctx.defer()
# Save files to the temp directory inside the bot's root directory
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)
await plik.save(file_path)
input_path = file_path
output_path = os.path.join(temp_dir, f"better_{plik.filename}")
file_type = filetype(file_path)
logger.info(f"Compression started for: {file_type}")
if file_type == "Video":
# Execute ffmpeg command to compress video
ffmpeg_command = f"ffmpeg -i {input_path} -c:v libx264 -crf 51 -b:v 100 -vf 'fps=10' -preset ultrafast -c:a libopus -b:a 500 {output_path}"
elif file_type == "Audio":
ffmpeg_command = f"ffmpeg -i {input_path} -ac 1 -b:a 500 -acodec libopus {output_path}"
elif file_type == "Image":
# Use ImageMagick's magick to heavily compress image (JPEG/PNG/GIF)
# For JPEG: quality 5, strip metadata, resize to 30% of the original size
# For PNG: magick to JPEG, then compress
# For GIF: optimize and reduce colors
ext = os.path.splitext(input_path)[1].lower()
if ext in [".jpg", ".jpeg"]:
img_command = f"magick {input_path} -strip -resize 30% -quality 5 {output_path}"
elif ext == ".gif":
# Optimize GIF: reduce colors, optimize layers, and resize
img_command = f"magick {input_path} -coalesce -resize 30% -colors 32 -strip -layers Optimize {output_path}"
else:
# magick non-JPEG images to JPEG for maximum compression
output_path = os.path.splitext(output_path)[0] + ".jpg"
img_command = f"magick {input_path} -strip -resize 30% -quality 5 {output_path}"
ffmpeg_command = img_command
else:
await ctx.respond("Plik jest nieobsługiwany", ephemeral=True)
logger.info("Unsupported file type")
return
try:
subprocess.run(ffmpeg_command, check=True, shell=True)
# Upload the compressed file
await ctx.respond(file=discord.File(output_path))
logger.info(f"Compression completed for: {file_type}")
# Delete compressed file
os.remove(input_path)
os.remove(output_path)
except subprocess.CalledProcessError as e:
await ctx.respond(f"Compression failed: {str(e)}", ephemeral=True)
if os.path.exists(input_path):
os.remove(input_path)
if os.path.exists(output_path):
os.remove(output_path)
def setup(bot):
bot.add_cog(Compression(bot))
|
