Redesign light music upload behavior

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
 ```
 

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,47 +266,123 @@ 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:
         """Upload tracks to device.
 
         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,

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,66 +228,76 @@ def podcasts_delete(light: Light, title):
 
 # -- Music commands -------------------------------------------------------------
 
+_VERBOSE_LIST_THRESHOLD = 20
+
 
 @music.command("upload")
 @with_light
 @click.argument("songs", nargs=-1, required=True)
 @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":