Trixy7/plugins/music/sources/radio.py

# -*- coding: utf-8 -*-
"""
Radio-Stream-Quelle fuer HTTP Radio (Icecast/Shoutcast).

Spielt HTTP-Audio-Streams ab — kein Download, direktes Streaming.
Unterstuetzt ICY-Metadaten fuer Artist/Titel-Anzeige.

Keine externen Abhaengigkeiten — nutzt nur aiohttp (bereits
im Projekt) fuer HTTP-Requests.

Konfiguration ueber config.json:
    "radio": {
        "enabled": true,
        "favorites": [
            {"name": "SWR3", "url": "https://liveradio.swr.de/..."},
            {"name": "1LIVE", "url": "https://wdr-1live-live.icecast.wdr.de/..."}
        ]
    }
"""

from __future__ import annotations

import asyncio
import re
from pathlib import Path
from typing import AsyncIterator, Any

from trixy_core.music.sources.base import MusicSource, SourceType, SourceState, SourceInfo
from trixy_core.music.track import Track, TrackMetadata, TrackState
from trixy_core.utils.debug import pinfo, pdebug, perror, pwarn

# Optionale Abhaengigkeit
_AIOHTTP_AVAILABLE = False
try:
    import aiohttp
    _AIOHTTP_AVAILABLE = True
except ImportError:
    pass


# Vordefinierte deutsche Radio-Sender als Fallback
_DEFAULT_STATIONS: list[dict[str, str]] = [
    {"name": "SWR3", "url": "https://liveradio.swr.de/sw282p3/swr3/play.mp3", "genre": "Pop"},
    {"name": "1LIVE", "url": "https://wdr-1live-live.icecast.wdr.de/wdr/1live/live/mp3/128/stream.mp3", "genre": "Pop/Rock"},
    {"name": "WDR 2", "url": "https://wdr-wdr2-rheinland.icecast.wdr.de/wdr/wdr2/rheinland/mp3/128/stream.mp3", "genre": "Pop"},
    {"name": "Bayern 3", "url": "https://streams.br.de/bayern3_2.m3u", "genre": "Pop"},
    {"name": "NDR 2", "url": "https://icecast.ndr.de/ndr/ndr2/niedersachsen/mp3/128/stream.mp3", "genre": "Pop/Rock"},
    {"name": "Fritz", "url": "https://rbb-fritz-live.cast.addradio.de/rbb/fritz/live/mp3/128/stream.mp3", "genre": "Alternative"},
    {"name": "Radio BOB!", "url": "https://streams.radiobob.de/bob-live/mp3-192/mediaplayer", "genre": "Rock"},
    {"name": "Klassik Radio", "url": "https://stream.klassikradio.de/live/mp3-192/stream.klassikradio.de/", "genre": "Klassik"},
    {"name": "Deutschlandfunk", "url": "https://st01.sslstream.dlf.de/dlf/01/128/mp3/stream.mp3", "genre": "Nachrichten"},
    {"name": "FluxFM", "url": "https://streams.fluxfm.de/live/mp3-320/audio/", "genre": "Alternative/Indie"},
    {"name": "bigFM", "url": "https://streams.bigfm.de/bigfm-deutschland-128-mp3", "genre": "Pop/Dance"},
    {"name": "bigFM Charts", "url": "https://streams.bigfm.de/bigfm-charts-128-mp3", "genre": "Charts"},
    {"name": "bigFM Nitrox", "url": "https://streams.bigfm.de/bigfm-nitrox-128-mp3", "genre": "Dance/EDM"},
]


class RadioStation:
    """Ein einzelner Radio-Sender."""

    def __init__(
        self,
        name: str,
        url: str,
        genre: str = "",
        country: str = "DE",
        description: str = "",
    ) -> None:
        self.name = name
        self.url = url
        self.genre = genre
        self.country = country
        self.description = description

    def to_track(self) -> Track:
        """Konvertiert den Sender in ein Track-Objekt."""
        return Track(
            uri=f"radio:{self.name}",
            source_type="radio",
            source_id=self.name,
            metadata=TrackMetadata(
                title=self.name,
                artist="Live Radio",
                genre=self.genre,
                duration_ms=0,  # Live = keine Dauer
                custom={"country": self.country, "stream_url": self.url},
            ),
            state=TrackState.AVAILABLE,
            extra={
                "stream_url": self.url,
                "is_live": True,
                "genre": self.genre,
                "country": self.country,
                "description": self.description,
            },
        )


class RadioSource(MusicSource):
    """Radio-Stream-Quelle fuer HTTP Radio (Icecast/Shoutcast)."""

    def __init__(
        self,
        name: str = "Radio",
        favorites: list[dict[str, str]] | None = None,
        include_defaults: bool = True,
        buffer_size: int = 16384,
        icy_metadata: bool = True,
    ) -> None:
        """
        Initialisiert die Radio-Quelle.

        Args:
            name: Name der Quelle
            favorites: Liste von Favoriten-Sendern [{name, url, genre}]
            include_defaults: Standard-Sender hinzufuegen wenn keine Favoriten
            buffer_size: Stream-Buffer-Groesse in Bytes
            icy_metadata: ICY-Metadaten anfordern (Artist/Titel)
        """
        super().__init__(name=name, source_type=SourceType.URL)
        self._favorites = favorites or []
        self._include_defaults = include_defaults
        self._buffer_size = buffer_size
        self._icy_metadata = icy_metadata
        self._stations: list[RadioStation] = []
        self._current_metadata: dict[str, str] = {}

    @property
    def stations(self) -> list[RadioStation]:
        return list(self._stations)

    @property
    def station_count(self) -> int:
        return len(self._stations)

    @property
    def current_metadata(self) -> dict[str, str]:
        """Aktueller ICY-Metadaten-Stand (title, artist)."""
        return dict(self._current_metadata)

    async def connect(self) -> bool:
        """Laedt Sender-Liste und prueft Verfuegbarkeit."""
        if not _AIOHTTP_AVAILABLE:
            pwarn("RadioSource: aiohttp nicht installiert")
            self._state = SourceState.UNAVAILABLE
            return False

        self._stations = []

        # Favoriten laden
        for fav in self._favorites:
            name = fav.get("name", "")
            url = fav.get("url", "")
            if name and url:
                self._stations.append(RadioStation(
                    name=name,
                    url=url,
                    genre=fav.get("genre", ""),
                    country=fav.get("country", "DE"),
                    description=fav.get("description", ""),
                ))

        # Standard-Sender als Fallback
        if not self._stations and self._include_defaults:
            for default in _DEFAULT_STATIONS:
                self._stations.append(RadioStation(
                    name=default["name"],
                    url=default["url"],
                    genre=default.get("genre", ""),
                ))

        if not self._stations:
            pwarn("RadioSource: Keine Sender konfiguriert")
            self._state = SourceState.UNAVAILABLE
            return False

        # Tracks registrieren
        for station in self._stations:
            self._add_track(station.to_track())

        self._state = SourceState.AVAILABLE
        pinfo(f"RadioSource: {len(self._stations)} Sender geladen")
        return True

    async def disconnect(self) -> None:
        """Trennt die Verbindung."""
        self._stations = []
        self._clear_tracks()
        self._current_metadata = {}
        self._state = SourceState.UNAVAILABLE

    async def scan(self) -> int:
        """Gibt Anzahl der Sender zurueck."""
        return len(self._stations)

    async def get_track(self, track_id: str) -> Track | None:
        """Holt Track nach ID."""
        return self._tracks.get(track_id)

    async def get_metadata(self, track: Track) -> TrackMetadata | None:
        """Liefert Metadaten eines Tracks."""
        return track.metadata

    async def search(self, query: str, limit: int = 5) -> list[Track]:
        """
        Sucht in der Sender-Liste nach passenden Sendern.

        Durchsucht Name, Genre und Beschreibung.

        Args:
            query: Suchstring (z.B. "SWR3", "Rock", "Klassik")
            limit: Maximale Ergebnisse

        Returns:
            Liste passender Sender als Tracks
        """
        if self._state != SourceState.AVAILABLE:
            return []

        query_lower = query.lower()
        results: list[tuple[int, Track]] = []

        for station in self._stations:
            score = 0

            # Exakter Name-Match (hoechste Prioritaet)
            if query_lower == station.name.lower():
                score = 100
            # Name enthaelt Query
            elif query_lower in station.name.lower():
                score = 80
            # Genre-Match
            elif station.genre and query_lower in station.genre.lower():
                score = 60
            # Beschreibung
            elif station.description and query_lower in station.description.lower():
                score = 40
            # Wort-Teilmatch im Namen
            else:
                for word in query_lower.split():
                    if word in station.name.lower():
                        score = max(score, 30)

            if score > 0:
                results.append((score, station.to_track()))

        # Nach Score sortieren, Limit anwenden
        results.sort(key=lambda x: x[0], reverse=True)
        return [track for _, track in results[:limit]]

    async def prepare_track(self, track: Track) -> bool:
        """
        Radio-Streams brauchen keine Vorbereitung.

        Setzt den Track auf AVAILABLE — das eigentliche Streaming
        passiert in get_audio_stream().
        """
        stream_url = track.extra.get("stream_url", "")
        if not stream_url:
            track.state = TrackState.ERROR
            track.error_message = "Keine Stream-URL"
            return False

        track.state = TrackState.AVAILABLE
        return True

    async def get_audio_stream(self, track: Track) -> AsyncIterator[bytes]:
        """
        Streamt Audio-Daten direkt vom HTTP-Radio-Server.

        Unterstuetzt ICY-Metadaten fuer Live-Titel-Anzeige.
        Laeuft endlos bis der Consumer aufhoert zu lesen.
        """
        if not _AIOHTTP_AVAILABLE:
            return

        stream_url = track.extra.get("stream_url", "")
        if not stream_url:
            return

        headers = {"User-Agent": "Trixy/1.0"}
        if self._icy_metadata:
            headers["Icy-MetaData"] = "1"

        try:
            timeout = aiohttp.ClientTimeout(total=None, connect=10)
            async with aiohttp.ClientSession(timeout=timeout) as session:
                async with session.get(stream_url, headers=headers) as resp:
                    if resp.status != 200:
                        perror(f"RadioSource: HTTP {resp.status} fuer {stream_url}")
                        return

                    # ICY-Metadaten-Intervall ermitteln
                    icy_interval = int(resp.headers.get("icy-metaint", "0"))

                    track.state = TrackState.PLAYING

                    if icy_interval > 0 and self._icy_metadata:
                        async for chunk in self._read_with_icy(
                            resp, icy_interval, track
                        ):
                            yield chunk
                    else:
                        async for chunk in resp.content.iter_chunked(
                            self._buffer_size
                        ):
                            yield chunk

        except asyncio.CancelledError:
            pdebug(f"RadioSource: Stream abgebrochen: {track.metadata.title}")
        except Exception as e:
            pdebug(f"RadioSource: Stream-Fehler: {e}")
        finally:
            track.state = TrackState.AVAILABLE

    async def _read_with_icy(
        self,
        resp: Any,
        interval: int,
        track: Track,
    ) -> AsyncIterator[bytes]:
        """
        Liest HTTP-Stream mit ICY-Metadaten-Extraktion.

        ICY-Protokoll: Alle `interval` Bytes kommt ein Metadaten-Block:
        - 1 Byte: Laenge * 16
        - N Bytes: Metadaten-String (z.B. "StreamTitle='Artist - Title';")
        - Danach wieder `interval` Bytes Audio
        """
        buffer = b""

        async for raw_chunk in resp.content.iter_any():
            buffer += raw_chunk

            while len(buffer) >= interval:
                # Audio-Daten bis zum Metadaten-Marker
                audio_chunk = buffer[:interval]
                buffer = buffer[interval:]

                yield audio_chunk

                # Metadaten-Block lesen
                if len(buffer) < 1:
                    # Mehr Daten brauchen
                    break

                meta_len = buffer[0] * 16
                buffer = buffer[1:]

                if meta_len > 0:
                    if len(buffer) < meta_len:
                        # Noch nicht genug Daten — warten
                        # Metadaten-Byte zuruecklegen
                        buffer = bytes([meta_len // 16]) + buffer
                        break

                    meta_bytes = buffer[:meta_len]
                    buffer = buffer[meta_len:]

                    # Metadaten parsen
                    self._parse_icy_metadata(
                        meta_bytes.decode("utf-8", errors="replace"), track
                    )

    def _parse_icy_metadata(self, meta_str: str, track: Track) -> None:
        """
        Parst ICY-Metadaten-String.

        Format: StreamTitle='Artist - Title';StreamUrl='...';
        """
        match = re.search(r"StreamTitle='([^']*)'", meta_str)
        if not match:
            return

        stream_title = match.group(1).strip()
        if not stream_title:
            return

        # Artist - Title aufteilen
        artist = ""
        title = stream_title

        if " - " in stream_title:
            parts = stream_title.split(" - ", 1)
            artist = parts[0].strip()
            title = parts[1].strip()

        # Metadaten aktualisieren
        self._current_metadata = {"artist": artist, "title": title}

        # Track-Metadaten live aktualisieren
        if title:
            track.metadata.title = f"{track.extra.get('station_name', track.metadata.title)}"
            track.metadata.custom["now_playing_artist"] = artist
            track.metadata.custom["now_playing_title"] = title

        pdebug(f"RadioSource: ICY → {artist} - {title}" if artist else f"RadioSource: ICY → {title}")

    def get_info(self) -> SourceInfo:
        """Gibt Informationen ueber die Radio-Quelle zurueck."""
        return SourceInfo(
            source_type=self.source_type,
            name=self.name,
            description=f"{len(self._stations)} Sender konfiguriert",
            state=self._state,
            track_count=len(self._stations),
            supports_search=True,
            supports_metadata=True,
            supports_artwork=False,
            supports_streaming=True,
            metadata={
                "station_count": len(self._stations),
                "stations": [s.name for s in self._stations],
            },
        )

    async def check_stream(self, url: str, timeout: float = 5.0) -> bool:
        """
        Prueft ob ein Stream-URL erreichbar ist.

        Args:
            url: Stream-URL
            timeout: Timeout in Sekunden

        Returns:
            True wenn erreichbar
        """
        if not _AIOHTTP_AVAILABLE:
            return False

        try:
            client_timeout = aiohttp.ClientTimeout(total=timeout)
            async with aiohttp.ClientSession(timeout=client_timeout) as session:
                async with session.head(url) as resp:
                    return resp.status == 200
        except Exception:
            return False