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
|
import mimetypes
import os
import subprocess
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)
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}")
if file_type == "Video":
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}"
elif file_type == "Audio":
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}"
elif file_type == "Image":
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}"
else:
logger.info(f"Unsupported file type: {file_type}")
os.remove(file_path)
return None
# Run the compression
subprocess.run(command, check=True, shell=True)
if 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)
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))
|
