diff --git a/README.md b/README.md index 5fe0320..cabcba4 100644 --- a/README.md +++ b/README.md @@ -78,15 +78,15 @@ After the first login, your auth token will be cached. Tokens are good for 30 da ```sh # Upload tracks -# Overwrite existing matching tracks (match on file title metadata) -light music upload song1 song2 song3 --match-title-by metadata +# Files matching an existing track (by title+artist) are skipped by default +light music upload song1 song2 song3 # Upload tracks -# Overwrite existing matching tracks (match on filename) -light music upload song1 song2 song3 --match-title-by filename +# Delete-and-replace matching existing tracks instead of skipping them +light music upload song1 song2 song3 --overwrite # Upload tracks -# Don't overwrite existing tracks +# Allow uploading duplicate tracks light music upload --allow-duplicates song1 song2 song3 ``` diff --git a/light_api/light_api/music.py b/light_api/light_api/music.py index 22e2fa1..83ef77f 100644 --- a/light_api/light_api/music.py +++ b/light_api/light_api/music.py @@ -7,7 +7,7 @@ import re import tempfile from enum import StrEnum -from typing import Callable, Literal +from typing import Callable from dataclasses import dataclass from mutagen._file import File @@ -266,11 +266,93 @@ def delete_tracks_by_artist_regex(self, pattern: str) -> None: """ self.delete_tracks_predicate(lambda t: bool(re.match(pattern, t.artist))) + def _track_identity(self, file_path: str) -> tuple[str, str]: + """Return (title, artist) for an audio file at file_path. + + If the track has metadata for a field: use that metadata. + If no metadata for a field: titles will fall back to just 'filename', and + artists will fall back to "Unknown", matching the dashboard's implementation. + + Args: + file_path: File path of audio file to process. + + Returns: + (title, artist) tuple representing that file's content. + """ + tags = File(file_path, easy=True) + title = (tags.get("title", [None])[0] if tags else None) or os.path.splitext( + os.path.basename(file_path) + )[0] + artist = (tags.get("artist", [None])[0] if tags else None) or "Unknown" + return title, artist + + def _find_matching_track(self, title: str, artist: str) -> LightTrack | None: + """Find an existing track with an exact (title, artist) match.""" + for t in self._tracks: + if t.title == title and t.artist == artist: + return t + return None + + def find_upload_matches(self, files: list[str]) -> dict[str, LightTrack]: + """Given a list of local audio files, find those that already exist on the device and + return the matching existing LightTrack instances. + + Args: + files: A list of paths to audio files. + + Returns: + A {file_path: LightTrack} dict. + """ + self._init_tracks() + matches: dict[str, LightTrack] = {} + for file_path in files: + title, artist = self._track_identity(file_path) + match = self._find_matching_track(title, artist) + if match is not None: + matches[file_path] = match + return matches + + def _resolve_upload_plan( + self, files: list[str], allow_duplicates: bool, overwrite: bool + ) -> tuple[list[str], list[LightTrack]]: + """Return the subset of files to upload and the subset of tracks to overwrite after + applying allow_duplicates/overwrite behavior flags. + + Returns: + Tuple of lists. First item is list[str] of files to upload. Second item is list[LightTrack] + of files to be overwritten. + """ + if allow_duplicates: + return (files, []) + + matches = self.find_upload_matches(files) + to_delete = list({t.audio_id: t for t in matches.values()}.values()) if overwrite else [] + + to_upload = [] + for file_path in files: + match = matches.get(file_path) + + if match is None or overwrite: + to_upload.append(file_path) + else: + log.info(f"Skipping {file_path!r}: matches existing ({match.title!r}, {match.artist!r})") + + return (to_upload, to_delete) + + @staticmethod + def filter_valid_tracks(files: list[str]) -> tuple[list[str], list[str]]: + """Verify validity of audio files, returning (valid, invalid).""" + valid = [] + invalid = [] + for file_path in files: + (valid if os.path.exists(file_path) else invalid).append(file_path) + return valid, invalid + def upload_tracks( self, files: list[str], allow_duplicates: bool = False, - match_title_by: Literal["metadata", "filename"] = "metadata", + overwrite: bool = False, convert_flac: bool = True, on_progress: "Callable[[str, int, int], None] | None" = None, ) -> None: @@ -278,35 +360,29 @@ def upload_tracks( Args: files: List of paths to audio files to upload. - allow_duplicates: If False (default), existing tracks with matching titles are - deleted before uploading. If True, duplicates are kept. - match_title_by: How to determine a track's title for duplicate detection. - "metadata" (default) reads the title from the file's ID3/audio tags. - "filename" uses the filename (without extension) as the title. + allow_duplicates: If True, skip duplicate checking entirely and always upload, + potentially creating multiple tracks with the same title/artist. + overwrite: If True, delete a file's matching existing track (if any) before + uploading it. If False (default), files matching an existing track + are skipped instead, leaving the existing track untouched. """ - manual_update_cmds = [] - - if not allow_duplicates: - titles: list[str] - - if match_title_by == "metadata": - titles = [] - for s in files: - f = File(s, easy=True) - if f is None: - raise ValueError(f"Could not read metadata from {s}") - titles.append(f.get("title", ["Unknown Title"])[0]) - else: - titles = [os.path.splitext(os.path.basename(s))[0] for s in files] + if overwrite and allow_duplicates: + raise ValueError("overwrite and allow_duplicates are mutually exclusive") - self.delete_tracks_by_title(titles) + valid_files, invalid_files = self.filter_valid_tracks(files) + for file_path in invalid_files: + log.warning(f"File not found, skipping: {file_path}") - for file_path in files: - log.info(f"Uploading {file_path}") + to_upload, to_delete = self._resolve_upload_plan( + valid_files, allow_duplicates, overwrite + ) - if not os.path.exists(file_path): - log.warning(f"File not found, skipping: {file_path}") - continue + if to_delete: + audio_ids = {t.audio_id for t in to_delete} + self.delete_tracks_predicate(lambda t: t.audio_id in audio_ids) + + for file_path in to_upload: + log.info(f"Uploading {file_path}") tmp_path = None try: @@ -367,30 +443,12 @@ def _chunks(path: str, total: int, filename: str): raise RuntimeError( f"Upload {os.path.basename(upload_path)}: {put_resp.status_code} {put_resp.text}" ) - - # The Light API has an issue where it won't set title/artist metadata properly - # when uploading non-mp3 files. Give user list of commands to patch manually after - # upload since files are still processing and a patch immediately after will fail. - if content_type != "audio/mpeg": - path = os.path.basename(upload_path) - tags = File(upload_path, easy=True) - title = tags.get("title", ["Unknown"])[0] if tags else "Unknown" - artist = tags.get("artist", ["Unknown"])[0] if tags else "Unknown" - album = tags.get("album", [""])[0] if tags else "" - cmd = f'light music update "{path}" --new-title "{title}" --new-artist "{artist}" --new-album "{album}"' - manual_update_cmds.append(cmd) finally: if tmp_path: os.unlink(tmp_path) log.info("All uploads complete") - if len(manual_update_cmds) > 0: - log.warning( - "Manual metadata fixes needed:\n" - + "\n".join(f" {cmd} ;" for cmd in manual_update_cmds) - ) - def update_track_metadata( self, audio_id: str, diff --git a/light_cli_tui/light_cli_tui/cli.py b/light_cli_tui/light_cli_tui/cli.py index aae5d84..715c21d 100644 --- a/light_cli_tui/light_cli_tui/cli.py +++ b/light_cli_tui/light_cli_tui/cli.py @@ -43,7 +43,9 @@ def invoke(self, ctx: click.Context): raise -@click.group(cls=JsonAwareGroup, context_settings={"help_option_names": ["-h", "--help"]}) +@click.group( + cls=JsonAwareGroup, context_settings={"help_option_names": ["-h", "--help"]} +) @click.version_option(package_name="light-phone-cli-tui", prog_name="light") @click.option("--email", default=None, help="Light account email address.") @click.option("--email-file", default=None, help="Path to file containing email.") @@ -99,9 +101,7 @@ def cli( `--device-id` (mutually exclusive). """ if (phone_number or phone_number_file) and (device_id or device_id_file): - raise click.UsageError( - "--phone-number and --device-id are mutually exclusive." - ) + raise click.UsageError("--phone-number and --device-id are mutually exclusive.") logging.basicConfig(format="%(name)s %(levelname)s %(message)s") logging.getLogger("light").setLevel(log_level.upper()) @@ -228,6 +228,8 @@ def podcasts_delete(light: Light, title): # -- Music commands ------------------------------------------------------------- +_VERBOSE_LIST_THRESHOLD = 20 + @music.command("upload") @with_light @@ -235,59 +237,67 @@ def podcasts_delete(light: Light, title): @click.option( "--allow-duplicates", is_flag=True, - help="Skip duplicate checking and always upload.", + help="Allow uploading duplicate tracks.", ) @click.option( - "--match-title-by", - "-m", - type=click.Choice(["filename", "metadata"]), - default="metadata", - show_default=True, - help="How to match existing tracks when checking for duplicates.", + "--overwrite", + is_flag=True, + help="Overwrite existing matching tracks.", ) @click.option( "--no-convert-flac", is_flag=True, default=False, - help="Skip FLAC to MP3 conversion (conversion is on by default to preserve metadata).", + help="Skip FLAC to MP3 conversion. Conversion is on by default to preserve metadata.", ) -def music_upload(light: Light, songs, allow_duplicates, match_title_by, no_convert_flac): - """Upload one or more audio files to your device. - - Duplicate detection is on by default - existing tracks with a matching - title will be replaced. Use `--allow-duplicates` to skip this. +@click.option( + "--verbose", + "-v", + is_flag=True, + default=False, + help="Show the full list of affected tracks, even for large batches.", +) +def music_upload(light: Light, songs, allow_duplicates, overwrite, no_convert_flac, verbose): + """Upload audio files to your device. **Example:** `light music upload track1.mp3 track2.mp3` """ - files = list(songs) - - if not allow_duplicates: - from mutagen._file import File as MutagenFile - import os as _os - - if match_title_by == "metadata": - titles = [] - for s in files: - f = MutagenFile(s, easy=True) - titles.append( - f.get("title", [_os.path.splitext(_os.path.basename(s))[0]])[0] - if f - else _os.path.splitext(_os.path.basename(s))[0] - ) + if overwrite and allow_duplicates: + raise click.UsageError( + "--overwrite and --allow-duplicates are mutually exclusive." + ) + + files, invalid_files = light.music.filter_valid_tracks(list(songs)) + for file_path in invalid_files: + console.print(f"[yellow]File not found, skipping: {file_path}[/yellow]") + + matches = light.music.find_upload_matches(files) + + skip_count = len(matches) if matches and not overwrite and not allow_duplicates else 0 + upload_count = len(files) - skip_count + + console.print(f"{upload_count} track{'s' if upload_count != 1 else ''} will be uploaded") + if matches: + if skip_count: + console.print( + f"{skip_count} existing track{'s' if skip_count != 1 else ''} will be skipped:" + ) else: - titles = [_os.path.splitext(_os.path.basename(s))[0] for s in files] + verb = "duplicated" if allow_duplicates else "overwritten" + console.print( + f"{len(matches)} of these already exist and will be {verb}:" + ) - existing = light.music.get_tracks() - to_overwrite = [t for t in existing if t.title in set(titles)] + if not verbose and len(matches) > _VERBOSE_LIST_THRESHOLD: + console.print("[dim]Use --verbose/-v to show full list.[/dim]") + else: + for file_path, t in matches.items(): + console.print(f" {file_path} -> {t.artist} — {t.title}") - if to_overwrite: - console.print(f"Tracks to overwrite ({len(to_overwrite)}):") - for t in to_overwrite: - console.print(f" {t.artist} — {t.title}") - if not click.confirm("Proceed?"): - return + if not click.confirm("Proceed?"): + return with Progress( TextColumn("[progress.description]{task.description}"), @@ -310,7 +320,7 @@ def on_progress(filename: str, sent: int, total: int) -> None: light.music.upload_tracks( files, allow_duplicates=allow_duplicates, - match_title_by=match_title_by, + overwrite=overwrite, convert_flac=not no_convert_flac, on_progress=on_progress, ) @@ -480,7 +490,9 @@ def notes_list(light: Light, show_id=False, content_preview=False): def render_human_readable(): if content_preview: - console.print(f"[dim]Content preview enabled. This might take a while.[/dim]") + console.print( + f"[dim]Content preview enabled. This might take a while.[/dim]" + ) for i, note in enumerate(all_notes, 1): if note.note_type == "audio": diff --git a/tests/test_api.py b/tests/test_api.py index 120d41d..6cd86d5 100644 --- a/tests/test_api.py +++ b/tests/test_api.py @@ -5,11 +5,11 @@ import respx import httpx from types import SimpleNamespace -from unittest.mock import patch +from unittest.mock import MagicMock, patch from light_api.client import Light from light_api.notes import LightNote -from light_api.music import LightTrack +from light_api.music import LightMusic, LightTrack API = "https://production.lightphonecloud.com" @@ -363,6 +363,179 @@ def test_raises_on_error_response(self, f_devices, f_tools): light.music.get_tracks() +class TestFindMatchingTrack: + def test_ignores_cross_artist_title_collision(self): + """Regression test for #18: "Playing God" by Polyphia must not match an + existing "Playing God" by Paramore just because titles collide.""" + light = make_light() + light.music._tracks = [ + LightTrack( + playlist_item_id="1", audio_id="a1", + title="Playing God", artist="Paramore", album="", + ), + ] + + assert light.music._find_matching_track("Playing God", "Paramore") is not None + assert light.music._find_matching_track("Playing God", "Polyphia") is None + + def test_matches_untagged_tracks_by_exact_unknown_artist(self): + """A file with no tags resolves to artist="Unknown" and should exact-match + an existing track that was itself uploaded with no metadata.""" + light = make_light() + light.music._tracks = [ + LightTrack( + playlist_item_id="2", audio_id="a2", + title="Some Old Rip", artist="Unknown", album="", + ), + ] + + match = light.music._find_matching_track("Some Old Rip", "Unknown") + assert match is not None and match.audio_id == "a2" + + +class TestTrackIdentity: + def test_reads_title_and_artist_from_tags(self): + light = make_light() + with patch("light_api.music.File", return_value={"title": ["Song"], "artist": ["Artist"]}): + assert light.music._track_identity("song.mp3") == ("Song", "Artist") + + def test_falls_back_to_filename_and_unknown_when_tags_missing(self): + light = make_light() + with patch("light_api.music.File", return_value=None): + title, artist = light.music._track_identity("/path/Some Song.mp3") + assert title == "Some Song" + assert artist == "Unknown" + + def test_falls_back_per_field_when_only_one_tag_is_missing(self): + light = make_light() + with patch("light_api.music.File", return_value={"title": ["Real Title"]}): + title, artist = light.music._track_identity("/path/Filename.mp3") + assert title == "Real Title" + assert artist == "Unknown" + + +class TestFindUploadMatches: + def test_ignores_cross_artist_title_collision(self): + light = make_light() + light.music._tracks = [ + LightTrack( + playlist_item_id="1", audio_id="a1", + title="Playing God", artist="Paramore", album="", + ), + LightTrack( + playlist_item_id="2", audio_id="a2", + title="New Song", artist="New Artist", album="", + ), + ] + + tags_by_path = { + "playing_god_polyphia.mp3": {"title": ["Playing God"], "artist": ["Polyphia"]}, + "new_song.mp3": {"title": ["New Song"], "artist": ["New Artist"]}, + } + + with patch("light_api.music.File", side_effect=lambda p, easy=True: tags_by_path[p]): + matches = light.music.find_upload_matches( + ["playing_god_polyphia.mp3", "new_song.mp3"] + ) + + assert "playing_god_polyphia.mp3" not in matches + assert matches["new_song.mp3"].audio_id == "a2" + + +class TestResolveUploadPlan: + """Unit tests for LightMusic._resolve_upload_plan's skip/overwrite/allow_duplicates + filtering. Pure computation - no mocking of delete_tracks_predicate needed.""" + + def _light_with_track(self): + light = make_light() + light.music._tracks = [ + LightTrack( + playlist_item_id="1", audio_id="a1", + title="Song", artist="Artist", album="", + ), + ] + return light + + def test_allow_duplicates_returns_files_unchanged_and_nothing_to_delete(self): + light = self._light_with_track() + + to_upload, to_delete = light.music._resolve_upload_plan( + ["match.mp3", "new.mp3"], allow_duplicates=True, overwrite=False + ) + + assert to_upload == ["match.mp3", "new.mp3"] + assert to_delete == [] + + def test_default_skips_matching_files_without_deleting(self): + light = self._light_with_track() + + tags_by_path = { + "match.mp3": {"title": ["Song"], "artist": ["Artist"]}, + "new.mp3": {"title": ["New Song"], "artist": ["New Artist"]}, + } + with patch("light_api.music.File", side_effect=lambda p, easy=True: tags_by_path[p]): + to_upload, to_delete = light.music._resolve_upload_plan( + ["match.mp3", "new.mp3"], allow_duplicates=False, overwrite=False + ) + + assert to_upload == ["new.mp3"] + assert to_delete == [] + + def test_overwrite_returns_matches_to_delete_and_still_uploads_them(self): + light = self._light_with_track() + + tags_by_path = { + "match.mp3": {"title": ["Song"], "artist": ["Artist"]}, + "new.mp3": {"title": ["New Song"], "artist": ["New Artist"]}, + } + with patch("light_api.music.File", side_effect=lambda p, easy=True: tags_by_path[p]): + to_upload, to_delete = light.music._resolve_upload_plan( + ["match.mp3", "new.mp3"], allow_duplicates=False, overwrite=True + ) + + assert to_upload == ["match.mp3", "new.mp3"] + assert to_delete == [light.music._tracks[0]] + + +class TestUploadTracksExcludesMissingFiles: + """overwrite=True must never delete a track whose replacement file doesn't exist on disk.""" + + def test_missing_file_never_reaches_matching_or_deletion(self): + light = make_light() + light.music._tracks = [ + LightTrack( + playlist_item_id="1", audio_id="a1", + title="Song", artist="Artist", album="", + ), + ] + light.music.delete_tracks_predicate = MagicMock() + + # Tags are mocked to guarantee a match *would* occur if File() were ever + # called on this path - but the path doesn't exist, so matching/deletion + # must never even be attempted for it. + with patch( + "light_api.music.File", + return_value={"title": ["Song"], "artist": ["Artist"]}, + ) as mock_file: + light.music.upload_tracks(["/nonexistent/match.mp3"], overwrite=True) + + mock_file.assert_not_called() + light.music.delete_tracks_predicate.assert_not_called() + + +class TestFilterValidTracks: + def test_splits_existing_and_missing_paths(self, tmp_path): + real_file = tmp_path / "song.mp3" + real_file.write_bytes(b"") + + valid, invalid = LightMusic.filter_valid_tracks( + [str(real_file), "/nonexistent/missing.mp3"] + ) + + assert valid == [str(real_file)] + assert invalid == ["/nonexistent/missing.mp3"] + + def make_note(overrides: dict | None = None) -> LightNote: data = dict(id="note-1", file_id="file-1", note_type="text", title="old title", updated_at="2026-01-01T00:00:00") if overrides: