Compare commits

...

2 Commits

9 changed files with 134 additions and 95 deletions
+5
View File
@@ -49,8 +49,13 @@ Set credentials via environment variable or `.env` file:
```env ```env
SPOTIFY_SP_DC=your_cookie_value SPOTIFY_SP_DC=your_cookie_value
QQ_MUSIC_API_URL=https://api.example.com QQ_MUSIC_API_URL=https://api.example.com
LRCFETCH_PLAYER=spotify
``` ```
- `SPOTIFY_SP_DC` — required for Spotify source. Defaults to empty (disabled Spotify source).
- `QQ_MUSIC_API_URL` — required for QQ Music source. Defaults to empty (disabled QQ Music source).
- `LRCFETCH_PLAYER` — preferred MPRIS player when multiple are active. Defaults to `spotify`. Only used when no `--player` flag is given and more than one player (or none of them) is currently playing.
Shell completion (zsh/fish/bash): Shell completion (zsh/fish/bash):
```bash ```bash
+1 -1
View File
@@ -1 +1 @@
__version__ = "0.1.3" __version__ = "0.1.4"
+10 -1
View File
@@ -15,6 +15,7 @@ from .config import enable_debug
from .models import TrackMeta, CacheStatus from .models import TrackMeta, CacheStatus
from .mpris import get_current_track from .mpris import get_current_track
from .core import LrcManager, FetcherMethodType from .core import LrcManager, FetcherMethodType
from .lrc import get_sidecar_path
app = cyclopts.App( app = cyclopts.App(
@@ -174,7 +175,7 @@ def export(
str | None, str | None,
cyclopts.Parameter( cyclopts.Parameter(
name=["--output", "-o"], name=["--output", "-o"],
help="Output file path (default: <Artist> - <Title>.lrc).", help="Output file path (default: same directory as audio file with .lrc extension, or current directory if not available).",
), ),
] = None, ] = None,
method: Annotated[ method: Annotated[
@@ -202,6 +203,14 @@ def export(
sys.exit(1) sys.exit(1)
# Build default output path # Build default output path
if not output:
if track.url:
lrc_path = get_sidecar_path(track.url, ensure_exists=False)
if lrc_path:
output = str(lrc_path)
logger.info(f"Exporting to sidecar path: {output}")
# Fallback to current directory with sanitized filename
if not output: if not output:
filename = ( filename = (
f"{track.artist} - {track.title}.lrc" f"{track.artist} - {track.title}.lrc"
+3
View File
@@ -58,6 +58,9 @@ LRCLIB_SEARCH_URL = "https://lrclib.net/api/search"
# QQ Music API (self-hosted proxy) # QQ Music API (self-hosted proxy)
QQ_MUSIC_API_URL = os.environ.get("QQ_MUSIC_API_URL", "").rstrip("/") QQ_MUSIC_API_URL = os.environ.get("QQ_MUSIC_API_URL", "").rstrip("/")
# Player preference (used when multiple MPRIS players are active)
PREFERRED_PLAYER = os.environ.get("LRCFETCH_PLAYER", "spotify")
# User-Agents # User-Agents
UA_BROWSER = "Mozilla/5.0 (X11; Linux x86_64; rv:148.0) Gecko/20100101 Firefox/148.0" 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)" UA_LRCFETCH = "LRCFetch (https://github.com/Uyanide/lrcfetch)"
+16 -14
View File
@@ -10,16 +10,14 @@ Priority:
2. Embedded lyrics in audio metadata (FLAC, MP3 USLT/SYLT tags) 2. Embedded lyrics in audio metadata (FLAC, MP3 USLT/SYLT tags)
""" """
import os
from typing import Optional from typing import Optional
from urllib.parse import unquote
from loguru import logger from loguru import logger
from mutagen._file import File from mutagen._file import File
from mutagen.flac import FLAC from mutagen.flac import FLAC
from .base import BaseFetcher from .base import BaseFetcher
from ..models import TrackMeta, LyricResult from ..models import TrackMeta, LyricResult
from ..lrc import detect_sync_status from ..lrc import detect_sync_status, get_audio_path, get_sidecar_path
class LocalFetcher(BaseFetcher): class LocalFetcher(BaseFetcher):
@@ -32,16 +30,15 @@ class LocalFetcher(BaseFetcher):
if not track.is_local or not track.url: if not track.is_local or not track.url:
return None return None
file_path = unquote(track.url.replace("file://", "", 1)) audio_path = get_audio_path(track.url, ensure_exists=False)
if not os.path.exists(file_path): if not audio_path:
logger.debug(f"Local: file does not exist: {file_path}") logger.debug(f"Local: audio URL is not a valid file path: {track.url}")
return None return None
logger.info(f"Local: checking for lyrics near {file_path}") lrc_path = get_sidecar_path(
track.url, ensure_audio_exists=False, ensure_exists=True
# Sidecar .lrc file )
lrc_path = os.path.splitext(file_path)[0] + ".lrc" if lrc_path:
if os.path.exists(lrc_path):
try: try:
with open(lrc_path, "r", encoding="utf-8") as f: with open(lrc_path, "r", encoding="utf-8") as f:
content = f.read().strip() content = f.read().strip()
@@ -53,10 +50,15 @@ class LocalFetcher(BaseFetcher):
) )
except Exception as e: except Exception as e:
logger.error(f"Local: error reading {lrc_path}: {e}") logger.error(f"Local: error reading {lrc_path}: {e}")
else:
logger.debug(f"Local: no .lrc sidecar found for {audio_path}")
# Embedded metadata # Embedded metadata
if not audio_path.exists():
logger.debug(f"Local: audio file does not exist: {audio_path}")
return None
try: try:
audio = File(file_path) audio = File(audio_path)
if audio is not None: if audio is not None:
lyrics = None lyrics = None
@@ -83,7 +85,7 @@ class LocalFetcher(BaseFetcher):
else: else:
logger.debug("Local: no embedded lyrics found") logger.debug("Local: no embedded lyrics found")
except Exception as e: except Exception as e:
logger.error(f"Local: error reading metadata for {file_path}: {e}") logger.error(f"Local: error reading metadata for {audio_path}: {e}")
logger.debug(f"Local: no lyrics found for {file_path}") logger.debug(f"Local: no lyrics found for {audio_path}")
return None return None
+31
View File
@@ -5,6 +5,9 @@ Description: Shared LRC time-tag utilities
""" """
import re import re
from pathlib import Path
from typing import Optional
from urllib.parse import unquote
from .models import CacheStatus from .models import CacheStatus
@@ -93,3 +96,31 @@ def detect_sync_status(text: str) -> CacheStatus:
return ( return (
CacheStatus.SUCCESS_SYNCED if is_synced(text) else CacheStatus.SUCCESS_UNSYNCED CacheStatus.SUCCESS_SYNCED if is_synced(text) else CacheStatus.SUCCESS_UNSYNCED
) )
def get_audio_path(audio_url: str, ensure_exists: bool = False) -> Optional[Path]:
"""Convert file:// URL to Path, return None if invalid or (if ensure_exists) file doesn't exist."""
if not audio_url.startswith("file://"):
return None
file_path = unquote(audio_url.replace("file://", "", 1))
path = Path(file_path)
if ensure_exists and not path.exists():
return None
return path
def get_sidecar_path(
audio_url: str, ensure_audio_exists: bool = False, ensure_exists: bool = False
) -> Optional[Path]:
"""Given a file:// URL, return the corresponding .lrc sidecar path.
If ensure_audio_exists is True, return None if the audio file does not exist.
If ensure_exists is True, return None if the .lrc file does not exist.
"""
audio_path = get_audio_path(audio_url, ensure_exists=ensure_audio_exists)
if not audio_path:
return None
lrc_path = audio_path.with_suffix(".lrc")
if ensure_exists and not lrc_path.exists():
return None
return lrc_path
+66 -77
View File
@@ -9,14 +9,13 @@ from dbus_next.aio.message_bus import MessageBus
from dbus_next.constants import BusType from dbus_next.constants import BusType
from dbus_next.message import Message from dbus_next.message import Message
from lrcfetch.models import TrackMeta from lrcfetch.models import TrackMeta
from lrcfetch.config import PREFERRED_PLAYER
from loguru import logger from loguru import logger
from typing import Optional, List, Any from typing import Optional, List, Any
import subprocess
async def _get_active_players( async def _list_mpris_players(bus: MessageBus) -> List[str]:
bus: MessageBus, specific_player: Optional[str] = None """List all MPRIS player bus names."""
) -> List[str]:
try: try:
reply = await bus.call( reply = await bus.call(
Message( Message(
@@ -28,22 +27,70 @@ async def _get_active_players(
) )
if not reply or not reply.body: if not reply or not reply.body:
return [] return []
return [
names = reply.body[0] name for name in reply.body[0] if name.startswith("org.mpris.MediaPlayer2.")
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: except Exception as e:
logger.error(f"Failed to list DBus names: {e}") logger.error(f"Failed to list DBus names: {e}")
return [] return []
async def _get_playback_status(bus: MessageBus, player_name: str) -> Optional[str]:
"""Get PlaybackStatus ('Playing', 'Paused', 'Stopped') for a player."""
try:
introspection = await bus.introspect(player_name, "/org/mpris/MediaPlayer2")
proxy = bus.get_proxy_object(
player_name, "/org/mpris/MediaPlayer2", introspection
)
props = proxy.get_interface("org.freedesktop.DBus.Properties")
status_var = await getattr(props, "call_get")(
"org.mpris.MediaPlayer2.Player", "PlaybackStatus"
)
return status_var.value if status_var else None
except Exception as e:
logger.debug(f"Could not get playback status for {player_name}: {e}")
return None
async def _select_player(
bus: MessageBus, specific_player: Optional[str] = None
) -> Optional[str]:
"""Select the best MPRIS player.
When specific_player is given, filter by name match.
Otherwise: prefer the currently playing player. If multiple are playing,
prefer the one matching LRCFETCH_PLAYER env var (default: spotify).
"""
players = await _list_mpris_players(bus)
if not players:
return None
if specific_player:
players = [p for p in players if specific_player.lower() in p.lower()]
return players[0] if players else None
# Check playback status for each player
playing = []
for p in players:
status = await _get_playback_status(bus, p)
logger.debug(f"Player {p}: {status}")
if status == "Playing":
playing.append(p)
candidates = playing if playing else players
if len(candidates) == 1:
return candidates[0]
# Multiple candidates: prefer LRCFETCH_PLAYER
preferred = PREFERRED_PLAYER.lower()
if preferred:
for p in candidates:
if preferred in p.lower():
return p
return candidates[0]
async def _fetch_metadata_dbus( async def _fetch_metadata_dbus(
specific_player: Optional[str] = None, specific_player: Optional[str] = None,
) -> Optional[TrackMeta]: ) -> Optional[TrackMeta]:
@@ -55,14 +102,13 @@ async def _fetch_metadata_dbus(
return None return None
try: try:
players = await _get_active_players(bus, specific_player) player_name = await _select_player(bus, specific_player)
if not players: if not player_name:
logger.debug( logger.debug(
f"No active MPRIS players found via DBus{' for ' + specific_player if specific_player else ''}." f"No active MPRIS players found via DBus{' for ' + specific_player if specific_player else ''}."
) )
return None return None
player_name = players[0]
logger.debug(f"Using player: {player_name}") logger.debug(f"Using player: {player_name}")
introspection = await bus.introspect(player_name, "/org/mpris/MediaPlayer2") introspection = await bus.introspect(player_name, "/org/mpris/MediaPlayer2")
@@ -134,66 +180,9 @@ async def _fetch_metadata_dbus(
bus.disconnect() 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]: def get_current_track(player_name: Optional[str] = None) -> Optional[TrackMeta]:
try: try:
meta = asyncio.run(_fetch_metadata_dbus(player_name)) return asyncio.run(_fetch_metadata_dbus(player_name))
if meta:
return meta
except Exception as e: except Exception as e:
logger.error(f"DBus async loop failed: {e}") logger.error(f"DBus async loop failed: {e}")
return None
return _fetch_metadata_subprocess(player_name)
+1 -1
View File
@@ -4,7 +4,7 @@ build-backend = "hatchling.build"
[project] [project]
name = "lrcfetch" name = "lrcfetch"
version = "0.1.3" version = "0.1.4"
description = "Fetch line-synced lyrics for your music player." description = "Fetch line-synced lyrics for your music player."
readme = "README.md" readme = "README.md"
requires-python = ">=3.13" requires-python = ">=3.13"
Generated
+1 -1
View File
@@ -153,7 +153,7 @@ wheels = [
[[package]] [[package]]
name = "lrcfetch" name = "lrcfetch"
version = "0.1.3" version = "0.1.4"
source = { editable = "." } source = { editable = "." }
dependencies = [ dependencies = [
{ name = "cyclopts" }, { name = "cyclopts" },