150 lines
5.3 KiB
Python
150 lines
5.3 KiB
Python
"""ICY (SHOUTcast/Icecast) metadata poller — reads live track titles from radio streams."""
|
|
import asyncio
|
|
import logging
|
|
import re
|
|
import ssl
|
|
from typing import Awaitable, Callable
|
|
from urllib.parse import urlparse
|
|
|
|
log = logging.getLogger(__name__)
|
|
|
|
_TITLE_RE = re.compile(r"StreamTitle='([^']*)';")
|
|
|
|
# Some servers send a "StreamUrl='...'" which we can ignore
|
|
_CHUNK = 8192
|
|
|
|
|
|
class IcyMetadataPoller:
|
|
"""Polls an Icecast/SHOUTcast stream for ICY metadata and fires a callback on title change.
|
|
|
|
Usage:
|
|
poller = IcyMetadataPoller("http://...", callback)
|
|
task = asyncio.create_task(poller.run())
|
|
...
|
|
poller.stop()
|
|
task.cancel()
|
|
"""
|
|
|
|
def __init__(
|
|
self,
|
|
url: str,
|
|
on_title: Callable[[str], Awaitable[None]],
|
|
):
|
|
self._url = url
|
|
self._on_title = on_title
|
|
self._stop = asyncio.Event()
|
|
self._last_title: str | None = None
|
|
|
|
# ── public API ──────────────────────────────────────────────────────────
|
|
|
|
async def run(self) -> None:
|
|
"""Main loop — connect, poll, reconnect on failure."""
|
|
while not self._stop.is_set():
|
|
try:
|
|
await self._poll_loop()
|
|
except asyncio.CancelledError:
|
|
return
|
|
except Exception:
|
|
log.debug(f"ICY poller disconnected, reconnecting in 10s: {self._url}")
|
|
# Back off before reconnecting
|
|
try:
|
|
await asyncio.wait_for(self._stop.wait(), timeout=10.0)
|
|
return # stop() was called
|
|
except asyncio.TimeoutError:
|
|
pass
|
|
|
|
def stop(self) -> None:
|
|
"""Signal the poller to stop (doesn't cancel the task)."""
|
|
self._stop.set()
|
|
|
|
# ── internals ───────────────────────────────────────────────────────────
|
|
|
|
async def _poll_loop(self) -> None:
|
|
parsed = urlparse(self._url)
|
|
host = parsed.hostname
|
|
port = parsed.port or (443 if parsed.scheme == "https" else 80)
|
|
path = parsed.path or "/"
|
|
if parsed.query:
|
|
path += "?" + parsed.query
|
|
|
|
# TLS for HTTPS
|
|
ssl_ctx = ssl.create_default_context() if parsed.scheme == "https" else None
|
|
|
|
reader, writer = await asyncio.open_connection(host, port, ssl=ssl_ctx)
|
|
try:
|
|
# Send HTTP GET with Icy-MetaData header
|
|
request = (
|
|
f"GET {path} HTTP/1.0\r\n"
|
|
f"Host: {host}\r\n"
|
|
f"Icy-MetaData: 1\r\n"
|
|
f"User-Agent: yuuka-radio/1.0\r\n"
|
|
f"\r\n"
|
|
)
|
|
writer.write(request.encode())
|
|
await writer.drain()
|
|
|
|
# Read status line + headers
|
|
headers = await self._read_headers(reader)
|
|
|
|
metaint = int(headers.get("icy-metaint", 0))
|
|
if metaint <= 0:
|
|
log.debug(f"No ICY metadata on {self._url}")
|
|
return # stream has no metadata support
|
|
|
|
# Poll: skip audio → read metadata length → parse title
|
|
while not self._stop.is_set():
|
|
await self._skip(reader, metaint)
|
|
|
|
length_byte = await reader.readexactly(1)
|
|
length = length_byte[0] * 16
|
|
if length > 0:
|
|
raw = await reader.readexactly(length)
|
|
title = self._extract_title(raw.decode("utf-8", errors="replace"))
|
|
if title and title != self._last_title:
|
|
self._last_title = title
|
|
try:
|
|
await self._on_title(title)
|
|
except Exception:
|
|
log.debug("ICY callback error", exc_info=True)
|
|
finally:
|
|
writer.close()
|
|
try:
|
|
await writer.wait_closed()
|
|
except Exception:
|
|
pass
|
|
|
|
@staticmethod
|
|
async def _read_headers(reader: asyncio.StreamReader) -> dict[str, str]:
|
|
"""Read HTTP response headers into a lowercase-keyed dict."""
|
|
headers: dict[str, str] = {}
|
|
while True:
|
|
line = await reader.readline()
|
|
if not line:
|
|
break
|
|
text = line.decode("utf-8", errors="replace").strip()
|
|
if not text:
|
|
break # empty line = end of headers
|
|
if ":" in text:
|
|
key, _, value = text.partition(":")
|
|
headers[key.strip().lower()] = value.strip()
|
|
return headers
|
|
|
|
@staticmethod
|
|
async def _skip(reader: asyncio.StreamReader, n: int) -> None:
|
|
"""Read and discard exactly *n* bytes from the stream."""
|
|
remaining = n
|
|
while remaining > 0:
|
|
chunk = await reader.read(min(remaining, _CHUNK))
|
|
if not chunk:
|
|
raise ConnectionError("ICY stream ended unexpectedly")
|
|
remaining -= len(chunk)
|
|
|
|
@staticmethod
|
|
def _extract_title(metadata: str) -> str | None:
|
|
"""Parse ``StreamTitle='Artist - Title';`` from an ICY metadata chunk."""
|
|
m = _TITLE_RE.search(metadata)
|
|
if not m:
|
|
return None
|
|
title = m.group(1).strip()
|
|
return title if title else None
|