1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
|
import asyncio
import hashlib
import logging
import time
from abc import ABC, abstractmethod
from click import launch
from rich.prompt import Prompt
from ..client import Client, QobuzClient, SoundcloudClient, TidalClient
from ..config import Config
from ..console import console
from ..exceptions import AuthenticationError, MissingCredentialsError
logger = logging.getLogger("streamrip")
class CredentialPrompter(ABC):
client: Client
def __init__(self, config: Config, client: Client):
self.config = config
self.client = self.type_check_client(client)
@abstractmethod
def has_creds(self) -> bool:
raise NotImplementedError
@abstractmethod
async def prompt_and_login(self):
"""Prompt for credentials in the appropriate way,
and save them to the configuration.
"""
raise NotImplementedError
@abstractmethod
def save(self):
"""Save current config to file"""
raise NotImplementedError
@abstractmethod
def type_check_client(self, client: Client):
raise NotImplementedError
class QobuzPrompter(CredentialPrompter):
client: QobuzClient
def has_creds(self) -> bool:
c = self.config.session.qobuz
return c.email_or_userid != "" and c.password_or_token != ""
async def prompt_and_login(self):
if not self.has_creds():
self._prompt_creds_and_set_session_config()
while True:
try:
await self.client.login()
break
except AuthenticationError:
console.print("[yellow]Invalid credentials, try again.")
self._prompt_creds_and_set_session_config()
except MissingCredentialsError:
self._prompt_creds_and_set_session_config()
def _prompt_creds_and_set_session_config(self):
email = Prompt.ask("Enter your Qobuz email")
pwd_input = Prompt.ask("Enter your Qobuz password (invisible)", password=True)
pwd = hashlib.md5(pwd_input.encode("utf-8")).hexdigest()
console.print(
f"[green]Credentials saved to config file at [bold cyan]{self.config.path}",
)
c = self.config.session.qobuz
c.use_auth_token = False
c.email_or_userid = email
c.password_or_token = pwd
def save(self):
c = self.config.session.qobuz
cf = self.config.file.qobuz
cf.use_auth_token = False
cf.email_or_userid = c.email_or_userid
cf.password_or_token = c.password_or_token
self.config.file.set_modified()
def type_check_client(self, client) -> QobuzClient:
assert isinstance(client, QobuzClient)
return client
class TidalPrompter(CredentialPrompter):
timeout_s: int = 600 # 5 mins to login
client: TidalClient
def has_creds(self) -> bool:
return len(self.config.session.tidal.access_token) > 0
async def prompt_and_login(self):
device_code, uri = await self.client._get_device_code()
login_link = f"https://{uri}"
console.print(
f"Go to [blue underline]{login_link}[/blue underline] to log into Tidal within 5 minutes.",
)
launch(login_link)
start = time.time()
elapsed = 0.0
info = {}
while elapsed < self.timeout_s:
elapsed = time.time() - start
status, info = await self.client._get_auth_status(device_code)
if status == 2:
# pending
await asyncio.sleep(4)
continue
elif status == 0:
# successful
break
else:
raise Exception
c = self.config.session.tidal
c.user_id = info["user_id"] # type: ignore
c.country_code = info["country_code"] # type: ignore
c.access_token = info["access_token"] # type: ignore
c.refresh_token = info["refresh_token"] # type: ignore
c.token_expiry = info["token_expiry"] # type: ignore
self.client._update_authorization_from_config()
self.client.logged_in = True
self.save()
def type_check_client(self, client) -> TidalClient:
assert isinstance(client, TidalClient)
return client
def save(self):
c = self.config.session.tidal
cf = self.config.file.tidal
cf.user_id = c.user_id
cf.country_code = c.country_code
cf.access_token = c.access_token
cf.refresh_token = c.refresh_token
cf.token_expiry = c.token_expiry
self.config.file.set_modified()
class SoundcloudPrompter(CredentialPrompter):
def has_creds(self) -> bool:
return True
async def prompt_and_login(self):
pass
def save(self):
pass
def type_check_client(self, client) -> SoundcloudClient:
assert isinstance(client, SoundcloudClient)
return client
PROMPTERS = {
"qobuz": QobuzPrompter,
"tidal": TidalPrompter,
"soundcloud": SoundcloudPrompter,
}
def get_prompter(client: Client, config: Config) -> CredentialPrompter:
"""Return an instance of a prompter."""
p = PROMPTERS[client.source]
return p(config, client)
|
