From 9cf5a8c57d0b6725c5ea43dd31565e8ffa872555 Mon Sep 17 00:00:00 2001 From: Xoconoch Date: Sat, 23 Aug 2025 07:52:12 -0600 Subject: [PATCH] fix: m3u file reappended everything when re-downloading the same playlist. --- deezspot/libutils/write_m3u.py | 127 ++++++++++++++++++++++++++++----- 1 file changed, 110 insertions(+), 17 deletions(-) diff --git a/deezspot/libutils/write_m3u.py b/deezspot/libutils/write_m3u.py index afcbde1..eabbeef 100644 --- a/deezspot/libutils/write_m3u.py +++ b/deezspot/libutils/write_m3u.py @@ -88,32 +88,125 @@ def _get_track_info(track: Track) -> tuple: return 'Unknown Artist', 'Unknown Title' +def _read_m3u_entries(m3u_path: str) -> List[tuple]: + """Parse existing m3u into a list of (extinf_line, path_line) entries after header.""" + entries: List[tuple] = [] + if not os.path.exists(m3u_path): + return entries + try: + with open(m3u_path, "r", encoding="utf-8") as f: + lines = [line.rstrip('\n') for line in f] + except OSError: + return entries + # Skip header if present + idx = 0 + if idx < len(lines) and lines[idx].strip() == "#EXTM3U": + idx += 1 + while idx < len(lines): + line = lines[idx] + if not line: + idx += 1 + continue + if line.startswith("#EXTINF:"): + extinf = line + path = "" + if idx + 1 < len(lines): + path = lines[idx + 1] + entries.append((extinf, path)) + idx += 2 + else: + # Path-only entry + entries.append(("", line)) + idx += 1 + return entries + + +def _write_m3u_entries(m3u_path: str, entries: List[tuple]) -> None: + """Write header and provided entries back to file.""" + # Ensure folder exists + os.makedirs(os.path.dirname(m3u_path), exist_ok=True) + with open(m3u_path, "w", encoding="utf-8") as m3u_file: + m3u_file.write("#EXTM3U\n") + for extinf, path in entries: + # Skip empty placeholders + if not path and not extinf: + continue + if extinf: + m3u_file.write(f"{extinf}\n") + if path: + m3u_file.write(f"{path}\n") + + def append_track_to_m3u(m3u_path: str, track: Union[str, Track]) -> None: - """Append a single track to m3u with EXTINF and a resolved path.""" + """Append a single track to m3u with EXTINF and a resolved path. + Idempotent behavior: if entry for same path exists, it is updated/moved to the desired position; if an entry exists at the desired position but differs, it is overwritten. + """ ensure_m3u_header(m3u_path) + # Prepare entries and base dir + playlist_m3u_dir = os.path.dirname(m3u_path) + entries = _read_m3u_entries(m3u_path) + + # Handle simple string path case: dedupe by path, append if new if isinstance(track, str): resolved = _resolve_existing_song_path(track) if not resolved: return - playlist_m3u_dir = os.path.dirname(m3u_path) relative_path = os.path.relpath(resolved, start=playlist_m3u_dir) - with open(m3u_path, "a", encoding="utf-8") as m3u_file: - m3u_file.write(f"{relative_path}\n") + # Remove existing entries with same path + new_entries = [(e, p) for (e, p) in entries if p != relative_path] + # If nothing changed and path already existed identically, skip write + if len(new_entries) == len(entries) and any(p == relative_path for _, p in entries): + return + new_entries.append(("", relative_path)) + _write_m3u_entries(m3u_path, new_entries) + return + + # Validate Track object + if (not isinstance(track, Track) or + not track.success or + not hasattr(track, 'song_path')): + return + + resolved = _resolve_existing_song_path(track.song_path) + if not resolved: + return + + relative_path = os.path.relpath(resolved, start=playlist_m3u_dir) + duration = _get_track_duration_seconds(track) + artist, title = _get_track_info(track) + extinf_line = f"#EXTINF:{duration},{artist} - {title}" + new_entry = (extinf_line, relative_path) + + # Determine target playlist position (1-based). Fallback to 0 (append behavior) if missing/invalid. + position = 0 + try: + if hasattr(track, 'tags') and track.tags: + raw_pos = track.tags.get('playlistnum') + if raw_pos is not None: + position = int(raw_pos) + except (ValueError, TypeError): + position = 0 + + # Remove duplicates by path first (to avoid multiple occurrences upon re-download) + entries = [(e, p) for (e, p) in entries if p != relative_path] + + if position >= 1 and position - 1 < len(entries): + # If there is already something at that position, only overwrite if different + current = entries[position - 1] + if current != new_entry: + entries[position - 1] = new_entry + else: + # Exact match at the correct position: nothing to do + return else: - if (not isinstance(track, Track) or - not track.success or - not hasattr(track, 'song_path')): + # If position beyond current length or not provided: append unless identical exists (path dedup above already handled) + # Avoid appending if an identical entry already exists (by both extinf and path) + if any(e == new_entry for e in entries): return - resolved = _resolve_existing_song_path(track.song_path) - if not resolved: - return - playlist_m3u_dir = os.path.dirname(m3u_path) - relative_path = os.path.relpath(resolved, start=playlist_m3u_dir) - duration = _get_track_duration_seconds(track) - artist, title = _get_track_info(track) - with open(m3u_path, "a", encoding="utf-8") as m3u_file: - m3u_file.write(f"#EXTINF:{duration},{artist} - {title}\n") - m3u_file.write(f"{relative_path}\n") + # If position is within range but entries is shorter, we won't insert placeholders; we simply append to end. + entries.append(new_entry) + + _write_m3u_entries(m3u_path, entries) def write_tracks_to_m3u(output_dir: str, playlist_name: str, tracks: List[Track]) -> str: