music-assistant-server

1"""Various helpers and utils for the DLNA Player Provider."""
2
3from __future__ import annotations
4
5import xml.etree.ElementTree as ET
6from typing import TYPE_CHECKING
7
8from aiohttp.web import Request, Response
9from async_upnp_client.const import HttpRequest
10from async_upnp_client.event_handler import UpnpEventHandler, UpnpNotifyServer
11
12if TYPE_CHECKING:
13    from async_upnp_client.client import UpnpRequester
14
15    from music_assistant import MusicAssistant
16
17
18class DLNANotifyServer(UpnpNotifyServer):  # type: ignore[misc,unused-ignore]
19    """Notify server for async_upnp_client which uses the MA webserver."""
20
21    def __init__(
22        self,
23        requester: UpnpRequester,
24        mass: MusicAssistant,
25    ) -> None:
26        """Initialize."""
27        self.mass = mass
28        self.event_handler = UpnpEventHandler(self, requester)
29        self.mass.streams.register_dynamic_route("/notify", self._handle_request, method="NOTIFY")
30
31    @property
32    def callback_url(self) -> str:
33        """Return callback URL on which we are callable."""
34        return f"{self.mass.streams.base_url}/notify"
35
36    async def _handle_request(self, request: Request) -> Response:
37        """Handle incoming requests."""
38        if request.method != "NOTIFY":
39            return Response(status=405)
40
41        # Some DLNA devices (e.g. Denon HEOS) send NOTIFY bodies that are not
42        # valid UTF-8 when track metadata contains non-ASCII characters in the
43        # device's native encoding. Decode leniently so we don't drop the event.
44        body_bytes = await request.read()
45        body = body_bytes.decode("utf-8", errors="replace")
46
47        # transform aiohttp request to async_upnp_client request
48        http_request = HttpRequest(
49            method=request.method,
50            url=str(request.url),
51            headers=request.headers,
52            body=body,
53        )
54
55        try:
56            status = await self.event_handler.handle_notify(http_request)
57        except ET.ParseError as err:
58            self.mass.logger.debug(
59                "Ignoring malformed XML in DLNA notify from %s: %s",
60                request.remote,
61                err,
62            )
63            return Response(status=400)
64
65        return Response(status=status)
66