music-assistant-server

1"""Helpers for creating/parsing URI's."""
2
3import asyncio
4import os
5import re
6
7from music_assistant_models.enums import MediaType
8from music_assistant_models.errors import InvalidProviderID, InvalidProviderURI
9from music_assistant_models.helpers import create_uri as create_uri_org
10
11base62_length22_id_pattern = re.compile(r"^[a-zA-Z0-9]{22}$")
12
13# create alias to original create_uri function
14create_uri = create_uri_org
15
16
17def valid_base62_length22(item_id: str) -> bool:
18    """Validate Spotify style ID."""
19    return bool(base62_length22_id_pattern.match(item_id))
20
21
22def valid_id(provider: str, item_id: str) -> bool:
23    """Validate Provider ID."""
24    if provider == "spotify":
25        return valid_base62_length22(item_id)
26    return True
27
28
29async def parse_uri(uri: str, validate_id: bool = False) -> tuple[MediaType, str, str]:
30    """Try to parse URI to Mass identifiers.
31
32    Returns Tuple: MediaType, provider_instance_id_or_domain, item_id
33    """
34    try:
35        if uri.startswith("https://open."):
36            # public share URL (e.g. Spotify or Qobuz, not sure about others)
37            # https://open.spotify.com/playlist/5lH9NjOeJvctAO92ZrKQNB?si=04a63c8234ac413e
38            provider_instance_id_or_domain = uri.split(".")[1]
39            media_type_str = uri.split("/")[3]
40            media_type = MediaType(media_type_str)
41            item_id = uri.split("/")[4].split("?")[0]
42        elif uri.startswith("https://tidal.com/browse/"):
43            # Tidal public share URL
44            # https://tidal.com/browse/track/123456
45            provider_instance_id_or_domain = "tidal"
46            media_type_str = uri.split("/")[4]
47            media_type = MediaType(media_type_str)
48            item_id = uri.split("/")[5].split("?")[0]
49        elif uri.startswith(("http://", "https://", "rtsp://", "rtmp://")):
50            # Translate a plain URL to the builtin provider
51            provider_instance_id_or_domain = "builtin"
52            media_type = MediaType.UNKNOWN
53            item_id = uri
54        elif "://" in uri and len(uri.split("/")) >= 4:
55            # music assistant-style uri
56            # provider://media_type/item_id
57            provider_instance_id_or_domain, rest = uri.split("://", 1)
58            media_type_str, item_id = rest.split("/", 1)
59            media_type = MediaType(media_type_str)
60        elif ":" in uri and len(uri.split(":")) == 3:
61            # spotify new-style uri
62            provider_instance_id_or_domain, media_type_str, item_id = uri.split(":")
63            media_type = MediaType(media_type_str)
64        elif "/" in uri and await asyncio.to_thread(os.path.isfile, uri):
65            # Translate a local file (which is not from a file provider!) to the builtin provider
66            provider_instance_id_or_domain = "builtin"
67            media_type = MediaType.UNKNOWN
68            item_id = uri
69        else:
70            raise KeyError
71    except (TypeError, AttributeError, ValueError, KeyError) as err:
72        msg = f"Not a valid Music Assistant uri: {uri}"
73        raise InvalidProviderURI(msg) from err
74    if validate_id and not valid_id(provider_instance_id_or_domain, item_id):
75        msg = f"Invalid {provider_instance_id_or_domain} ID: {item_id} found in URI: {uri}"
76        raise InvalidProviderID(msg)
77    return (media_type, provider_instance_id_or_domain, item_id)
78