498 lines
19 KiB
Python
Executable File
498 lines
19 KiB
Python
Executable File
#!/usr/bin/env python3
|
||
import requests
|
||
import time
|
||
import re
|
||
import subprocess
|
||
import os
|
||
import logging
|
||
from threading import Thread
|
||
from pydbus import SessionBus
|
||
from pydbus.generic import signal
|
||
from gi.repository import GLib
|
||
from bs4 import BeautifulSoup
|
||
|
||
# --- CONFIGURATION & XDG PATHS ---
|
||
BASE_URL = "http://localhost:5666"
|
||
AUTH = ('winamp', 'llama')
|
||
POSSIBLE_CREDS = [('winamp', 'llama'), ('winamp', ''), None]
|
||
APP_ID = "org.mpris.MediaPlayer2.winamp"
|
||
DEFAULT_ART = "https://webamp.org/favicon.ico"
|
||
|
||
# XDG Standard Paths
|
||
XDG_STATE_HOME = os.environ.get("XDG_STATE_HOME", os.path.expanduser("~/.local/state"))
|
||
XDG_RUNTIME_DIR = os.environ.get("XDG_RUNTIME_DIR", f"/run/user/{os.getuid()}")
|
||
|
||
LOG_DIR = os.path.join(XDG_STATE_HOME, "winamp-mpris")
|
||
LOG_FILE = os.path.join(LOG_DIR, "bridge.log")
|
||
PID_FILE = os.path.join(XDG_RUNTIME_DIR, "winamp-mpris.pid")
|
||
|
||
# Ensure log directory exists
|
||
os.makedirs(LOG_DIR, exist_ok=True)
|
||
|
||
# Configure Logging
|
||
logging.basicConfig(
|
||
level=logging.INFO,
|
||
format='%(asctime)s [%(levelname)s] %(message)s',
|
||
handlers=[
|
||
logging.FileHandler(LOG_FILE),
|
||
logging.StreamHandler()
|
||
]
|
||
)
|
||
logger = logging.getLogger("winamp-mpris")
|
||
|
||
def authenticated_get(url, timeout=2):
|
||
"""Helper to perform GET with authentication fallback."""
|
||
global AUTH
|
||
try:
|
||
r = requests.get(url, auth=AUTH, timeout=timeout)
|
||
if r.status_code != 401:
|
||
return r
|
||
except requests.RequestException as e:
|
||
raise e
|
||
|
||
for cred in POSSIBLE_CREDS:
|
||
if cred == AUTH:
|
||
continue
|
||
try:
|
||
r = requests.get(url, auth=cred, timeout=timeout)
|
||
if r.status_code != 401:
|
||
logger.info(f"Authentication fallback successful: switched to {cred}")
|
||
AUTH = cred
|
||
return r
|
||
except requests.RequestException:
|
||
continue
|
||
return r
|
||
|
||
def write_pid_file():
|
||
try:
|
||
with open(PID_FILE, "w") as f:
|
||
f.write(str(os.getpid()))
|
||
except Exception as e:
|
||
logger.error(f"Failed to write PID file: {e}")
|
||
|
||
class WinampMPRIS:
|
||
"""
|
||
MPRIS2 specification implementation for Winamp Web Interface.
|
||
Optimized for Plasma 6 property change detection.
|
||
"""
|
||
dbus = f"""
|
||
<node>
|
||
<interface name="org.mpris.MediaPlayer2">
|
||
<property name="CanQuit" type="b" access="read" />
|
||
<property name="CanRaise" type="b" access="read" />
|
||
<property name="HasTrackList" type="b" access="read" />
|
||
<property name="Identity" type="s" access="read" />
|
||
<property name="DesktopEntry" type="s" access="read" />
|
||
<property name="SupportedUriSchemes" type="as" access="read" />
|
||
<property name="SupportedMimeTypes" type="as" access="read" />
|
||
</interface>
|
||
<interface name="org.mpris.MediaPlayer2.Player">
|
||
<method name="Next" />
|
||
<method name="Previous" />
|
||
<method name="Pause" />
|
||
<method name="PlayPause" />
|
||
<method name="Stop" />
|
||
<method name="Play" />
|
||
<method name="Seek">
|
||
<arg type="x" name="Offset" direction="in" />
|
||
</method>
|
||
<method name="SetPosition">
|
||
<arg type="o" name="TrackId" direction="in" />
|
||
<arg type="x" name="Position" direction="in" />
|
||
</method>
|
||
<method name="OpenUri">
|
||
<arg type="s" name="Uri" direction="in" />
|
||
</method>
|
||
<property name="PlaybackStatus" type="s" access="read">
|
||
<annotation name="org.freedesktop.DBus.Property.EmitsChangedSignal" value="true"/>
|
||
</property>
|
||
<property name="LoopStatus" type="s" access="readwrite">
|
||
<annotation name="org.freedesktop.DBus.Property.EmitsChangedSignal" value="true"/>
|
||
</property>
|
||
<property name="Rate" type="d" access="readwrite" />
|
||
<property name="Shuffle" type="b" access="readwrite">
|
||
<annotation name="org.freedesktop.DBus.Property.EmitsChangedSignal" value="true"/>
|
||
</property>
|
||
<property name="Metadata" type="a{{sv}}" access="read">
|
||
<annotation name="org.freedesktop.DBus.Property.EmitsChangedSignal" value="true"/>
|
||
</property>
|
||
<property name="CanGoNext" type="b" access="read" />
|
||
<property name="CanGoPrevious" type="b" access="read" />
|
||
<property name="CanPlay" type="b" access="read" />
|
||
<property name="CanPause" type="b" access="read" />
|
||
<property name="CanControl" type="b" access="read" />
|
||
<property name="CanSeek" type="b" access="read" />
|
||
<property name="Position" type="x" access="read">
|
||
<annotation name="org.freedesktop.DBus.Property.EmitsChangedSignal" value="false"/>
|
||
</property>
|
||
<property name="Volume" type="d" access="readwrite">
|
||
<annotation name="org.freedesktop.DBus.Property.EmitsChangedSignal" value="true"/>
|
||
</property>
|
||
</interface>
|
||
</node>
|
||
"""
|
||
|
||
# Signal for org.freedesktop.DBus.Properties.PropertiesChanged
|
||
PropertiesChanged = signal()
|
||
|
||
def __init__(self):
|
||
self._title = "Winamp"
|
||
self._artist = "Unknown"
|
||
self._album = ""
|
||
self._status = "Stopped"
|
||
self._art_url = DEFAULT_ART
|
||
self._last_position_us = 0
|
||
self._total_length_us = 0
|
||
self._last_update_ts = time.time()
|
||
self._shuffle = False
|
||
self._loop_status = "None"
|
||
self._volume = 1.0
|
||
|
||
def _request(self, endpoint):
|
||
logger.info(f"COMMAND RECEIVED: {endpoint}")
|
||
try:
|
||
r = authenticated_get(f"{BASE_URL}/{endpoint}", timeout=2)
|
||
if r.status_code == 401:
|
||
msg = "401 Unauthorized: All authentication attempts failed (winamp:llama, winamp:, anon). Check plugin config."
|
||
logger.warning(f"ERROR: {msg}")
|
||
subprocess.run(["notify-send", "-u", "critical", "-t", "10000", "Winamp Bridge Auth Error", msg])
|
||
elif r.status_code != 200:
|
||
msg = f"Failed to send '{endpoint}' to Winamp (Status {r.status_code})."
|
||
logger.error(f"ERROR: {msg}")
|
||
subprocess.run(["notify-send", "-u", "critical", "-t", "3000", "Winamp Bridge Error", msg])
|
||
except Exception as e:
|
||
msg = f"Connection error while sending '{endpoint}': {e}"
|
||
logger.warning(f"ERROR: {msg}")
|
||
subprocess.run(["notify-send", "-u", "critical", "-t", "3000", "Winamp Bridge Offline", msg])
|
||
|
||
# MPRIS Methods
|
||
def Next(self): self._request("next")
|
||
def Previous(self): self._request("prev")
|
||
def Pause(self): self._request("pause")
|
||
def PlayPause(self): self._request("pause")
|
||
def Play(self): self._request("play")
|
||
def Stop(self): self._request("stop")
|
||
def Seek(self, offset): pass
|
||
def SetPosition(self, track_id, position): pass
|
||
def OpenUri(self, uri):
|
||
# Found /url?p&<url> in source code - plays immediately
|
||
self._request(f"url?p&{uri}")
|
||
|
||
# --- Root Properties ---
|
||
@property
|
||
def CanQuit(self): return False
|
||
@property
|
||
def CanRaise(self): return False
|
||
@property
|
||
def HasTrackList(self): return False
|
||
@property
|
||
def Identity(self): return "Winamp Web Bridge"
|
||
@property
|
||
def DesktopEntry(self): return ""
|
||
@property
|
||
def SupportedUriSchemes(self): return []
|
||
@property
|
||
def SupportedMimeTypes(self): return []
|
||
|
||
# --- Player Properties ---
|
||
@property
|
||
def PlaybackStatus(self): return self._status
|
||
|
||
@property
|
||
def LoopStatus(self): return self._loop_status
|
||
@LoopStatus.setter
|
||
def LoopStatus(self, value):
|
||
# MPRIS: "None", "Track", "Playlist"
|
||
# Winamp Web Interface has "on" or "off" for Repeat (playlist-wide)
|
||
# We'll map "Playlist" and "Track" both to Repeat ON.
|
||
if value in ["Playlist", "Track"]:
|
||
self._request("playmode?repeat=on")
|
||
else:
|
||
self._request("playmode?repeat=off")
|
||
|
||
@property
|
||
def Rate(self): return 1.0
|
||
@Rate.setter
|
||
def Rate(self, value): pass
|
||
|
||
@property
|
||
def Shuffle(self): return self._shuffle
|
||
@Shuffle.setter
|
||
def Shuffle(self, value):
|
||
if value:
|
||
self._request("playmode?random=on")
|
||
else:
|
||
self._request("playmode?random=off")
|
||
|
||
@property
|
||
def CanGoNext(self): return True
|
||
@property
|
||
def CanGoPrevious(self): return True
|
||
@property
|
||
def CanPlay(self): return True
|
||
@property
|
||
def CanPause(self): return True
|
||
@property
|
||
def CanControl(self): return True
|
||
@property
|
||
def CanSeek(self): return False
|
||
|
||
@property
|
||
def Position(self):
|
||
if self._status == "Playing":
|
||
now = time.time()
|
||
elapsed_us = int((now - self._last_update_ts) * 1e6)
|
||
current_pos = self._last_position_us + elapsed_us
|
||
if self._total_length_us > 0:
|
||
return min(current_pos, self._total_length_us)
|
||
return current_pos
|
||
return self._last_position_us
|
||
|
||
@property
|
||
def Volume(self): return self._volume
|
||
@Volume.setter
|
||
def Volume(self, value):
|
||
self._volume = max(0.0, min(1.0, value))
|
||
# Plugin source shows /vol?volume= accepts 0-10
|
||
vol_10 = int(self._volume * 10)
|
||
self._request(f"vol?volume={vol_10}")
|
||
|
||
@property
|
||
def Metadata(self):
|
||
# Metadata is a{sv}, so values MUST be wrapped in GLib.Variant
|
||
return {
|
||
'mpris:trackid': GLib.Variant('o', '/org/mpris/MediaPlayer2/winamp/track/0'),
|
||
'mpris:length': GLib.Variant('x', self._total_length_us),
|
||
'xesam:title': GLib.Variant('s', self._title),
|
||
'xesam:artist': GLib.Variant('as', [self._artist]),
|
||
'xesam:album': GLib.Variant('s', self._album),
|
||
'mpris:artUrl': GLib.Variant('s', self._art_url)
|
||
}
|
||
|
||
def parse_time_to_us(time_str):
|
||
"""Parses MM:SS or H:MM:SS to microseconds."""
|
||
try:
|
||
parts = list(map(int, time_str.split(':')))
|
||
if len(parts) == 2: # MM:SS
|
||
return (parts[0] * 60 + parts[1]) * 1_000_000
|
||
elif len(parts) == 3: # H:MM:SS
|
||
return (parts[0] * 3600 + parts[1] * 60 + parts[2]) * 1_000_000
|
||
except Exception:
|
||
pass
|
||
return 0
|
||
|
||
def get_winamp_window_title():
|
||
"""Extracts the Winamp window title using wmctrl."""
|
||
try:
|
||
output = subprocess.check_output(["wmctrl", "-l"], stderr=subprocess.STDOUT).decode('utf-8')
|
||
for line in output.splitlines():
|
||
# Match the typical Winamp window title pattern
|
||
if " - Winamp" in line:
|
||
# wmctrl output: 0x03000001 0 Compy-686 ARTMS–<Dall>–Unf/Air | 0:06/2:45 - Winamp
|
||
parts = line.split(None, 3)
|
||
if len(parts) >= 4:
|
||
return parts[3]
|
||
except Exception:
|
||
pass
|
||
return None
|
||
|
||
def fetch_album_art(artist, album):
|
||
"""Fetches album art from iTunes Search API."""
|
||
if not artist or not album or artist == "Unknown":
|
||
return DEFAULT_ART
|
||
|
||
try:
|
||
query = f"{artist} {album}"
|
||
url = f"https://itunes.apple.com/search?term={query}&entity=album&limit=1"
|
||
r = requests.get(url, timeout=3)
|
||
if r.status_code == 200:
|
||
data = r.json()
|
||
if data.get("resultCount", 0) > 0:
|
||
art_100 = data["results"][0].get("artworkUrl100", DEFAULT_ART)
|
||
# Upgrade to 600x600 for better quality
|
||
return art_100.replace("100x100bb.jpg", "600x600bb.jpg")
|
||
except Exception as e:
|
||
print(f"Art fetch error: {e}")
|
||
return DEFAULT_ART
|
||
|
||
def update_loop(player):
|
||
last_known_title = ""
|
||
last_known_status = ""
|
||
last_known_album = ""
|
||
last_time_str = ""
|
||
last_known_shuffle = False
|
||
last_known_loop = "None"
|
||
offline_logged = False
|
||
auth_error_logged = False
|
||
|
||
while True:
|
||
try:
|
||
# 1. Try Window Title (Primary)
|
||
window_title = get_winamp_window_title()
|
||
title_parsed = False
|
||
|
||
if window_title:
|
||
# Pattern: Artist–Album–Title | 0:06/2:45 - Winamp
|
||
# Only use window title for metadata as time is unreliable here
|
||
match = re.search(r"(.*?) \| .* - Winamp", window_title)
|
||
if match:
|
||
metadata_raw = match.group(1).strip()
|
||
|
||
# Split metadata (Artist–Album–Title)
|
||
meta_parts = [p.strip() for p in re.split(r'[–—-]', metadata_raw)]
|
||
|
||
if len(meta_parts) >= 3:
|
||
player._artist = meta_parts[0]
|
||
player._album = meta_parts[1]
|
||
player._title = " – ".join(meta_parts[2:])
|
||
elif len(meta_parts) == 2:
|
||
player._artist = meta_parts[0]
|
||
player._title = meta_parts[1]
|
||
player._album = ""
|
||
else:
|
||
player._title = metadata_raw
|
||
player._artist = "Unknown"
|
||
player._album = ""
|
||
|
||
title_parsed = True
|
||
|
||
# 2. Poll Web UI (Status and Time source)
|
||
try:
|
||
r = authenticated_get(f"{BASE_URL}/main", timeout=1)
|
||
if r.status_code == 200:
|
||
offline_logged = False
|
||
auth_error_logged = False
|
||
soup = BeautifulSoup(r.text, 'html.parser')
|
||
p_tags = soup.find_all('p')
|
||
status_raw = p_tags[0].text if p_tags else ""
|
||
|
||
new_status = "Stopped"
|
||
if "Playing" in status_raw: new_status = "Playing"
|
||
elif "Paused" in status_raw: new_status = "Paused"
|
||
|
||
player._status = new_status
|
||
|
||
# Parse Repeat and Random status
|
||
# <p>Repeat is <b>off</b> ... <br>Random is <b>off</b> ... </p>
|
||
repeat_tag = None
|
||
for p in p_tags:
|
||
if "Repeat is" in p.text:
|
||
repeat_tag = p
|
||
break
|
||
|
||
if repeat_tag:
|
||
# Extract "on" or "off" status
|
||
text = repeat_tag.get_text()
|
||
is_repeat = False
|
||
is_random = False
|
||
|
||
m_rep = re.search(r"Repeat is\s*(?:<b>)?(\w+)(?:</b>)?", text, re.I)
|
||
if m_rep:
|
||
is_repeat = m_rep.group(1).lower() == "on"
|
||
|
||
m_rand = re.search(r"Random is\s*(?:<b>)?(\w+)(?:</b>)?", text, re.I)
|
||
if m_rand:
|
||
is_random = m_rand.group(1).lower() == "on"
|
||
|
||
player._loop_status = "Playlist" if is_repeat else "None"
|
||
player._shuffle = is_random
|
||
|
||
# Always update time from Web UI
|
||
match_time = re.search(r"\((\d+:?\d*:\d+) / (\d+:?\d*:\d+)\)", status_raw)
|
||
if match_time:
|
||
current_time_str = match_time.group(1)
|
||
if current_time_str != last_time_str:
|
||
player._last_position_us = parse_time_to_us(current_time_str)
|
||
player._total_length_us = parse_time_to_us(match_time.group(2))
|
||
player._last_update_ts = time.time()
|
||
last_time_str = current_time_str
|
||
|
||
if not title_parsed:
|
||
# Web UI Fallback for metadata
|
||
match_meta = re.search(r"track \d+ - (.*?) - (.*?) -", status_raw)
|
||
if match_meta:
|
||
player._artist = match_meta.group(1).strip()
|
||
player._title = match_meta.group(2).strip()
|
||
player._album = ""
|
||
elif "track" in status_raw:
|
||
player._title = status_raw.split('-')[0].replace("Playing track ", "").strip()
|
||
player._artist = "Unknown"
|
||
player._album = ""
|
||
elif r.status_code == 401:
|
||
if not auth_error_logged:
|
||
msg = "401 Unauthorized: All authentication attempts failed (winamp:llama, winamp:, anon). Check plugin config."
|
||
logger.warning(f"ERROR: {msg}")
|
||
subprocess.run(["notify-send", "-u", "critical", "-t", "10000", "Winamp Bridge Auth Error", msg])
|
||
auth_error_logged = True
|
||
player._status = "Stopped"
|
||
except requests.RequestException:
|
||
if not window_title:
|
||
if not offline_logged:
|
||
logger.info("Winamp Web Interface offline and no window found.")
|
||
offline_logged = True
|
||
player._status = "Stopped"
|
||
|
||
if (player._status != last_known_status or
|
||
player._title != last_known_title or
|
||
player._album != last_known_album or
|
||
player._shuffle != last_known_shuffle or
|
||
player._loop_status != last_known_loop):
|
||
|
||
# Fetch art if album/artist changed
|
||
if player._artist != "Unknown" and player._album:
|
||
player._art_url = fetch_album_art(player._artist, player._album)
|
||
else:
|
||
player._art_url = DEFAULT_ART
|
||
|
||
last_known_status = player._status
|
||
last_known_title = player._title
|
||
last_known_album = player._album
|
||
last_known_shuffle = player._shuffle
|
||
last_known_loop = player._loop_status
|
||
|
||
album_str = f" [{player._album}]" if player._album else ""
|
||
logger.info(f"UPDATE: [{player._status}] Shuffle: {player._shuffle}, Loop: {player._loop_status}, {player._artist}{album_str} - {player._title}")
|
||
|
||
player.PropertiesChanged(
|
||
"org.mpris.MediaPlayer2.Player",
|
||
{
|
||
"PlaybackStatus": player.PlaybackStatus,
|
||
"Metadata": player.Metadata,
|
||
"Shuffle": player.Shuffle,
|
||
"LoopStatus": player.LoopStatus,
|
||
"Volume": player.Volume
|
||
},
|
||
[]
|
||
)
|
||
|
||
except Exception as e:
|
||
logger.error(f"Update error: {e}")
|
||
|
||
time.sleep(1)
|
||
|
||
|
||
if __name__ == "__main__":
|
||
write_pid_file()
|
||
bus = SessionBus()
|
||
player_logic = WinampMPRIS()
|
||
bus.publish(APP_ID, ("/org/mpris/MediaPlayer2", player_logic))
|
||
|
||
GLib.timeout_add(1500, lambda: player_logic.PropertiesChanged(
|
||
"org.mpris.MediaPlayer2.Player",
|
||
{
|
||
"PlaybackStatus": player_logic.PlaybackStatus,
|
||
"Metadata": player_logic.Metadata
|
||
},
|
||
[]
|
||
))
|
||
|
||
thread = Thread(target=update_loop, args=(player_logic,), daemon=True)
|
||
thread.start()
|
||
|
||
logger.info(f"--- Winamp Bridge Started (Window Title + Web UI + Album Art) ---")
|
||
loop = GLib.MainLoop()
|
||
try:
|
||
loop.run()
|
||
except KeyboardInterrupt:
|
||
pass
|