36058c318a
- fetch() uses a retrying session (connection resets, 5xx, 429 Retry-After) with backoff, so a transient drop mid-fetch no longer aborts the run. - tqdm progress bars on all paginated fetches (driven by paging `total`). - Flow.fetch_spotify(): drop stray `self` from the @staticmethod. - gui.sh runs the venv's streamlit from the script dir; add requirements.txt. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
136 lines
No EOL
4.6 KiB
Python
136 lines
No EOL
4.6 KiB
Python
|
|
from tqdm import tqdm
|
|
import src.lyrics.lyrics as Lyrics
|
|
|
|
from Env import client_id, client_secret, local_library_path
|
|
from Env import work_dir
|
|
|
|
from src.Library import *
|
|
from src.LibrarySerializer import LibrarySaver, LibraryLoader
|
|
|
|
from src.backends.spotify.SpotifyWebAPI import fetch_all
|
|
from src.backends.spotify.ParseSpotify import Parser
|
|
|
|
from src.backends.local.LocalLibraryIndexer import update_local_library
|
|
from src.backends.local.ParseLocal import LocalParser
|
|
from src.backends.itunes.ParseItunesXml import ParseItunesXml
|
|
|
|
|
|
from src.mappers.RemoteLibraryResolver import resolve_remote_tracks
|
|
from src.StatGenerator import log_stats
|
|
|
|
|
|
class Flow:
|
|
spotify_library : Library = None
|
|
local_library : Library = None
|
|
itunes_library : Library = None
|
|
|
|
@staticmethod
|
|
def fetch_spotify():
|
|
fetch_all(client_id, client_secret, work_dir)
|
|
|
|
@staticmethod
|
|
def fetch_local():
|
|
update_local_library(local_library_path, work_dir)
|
|
|
|
@staticmethod
|
|
def parse_spotify_library():
|
|
parser = Parser()
|
|
|
|
library_parsed = parser.parse(
|
|
work_dir / "spotify" / "playlistTracks.json",
|
|
work_dir / "spotify" / "tracks.json",
|
|
work_dir / "spotify" / "top_tracks.json",
|
|
work_dir / "spotify" / "top_artists.json",
|
|
)
|
|
|
|
LibrarySaver(library_parsed).save(work_dir / "spotify_library.json")
|
|
|
|
@staticmethod
|
|
def parse_itunes_library():
|
|
parser = ParseItunesXml()
|
|
|
|
library_parsed = parser.parse(
|
|
work_dir / "itunes" / "Library.xml",
|
|
work_dir / "itunes" / "Library.json"
|
|
)
|
|
|
|
LibrarySaver(library_parsed).save(work_dir / "itunes_library.json")
|
|
|
|
@staticmethod
|
|
def parse_local_library():
|
|
parser = LocalParser()
|
|
library_parsed = parser.parse(work_dir / "local" / "local_library.json")
|
|
LibrarySaver(library_parsed).save(work_dir / "local_library.json")
|
|
|
|
|
|
def fetch_and_save_lyrics(self):
|
|
"""
|
|
Fetches and writes lyrics for tracks that don't have lyrics yet.
|
|
Displays stats and uses a progress bar.
|
|
"""
|
|
|
|
# ---------- Step 1: Collect tracks without lyrics ----------
|
|
tracks_missing_lyrics = [
|
|
track for track in self.local_library.tracks
|
|
if len(track.artists) > 0 and not track.has_lyrics
|
|
]
|
|
|
|
total_tracks = len(self.local_library.tracks)
|
|
missing_count = len(tracks_missing_lyrics)
|
|
|
|
print(f"[INFO] Total tracks in library: {total_tracks}")
|
|
print(f"[INFO] Tracks missing lyrics: {missing_count}")
|
|
|
|
if missing_count == 0:
|
|
print("[INFO] No tracks need lyrics. Exiting.")
|
|
return
|
|
|
|
# ---------- Step 2: Fetch and write lyrics with progress bar ----------
|
|
for track in tqdm(tracks_missing_lyrics, desc="Fetching lyrics", unit="track"):
|
|
artist_name = track.artists[0].title
|
|
album_title = track.album.title if track.album else "Unknown Album"
|
|
print(f"\n[INFO] Processing: '{track.title}' by '{artist_name}' on '{album_title}'")
|
|
|
|
lyrics_text = Lyrics.fetch_lyrics(artist_name, album_title, track.title, track.duration_ms)
|
|
|
|
if lyrics_text:
|
|
print("[SUCCESS] Lyrics found, writing to file...")
|
|
file_path = local_library_path / track.local_path
|
|
try:
|
|
Lyrics.write_lyrics(file_path, lyrics_text)
|
|
print(f"[SUCCESS] Lyrics written for '{track.title}'")
|
|
except Exception as e:
|
|
print(f"[ERROR] Failed to write lyrics for '{track.title}': {e}")
|
|
else:
|
|
print(f"[WARNING] Lyrics not found for '{track.title}'")
|
|
|
|
|
|
def load_libraries(self):
|
|
self.spotify_library = LibraryLoader().load(work_dir / "spotify_library.json")
|
|
self.itunes_library = LibraryLoader().load(work_dir / "itunes_library.json")
|
|
self.local_library = LibraryLoader().load(work_dir / "local_library.json")
|
|
|
|
def save_libraries(self):
|
|
LibrarySaver(self.spotify_library).save(work_dir / "spotify_library.json")
|
|
LibrarySaver(self.local_library).save(work_dir / "local_library.json")
|
|
|
|
def map_local_to_spotify(self):
|
|
resolve_remote_tracks(client_id, client_secret, self.local_library, self.spotify_library, work_dir)
|
|
|
|
def log_stats(self):
|
|
log_stats(work_dir / "remote_to_local_map.json", work_dir / "stats", self.local_library, self.spotify_library)
|
|
|
|
|
|
def run(self, arg):
|
|
self.fetch_local()
|
|
# self.do_fetch_spotify()
|
|
|
|
self.parse_local_library()
|
|
self.parse_spotify_library()
|
|
|
|
self.load_libraries()
|
|
|
|
# self.map_local_to_spotify()
|
|
|
|
return True |