init
This commit is contained in:
@@ -0,0 +1 @@
|
||||
__version__ = "0.1.0"
|
||||
@@ -0,0 +1,4 @@
|
||||
from lrcfetch.cli import run
|
||||
|
||||
if __name__ == "__main__":
|
||||
run()
|
||||
@@ -0,0 +1,275 @@
|
||||
"""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
|
||||
|
||||
|
||||
def _generate_key(track: TrackMeta, source: str) -> str:
|
||||
"""Generate a unique cache key from track metadata and source.
|
||||
|
||||
The key is scoped by source so that different fetchers can cache
|
||||
independently for the same track (e.g. Spotify synced vs Netease unsynced).
|
||||
"""
|
||||
# Spotify tracks always use their track ID as the primary identifier
|
||||
if track.trackid and source == "spotify":
|
||||
return f"spotify:{track.trackid}"
|
||||
|
||||
parts = []
|
||||
if track.artist:
|
||||
parts.append(track.artist)
|
||||
if track.title:
|
||||
parts.append(track.title)
|
||||
if track.album:
|
||||
parts.append(track.album)
|
||||
if track.length:
|
||||
parts.append(str(track.length))
|
||||
|
||||
# Fall back to URL for local files
|
||||
if not parts and track.url:
|
||||
return f"{source}:url:{track.url}"
|
||||
|
||||
if not parts:
|
||||
raise ValueError("Insufficient metadata to generate cache key")
|
||||
|
||||
raw = "|".join(parts)
|
||||
digest = hashlib.sha256(raw.encode()).hexdigest()
|
||||
return f"{source}:{digest}"
|
||||
|
||||
|
||||
class CacheEngine:
|
||||
def __init__(self, db_path: str = DB_PATH):
|
||||
self.db_path = db_path
|
||||
self._init_db()
|
||||
|
||||
def _init_db(self) -> None:
|
||||
"""Create or migrate the cache table."""
|
||||
with sqlite3.connect(self.db_path) as conn:
|
||||
conn.execute("""
|
||||
CREATE TABLE IF NOT EXISTS cache (
|
||||
key TEXT PRIMARY KEY,
|
||||
source TEXT NOT NULL,
|
||||
status TEXT NOT NULL,
|
||||
lyrics TEXT,
|
||||
created_at INTEGER NOT NULL,
|
||||
expires_at INTEGER,
|
||||
artist TEXT,
|
||||
title TEXT,
|
||||
album TEXT
|
||||
)
|
||||
""")
|
||||
conn.commit()
|
||||
|
||||
# ------------------------------------------------------------------
|
||||
# Read
|
||||
# ------------------------------------------------------------------
|
||||
|
||||
def get(self, track: TrackMeta, source: str) -> Optional[LyricResult]:
|
||||
"""Look up a cached result for *track* from *source*.
|
||||
|
||||
Returns None on cache miss or expiration.
|
||||
"""
|
||||
try:
|
||||
key = _generate_key(track, source)
|
||||
except ValueError:
|
||||
return None
|
||||
|
||||
with sqlite3.connect(self.db_path) as conn:
|
||||
row = conn.execute(
|
||||
"SELECT status, lyrics, source, expires_at FROM cache WHERE key = ?",
|
||||
(key,),
|
||||
).fetchone()
|
||||
|
||||
if not row:
|
||||
logger.debug(f"Cache miss: {source} / {track.display_name()}")
|
||||
return None
|
||||
|
||||
status_str, lyrics, src, expires_at = row
|
||||
|
||||
# Check TTL expiration
|
||||
if expires_at and expires_at < int(time.time()):
|
||||
logger.debug(f"Cache expired: {source} / {track.display_name()}")
|
||||
conn.execute("DELETE FROM cache WHERE key = ?", (key,))
|
||||
conn.commit()
|
||||
return None
|
||||
|
||||
remaining = expires_at - int(time.time()) if expires_at else None
|
||||
logger.debug(
|
||||
f"Cache hit: {source} / {track.display_name()} "
|
||||
f"[{status_str}, ttl={remaining}s]"
|
||||
)
|
||||
return LyricResult(
|
||||
status=CacheStatus(status_str),
|
||||
lyrics=lyrics,
|
||||
source=src,
|
||||
ttl=remaining,
|
||||
)
|
||||
|
||||
def get_best(self, track: TrackMeta, sources: list[str]) -> Optional[LyricResult]:
|
||||
"""Return the best cached result across *sources* (synced > unsynced).
|
||||
|
||||
Skips negative statuses (NOT_FOUND, NETWORK_ERROR) — those are only
|
||||
consulted per-source to avoid redundant fetches.
|
||||
"""
|
||||
best: Optional[LyricResult] = None
|
||||
for src in sources:
|
||||
cached = self.get(track, src)
|
||||
if not cached:
|
||||
continue
|
||||
if cached.status == CacheStatus.SUCCESS_SYNCED:
|
||||
return cached # Can't do better
|
||||
if cached.status == CacheStatus.SUCCESS_UNSYNCED and best is None:
|
||||
best = cached
|
||||
return best
|
||||
|
||||
# ------------------------------------------------------------------
|
||||
# Write
|
||||
# ------------------------------------------------------------------
|
||||
|
||||
def set(
|
||||
self,
|
||||
track: TrackMeta,
|
||||
source: str,
|
||||
result: LyricResult,
|
||||
ttl_seconds: Optional[int] = None,
|
||||
) -> None:
|
||||
"""Store a lyric result in the cache."""
|
||||
try:
|
||||
key = _generate_key(track, source)
|
||||
except ValueError:
|
||||
logger.warning("Cannot cache: insufficient track metadata.")
|
||||
return
|
||||
|
||||
now = int(time.time())
|
||||
expires_at = now + ttl_seconds if ttl_seconds else None
|
||||
|
||||
with sqlite3.connect(self.db_path) as conn:
|
||||
conn.execute(
|
||||
"""INSERT OR REPLACE INTO cache
|
||||
(key, source, status, lyrics, created_at, expires_at,
|
||||
artist, title, album)
|
||||
VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)""",
|
||||
(
|
||||
key,
|
||||
source,
|
||||
result.status.value,
|
||||
result.lyrics,
|
||||
now,
|
||||
expires_at,
|
||||
track.artist,
|
||||
track.title,
|
||||
track.album,
|
||||
),
|
||||
)
|
||||
conn.commit()
|
||||
logger.debug(
|
||||
f"Cached: {source} / {track.display_name()} "
|
||||
f"[{result.status.value}, ttl={ttl_seconds}s]"
|
||||
)
|
||||
|
||||
# ------------------------------------------------------------------
|
||||
# Delete
|
||||
# ------------------------------------------------------------------
|
||||
|
||||
def clear_all(self) -> None:
|
||||
"""Remove every entry from the cache."""
|
||||
with sqlite3.connect(self.db_path) as conn:
|
||||
conn.execute("DELETE FROM cache")
|
||||
conn.commit()
|
||||
logger.info("Cache cleared.")
|
||||
|
||||
def clear_track(self, track: TrackMeta) -> None:
|
||||
"""Remove all cached entries (every source) for a single track."""
|
||||
conditions, params = self._track_where(track)
|
||||
if not conditions:
|
||||
logger.info(f"No cache entries found for {track.display_name()}.")
|
||||
return
|
||||
where = " AND ".join(conditions)
|
||||
with sqlite3.connect(self.db_path) as conn:
|
||||
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()}.")
|
||||
else:
|
||||
logger.info(f"No cache entries found for {track.display_name()}.")
|
||||
|
||||
def prune(self) -> int:
|
||||
"""Remove all expired entries. Returns the number of rows deleted."""
|
||||
with sqlite3.connect(self.db_path) as conn:
|
||||
cur = conn.execute(
|
||||
"DELETE FROM cache WHERE expires_at IS NOT NULL AND expires_at < ?",
|
||||
(int(time.time()),),
|
||||
)
|
||||
conn.commit()
|
||||
count = cur.rowcount
|
||||
logger.info(f"Pruned {count} expired cache entries.")
|
||||
return count
|
||||
|
||||
@staticmethod
|
||||
def _track_where(track: TrackMeta) -> tuple[list[str], list[str]]:
|
||||
"""Build WHERE conditions to match a track across all sources."""
|
||||
conditions: list[str] = []
|
||||
params: list[str] = []
|
||||
if track.artist:
|
||||
conditions.append("artist = ?")
|
||||
params.append(track.artist)
|
||||
if track.title:
|
||||
conditions.append("title = ?")
|
||||
params.append(track.title)
|
||||
if track.album:
|
||||
conditions.append("album = ?")
|
||||
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)."""
|
||||
conditions, params = self._track_where(track)
|
||||
if not conditions:
|
||||
return []
|
||||
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()]
|
||||
|
||||
def query_all(self) -> list[dict]:
|
||||
"""Return every row in the cache table."""
|
||||
with sqlite3.connect(self.db_path) as conn:
|
||||
conn.row_factory = sqlite3.Row
|
||||
return [dict(r) for r in conn.execute("SELECT * FROM cache").fetchall()]
|
||||
|
||||
def stats(self) -> dict:
|
||||
"""Return aggregate cache statistics."""
|
||||
now = int(time.time())
|
||||
with sqlite3.connect(self.db_path) as conn:
|
||||
total = conn.execute("SELECT COUNT(*) FROM cache").fetchone()[0]
|
||||
expired = conn.execute(
|
||||
"SELECT COUNT(*) FROM cache WHERE expires_at IS NOT NULL AND expires_at < ?",
|
||||
(now,),
|
||||
).fetchone()[0]
|
||||
by_status = dict(
|
||||
conn.execute(
|
||||
"SELECT status, COUNT(*) FROM cache GROUP BY status"
|
||||
).fetchall()
|
||||
)
|
||||
by_source = dict(
|
||||
conn.execute(
|
||||
"SELECT source, COUNT(*) FROM cache GROUP BY source"
|
||||
).fetchall()
|
||||
)
|
||||
return {
|
||||
"total": total,
|
||||
"expired": expired,
|
||||
"active": total - expired,
|
||||
"by_status": by_status,
|
||||
"by_source": by_source,
|
||||
}
|
||||
+318
@@ -0,0 +1,318 @@
|
||||
"""CLI interface for lrcfetch."""
|
||||
|
||||
import typer
|
||||
import time
|
||||
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
|
||||
|
||||
app = typer.Typer(
|
||||
help="LRCFetch — Fetch lyrics for tracks.",
|
||||
add_completion=False,
|
||||
)
|
||||
|
||||
manager = LrcManager()
|
||||
|
||||
|
||||
@app.callback()
|
||||
def main(
|
||||
debug: bool = typer.Option(False, "--debug", "-d", help="Enable debug logging."),
|
||||
):
|
||||
if debug:
|
||||
enable_debug()
|
||||
|
||||
|
||||
# ------------------------------------------------------------------
|
||||
# fetch
|
||||
# ------------------------------------------------------------------
|
||||
|
||||
|
||||
@app.command()
|
||||
def fetch(
|
||||
method: Optional[str] = typer.Option(
|
||||
None, "--method", help="Force a specific source (local, spotify, lrclib, lrclib-search, netease)."
|
||||
),
|
||||
player: Optional[str] = typer.Option(
|
||||
None, "--player", "-p", help="Target a specific MPRIS player."
|
||||
),
|
||||
no_cache: bool = typer.Option(
|
||||
False, "--no-cache", help="Bypass the cache for this request."
|
||||
),
|
||||
only_synced: bool = typer.Option(
|
||||
False, "--only-synced", help="Only accept synced (timed) lyrics."
|
||||
),
|
||||
):
|
||||
"""Fetch and print lyrics for the currently playing track."""
|
||||
track = get_current_track(player)
|
||||
|
||||
if not track:
|
||||
logger.error("No active playing track found.")
|
||||
raise typer.Exit(1)
|
||||
|
||||
logger.info(f"Track: {track.display_name()}")
|
||||
|
||||
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.")
|
||||
raise typer.Exit(1)
|
||||
|
||||
if only_synced and result.status != CacheStatus.SUCCESS_SYNCED:
|
||||
logger.error("Only unsynced lyrics available (--only-synced requested).")
|
||||
raise typer.Exit(1)
|
||||
|
||||
print(result.lyrics)
|
||||
|
||||
|
||||
# ------------------------------------------------------------------
|
||||
# search
|
||||
# ------------------------------------------------------------------
|
||||
|
||||
|
||||
@app.command()
|
||||
def search(
|
||||
title: str = typer.Option(..., "--title", "-t", help="Track title."),
|
||||
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:///...)."),
|
||||
method: Optional[str] = typer.Option(
|
||||
None, "--method", help="Force a specific source."
|
||||
),
|
||||
no_cache: bool = typer.Option(
|
||||
False, "--no-cache", help="Bypass the cache for this request."
|
||||
),
|
||||
only_synced: bool = typer.Option(
|
||||
False, "--only-synced", help="Only accept synced (timed) lyrics."
|
||||
),
|
||||
):
|
||||
"""Search for lyrics by metadata (bypasses MPRIS)."""
|
||||
track = TrackMeta(
|
||||
title=title,
|
||||
artist=artist,
|
||||
album=album,
|
||||
trackid=trackid,
|
||||
length=length,
|
||||
url=url,
|
||||
)
|
||||
|
||||
logger.info(f"Track: {track.display_name()}")
|
||||
|
||||
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.")
|
||||
raise typer.Exit(1)
|
||||
|
||||
if only_synced and result.status != CacheStatus.SUCCESS_SYNCED:
|
||||
logger.error("Only unsynced lyrics available (--only-synced requested).")
|
||||
raise typer.Exit(1)
|
||||
|
||||
print(result.lyrics)
|
||||
|
||||
|
||||
# ------------------------------------------------------------------
|
||||
# export
|
||||
# ------------------------------------------------------------------
|
||||
|
||||
|
||||
@app.command()
|
||||
def export(
|
||||
output: Optional[str] = typer.Option(
|
||||
None, "--output", "-o", help="Output file path (default: <Artist> - <Title>.lrc)."
|
||||
),
|
||||
method: Optional[str] = typer.Option(
|
||||
None, "--method", help="Force a specific source."
|
||||
),
|
||||
player: Optional[str] = typer.Option(
|
||||
None, "--player", "-p", help="Target a specific MPRIS player."
|
||||
),
|
||||
no_cache: bool = typer.Option(False, "--no-cache", help="Bypass cache."),
|
||||
overwrite: bool = typer.Option(
|
||||
False, "--overwrite", "-f", help="Overwrite existing file."
|
||||
),
|
||||
):
|
||||
"""Export lyrics of the current track to a .lrc file."""
|
||||
track = get_current_track(player)
|
||||
if not track:
|
||||
logger.error("No active playing track found.")
|
||||
raise typer.Exit(1)
|
||||
|
||||
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)
|
||||
|
||||
# Build default output path
|
||||
if not output:
|
||||
filename = (
|
||||
f"{track.artist} - {track.title}.lrc"
|
||||
if track.artist and track.title
|
||||
else "lyrics.lrc"
|
||||
)
|
||||
# Sanitize filename
|
||||
filename = "".join(
|
||||
c for c in filename if c.isalpha() or c.isdigit() or c in " -_."
|
||||
).rstrip()
|
||||
output = os.path.join(os.getcwd(), filename)
|
||||
|
||||
if os.path.exists(output) and not overwrite:
|
||||
logger.error(f"File exists: {output} (use -f to overwrite)")
|
||||
raise typer.Exit(1)
|
||||
|
||||
try:
|
||||
with open(output, "w", encoding="utf-8") as f:
|
||||
f.write(result.lyrics)
|
||||
logger.info(f"Exported lyrics to {output}")
|
||||
except Exception as e:
|
||||
logger.error(f"Failed to write file: {e}")
|
||||
raise typer.Exit(1)
|
||||
|
||||
|
||||
# ------------------------------------------------------------------
|
||||
# cache
|
||||
# ------------------------------------------------------------------
|
||||
|
||||
|
||||
@app.command()
|
||||
def cache(
|
||||
clear: bool = typer.Option(False, "--clear", help="Clear the entire cache."),
|
||||
clear_current: bool = typer.Option(
|
||||
False, "--clear-current", help="Clear cache for the current track."
|
||||
),
|
||||
prune: bool = typer.Option(False, "--prune", help="Remove expired entries."),
|
||||
stats: bool = typer.Option(False, "--stats", help="Show cache statistics."),
|
||||
query: bool = typer.Option(
|
||||
False, "--query", "-q", help="Show detailed cache info for the current track."
|
||||
),
|
||||
query_all: bool = typer.Option(
|
||||
False, "--query-all", help="Dump all cache entries."
|
||||
),
|
||||
):
|
||||
"""Manage the local SQLite cache."""
|
||||
if clear:
|
||||
manager.cache.clear_all()
|
||||
return
|
||||
|
||||
if clear_current:
|
||||
track = get_current_track()
|
||||
if not track:
|
||||
logger.error("No active playing track found.")
|
||||
raise typer.Exit(1)
|
||||
manager.cache.clear_track(track)
|
||||
return
|
||||
|
||||
if prune:
|
||||
manager.cache.prune()
|
||||
return
|
||||
|
||||
if stats:
|
||||
s = manager.cache.stats()
|
||||
print("=== Cache Statistics ===")
|
||||
print(f"Total entries : {s['total']}")
|
||||
print(f"Active : {s['active']}")
|
||||
print(f"Expired : {s['expired']}")
|
||||
if s["by_status"]:
|
||||
print("\nBy status:")
|
||||
for status, count in s["by_status"].items():
|
||||
print(f" {status}: {count}")
|
||||
if s["by_source"]:
|
||||
print("\nBy source:")
|
||||
for source, count in s["by_source"].items():
|
||||
print(f" {source}: {count}")
|
||||
return
|
||||
|
||||
if query:
|
||||
track = get_current_track()
|
||||
if not track:
|
||||
logger.error("No active playing track found.")
|
||||
raise typer.Exit(1)
|
||||
_print_track_cache(track)
|
||||
return
|
||||
|
||||
if query_all:
|
||||
rows = manager.cache.query_all()
|
||||
if not rows:
|
||||
print("Cache is empty.")
|
||||
return
|
||||
for row in rows:
|
||||
_print_cache_row(row)
|
||||
print()
|
||||
return
|
||||
|
||||
logger.info(
|
||||
"No action specified. Try --stats, --query, --query-all, "
|
||||
"--prune, --clear, or --clear-current."
|
||||
)
|
||||
|
||||
|
||||
def _print_track_cache(track: TrackMeta) -> None:
|
||||
"""Print all cached entries for a given track."""
|
||||
print(f"Track: {track.display_name()}")
|
||||
if track.album:
|
||||
print(f"Album: {track.album}")
|
||||
if track.length:
|
||||
secs = track.length / 1000.0
|
||||
print(f"Duration: {int(secs // 60)}:{secs % 60:05.2f}")
|
||||
print()
|
||||
|
||||
rows = manager.cache.query_track(track)
|
||||
if not rows:
|
||||
print(" (no cache entries)")
|
||||
return
|
||||
|
||||
for row in rows:
|
||||
_print_cache_row(row, indent=" ")
|
||||
|
||||
|
||||
def _print_cache_row(row: dict, indent: str = "") -> None:
|
||||
"""Pretty-print a single cache row."""
|
||||
now = int(time.time())
|
||||
source = row.get("source", "?")
|
||||
status = row.get("status", "?")
|
||||
artist = row.get("artist", "")
|
||||
title = row.get("title", "")
|
||||
album = row.get("album", "")
|
||||
created = row.get("created_at", 0)
|
||||
expires = row.get("expires_at")
|
||||
lyrics = row.get("lyrics", "")
|
||||
|
||||
name = f"{artist} - {title}" if artist and title else row.get("key", "?")
|
||||
print(f"{indent}[{source}] {name}")
|
||||
if album:
|
||||
print(f"{indent} Album : {album}")
|
||||
print(f"{indent} Status : {status}")
|
||||
if created:
|
||||
age = now - created
|
||||
print(f"{indent} Cached : {age // 3600}h {(age % 3600) // 60}m ago")
|
||||
if expires:
|
||||
remaining = expires - now
|
||||
if remaining > 0:
|
||||
print(f"{indent} Expires : in {remaining // 3600}h {(remaining % 3600) // 60}m")
|
||||
else:
|
||||
print(f"{indent} Expires : EXPIRED")
|
||||
else:
|
||||
print(f"{indent} Expires : never")
|
||||
if lyrics:
|
||||
line_count = len(lyrics.splitlines())
|
||||
print(f"{indent} Lyrics : {line_count} lines")
|
||||
|
||||
|
||||
def run():
|
||||
app()
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
run()
|
||||
@@ -0,0 +1,77 @@
|
||||
"""Global configuration constants and logger setup."""
|
||||
|
||||
import os
|
||||
import sys
|
||||
from pathlib import Path
|
||||
from platformdirs import user_cache_dir, user_config_dir
|
||||
from dotenv import load_dotenv
|
||||
from loguru import logger
|
||||
|
||||
# ─── Application ─────────────────────────────────────────────────────
|
||||
APP_NAME = "lrcfetch"
|
||||
APP_AUTHOR = "Uyanide"
|
||||
|
||||
# ─── .env loading (XDG config dir first, then project-local) ─────────
|
||||
_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)
|
||||
|
||||
# ─── HTTP ────────────────────────────────────────────────────────────
|
||||
HTTP_TIMEOUT = 10.0
|
||||
|
||||
# ─── Cache TTLs (seconds) ───────────────────────────────────────────
|
||||
TTL_SYNCED = 86400 * 30 # 30 days
|
||||
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
|
||||
|
||||
# ─── Spotify ─────────────────────────────────────────────────────────
|
||||
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"
|
||||
SPOTIFY_SECRET_URL = (
|
||||
"https://raw.githubusercontent.com/xyloflake/spot-secrets-go"
|
||||
"/refs/heads/main/secrets/secrets.json"
|
||||
)
|
||||
SPOTIFY_SP_DC = os.environ.get("SPOTIFY_SP_DC", "")
|
||||
|
||||
# ─── Netease ─────────────────────────────────────────────────────────
|
||||
NETEASE_SEARCH_URL = "https://music.163.com/api/cloudsearch/pc"
|
||||
NETEASE_LYRIC_URL = "https://interface3.music.163.com/api/song/lyric"
|
||||
|
||||
# ─── LRCLIB ──────────────────────────────────────────────────────────
|
||||
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"
|
||||
)
|
||||
UA_LRCFETCH = "LRCFetch (https://github.com/Uyanide/lrcfetch)"
|
||||
|
||||
# ─── Paths ───────────────────────────────────────────────────────────
|
||||
CACHE_DIR = user_cache_dir(APP_NAME, APP_AUTHOR)
|
||||
DB_PATH = os.path.join(CACHE_DIR, "cache.db")
|
||||
|
||||
os.makedirs(CACHE_DIR, exist_ok=True)
|
||||
|
||||
# ─── Logger ──────────────────────────────────────────────────────────
|
||||
_LOG_FORMAT = (
|
||||
"<green>{time:YYYY-MM-DD HH:mm:ss}</green> | "
|
||||
"<level>{level: <8}</level> | "
|
||||
"<cyan>{name}</cyan>:<cyan>{function}</cyan>:<cyan>{line}</cyan> - "
|
||||
"<level>{message}</level>"
|
||||
)
|
||||
|
||||
logger.remove()
|
||||
logger.add(sys.stderr, format=_LOG_FORMAT, level="INFO")
|
||||
|
||||
|
||||
def enable_debug() -> None:
|
||||
"""Switch logger to DEBUG level."""
|
||||
logger.remove()
|
||||
logger.add(sys.stderr, format=_LOG_FORMAT, level="DEBUG")
|
||||
@@ -0,0 +1,187 @@
|
||||
"""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
|
||||
3. Cache every result (success, not-found, or error) per source
|
||||
4. Return the best result (synced > unsynced > None)
|
||||
"""
|
||||
|
||||
import re
|
||||
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.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
|
||||
|
||||
# Matches any LRC time tag at the start of a line: [mm:ss.cc] or [mm:ss.ccc]
|
||||
_LRC_LINE_RE = re.compile(r"^\[(\d{2}:\d{2}\.\d{2,3})\]", re.MULTILINE)
|
||||
|
||||
|
||||
def _normalize_unsynced(lyrics: str) -> str:
|
||||
"""Normalize unsynced lyrics so every line has a [00:00.00] tag.
|
||||
|
||||
- Lines that already have time tags: replace with [00:00.00]
|
||||
- Lines without time tags: prepend [00:00.00]
|
||||
- Blank lines are kept as-is
|
||||
"""
|
||||
out: list[str] = []
|
||||
for line in lyrics.splitlines():
|
||||
stripped = line.strip()
|
||||
if not stripped:
|
||||
out.append("")
|
||||
continue
|
||||
# Strip existing time tag(s) from the beginning
|
||||
cleaned = _LRC_LINE_RE.sub("", stripped)
|
||||
# Could have multiple tags like [00:12.34][00:56.78]text
|
||||
while _LRC_LINE_RE.match(cleaned):
|
||||
cleaned = _LRC_LINE_RE.sub("", cleaned)
|
||||
out.append(f"[00:00.00]{cleaned}")
|
||||
return "\n".join(out)
|
||||
|
||||
|
||||
# Maps CacheStatus to the default TTL used when storing results
|
||||
_STATUS_TTL: dict[CacheStatus, int] = {
|
||||
CacheStatus.SUCCESS_SYNCED: TTL_SYNCED,
|
||||
CacheStatus.SUCCESS_UNSYNCED: TTL_UNSYNCED,
|
||||
CacheStatus.NOT_FOUND: TTL_NOT_FOUND,
|
||||
CacheStatus.NETWORK_ERROR: TTL_NETWORK_ERROR,
|
||||
}
|
||||
|
||||
|
||||
class LrcManager:
|
||||
"""Main entry point for fetching lyrics with caching."""
|
||||
|
||||
def __init__(self) -> None:
|
||||
self.cache = CacheEngine()
|
||||
self.fetchers: dict[str, BaseFetcher] = {
|
||||
"local": LocalFetcher(),
|
||||
"spotify": SpotifyFetcher(),
|
||||
"lrclib": LrclibFetcher(),
|
||||
"lrclib-search": LrclibSearchFetcher(),
|
||||
"netease": NeteaseFetcher(),
|
||||
}
|
||||
|
||||
def _build_sequence(
|
||||
self, track: TrackMeta, force_method: Optional[str] = None
|
||||
) -> list[BaseFetcher]:
|
||||
"""Determine the ordered list of fetchers to try."""
|
||||
if force_method:
|
||||
if force_method not in self.fetchers:
|
||||
logger.error(f"Unknown method: {force_method}")
|
||||
return []
|
||||
return [self.fetchers[force_method]]
|
||||
|
||||
sequence: list[BaseFetcher] = []
|
||||
if track.is_local:
|
||||
sequence.append(self.fetchers["local"])
|
||||
if track.trackid:
|
||||
sequence.append(self.fetchers["spotify"])
|
||||
if track.is_complete:
|
||||
sequence.append(self.fetchers["lrclib"])
|
||||
if track.title:
|
||||
sequence.append(self.fetchers["lrclib-search"])
|
||||
sequence.append(self.fetchers["netease"])
|
||||
|
||||
logger.debug(
|
||||
f"Fallback sequence: {[f.source_name for f in sequence]}"
|
||||
)
|
||||
return sequence
|
||||
|
||||
def fetch_for_track(
|
||||
self,
|
||||
track: TrackMeta,
|
||||
force_method: Optional[str] = None,
|
||||
bypass_cache: bool = False,
|
||||
) -> Optional[LyricResult]:
|
||||
"""Fetch lyrics for *track* using the fallback pipeline.
|
||||
|
||||
Each source is checked against the cache independently:
|
||||
- Cache hit with synced lyrics → return immediately
|
||||
- Cache hit with negative status (NOT_FOUND / NETWORK_ERROR) → skip source
|
||||
- Cache miss or unsynced → call fetcher, then cache the result
|
||||
|
||||
After all sources are tried, returns the best result found
|
||||
(synced > unsynced > None).
|
||||
"""
|
||||
logger.info(f"Fetching lyrics for: {track.display_name()}")
|
||||
|
||||
sequence = self._build_sequence(track, force_method)
|
||||
if not sequence:
|
||||
return None
|
||||
|
||||
# Best result seen so far (synced wins over unsynced)
|
||||
best_result: Optional[LyricResult] = None
|
||||
|
||||
for fetcher in sequence:
|
||||
source = fetcher.source_name
|
||||
|
||||
# Cache check
|
||||
if not bypass_cache:
|
||||
cached = self.cache.get(track, source)
|
||||
if cached:
|
||||
if cached.status == CacheStatus.SUCCESS_SYNCED:
|
||||
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)")
|
||||
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")
|
||||
continue
|
||||
else:
|
||||
logger.debug(f"[{source}] cache bypassed")
|
||||
|
||||
# Fetch
|
||||
logger.debug(f"[{source}] calling fetcher...")
|
||||
result = fetcher.fetch(track)
|
||||
|
||||
if not result:
|
||||
logger.debug(f"[{source}] returned None (no result)")
|
||||
continue
|
||||
|
||||
# Cache the result
|
||||
ttl = result.ttl or _STATUS_TTL.get(result.status, TTL_NOT_FOUND)
|
||||
self.cache.set(track, source, result, ttl_seconds=ttl)
|
||||
|
||||
# Evaluate result
|
||||
if result.status == CacheStatus.SUCCESS_SYNCED:
|
||||
logger.info(f"[{source}] got synced lyrics")
|
||||
return result
|
||||
|
||||
if result.status == CacheStatus.SUCCESS_UNSYNCED:
|
||||
logger.debug(f"[{source}] got unsynced lyrics (continuing)")
|
||||
if best_result is None:
|
||||
best_result = result
|
||||
|
||||
# NOT_FOUND / NETWORK_ERROR: already cached, try next
|
||||
|
||||
# Return best available
|
||||
if best_result:
|
||||
# Normalize unsynced lyrics: set all timestamps to [00:00.00]
|
||||
if (
|
||||
best_result.status == CacheStatus.SUCCESS_UNSYNCED
|
||||
and best_result.lyrics
|
||||
):
|
||||
best_result = LyricResult(
|
||||
status=best_result.status,
|
||||
lyrics=_normalize_unsynced(best_result.lyrics),
|
||||
source=best_result.source,
|
||||
ttl=best_result.ttl,
|
||||
)
|
||||
logger.info(
|
||||
f"Returning unsynced lyrics from {best_result.source} "
|
||||
f"(no synced source found)"
|
||||
)
|
||||
else:
|
||||
logger.info(f"No lyrics found for {track.display_name()}")
|
||||
|
||||
return best_result
|
||||
@@ -0,0 +1,16 @@
|
||||
from abc import ABC, abstractmethod
|
||||
from typing import Optional
|
||||
from lrcfetch.models import TrackMeta, LyricResult
|
||||
|
||||
|
||||
class BaseFetcher(ABC):
|
||||
@property
|
||||
@abstractmethod
|
||||
def source_name(self) -> str:
|
||||
"""Name of the fetcher source."""
|
||||
pass
|
||||
|
||||
@abstractmethod
|
||||
def fetch(self, track: TrackMeta) -> Optional[LyricResult]:
|
||||
"""Fetch lyrics for the given track. Returns None if unable to fetch."""
|
||||
pass
|
||||
@@ -0,0 +1,99 @@
|
||||
"""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)
|
||||
"""
|
||||
|
||||
import re
|
||||
import os
|
||||
from typing import Optional
|
||||
from loguru import logger
|
||||
from lrcfetch.models import TrackMeta, LyricResult, CacheStatus
|
||||
from lrcfetch.fetchers.base import BaseFetcher
|
||||
from mutagen._file import File
|
||||
from mutagen.flac import FLAC
|
||||
|
||||
# Matches LRC time tags like [00:12.34] or [01:23.456]
|
||||
_LRC_TIME_TAG_RE = re.compile(r"\[\d{2}:\d{2}\.\d{2,3}\]")
|
||||
# Matches time tags that are all zeros
|
||||
_ZERO_TIME_TAG_RE = re.compile(r"^\[00:00\.0{2,3}\]$")
|
||||
|
||||
|
||||
def _detect_sync_status(text: str) -> CacheStatus:
|
||||
"""Determine whether lyrics text contains meaningful LRC time tags.
|
||||
|
||||
Returns UNSYNCED if no tags exist or all tags are [00:00.00].
|
||||
"""
|
||||
tags = _LRC_TIME_TAG_RE.findall(text)
|
||||
if not tags:
|
||||
return CacheStatus.SUCCESS_UNSYNCED
|
||||
for tag in tags:
|
||||
if not _ZERO_TIME_TAG_RE.match(tag):
|
||||
return CacheStatus.SUCCESS_SYNCED
|
||||
return CacheStatus.SUCCESS_UNSYNCED
|
||||
|
||||
|
||||
class LocalFetcher(BaseFetcher):
|
||||
@property
|
||||
def source_name(self) -> str:
|
||||
return "local"
|
||||
|
||||
def fetch(self, track: TrackMeta) -> Optional[LyricResult]:
|
||||
"""Attempt to read lyrics from local filesystem."""
|
||||
if not track.is_local or not track.url:
|
||||
return None
|
||||
|
||||
file_path = track.url.replace("file://", "", 1)
|
||||
if not os.path.exists(file_path):
|
||||
logger.debug(f"Local: file does not exist: {file_path}")
|
||||
return None
|
||||
|
||||
logger.info(f"Local: checking for lyrics near {file_path}")
|
||||
|
||||
# Sidecar .lrc file
|
||||
lrc_path = os.path.splitext(file_path)[0] + ".lrc"
|
||||
if os.path.exists(lrc_path):
|
||||
try:
|
||||
with open(lrc_path, "r", encoding="utf-8") as f:
|
||||
content = f.read().strip()
|
||||
if content:
|
||||
status = _detect_sync_status(content)
|
||||
logger.info(f"Local: found .lrc sidecar ({status.value})")
|
||||
return LyricResult(
|
||||
status=status, lyrics=content, source=self.source_name
|
||||
)
|
||||
except Exception as e:
|
||||
logger.error(f"Local: error reading {lrc_path}: {e}")
|
||||
|
||||
# Embedded metadata
|
||||
try:
|
||||
audio = File(file_path)
|
||||
if audio is not None:
|
||||
lyrics = None
|
||||
|
||||
if isinstance(audio, FLAC):
|
||||
# FLAC stores lyrics in vorbis comment tags
|
||||
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():
|
||||
if key.startswith("USLT") or key.startswith("SYLT"):
|
||||
lyrics = str(audio.tags[key])
|
||||
break
|
||||
|
||||
if lyrics:
|
||||
status = _detect_sync_status(lyrics)
|
||||
logger.info(f"Local: found embedded lyrics ({status.value})")
|
||||
return LyricResult(
|
||||
status=status,
|
||||
lyrics=lyrics.strip(),
|
||||
source=f"{self.source_name} (embedded)",
|
||||
)
|
||||
else:
|
||||
logger.debug("Local: no embedded lyrics found")
|
||||
except Exception as e:
|
||||
logger.error(f"Local: error reading metadata for {file_path}: {e}")
|
||||
|
||||
logger.debug(f"Local: no lyrics found for {file_path}")
|
||||
return None
|
||||
@@ -0,0 +1,94 @@
|
||||
"""LRCLIB fetcher — queries lrclib.net for synced/plain lyrics.
|
||||
|
||||
Requires complete track metadata (artist, title, album, duration).
|
||||
"""
|
||||
|
||||
import httpx
|
||||
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 (
|
||||
HTTP_TIMEOUT,
|
||||
TTL_UNSYNCED,
|
||||
TTL_NOT_FOUND,
|
||||
TTL_NETWORK_ERROR,
|
||||
LRCLIB_API_URL,
|
||||
UA_LRCFETCH,
|
||||
)
|
||||
|
||||
|
||||
class LrclibFetcher(BaseFetcher):
|
||||
@property
|
||||
def source_name(self) -> str:
|
||||
return "lrclib"
|
||||
|
||||
def fetch(self, track: TrackMeta) -> Optional[LyricResult]:
|
||||
"""Fetch lyrics from LRCLIB. Requires complete metadata."""
|
||||
if not track.is_complete:
|
||||
logger.debug("LRCLIB: skipped — incomplete metadata")
|
||||
return None
|
||||
|
||||
params = {
|
||||
"track_name": track.title,
|
||||
"artist_name": track.artist,
|
||||
"album_name": track.album,
|
||||
"duration": track.length / 1000.0 if track.length else 0,
|
||||
}
|
||||
|
||||
url = f"{LRCLIB_API_URL}?{urlencode(params)}"
|
||||
logger.info(f"LRCLIB: fetching lyrics for {track.display_name()}")
|
||||
|
||||
try:
|
||||
with httpx.Client(timeout=HTTP_TIMEOUT) as client:
|
||||
resp = client.get(url, headers={"User-Agent": UA_LRCFETCH})
|
||||
|
||||
if resp.status_code == 404:
|
||||
logger.debug(f"LRCLIB: not found for {track.display_name()}")
|
||||
return LyricResult(status=CacheStatus.NOT_FOUND, ttl=TTL_NOT_FOUND)
|
||||
|
||||
if resp.status_code != 200:
|
||||
logger.error(f"LRCLIB: API returned {resp.status_code}")
|
||||
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)
|
||||
|
||||
synced = data.get("syncedLyrics")
|
||||
unsynced = data.get("plainLyrics")
|
||||
|
||||
if isinstance(synced, str) and synced.strip():
|
||||
logger.info(
|
||||
f"LRCLIB: got synced lyrics ({len(synced.splitlines())} lines)"
|
||||
)
|
||||
return LyricResult(
|
||||
status=CacheStatus.SUCCESS_SYNCED,
|
||||
lyrics=synced.strip(),
|
||||
source=self.source_name,
|
||||
)
|
||||
elif isinstance(unsynced, str) and unsynced.strip():
|
||||
logger.info(
|
||||
f"LRCLIB: got unsynced lyrics ({len(unsynced.splitlines())} lines)"
|
||||
)
|
||||
return LyricResult(
|
||||
status=CacheStatus.SUCCESS_UNSYNCED,
|
||||
lyrics=unsynced.strip(),
|
||||
source=self.source_name,
|
||||
ttl=TTL_UNSYNCED,
|
||||
)
|
||||
else:
|
||||
logger.debug(f"LRCLIB: empty response for {track.display_name()}")
|
||||
return LyricResult(status=CacheStatus.NOT_FOUND, ttl=TTL_NOT_FOUND)
|
||||
|
||||
except httpx.HTTPError as e:
|
||||
logger.error(f"LRCLIB: HTTP error: {e}")
|
||||
return LyricResult(status=CacheStatus.NETWORK_ERROR, ttl=TTL_NETWORK_ERROR)
|
||||
except Exception as e:
|
||||
logger.error(f"LRCLIB: unexpected error: {e}")
|
||||
return None
|
||||
@@ -0,0 +1,136 @@
|
||||
"""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.
|
||||
"""
|
||||
|
||||
import httpx
|
||||
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 (
|
||||
HTTP_TIMEOUT,
|
||||
TTL_UNSYNCED,
|
||||
TTL_NOT_FOUND,
|
||||
TTL_NETWORK_ERROR,
|
||||
DURATION_TOLERANCE_MS,
|
||||
LRCLIB_SEARCH_URL,
|
||||
UA_LRCFETCH,
|
||||
)
|
||||
|
||||
|
||||
class LrclibSearchFetcher(BaseFetcher):
|
||||
@property
|
||||
def source_name(self) -> str:
|
||||
return "lrclib-search"
|
||||
|
||||
def fetch(self, track: TrackMeta) -> Optional[LyricResult]:
|
||||
"""Search LRCLIB for lyrics. Requires at least a title."""
|
||||
if not track.title:
|
||||
logger.debug("LRCLIB-search: skipped — no title")
|
||||
return None
|
||||
|
||||
params: dict[str, str] = {"track_name": track.title}
|
||||
if track.artist:
|
||||
params["artist_name"] = track.artist
|
||||
if track.album:
|
||||
params["album_name"] = track.album
|
||||
|
||||
url = f"{LRCLIB_SEARCH_URL}?{urlencode(params)}"
|
||||
logger.info(f"LRCLIB-search: searching for {track.display_name()}")
|
||||
|
||||
try:
|
||||
with httpx.Client(timeout=HTTP_TIMEOUT) as client:
|
||||
resp = client.get(url, headers={"User-Agent": UA_LRCFETCH})
|
||||
|
||||
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)
|
||||
|
||||
data = resp.json()
|
||||
|
||||
if not isinstance(data, list) or len(data) == 0:
|
||||
logger.debug(f"LRCLIB-search: no results for {track.display_name()}")
|
||||
return LyricResult(status=CacheStatus.NOT_FOUND, ttl=TTL_NOT_FOUND)
|
||||
|
||||
logger.debug(f"LRCLIB-search: got {len(data)} candidates")
|
||||
|
||||
# Select best match by duration
|
||||
best = self._select_best(data, track)
|
||||
if best is None:
|
||||
logger.debug("LRCLIB-search: no valid candidate found")
|
||||
return LyricResult(status=CacheStatus.NOT_FOUND, ttl=TTL_NOT_FOUND)
|
||||
|
||||
# Extract lyrics
|
||||
synced = best.get("syncedLyrics")
|
||||
unsynced = best.get("plainLyrics")
|
||||
|
||||
if isinstance(synced, str) and synced.strip():
|
||||
logger.info(
|
||||
f"LRCLIB-search: got synced lyrics ({len(synced.splitlines())} lines)"
|
||||
)
|
||||
return LyricResult(
|
||||
status=CacheStatus.SUCCESS_SYNCED,
|
||||
lyrics=synced.strip(),
|
||||
source=self.source_name,
|
||||
)
|
||||
elif isinstance(unsynced, str) and unsynced.strip():
|
||||
logger.info(
|
||||
f"LRCLIB-search: got unsynced lyrics ({len(unsynced.splitlines())} lines)"
|
||||
)
|
||||
return LyricResult(
|
||||
status=CacheStatus.SUCCESS_UNSYNCED,
|
||||
lyrics=unsynced.strip(),
|
||||
source=self.source_name,
|
||||
ttl=TTL_UNSYNCED,
|
||||
)
|
||||
else:
|
||||
logger.debug("LRCLIB-search: best candidate has empty lyrics")
|
||||
return LyricResult(status=CacheStatus.NOT_FOUND, ttl=TTL_NOT_FOUND)
|
||||
|
||||
except httpx.HTTPError as e:
|
||||
logger.error(f"LRCLIB-search: HTTP error: {e}")
|
||||
return LyricResult(status=CacheStatus.NETWORK_ERROR, ttl=TTL_NETWORK_ERROR)
|
||||
except Exception as e:
|
||||
logger.error(f"LRCLIB-search: unexpected error: {e}")
|
||||
return None
|
||||
|
||||
@staticmethod
|
||||
def _select_best(candidates: list[dict], track: TrackMeta) -> Optional[dict]:
|
||||
"""Pick the best candidate, preferring synced lyrics and closest duration."""
|
||||
if track.length is not None:
|
||||
track_s = track.length / 1000.0
|
||||
best: Optional[dict] = None
|
||||
best_diff = float("inf")
|
||||
|
||||
for item in candidates:
|
||||
if not isinstance(item, dict):
|
||||
continue
|
||||
duration = item.get("duration")
|
||||
if not isinstance(duration, (int, float)):
|
||||
continue
|
||||
diff = abs(duration - track_s) * 1000 # compare in ms
|
||||
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):
|
||||
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)")
|
||||
return best
|
||||
|
||||
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():
|
||||
return item
|
||||
return candidates[0] if isinstance(candidates[0], dict) else None
|
||||
@@ -0,0 +1,216 @@
|
||||
"""Netease Cloud Music fetcher.
|
||||
|
||||
Uses the public cloudsearch API for searching and the song/lyric API for
|
||||
retrieving lyrics. No authentication required.
|
||||
|
||||
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 re
|
||||
import httpx
|
||||
from typing import Optional
|
||||
from loguru import logger
|
||||
from lrcfetch.models import TrackMeta, LyricResult, CacheStatus
|
||||
from lrcfetch.fetchers.base import BaseFetcher
|
||||
from lrcfetch.config import (
|
||||
HTTP_TIMEOUT,
|
||||
TTL_NOT_FOUND,
|
||||
TTL_NETWORK_ERROR,
|
||||
DURATION_TOLERANCE_MS,
|
||||
NETEASE_SEARCH_URL,
|
||||
NETEASE_LYRIC_URL,
|
||||
UA_BROWSER,
|
||||
)
|
||||
|
||||
# Matches LRC time tags like [00:12.34] or [01:23.456]
|
||||
_LRC_TIME_TAG_RE = re.compile(r"\[\d{2}:\d{2}\.\d{2,3}\]")
|
||||
# Matches time tags that are all zeros: [00:00.00] or [00:00.000]
|
||||
_ZERO_TIME_TAG_RE = re.compile(r"^\[00:00\.0{2,3}\]")
|
||||
|
||||
_HEADERS = {
|
||||
"User-Agent": UA_BROWSER,
|
||||
"Referer": "https://music.163.com/",
|
||||
}
|
||||
|
||||
|
||||
def _is_synced_lrc(text: str) -> bool:
|
||||
"""Check whether *text* contains actual LRC time tags with non-zero times.
|
||||
|
||||
Returns False if:
|
||||
- No time tags at all
|
||||
- All time tags are [00:00.00] (unsynced disguised as synced)
|
||||
"""
|
||||
lines_with_tags = _LRC_TIME_TAG_RE.findall(text)
|
||||
if not lines_with_tags:
|
||||
return False
|
||||
# Check if ALL tags are zero — if so, it's unsynced
|
||||
for tag in lines_with_tags:
|
||||
if not _ZERO_TIME_TAG_RE.match(tag):
|
||||
return True # Found at least one non-zero tag
|
||||
return False
|
||||
|
||||
|
||||
class NeteaseFetcher(BaseFetcher):
|
||||
@property
|
||||
def source_name(self) -> str:
|
||||
return "netease"
|
||||
|
||||
def _search(self, track: TrackMeta, limit: int = 10) -> Optional[int]:
|
||||
"""Search Netease and return the best-matching song ID.
|
||||
|
||||
When ``track.length`` is available, candidates are ranked by duration
|
||||
difference and only accepted if within ``DURATION_TOLERANCE_MS``.
|
||||
"""
|
||||
query = f"{track.artist or ''} {track.title or ''}".strip()
|
||||
if not query:
|
||||
return None
|
||||
|
||||
logger.debug(f"Netease: searching for '{query}' (limit={limit})")
|
||||
|
||||
try:
|
||||
with httpx.Client(timeout=HTTP_TIMEOUT) as client:
|
||||
resp = client.post(
|
||||
NETEASE_SEARCH_URL,
|
||||
headers=_HEADERS,
|
||||
data={"s": query, "type": "1", "limit": str(limit), "offset": "0"},
|
||||
)
|
||||
resp.raise_for_status()
|
||||
result = resp.json()
|
||||
|
||||
# Validate response
|
||||
if not isinstance(result, dict):
|
||||
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")
|
||||
return None
|
||||
|
||||
songs = result_body.get("songs")
|
||||
if not isinstance(songs, list) or len(songs) == 0:
|
||||
logger.debug("Netease: search returned 0 results")
|
||||
return None
|
||||
|
||||
logger.debug(f"Netease: search returned {len(songs)} candidates")
|
||||
|
||||
# Duration-based best-match selection
|
||||
if track.length is not None:
|
||||
track_ms = track.length
|
||||
best_id: Optional[int] = None
|
||||
best_diff = float("inf")
|
||||
|
||||
for song in songs:
|
||||
if not isinstance(song, dict):
|
||||
continue
|
||||
sid = song.get("id")
|
||||
name = song.get("name", "?")
|
||||
duration = song.get("dt") # milliseconds
|
||||
if not isinstance(duration, int):
|
||||
logger.debug(f" candidate {sid} '{name}': no duration, skipped")
|
||||
continue
|
||||
diff = abs(duration - track_ms)
|
||||
logger.debug(
|
||||
f" candidate {sid} '{name}': "
|
||||
f"duration={duration}ms, diff={diff}ms"
|
||||
)
|
||||
if diff < best_diff:
|
||||
best_diff = diff
|
||||
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)"
|
||||
)
|
||||
return best_id
|
||||
|
||||
logger.debug(
|
||||
f"Netease: no candidate within {DURATION_TOLERANCE_MS}ms "
|
||||
f"(best diff={best_diff}ms)"
|
||||
)
|
||||
return None
|
||||
|
||||
# No duration info — take the first result
|
||||
first = songs[0]
|
||||
if not isinstance(first, dict) or "id" not in first:
|
||||
logger.error("Netease: first search result has no 'id'")
|
||||
return None
|
||||
logger.debug(
|
||||
f"Netease: no duration available, using first result "
|
||||
f"id={first['id']} '{first.get('name', '?')}'"
|
||||
)
|
||||
return first["id"]
|
||||
|
||||
except Exception as e:
|
||||
logger.error(f"Netease: search failed: {e}")
|
||||
return None
|
||||
|
||||
def _get_lyric(self, song_id: int) -> Optional[LyricResult]:
|
||||
"""Fetch lyrics for a given Netease song ID."""
|
||||
logger.debug(f"Netease: fetching lyrics for song_id={song_id}")
|
||||
|
||||
try:
|
||||
with httpx.Client(timeout=HTTP_TIMEOUT) as client:
|
||||
resp = client.post(
|
||||
NETEASE_LYRIC_URL,
|
||||
headers=_HEADERS,
|
||||
data={
|
||||
"id": str(song_id),
|
||||
"cp": "false",
|
||||
"tv": "0",
|
||||
"lv": "0",
|
||||
"rv": "0",
|
||||
"kv": "0",
|
||||
"yv": "0",
|
||||
"ytv": "0",
|
||||
"yrv": "0",
|
||||
},
|
||||
)
|
||||
resp.raise_for_status()
|
||||
data = resp.json()
|
||||
|
||||
# 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)
|
||||
|
||||
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}")
|
||||
return LyricResult(status=CacheStatus.NOT_FOUND, ttl=TTL_NOT_FOUND)
|
||||
|
||||
lrc: str = lrc_obj.get("lyric", "")
|
||||
if not isinstance(lrc, str) or not lrc.strip():
|
||||
logger.debug(f"Netease: empty lyrics for song_id={song_id}")
|
||||
return LyricResult(status=CacheStatus.NOT_FOUND, ttl=TTL_NOT_FOUND)
|
||||
|
||||
# Determine sync status
|
||||
synced = _is_synced_lrc(lrc)
|
||||
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)"
|
||||
)
|
||||
return LyricResult(
|
||||
status=status, lyrics=lrc.strip(), source=self.source_name
|
||||
)
|
||||
|
||||
except Exception as e:
|
||||
logger.error(f"Netease: lyric fetch failed for song_id={song_id}: {e}")
|
||||
return LyricResult(status=CacheStatus.NETWORK_ERROR, ttl=TTL_NETWORK_ERROR)
|
||||
|
||||
def fetch(self, track: TrackMeta) -> Optional[LyricResult]:
|
||||
"""Search for the track and fetch its lyrics."""
|
||||
query = f"{track.artist or ''} {track.title or ''}".strip()
|
||||
if not query:
|
||||
logger.debug("Netease: skipped — insufficient metadata")
|
||||
return None
|
||||
|
||||
logger.info(f"Netease: fetching lyrics for {track.display_name()}")
|
||||
song_id = self._search(track)
|
||||
if not song_id:
|
||||
logger.debug(f"Netease: no match found for {track.display_name()}")
|
||||
return LyricResult(status=CacheStatus.NOT_FOUND, ttl=TTL_NOT_FOUND)
|
||||
|
||||
return self._get_lyric(song_id)
|
||||
@@ -0,0 +1,323 @@
|
||||
"""Spotify fetcher — obtains synced lyrics via Spotify's internal color-lyrics API.
|
||||
|
||||
Authentication flow (mirrors spotify-lyrics Go implementation):
|
||||
1. Fetch server time from Spotify
|
||||
2. Fetch TOTP secret from xyloflake/spot-secrets-go
|
||||
3. Generate a TOTP code and exchange it (with SP_DC cookie) for an access token
|
||||
4. Request lyrics using the access token
|
||||
|
||||
The secret and token are cached on the instance to avoid redundant network
|
||||
calls within the same session.
|
||||
|
||||
Requires SPOTIFY_SP_DC environment variable to be set.
|
||||
"""
|
||||
|
||||
import httpx
|
||||
import time
|
||||
import struct
|
||||
import hmac
|
||||
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 (
|
||||
HTTP_TIMEOUT,
|
||||
TTL_NOT_FOUND,
|
||||
TTL_NETWORK_ERROR,
|
||||
SPOTIFY_TOKEN_URL,
|
||||
SPOTIFY_LYRICS_URL,
|
||||
SPOTIFY_SERVER_TIME_URL,
|
||||
SPOTIFY_SECRET_URL,
|
||||
SPOTIFY_SP_DC,
|
||||
UA_BROWSER,
|
||||
)
|
||||
|
||||
|
||||
class SpotifyFetcher(BaseFetcher):
|
||||
def __init__(self) -> None:
|
||||
# Session-level caches to avoid refetching within the same run
|
||||
self._cached_secret: Optional[Tuple[str, int]] = None
|
||||
self._cached_token: Optional[str] = None
|
||||
self._token_expires_at: float = 0.0
|
||||
|
||||
@property
|
||||
def source_name(self) -> str:
|
||||
return "spotify"
|
||||
|
||||
# ─── Auth helpers ────────────────────────────────────────────────
|
||||
|
||||
def _get_server_time(self, client: httpx.Client) -> Optional[int]:
|
||||
"""Fetch Spotify's server timestamp (seconds since epoch)."""
|
||||
try:
|
||||
res = client.get(SPOTIFY_SERVER_TIME_URL, timeout=HTTP_TIMEOUT)
|
||||
res.raise_for_status()
|
||||
data = res.json()
|
||||
if not isinstance(data, dict) or "serverTime" not in data:
|
||||
logger.error(f"Spotify: unexpected server-time response: {data}")
|
||||
return None
|
||||
server_time = data["serverTime"]
|
||||
logger.debug(f"Spotify: server time = {server_time}")
|
||||
return server_time
|
||||
except Exception as e:
|
||||
logger.error(f"Spotify: failed to fetch server time: {e}")
|
||||
return None
|
||||
|
||||
def _get_secret(self, client: httpx.Client) -> Optional[Tuple[str, int]]:
|
||||
"""Fetch and decode the TOTP secret. Cached after first success.
|
||||
|
||||
Response format: [{version: int, secret: str}, ...]
|
||||
Each character in *secret* is XOR-decoded with ``(index % 33) + 9``.
|
||||
"""
|
||||
if self._cached_secret is not None:
|
||||
logger.debug("Spotify: using cached TOTP secret")
|
||||
return self._cached_secret
|
||||
|
||||
try:
|
||||
res = client.get(SPOTIFY_SECRET_URL, timeout=HTTP_TIMEOUT)
|
||||
res.raise_for_status()
|
||||
data = res.json()
|
||||
|
||||
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 '?'})")
|
||||
return None
|
||||
|
||||
last = data[-1]
|
||||
if "secret" not in last or "version" not in last:
|
||||
logger.error(f"Spotify: malformed secret entry: {list(last.keys())}")
|
||||
return None
|
||||
|
||||
secret_raw = last["secret"]
|
||||
version = last["version"]
|
||||
|
||||
# XOR decode
|
||||
parts = []
|
||||
for i, char in enumerate(secret_raw):
|
||||
parts.append(str(ord(char) ^ ((i % 33) + 9)))
|
||||
secret = "".join(parts)
|
||||
|
||||
logger.debug(f"Spotify: decoded secret v{version} (len={len(secret)})")
|
||||
self._cached_secret = (secret, version)
|
||||
return self._cached_secret
|
||||
|
||||
except Exception as e:
|
||||
logger.error(f"Spotify: failed to fetch secret: {e}")
|
||||
return None
|
||||
|
||||
@staticmethod
|
||||
def _generate_totp(server_time_s: int, secret: str) -> str:
|
||||
"""Generate a 6-digit TOTP code compatible with Spotify's auth.
|
||||
|
||||
Uses HMAC-SHA1 with a 30-second period, matching the Go reference.
|
||||
"""
|
||||
counter = server_time_s // 30
|
||||
counter_bytes = struct.pack(">Q", counter)
|
||||
|
||||
mac = hmac.new(secret.encode(), counter_bytes, hashlib.sha1).digest()
|
||||
|
||||
offset = mac[-1] & 0x0F
|
||||
binary_code = (
|
||||
(mac[offset] & 0x7F) << 24
|
||||
| (mac[offset + 1] & 0xFF) << 16
|
||||
| (mac[offset + 2] & 0xFF) << 8
|
||||
| (mac[offset + 3] & 0xFF)
|
||||
)
|
||||
|
||||
code = binary_code % (10**6)
|
||||
return str(code).zfill(6)
|
||||
|
||||
def _get_token(self) -> Optional[str]:
|
||||
"""Obtain a Spotify access token. Cached until expiry.
|
||||
|
||||
Requires SP_DC cookie (set via SPOTIFY_SP_DC env var).
|
||||
"""
|
||||
# Return cached token if still valid (with 30s safety margin)
|
||||
if self._cached_token and time.time() < self._token_expires_at - 30:
|
||||
logger.debug("Spotify: using cached access token")
|
||||
return self._cached_token
|
||||
|
||||
if not SPOTIFY_SP_DC:
|
||||
logger.error(
|
||||
"Spotify: SPOTIFY_SP_DC env var not set — "
|
||||
"cannot authenticate with Spotify"
|
||||
)
|
||||
return None
|
||||
|
||||
headers = {
|
||||
"User-Agent": UA_BROWSER,
|
||||
"Cookie": f"sp_dc={SPOTIFY_SP_DC}",
|
||||
}
|
||||
|
||||
with httpx.Client(headers=headers) as client:
|
||||
# Step 1: server time
|
||||
server_time = self._get_server_time(client)
|
||||
if server_time is None:
|
||||
return None
|
||||
|
||||
# Step 2: secret
|
||||
secret_data = self._get_secret(client)
|
||||
if secret_data is None:
|
||||
return None
|
||||
|
||||
secret, version = secret_data
|
||||
|
||||
# Step 3: TOTP
|
||||
totp = self._generate_totp(server_time, secret)
|
||||
logger.debug(f"Spotify: generated TOTP v{version}: {totp}")
|
||||
|
||||
# Step 4: exchange for token
|
||||
params = {
|
||||
"reason": "transport",
|
||||
"productType": "web-player",
|
||||
"totp": totp,
|
||||
"totpVer": str(version),
|
||||
"ts": str(int(time.time())),
|
||||
}
|
||||
|
||||
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}"
|
||||
)
|
||||
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__}")
|
||||
return None
|
||||
|
||||
token = body["accessToken"]
|
||||
is_anonymous = body.get("isAnonymous", False)
|
||||
if is_anonymous:
|
||||
logger.warning(
|
||||
"Spotify: received anonymous token — SP_DC may be invalid"
|
||||
)
|
||||
|
||||
# Cache with reported expiry
|
||||
expires_ms = body.get("accessTokenExpirationTimestampMs", 0)
|
||||
if expires_ms and expires_ms > int(time.time() * 1000):
|
||||
self._token_expires_at = expires_ms / 1000.0
|
||||
else:
|
||||
logger.warning("Spotify: token expiry missing or invalid")
|
||||
self._token_expires_at = time.time() + 3600
|
||||
|
||||
self._cached_token = token
|
||||
logger.debug("Spotify: obtained access token")
|
||||
return token
|
||||
|
||||
except Exception as e:
|
||||
logger.error(f"Spotify: token request failed: {e}")
|
||||
return None
|
||||
|
||||
# ─── Lyrics ──────────────────────────────────────────────────────
|
||||
|
||||
@staticmethod
|
||||
def _format_lrc_line(start_ms: int, words: str) -> str:
|
||||
"""Format a single lyric line as LRC ``[mm:ss.cc]text``."""
|
||||
minutes = start_ms // 60000
|
||||
seconds = (start_ms // 1000) % 60
|
||||
centiseconds = round((start_ms % 1000) / 10.0)
|
||||
return f"[{minutes:02d}:{seconds:02d}.{centiseconds:02.0f}]{words}"
|
||||
|
||||
@staticmethod
|
||||
def _is_truly_synced(lines: list[dict]) -> bool:
|
||||
"""Check if lyrics are actually synced (not all timestamps zero)."""
|
||||
for line in lines:
|
||||
try:
|
||||
ms = int(line.get("startTimeMs", "0"))
|
||||
if ms > 0:
|
||||
return True
|
||||
except (ValueError, TypeError):
|
||||
continue
|
||||
return False
|
||||
|
||||
def fetch(self, track: TrackMeta) -> Optional[LyricResult]:
|
||||
"""Fetch lyrics for a Spotify track by its track ID."""
|
||||
if not track.trackid:
|
||||
logger.debug("Spotify: skipped — no trackid in metadata")
|
||||
return None
|
||||
|
||||
logger.info(f"Spotify: fetching lyrics for trackid={track.trackid}")
|
||||
|
||||
token = self._get_token()
|
||||
if not token:
|
||||
logger.error("Spotify: cannot fetch lyrics without a token")
|
||||
return LyricResult(status=CacheStatus.NETWORK_ERROR, ttl=TTL_NETWORK_ERROR)
|
||||
|
||||
url = f"{SPOTIFY_LYRICS_URL}{track.trackid}?format=json&market=from_token"
|
||||
headers = {
|
||||
"User-Agent": UA_BROWSER,
|
||||
"Authorization": f"Bearer {token}",
|
||||
"App-Platform": "WebPlayer",
|
||||
}
|
||||
|
||||
try:
|
||||
with httpx.Client(timeout=HTTP_TIMEOUT) as client:
|
||||
res = client.get(url, headers=headers)
|
||||
|
||||
if res.status_code == 404:
|
||||
logger.debug(f"Spotify: 404 for trackid={track.trackid}")
|
||||
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}")
|
||||
return LyricResult(
|
||||
status=CacheStatus.NETWORK_ERROR, ttl=TTL_NETWORK_ERROR
|
||||
)
|
||||
|
||||
data = res.json()
|
||||
|
||||
# Validate response structure
|
||||
if not isinstance(data, dict) or "lyrics" not in data:
|
||||
logger.error(f"Spotify: unexpected lyrics response structure")
|
||||
return LyricResult(
|
||||
status=CacheStatus.NETWORK_ERROR, ttl=TTL_NETWORK_ERROR
|
||||
)
|
||||
|
||||
lyrics_data = data["lyrics"]
|
||||
sync_type = lyrics_data.get("syncType", "")
|
||||
lines = lyrics_data.get("lines", [])
|
||||
|
||||
if not isinstance(lines, list) or len(lines) == 0:
|
||||
logger.debug("Spotify: response contained no lyric lines")
|
||||
return LyricResult(status=CacheStatus.NOT_FOUND, ttl=TTL_NOT_FOUND)
|
||||
|
||||
# Determine sync status
|
||||
# syncType == "LINE_SYNCED" AND at least one non-zero timestamp
|
||||
is_synced = sync_type == "LINE_SYNCED" and self._is_truly_synced(lines)
|
||||
|
||||
# Convert to LRC
|
||||
lrc_lines: list[str] = []
|
||||
for line in lines:
|
||||
words = line.get("words", "")
|
||||
if not isinstance(words, str):
|
||||
continue
|
||||
try:
|
||||
ms = int(line.get("startTimeMs", "0"))
|
||||
except (ValueError, TypeError):
|
||||
ms = 0
|
||||
|
||||
if is_synced:
|
||||
lrc_lines.append(self._format_lrc_line(ms, words))
|
||||
else:
|
||||
# Unsynced: emit with zero timestamps
|
||||
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)"
|
||||
)
|
||||
return LyricResult(status=status, lyrics=content, source=self.source_name)
|
||||
|
||||
except Exception as e:
|
||||
logger.error(f"Spotify: lyrics fetch failed: {e}")
|
||||
return LyricResult(status=CacheStatus.NETWORK_ERROR, ttl=TTL_NETWORK_ERROR)
|
||||
@@ -0,0 +1,54 @@
|
||||
"""Data models for lrcfetch."""
|
||||
|
||||
from pydantic import BaseModel, ConfigDict
|
||||
from enum import Enum
|
||||
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"
|
||||
NETWORK_ERROR = "NETWORK_ERROR"
|
||||
|
||||
|
||||
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
|
||||
album: Optional[str] = None
|
||||
artist: Optional[str] = None
|
||||
title: Optional[str] = None
|
||||
url: Optional[str] = None # Playback URL (file:// for local files)
|
||||
|
||||
@property
|
||||
def is_local(self) -> bool:
|
||||
"""True when the track is a local file (file:// URL)."""
|
||||
return bool(self.url and self.url.startswith("file://"))
|
||||
|
||||
@property
|
||||
def is_complete(self) -> bool:
|
||||
"""True when all fields required by LRCLIB are present."""
|
||||
return all([self.length, self.album, self.title, self.artist])
|
||||
|
||||
def display_name(self) -> str:
|
||||
"""Human-readable representation for logging."""
|
||||
parts = []
|
||||
if self.artist:
|
||||
parts.append(self.artist)
|
||||
if self.title:
|
||||
parts.append(self.title)
|
||||
return " - ".join(parts) if parts else self.trackid or self.url or "(unknown)"
|
||||
|
||||
|
||||
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)
|
||||
@@ -0,0 +1,169 @@
|
||||
import asyncio
|
||||
from dbus_next.aio.message_bus import MessageBus
|
||||
from dbus_next.constants import BusType
|
||||
from dbus_next.message import Message
|
||||
from lrcfetch.models import TrackMeta
|
||||
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]:
|
||||
try:
|
||||
reply = await bus.call(
|
||||
Message(
|
||||
destination="org.freedesktop.DBus",
|
||||
path="/org/freedesktop/DBus",
|
||||
interface="org.freedesktop.DBus",
|
||||
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]:
|
||||
bus = None
|
||||
try:
|
||||
bus = await MessageBus(bus_type=BusType.SESSION).connect()
|
||||
except Exception as e:
|
||||
logger.error(f"Failed to connect to DBus: {e}")
|
||||
return None
|
||||
|
||||
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 ''}.")
|
||||
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)
|
||||
|
||||
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")
|
||||
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)
|
||||
if trackid:
|
||||
trackid = trackid.value
|
||||
if isinstance(trackid, str):
|
||||
if trackid.startswith("spotify:track:"):
|
||||
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
|
||||
|
||||
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
|
||||
|
||||
return TrackMeta(
|
||||
trackid=trackid,
|
||||
length=length,
|
||||
album=album,
|
||||
artist=artist,
|
||||
title=title,
|
||||
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]:
|
||||
"""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)
|
||||
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
|
||||
|
||||
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
|
||||
)
|
||||
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))
|
||||
if meta:
|
||||
return meta
|
||||
except Exception as e:
|
||||
logger.error(f"DBus async loop failed: {e}")
|
||||
|
||||
return _fetch_metadata_subprocess(player_name)
|
||||
Reference in New Issue
Block a user