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