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
|
# 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 asyncio
import logging
import os
import yt_dlp
from youtubesearchpython import VideosSearch
logger = logging.getLogger(__name__)
class MediaDownloader:
"""Handles downloading media from various sources."""
@staticmethod
async def download(ctx, url):
"""Download audio from YouTube URL."""
# Create temp directory if it doesn't exist
temp_dir = os.path.join("cogs", "temp", "music")
os.makedirs(temp_dir, exist_ok=True)
ytdl_opts = {
"outtmpl": os.path.join(temp_dir, "%(title)s.%(ext)s"),
"postprocessors": [
{
"key": "FFmpegExtractAudio",
"preferredcodec": "mp3",
"preferredquality": "192",
}
],
"force_ipv4": True,
}
try:
with yt_dlp.YoutubeDL(ytdl_opts) as ydl:
info_dict = ydl.extract_info(url, download=True)
# Get the base filename without extension
base_filename = os.path.splitext(ydl.prepare_filename(info_dict))[0]
# Add .mp3 extension since we're extracting audio
filename = f"{base_filename}.mp3"
logger.info(f"Downloaded file: {filename}")
return filename
except Exception as e:
logger.error(f"Error downloading video: {e}")
await ctx.respond("eror pobirajync wideło", ephemeral=True)
@staticmethod
async def get_url(ctx, query):
"""Search YouTube and get URL for a query."""
search = VideosSearch(query, limit=1)
if search.result()["result"]:
title = search.result()["result"][0]["title"]
url = search.result()["result"][0]["link"]
logger.info(f"Found video: {title}")
return url, title
else:
logger.warning("No results found for the query.")
await ctx.respond("nic ni ma", ephemeral=True)
return None
class AudioCompressor:
"""Handles audio compression using FFmpeg."""
@staticmethod
async def compress(ctx, filename):
"""Compress audio file to opus format."""
if not filename:
await ctx.respond("dzie wideło", ephemeral=True)
return None
# Create temp directory if it doesn't exist
temp_dir = os.path.join(
os.path.dirname(os.path.dirname(os.path.abspath(__file__))),
"temp",
"compressed",
)
os.makedirs(temp_dir, exist_ok=True)
# Create a temporary output filename in the temp directory
base_name = os.path.basename(filename)
output_filename = os.path.join(
temp_dir, f"{os.path.splitext(base_name)[0]}_compressed.opus"
)
# Ensure the output directory exists
os.makedirs(os.path.dirname(output_filename) or ".", exist_ok=True)
# Use proper quoting for filenames with spaces
ffmpeg_command = [
"ffmpeg",
"-y", # Overwrite output file if it exists
"-i",
filename,
"-ac",
"1", # Mono audio
"-b:a",
"500", # 500k bitrate
"-acodec",
"libopus",
output_filename,
]
try:
logger.info(f"Running FFmpeg command: {' '.join(ffmpeg_command)}")
process = await asyncio.create_subprocess_exec(
*ffmpeg_command,
stdout=asyncio.subprocess.PIPE,
stderr=asyncio.subprocess.PIPE,
)
_, stderr = await process.communicate()
if process.returncode != 0:
error_msg = stderr.decode()
logger.error(f"FFmpeg error ({process.returncode}): {error_msg}")
await ctx.respond("eror kompresujync pilk", ephemeral=True)
return None
if not os.path.exists(output_filename):
logger.error(f"Output file was not created: {output_filename}")
await ctx.respond(
"eror kompresujync pilk - output file not created", ephemeral=True
)
return None
logger.info(f"Successfully compressed file: {output_filename}")
return output_filename
except Exception as e:
logger.error(f"Error compressing file: {str(e)}", exc_info=True)
await ctx.respond(f"eror kompresujync pilk: {str(e)}", ephemeral=True)
return None
|
