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
|
#!/bin/python3
import argparse
import os
import re
import subprocess
import sys
import threading
from concurrent.futures import ThreadPoolExecutor, as_completed
from mutagen.easyid3 import EasyID3
from mutagen.mp3 import MP3
from ytmusicapi import YTMusic
# Initialize the YTMusic API
ytmusic = YTMusic()
# Constants for maximum threads and print lock
MAX_THREADS = 8
print_lock = threading.Lock()
# Function to sanitize file names
def sanitize_filename(name):
return re.sub(r'[\\/*?:"<>|]', "_", name)
# Function to read song names from a custom input file
def read_song_names(filename="$HOME/.config/pamus/list.txt"):
filename = filename.replace("$HOME", os.environ["HOME"]) # Replace $HOME with the full path
with open(filename, "r") as f:
return [
line.split("#", 1)[0].strip()
for line in f
if line.strip() and not line.strip().startswith("#")
]
# Function to download audio from YouTube using yt-dlp
def download_audio(video_id, sanitized_title, output_dir, audio_format, verbose=False):
os.makedirs(output_dir, exist_ok=True)
url = f"https://www.youtube.com/watch?v={video_id}"
output_template = os.path.join(output_dir, f"{sanitized_title}.%(ext)s")
command = [
"yt-dlp",
"-x",
"--audio-format", audio_format, # Use the specified audio format
"--quiet" if not verbose else "",
"--no-warnings",
"--output", output_template,
url
]
subprocess.run(command, check=True)
# Function to tag the audio file with metadata
def tag_audio(file_path, metadata):
audio = MP3(file_path, ID3=EasyID3)
audio["title"] = metadata.get("title", "")
audio["artist"] = ", ".join(metadata.get("artists", []))
audio["album"] = metadata.get("album", "")
audio["genre"] = metadata.get("genre", "")
audio.save()
# Function to process each song
def process_song(song_query, output_dir, audio_format, verbose):
try:
song = ytmusic.search(song_query, filter="songs")[0]
title = song["title"]
sanitized_title = sanitize_filename(title)
audio_path = os.path.join(output_dir, f"{sanitized_title}.{audio_format}")
if os.path.exists(audio_path):
with print_lock:
print(f"Already downloaded: {title}")
return
with print_lock:
print(f"Downloading: {title}")
download_audio(song["videoId"], sanitized_title, output_dir, audio_format, verbose)
metadata = {
"title": title,
"artists": [a["name"] for a in song["artists"]],
"album": song.get("album", {}).get("name", ""),
"genre": song.get("resultType", "")
}
if os.path.exists(audio_path):
tag_audio(audio_path, metadata)
with print_lock:
print(f"Done: {title}")
else:
with print_lock:
print(f"Download failed: {title}", file=sys.stderr)
except Exception as e:
with print_lock:
print(f"Error processing '{song_query}': {e}", file=sys.stderr)
# Main function to handle the execution
def main():
# Parse command-line arguments for output directory, verbosity, input file, and audio format
parser = argparse.ArgumentParser(description="Download and tag songs from YouTube Music")
parser.add_argument(
"-o", "--output-dir",
type=str,
default="$HOME/music", # Default download path
help="Directory to save the downloaded audio files"
)
parser.add_argument(
"-v", "--verbose",
action="store_true",
help="Enable verbose output"
)
parser.add_argument(
"-i", "--input-file",
type=str,
default="$HOME/.config/pamus/list.txt", # Default input file path
help="Input file containing song names to download"
)
parser.add_argument(
"-f", "--audio-format",
type=str,
choices=["mp3", "aac", "flac", "opus", "wav"],
default="mp3",
help="Audio format to download (default: mp3)"
)
args = parser.parse_args()
output_dir = args.output_dir
verbose = args.verbose
input_file = args.input_file
audio_format = args.audio_format
# Replace $HOME with the actual home directory path for input file and output directory
output_dir = output_dir.replace("$HOME", os.environ["HOME"])
input_file = input_file.replace("$HOME", os.environ["HOME"])
song_names = read_song_names(input_file)
with ThreadPoolExecutor(max_workers=MAX_THREADS) as executor:
futures = [executor.submit(process_song, name, output_dir, audio_format, verbose) for name in song_names]
for _ in as_completed(futures):
pass # handled in process_song()
if __name__ == "__main__":
main()
|
