lyrics fetcher
This commit is contained in:
parent
9c9ddb7a9e
commit
5d38a146a1
8 changed files with 277 additions and 23 deletions
51
Flow.py
51
Flow.py
|
|
@ -1,7 +1,11 @@
|
|||
|
||||
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.spotify.SpotifyWebAPI import fetch_all
|
||||
|
|
@ -15,8 +19,8 @@ from src.StatGenerator import log_stats
|
|||
|
||||
|
||||
class Flow:
|
||||
spotify_library = None
|
||||
local_library = None
|
||||
spotify_library : Library = None
|
||||
local_library : Library = None
|
||||
|
||||
@staticmethod
|
||||
def fetch_spotify(self):
|
||||
|
|
@ -45,6 +49,49 @@ class Flow:
|
|||
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.local_library = LibraryLoader().load(work_dir / "local_library.json")
|
||||
|
|
|
|||
20
main.py
20
main.py
|
|
@ -1,7 +1,21 @@
|
|||
|
||||
|
||||
import Flow
|
||||
|
||||
if __name__ == '__main__':
|
||||
def test1():
|
||||
flow = Flow.Flow()
|
||||
|
||||
# flow.fetch_local()
|
||||
flow.parse_local_library()
|
||||
flow.load_libraries()
|
||||
|
||||
# flow.map_local_to_spotify()
|
||||
# flow.log_stats()
|
||||
|
||||
flow.fetch_and_save_lyrics()
|
||||
|
||||
|
||||
def main():
|
||||
flow = Flow.Flow()
|
||||
|
||||
# flow.fetch_spotify()
|
||||
|
|
@ -17,3 +31,7 @@ if __name__ == '__main__':
|
|||
flow.log_stats()
|
||||
|
||||
print("asd")
|
||||
|
||||
|
||||
test1()
|
||||
# main()
|
||||
|
|
|
|||
|
|
@ -22,6 +22,7 @@ class Track:
|
|||
self.duration_ms : int = None
|
||||
self.added_at : datetime = None
|
||||
self.local_path : Path = None
|
||||
self.has_lyrics : bool = False
|
||||
|
||||
|
||||
class Album:
|
||||
|
|
|
|||
|
|
@ -45,6 +45,7 @@ class LibrarySaver:
|
|||
"duration_ms": track.duration_ms,
|
||||
"added_at": _dt_to_str(track.added_at),
|
||||
"local_path": track.local_path,
|
||||
"has_lyrics": track.has_lyrics,
|
||||
}
|
||||
|
||||
@staticmethod
|
||||
|
|
@ -119,6 +120,7 @@ class LibraryLoader:
|
|||
track.local_path = data.get("local_path")
|
||||
track.artists = [self.get_artist(artist_id) for artist_id in data.get("artists")]
|
||||
track.album = self.get_album(data.get("album"))
|
||||
track.has_lyrics = data.get("has_lyrics")
|
||||
|
||||
self.track_map[track.id] = track
|
||||
return track
|
||||
|
|
|
|||
|
|
@ -1,8 +1,8 @@
|
|||
import os
|
||||
import fnmatch
|
||||
import mutagen
|
||||
import eyed3
|
||||
from mutagen.flac import FLAC
|
||||
from mutagen import File as MutagenFile
|
||||
|
||||
from src.helpers import *
|
||||
|
||||
|
|
@ -11,6 +11,52 @@ music_extensions = ('*.mp3', '*.flac', '*.wav', '*.aac', '*.ogg', '*.m4a')
|
|||
exclude_directories = ("*mary--*", "*example-word*")
|
||||
|
||||
|
||||
from mutagen.id3 import ID3
|
||||
from mutagen.id3 import ID3NoHeaderError
|
||||
|
||||
|
||||
def detect_lyrics(abs_path, ext):
|
||||
ext = ext.lower()
|
||||
|
||||
# ---------- MP3 ----------
|
||||
if ext == ".mp3":
|
||||
try:
|
||||
tags = ID3(abs_path)
|
||||
except ID3NoHeaderError:
|
||||
return False
|
||||
|
||||
return bool(
|
||||
tags.getall("SYLT") or
|
||||
tags.getall("USLT")
|
||||
)
|
||||
|
||||
# ---------- FLAC / OGG / OPUS ----------
|
||||
if ext in {".flac", ".ogg", ".opus"}:
|
||||
try:
|
||||
audio = MutagenFile(abs_path)
|
||||
if not audio or not audio.tags:
|
||||
return False
|
||||
|
||||
for key in audio.tags.keys():
|
||||
k = key.upper()
|
||||
if k in {"LYRICS", "LRC", "SYNCEDLYRICS"}:
|
||||
if audio.tags[key]:
|
||||
return True
|
||||
except Exception:
|
||||
return False
|
||||
|
||||
# ---------- Fallback ----------
|
||||
try:
|
||||
audio = MutagenFile(abs_path)
|
||||
if audio and audio.tags:
|
||||
for key in audio.tags.keys():
|
||||
if "LYRIC" in key.upper():
|
||||
return True
|
||||
except Exception:
|
||||
pass
|
||||
|
||||
return False
|
||||
|
||||
def update_local_library(root_dir, output_dir):
|
||||
|
||||
set_workdir(output_dir / "local")
|
||||
|
|
@ -29,50 +75,77 @@ def update_local_library(root_dir, output_dir):
|
|||
return prev_id
|
||||
|
||||
for dir_path, dir_names, filenames in os.walk(root_dir):
|
||||
dir_names[:] = [d for d in dir_names if not any(fnmatch.fnmatch(d, exclude) for exclude in exclude_directories)]
|
||||
filtered_filenames = [filename for filename in filenames if any(fnmatch.fnmatch(filename, ext) for ext in music_extensions)]
|
||||
dir_names[:] = [
|
||||
d for d in dir_names
|
||||
if not any(fnmatch.fnmatch(d, exclude) for exclude in exclude_directories)
|
||||
]
|
||||
|
||||
filtered_filenames = [
|
||||
filename for filename in filenames
|
||||
if any(fnmatch.fnmatch(filename, ext) for ext in music_extensions)
|
||||
]
|
||||
|
||||
for filename in filtered_filenames:
|
||||
song_name, track_type = os.path.splitext(filename)
|
||||
relative_path = os.path.relpath(os.path.join(dir_path, filename), root_dir)
|
||||
song_id = get_new_song_id(song_id, relative_path)
|
||||
|
||||
new_library[song_id] = {
|
||||
"name": song_name,
|
||||
"path": relative_path,
|
||||
"type": track_type,
|
||||
"artist": "",
|
||||
"album": ""
|
||||
"album": "",
|
||||
"duration": None
|
||||
}
|
||||
|
||||
for track_id, track in new_library.items():
|
||||
abs_path = os.path.join(root_dir, track['path'])
|
||||
match track['type']:
|
||||
|
||||
track_abs_path = os.path.join(root_dir, track["path"])
|
||||
track["has_lyrics"] = detect_lyrics(track_abs_path, track['type'].lower())
|
||||
|
||||
match track['type'].lower():
|
||||
case ".mp3":
|
||||
mp3track = eyed3.load(abs_path)
|
||||
if not mp3track:
|
||||
print(f"Failed to load mp3 {abs_path} using just name of the file as track name")
|
||||
if not mp3track or not mp3track.info:
|
||||
print(f"Failed to load mp3 {abs_path}")
|
||||
track['name'] = track['path']
|
||||
else:
|
||||
track['duration'] = mp3track.info.time_secs
|
||||
|
||||
tag = mp3track.tag
|
||||
if not tag:
|
||||
print(f"Invalid mp3 header {abs_path} using just name of the file as track name")
|
||||
track['name'] = track['path']
|
||||
else:
|
||||
if tag:
|
||||
track['artist'] = tag.artist
|
||||
track['name'] = tag.title
|
||||
track['album'] = tag.album
|
||||
|
||||
case ".flac":
|
||||
try:
|
||||
metadata = FLAC(abs_path).tags
|
||||
if 'artist' in metadata:
|
||||
track['artist'] = " ".join(metadata['artist'])
|
||||
if 'TITLE' in metadata:
|
||||
track['name'] = " ".join(metadata['TITLE'])
|
||||
if 'album' in metadata:
|
||||
track['album'] = " ".join(metadata['album'])
|
||||
audio = FLAC(abs_path)
|
||||
|
||||
if audio.info:
|
||||
track['duration'] = audio.info.length
|
||||
|
||||
metadata = audio.tags
|
||||
if metadata:
|
||||
if 'artist' in metadata:
|
||||
track['artist'] = " ".join(metadata['artist'])
|
||||
if 'TITLE' in metadata:
|
||||
track['name'] = " ".join(metadata['TITLE'])
|
||||
if 'album' in metadata:
|
||||
track['album'] = " ".join(metadata['album'])
|
||||
|
||||
except mutagen.MutagenError:
|
||||
print(f"Invalid flac header {abs_path} using just name of the file as track name")
|
||||
print(f"Invalid flac header {abs_path}")
|
||||
track['name'] = track['path']
|
||||
|
||||
case _:
|
||||
try:
|
||||
audio = MutagenFile(abs_path)
|
||||
if audio and audio.info:
|
||||
track['duration'] = audio.info.length
|
||||
except Exception:
|
||||
pass
|
||||
|
||||
save_data(new_library, "local_library")
|
||||
|
|
|
|||
|
|
@ -1,7 +1,7 @@
|
|||
|
||||
from pathlib import Path
|
||||
|
||||
def generate_playlists(local_library, library_path, remote_playlists, link_pattern, output_dir):
|
||||
def ():
|
||||
output_dir = Path(output_dir)
|
||||
output_playlists = []
|
||||
|
||||
|
|
|
|||
|
|
@ -116,6 +116,8 @@ class LocalParser:
|
|||
track.artists = self.parse_artists(data.get("artist"))
|
||||
track.album = self.parse_album(data.get("album"))
|
||||
track.local_path = data.get("path")
|
||||
track.duration_ms = data.get("duration")
|
||||
track.has_lyrics = data.get("has_lyrics")
|
||||
|
||||
self.track_map[track.id] = track
|
||||
self.library.tracks.append(track)
|
||||
|
|
|
|||
111
src/lyrics/lyrics.py
Normal file
111
src/lyrics/lyrics.py
Normal file
|
|
@ -0,0 +1,111 @@
|
|||
import requests
|
||||
from mutagen import File
|
||||
from mutagen.id3 import ID3, USLT, ID3NoHeaderError
|
||||
import os
|
||||
|
||||
def fetch_lyrics(artist: str, album: str, track: str, duration: int = 0) -> str:
|
||||
|
||||
skip_artists = [
|
||||
"shurupis",
|
||||
"unknown"
|
||||
]
|
||||
|
||||
skip_album = [
|
||||
"unknown"
|
||||
]
|
||||
|
||||
if artist in skip_artists or album in skip_album:
|
||||
return ""
|
||||
|
||||
|
||||
url = "https://lrclib.net/api/get"
|
||||
params = {
|
||||
"artist_name": artist,
|
||||
"track_name": track,
|
||||
"album_name": album,
|
||||
"duration": duration,
|
||||
}
|
||||
try:
|
||||
resp = requests.get(url, params=params, timeout=10)
|
||||
if resp.status_code != 200:
|
||||
try:
|
||||
err = resp.json()
|
||||
print(err.get("error"), "-", err.get("message"))
|
||||
except ValueError:
|
||||
print("HTTPError", resp.status_code)
|
||||
return ""
|
||||
data = resp.json()
|
||||
if "error" in data:
|
||||
print(data.get("error"), "-", data.get("message"))
|
||||
return ""
|
||||
return data.get("syncedLyrics", "")
|
||||
except requests.RequestException as e:
|
||||
print(type(e).__name__, "-", str(e))
|
||||
return ""
|
||||
except ValueError as e:
|
||||
print(type(e).__name__, "-", str(e))
|
||||
return ""
|
||||
|
||||
|
||||
def write_lyrics(file_path, lyrics_text, lang="eng"):
|
||||
"""
|
||||
Writes lyrics text into tags recognized by Navidrome.
|
||||
Provides detailed logs and success/failure info.
|
||||
"""
|
||||
|
||||
ext = os.path.splitext(file_path)[1].lower()
|
||||
print(f"[INFO] Processing file: {file_path} (extension: {ext})")
|
||||
|
||||
try:
|
||||
# ---------- MP3 ----------
|
||||
if ext == ".mp3":
|
||||
print("[INFO] Detected MP3 file")
|
||||
try:
|
||||
tags = ID3(file_path)
|
||||
print("[INFO] Existing ID3 tags loaded")
|
||||
except ID3NoHeaderError:
|
||||
tags = ID3()
|
||||
print("[INFO] No ID3 header found, creating new tags")
|
||||
|
||||
tags.delall("USLT")
|
||||
print("[INFO] Removed existing USLT frames (lyrics)")
|
||||
|
||||
tags.add(
|
||||
USLT(
|
||||
encoding=3, # UTF-8
|
||||
lang=lang,
|
||||
desc="",
|
||||
text=lyrics_text
|
||||
)
|
||||
)
|
||||
tags.save(file_path)
|
||||
print("[SUCCESS] Lyrics written to MP3 tags")
|
||||
return
|
||||
|
||||
# ---------- FLAC / OGG / OPUS ----------
|
||||
audio = File(file_path)
|
||||
if audio is None:
|
||||
print(f"[ERROR] Unsupported or unrecognized file format: {file_path}")
|
||||
return
|
||||
|
||||
if ext in {".flac", ".ogg", ".opus"}:
|
||||
print(f"[INFO] Detected {ext.upper()} file")
|
||||
audio["LRC"] = lyrics_text
|
||||
audio["SYNCEDLYRICS"] = lyrics_text
|
||||
audio.save()
|
||||
print(f"[SUCCESS] Lyrics written to {ext.upper()} tags")
|
||||
return
|
||||
|
||||
# ---------- Fallback ----------
|
||||
if audio.tags is not None:
|
||||
print("[INFO] Using fallback tag writing")
|
||||
audio["LYRICS"] = lyrics_text
|
||||
audio.save()
|
||||
print("[SUCCESS] Lyrics written using fallback tags")
|
||||
return
|
||||
|
||||
print(f"[WARNING] No suitable tag frame found for {file_path}")
|
||||
|
||||
except Exception as e:
|
||||
print(f"[ERROR] Failed to write lyrics to {file_path}: {e}")
|
||||
|
||||
Loading…
Add table
Add a link
Reference in a new issue