/
/
/
1"""Helpers for creating/parsing URI's."""
2
3import asyncio
4import os
5import re
6from typing import Final
7
8from music_assistant_models.enums import MediaType
9from music_assistant_models.errors import InvalidProviderID, InvalidProviderURI
10from music_assistant_models.helpers import create_uri as create_uri_org
11
12base62_length22_id_pattern = re.compile(r"^[a-zA-Z0-9]{22}$")
13
14# plain stream URLs that resolve to the builtin provider, which takes the URL as its item_id
15BUILTIN_URL_SCHEMES: Final[tuple[str, ...]] = ("http://", "https://", "rtsp://", "rtmp://")
16
17# create alias to original create_uri function
18create_uri = create_uri_org
19
20
21def valid_base62_length22(item_id: str) -> bool:
22 """Validate Spotify style ID."""
23 return bool(base62_length22_id_pattern.match(item_id))
24
25
26def valid_id(provider: str, item_id: str) -> bool:
27 """Validate Provider ID."""
28 if provider == "spotify":
29 return valid_base62_length22(item_id)
30 return True
31
32
33async def parse_uri(uri: str, validate_id: bool = False) -> tuple[MediaType, str, str]: # noqa: PLR0915
34 """
35 Try to parse URI to Mass identifiers.
36
37 Returns Tuple: MediaType, provider_instance_id_or_domain, item_id
38 """
39 try:
40 if uri.startswith("https://open."):
41 # public share URL (e.g. Spotify or Qobuz, not sure about others)
42 # https://open.spotify.com/playlist/5lH9NjOeJvctAO92ZrKQNB?si=04a63c8234ac413e
43 provider_instance_id_or_domain = uri.split(".")[1]
44 media_type_str = uri.split("/")[3]
45 media_type = MediaType(media_type_str)
46 item_id = uri.split("/")[4].split("?", maxsplit=1)[0]
47 elif uri.startswith("https://tidal.com/browse/"):
48 # Tidal public share URL
49 # https://tidal.com/browse/track/123456
50 provider_instance_id_or_domain = "tidal"
51 media_type_str = uri.split("/")[4]
52 media_type = MediaType(media_type_str)
53 item_id = uri.split("/")[5].split("?", maxsplit=1)[0]
54 elif uri.startswith("https://music.apple.com/"):
55 # Apple Music share URL
56 # https://music.apple.com/{storefront}/{type}/{slug}/{id}
57 _apple_type_map = {
58 "station": MediaType.PLAYLIST,
59 "playlist": MediaType.PLAYLIST,
60 "album": MediaType.ALBUM,
61 "artist": MediaType.ARTIST,
62 "song": MediaType.TRACK,
63 }
64 parts = uri.rstrip("/").split("?")[0].split("/")
65 # parts: ['https:', '', 'music.apple.com', '{sf}', '{type}', '{slug}', '{id}']
66 # or: ['https:', '', 'music.apple.com', '{sf}', '{type}', '{id}'] (no slug)
67 # Track share links are album URLs with a ?i=<track_id> query param
68 query = uri.split("?", 1)[1] if "?" in uri else ""
69 track_id_from_query = next(
70 (p.split("=", 1)[1] for p in query.split("&") if p.startswith("i=")),
71 None,
72 )
73 if len(parts) >= 6:
74 apple_type = parts[4]
75 if apple_type == "album" and track_id_from_query:
76 provider_instance_id_or_domain = "apple_music"
77 media_type = MediaType.TRACK
78 item_id = track_id_from_query
79 elif apple_type in _apple_type_map:
80 item_id = parts[-1]
81 if not item_id:
82 raise KeyError
83 provider_instance_id_or_domain = "apple_music"
84 media_type = _apple_type_map[apple_type]
85 else:
86 raise KeyError
87 else:
88 raise KeyError
89 elif uri.startswith(("https://www.deezer.com/", "https://deezer.com/")):
90 # Deezer share URL
91 # https://www.deezer.com/track/123456
92 # https://www.deezer.com/en/track/123456 (with locale)
93 # https://deezer.com/album/789
94 _deezer_type_map = {
95 "track": MediaType.TRACK,
96 "album": MediaType.ALBUM,
97 "artist": MediaType.ARTIST,
98 "playlist": MediaType.PLAYLIST,
99 "show": MediaType.PODCAST,
100 "episode": MediaType.PODCAST_EPISODE,
101 }
102 parts = uri.rstrip("/").split("?")[0].split("/")
103 # Find the type segment by checking against the known map
104 deezer_type = None
105 deezer_id = None
106 for i, part in enumerate(parts):
107 if part in _deezer_type_map and i + 1 < len(parts):
108 deezer_type = part
109 deezer_id = parts[i + 1]
110 break
111 if deezer_type is None or not deezer_id or not deezer_id.isdigit():
112 raise KeyError
113 provider_instance_id_or_domain = "deezer"
114 media_type = _deezer_type_map[deezer_type]
115 item_id = deezer_id
116 elif uri.startswith(BUILTIN_URL_SCHEMES):
117 # Translate a plain URL to the builtin provider
118 provider_instance_id_or_domain = "builtin"
119 media_type = MediaType.UNKNOWN
120 item_id = uri
121 elif "://" in uri and len(uri.split("/")) >= 4:
122 # music assistant-style uri
123 # provider://media_type/item_id
124 provider_instance_id_or_domain, rest = uri.split("://", 1)
125 media_type_str, item_id = rest.split("/", 1)
126 media_type = MediaType(media_type_str)
127 elif ":" in uri and len(uri.split(":")) == 3:
128 # spotify new-style uri
129 provider_instance_id_or_domain, media_type_str, item_id = uri.split(":")
130 media_type = MediaType(media_type_str)
131 elif "/" in uri and await asyncio.to_thread(os.path.isfile, uri):
132 # Translate a local file (which is not from a file provider!) to the builtin provider
133 provider_instance_id_or_domain = "builtin"
134 media_type = MediaType.UNKNOWN
135 item_id = uri
136 else:
137 raise KeyError
138 except (TypeError, AttributeError, ValueError, KeyError) as err:
139 msg = f"Not a valid Music Assistant uri: {uri}"
140 raise InvalidProviderURI(msg) from err
141 if validate_id and not valid_id(provider_instance_id_or_domain, item_id):
142 msg = f"Invalid {provider_instance_id_or_domain} ID: {item_id} found in URI: {uri}"
143 raise InvalidProviderID(msg)
144 return (media_type, provider_instance_id_or_domain, item_id)
145