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
SPOTIFY_SP_DC=your_cookie_value
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):
```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 .mpris import get_current_track
from .core import LrcManager, FetcherMethodType
from .lrc import get_sidecar_path
app = cyclopts.App(
@@ -174,7 +175,7 @@ def export(
str | None,
cyclopts.Parameter(
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,
method: Annotated[
@@ -202,6 +203,14 @@ def export(
sys.exit(1)
# 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:
filename = (
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_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
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)"
+16 -14
View File
@@ -10,16 +10,14 @@ Priority:
2. Embedded lyrics in audio metadata (FLAC, MP3 USLT/SYLT tags)
"""
import os
from typing import Optional
from urllib.parse import unquote
from loguru import logger
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
from ..lrc import detect_sync_status, get_audio_path, get_sidecar_path
class LocalFetcher(BaseFetcher):
@@ -32,16 +30,15 @@ class LocalFetcher(BaseFetcher):
if not track.is_local or not track.url:
return None
file_path = unquote(track.url.replace("file://", "", 1))
if not os.path.exists(file_path):
logger.debug(f"Local: file does not exist: {file_path}")
audio_path = get_audio_path(track.url, ensure_exists=False)
if not audio_path:
logger.debug(f"Local: audio URL is not a valid file path: {track.url}")
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):
lrc_path = get_sidecar_path(
track.url, ensure_audio_exists=False, ensure_exists=True
)
if lrc_path:
try:
with open(lrc_path, "r", encoding="utf-8") as f:
content = f.read().strip()
@@ -53,10 +50,15 @@ class LocalFetcher(BaseFetcher):
)
except Exception as e:
logger.error(f"Local: error reading {lrc_path}: {e}")
else:
logger.debug(f"Local: no .lrc sidecar found for {audio_path}")
# Embedded metadata
if not audio_path.exists():
logger.debug(f"Local: audio file does not exist: {audio_path}")
return None
try:
audio = File(file_path)
audio = File(audio_path)
if audio is not None:
lyrics = None
@@ -83,7 +85,7 @@ class LocalFetcher(BaseFetcher):
else:
logger.debug("Local: no embedded lyrics found")
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
+31
View File
@@ -5,6 +5,9 @@ Description: Shared LRC time-tag utilities
"""
import re
from pathlib import Path
from typing import Optional
from urllib.parse import unquote
from .models import CacheStatus
@@ -93,3 +96,31 @@ def detect_sync_status(text: str) -> CacheStatus:
return (
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.message import Message
from lrcfetch.models import TrackMeta
from lrcfetch.config import PREFERRED_PLAYER
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 _list_mpris_players(bus: MessageBus) -> List[str]:
"""List all MPRIS player bus names."""
try:
reply = await bus.call(
Message(
@@ -28,22 +27,70 @@ async def _get_active_players(
)
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
return [
name for name in reply.body[0] if name.startswith("org.mpris.MediaPlayer2.")
]
except Exception as e:
logger.error(f"Failed to list DBus names: {e}")
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(
specific_player: Optional[str] = None,
) -> Optional[TrackMeta]:
@@ -55,14 +102,13 @@ async def _fetch_metadata_dbus(
return None
try:
players = await _get_active_players(bus, specific_player)
if not players:
player_name = await _select_player(bus, specific_player)
if not player_name:
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")
@@ -134,66 +180,9 @@ async def _fetch_metadata_dbus(
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
return asyncio.run(_fetch_metadata_dbus(player_name))
except Exception as e:
logger.error(f"DBus async loop failed: {e}")
return _fetch_metadata_subprocess(player_name)
return None
+1 -1
View File
@@ -4,7 +4,7 @@ build-backend = "hatchling.build"
[project]
name = "lrcfetch"
version = "0.1.3"
version = "0.1.4"
description = "Fetch line-synced lyrics for your music player."
readme = "README.md"
requires-python = ">=3.13"
Generated
+1 -1
View File
@@ -153,7 +153,7 @@ wheels = [
[[package]]
name = "lrcfetch"
version = "0.1.3"
version = "0.1.4"
source = { editable = "." }
dependencies = [
{ name = "cyclopts" },