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
|
# 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 asyncio
import logging
from pathlib import Path
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
|
