124 lines
3 KiB
Python
124 lines
3 KiB
Python
from dotenv import load_dotenv
|
||
load_dotenv()
|
||
|
||
from spotify import (
|
||
spotify_client,
|
||
get_playlists,
|
||
get_playlist_tracks,
|
||
get_liked_tracks,
|
||
)
|
||
|
||
from subsonic import (
|
||
search_song,
|
||
create_playlist,
|
||
add_to_playlist,
|
||
get_playlist_by_name,
|
||
get_playlist_songs,
|
||
)
|
||
|
||
from matcher import make_query
|
||
from utils import parse_selection
|
||
|
||
|
||
LIKED_NAME = "Spotify – Liked Songs"
|
||
|
||
|
||
def main():
|
||
sp = spotify_client()
|
||
playlists = get_playlists(sp)
|
||
|
||
print("\nAvailable sources:\n")
|
||
|
||
sources = []
|
||
|
||
# Normal playlists
|
||
for i, p in enumerate(playlists, 1):
|
||
print(f"{i:2d}. 📀 {p['name']}")
|
||
sources.append(("playlist", p))
|
||
|
||
# Liked songs option
|
||
liked_index = len(sources) + 1
|
||
print(f"{liked_index:2d}. ❤️ Liked Songs")
|
||
sources.append(("liked", None))
|
||
|
||
choice = input("\nSelect (e.g. 1,3,5 | 2-6 | all): ")
|
||
selected_indices = parse_selection(choice, len(sources))
|
||
|
||
if not selected_indices:
|
||
print("Nothing selected.")
|
||
return
|
||
|
||
report = {}
|
||
|
||
for idx in selected_indices:
|
||
source_type, payload = sources[idx]
|
||
|
||
if source_type == "liked":
|
||
name = LIKED_NAME
|
||
print(f"\n▶ Migrating: {name}")
|
||
tracks = get_liked_tracks(sp)
|
||
else:
|
||
name = payload["name"]
|
||
print(f"\n▶ Migrating playlist: {name}")
|
||
tracks = get_playlist_tracks(sp, payload["id"])
|
||
|
||
existing = get_playlist_by_name(name)
|
||
if existing:
|
||
pid = existing["id"]
|
||
existing_songs = get_playlist_songs(pid)
|
||
print(" ↺ Playlist exists on Subsonic, retrying missing tracks")
|
||
else:
|
||
pid = create_playlist(name)
|
||
existing_songs = set()
|
||
print(" + Created new Subsonic playlist")
|
||
|
||
added_now = []
|
||
skipped = []
|
||
missing = []
|
||
|
||
for t in tracks:
|
||
query = make_query(t)
|
||
sid = search_song(query)
|
||
|
||
if not sid:
|
||
missing.append(query)
|
||
print(f" ✗ {query}")
|
||
continue
|
||
|
||
if sid in existing_songs:
|
||
skipped.append(query)
|
||
print(f" ↷ Already exists: {query}")
|
||
continue
|
||
|
||
added_now.append((query, sid))
|
||
print(f" ✔ {query}")
|
||
|
||
if added_now:
|
||
add_to_playlist(pid, [sid for _, sid in added_now])
|
||
|
||
report[name] = {
|
||
"added": [q for q, _ in added_now],
|
||
"skipped": skipped,
|
||
"missing": missing,
|
||
}
|
||
|
||
print_summary(report)
|
||
|
||
|
||
def print_summary(report):
|
||
print("\n===== Migration Summary =====\n")
|
||
|
||
for name, data in report.items():
|
||
print(f"📀 {name}")
|
||
print(f" ✔ Added now: {len(data['added'])}")
|
||
print(f" ↷ Already existed: {len(data['skipped'])}")
|
||
print(f" ✗ Still missing: {len(data['missing'])}")
|
||
|
||
if data["missing"]:
|
||
for m in data["missing"]:
|
||
print(f" - {m}")
|
||
print()
|
||
|
||
|
||
if __name__ == "__main__":
|
||
main()
|