main
1#!/usr/bin/env python3
2"""
3Music Playlist Downloader
4Downloads DJ podcasts/radio shows from Mixcloud and SoundCloud,
5and generates M3U playlists.
6"""
7
8import argparse
9import logging
10import os
11import subprocess
12import sys
13from dataclasses import dataclass
14from pathlib import Path
15from typing import List
16
17import yaml
18
19
20@dataclass
21class MixcloudShow:
22 """Configuration for a Mixcloud show."""
23
24 handle: str
25 artist: str
26 show: str
27 beets_tags: dict = None # Optional per-show metadata
28
29 def __post_init__(self):
30 if self.beets_tags is None:
31 self.beets_tags = {}
32
33
34@dataclass
35class SoundcloudShow:
36 """Configuration for a SoundCloud show."""
37
38 url: str
39 artist: str
40 show: str
41 beets_tags: dict = None # Optional per-show metadata
42
43 def __post_init__(self):
44 if self.beets_tags is None:
45 self.beets_tags = {}
46
47
48@dataclass
49class BeetsConfig:
50 """Beets integration configuration."""
51
52 enable: bool = False
53 import_after_download: bool = True
54 write_tags: bool = True
55 default_tags: dict = None
56
57 def __post_init__(self):
58 if self.default_tags is None:
59 self.default_tags = {}
60
61
62@dataclass
63class Config:
64 """Main configuration."""
65
66 base_dir: Path
67 mixcloud_shows: List[MixcloudShow]
68 soundcloud_shows: List[SoundcloudShow]
69 yt_dlp_options: dict
70 beets: BeetsConfig
71
72
73def load_config(config_path: Path) -> Config:
74 """Load configuration from YAML file."""
75 with open(config_path) as f:
76 data = yaml.safe_load(f)
77
78 mixcloud_shows = [
79 MixcloudShow(**show) for show in data.get("mixcloud_shows", [])
80 ]
81 soundcloud_shows = [
82 SoundcloudShow(**show) for show in data.get("soundcloud_shows", [])
83 ]
84
85 # Load beets config if present
86 beets_data = data.get("beets", {})
87 beets_config = BeetsConfig(
88 enable=beets_data.get("enable", False),
89 import_after_download=beets_data.get("import_after_download", True),
90 write_tags=beets_data.get("write_tags", True),
91 default_tags=beets_data.get("default_tags", {}),
92 )
93
94 return Config(
95 base_dir=Path(data.get("base_dir", "/neo/music")),
96 mixcloud_shows=mixcloud_shows,
97 soundcloud_shows=soundcloud_shows,
98 yt_dlp_options=data.get("yt_dlp_options", {}),
99 beets=beets_config,
100 )
101
102
103def build_yt_dlp_command(
104 url: str,
105 output_template: str,
106 artist: str,
107 archive_file: Path,
108 yt_dlp_options: dict,
109) -> List[str]:
110 """Build yt-dlp command with options."""
111 cmd = ["yt-dlp"]
112
113 # Add configured options
114 if yt_dlp_options.get("ignore_errors", True):
115 cmd.append("-i")
116 if yt_dlp_options.get("continue", True):
117 cmd.append("-c")
118
119 format_opt = yt_dlp_options.get("format", "best")
120 cmd.extend(["-f", format_opt])
121
122 if yt_dlp_options.get("add_metadata", True):
123 cmd.append("--add-metadata")
124 if yt_dlp_options.get("embed_thumbnail", True):
125 cmd.append("--embed-thumbnail")
126
127 # Download archive for deduplication
128 cmd.extend(["--download-archive", str(archive_file)])
129
130 # Parse metadata
131 cmd.extend(
132 [
133 "--parse-metadata",
134 "title:%(album)s",
135 "--parse-metadata",
136 f"{artist}:%(artist)s",
137 ]
138 )
139
140 # Output template
141 cmd.extend(["--output", output_template, url])
142
143 return cmd
144
145
146def download_mixcloud_show(
147 show: MixcloudShow, base_dir: Path, yt_dlp_options: dict
148) -> bool:
149 """Download a Mixcloud show."""
150 url = f"https://www.mixcloud.com/{show.handle}/"
151 output_dir = base_dir / show.show
152 output_dir.mkdir(parents=True, exist_ok=True)
153
154 output_template = str(output_dir / "%(title)s-%(id)s.%(ext)s")
155 archive_file = output_dir / ".downloaded.txt"
156
157 logging.info(f"Downloading {show.show} by {show.artist}...")
158 cmd = build_yt_dlp_command(
159 url, output_template, show.artist, archive_file, yt_dlp_options
160 )
161
162 try:
163 subprocess.run(cmd, check=True, capture_output=False)
164 return True
165 except subprocess.CalledProcessError as e:
166 logging.warning(f"Failed to download {show.show}: {e}")
167 return False
168
169
170def download_soundcloud_show(
171 show: SoundcloudShow, base_dir: Path, yt_dlp_options: dict
172) -> bool:
173 """Download a SoundCloud show."""
174 output_dir = base_dir / show.show
175 output_dir.mkdir(parents=True, exist_ok=True)
176
177 output_template = str(output_dir / "%(title)s-%(id)s.%(ext)s")
178 archive_file = output_dir / ".downloaded.txt"
179
180 logging.info(f"Downloading {show.show} by {show.artist}...")
181 cmd = build_yt_dlp_command(
182 show.url, output_template, show.artist, archive_file, yt_dlp_options
183 )
184
185 try:
186 subprocess.run(cmd, check=True, capture_output=False)
187 return True
188 except subprocess.CalledProcessError as e:
189 logging.warning(f"Failed to download {show.show}: {e}")
190 return False
191
192
193def generate_playlist(
194 artist: str, show: str, base_dir: Path, playlist_dir: Path
195):
196 """Generate M3U playlist for a show."""
197 show_dir = base_dir / show
198 if not show_dir.exists():
199 logging.warning(f"Show directory does not exist: {show_dir}")
200 return
201
202 # Find all audio files (exclude archive file)
203 audio_extensions = {".m4a", ".mp3", ".opus", ".ogg", ".flac"}
204 audio_files = sorted(
205 [
206 f
207 for f in show_dir.iterdir()
208 if f.is_file()
209 and f.suffix.lower() in audio_extensions
210 and not f.name.startswith(".") # Exclude hidden files
211 ]
212 )
213
214 if not audio_files:
215 logging.warning(f"No audio files found in {show_dir}")
216 return
217
218 # Generate playlist filename
219 playlist_name = f"Mix - {artist} - {show}.m3u"
220 playlist_path = playlist_dir / playlist_name
221
222 logging.info(
223 f"Generating playlist: {playlist_name} ({len(audio_files)} files)"
224 )
225
226 # Write M3U playlist with relative paths
227 with open(playlist_path, "w", encoding="utf-8") as f:
228 f.write("#EXTM3U\n")
229 for audio_file in audio_files:
230 # Use relative path from playlist to audio file
231 relative_path = os.path.relpath(audio_file, playlist_dir)
232 f.write(f"{relative_path}\n")
233
234
235def import_to_beets(
236 base_dir: Path,
237 artist: str,
238 show: str,
239 show_beets_tags: dict,
240 beets_config: BeetsConfig,
241) -> bool:
242 """Import show to beets database with merged metadata."""
243 if not beets_config.enable:
244 return True # Skip if disabled
245
246 show_dir = base_dir / show
247 if not show_dir.exists():
248 logging.warning(f"Show directory does not exist: {show_dir}")
249 return False
250
251 # Merge tags: default_tags < show_beets_tags
252 merged_tags = {**beets_config.default_tags, **show_beets_tags}
253
254 # Always set artist and album from show config
255 merged_tags["artist"] = artist
256 merged_tags["album"] = show
257
258 # Build beets import command
259 cmd = [
260 "beet",
261 "import",
262 "-C", # Don't move files (keep in place)
263 "-A", # Don't autotag (skip MusicBrainz)
264 "-q", # Quiet mode
265 ]
266
267 # Add all merged tags
268 for key, value in merged_tags.items():
269 cmd.extend(["--set", f"{key}={value}"])
270
271 cmd.append(str(show_dir))
272
273 try:
274 result = subprocess.run(
275 cmd, check=True, capture_output=True, text=True
276 )
277 logging.debug(f"Beets import output: {result.stdout}")
278
279 # Write metadata to file tags if enabled
280 if beets_config.write_tags:
281 write_cmd = ["beet", "write", "-q", f"album:{show}"]
282 subprocess.run(write_cmd, check=False, capture_output=True)
283
284 logging.info(f"✓ Imported {show} to beets")
285 return True
286 except subprocess.CalledProcessError as e:
287 logging.warning(f"Failed to import {show} to beets: {e.stderr}")
288 return False
289
290
291def main():
292 """Main entry point."""
293 parser = argparse.ArgumentParser(
294 description="Download music podcasts and generate playlists"
295 )
296 parser.add_argument(
297 "--config",
298 type=Path,
299 default="/neo/music/music-playlist-dl.yaml",
300 help="Path to configuration file",
301 )
302 parser.add_argument(
303 "--verbose", "-v", action="store_true", help="Enable verbose logging"
304 )
305 parser.add_argument(
306 "--import-existing",
307 action="store_true",
308 help="Import all existing files to beets (run once after enabling)",
309 )
310 args = parser.parse_args()
311
312 # Setup logging
313 logging.basicConfig(
314 level=logging.DEBUG if args.verbose else logging.INFO,
315 format="%(asctime)s - %(levelname)s - %(message)s",
316 )
317
318 # Load configuration
319 try:
320 config = load_config(args.config)
321 except Exception as e:
322 logging.error(f"Failed to load configuration: {e}")
323 sys.exit(1)
324
325 # Setup directories
326 base_dir = config.base_dir
327 playlist_dir = base_dir.parent / "playlists"
328 base_dir.mkdir(parents=True, exist_ok=True)
329 playlist_dir.mkdir(parents=True, exist_ok=True)
330
331 logging.info(
332 f"Starting music podcast downloads to: {base_dir}"
333 )
334 logging.info("=" * 60)
335
336 # Download Mixcloud shows
337 for show in config.mixcloud_shows:
338 download_mixcloud_show(show, base_dir, config.yt_dlp_options)
339
340 # Download SoundCloud shows
341 for show in config.soundcloud_shows:
342 download_soundcloud_show(show, base_dir, config.yt_dlp_options)
343
344 logging.info("=" * 60)
345 logging.info("Generating playlists...")
346
347 # Generate playlists for all shows
348 for show in config.mixcloud_shows:
349 generate_playlist(show.artist, show.show, base_dir, playlist_dir)
350
351 for show in config.soundcloud_shows:
352 generate_playlist(show.artist, show.show, base_dir, playlist_dir)
353
354 # Import to beets if enabled
355 if config.beets.enable:
356 if args.import_existing:
357 logging.info("=" * 60)
358 logging.info("Importing all existing files to beets...")
359 for show in config.mixcloud_shows:
360 import_to_beets(
361 base_dir,
362 show.artist,
363 show.show,
364 show.beets_tags,
365 config.beets,
366 )
367 for show in config.soundcloud_shows:
368 import_to_beets(
369 base_dir,
370 show.artist,
371 show.show,
372 show.beets_tags,
373 config.beets,
374 )
375 logging.info("Beets import complete!")
376 sys.exit(0)
377
378 elif config.beets.import_after_download:
379 logging.info("=" * 60)
380 logging.info("Importing new downloads to beets...")
381 for show in config.mixcloud_shows:
382 import_to_beets(
383 base_dir,
384 show.artist,
385 show.show,
386 show.beets_tags,
387 config.beets,
388 )
389 for show in config.soundcloud_shows:
390 import_to_beets(
391 base_dir,
392 show.artist,
393 show.show,
394 show.beets_tags,
395 config.beets,
396 )
397
398 logging.info("=" * 60)
399 logging.info("Download complete!")
400
401
402if __name__ == "__main__":
403 main()