Uyanide/lrx-cli
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat: --player bypasses players_blacklist

Browse Source
This commit is contained in:
Uyanide
2026-04-10 15:19:51 +02:00
parent b18d860aca
commit 794f42b42b
6 changed files with 65 additions and 89 deletions
Download Patch File Download Diff File Expand all files Collapse all files
README.md
+1 -1
View File
@@ -112,7 +112,7 @@ optional; all values have defaults. Unknown keys are rejected with an error.
```toml
[general]
preferred_player = "spotify" # preferred MPRIS player when multiple are active
player_blacklist = ["firefox", "zen", "chrome", "chromium", "vivaldi", "edge", "opera", "mpv"]
player_blacklist = ["firefox", "zen", "chrome", "chromium", "vivaldi", "edge", "opera", "mpv"] # bypassed by --player/-p
http_timeout = 10.0 # seconds
[credentials]
src/lrx_cli/cli.py
+1 -1
View File
@@ -71,7 +71,7 @@ def launcher(
str | None,
cyclopts.Parameter(
name=["--player", "-p"],
help="Target a specific MPRIS player using its DBus name or a portion thereof.",
help="Target a specific MPRIS player using its DBus name or a portion thereof. Bypasses player_blacklist.",
),
] = None,
db_path: Annotated[
src/lrx_cli/mpris.py
+44 -29
View File
@@ -17,11 +17,8 @@ from .config import DEFAULT_PLAYER_BLACKLIST, DEFAULT_PREFERRED_PLAYER
from .models import TrackMeta
async def _list_mpris_players(
bus: MessageBus,
player_blacklist: tuple[str, ...],
) -> List[str]:
"""List all MPRIS player bus names, excluding blacklisted entries."""
async def _list_mpris_players(bus: MessageBus) -> List[str]:
"""List all MPRIS player bus names without any filtering."""
try:
reply = await bus.call(
Message(
@@ -34,10 +31,7 @@ async def _list_mpris_players(
if not reply or not reply.body:
return []
return [
name
for name in reply.body[0]
if name.startswith("org.mpris.MediaPlayer2.")
and not any(x.lower() in name.lower() for x in player_blacklist)
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}")
@@ -61,6 +55,32 @@ async def _get_playback_status(bus: MessageBus, player_name: str) -> Optional[st
return None
def pick_active_player(
all_names: list[str],
playing: list[str],
preferred: str,
last_active: str | None = None,
) -> str | None:
"""Select the best MPRIS player by play state, preferred keyword, and continuity.
Priority: single playing > preferred keyword among playing > preferred keyword
among all candidates > last active > first candidate.
"""
if not all_names:
return None
if len(playing) == 1:
return playing[0]
candidates = playing if playing else all_names
preferred_lower = preferred.lower().strip()
if preferred_lower:
for name in candidates:
if preferred_lower in name.lower():
return name
if last_active and last_active in all_names:
return last_active
return candidates[0] if candidates else None
async def _select_player(
bus: MessageBus,
specific_player: Optional[str],
@@ -69,38 +89,33 @@ async def _select_player(
) -> Optional[str]:
"""Select the best MPRIS player.
When specific_player is given, filter by name match.
When specific_player is given, it bypasses player_blacklist and filters by name.
Otherwise: prefer the currently playing player. If multiple are playing,
prefer the one matching preferred_player (default: spotify).
"""
players = await _list_mpris_players(bus, player_blacklist)
if not players:
all_names = await _list_mpris_players(bus)
if not all_names:
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
# --player bypasses player_blacklist so the user can target any player
matched = [p for p in all_names if specific_player.lower() in p.lower()]
return matched[0] if matched else None
# Check playback status for each player
playing = []
for p in players:
# auto-selection: apply blacklist before choosing
candidates = [
p
for p in all_names
if not any(x.lower() in p.lower() for x in player_blacklist)
]
playing: list[str] = []
for p in candidates:
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 preferred_player
preferred = preferred_player.lower()
if preferred:
for p in candidates:
if preferred in p.lower():
return p
return candidates[0]
return pick_active_player(candidates, playing, preferred_player)
async def _fetch_metadata_dbus(
src/lrx_cli/watch/player.py
+14 -41
View File
@@ -17,6 +17,7 @@ from dbus_next.message import Message
from loguru import logger
from ..models import TrackMeta
from ..mpris import pick_active_player
def _variant_value(item: object) -> object | None:
@@ -40,25 +41,6 @@ class PlayerTarget:
"""Constraint for choosing which players are visible to watch."""
hint: Optional[str] = None
player_blacklist: tuple[str, ...] = ()
def validation_error(self) -> str | None:
"""Return validation message when hint conflicts with blacklist, else None."""
normalized_hint = self.normalized_hint
if not normalized_hint:
return None
for blocked in self.player_blacklist:
normalized_blocked = blocked.strip().lower()
if not normalized_blocked:
continue
if _keyword_match(normalized_hint, normalized_blocked) or _keyword_match(
normalized_blocked, normalized_hint
):
return (
f"Requested player '{self.hint}' is blocked by "
f"PLAYER_BLACKLIST entry '{blocked}'."
)
return None
@property
def normalized_hint(self) -> str:
@@ -103,7 +85,7 @@ class PlayerMonitor:
self._on_players_changed = on_players_changed
self._on_seeked = on_seeked
self._on_playback_status = on_playback_status
self._target = target or PlayerTarget(player_blacklist=self._player_blacklist)
self._target = target or PlayerTarget()
self.players: dict[str, PlayerState] = {}
self._bus: MessageBus | None = None
self._props_cache: dict[str, object] = {}
@@ -169,7 +151,11 @@ class PlayerMonitor:
logger.debug(f"Failed to add DBus match rule {rule}: {e}")
async def _list_mpris_players(self) -> list[str]:
"""List visible MPRIS players after applying blacklist and target filter."""
"""List visible MPRIS players after applying target filter and optional blacklist.
The blacklist is skipped when an explicit player hint is active so that
``--player`` can target any player regardless of PLAYER_BLACKLIST.
"""
if not self._bus:
return []
try:
@@ -184,10 +170,14 @@ class PlayerMonitor:
if not reply or not reply.body:
return []
out: list[str] = []
hint_active = bool(self._target.normalized_hint)
for name in reply.body[0]:
if not name.startswith("org.mpris.MediaPlayer2."):
continue
if any(x.lower() in name.lower() for x in self._player_blacklist):
# --player bypasses the blacklist; only filter when no hint is given
if not hint_active and any(
x.lower() in name.lower() for x in self._player_blacklist
):
continue
if not self._target.allows(name):
continue
@@ -406,23 +396,6 @@ class ActivePlayerSelector:
"""Select active player by playing state, preferred keyword, and continuity."""
if not players:
return None
all_names = list(players.keys())
playing = [name for name, st in players.items() if st.status == "Playing"]
if len(playing) == 1:
# unambiguous — only one player is currently playing
return playing[0]
preferred = preferred_player.lower().strip()
# when multiple players are playing, narrow candidates to those; otherwise
# fall back to all known players so a paused preferred player still wins
candidates = playing if playing else list(players.keys())
if preferred:
for name in candidates:
if preferred in name.lower():
return name
# preserve the last selection to avoid jitter when nothing else changes
if last_active and last_active in players:
return last_active
return candidates[0] if candidates else None
return pick_active_player(all_names, playing, preferred_player, last_active)
src/lrx_cli/watch/session.py
+1 -9
View File
@@ -129,10 +129,7 @@ class WatchCoordinator:
self._emit_scheduled = False
self._calibration_task = None
self._target = PlayerTarget(
hint=player_hint,
player_blacklist=self._config.general.player_blacklist,
)
self._target = PlayerTarget(hint=player_hint)
self._control = ControlServer(socket_path=config.watch.socket_path)
self._player_monitor = PlayerMonitor(
@@ -156,11 +153,6 @@ class WatchCoordinator:
async def run(self) -> bool:
"""Run watch workflow and return success flag."""
target_issue = self._target.validation_error()
if target_issue:
logger.error(target_issue)
return False
logger.info(
"watch session starting (player filter: {})",
self._player_hint or "<none>",
tests/test_watch.py
+4 -8
View File
@@ -41,14 +41,10 @@ def test_player_target_filters_by_case_insensitive_substring() -> None:
assert target.allows("org.mpris.MediaPlayer2.mpd") is False
def test_player_target_reports_blacklisted_hint() -> None:
target = PlayerTarget("spot", player_blacklist=("spotify",))
assert target.validation_error() is not None
def test_player_target_non_blacklisted_hint_is_valid() -> None:
target = PlayerTarget("mpd", player_blacklist=("spotify",))
assert target.validation_error() is None
def test_player_target_hint_allows_regardless_of_blacklist() -> None:
# --player bypasses PLAYER_BLACKLIST; PlayerTarget.allows() reflects the hint only
target = PlayerTarget("spot")
assert target.allows("org.mpris.MediaPlayer2.spotify") is True
# ActivePlayerSelector