uris, external-ids and with-bar can be set in config. flags override file, and file overrides no flag

also, not saved playlists can be exported now if id is provided

Author: donmerendolo

exportify-cli.py

@@ -87,29 +87,30 @@ def load_config(config_path: Path) -> configparser.ConfigParser:
         return config
     # Prompt user to create config
     logger.info(f"Config not found or invalid at {config_path}, creating new.")
-    click.echo(
-        """
-File "config.cfg" not found or invalid. Let's create it.
+    click.echo("""File "config.cfg" not found or invalid. Let's create it.
 
 1. Go to Spotify Developer Dashboard (https://developer.spotify.com/dashboard).
 2. Create a new app.
 3. Set a name and description for your app.
 4. Add a redirect URI (e.g. http://127.0.0.1:3000/callback).
 
-Now copy the Client ID, Client Secret and Redirect URI and paste below:
-"""
-    )
+Now after creating the app, press the Settings button on the upper right corner.
+Copy the Client ID, Client Secret and Redirect URI and paste them below.""")
+
     spotify_cfg = {
         "client_id": click.prompt("Spotify Client ID", type=str),
         "client_secret": click.prompt(
-            "Spotify Client Secret", hide_input=True, type=str
+            "Spotify Client Secret",
+            hide_input=True,
+            type=str,
         ),
         "redirect_uri": click.prompt(
             "Redirect URI",
             type=str,
-            default="http://127.0.0.1:3000/callback",
+            default="http://127.0.1:3000/callback",
         ),
     }
+
     config["spotify"] = spotify_cfg
     if not validate_config(config):
         logger.error("Invalid Spotify configuration.")
@@ -144,20 +145,23 @@ def sanitize_filename(name: str, ext: str) -> str:
 
 
 def write_file(file_path: Path, data: list[dict], file_format: str = "csv") -> None:
-    """Write list of dicts to file (CSV or JSON)."""
+    """Write list of dicts to file."""
     if not data:
         logger.warning("No data to write; skipping file.")
         return
+
     if file_format == "csv":
         headers = list(data[0].keys())
         with file_path.open("w", newline="", encoding="utf-8") as csvfile:
             writer = csv.DictWriter(csvfile, fieldnames=headers)
             writer.writeheader()
             for row in data:
                 writer.writerow(row)
+
     elif file_format == "json":
         with file_path.open("w", encoding="utf-8") as jsonfile:
             json.dump(data, jsonfile, ensure_ascii=False, indent=4)
+
     logger.info(f"Exported to {file_path}")
 
 
@@ -172,6 +176,7 @@ def __init__(
         external_ids: bool,
         with_bar_flag: bool,
     ) -> None:
+        """Initialize the exporter with a Spotify client."""
         self.spotify = spotify_client
         self.file_format = file_format
         self.include_uris = include_uris
@@ -188,10 +193,110 @@ def _fetch_all_items(
         show_bar: bool = True,
         **kwargs: Any,
     ) -> list[dict]:
-        # ... existing pagination and batching logic unchanged ...
-        pass  # existing implementation here
+        """Fetch all paginated or batched items from a Spotify endpoint.
+
+        - If the first positional arg is a list, treats it as an ID list for a batch
+          endpoint (e.g. `self.spotify.albums`), calls in chunks of 20, and returns
+          results[key] aggregated.
+        - Otherwise treats as a paginated endpoint, calling `fetch_func(*args, **kwargs)`
+          then .next() until exhausted, aggregating results["items"] or results[key].
+        """
+        items: list[dict] = []
+
+        # --- Batch mode (e.g. spotify.albums) ---
+        if args and isinstance(args[0], list):
+            id_list: list[str] = args[0]
+            total = len(id_list)
+            desc_text = desc or fetch_func.__name__
+            # build 20-item batches
+            batches = [id_list[i : i + 20] for i in range(0, total, 20)]
+
+            pbar = (
+                tqdm(total=total, desc=desc_text, unit="album", bar_format=bar_format)
+                if show_bar and self.with_bar_flag
+                else None
+            )
+
+            for batch in batches:
+                results = fetch_func(batch, *args[1:], **kwargs)
+                page_items = results.get(key, [])
+
+                fetched_ids = [i.get("id") for i in page_items if i and i.get("id")]
+                page_items = [i for i in page_items if i]
+
+                unfetched_ids = list(set(batch) - set(fetched_ids))
+                shows = []
+                for id in unfetched_ids:
+                    try:
+                        if unfetched_ids:
+                            shows.append(self.spotify.show(id))
+                    except spotipy.SpotifyException as e:
+                        logger.warning(f"Failed to fetch show for ID {id}: {e}")
+
+                for i in shows:
+                    i["label"] = i.get("publisher")
+                page_items.extend(shows)
+
+                items.extend(page_items)
+
+                if pbar:
+                    pbar.update(len(page_items))
+
+            if pbar:
+                pbar.close()
+            return items
+
+        # --- Paginated mode (e.g. playlist_tracks, saved_tracks) ---
+        # Initial fetch
+        results = fetch_func(*args, **kwargs)
+        page_items = results.get(key, [])
+        items.extend(page_items)
+
+        total = results.get("total")
+        # fallback to a name field or the function name if no desc given
+        desc_text = desc or (results.get("name") or fetch_func.__name__)
+
+        pbar = (
+            tqdm(total=total, desc=desc_text, unit="track", bar_format=bar_format)
+            if show_bar and self.with_bar_flag
+            else None
+        )
+        if pbar:
+            pbar.update(len(page_items))
+
+        # iterate through all pages
+        while len(items) < total:
+            results = self.spotify.next(results)
+            page_items = results.get(key, [])
+            items.extend(page_items)
+            if pbar:
+                pbar.update(len(page_items))
+
+        if pbar:
+            pbar.close()
+
+        def _episode_to_track(item: dict) -> dict:
+            """Convert episode to track if applicable."""
+            if item.get("track"):
+                track = item.get("track")
+            else:
+                return
+
+            if track.get("type") == "episode":
+                episode_id = track.get("id")
+                episode = self.spotify.episode(episode_id)
+                track["release_date"] = episode.get("release_date")
+
+                for artist in track.get("artists", []):
+                    artist["name"] = artist.get("type")
+
+        for i in items:
+            _episode_to_track(i)
+
+        return items
 
     def get_playlists(self) -> list[dict]:
+        """Retrieve all user playlists plus liked songs."""
         items = self._fetch_all_items(
             self.spotify.current_user_playlists,
             "items",
@@ -207,8 +312,109 @@ def get_playlists(self) -> list[dict]:
         return [liked, *items]
 
     def export_playlist(self, playlist: dict, output_dir: Path) -> None:
-        # ... existing export logic unchanged ...
-        pass  # existing implementation here
+        """Export a single playlist to CSV file."""
+        name, pid = playlist["name"], playlist["id"]
+        output_dir.mkdir(parents=True, exist_ok=True)
+        filepath = output_dir / sanitize_filename(name, self.file_format)
+
+        # Format description for progress bar
+        desc = (
+            name[: DESC_LENGTH - 2] + "...: "
+            if len(name) > DESC_LENGTH - 2
+            else f"{name}: ".ljust(DESC_LENGTH + 3)
+        )
+
+        # Fetch tracks
+        if pid == "liked_songs":
+            items = self._fetch_all_items(
+                self.spotify.current_user_saved_tracks,
+                "items",
+                desc=desc,
+            )
+        else:
+            items = self._fetch_all_items(
+                self.spotify.playlist_tracks,
+                "items",
+                pid,
+                desc=desc,
+            )
+
+        # Found a track like this in a playlist (replacing [user_id]
+        # with the actual user id). I could remove it from the list
+        # since it has no info about the track, but one could hypothetically
+        # do detective work out of 'added_at' to find the track (or thing)
+        # it originally was (I guess), so I'll leave it there.
+        #
+        # No idea on how it was generated.
+        #
+        # {'added_at': '2022-06-30T21:08:13Z',
+        #  'added_by': {'external_urls': {'spotify': 'https://open.spotify.com/user/[user_id]'},
+        #               'href': 'https://api.spotify.com/v1/users/[user_id]',
+        #               'id': '[user_id]',
+        #               'type': 'user',
+        #               'uri': 'spotify:user:[user_id]'},
+        #  'is_local': False,
+        #  'primary_color': None,
+        #  'track': None,
+        #  'video_thumbnail': {'url': None}}
+
+        # Batch fetch album details
+        album_ids = list(
+            {
+                i.get("track").get("album").get("id")
+                for i in items
+                if i.get("track")
+                and i.get("track").get("album")
+                and i.get("track").get("album").get("id")
+            },
+        )
+        album_items = self._fetch_all_items(
+            self.spotify.albums,
+            "albums",
+            album_ids,
+            desc="Fetching album details: ",
+        )
+        albums = {a.get("id"): a for a in album_items if a}
+
+        # Build export data
+        export_data = []
+        for i in items:
+            track = i.get("track") or {}
+            album = albums.get(track.get("album", {}).get("id"), {})
+            artists = [a["name"] for a in track.get("artists", [])]
+            artist_uris = [a["uri"] for a in track.get("artists", [])]
+
+            record = {
+                "Track URI": track.get("uri"),
+                "Artist URI(s)": artist_uris,
+                "Album URI": album.get("uri"),
+                "Track Name": track.get("name"),
+                "Album Name": album.get("name"),
+                "Artist Name(s)": artists,
+                "Release Date": album.get("release_date")
+                or (track.get("release_date")),
+                "Duration_ms": track.get("duration_ms"),
+                "Popularity": track.get("popularity"),
+                "Added By": i.get("added_by", {}).get("id"),
+                "Added At": i.get("added_at"),
+                "Record Label": album.get("label"),
+                "Track ISRC": track.get("external_ids", {}).get("isrc"),
+                "Album UPC": album.get("external_ids", {}).get("upc"),
+            }
+
+            if not self.include_uris:
+                record.pop("Artist URI(s)", None)
+                record.pop("Album URI", None)
+            if not self.external_ids:
+                record.pop("Track ISRC", None)
+                record.pop("Album UPC", None)
+
+            export_data.append(record)
+
+        write_file(filepath, export_data, self.file_format)
+        click.echo(
+            f"Exported {len(export_data)} tracks from '{name}' to {filepath}",
+        )
 
 
 @click.command()
@@ -245,6 +451,7 @@ def export_playlist(self, playlist: dict, output_dir: Path) -> None:
 @click.option(
     "--playlist",
     "-p",
+    "playlist",
     multiple=True,
     help="Names or IDs of playlists to export",
 )
@@ -271,7 +478,7 @@ def export_playlist(self, playlist: dict, output_dir: Path) -> None:
     "--with-bar/--no-bar",
     "with_bar_flag",
     default=None,
-    help="Show progress bar (overrides config)",
+    help="Show or hide progress bar (overrides config)",
 )
 def main(
     config: str,
@@ -306,6 +513,7 @@ def main(
     )
 
     client = init_spotify_client(cfg)
+
     exporter = SpotifyExporter(
         spotify_client=client,
         file_format=file_format,
@@ -314,41 +522,64 @@ def main(
         with_bar_flag=bar_flag,
     )
 
-    playlists = exporter.get_playlists()
+    fetched_playlists = exporter.get_playlists()
+
     if list_only:
-        playlist_data = [[p["name"], p["id"], p["tracks"]["total"]] for p in playlists]
+        playlist_data = [
+            [p["name"], p["id"], p["tracks"]["total"]] for p in fetched_playlists
+        ]
         terminal_width = os.get_terminal_size().columns
+
         click.echo(
             tabulate(
                 playlist_data,
                 headers=["Name", "ID", "Tracks"],
                 tablefmt="simple",
+                # 34 is the width of ID and Tracks columns + padding
                 maxcolwidths=[terminal_width - 34, None, None],
-            )
+            ),
         )
         sys.exit(0)
 
     # Determine targets
     targets = []
     if export_all:
-        targets = playlists
+        targets = fetched_playlists
     else:
-        for p in playlists:
+        # Exact matches first
+        for p in fetched_playlists:
             if p["name"] in playlist or p["id"] in playlist:
                 targets.append(p)
+        # For unmatched inputs, try unique prefix match
         for term in playlist:
             if any(p for p in targets if p["name"] == term or p["id"] == term):
                 continue
+
             matches = [
-                p for p in playlists if p["name"].lower().startswith(term.lower())
+                p
+                for p in fetched_playlists
+                if p["name"].lower().startswith(term.lower())
             ]
             if len(matches) == 1:
                 targets.append(matches[0])
             elif len(matches) > 1:
                 click.echo(
-                    f"Ambiguous prefix '{term}': matches {', '.join(p['name'] for p in matches)}. Skipping."
+                    f"Ambiguous prefix '{term}': matches "
+                    f"{', '.join(p['name'] for p in matches)}. Skipping.",
                 )
+
+            # User may be trying to export a playlist they have not saved
+            elif term.isalnum() and len(term) == 22:
+                try:
+                    pl = client.playlist(term)
+                    if pl:
+                        targets.append(pl)
+                except spotipy.SpotifyException as e:
+                    logger.warning(f"Failed to fetch playlist {term}: {e}")
+
+    # Deduplicate
     targets = list({p["id"]: p for p in targets}.values())
+
     if not targets:
         click.echo("No matching playlists found.")
         sys.exit(1)