diff --git a/pyproject.toml b/pyproject.toml index 6145765..6ab6e2d 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -4,7 +4,7 @@ build-backend = "hatchling.build" [project] name = "lrx-cli" -version = "0.7.11" +version = "0.7.12" description = "Fetch line-synced lyrics for your music player." readme = "README.md" requires-python = ">=3.13" diff --git a/src/lrx_cli/config.py b/src/lrx_cli/config.py index 3bb9cae..7779218 100644 --- a/src/lrx_cli/config.py +++ b/src/lrx_cli/config.py @@ -61,7 +61,7 @@ MULTI_CANDIDATE_DELAY_S = 0.2 # delay between sequential lyric fetches LEGACY_CONFIDENCE = 50.0 # User-Agents -UA_BROWSER = "Mozilla/5.0 (X11; Linux x86_64; rv:152.0) Gecko/20100101 Firefox/152.0" +UA_BROWSER = "Mozilla/5.0 (Macintosh; Intel Mac OS X 15_8_0) AppleWebKit/605.1.15 (KHTML, like Gecko) Version/27.0 Safari/605.1.15" UA_LRX = f"LRX-CLI {APP_VERSION} (https://github.com/Uyanide/lrx-cli)" MUSIXMATCH_COOLDOWN_MS = 600_000 # 10 minutes diff --git a/src/lrx_cli/mpris.py b/src/lrx_cli/mpris.py index 3ae9604..4a30f27 100644 --- a/src/lrx_cli/mpris.py +++ b/src/lrx_cli/mpris.py @@ -8,7 +8,7 @@ from __future__ import annotations import asyncio from dbus_next.aio.message_bus import MessageBus -from dbus_next.constants import BusType +from dbus_next.constants import BusType, MessageType from dbus_next.message import Message from loguru import logger from typing import Optional, List, Any @@ -38,17 +38,39 @@ async def _list_mpris_players(bus: MessageBus) -> List[str]: return [] +async def get_player_property( + bus: MessageBus, player_name: str, prop: str +) -> Optional[Any]: + """Read an org.mpris.MediaPlayer2.Player property with a raw Properties.Get. + + Raw calls keep property access independent of the peer's introspection data. + """ + reply = await bus.call( + Message( + destination=player_name, + path="/org/mpris/MediaPlayer2", + interface="org.freedesktop.DBus.Properties", + member="Get", + signature="ss", + body=["org.mpris.MediaPlayer2.Player", prop], + ) + ) + if reply is None: + return None + if reply.message_type != MessageType.METHOD_RETURN: + raise RuntimeError( + f"Properties.Get({prop}) failed: " + f"{reply.body[0] if reply.body else reply.error_name or 'no error body'}" + ) + if not reply.body: + return None + return reply.body[0] + + 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" - ) + status_var = await get_player_property(bus, player_name, "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}") @@ -128,13 +150,15 @@ async def _fetch_metadata_dbus( specific_player: Optional[str], preferred_player: str, player_blacklist: tuple[str, ...], + bus: Optional[MessageBus] = 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 + owns_bus = bus is None + if bus is 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: player_name = await _select_player( @@ -148,20 +172,8 @@ async def _fetch_metadata_dbus( 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" - ) + metadata_var = await get_player_property(bus, player_name, "Metadata") if not metadata_var: logger.error("Empty metadata received.") return None @@ -215,7 +227,7 @@ async def _fetch_metadata_dbus( return None finally: - if bus: + if owns_bus and bus: bus.disconnect() diff --git a/src/lrx_cli/watch/player.py b/src/lrx_cli/watch/player.py index a9baade..d789cb4 100644 --- a/src/lrx_cli/watch/player.py +++ b/src/lrx_cli/watch/player.py @@ -17,7 +17,7 @@ from dbus_next.message import Message from loguru import logger from ..models import TrackMeta -from ..mpris import pick_active_player +from ..mpris import get_player_property, pick_active_player def _variant_value(item: object) -> object | None: @@ -70,7 +70,6 @@ class PlayerMonitor: _target: PlayerTarget players: dict[str, PlayerState] _bus: MessageBus | None - _props_cache: dict[str, object] def __init__( self, @@ -88,7 +87,6 @@ class PlayerMonitor: self._target = target or PlayerTarget() self.players: dict[str, PlayerState] = {} self._bus: MessageBus | None = None - self._props_cache: dict[str, object] = {} async def start(self) -> None: """Start DBus monitoring and populate initial player snapshot.""" @@ -99,33 +97,10 @@ class PlayerMonitor: async def close(self) -> None: """Stop DBus monitoring and close bus connection.""" - self._props_cache.clear() if self._bus: self._bus.disconnect() self._bus = None - async def _get_player_props(self, bus_name: str) -> object | None: - """Return cached DBus Properties interface for player, creating it if missing.""" - if not self._bus: - return None - if bus_name in self._props_cache: - return self._props_cache[bus_name] - - try: - introspection = await self._bus.introspect( - bus_name, "/org/mpris/MediaPlayer2" - ) - proxy = self._bus.get_proxy_object( - bus_name, "/org/mpris/MediaPlayer2", introspection - ) - props = proxy.get_interface("org.freedesktop.DBus.Properties") - self._props_cache[bus_name] = props - return props - except Exception as e: - logger.debug(f"Failed to prepare DBus props for {bus_name}: {e}") - self._props_cache.pop(bus_name, None) - return None - async def _add_match_rules(self) -> None: """Register signal subscriptions needed by monitor.""" if not self._bus: @@ -190,16 +165,13 @@ class PlayerMonitor: async def _fetch_player_state(self, bus_name: str) -> Optional[PlayerState]: """Read current playback status and metadata from one player service.""" - props = await self._get_player_props(bus_name) - if props is None: + if not self._bus: return None try: - status_var = await getattr(props, "call_get")( - "org.mpris.MediaPlayer2.Player", "PlaybackStatus" - ) - metadata_var = await getattr(props, "call_get")( - "org.mpris.MediaPlayer2.Player", "Metadata" + status_var = await get_player_property( + self._bus, bus_name, "PlaybackStatus" ) + metadata_var = await get_player_property(self._bus, bus_name, "Metadata") status = status_var.value if status_var else "Stopped" track = self._track_from_metadata( metadata_var.value if metadata_var else {} @@ -207,7 +179,6 @@ class PlayerMonitor: return PlayerState(bus_name=bus_name, status=status, track=track) except Exception as e: logger.debug(f"Failed to read state for {bus_name}: {e}") - self._props_cache.pop(bus_name, None) return None def _track_from_metadata(self, metadata: dict[str, object]) -> Optional[TrackMeta]: @@ -270,9 +241,6 @@ class PlayerMonitor: added = sorted(after - before) removed = sorted(before - after) - for bus_name in removed: - self._props_cache.pop(bus_name, None) - self.players = updated if added or removed: @@ -371,19 +339,15 @@ class PlayerMonitor: async def get_position_ms(self, bus_name: str) -> Optional[int]: """Read player-reported position in milliseconds.""" - props = await self._get_player_props(bus_name) - if props is None: + if not self._bus: return None try: - position_var = await getattr(props, "call_get")( - "org.mpris.MediaPlayer2.Player", "Position" - ) + position_var = await get_player_property(self._bus, bus_name, "Position") if position_var is None: return None return max(0, int(position_var.value) // 1000) except Exception as e: logger.debug(f"Failed to read position from {bus_name}: {e}") - self._props_cache.pop(bus_name, None) return None diff --git a/tests/test_mpris.py b/tests/test_mpris.py new file mode 100644 index 0000000..41e0314 --- /dev/null +++ b/tests/test_mpris.py @@ -0,0 +1,158 @@ +"""Regression tests: MPRIS property reads must be issued as raw +org.freedesktop.DBus.Properties.Get messages, independent of introspection. +""" + +import asyncio +from typing import Callable, Optional + +import pytest +from dbus_next.constants import MessageType +from dbus_next.message import Message +from dbus_next.signature import Variant + +from lrx_cli.models import TrackMeta +from lrx_cli.mpris import _fetch_metadata_dbus, _get_playback_status +from lrx_cli.watch.player import PlayerMonitor + +PLAYER = "org.mpris.MediaPlayer2.generic1" + +PLAYER_METADATA = { + "mpris:trackid": Variant("o", "/org/example/Tracks/42"), + "mpris:length": Variant("x", 239_761_000), + "mpris:artUrl": Variant("s", "http://example.com/cover.jpg"), + "xesam:album": Variant("s", "album"), + "xesam:artist": Variant("as", ["artist"]), + "xesam:title": Variant("s", "title"), + "xesam:url": Variant("s", "http://example.com/stream"), +} + + +class FakeBus: + """Test double exposing only the low-level call/disconnect surface.""" + + def __init__(self, reply: Callable[[Message], Optional[Message]]) -> None: + self._reply = reply + self.calls: list[Message] = [] + self.disconnected = False + + async def call(self, message: Message) -> Optional[Message]: + self.calls.append(message) + return self._reply(message) + + def disconnect(self) -> None: + self.disconnected = True + + +def _return(signature: str, body: list) -> Message: + return Message( + message_type=MessageType.METHOD_RETURN, + reply_serial=1, + signature=signature, + body=body, + ) + + +def _names_reply(names: list[str]) -> Message: + return _return("as", [names]) + + +def _metadata_reply() -> Message: + return _return("v", [Variant("a{sv}", PLAYER_METADATA)]) + + +def _state_reply(msg: Message) -> Optional[Message]: + """Reply to ListNames and the PlaybackStatus/Metadata property reads.""" + if msg.member == "ListNames": + return _names_reply([PLAYER]) + if msg.member == "Get": + if msg.body[1] == "PlaybackStatus": + return _return("v", [Variant("s", "Playing")]) + if msg.body[1] == "Metadata": + return _metadata_reply() + return None + + +def _monitor(bus: FakeBus) -> PlayerMonitor: + monitor = PlayerMonitor( + on_players_changed=lambda: None, + on_seeked=lambda bus_name, position_ms: None, + on_playback_status=lambda bus_name, status: None, + player_blacklist=(), + ) + monitor._bus = bus # type: ignore[assignment] + return monitor + + +def test_get_playback_status_sends_raw_properties_get() -> None: + def reply(msg: Message) -> Optional[Message]: + if msg.member == "Get": + return _return("v", [Variant("s", "Playing")]) + return None + + bus = FakeBus(reply) + status = asyncio.run(_get_playback_status(bus, PLAYER)) # type: ignore[arg-type] + + assert status == "Playing" + request = bus.calls[0] + assert request.destination == PLAYER + assert request.path == "/org/mpris/MediaPlayer2" + assert request.interface == "org.freedesktop.DBus.Properties" + assert request.member == "Get" + assert request.body == ["org.mpris.MediaPlayer2.Player", "PlaybackStatus"] + + +def test_fetch_metadata_dbus_reads_metadata_via_raw_properties_get() -> None: + bus = FakeBus(_state_reply) + track = asyncio.run(_fetch_metadata_dbus(None, "", (), bus=bus)) # type: ignore[arg-type] + + # trackid without a Spotify prefix normalizes to None + assert track == TrackMeta( + trackid=None, + length=239_761, + album="album", + artist="artist", + title="title", + url="http://example.com/stream", + ) + assert not any(msg.member == "Introspect" for msg in bus.calls) + # injected bus stays owned by the caller + assert bus.disconnected is False + + +def test_fetch_metadata_dbus_disconnects_self_created_bus( + monkeypatch: pytest.MonkeyPatch, +) -> None: + bus = FakeBus(_state_reply) + + class FakeMessageBus: + @staticmethod + async def connect() -> FakeBus: + return bus + + monkeypatch.setattr("lrx_cli.mpris.MessageBus", lambda bus_type: FakeMessageBus) + track = asyncio.run(_fetch_metadata_dbus(None, "", ())) + + assert track is not None + assert track.title == "title" + assert bus.disconnected is True + + +def test_player_monitor_refresh_reads_state_via_raw_properties_get() -> None: + monitor = _monitor(FakeBus(_state_reply)) + asyncio.run(monitor.refresh()) + + state = monitor.players.get(PLAYER) + assert state is not None + assert state.status == "Playing" + assert state.track is not None + assert state.track.title == "title" + + +def test_player_monitor_get_position_ms_reads_raw_position() -> None: + def reply(msg: Message) -> Optional[Message]: + if msg.member == "Get": + return _return("v", [Variant("x", 231_008_000)]) + return None + + monitor = _monitor(FakeBus(reply)) + assert asyncio.run(monitor.get_position_ms(PLAYER)) == 231_008 diff --git a/uv.lock b/uv.lock index 265727c..3d182f1 100644 --- a/uv.lock +++ b/uv.lock @@ -153,7 +153,7 @@ wheels = [ [[package]] name = "lrx-cli" -version = "0.7.11" +version = "0.7.12" source = { editable = "." } dependencies = [ { name = "cyclopts" },