import os import click from dotenv import load_dotenv load_dotenv() from tunesynctool import PlaylistSynchronizer, Configuration, SubsonicDriver, SpotifyDriver, Track, TrackMatcher from tunesynctool.integrations import Musicbrainz from pyarr import LidarrAPI from tqdm import tqdm from interactive import InteractiveApp import musicbrainzngs config = Configuration.from_env() @click.group() def cli(): """Spotidrome tool CLI.""" pass # @cli.command("lidarr_sync") # @click.option('--tracks', help='List of tracks to sync with Lidarr') def lidarr_sync(tracks: list[Track]): """Sync tracks with Lidarr.""" # Initialize the Lidarr API lidarr_api = LidarrAPI( os.getenv("LIDARR_BASE_URL"), os.getenv("LIDARR_API_KEY"), ) for track in tqdm(tracks, desc="Syncing tracks with Lidarr"): mbid = track.musicbrainz_id if not mbid: mbid = Musicbrainz.id_from_track(track) if not mbid: tqdm.write(f"Track {track.title} by {track.primary_artist} has no MusicBrainz ID.") continue releaseGroup = musicbrainzngs.search_release_groups(query=f"{track.album_name} {track.primary_artist}") if not releaseGroup: tqdm.write(f"Release group not found for track {track.title} by {track.primary_artist}.") continue results = lidarr_api.lookup_album(f"lidarr:{releaseGroup['release-group-list'][0]['id']}") if results: if results[0]['monitored']: tqdm.write(f"Album {track.album_name} already monitored on Lidarr.") monitored = lidarr_api.get_album(foreignAlbumId=results[0]['foreignAlbumId']) if monitored[0]['statistics']['percentOfTracks'] != 100: tqdm.write(f"Album {track.album_name} not complete on Lidarr, searching for missing tracks.") releases = lidarr_api.get_release(artistId=results[0]['artistId'], albumId=results[0]['id']) tqdm.write(f"Found {len(releases)} releases for album {track.album_name}.") if len(releases) == 0: tqdm.write(f"No releases found for album {track.album_name}.") continue tqdm.write(f"Adding release {releases[0]['guid']} to Lidarr.") lidarr_api._post(path="v1/release",data={ "guid": releases[0]['guid'], "indexerId": releases[0]['indexerId'], }) tqdm.write(f"Release {releases[0]['guid']} grabbed.") continue click.echo(f"Adding album {track.album_name} to Lidarr.") lidarr_api.add_album(results[0], search_for_new_album=True, artist_search_for_missing_albums=True, root_dir=os.getenv("LIDARR_ROOT_DIR")) click.echo(f"Album {track.album_name} added to Lidarr.") else: click.echo(f"Album {track.album_name} not found on Lidarr.") continue @cli.command("sync") @click.option('--spotify-playlist-id', help='Spotify playlist ID') @click.option('--subsonic-playlist-id', help='Subsonic playlist ID') @click.option('--lidarr', default=True, help='Monitor Artists on Lidarr, then search for albums and download.') def sync(spotify_playlist_id, subsonic_playlist_id, lidarr): """Sync a Spotify playlist with Subsonic. """ if not spotify_playlist_id: raise click.BadParameter('Spotify playlist ID is required.') # Initialize the Spotify driver spotify_driver = SpotifyDriver(config) # Initialize the Subsonic driver subsonic_driver = SubsonicDriver(config) # Initialize the playlist synchronizer synchronizer = PlaylistSynchronizer(spotify_driver, subsonic_driver, config) # Get the tracks from the Spotify playlist tracks = spotify_driver.get_playlist_tracks(spotify_playlist_id, limit=500) if lidarr: lidarr_sync(tracks) # Synchronize the playlist synchronizer.sync(spotify_playlist_id, subsonic_playlist_id) click.echo("Sync completed.") @cli.command("transfer") @click.option('--spotify-playlist-id', help='Spotify playlist ID') @click.option('--lidarr', default=True, help='Monitor Artists on Lidarr, then search for albums and download.') def transfer(spotify_playlist_id, lidarr): """Transfer a Spotify playlist with Subsonic. """ if not spotify_playlist_id: raise click.BadParameter('Spotify playlist ID is required.') # Initialize the Spotify driver spotify_driver = SpotifyDriver(config) # Initialize the Subsonic driver subsonic_driver = SubsonicDriver(config) source_tracks = spotify_driver.get_playlist_tracks(spotify_playlist_id) source_playlist = spotify_driver.get_playlist(spotify_playlist_id) if lidarr: lidarr_sync(source_tracks) click.echo("Lidarr sync started. Please check Lidarr for progress. Run `sync` when completed.") matcher = TrackMatcher(subsonic_driver) matched_tracks: list[Track] = [] click.echo(f"Matching tracks from {source_playlist.name}...") click.echo(f"Total tracks to match: {len(source_tracks)}") for track in tqdm(source_tracks, desc="Matching tracks"): tqdm.write(f"Matching track: {track.title} by {track.primary_artist}") matched_track = matcher.find_match(track) if matched_track: matched_tracks.append(matched_track) tqdm.write(f"Matched track: {matched_track.title} by {matched_track.primary_artist}") else: tqdm.write(f"Track not found: {track.title} by {track.primary_artist}") try: target_playlist = subsonic_driver.create_playlist(source_playlist.name) subsonic_driver.add_tracks_to_playlist(playlist_id=target_playlist.service_id, track_ids=(track.service_id for track in matched_tracks)) click.echo("Transfer completed.") except Exception as e: click.echo(f"Error during transfer: {e}") @cli.command("list") @click.option('--platform', type=click.Choice(['spotify', 'subsonic']), help='Platform to list playlists from') def list_playlists(platform): """List playlists from a platform.""" if not platform: raise click.BadParameter('Platform is required.') if platform == 'spotify': driver = SpotifyDriver(config) elif platform == 'subsonic': driver = SubsonicDriver(config) playlists = driver.get_user_playlists() for playlist in playlists: click.echo(f"{playlist.name} ({playlist.service_id})") @cli.command("interactive") def interactive(): """Interactive mode.""" app = InteractiveApp() app.run() if __name__ == '__main__': cli()