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
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
|
# 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 os
import tempfile
import logging
import requests
import discord
import re
import yt_dlp
from discord.ext import commands
import aiofiles
logger = logging.getLogger(__name__)
class Downloader(commands.Cog):
def __init__(self, bot):
self.bot = bot
URL_PATTERN = re.compile(
r'https://(?:[A-Za-z0-9-]+\.)*[A-Za-z0-9-]{1,63}\.[A-Za-z]{2,63}/\S*'
)
@commands.message_command(
name="download",
description="Download and convert video/audio from internet"
)
async def message_download(self, ctx, message: discord.Message):
# check if message has a valid url
url = [m.group(0) for m in self.URL_PATTERN.finditer(message.content)]
if url:
for url in url:
await self.download(ctx=ctx, url=url, fformat="mp4")
else:
await ctx.respond("Brak załączników w wiadomości", ephemeral=True)
@commands.slash_command(
name="download",
description="Download and convert video/audio from internet"
)
async def download(
self,
ctx: discord.ApplicationContext,
url: str = discord.Option(str, description="Link media", required=True),
catbox: bool = discord.Option(bool, description="wysłać na serwer??", required=False, default=False),
fformat: str = discord.Option(str, description="jaki format? (mp4, mp3, wav, etc.)", required=False, default="mp4"),
filename_user: str = discord.Option(str, description="jaka nazwa pliku?", required=False)
):
"""Download and convert media from a URL to the specified format"""
await ctx.defer()
# Call the function and get the response
response = await self.download_standalone(url=url, catbox=catbox, fformat=fformat, filename_user=filename_user)
if isinstance(response, dict):
if response["status"] == "success":
if response["type"] == "url":
await ctx.respond(f"pobioron i wysyłon: {response['content']}")
elif response["type"] == "file":
file_data = response["content"]
await ctx.respond(file=discord.File(file_data))
else:
logger.error(f"Error: {response['content']}")
if 'details' in response:
logger.error(f"Details: {response['details']}")
@staticmethod
async def download_standalone(url: str, catbox: bool = False, fformat: str = "mp4", filename_user: str | None = None):
try:
logger.info(f"Downloading {url} as {fformat}")
with tempfile.TemporaryDirectory(delete=False) as temp_dir:
ydl_opts = Downloader._get_ydl_options(fformat, temp_dir)
with yt_dlp.YoutubeDL(ydl_opts) as ydl:
ydl.download([url])
actual_file, filename = Downloader._find_and_prepare_file(temp_dir, filename_user)
if Downloader._should_upload_to_server(actual_file, catbox):
return await Downloader._upload_to_server(actual_file, filename, temp_dir)
else:
return Downloader._return_file_response(actual_file, filename, temp_dir)
except Exception as e:
return Downloader._create_error_response("Download failed", str(e), None)
@staticmethod
def _get_ydl_options(fformat: str, temp_dir: str) -> dict:
ydl_opts = {
'outtmpl': os.path.join(temp_dir, '%(title)s.%(ext)s'),
'format': 'bestaudio/best' if fformat in ['mp3', 'wav', 'm4a'] else 'bestvideo[ext=mp4]+bestaudio[ext=m4a]/best[ext=mp4]/best',
'merge_output_format': 'mp4',
'postprocessors': [],
'remote-components': 'ejs:github',
'js-runtimes': {'deno': {'path': '/usr/bin/deno'}},
'o': '"%(title)s.%(ext)s"'
}
if fformat in ['mp3', 'wav', 'm4a']:
ydl_opts['postprocessors'].append({
'key': 'FFmpegExtractAudio',
'preferredcodec': fformat,
'preferredquality': '192',
})
elif fformat != 'mp4':
ydl_opts['postprocessors'].append({
'key': 'FFmpegVideoConvertor',
'preferedformat': fformat,
})
return ydl_opts
@staticmethod
def _find_and_prepare_file(temp_dir: str, filename_user: str) -> tuple:
files = [f for f in os.listdir(temp_dir) if os.path.isfile(os.path.join(temp_dir, f))]
if not files:
raise FileNotFoundError("No file was downloaded")
actual_file = os.path.join(temp_dir, files[0])
extension = actual_file.split('.')[-1]
if not filename_user:
filename = os.path.basename(actual_file)
else:
filename = filename_user + '.' + extension
actual_file = os.path.join(temp_dir, filename)
if not os.path.exists(actual_file):
raise FileNotFoundError(f"Downloaded file not found: {actual_file}")
return actual_file, filename
@staticmethod
def _should_upload_to_server(actual_file: str, catbox: bool) -> bool:
return os.path.getsize(actual_file) > 1024*1024*10 or catbox
@staticmethod
async def _upload_to_server(actual_file: str, filename: str, temp_dir: str) -> dict:
try:
async with aiofiles.open(actual_file, 'rb') as f:
content = await f.read()
files = {'f': (filename, content, '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]:
return Downloader._handle_upload_success(response, filename, temp_dir)
else:
return Downloader._create_error_response(
f"Server returned {response.status_code}",
response.text or 'Unknown error',
temp_dir
)
except Exception as upload_error:
return Downloader._create_error_response("Upload failed", str(upload_error), temp_dir)
@staticmethod
def _handle_upload_success(response: requests.Response, filename: str, temp_dir: str) -> dict:
try:
result = response.json()
if 'fileurl' in result:
return {
"status": "success",
"type": "url",
"content": result['fileurl'],
"filename": filename,
"temp_dir": temp_dir
}
else:
return Downloader._create_error_response(
"Invalid response from file server",
str(result),
temp_dir
)
except ValueError:
return Downloader._create_error_response(
"Failed to parse server response",
response.text[:200],
temp_dir
)
@staticmethod
def _return_file_response(actual_file: str, filename: str, temp_dir: str) -> dict:
return {
"status": "success",
"type": "file",
"content": actual_file,
"filename": filename,
"temp_dir": temp_dir
}
@staticmethod
def _create_error_response(content: str, details: str, temp_dir: str) -> dict:
return {
"status": "error",
"type": "error",
"content": content,
"details": details,
"temp_dir": temp_dir
}
def setup(bot):
bot.add_cog(Downloader(bot))
|
