Initial file written.

Signed-off-by: Cliff Hill <xlorep@darkhelm.org>

src/playlist/loader.py

@@ -0,0 +1,284 @@
+from plexapi.server import PlexServer
+import sys
+import os
+import toml
+import argparse
+from config_path import ConfigPath
+import asyncio
+from concurrent.futures import ThreadPoolExecutor
+import aiofiles
+from dataclasses import dataclass
+
+# Configuration file setup using config-path
+CONFIG_DIR: ConfigPath = ConfigPath("PlexPlaylist", "Plex", ".config")
+CONFIG_FILE: str = CONFIG_DIR.path("plex_config.toml")
+
+# Executor for blocking operations
+executor: ThreadPoolExecutor = ThreadPoolExecutor(max_workers=4)
+
+@dataclass
+class Track:
+    """Dataclass representing a music track from Plex.
+
+    Attributes:
+        track_number: The track number or "N/A" if not available.
+        disc_number: The disc number or "N/A" if not applicable.
+        title: The title of the track.
+        track_artist: The artist of the track.
+        album_title: The title of the album.
+        album_artist: The artist of the album.
+        year: The release year of the album or "N/A".
+        genres: List of genres associated with the track.
+        date_added: Date the track was added to the library (YYYY-MM-DD).
+        last_played: Date the track was last played (YYYY-MM-DD) or "Never".
+        play_count: Number of times the track has been played.
+        rating: User rating of the track or "Not Rated".
+    """
+    track_number: str
+    disc_number: str
+    title: str
+    track_artist: str
+    album_title: str
+    album_artist: str
+    year: str
+    genres: list[str]
+    date_added: str
+    last_played: str
+    play_count: int
+    rating: str
+
+    def __str__(self) -> str:
+        """String representation of the track for printing.
+
+        Returns:
+            Formatted string with all track details.
+        """
+        output: list[str] = [f"Track #{self.track_number}"]
+        if self.disc_number != "N/A":
+            output.append(f"Disc #{self.disc_number}")
+        output.extend([
+            f"Title: {self.title}",
+            f"Track Artist: {self.track_artist}",
+            f"Album: {self.album_title}",
+            f"Album Artist: {self.album_artist}",
+            f"Year: {self.year}",
+            f"Genre: {', '.join(self.genres) if self.genres else 'N/A'}",
+            f"Date Added: {self.date_added}",
+            f"Last Played: {self.last_played}",
+            f"Play Count: {self.play_count}",
+            f"Rating: {self.rating}",
+            "-" * 30
+        ])
+        return "\n".join(output)
+
+async def save_config(baseurl: str, token: str) -> None:
+    """Save Plex configuration to a local TOML file asynchronously.
+
+    Args:
+        baseurl: The URL of the Plex server (e.g., http://192.168.1.100:32400).
+        token: The authentication token for the Plex server.
+
+    Raises:
+        Exception: If there's an error writing to the config file.
+    """
+    loop: asyncio.AbstractEventLoop = asyncio.get_event_loop()
+    await loop.run_in_executor(executor, os.makedirs, os.path.dirname(CONFIG_FILE), {"exist_ok": True})
+    
+    config: dict[str, dict[str, str]] = {
+        "plex": {
+            "baseurl": baseurl,
+            "token": token
+        }
+    }
+    async with aiofiles.open(CONFIG_FILE, 'w') as f:
+        await f.write(toml.dumps(config))
+    print("Configuration saved successfully")
+
+async def load_config() -> tuple[str | None, str | None]:
+    """Load Plex configuration from local TOML file asynchronously.
+
+    Returns:
+        A tuple of (baseurl, token) if found, else (None, None).
+    """
+    if os.path.exists(CONFIG_FILE):
+        async with aiofiles.open(CONFIG_FILE, 'r') as f:
+            content: str = await f.read()
+            config: dict[str, dict[str, str]] = toml.loads(content)
+            return config.get("plex", {}).get("baseurl"), config.get("plex", {}).get("token")
+    return None, None
+
+async def connect_to_plex(baseurl: str, token: str) -> PlexServer | None:
+    """Connect to Plex server with given credentials asynchronously.
+
+    Args:
+        baseurl: The URL of the Plex server.
+        token: The authentication token for the Plex server.
+
+    Returns:
+        PlexServer instance if successful, None if connection fails.
+    """
+    loop: asyncio.AbstractEventLoop = asyncio.get_event_loop()
+    try:
+        plex: PlexServer = await loop.run_in_executor(executor, PlexServer, baseurl, token)
+        return plex
+    except Exception as e:
+        print(f"Error connecting to Plex server: {e}")
+        return None
+
+async def process_track_batch(tracks: list, batch_start: int, batch_end: int, total_tracks: int) -> list[Track]:
+    """Process a batch of tracks asynchronously and return a list of Track objects.
+
+    Args:
+        tracks: List of raw Plex track objects.
+        batch_start: Starting index of the batch.
+        batch_end: Ending index of the batch.
+        total_tracks: Total number of tracks for progress reporting.
+
+    Returns:
+        List of processed Track objects for this batch.
+    """
+    loop: asyncio.AbstractEventLoop = asyncio.get_event_loop()
+    print(f"\nProcessing tracks {batch_start + 1}-{batch_end} of {total_tracks}")
+    
+    batch_tracks: list[Track] = []
+    for track in tracks[batch_start:batch_end]:
+        track_number: str = track.index or "N/A"
+        disc_number: str = track.parentIndex if hasattr(track, 'parentIndex') else "N/A"
+        track_title: str = track.title or "Unknown Title"
+        
+        # Wrap potentially blocking calls
+        track_artist: str = track.originalTitle or (
+            await loop.run_in_executor(executor, lambda t=track: t.artist().title if t.artist() else "Unknown Artist")
+        )
+        album_title: str = await loop.run_in_executor(executor, lambda t=track: t.album().title if t.album() else "Unknown Album")
+        album_artist: str = await loop.run_in_executor(executor, lambda t=track: 
+            t.album().artist().title if t.album() and t.album().artist() else "Unknown Artist")
+        year: str = await loop.run_in_executor(executor, lambda t=track: t.album().year if t.album() else "N/A")
+        
+        genres: list = track.genres if hasattr(track, 'genres') and track.genres else []
+        genre_list: list[str] = [genre.tag for genre in genres] if genres else []
+        date_added: str = track.addedAt.strftime('%Y-%m-%d') if track.addedAt else "N/A"
+        last_played: str = track.lastViewedAt.strftime('%Y-%m-%d') if track.lastViewedAt else "Never"
+        play_count: int = track.viewCount if track.viewCount is not None else 0
+        rating: str = track.userRating if track.userRating is not None else "Not Rated"
+        
+        # Create Track instance and append to batch list
+        track_data: Track = Track(
+            track_number=track_number,
+            disc_number=disc_number,
+            title=track_title,
+            track_artist=track_artist,
+            album_title=album_title,
+            album_artist=album_artist,
+            year=year,
+            genres=genre_list,
+            date_added=date_added,
+            last_played=last_played,
+            play_count=play_count,
+            rating=rating
+        )
+        batch_tracks.append(track_data)
+    
+    return batch_tracks
+
+async def get_music_tracks(plex: PlexServer, batch_size: int = 1000) -> list[Track]:
+    """Retrieve all music tracks from Plex server in batches and return combined list.
+
+    Args:
+        plex: Connected PlexServer instance.
+        batch_size: Number of tracks to process per batch. Defaults to 1000.
+
+    Returns:
+        Combined list of all processed Track objects.
+
+    Raises:
+        Exception: If there's an error retrieving tracks from the Plex server.
+    """
+    try:
+        # Get all music libraries
+        loop: asyncio.AbstractEventLoop = asyncio.get_event_loop()
+        sections: list = await loop.run_in_executor(executor, lambda: [s for s in plex.library.sections() if s.type == 'artist'])
+        
+        if not sections:
+            print("No music libraries found!")
+            return []
+
+        # Collect all tracks from all music sections
+        all_tracks: list = []
+        for section in sections:
+            print(f"\nCollecting tracks from library: {section.title}")
+            tracks: list = await loop.run_in_executor(executor, section.search, {'libtype': 'track'})
+            all_tracks.extend(tracks)
+        
+        total_tracks: int = len(all_tracks)
+        print(f"\nFound {total_tracks} tracks across all libraries")
+        print("-" * 50)
+        
+        if total_tracks == 0:
+            print("No tracks found!")
+            return []
+
+        # Create tasks for each batch
+        tasks: list[asyncio.Task] = []
+        for batch_start in range(0, total_tracks, batch_size):
+            batch_end: int = min(batch_start + batch_size, total_tracks)
+            tasks.append(process_track_batch(all_tracks, batch_start, batch_end, total_tracks))
+        
+        # Run all batch tasks concurrently and gather results
+        batch_results: list[list[Track]] = await asyncio.gather(*tasks)
+        
+        # Efficiently combine all tracks from batches
+        combined_tracks: list[Track] = []
+        for batch in batch_results:
+            combined_tracks.extend(batch)
+        
+        # Print all tracks
+        for track in combined_tracks:
+            print(track)
+        
+        return combined_tracks
+                        
+    except Exception as e:
+        print(f"Error retrieving music tracks: {e}")
+        return []
+
+async def main() -> None:
+    """Main entry point for the Plex Playlist Track Lister.
+
+    Parses command-line arguments, manages configuration, and retrieves tracks.
+    """
+    parser: argparse.ArgumentParser = argparse.ArgumentParser(description="Plex Playlist Track Lister")
+    parser.add_argument('--login', action='store_true', help="Force login and update credentials")
+    parser.add_argument('--batch-size', type=int, default=1000, 
+                        help="Number of tracks to process per batch (default: 1000)")
+    args: argparse.Namespace = parser.parse_args()
+
+    # Try to load existing configuration
+    baseurl: str | None
+    token: str | None
+    baseurl, token = await load_config()
+
+    # If no config exists or login is forced
+    if args.login or not (baseurl and token):
+        baseurl = input("Enter Plex server URL (e.g., http://192.168.1.100:32400): ").strip()
+        token = input("Enter Plex token: ").strip()
+        
+        # Test the connection
+        plex: PlexServer | None = await connect_to_plex(baseurl, token)
+        if plex:
+            await save_config(baseurl, token)
+        else:
+            print("Failed to connect with provided credentials. Exiting.")
+            sys.exit(1)
+    else:
+        plex = await connect_to_plex(baseurl, token)
+        if not plex:
+            print("Stored credentials failed. Please use --login to update them.")
+            sys.exit(1)
+
+    # Get and display music tracks with specified batch size
+    tracks: list[Track] = await get_music_tracks(plex, batch_size=args.batch_size)
+    print(f"\nTotal tracks retrieved: {len(tracks)}")
+
+if __name__ == "__main__":
+    asyncio.run(main())