Compare commits

..

2 Commits

19 changed files with 400 additions and 77 deletions
+41 -15
View File
@@ -18,25 +18,51 @@ Lyrics are fetched using a fallback pipeline (first synced result wins):
See `lrcfetch --help` for full command reference. Common use cases: See `lrcfetch --help` for full command reference. Common use cases:
```bash - Fetch lyrics for the currently playing track:
# Fetch lyrics for the currently playing track
lrcfetch fetch
# Search by metadata (bypasses MPRIS) ```bash
lrcfetch search -t "Song Title" -a "Artist" lrcfetch fetch
```
# Export to .lrc file using a specific player or source to fetch from:
lrcfetch export
# Force a specific source ```bash
lrcfetch fetch --method spotify lrcfetch --player mpd fetch --method lrclib-search
```
# Cache management - Search by metadata (bypasses MPRIS):
lrcfetch cache stats # show cache statistics
lrcfetch cache query # query cache for current track ```bash
lrcfetch cache clear # clears cache of current track lrcfetch search -t "My Love" -a "Westlife"
lrcfetch cache clear --all # clears entire cache lrcfetch search --trackid "5p0ietGkLNEqx1Z7ijkw5g"
``` ```
or for a local file:
```bash
lrcfetch search --path "/path/to/Westlife - My Love.flac"
```
- Export to sidecar `.lrc` file:
```bash
lrcfetch export
```
or to a custom path:
```bash
lrcfetch export --output /path/to/lyrics.lrc
```
- Cache management:
```bash
lrcfetch cache stats # show cache statistics
lrcfetch cache query # query cache for current track
lrcfetch cache clear # clears cache of current track
lrcfetch cache clear --all # clears entire cache
```
## Configuration ## Configuration
+1 -1
View File
@@ -1 +1 @@
__version__ = "0.1.4" __version__ = "0.1.5"
+25 -4
View File
@@ -7,14 +7,17 @@ Description: CLI interface
import sys import sys
import time import time
import os import os
from pathlib import Path
from typing import Annotated from typing import Annotated
from urllib.parse import quote
import cyclopts import cyclopts
from loguru import logger from loguru import logger
from .config import enable_debug from .config import enable_debug
from .models import TrackMeta, CacheStatus from .models import TrackMeta, CacheStatus
from .mpris import get_current_track from .mpris import get_current_track
from .core import LrcManager, FetcherMethodType from .core import LrcManager
from .fetchers import FetcherMethodType
from .lrc import get_sidecar_path from .lrc import get_sidecar_path
@@ -108,8 +111,8 @@ def fetch(
def search( def search(
*, *,
title: Annotated[ title: Annotated[
str, cyclopts.Parameter(name=["--title", "-t"], help="Track title.") str | None, cyclopts.Parameter(name=["--title", "-t"], help="Track title.")
], ] = None,
artist: Annotated[ artist: Annotated[
str | None, cyclopts.Parameter(name=["--artist", "-a"], help="Artist name.") str | None, cyclopts.Parameter(name=["--artist", "-a"], help="Artist name.")
] = None, ] = None,
@@ -122,7 +125,17 @@ def search(
), ),
] = None, ] = None,
url: Annotated[ url: Annotated[
str | None, cyclopts.Parameter(help="Local file URL (file:///...).") str | None,
cyclopts.Parameter(
help="Local file URL (file:///...). Mutually exclusive with --path."
),
] = None,
path: Annotated[
str | None,
cyclopts.Parameter(
name=["--path"],
help="Local audio file path. Mutually exclusive with --url.",
),
] = None, ] = None,
method: Annotated[ method: Annotated[
FetcherMethodType | None, cyclopts.Parameter(help="Force a specific source.") FetcherMethodType | None, cyclopts.Parameter(help="Force a specific source.")
@@ -141,6 +154,14 @@ def search(
] = False, ] = False,
): ):
"""Search for lyrics by metadata (bypasses MPRIS).""" """Search for lyrics by metadata (bypasses MPRIS)."""
if url and path:
logger.error("--url and --path are mutually exclusive.")
sys.exit(1)
if path:
resolved = str(Path(path).resolve())
url = "file://" + quote(resolved, safe="/")
track = TrackMeta( track = TrackMeta(
title=title, title=title,
artist=artist, artist=artist,
+9 -47
View File
@@ -14,39 +14,14 @@ Fetch pipeline:
from typing import Optional from typing import Optional
from loguru import logger from loguru import logger
from typing import Literal
from .fetchers.netease import NeteaseFetcher from .fetchers import FetcherMethodType, create_fetchers
from .fetchers.qqmusic import QQMusicFetcher
from .fetchers.lrclib_search import LrclibSearchFetcher
from .fetchers.lrclib import LrclibFetcher
from .fetchers.spotify import SpotifyFetcher
from .fetchers.local import LocalFetcher
from .fetchers.cache_search import CacheSearchFetcher
from .fetchers.base import BaseFetcher from .fetchers.base import BaseFetcher
from .cache import CacheEngine from .cache import CacheEngine
from .lrc import LRC_LINE_RE, normalize_tags from .lrc import LRC_LINE_RE, normalize_tags
from .config import TTL_SYNCED, TTL_UNSYNCED, TTL_NOT_FOUND, TTL_NETWORK_ERROR from .config import TTL_SYNCED, TTL_UNSYNCED, TTL_NOT_FOUND, TTL_NETWORK_ERROR
from .models import TrackMeta, LyricResult, CacheStatus from .models import TrackMeta, LyricResult, CacheStatus
from .enrichers import enrich_track
METHODS = (
"local",
"cache-search",
"spotify",
"lrclib",
"lrclib-search",
"netease",
"qqmusic",
)
FetcherMethodType = Literal[
"local",
"cache-search",
"spotify",
"lrclib",
"lrclib-search",
"netease",
"qqmusic",
]
def _normalize_unsynced(lyrics: str) -> str: def _normalize_unsynced(lyrics: str) -> str:
@@ -81,23 +56,9 @@ _STATUS_TTL: dict[CacheStatus, Optional[int]] = {
class LrcManager: class LrcManager:
"""Main entry point for fetching lyrics with caching.""" """Main entry point for fetching lyrics with caching."""
# Fetchers that manage their own cache logic (skip per-source cache check)
_SELF_CACHED = frozenset({"cache-search"})
def __init__(self) -> None: def __init__(self) -> None:
self.cache = CacheEngine() self.cache = CacheEngine()
self.fetchers: dict[FetcherMethodType, BaseFetcher] = { self.fetchers = create_fetchers(self.cache)
"local": LocalFetcher(),
"cache-search": CacheSearchFetcher(self.cache),
"spotify": SpotifyFetcher(),
"lrclib": LrclibFetcher(),
"lrclib-search": LrclibSearchFetcher(),
"netease": NeteaseFetcher(),
"qqmusic": QQMusicFetcher(),
}
assert set(self.fetchers) == set(METHODS), (
f"METHODS and fetchers out of sync: {set(METHODS) ^ set(self.fetchers)}"
)
def _build_sequence( def _build_sequence(
self, track: TrackMeta, force_method: Optional[FetcherMethodType] = None self, track: TrackMeta, force_method: Optional[FetcherMethodType] = None
@@ -142,6 +103,7 @@ class LrcManager:
After all sources are tried, returns the best result found After all sources are tried, returns the best result found
(synced > unsynced > None). (synced > unsynced > None).
""" """
track = enrich_track(track)
logger.info(f"Fetching lyrics for: {track.display_name()}") logger.info(f"Fetching lyrics for: {track.display_name()}")
sequence = self._build_sequence(track, force_method) sequence = self._build_sequence(track, force_method)
@@ -155,7 +117,7 @@ class LrcManager:
source = fetcher.source_name source = fetcher.source_name
# Cache check (skip for fetchers that handle their own caching) # Cache check (skip for fetchers that handle their own caching)
if not bypass_cache and source not in self._SELF_CACHED: if not bypass_cache and not fetcher.self_cached:
cached = self.cache.get(track, source) cached = self.cache.get(track, source)
if cached: if cached:
if cached.status == CacheStatus.SUCCESS_SYNCED: if cached.status == CacheStatus.SUCCESS_SYNCED:
@@ -176,12 +138,12 @@ class LrcManager:
f"[{source}] cache hit: {cached.status.value}, skipping" f"[{source}] cache hit: {cached.status.value}, skipping"
) )
continue continue
else: elif not fetcher.self_cached:
logger.debug(f"[{source}] cache bypassed") logger.debug(f"[{source}] cache bypassed")
# Fetch # Fetch
logger.debug(f"[{source}] calling fetcher...") logger.debug(f"[{source}] calling fetcher...")
result = fetcher.fetch(track) result = fetcher.fetch(track, bypass_cache=bypass_cache)
if not result: if not result:
logger.debug(f"[{source}] returned None (no result)") logger.debug(f"[{source}] returned None (no result)")
@@ -196,8 +158,8 @@ class LrcManager:
ttl=result.ttl, ttl=result.ttl,
) )
# Cache the normalized result (skip for read-only fetchers) # Cache the normalized result (skip for self-cached fetchers)
if source not in self._SELF_CACHED: if not fetcher.self_cached:
ttl = result.ttl or _STATUS_TTL.get(result.status, TTL_NOT_FOUND) ttl = result.ttl or _STATUS_TTL.get(result.status, TTL_NOT_FOUND)
self.cache.set(track, source, result, ttl_seconds=ttl) self.cache.set(track, source, result, ttl_seconds=ttl)
+39
View File
@@ -0,0 +1,39 @@
"""
Author: Uyanide pywang0608@foxmail.com
Date: 2026-03-31 06:09:11
Description: Metadata enrichment pipeline
"""
from loguru import logger
from .base import BaseEnricher
from .audio_tag import AudioTagEnricher
from .file_name import FileNameEnricher
from ..models import TrackMeta
# Enrichers run in order; earlier ones have higher priority.
_ENRICHERS: list[BaseEnricher] = [
AudioTagEnricher(),
FileNameEnricher(),
]
def enrich_track(track: TrackMeta) -> TrackMeta:
"""Run all enrichers and return a track with missing fields filled in.
Each enricher sees the cumulative state (earlier enrichers' results
are already applied). A field is only set if it is currently None.
"""
for enricher in _ENRICHERS:
try:
result = enricher.enrich(track)
except Exception as e:
logger.warning(f"Enricher {enricher.name} failed: {e}")
continue
if not result:
continue
# Only apply fields that are still None
updates = {k: v for k, v in result.items() if getattr(track, k, None) is None}
if updates:
track = track.model_copy(update=updates)
return track
+78
View File
@@ -0,0 +1,78 @@
"""
Author: Uyanide pywang0608@foxmail.com
Date: 2026-03-31 06:11:27
Description: Enricher that reads metadata from audio file tags (mutagen)
"""
from typing import Optional
from loguru import logger
from mutagen._file import File, FileType
from .base import BaseEnricher
from ..models import TrackMeta
from ..lrc import get_audio_path
class AudioTagEnricher(BaseEnricher):
"""Extract title, artist, album, and duration from audio file tags."""
@property
def name(self) -> str:
return "audio-tag"
def enrich(self, track: TrackMeta) -> Optional[dict]:
if not track.is_local or not track.url:
return None
audio_path = get_audio_path(track.url, ensure_exists=True)
if not audio_path:
return None
try:
audio = File(audio_path)
except Exception as e:
logger.debug(f"AudioTag: failed to read {audio_path}: {e}")
return None
if audio is None:
return None
updates: dict = {}
# Try common tag names (vorbis comments, ID3, MP4)
title = _first_tag(audio, "title", "TIT2", "\xa9nam")
if title and not track.title:
updates["title"] = title
artist = _first_tag(audio, "artist", "TPE1", "\xa9ART")
if artist and not track.artist:
updates["artist"] = artist
album = _first_tag(audio, "album", "TALB", "\xa9alb")
if album and not track.album:
updates["album"] = album
if not track.length and audio.info and hasattr(audio.info, "length"):
length_ms = int(audio.info.length * 1000)
if length_ms > 0:
updates["length"] = length_ms
if updates:
logger.debug(f"AudioTag: enriched fields: {list(updates.keys())}")
return updates or None
def _first_tag(audio: FileType, *keys: str) -> Optional[str]:
"""Return the first non-empty string value found among the given tag keys."""
if not audio.tags:
return None
for key in keys:
val = audio.tags.get(key)
if val is None:
continue
# mutagen returns lists for vorbis, single values for ID3
if isinstance(val, list):
val = val[0] if val else None
if val:
return str(val).strip()
return None
+31
View File
@@ -0,0 +1,31 @@
"""
Author: Uyanide pywang0608@foxmail.com
Date: 2026-03-31 06:08:16
Description: Base class for metadata enrichers
"""
from abc import ABC, abstractmethod
from typing import Optional
from ..models import TrackMeta
class BaseEnricher(ABC):
"""Attempts to fill missing fields on a TrackMeta.
Each enricher inspects the track, and returns a dict of field names
to values for any fields it can provide. Only fields that are
currently ``None`` on the track will actually be applied.
"""
@property
@abstractmethod
def name(self) -> str: ...
@abstractmethod
def enrich(self, track: TrackMeta) -> Optional[dict]:
"""Return a dict of {field_name: value} for fields this enricher can fill.
Return None or an empty dict if nothing can be contributed.
"""
...
+83
View File
@@ -0,0 +1,83 @@
"""
Author: Uyanide pywang0608@foxmail.com
Date: 2026-03-31 06:08:44
Description: Enricher that parses metadata from the audio file path
"""
import re
from typing import Optional
from loguru import logger
from .base import BaseEnricher
from ..models import TrackMeta
from ..lrc import get_audio_path
# Common track-number prefixes: "01 - ", "01. ", "1 - ", etc.
_TRACK_NUM_RE = re.compile(r"^\d{1,3}[\s.\-]+")
class FileNameEnricher(BaseEnricher):
"""Derive artist / title from the file path when tags are unavailable.
Heuristics (applied to the stem of the filename):
- "Artist - Title" → artist, title
- "01 - Title" → title only (leading track number stripped)
- "Title" → title only
If artist is still missing after parsing the filename, the parent
directory name is used as a guess (common layout: ``Artist/Album/track``).
"""
@property
def name(self) -> str:
return "file-name"
def enrich(self, track: TrackMeta) -> Optional[dict]:
if not track.is_local or not track.url:
return None
audio_path = get_audio_path(track.url, ensure_exists=False)
if not audio_path:
return None
updates: dict = {}
stem = audio_path.stem
# Try "Artist - Title" split
if " - " in stem:
left, right = stem.split(" - ", 1)
left = _TRACK_NUM_RE.sub("", left).strip()
right = right.strip()
if left and right:
# Both sides non-empty after stripping track number
if not track.artist:
updates["artist"] = left
if not track.title:
updates["title"] = right
elif right:
# Left was only a track number → right is the title
if not track.title:
updates["title"] = right
else:
# No separator: strip track number, remainder is title
title_guess = _TRACK_NUM_RE.sub("", stem).strip()
if title_guess and not track.title:
updates["title"] = title_guess
# Use parent directory as artist fallback
# Typical layout: /Music/Artist/Album/01 - Track.flac
if not track.artist and "artist" not in updates:
parents = audio_path.parents
if len(parents) >= 2:
album_dir = parents[0].name
artist_dir = parents[1].name
if artist_dir and artist_dir not in (".", "/"):
updates["artist"] = artist_dir
if not track.album and album_dir and album_dir != artist_dir:
updates["album"] = album_dir
if updates:
logger.debug(f"FileName: enriched fields: {list(updates.keys())}")
return updates or None
+54
View File
@@ -0,0 +1,54 @@
"""
Author: Uyanide pywang0608@foxmail.com
Date: 2026-03-25 02:33:26
Description: Fetcher pipeline — registry and types
"""
from typing import Literal
from .base import BaseFetcher
from .local import LocalFetcher
from .cache_search import CacheSearchFetcher
from .spotify import SpotifyFetcher
from .lrclib import LrclibFetcher
from .lrclib_search import LrclibSearchFetcher
from .netease import NeteaseFetcher
from .qqmusic import QQMusicFetcher
from ..cache import CacheEngine
METHODS = (
"local",
"cache-search",
"spotify",
"lrclib",
"lrclib-search",
"netease",
"qqmusic",
)
FetcherMethodType = Literal[
"local",
"cache-search",
"spotify",
"lrclib",
"lrclib-search",
"netease",
"qqmusic",
]
def create_fetchers(cache: CacheEngine) -> dict[str, BaseFetcher]:
"""Instantiate all fetchers. Returns a dict keyed by source name."""
fetchers: dict[str, BaseFetcher] = {
"local": LocalFetcher(),
"cache-search": CacheSearchFetcher(cache),
"spotify": SpotifyFetcher(),
"lrclib": LrclibFetcher(),
"lrclib-search": LrclibSearchFetcher(),
"netease": NeteaseFetcher(),
"qqmusic": QQMusicFetcher(),
}
assert set(fetchers) == set(METHODS), (
f"METHODS and fetchers out of sync: {set(METHODS) ^ set(fetchers)}"
)
return fetchers
+8 -1
View File
@@ -17,7 +17,14 @@ class BaseFetcher(ABC):
"""Name of the fetcher source.""" """Name of the fetcher source."""
pass pass
@property
def self_cached(self) -> bool:
"""True if this fetcher manages its own cache (skip per-source cache check)."""
return False
@abstractmethod @abstractmethod
def fetch(self, track: TrackMeta) -> Optional[LyricResult]: def fetch(
self, track: TrackMeta, bypass_cache: bool = False
) -> Optional[LyricResult]:
"""Fetch lyrics for the given track. Returns None if unable to fetch.""" """Fetch lyrics for the given track. Returns None if unable to fetch."""
pass pass
+11 -1
View File
@@ -26,7 +26,17 @@ class CacheSearchFetcher(BaseFetcher):
def source_name(self) -> str: def source_name(self) -> str:
return "cache-search" return "cache-search"
def fetch(self, track: TrackMeta) -> Optional[LyricResult]: @property
def self_cached(self) -> bool:
return True
def fetch(
self, track: TrackMeta, bypass_cache: bool = False
) -> Optional[LyricResult]:
if bypass_cache:
logger.debug("Cache-search: bypassed by caller")
return None
if not track.title: if not track.title:
logger.debug("Cache-search: skipped — no title") logger.debug("Cache-search: skipped — no title")
return None return None
+3 -1
View File
@@ -25,7 +25,9 @@ class LocalFetcher(BaseFetcher):
def source_name(self) -> str: def source_name(self) -> str:
return "local" return "local"
def fetch(self, track: TrackMeta) -> Optional[LyricResult]: def fetch(
self, track: TrackMeta, bypass_cache: bool = False
) -> Optional[LyricResult]:
"""Attempt to read lyrics from local filesystem.""" """Attempt to read lyrics from local filesystem."""
if not track.is_local or not track.url: if not track.is_local or not track.url:
return None return None
+3 -1
View File
@@ -30,7 +30,9 @@ class LrclibFetcher(BaseFetcher):
def source_name(self) -> str: def source_name(self) -> str:
return "lrclib" return "lrclib"
def fetch(self, track: TrackMeta) -> Optional[LyricResult]: def fetch(
self, track: TrackMeta, bypass_cache: bool = False
) -> Optional[LyricResult]:
"""Fetch lyrics from LRCLIB. Requires complete metadata.""" """Fetch lyrics from LRCLIB. Requires complete metadata."""
if not track.is_complete: if not track.is_complete:
logger.debug("LRCLIB: skipped — incomplete metadata") logger.debug("LRCLIB: skipped — incomplete metadata")
+3 -1
View File
@@ -32,7 +32,9 @@ class LrclibSearchFetcher(BaseFetcher):
def source_name(self) -> str: def source_name(self) -> str:
return "lrclib-search" return "lrclib-search"
def fetch(self, track: TrackMeta) -> Optional[LyricResult]: def fetch(
self, track: TrackMeta, bypass_cache: bool = False
) -> Optional[LyricResult]:
"""Search LRCLIB for lyrics. Requires at least a title.""" """Search LRCLIB for lyrics. Requires at least a title."""
if not track.title: if not track.title:
logger.debug("LRCLIB-search: skipped — no title") logger.debug("LRCLIB-search: skipped — no title")
+3 -1
View File
@@ -194,7 +194,9 @@ class NeteaseFetcher(BaseFetcher):
logger.error(f"Netease: lyric fetch failed for song_id={song_id}: {e}") logger.error(f"Netease: lyric fetch failed for song_id={song_id}: {e}")
return LyricResult(status=CacheStatus.NETWORK_ERROR, ttl=TTL_NETWORK_ERROR) return LyricResult(status=CacheStatus.NETWORK_ERROR, ttl=TTL_NETWORK_ERROR)
def fetch(self, track: TrackMeta) -> Optional[LyricResult]: def fetch(
self, track: TrackMeta, bypass_cache: bool = False
) -> Optional[LyricResult]:
"""Search for the track and fetch its lyrics.""" """Search for the track and fetch its lyrics."""
query = f"{track.artist or ''} {track.title or ''}".strip() query = f"{track.artist or ''} {track.title or ''}".strip()
if not query: if not query:
+3 -1
View File
@@ -155,7 +155,9 @@ class QQMusicFetcher(BaseFetcher):
logger.error(f"QQMusic: lyric fetch failed for mid={mid}: {e}") logger.error(f"QQMusic: lyric fetch failed for mid={mid}: {e}")
return LyricResult(status=CacheStatus.NETWORK_ERROR, ttl=TTL_NETWORK_ERROR) return LyricResult(status=CacheStatus.NETWORK_ERROR, ttl=TTL_NETWORK_ERROR)
def fetch(self, track: TrackMeta) -> Optional[LyricResult]: def fetch(
self, track: TrackMeta, bypass_cache: bool = False
) -> Optional[LyricResult]:
"""Search for the track and fetch its lyrics.""" """Search for the track and fetch its lyrics."""
if not QQ_MUSIC_API_URL: if not QQ_MUSIC_API_URL:
logger.debug("QQMusic: skipped — QQ_MUSIC_API_URL not configured") logger.debug("QQMusic: skipped — QQ_MUSIC_API_URL not configured")
+3 -1
View File
@@ -274,7 +274,9 @@ class SpotifyFetcher(BaseFetcher):
continue continue
return False return False
def fetch(self, track: TrackMeta) -> Optional[LyricResult]: def fetch(
self, track: TrackMeta, bypass_cache: bool = False
) -> Optional[LyricResult]:
"""Fetch lyrics for a Spotify track by its track ID.""" """Fetch lyrics for a Spotify track by its track ID."""
if not track.trackid: if not track.trackid:
logger.debug("Spotify: skipped — no trackid in metadata") logger.debug("Spotify: skipped — no trackid in metadata")
+1 -1
View File
@@ -4,7 +4,7 @@ build-backend = "hatchling.build"
[project] [project]
name = "lrcfetch" name = "lrcfetch"
version = "0.1.4" version = "0.1.5"
description = "Fetch line-synced lyrics for your music player." description = "Fetch line-synced lyrics for your music player."
readme = "README.md" readme = "README.md"
requires-python = ">=3.13" requires-python = ">=3.13"
Generated
+1 -1
View File
@@ -153,7 +153,7 @@ wheels = [
[[package]] [[package]]
name = "lrcfetch" name = "lrcfetch"
version = "0.1.4" version = "0.1.5"
source = { editable = "." } source = { editable = "." }
dependencies = [ dependencies = [
{ name = "cyclopts" }, { name = "cyclopts" },