/
/
/
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