Compare commits
2 Commits
34dfe7d042
...
02abfe636f
| Author | SHA1 | Date | |
|---|---|---|---|
|
02abfe636f
|
|||
|
1a301deb40
|
@@ -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 @@
|
||||
__version__ = "0.1.3"
|
||||
__version__ = "0.1.4"
|
||||
|
||||
+10
-1
@@ -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"
|
||||
|
||||
@@ -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
@@ -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
|
||||
|
||||
@@ -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
@@ -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
@@ -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"
|
||||
|
||||
Reference in New Issue
Block a user