Improve track matching and add comprehensive job progress tracking

musicsync/cli.py

@@ -2,18 +2,20 @@
 
 import argparse
 import logging
+import shutil
 import sys
+import time
 from collections.abc import Callable, Sequence
 from pathlib import Path
 
 from . import __version__
 from .config import SERVICES, config_path, ensure_config_file
 from .csvio import read_tracks, slug, write_tracks
 from .errors import ConfigError, MusicSyncError, ProviderError
-from .models import Match, Track
+from .models import Track
 from .providers.base import Provider
 from .services import Services
-from .sync import LIKED, Progress, import_tracks, select_tracks
+from .sync import LIKED, Miss, Progress, Step, import_albums, import_tracks, remaining, select_tracks
 
 type Handler = Callable[[argparse.Namespace, Services], int]
 
@@ -22,35 +24,67 @@ def _warn(message: str) -> None:
     print(message, file=sys.stderr)
 
 
-def _progress(quiet: bool) -> Progress | None:
-    """A progress display on stderr: a live line on a terminal, a line per 50 tracks in a log."""
-    if quiet:
-        return None
-    tty = sys.stderr.isatty()
+def _seconds(seconds: float) -> str:
+    """A rough duration for people: "40 s", "3 min"."""
+    return f"{max(round(seconds / 10) * 10, 10)} s" if seconds < 60 else f"{round(seconds / 60)} min"
+
+
+class _ProgressLine:
+    """Progress on stderr. On a terminal: one line per step of the work, redrawn in place. In a log file:
+    a line per 50 tracks, and one when a step is complete."""
+
+    def __init__(self) -> None:
+        self.tty = sys.stderr.isatty()
+        self.text = ""  # the step the current line is about
+        self.complete = True  # the last step reported was complete, so no line is left open
+        self.started = time.monotonic()
+
+    def __call__(self, step: Step) -> None:
+        if step.text != self.text or self.complete:  # a new step, or the same kind of step for the next list
+            if self.tty and not self.complete:
+                print(file=sys.stderr, flush=True)  # end the line the previous step left open
+            self.text, self.started = step.text, time.monotonic()
+        self.complete = step.total is not None and step.done >= step.total
+        if self.tty:
+            width = max(shutil.get_terminal_size().columns - 1, 40)
+            end = "\n" if self.complete else ""
+            print(f"\r{self.line(step)[:width]:<{width}}", end=end, file=sys.stderr, flush=True)
+        elif self.complete or (step.done and step.done % 50 == 0):
+            print(self.line(step, current=False), file=sys.stderr, flush=True)
+
+    def line(self, step: Step, *, current: bool = True) -> str:
+        """ "Finding the tracks on Tidal  [#####---]  45/300  40 found, 5 not found  1 min left  Artist - Title" """
+        if step.total is None:
+            parts = [step.text, f"{step.done} so far"]
+        else:
+            filled = round(20 * step.done / step.total) if step.total else 20
+            parts = [step.text, f"[{'#' * filled}{'-' * (20 - filled)}]", f"{step.done}/{step.total}"]
+        if step.phase == "match":
+            parts.append(f"{step.found} found, {step.done - step.found} not found")
+        if (left := remaining(step.done, step.total, time.monotonic() - self.started)) is not None:
+            parts.append(f"{_seconds(left)} left")
+        if current and step.track and not self.complete:
+            parts.append(str(step.track))
+        return "  ".join(parts)
 
-    def show(done: int, total: int, track: Track, _match: Match | None) -> None:
-        if tty:
-            end = "\n" if done == total else ""
-            print(f"\r[{done}/{total}] {str(track)[:60]:<60}", end=end, file=sys.stderr, flush=True)
-        elif done % 50 == 0 or done == total:
-            print(f"[{done}/{total}]", file=sys.stderr, flush=True)
 
-    return show
+def _progress(quiet: bool) -> Progress | None:
+    return None if quiet else _ProgressLine()
 
 
-def _report_unmatched(unmatched: list[Track], path: str | None) -> None:
-    """Show the tracks that were not found, or save them all when ``--unmatched`` was given."""
-    if not unmatched:
+def _report_unmatched(misses: list[Miss], path: str | None) -> None:
+    """Show the tracks that were not found and why, or save them all when ``--unmatched`` was given."""
+    if not misses:
         return
     if path:
-        write_tracks(path, unmatched)
-        print(f"{len(unmatched)} tracks without a match written to {path}")
+        write_tracks(path, [miss.track for miss in misses])
+        print(f"{len(misses)} tracks without a match written to {path}")
         return
     print("\nNot found:")
-    for track in unmatched[:10]:
-        print(f"  {track}")
-    if len(unmatched) > 10:
-        print(f"  ... and {len(unmatched) - 10} more (use --unmatched FILE to save them all)")
+    for miss in misses[:10]:
+        print(f"  {miss.track}: {miss.why}")
+    if len(misses) > 10:
+        print(f"  ... and {len(misses) - 10} more (use --unmatched FILE to save them all)")
 
 
 # --- commands ---------------------------------------------------------------------------------------
@@ -95,7 +129,9 @@ def cmd_playlists(args: argparse.Namespace, services: Services) -> int:
 
 def cmd_export(args: argparse.Namespace, services: Services) -> int:
     provider = services.provider(args.service)
-    selected = select_tracks(provider, liked=args.liked, playlist=args.playlist, on_skip=_warn)
+    selected = select_tracks(
+        provider, liked=args.liked, playlist=args.playlist, on_skip=_warn, progress=_progress(args.quiet)
+    )
     if args.all:  # one CSV per playlist, in a folder
         folder = Path(args.output or f"music-sync-export/{args.service}")
         for name, tracks in selected:
@@ -113,39 +149,42 @@ def cmd_import(args: argparse.Namespace, services: Services) -> int:
     if not tracks:
         print(f"No tracks found in {args.file}.")
         return 1
-    result = import_tracks(
-        services.provider(args.service),
-        tracks,
-        args.playlist,
-        min_score=args.min_score,
-        dry_run=args.dry_run,
-        progress=_progress(args.quiet),
-    )
+    provider, progress = services.provider(args.service), _progress(args.quiet)
+    if args.albums:  # a row per album, as csv2tidal took them
+        result = import_albums(provider, tracks, min_score=args.min_score, dry_run=args.dry_run, progress=progress)
+    else:
+        playlist = None if args.to_favorites else args.playlist
+        result = import_tracks(
+            provider, tracks, playlist, min_score=args.min_score, dry_run=args.dry_run, progress=progress
+        )
     print(result.summary())
-    _report_unmatched(result.unmatched, args.unmatched)
+    _report_unmatched(result.misses, args.unmatched)
     return 0
 
 
 def cmd_transfer(args: argparse.Namespace, services: Services) -> int:
     if args.source == args.target:
         raise ProviderError("Source and target are the same service.")
-    if args.to_playlist and args.all:
-        raise ProviderError("--to-playlist cannot be combined with --all.")
+    if args.to_playlist and (args.all or args.sync_favorites):
+        raise ProviderError("--to-playlist cannot be combined with --all or --sync-favorites.")
     source, target = services.provider(args.source), services.provider(args.target)
-    unmatched: list[Track] = []
-    for name, tracks in select_tracks(source, liked=args.liked, playlist=args.playlist, on_skip=_warn):
-        destination = args.to_playlist or (f"{LIKED} (from {args.source.title()})" if name == LIKED else name)
+    progress = _progress(args.quiet)
+    misses: list[Miss] = []
+    liked = args.liked or args.sync_favorites
+    selected = select_tracks(source, liked=liked, playlist=args.playlist, on_skip=_warn, progress=progress)
+    for name, tracks in selected:
+        # None: the favourites (liked songs) of the target
+        destination = (
+            None
+            if args.sync_favorites
+            else args.to_playlist or (f"{LIKED} (from {args.source.title()})" if name == LIKED else name)
+        )
         result = import_tracks(
-            target,
-            tracks,
-            destination,
-            min_score=args.min_score,
-            dry_run=args.dry_run,
-            progress=_progress(args.quiet),
+            target, tracks, destination, min_score=args.min_score, dry_run=args.dry_run, progress=progress
         )
         print(result.summary())
-        unmatched.extend(result.unmatched)
-    _report_unmatched(unmatched, args.unmatched)
+        misses.extend(result.misses)
+    _report_unmatched(misses, args.unmatched)
     return 0
 
 
@@ -215,12 +254,14 @@ def cmd_doctor(args: argparse.Namespace, services: Services) -> int:
 # --- argument parsing ---------------------------------------------------------------------------------
 
 
-def _add_selection(parser: argparse.ArgumentParser) -> None:
+def _add_selection(parser: argparse.ArgumentParser, *, favorites: bool = False) -> None:
     """--liked, --playlist or --all: which tracks to work on."""
     group = parser.add_mutually_exclusive_group(required=True)
     group.add_argument("--liked", action="store_true", help="your liked / saved songs")
     group.add_argument("--playlist", metavar="NAME_OR_ID", help="one playlist, by name or id")
     group.add_argument("--all", action="store_true", help="liked songs and every playlist you can read")
+    if favorites:  # the name spotify_to_tidal uses
+        group.add_argument("--sync-favorites", action="store_true", help="liked songs, to the target's favourites")
 
 
 def _add_matching_options(parser: argparse.ArgumentParser) -> None:
@@ -257,16 +298,20 @@ def add(name: str, handler: Handler, help: str, *, service: bool = True) -> argp
     export = add("export", cmd_export, "save liked songs or playlists as CSV")
     _add_selection(export)
     export.add_argument("-o", "--output", metavar="PATH", help="CSV file (a folder with --all)")
+    export.add_argument("-q", "--quiet", action="store_true", help="no progress output")
 
     imp = add("import", cmd_import, "add the tracks of a CSV file to a playlist")
     imp.add_argument("file", help="CSV file (see the README for the format)")
-    imp.add_argument("--playlist", default="Music-Sync import", help="playlist to add to, created if missing")
+    into = imp.add_mutually_exclusive_group()
+    into.add_argument("--playlist", default="Music-Sync import", help="playlist to add to, created if missing")
+    into.add_argument("--to-favorites", action="store_true", help="add to your favourites (liked songs) instead")
+    into.add_argument("--albums", action="store_true", help="rows are albums (artist,album), for your favourites")
     _add_matching_options(imp)
 
     transfer = add("transfer", cmd_transfer, "copy liked songs or playlists from one service to another", service=False)
     transfer.add_argument("source", choices=SERVICES, metavar="SOURCE")
     transfer.add_argument("target", choices=SERVICES, metavar="TARGET")
-    _add_selection(transfer)
+    _add_selection(transfer, favorites=True)
     transfer.add_argument("--to-playlist", metavar="NAME", help="playlist to fill (default: same name as the source)")
     _add_matching_options(transfer)
 

musicsync/config.py

@@ -10,6 +10,11 @@
 from .errors import ConfigError
 
 SERVICES = ("spotify", "tidal")
+# Where each service's developer apps are made.
+DEVELOPER_DASHBOARDS = {
+    "spotify": "https://developer.spotify.com/dashboard",
+    "tidal": "https://developer.tidal.com/dashboard",
+}
 TEMPLATE = Path(__file__).with_name("config_template.toml")
 
 
@@ -45,6 +50,11 @@ def ensure_config_file() -> bool:
     return True
 
 
+def default_redirect_uri(service: str) -> str:
+    """Where a service sends you back after logging in, unless the settings say otherwise."""
+    return f"http://127.0.0.1:8888/{service}/callback"
+
+
 def _locale_country() -> str:
     """The country of the system locale ("nl_NL.UTF-8" gives "NL"), else "US"."""
     for var in ("LC_ALL", "LC_MESSAGES", "LANG"):
@@ -70,12 +80,16 @@ class Settings:
     def is_configured(self, service: str) -> bool:
         return bool(self.services.get(service, ServiceConfig()).client_id)
 
+    def redirect_uri(self, service: str) -> str:
+        return self.services.get(service, ServiceConfig()).redirect_uri or default_redirect_uri(service)
+
     def require(self, service: str) -> ServiceConfig:
         """The credentials of ``service``, or a ConfigError that says how to set them."""
         if not self.is_configured(service):
             raise ConfigError(
-                f"{service.title()} client ID missing. Set {service.upper()}_CLIENT_ID or add "
-                f"client_id under [{service}] in {config_path()} (see the README)."
+                f"{service.title()} client ID missing. Create an app at {DEVELOPER_DASHBOARDS[service]} and put "
+                f"its Client ID under [{service}] in {config_path()}, or set {service.upper()}_CLIENT_ID "
+                "(see Setup in the README)."
             )
         return self.services[service]
 
@@ -98,9 +112,7 @@ def load_settings(path: Path | None = None, env: Mapping[str, str] | None = None
             client_id=env.get(f"{prefix}_CLIENT_ID") or section.get("client_id", ""),
             client_secret=env.get(f"{prefix}_CLIENT_SECRET") or section.get("client_secret", ""),
             # The web interface listens on 8888; the CLI login borrows the same address.
-            redirect_uri=env.get(f"{prefix}_REDIRECT_URI")
-            or section.get("redirect_uri")
-            or f"http://127.0.0.1:8888/{name}/callback",
+            redirect_uri=env.get(f"{prefix}_REDIRECT_URI") or section.get("redirect_uri") or default_redirect_uri(name),
         )
     country = env.get("MUSICSYNC_COUNTRY") or data.get("country") or _locale_country()
     return Settings(services, country.upper())

musicsync/csvio.py

@@ -48,7 +48,7 @@ def parse_csv(fh: TextIO, source: str = "the CSV") -> list[Track]:
         return []
 
     header = [_ALIASES.get(cell.strip().lower()) for cell in rows[0]]
-    if "title" in header:
+    if "title" in header or "album" in header:
         columns, body = header, rows[1:]
     else:
         columns, body = ["artists", "title"], rows  # no header row: "artist,title"
@@ -57,7 +57,7 @@ def parse_csv(fh: TextIO, source: str = "the CSV") -> list[Track]:
 
 
 def _track(record: dict[str, str]) -> Track | None:
-    if not (title := record.get("title")):
+    if not (title := record.get("title") or record.get("album")):  # a list of albums has no track titles
         return None
     ids = {}
     if record.get("spotify_uri", "").startswith("spotify:track:"):