/
/
/
1"""YouSee Musik authentication manager."""
2
3import re
4import time
5from typing import TYPE_CHECKING
6
7from music_assistant.constants import CONF_PASSWORD, CONF_USERNAME
8from music_assistant.helpers.util import (
9 lock,
10 try_parse_int,
11)
12from music_assistant.providers.yousee.api_client import JsonLike
13
14if TYPE_CHECKING:
15 from music_assistant.providers.yousee.provider import YouSeeMusikProvider
16
17
18class YouSeeAccessToken:
19 """YouSee Musik access token wrapper."""
20
21 def __init__(self, access_token: str) -> None:
22 """Initialize YouSeeAccessToken."""
23 self._access_token = access_token
24 self._token_parts = self._parse_access_token(access_token)
25
26 def is_expired(self) -> bool:
27 """Return True if token is expired."""
28 expires_at = try_parse_int(self._token_parts.get("ExpiresOn", 0))
29 return not expires_at or expires_at <= time.time()
30
31 def _parse_access_token(self, token: str) -> JsonLike:
32 return dict(part.split("=", 1) for part in token.split("&") if "=" in part)
33
34 def __str__(self) -> str:
35 """Return string representation of the access token."""
36 return self._access_token
37
38
39class YouSeeAuthManager:
40 """YouSee Musik authentication manager."""
41
42 def __init__(self, provider: "YouSeeMusikProvider"):
43 """Initialize YouSeeAuthManager."""
44 self._access_token: YouSeeAccessToken | None = None
45 self._refresh_token: str | None = None
46 self.mass = provider.mass
47 self.provider = provider
48 self.logger = provider.logger
49
50 def invalidate(self) -> None:
51 """Invalidate current access token."""
52 self._access_token = None
53
54 @lock
55 async def auth_token(self) -> YouSeeAccessToken | None:
56 """Authenticate and return access token."""
57 if self._access_token and not self._access_token.is_expired():
58 return self._access_token
59
60 # Try refresh token flow first
61 if self._refresh_token:
62 self.logger.debug("Trying to fetch refresh token")
63
64 async with self.mass.http_session.post(
65 "https://musik.yousee.dk/api/token", data={"refresh_token": self._refresh_token}
66 ) as refresh_response:
67 refresh_result = await refresh_response.json()
68 if refresh_result.get("status", 4) == 0:
69 access_token = refresh_result["tokenResult"]["access_token"]
70
71 self.logger.debug("Refresh token flow success")
72 self._access_token = YouSeeAccessToken(access_token)
73 self._refresh_token = refresh_result["tokenResult"]["refresh_token"]
74 return self._access_token
75
76 async with (
77 self.mass.http_session.get(
78 "https://musik.yousee.dk/api/delegatedlogin"
79 ) as delegate_response,
80 ):
81 post_action_re = re.search('action="([^"]+)"', await delegate_response.text())
82 if not post_action_re:
83 return None
84
85 cookies = delegate_response.cookies
86
87 async with self.mass.http_session.post(
88 f"https://login.yousee.dk{post_action_re.group(1)}",
89 data={
90 "pf.username": self.provider.config.get_value(CONF_USERNAME),
91 "pf.pass": self.provider.config.get_value(CONF_PASSWORD),
92 "pf.ok": "clicked",
93 "pf.adapterId": "MusicUsernamePasswordAdapter",
94 },
95 cookies=cookies,
96 ) as login_response:
97 access_token_re = re.search(
98 r'localStorage.setItem\("accesstoken", "([^"]+)"',
99 await login_response.text(),
100 )
101
102 refresh_token_re = re.search(
103 r'localStorage.setItem\("refreshtoken", "([^"]+)"',
104 await login_response.text(),
105 )
106
107 if not access_token_re or not refresh_token_re:
108 return None
109
110 access_token = access_token_re.group(1)
111 self._refresh_token = refresh_token_re.group(1)
112
113 self._access_token = YouSeeAccessToken(access_token)
114 self.logger.debug("Got new auth token")
115
116 return self._access_token
117