From 4182229ae23443d94e32662b9fc5479feb054ce1 Mon Sep 17 00:00:00 2001 From: Uyanide Date: Fri, 27 Mar 2026 12:52:45 +0100 Subject: [PATCH] =?UTF-8?q?=F0=9F=9A=A8=20lint?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- lrcfetch/cache.py | 32 +++++----- lrcfetch/cli.py | 60 ++++++++++--------- lrcfetch/config.py | 47 ++++++++------- lrcfetch/core.py | 46 ++++++++------ lrcfetch/fetchers/base.py | 9 ++- lrcfetch/fetchers/local.py | 18 ++++-- lrcfetch/fetchers/lrclib.py | 23 ++++--- lrcfetch/fetchers/lrclib_search.py | 46 ++++++++++---- lrcfetch/fetchers/netease.py | 48 ++++++++++----- lrcfetch/fetchers/spotify.py | 45 +++++++------- lrcfetch/lrc.py | 14 +++-- lrcfetch/models.py | 19 ++++-- lrcfetch/mpris.py | 96 ++++++++++++++++++++---------- 13 files changed, 316 insertions(+), 187 deletions(-) diff --git a/lrcfetch/cache.py b/lrcfetch/cache.py index da69e54..db39af4 100644 --- a/lrcfetch/cache.py +++ b/lrcfetch/cache.py @@ -1,13 +1,18 @@ -"""SQLite-based lyric cache with per-source storage and TTL expiration.""" +""" +Author: Uyanide pywang0608@foxmail.com +Date: 2026-03-25 10:18:03 +Description: SQLite-based lyric cache with per-source storage and TTL expiration +""" import sqlite3 import hashlib import time from typing import Optional -from lrcfetch.config import DB_PATH -from lrcfetch.models import TrackMeta, LyricResult, CacheStatus from loguru import logger +from .config import DB_PATH +from .models import TrackMeta, LyricResult, CacheStatus + def _generate_key(track: TrackMeta, source: str) -> str: """Generate a unique cache key from track metadata and source. @@ -64,9 +69,7 @@ class CacheEngine: """) conn.commit() - # ------------------------------------------------------------------ # Read - # ------------------------------------------------------------------ def get(self, track: TrackMeta, source: str) -> Optional[LyricResult]: """Look up a cached result for *track* from *source*. @@ -126,9 +129,7 @@ class CacheEngine: best = cached return best - # ------------------------------------------------------------------ # Write - # ------------------------------------------------------------------ def set( self, @@ -171,9 +172,7 @@ class CacheEngine: f"[{result.status.value}, ttl={ttl_seconds}s]" ) - # ------------------------------------------------------------------ # Delete - # ------------------------------------------------------------------ def clear_all(self) -> None: """Remove every entry from the cache.""" @@ -193,7 +192,9 @@ class CacheEngine: cur = conn.execute(f"DELETE FROM cache WHERE {where}", params) conn.commit() if cur.rowcount: - logger.info(f"Cleared {cur.rowcount} cache entries for {track.display_name()}.") + logger.info( + f"Cleared {cur.rowcount} cache entries for {track.display_name()}." + ) else: logger.info(f"No cache entries found for {track.display_name()}.") @@ -225,9 +226,7 @@ class CacheEngine: params.append(track.album) return conditions, params - # ------------------------------------------------------------------ # Query / inspect - # ------------------------------------------------------------------ def query_track(self, track: TrackMeta) -> list[dict]: """Return all cached rows for a given track (across all sources).""" @@ -237,9 +236,12 @@ class CacheEngine: where = " AND ".join(conditions) with sqlite3.connect(self.db_path) as conn: conn.row_factory = sqlite3.Row - return [dict(r) for r in conn.execute( - f"SELECT * FROM cache WHERE {where}", params - ).fetchall()] + return [ + dict(r) + for r in conn.execute( + f"SELECT * FROM cache WHERE {where}", params + ).fetchall() + ] def query_all(self) -> list[dict]: """Return every row in the cache table.""" diff --git a/lrcfetch/cli.py b/lrcfetch/cli.py index c47a57b..62beba7 100644 --- a/lrcfetch/cli.py +++ b/lrcfetch/cli.py @@ -1,4 +1,8 @@ -"""CLI interface for lrcfetch.""" +""" +Author: Uyanide pywang0608@foxmail.com +Date: 2026-03-26 02:04:39 +Description: CLI interface +""" import typer import time @@ -6,10 +10,10 @@ from typing import Optional from loguru import logger import os -from lrcfetch.config import enable_debug -from lrcfetch.models import TrackMeta, CacheStatus -from lrcfetch.mpris import get_current_track -from lrcfetch.core import LrcManager +from .config import enable_debug +from .models import TrackMeta, CacheStatus +from .mpris import get_current_track +from .core import LrcManager app = typer.Typer( help="LRCFetch — Fetch line-synced lyrics for your music player.", @@ -26,7 +30,10 @@ _player: Optional[str] = None def main( debug: bool = typer.Option(False, "--debug", "-d", help="Enable debug logging."), player: Optional[str] = typer.Option( - None, "--player", "-p", help="Target a specific MPRIS player using its DBus name or a portion thereof." + None, + "--player", + "-p", + help="Target a specific MPRIS player using its DBus name or a portion thereof.", ), ): global _player @@ -35,15 +42,15 @@ def main( _player = player -# ------------------------------------------------------------------ # fetch -# ------------------------------------------------------------------ @app.command() def fetch( method: Optional[str] = typer.Option( - None, "--method", help="Force a specific source (local, spotify, lrclib, lrclib-search, netease)." + None, + "--method", + help="Force a specific source (local, spotify, lrclib, lrclib-search, netease).", ), no_cache: bool = typer.Option( False, "--no-cache", help="Bypass the cache for this request." @@ -61,9 +68,7 @@ def fetch( logger.info(f"Track: {track.display_name()}") - result = manager.fetch_for_track( - track, force_method=method, bypass_cache=no_cache - ) + result = manager.fetch_for_track(track, force_method=method, bypass_cache=no_cache) if not result or not result.lyrics: logger.error("No lyrics found.") @@ -76,9 +81,7 @@ def fetch( print(result.lyrics) -# ------------------------------------------------------------------ # search -# ------------------------------------------------------------------ @app.command() @@ -87,8 +90,12 @@ def search( artist: Optional[str] = typer.Option(None, "--artist", "-a", help="Artist name."), album: Optional[str] = typer.Option(None, "--album", help="Album name."), trackid: Optional[str] = typer.Option(None, "--trackid", help="Spotify track ID."), - length: Optional[int] = typer.Option(None, "--length", "-l", help="Track duration in milliseconds."), - url: Optional[str] = typer.Option(None, "--url", help="Local file URL (file:///...)."), + length: Optional[int] = typer.Option( + None, "--length", "-l", help="Track duration in milliseconds." + ), + url: Optional[str] = typer.Option( + None, "--url", help="Local file URL (file:///...)." + ), method: Optional[str] = typer.Option( None, "--method", help="Force a specific source." ), @@ -111,9 +118,7 @@ def search( logger.info(f"Track: {track.display_name()}") - result = manager.fetch_for_track( - track, force_method=method, bypass_cache=no_cache - ) + result = manager.fetch_for_track(track, force_method=method, bypass_cache=no_cache) if not result or not result.lyrics: logger.error("No lyrics found.") @@ -126,15 +131,16 @@ def search( print(result.lyrics) -# ------------------------------------------------------------------ # export -# ------------------------------------------------------------------ @app.command() def export( output: Optional[str] = typer.Option( - None, "--output", "-o", help="Output file path (default: - .lrc)." + None, + "--output", + "-o", + help="Output file path (default: <Artist> - <Title>.lrc).", ), method: Optional[str] = typer.Option( None, "--method", help="Force a specific source." @@ -150,9 +156,7 @@ def export( logger.error("No active playing track found.") raise typer.Exit(1) - result = manager.fetch_for_track( - track, force_method=method, bypass_cache=no_cache - ) + result = manager.fetch_for_track(track, force_method=method, bypass_cache=no_cache) if not result or not result.lyrics: logger.error("No lyrics available to export.") raise typer.Exit(1) @@ -183,9 +187,7 @@ def export( raise typer.Exit(1) -# ------------------------------------------------------------------ # cache -# ------------------------------------------------------------------ @app.command() @@ -302,7 +304,9 @@ def _print_cache_row(row: dict, indent: str = "") -> None: if expires: remaining = expires - now if remaining > 0: - print(f"{indent} Expires : in {remaining // 3600}h {(remaining % 3600) // 60}m") + print( + f"{indent} Expires : in {remaining // 3600}h {(remaining % 3600) // 60}m" + ) else: print(f"{indent} Expires : EXPIRED") else: diff --git a/lrcfetch/config.py b/lrcfetch/config.py index a023f84..69fbec0 100644 --- a/lrcfetch/config.py +++ b/lrcfetch/config.py @@ -1,4 +1,8 @@ -"""Global configuration constants and logger setup.""" +""" +Author: Uyanide pywang0608@foxmail.com +Date: 2026-03-25 10:17:56 +Description: Global configuration constants and logger setup +""" import os import sys @@ -7,32 +11,32 @@ from platformdirs import user_cache_dir, user_config_dir from dotenv import load_dotenv from loguru import logger -# ─── Application ───────────────────────────────────────────────────── +# Application APP_NAME = "lrcfetch" APP_AUTHOR = "Uyanide" -# ─── Paths ─────────────────────────────────────────────────────────── +# Paths CACHE_DIR = user_cache_dir(APP_NAME, APP_AUTHOR) DB_PATH = os.path.join(CACHE_DIR, "cache.db") -# ─── .env loading (XDG config dir first, then project-local) ───────── +# .env loading _config_env = Path(user_config_dir(APP_NAME, APP_AUTHOR)) / ".env" -load_dotenv(_config_env) # ~/.config/lrcfetch/.env -load_dotenv() # .env in cwd (does NOT override existing vars) +load_dotenv(_config_env) # ~/.config/lrcfetch/.env +load_dotenv() # .env in cwd (does NOT override existing vars) -# ─── HTTP ──────────────────────────────────────────────────────────── +# HTTP HTTP_TIMEOUT = 10.0 -# ─── Cache TTLs (seconds) ─────────────────────────────────────────── -TTL_SYNCED = None # never expires -TTL_UNSYNCED = 86400 # 1 day -TTL_NOT_FOUND = 86400 * 3 # 3 days -TTL_NETWORK_ERROR = 3600 # 1 hour +# Cache TTLs (seconds) +TTL_SYNCED = None # never expires +TTL_UNSYNCED = 86400 # 1 day +TTL_NOT_FOUND = 86400 * 3 # 3 days +TTL_NETWORK_ERROR = 3600 # 1 hour -# ─── Search ────────────────────────────────────────────────────────── -DURATION_TOLERANCE_MS = 3000 # max duration mismatch for search matching +# Search +DURATION_TOLERANCE_MS = 3000 # max duration mismatch for search matching -# ─── Spotify ───────────────────────────────────────────────────────── +# Spotify related SPOTIFY_TOKEN_URL = "https://open.spotify.com/api/token" SPOTIFY_LYRICS_URL = "https://spclient.wg.spotify.com/color-lyrics/v2/track/" SPOTIFY_SERVER_TIME_URL = "https://open.spotify.com/api/server-time" @@ -43,24 +47,21 @@ SPOTIFY_SECRET_URL = ( SPOTIFY_SP_DC = os.environ.get("SPOTIFY_SP_DC", "") SPOTIFY_TOKEN_CACHE_FILE = os.path.join(CACHE_DIR, "spotify_token.json") -# ─── Netease ───────────────────────────────────────────────────────── +# Netease api NETEASE_SEARCH_URL = "https://music.163.com/api/cloudsearch/pc" NETEASE_LYRIC_URL = "https://interface3.music.163.com/api/song/lyric" -# ─── LRCLIB ────────────────────────────────────────────────────────── +# LRCLIB api LRCLIB_API_URL = "https://lrclib.net/api/get" LRCLIB_SEARCH_URL = "https://lrclib.net/api/search" -# ─── User-Agents ───────────────────────────────────────────────────── -UA_BROWSER = ( - "Mozilla/5.0 (X11; Linux x86_64; rv:148.0) " - "Gecko/20100101 Firefox/148.0" -) +# User-Agents +UA_BROWSER = "Mozilla/5.0 (X11; Linux x86_64; rv:148.0) Gecko/20100101 Firefox/148.0" UA_LRCFETCH = "LRCFetch (https://github.com/Uyanide/lrcfetch)" os.makedirs(CACHE_DIR, exist_ok=True) -# ─── Logger ────────────────────────────────────────────────────────── +# Logger _LOG_FORMAT = ( "<green>{time:YYYY-MM-DD HH:mm:ss}</green> | " "<level>{level: <8}</level> | " diff --git a/lrcfetch/core.py b/lrcfetch/core.py index ff6575e..303dfaa 100644 --- a/lrcfetch/core.py +++ b/lrcfetch/core.py @@ -1,5 +1,10 @@ -"""Core orchestrator — coordinates fetchers with cache-aware fallback. +""" +Author: Uyanide pywang0608@foxmail.com +Date: 2026-03-25 11:09:53 +Description: Core orchestrator — coordinates fetchers with cache-aware fallback +""" +""" Fetch pipeline: 1. Check cache for each source in the fallback sequence 2. For sources without a valid cache hit, call the fetcher @@ -9,16 +14,18 @@ Fetch pipeline: from typing import Optional from loguru import logger -from lrcfetch.models import TrackMeta, LyricResult, CacheStatus -from lrcfetch.config import TTL_SYNCED, TTL_UNSYNCED, TTL_NOT_FOUND, TTL_NETWORK_ERROR -from lrcfetch.lrc import LRC_LINE_RE, normalize_tags -from lrcfetch.cache import CacheEngine -from lrcfetch.fetchers.base import BaseFetcher -from lrcfetch.fetchers.local import LocalFetcher -from lrcfetch.fetchers.spotify import SpotifyFetcher -from lrcfetch.fetchers.lrclib import LrclibFetcher -from lrcfetch.fetchers.lrclib_search import LrclibSearchFetcher -from lrcfetch.fetchers.netease import NeteaseFetcher + +from .fetchers.netease import NeteaseFetcher +from .fetchers.lrclib_search import LrclibSearchFetcher +from .fetchers.lrclib import LrclibFetcher +from .fetchers.spotify import SpotifyFetcher +from .fetchers.local import LocalFetcher +from .fetchers.base import BaseFetcher +from .cache import CacheEngine +from .lrc import LRC_LINE_RE, normalize_tags +from .config import TTL_SYNCED, TTL_UNSYNCED, TTL_NOT_FOUND, TTL_NETWORK_ERROR +from .models import TrackMeta, LyricResult, CacheStatus + def _normalize_unsynced(lyrics: str) -> str: """Normalize unsynced lyrics so every line has a [00:00.00] tag. @@ -83,9 +90,7 @@ class LrcManager: sequence.append(self.fetchers["lrclib-search"]) sequence.append(self.fetchers["netease"]) - logger.debug( - f"Fallback sequence: {[f.source_name for f in sequence]}" - ) + logger.debug(f"Fallback sequence: {[f.source_name for f in sequence]}") return sequence def fetch_for_track( @@ -124,12 +129,19 @@ class LrcManager: logger.info(f"[{source}] cache hit: synced lyrics") return cached elif cached.status == CacheStatus.SUCCESS_UNSYNCED: - logger.debug(f"[{source}] cache hit: unsynced lyrics (continuing)") + logger.debug( + f"[{source}] cache hit: unsynced lyrics (continuing)" + ) if best_result is None: best_result = cached continue # Try next source for synced - elif cached.status in (CacheStatus.NOT_FOUND, CacheStatus.NETWORK_ERROR): - logger.debug(f"[{source}] cache hit: {cached.status.value}, skipping") + elif cached.status in ( + CacheStatus.NOT_FOUND, + CacheStatus.NETWORK_ERROR, + ): + logger.debug( + f"[{source}] cache hit: {cached.status.value}, skipping" + ) continue else: logger.debug(f"[{source}] cache bypassed") diff --git a/lrcfetch/fetchers/base.py b/lrcfetch/fetchers/base.py index 1dabc45..afa30ed 100644 --- a/lrcfetch/fetchers/base.py +++ b/lrcfetch/fetchers/base.py @@ -1,6 +1,13 @@ +""" +Author: Uyanide pywang0608@foxmail.com +Date: 2026-03-25 02:33:26 +Description: Base fetcher class and common interfaces +""" + from abc import ABC, abstractmethod from typing import Optional -from lrcfetch.models import TrackMeta, LyricResult + +from ..models import TrackMeta, LyricResult class BaseFetcher(ABC): diff --git a/lrcfetch/fetchers/local.py b/lrcfetch/fetchers/local.py index 31c34e3..ea25c4e 100644 --- a/lrcfetch/fetchers/local.py +++ b/lrcfetch/fetchers/local.py @@ -1,5 +1,10 @@ -"""Local fetcher — reads lyrics from .lrc sidecar files or embedded audio metadata. +""" +Author: Uyanide pywang0608@foxmail.com +Date: 2026-03-26 02:08:41 +Description: Local fetcher — reads lyrics from .lrc sidecar files or embedded audio metadata +""" +""" Priority: 1. Same-directory .lrc file (e.g. /path/to/track.lrc) 2. Embedded lyrics in audio metadata (FLAC, MP3 USLT/SYLT tags) @@ -9,12 +14,13 @@ import os from typing import Optional from urllib.parse import unquote from loguru import logger -from lrcfetch.models import TrackMeta, LyricResult, CacheStatus -from lrcfetch.fetchers.base import BaseFetcher -from lrcfetch.lrc import detect_sync_status from mutagen._file import File from mutagen.flac import FLAC +from .base import BaseFetcher +from ..models import TrackMeta, LyricResult +from ..lrc import detect_sync_status + class LocalFetcher(BaseFetcher): @property @@ -56,7 +62,9 @@ class LocalFetcher(BaseFetcher): if isinstance(audio, FLAC): # FLAC stores lyrics in vorbis comment tags - lyrics = (audio.get("lyrics") or audio.get("unsynclyrics") or [None])[0] + lyrics = ( + audio.get("lyrics") or audio.get("unsynclyrics") or [None] + )[0] elif hasattr(audio, "tags") and audio.tags: # MP3 / other: look for USLT or SYLT ID3 frames for key in audio.tags.keys(): diff --git a/lrcfetch/fetchers/lrclib.py b/lrcfetch/fetchers/lrclib.py index d49ba9a..2325dfa 100644 --- a/lrcfetch/fetchers/lrclib.py +++ b/lrcfetch/fetchers/lrclib.py @@ -1,16 +1,21 @@ -"""LRCLIB fetcher — queries lrclib.net for synced/plain lyrics. +""" +Author: Uyanide pywang0608@foxmail.com +Date: 2026-03-25 05:23:38 +Description: LRCLIB fetcher — queries lrclib.net for synced/plain lyrics +""" +""" Requires complete track metadata (artist, title, album, duration). """ -import httpx from typing import Optional +import httpx from loguru import logger from urllib.parse import urlencode -from lrcfetch.models import TrackMeta, LyricResult, CacheStatus -from lrcfetch.fetchers.base import BaseFetcher -from lrcfetch.config import ( +from .base import BaseFetcher +from ..models import TrackMeta, LyricResult, CacheStatus +from ..config import ( HTTP_TIMEOUT, TTL_UNSYNCED, TTL_NOT_FOUND, @@ -51,14 +56,18 @@ class LrclibFetcher(BaseFetcher): if resp.status_code != 200: logger.error(f"LRCLIB: API returned {resp.status_code}") - return LyricResult(status=CacheStatus.NETWORK_ERROR, ttl=TTL_NETWORK_ERROR) + return LyricResult( + status=CacheStatus.NETWORK_ERROR, ttl=TTL_NETWORK_ERROR + ) data = resp.json() # Validate response if not isinstance(data, dict): logger.error(f"LRCLIB: unexpected response type: {type(data).__name__}") - return LyricResult(status=CacheStatus.NETWORK_ERROR, ttl=TTL_NETWORK_ERROR) + return LyricResult( + status=CacheStatus.NETWORK_ERROR, ttl=TTL_NETWORK_ERROR + ) synced = data.get("syncedLyrics") unsynced = data.get("plainLyrics") diff --git a/lrcfetch/fetchers/lrclib_search.py b/lrcfetch/fetchers/lrclib_search.py index 88bec7d..37c6fca 100644 --- a/lrcfetch/fetchers/lrclib_search.py +++ b/lrcfetch/fetchers/lrclib_search.py @@ -1,5 +1,10 @@ -"""LRCLIB search fetcher — fuzzy search via lrclib.net /api/search. +""" +Author: Uyanide pywang0608@foxmail.com +Date: 2026-03-25 05:30:50 +Description: LRCLIB search fetcher — fuzzy search via lrclib.net /api/search +""" +""" Used when metadata is incomplete (no album or duration) but title is available. Selects the best match by duration when track length is known. """ @@ -9,9 +14,9 @@ from typing import Optional from loguru import logger from urllib.parse import urlencode -from lrcfetch.models import TrackMeta, LyricResult, CacheStatus -from lrcfetch.fetchers.base import BaseFetcher -from lrcfetch.config import ( +from .base import BaseFetcher +from ..models import TrackMeta, LyricResult, CacheStatus +from ..config import ( HTTP_TIMEOUT, TTL_UNSYNCED, TTL_NOT_FOUND, @@ -48,7 +53,9 @@ class LrclibSearchFetcher(BaseFetcher): if resp.status_code != 200: logger.error(f"LRCLIB-search: API returned {resp.status_code}") - return LyricResult(status=CacheStatus.NETWORK_ERROR, ttl=TTL_NETWORK_ERROR) + return LyricResult( + status=CacheStatus.NETWORK_ERROR, ttl=TTL_NETWORK_ERROR + ) data = resp.json() @@ -116,21 +123,38 @@ class LrclibSearchFetcher(BaseFetcher): if diff > DURATION_TOLERANCE_MS: continue # Prefer synced over unsynced at similar duration - has_synced = isinstance(item.get("syncedLyrics"), str) and item["syncedLyrics"].strip() - best_synced = best is not None and isinstance(best.get("syncedLyrics"), str) and best["syncedLyrics"].strip() - if diff < best_diff or (diff == best_diff and has_synced and not best_synced): + has_synced = ( + isinstance(item.get("syncedLyrics"), str) + and item["syncedLyrics"].strip() + ) + best_synced = ( + best is not None + and isinstance(best.get("syncedLyrics"), str) + and best["syncedLyrics"].strip() + ) + if diff < best_diff or ( + diff == best_diff and has_synced and not best_synced + ): best_diff = diff best = item if best is not None: - logger.debug(f"LRCLIB-search: selected id={best.get('id')} (diff={best_diff:.0f}ms)") + logger.debug( + f"LRCLIB-search: selected id={best.get('id')} (diff={best_diff:.0f}ms)" + ) return best - logger.debug(f"LRCLIB-search: no candidate within {DURATION_TOLERANCE_MS}ms") + logger.debug( + f"LRCLIB-search: no candidate within {DURATION_TOLERANCE_MS}ms" + ) return None # No duration — pick first with synced lyrics, or just first for item in candidates: - if isinstance(item, dict) and isinstance(item.get("syncedLyrics"), str) and item["syncedLyrics"].strip(): + if ( + isinstance(item, dict) + and isinstance(item.get("syncedLyrics"), str) + and item["syncedLyrics"].strip() + ): return item return candidates[0] if isinstance(candidates[0], dict) else None diff --git a/lrcfetch/fetchers/netease.py b/lrcfetch/fetchers/netease.py index bae0760..b8628b2 100644 --- a/lrcfetch/fetchers/netease.py +++ b/lrcfetch/fetchers/netease.py @@ -1,5 +1,10 @@ -"""Netease Cloud Music fetcher. +""" +Author: Uyanide pywang0608@foxmail.com +Date: 2026-03-25 11:04:51 +Description: Netease Cloud Music fetcher +""" +""" Uses the public cloudsearch API for searching and the song/lyric API for retrieving lyrics. No authentication required. @@ -7,13 +12,14 @@ Search results are filtered by duration when the track has a known length to avoid returning lyrics for the wrong version of a song. """ -import httpx from typing import Optional +import httpx from loguru import logger -from lrcfetch.models import TrackMeta, LyricResult, CacheStatus -from lrcfetch.fetchers.base import BaseFetcher -from lrcfetch.lrc import is_synced -from lrcfetch.config import ( + +from .base import BaseFetcher +from ..models import TrackMeta, LyricResult, CacheStatus +from ..lrc import is_synced +from ..config import ( HTTP_TIMEOUT, TTL_NOT_FOUND, TTL_NETWORK_ERROR, @@ -58,12 +64,14 @@ class NeteaseFetcher(BaseFetcher): # Validate response if not isinstance(result, dict): - logger.error(f"Netease: search returned non-dict: {type(result).__name__}") + logger.error( + f"Netease: search returned non-dict: {type(result).__name__}" + ) return None result_body = result.get("result") if not isinstance(result_body, dict): - logger.debug(f"Netease: search 'result' field missing or invalid") + logger.debug("Netease: search 'result' field missing or invalid") return None songs = result_body.get("songs") @@ -86,7 +94,9 @@ class NeteaseFetcher(BaseFetcher): name = song.get("name", "?") duration = song.get("dt") # milliseconds if not isinstance(duration, int): - logger.debug(f" candidate {sid} '{name}': no duration, skipped") + logger.debug( + f" candidate {sid} '{name}': no duration, skipped" + ) continue diff = abs(duration - track_ms) logger.debug( @@ -98,9 +108,7 @@ class NeteaseFetcher(BaseFetcher): best_id = sid if best_id is not None and best_diff <= DURATION_TOLERANCE_MS: - logger.debug( - f"Netease: selected id={best_id} (diff={best_diff}ms)" - ) + logger.debug(f"Netease: selected id={best_id} (diff={best_diff}ms)") return best_id logger.debug( @@ -150,12 +158,18 @@ class NeteaseFetcher(BaseFetcher): # Validate response if not isinstance(data, dict): - logger.error(f"Netease: lyric response is not dict: {type(data).__name__}") - return LyricResult(status=CacheStatus.NETWORK_ERROR, ttl=TTL_NETWORK_ERROR) + logger.error( + f"Netease: lyric response is not dict: {type(data).__name__}" + ) + return LyricResult( + status=CacheStatus.NETWORK_ERROR, ttl=TTL_NETWORK_ERROR + ) lrc_obj = data.get("lrc") if not isinstance(lrc_obj, dict): - logger.debug(f"Netease: no 'lrc' object in response for song_id={song_id}") + logger.debug( + f"Netease: no 'lrc' object in response for song_id={song_id}" + ) return LyricResult(status=CacheStatus.NOT_FOUND, ttl=TTL_NOT_FOUND) lrc: str = lrc_obj.get("lyric", "") @@ -165,7 +179,9 @@ class NeteaseFetcher(BaseFetcher): # Determine sync status synced = is_synced(lrc) - status = CacheStatus.SUCCESS_SYNCED if synced else CacheStatus.SUCCESS_UNSYNCED + status = ( + CacheStatus.SUCCESS_SYNCED if synced else CacheStatus.SUCCESS_UNSYNCED + ) logger.info( f"Netease: got {status.value} lyrics for song_id={song_id} " f"({len(lrc.splitlines())} lines)" diff --git a/lrcfetch/fetchers/spotify.py b/lrcfetch/fetchers/spotify.py index 7536772..935341e 100644 --- a/lrcfetch/fetchers/spotify.py +++ b/lrcfetch/fetchers/spotify.py @@ -1,8 +1,13 @@ -"""Spotify fetcher — obtains synced lyrics via Spotify's internal color-lyrics API. +""" +Author: Uyanide pywang0608@foxmail.com +Date: 2026-03-25 10:43:21 +Description: Spotify fetcher — obtains synced lyrics via Spotify's internal color-lyrics API. +""" -Authentication flow (mirrors spotify-lyrics Go implementation): +""" +Authentication flow: 1. Fetch server time from Spotify - 2. Fetch TOTP secret from xyloflake/spot-secrets-go + 2. Fetch TOTP secret 3. Generate a TOTP code and exchange it (with SP_DC cookie) for an access token 4. Request lyrics using the access token @@ -12,8 +17,8 @@ calls within the same session. Requires SPOTIFY_SP_DC environment variable to be set. """ -import json import httpx +import json import time import struct import hmac @@ -21,9 +26,9 @@ import hashlib from typing import Optional, Tuple from loguru import logger -from lrcfetch.models import TrackMeta, LyricResult, CacheStatus -from lrcfetch.fetchers.base import BaseFetcher -from lrcfetch.config import ( +from .base import BaseFetcher +from ..models import TrackMeta, LyricResult, CacheStatus +from ..config import ( HTTP_TIMEOUT, TTL_NOT_FOUND, TTL_NETWORK_ERROR, @@ -83,7 +88,8 @@ class SpotifyFetcher(BaseFetcher): if not isinstance(data, list) or len(data) == 0: logger.error( - f"Spotify: unexpected secrets response (type={type(data).__name__}, len={len(data) if isinstance(data, list) else '?'})") + f"Spotify: unexpected secrets response (type={type(data).__name__}, len={len(data) if isinstance(data, list) else '?'})" + ) return None last = data[-1] @@ -210,16 +216,15 @@ class SpotifyFetcher(BaseFetcher): try: res = client.get(SPOTIFY_TOKEN_URL, params=params, timeout=HTTP_TIMEOUT) if res.status_code != 200: - logger.error( - f"Spotify: token request returned {res.status_code}" - ) + logger.error(f"Spotify: token request returned {res.status_code}") return None body = res.json() if not isinstance(body, dict) or "accessToken" not in body: logger.error( - f"Spotify: unexpected token response keys: {list(body.keys()) if isinstance(body, dict) else type(body).__name__}") + f"Spotify: unexpected token response keys: {list(body.keys()) if isinstance(body, dict) else type(body).__name__}" + ) return None token = body["accessToken"] @@ -294,9 +299,7 @@ class SpotifyFetcher(BaseFetcher): if res.status_code == 404: logger.debug(f"Spotify: 404 for trackid={track.trackid}") - return LyricResult( - status=CacheStatus.NOT_FOUND, ttl=TTL_NOT_FOUND - ) + return LyricResult(status=CacheStatus.NOT_FOUND, ttl=TTL_NOT_FOUND) if res.status_code != 200: logger.error(f"Spotify: lyrics API returned {res.status_code}") @@ -308,7 +311,7 @@ class SpotifyFetcher(BaseFetcher): # Validate response structure if not isinstance(data, dict) or "lyrics" not in data: - logger.error(f"Spotify: unexpected lyrics response structure") + logger.error("Spotify: unexpected lyrics response structure") return LyricResult( status=CacheStatus.NETWORK_ERROR, ttl=TTL_NETWORK_ERROR ) @@ -343,11 +346,13 @@ class SpotifyFetcher(BaseFetcher): lrc_lines.append(f"[00:00.00]{words}") content = "\n".join(lrc_lines) - status = CacheStatus.SUCCESS_SYNCED if is_synced else CacheStatus.SUCCESS_UNSYNCED - - logger.info( - f"Spotify: got {status.value} lyrics ({len(lrc_lines)} lines)" + status = ( + CacheStatus.SUCCESS_SYNCED + if is_synced + else CacheStatus.SUCCESS_UNSYNCED ) + + logger.info(f"Spotify: got {status.value} lyrics ({len(lrc_lines)} lines)") return LyricResult(status=status, lyrics=content, source=self.source_name) except Exception as e: diff --git a/lrcfetch/lrc.py b/lrcfetch/lrc.py index 5a9eff2..11ece6f 100644 --- a/lrcfetch/lrc.py +++ b/lrcfetch/lrc.py @@ -1,10 +1,12 @@ -"""Shared LRC time-tag utilities. - -Handles detection, normalization, and sync-status checks for LRC lyrics. +""" +Author: Uyanide pywang0608@foxmail.com +Date: 2026-03-25 21:54:01 +Description: Shared LRC time-tag utilities """ import re -from lrcfetch.models import CacheStatus + +from .models import CacheStatus # Standard format: [mm:ss.cc] or [mm:ss.ccc] _STANDARD_TAG_RE = re.compile(r"\[\d{2}:\d{2}\.\d{2,3}\]") @@ -88,4 +90,6 @@ def is_synced(text: str) -> bool: def detect_sync_status(text: str) -> CacheStatus: """Determine whether lyrics contain meaningful LRC time tags.""" - return CacheStatus.SUCCESS_SYNCED if is_synced(text) else CacheStatus.SUCCESS_UNSYNCED + return ( + CacheStatus.SUCCESS_SYNCED if is_synced(text) else CacheStatus.SUCCESS_UNSYNCED + ) diff --git a/lrcfetch/models.py b/lrcfetch/models.py index e46d288..8666f2d 100644 --- a/lrcfetch/models.py +++ b/lrcfetch/models.py @@ -1,4 +1,8 @@ -"""Data models for lrcfetch.""" +""" +Author: Uyanide pywang0608@foxmail.com +Date: 2026-03-25 04:09:36 +Description: Data models +""" from pydantic import BaseModel, ConfigDict from enum import Enum @@ -7,6 +11,7 @@ from typing import Optional class CacheStatus(str, Enum): """Status of a cached lyric entry.""" + SUCCESS_SYNCED = "SUCCESS_SYNCED" SUCCESS_UNSYNCED = "SUCCESS_UNSYNCED" NOT_FOUND = "NOT_FOUND" @@ -15,14 +20,15 @@ class CacheStatus(str, Enum): class TrackMeta(BaseModel): """Metadata describing a track obtained from MPRIS or manual input.""" + model_config = ConfigDict(strict=True) - trackid: Optional[str] = None # Spotify track ID (without "spotify:track:" prefix) - length: Optional[int] = None # Duration in milliseconds + trackid: Optional[str] = None # Spotify track ID (without "spotify:track:" prefix) + length: Optional[int] = None # Duration in milliseconds album: Optional[str] = None artist: Optional[str] = None title: Optional[str] = None - url: Optional[str] = None # Playback URL (file:// for local files) + url: Optional[str] = None # Playback URL (file:// for local files) @property def is_local(self) -> bool: @@ -46,9 +52,10 @@ class TrackMeta(BaseModel): class LyricResult(BaseModel): """Result of a lyric fetch attempt, also used as cache record.""" + model_config = ConfigDict(strict=True) status: CacheStatus lyrics: Optional[str] = None - source: Optional[str] = None # Which fetcher produced this result - ttl: Optional[int] = None # Hint for cache TTL (seconds) + source: Optional[str] = None # Which fetcher produced this result + ttl: Optional[int] = None # Hint for cache TTL (seconds) diff --git a/lrcfetch/mpris.py b/lrcfetch/mpris.py index 398a257..c85c4b1 100644 --- a/lrcfetch/mpris.py +++ b/lrcfetch/mpris.py @@ -1,3 +1,9 @@ +""" +Author: Uyanide pywang0608@foxmail.com +Date: 2026-03-25 04:44:15 +Description: MPRIS integration for fetching track metadata +""" + import asyncio from dbus_next.aio.message_bus import MessageBus from dbus_next.constants import BusType @@ -7,34 +13,40 @@ from loguru import logger from typing import Optional, List, Any import subprocess -async def _get_active_players(bus: MessageBus, specific_player: Optional[str] = None) -> List[str]: + +async def _get_active_players( + bus: MessageBus, specific_player: Optional[str] = None +) -> List[str]: try: reply = await bus.call( Message( destination="org.freedesktop.DBus", path="/org/freedesktop/DBus", interface="org.freedesktop.DBus", - member="ListNames" + member="ListNames", ) ) if not reply or not reply.body: return [] - + names = reply.body[0] players = [name for name in names if name.startswith("org.mpris.MediaPlayer2.")] - + if specific_player: players = [p for p in players if specific_player.lower() in p.lower()] else: # Sort so that spotify is preferred players.sort(key=lambda x: 0 if "spotify" in x.lower() else 1) - + return players except Exception as e: logger.error(f"Failed to list DBus names: {e}") return [] -async def _fetch_metadata_dbus(specific_player: Optional[str] = None) -> Optional[TrackMeta]: + +async def _fetch_metadata_dbus( + specific_player: Optional[str] = None, +) -> Optional[TrackMeta]: bus = None try: bus = await MessageBus(bus_type=BusType.SESSION).connect() @@ -45,28 +57,34 @@ async def _fetch_metadata_dbus(specific_player: Optional[str] = None) -> Optiona try: players = await _get_active_players(bus, specific_player) if not players: - logger.debug(f"No active MPRIS players found via DBus{' for ' + specific_player if specific_player else ''}.") + logger.debug( + f"No active MPRIS players found via DBus{' for ' + specific_player if specific_player else ''}." + ) return None player_name = players[0] logger.debug(f"Using player: {player_name}") introspection = await bus.introspect(player_name, "/org/mpris/MediaPlayer2") - proxy = bus.get_proxy_object(player_name, "/org/mpris/MediaPlayer2", introspection) - + proxy = bus.get_proxy_object( + player_name, "/org/mpris/MediaPlayer2", introspection + ) + props_iface = proxy.get_interface("org.freedesktop.DBus.Properties") if not props_iface: logger.error(f"Player {player_name} doesn't support Properties interface.") return None - + try: - metadata_var: Any = await getattr(props_iface, "call_get")("org.mpris.MediaPlayer2.Player", "Metadata") + metadata_var: Any = await getattr(props_iface, "call_get")( + "org.mpris.MediaPlayer2.Player", "Metadata" + ) if not metadata_var: logger.error("Empty metadata received.") return None - + metadata = metadata_var.value - + # Extract trackid — MPRIS returns either "spotify:track:ID" # or a DBus object path like "/com/spotify/track/ID" trackid = metadata.get("mpris:trackid", None) @@ -77,21 +95,25 @@ async def _fetch_metadata_dbus(specific_player: Optional[str] = None) -> Optiona trackid = trackid.removeprefix("spotify:track:") elif trackid.startswith("/com/spotify/track/"): trackid = trackid.removeprefix("/com/spotify/track/") - + # Extract length (usually microseconds) length = metadata.get("mpris:length", None) if length: length = length.value // 1000 if isinstance(length.value, int) else None - + album = metadata.get("xesam:album", None) album = album.value if album else None - + artist = metadata.get("xesam:artist", None) - artist = artist.value[0] if artist and isinstance(artist.value, list) and artist.value else None - + artist = ( + artist.value[0] + if artist and isinstance(artist.value, list) and artist.value + else None + ) + title = metadata.get("xesam:title", None) title = title.value if title else None - + url = metadata.get("xesam:url", None) url = url.value if url else None @@ -101,63 +123,71 @@ async def _fetch_metadata_dbus(specific_player: Optional[str] = None) -> Optiona album=album, artist=artist, title=title, - url=url + url=url, ) except Exception as e: logger.error(f"Failed to get properties from {player_name}: {e}") return None - + finally: if bus: bus.disconnect() -def _fetch_metadata_subprocess(specific_player: Optional[str] = None) -> Optional[TrackMeta]: + +def _fetch_metadata_subprocess( + specific_player: Optional[str] = None, +) -> Optional[TrackMeta]: """Fallback using playerctl if dbus-next fails or session bus is problematic.""" logger.debug("Attempting to use playerctl as fallback.") try: # Check if playerctl exists subprocess.run(["playerctl", "--version"], capture_output=True, check=True) - + base_cmd = ["playerctl"] if specific_player: base_cmd.extend(["-p", specific_player]) - + def _get_prop(prop: str) -> Optional[str]: - res = subprocess.run(base_cmd + ["metadata", prop], capture_output=True, text=True) + res = subprocess.run( + base_cmd + ["metadata", prop], capture_output=True, text=True + ) if res.returncode == 0 and res.stdout.strip(): return res.stdout.strip() return None - + trackid = _get_prop("mpris:trackid") if trackid: if trackid.startswith("spotify:track:"): trackid = trackid.removeprefix("spotify:track:") elif trackid.startswith("/com/spotify/track/"): trackid = trackid.removeprefix("/com/spotify/track/") - + length_str = _get_prop("mpris:length") - length = int(length_str) // 1000 if length_str and length_str.isdigit() else None - + length = ( + int(length_str) // 1000 if length_str and length_str.isdigit() else None + ) + album = _get_prop("xesam:album") artist = _get_prop("xesam:artist") title = _get_prop("xesam:title") url = _get_prop("xesam:url") - + if not any([trackid, length, album, artist, title, url]): return None - + return TrackMeta( trackid=trackid, length=length, album=album, artist=artist, title=title, - url=url + url=url, ) except Exception as e: logger.debug(f"playerctl fallback failed: {e}") return None + def get_current_track(player_name: Optional[str] = None) -> Optional[TrackMeta]: try: meta = asyncio.run(_fetch_metadata_dbus(player_name)) @@ -165,5 +195,5 @@ def get_current_track(player_name: Optional[str] = None) -> Optional[TrackMeta]: return meta except Exception as e: logger.error(f"DBus async loop failed: {e}") - + return _fetch_metadata_subprocess(player_name)