76 lines
2.3 KiB
Python
76 lines
2.3 KiB
Python
import subprocess
|
|
import os
|
|
import requests
|
|
from pathlib import Path
|
|
from config import settings
|
|
|
|
def detect_query_type(query: str):
|
|
if "youtube.com/playlist" in query or "list=" in query:
|
|
return "playlist"
|
|
elif "youtube.com/watch" in query or "youtu.be/" in query:
|
|
return "video"
|
|
else:
|
|
return "search"
|
|
|
|
def download_song(query: str, single: bool = False):
|
|
Path(settings.MUSIC_DIR).mkdir(parents=True, exist_ok=True)
|
|
|
|
query_type = detect_query_type(query)
|
|
|
|
if query_type == "video":
|
|
yt_query = query
|
|
playlist_flag = "--no-playlist"
|
|
elif query_type == "playlist":
|
|
yt_query = query
|
|
playlist_flag = None
|
|
else:
|
|
yt_query = f"ytsearch1:{query.strip()}" if single else f"ytsearch20:{query.strip()}"
|
|
playlist_flag = "--no-playlist"
|
|
|
|
output_template = str(Path(settings.MUSIC_DIR) / "%(artist)s - %(title)s.%(ext)s")
|
|
|
|
command = [
|
|
"yt-dlp",
|
|
yt_query,
|
|
"--extract-audio",
|
|
"--audio-format", "mp3",
|
|
"--add-metadata",
|
|
"--output", output_template,
|
|
"--print", "%(title)s",
|
|
"--quiet"
|
|
]
|
|
|
|
if playlist_flag:
|
|
command.append(playlist_flag)
|
|
|
|
print(f"🎵 Starting download for: {query} (single={single})")
|
|
print(f"🛠️ Command: {' '.join(command)}")
|
|
|
|
result = subprocess.run(command, capture_output=True, text=True)
|
|
|
|
if result.returncode != 0:
|
|
print(result.stderr)
|
|
raise Exception(f"❌ Download failed:\n{result.stderr}")
|
|
|
|
downloaded_titles = [line.strip() for line in result.stdout.strip().split("\n") if line.strip()]
|
|
print(f"✅ Finished downloading:\n" + "\n".join(downloaded_titles))
|
|
|
|
for title in downloaded_titles:
|
|
filename = f"{title}.mp3"
|
|
full_path = os.path.join(settings.MUSIC_DIR, filename)
|
|
exists = os.path.exists(full_path)
|
|
print(f"📁 Checking file: {filename} → {'✅ Exists' if exists else '❌ Not found'}")
|
|
|
|
if settings.NAVIDROME_SCAN_URL:
|
|
try:
|
|
res = requests.get(settings.NAVIDROME_SCAN_URL, timeout=5)
|
|
res.raise_for_status()
|
|
print("🔄 Navidrome rescan triggered.")
|
|
except Exception as e:
|
|
print(f"⚠️ Failed to trigger Navidrome rescan: {e}")
|
|
|
|
return {
|
|
"query": query,
|
|
"downloaded": downloaded_titles
|
|
}
|