Trixy7/trixy_core/music/sources/url.py

# -*- coding: utf-8 -*-
"""
URL-Streams als Musikquelle.

Unterstützt HTTP/HTTPS Audio-URLs und Internet-Radio-Streams.
"""

import asyncio
import logging
import re
from dataclasses import dataclass
from pathlib import Path
from typing import AsyncIterator
from urllib.parse import urlparse

from trixy_core.music.sources.base import MusicSource, SourceType, SourceState
from trixy_core.music.track import Track, TrackMetadata, TrackState


@dataclass
class URLTrackInfo:
    """Informationen für URL-basierte Tracks."""
    url: str
    name: str = ""
    description: str = ""
    content_type: str = ""
    content_length: int = 0
    is_stream: bool = False  # True für Live-Streams (z.B. Radio)


class URLSource(MusicSource):
    """
    URL-basierte Musikquelle.

    Unterstützt:
    - Direkte Audio-URLs (MP3, OGG, etc.)
    - Internet-Radio-Streams
    - Podcast-URLs
    """

    # Unterstützte Content-Types
    SUPPORTED_CONTENT_TYPES = {
        "audio/mpeg": "mp3",
        "audio/mp3": "mp3",
        "audio/ogg": "ogg",
        "audio/vorbis": "ogg",
        "audio/opus": "opus",
        "audio/flac": "flac",
        "audio/wav": "wav",
        "audio/x-wav": "wav",
        "audio/aac": "aac",
        "audio/mp4": "m4a",
        "audio/x-m4a": "m4a",
        "application/ogg": "ogg",
        "audio/x-mpegurl": "m3u",
        "application/x-mpegURL": "m3u",
        "audio/x-scpls": "pls",
    }

    def __init__(
        self,
        name: str = "URL-Streams",
        timeout: float = 30.0,
        chunk_size: int = 65536,
    ) -> None:
        """
        Initialisiert die URL-Quelle.

        Args:
            name: Name der Quelle
            timeout: Timeout für HTTP-Anfragen (Sekunden)
            chunk_size: Chunk-Größe für Streaming (Bytes)
        """
        super().__init__(name=name, source_type=SourceType.URL)
        self.timeout = timeout
        self.chunk_size = chunk_size
        self.logger = logging.getLogger(__name__)

        # Registrierte URLs
        self._urls: dict[str, URLTrackInfo] = {}

    # ==========================================================================
    # URL-Management
    # ==========================================================================

    async def add_url(
        self,
        url: str,
        name: str = "",
        verify: bool = True,
    ) -> Track | None:
        """
        Fügt eine URL als Track hinzu.

        Args:
            url: Audio-URL
            name: Optionaler Name
            verify: URL validieren

        Returns:
            Erstellter Track oder None bei Fehler
        """
        if verify:
            info = await self._verify_url(url)
            if not info:
                return None
        else:
            info = URLTrackInfo(url=url, name=name or self._extract_name(url))

        if not info.name:
            info.name = name or self._extract_name(url)

        # Track erstellen
        metadata = TrackMetadata(
            title=info.name,
            format=info.content_type or "stream",
        )

        if info.content_length > 0:
            # Geschätzte Dauer (grob: 128kbps)
            metadata.duration_ms = int((info.content_length / 16000) * 1000)

        track = Track(
            uri=url,
            source_type="url",
            source_id=url,
            metadata=metadata,
            state=TrackState.AVAILABLE,
        )

        self._add_track(track)
        self._urls[track.id] = info

        self.logger.debug(f"URL hinzugefügt: {info.name}")
        return track

    async def add_urls(self, urls: list[str]) -> list[Track]:
        """
        Fügt mehrere URLs hinzu.

        Args:
            urls: Liste von Audio-URLs

        Returns:
            Liste der erstellten Tracks
        """
        tracks = []
        for url in urls:
            track = await self.add_url(url, verify=True)
            if track:
                tracks.append(track)
        return tracks

    def remove_url(self, track_id: str) -> bool:
        """
        Entfernt eine URL.

        Args:
            track_id: Track-ID

        Returns:
            True wenn entfernt
        """
        if track_id in self._tracks:
            self._remove_track(track_id)
            self._urls.pop(track_id, None)
            return True
        return False

    async def _verify_url(self, url: str) -> URLTrackInfo | None:
        """
        Verifiziert eine URL und holt Metadaten.

        Args:
            url: Zu prüfende URL

        Returns:
            URL-Info oder None bei Fehler
        """
        try:
            import aiohttp

            async with aiohttp.ClientSession() as session:
                async with session.head(
                    url,
                    timeout=aiohttp.ClientTimeout(total=self.timeout),
                    allow_redirects=True,
                ) as response:
                    if response.status != 200:
                        self.logger.warning(f"URL nicht erreichbar: {url} ({response.status})")
                        return None

                    content_type = response.headers.get("Content-Type", "").split(";")[0].strip()
                    content_length = int(response.headers.get("Content-Length", 0))

                    # Prüfen ob unterstützter Content-Type
                    if content_type not in self.SUPPORTED_CONTENT_TYPES:
                        # Versuche Dateiendung
                        parsed = urlparse(url)
                        ext = Path(parsed.path).suffix.lower()
                        if ext not in {".mp3", ".ogg", ".flac", ".wav", ".m4a", ".aac"}:
                            self.logger.warning(f"Nicht unterstützter Content-Type: {content_type}")
                            return None

                    # Stream erkennen (kein Content-Length = wahrscheinlich Stream)
                    is_stream = content_length == 0 and "icy" in response.headers.get("Server", "").lower()

                    return URLTrackInfo(
                        url=url,
                        name=self._extract_name(url),
                        content_type=self.SUPPORTED_CONTENT_TYPES.get(content_type, ""),
                        content_length=content_length,
                        is_stream=is_stream,
                    )

        except ImportError:
            self.logger.error("aiohttp nicht installiert")
        except asyncio.TimeoutError:
            self.logger.warning(f"Timeout bei URL-Prüfung: {url}")
        except Exception as e:
            self.logger.error(f"URL-Prüfung fehlgeschlagen: {e}")

        return None

    def _extract_name(self, url: str) -> str:
        """Extrahiert Namen aus URL."""
        parsed = urlparse(url)
        path = Path(parsed.path)

        if path.stem:
            # Dateiname ohne Extension
            name = path.stem
            # Underscores/Bindestriche durch Leerzeichen ersetzen
            name = re.sub(r"[-_]", " ", name)
            return name.title()

        # Hostname als Fallback
        return parsed.netloc or "Unbekannte URL"

    # ==========================================================================
    # MusicSource Implementation
    # ==========================================================================

    async def connect(self) -> bool:
        """Verbindet zur Quelle (keine echte Verbindung nötig)."""
        self._state = SourceState.AVAILABLE
        return True

    async def disconnect(self) -> None:
        """Trennt die Verbindung."""
        self._state = SourceState.UNAVAILABLE

    async def scan(self) -> int:
        """
        Scannt ist bei URL-Quelle nicht sinnvoll.

        Verifiziert stattdessen vorhandene URLs.

        Returns:
            Anzahl der verfügbaren Tracks
        """
        # Alle URLs verifizieren
        to_remove = []

        for track_id, info in self._urls.items():
            verified = await self._verify_url(info.url)
            if not verified:
                to_remove.append(track_id)
            else:
                # Track-Zustand aktualisieren
                track = self._tracks.get(track_id)
                if track:
                    track.state = TrackState.AVAILABLE

        # Nicht mehr erreichbare entfernen
        for track_id in to_remove:
            self._remove_track(track_id)
            self._urls.pop(track_id, None)

        self._state = SourceState.AVAILABLE
        return self.track_count

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

    async def get_audio_stream(self, track: Track) -> AsyncIterator[bytes]:
        """
        Liefert Audio-Daten als Stream.

        Args:
            track: Track zum Streamen

        Yields:
            Audio-Daten in Chunks
        """
        url = track.uri

        try:
            import aiohttp

            async with aiohttp.ClientSession() as session:
                async with session.get(
                    url,
                    timeout=aiohttp.ClientTimeout(
                        total=None,  # Kein Gesamt-Timeout für Streams
                        connect=self.timeout,
                        sock_read=self.timeout,
                    ),
                ) as response:
                    if response.status != 200:
                        self.logger.error(f"Stream-Fehler: {response.status}")
                        return

                    async for chunk in response.content.iter_chunked(self.chunk_size):
                        yield chunk

        except ImportError:
            self.logger.error("aiohttp nicht installiert")
        except asyncio.TimeoutError:
            self.logger.warning(f"Stream-Timeout: {url}")
        except asyncio.CancelledError:
            self.logger.debug(f"Stream abgebrochen: {url}")
        except Exception as e:
            self.logger.error(f"Stream-Fehler: {e}")

    async def get_metadata(self, track: Track) -> TrackMetadata | None:
        """
        Lädt Metadaten für einen Track.

        Bei URL-Streams begrenzt, da keine ID3-Tags verfügbar.
        """
        info = self._urls.get(track.id)
        if not info:
            return track.metadata

        # ICY-Metadaten für Radio-Streams abrufen
        if info.is_stream:
            icy_meta = await self._get_icy_metadata(track.uri)
            if icy_meta:
                # Titel aus StreamTitle parsen
                title = icy_meta.get("StreamTitle", "")
                if " - " in title:
                    artist, song = title.split(" - ", 1)
                    track.metadata.artist = artist.strip()
                    track.metadata.title = song.strip()
                else:
                    track.metadata.title = title

        return track.metadata

    async def _get_icy_metadata(self, url: str) -> dict | None:
        """
        Holt ICY-Metadaten von Radio-Streams.

        Args:
            url: Stream-URL

        Returns:
            Metadaten-Dict oder None
        """
        try:
            import aiohttp

            headers = {"Icy-MetaData": "1"}

            async with aiohttp.ClientSession() as session:
                async with session.get(
                    url,
                    headers=headers,
                    timeout=aiohttp.ClientTimeout(total=5),
                ) as response:
                    # ICY-Metadaten aus Header
                    icy_name = response.headers.get("icy-name", "")
                    icy_genre = response.headers.get("icy-genre", "")
                    icy_url = response.headers.get("icy-url", "")

                    return {
                        "name": icy_name,
                        "genre": icy_genre,
                        "url": icy_url,
                    }

        except Exception:
            pass
        return None

    # ==========================================================================
    # Playlist-Support
    # ==========================================================================

    async def load_playlist(self, url: str) -> list[Track]:
        """
        Lädt Playlist von URL (M3U, PLS).

        Args:
            url: Playlist-URL

        Returns:
            Liste der Tracks
        """
        tracks = []

        try:
            import aiohttp

            async with aiohttp.ClientSession() as session:
                async with session.get(url, timeout=aiohttp.ClientTimeout(total=self.timeout)) as response:
                    if response.status != 200:
                        return tracks

                    content = await response.text()
                    urls = self._parse_playlist(content, url)

                    for stream_url in urls:
                        track = await self.add_url(stream_url, verify=True)
                        if track:
                            tracks.append(track)

        except Exception as e:
            self.logger.error(f"Playlist-Laden fehlgeschlagen: {e}")

        return tracks

    def _parse_playlist(self, content: str, base_url: str) -> list[str]:
        """
        Parst Playlist-Inhalt.

        Args:
            content: Playlist-Inhalt
            base_url: Basis-URL für relative Pfade

        Returns:
            Liste von Stream-URLs
        """
        urls = []
        lines = content.strip().split("\n")

        for line in lines:
            line = line.strip()

            # Kommentare und leere Zeilen überspringen
            if not line or line.startswith("#"):
                continue

            # PLS-Format: File1=http://...
            if "=" in line and line.split("=")[0].startswith("File"):
                url = line.split("=", 1)[1].strip()
                urls.append(url)
            elif line.startswith("http://") or line.startswith("https://"):
                urls.append(line)

        return urls

    def get_info(self):
        """Liefert Informationen."""
        info = super().get_info()
        info.description = f"{self.track_count} URL-Streams"
        info.supports_streaming = True
        return info