/
/
/
1"""
2Setup flow for the QQ Music provider.
3
4QQ Music has no official OAuth: authentication is a QR code the user scans with either
5the QQ or the WeChat mobile app. The flow first asks which app to use, then shows a QR
6code the user scans and confirms. Whenever the scan window elapses the QR code is minted
7afresh and the progress step is re-emitted in place, so the shown code never silently
8goes stale. The resulting credential is persisted as setup_data; the QR identifier/type
9and page URL that used to be smuggled through hidden config values now live only as
10locals for the duration of the flow.
11"""
12
13from __future__ import annotations
14
15import asyncio
16import base64
17from typing import TYPE_CHECKING
18
19from music_assistant_models.config_entries import ConfigEntry, ConfigValueOption
20from music_assistant_models.enums import ConfigEntryType
21from qqmusic_api import Client as QQClient
22from qqmusic_api.models.login import QRCodeLoginEvents, QRLoginType
23
24from music_assistant.models.setup_flow import AbortFlow, SetupFlowError, StepExpiredError
25
26from . import _store_credential
27
28if TYPE_CHECKING:
29 from music_assistant_models.config_entries import ConfigValueType
30 from qqmusic_api import Credential
31 from qqmusic_api.models.login import QR
32
33 from music_assistant.models.setup_flow import SetupSession
34
35# flow-local key for the QQ vs WeChat login method select (never persisted)
36CONF_METHOD = "method"
37METHOD_QQ = "qq"
38METHOD_WX = "wx"
39# scan window (seconds) a shown QR code is valid for before it is refreshed in place
40_QR_EXPIRES_IN = 120.0
41# delay (seconds) between poll attempts against the QR check endpoint
42_QR_POLL_INTERVAL = 1.5
43
44
45async def run_setup(session: SetupSession) -> None:
46 """
47 Run the QQ Music QR login flow: pick QQ/WeChat, scan a QR, store the credential.
48
49 :param session: The setup session driving the flow.
50 """
51 method_default = str(
52 session.context.setup_data.get(CONF_METHOD)
53 or session.context.values.get(CONF_METHOD)
54 or METHOD_QQ
55 )
56 client = QQClient()
57 try:
58 errors: dict[str, str] | None = None
59 while True:
60 values = await session.form(
61 [
62 ConfigEntry(
63 key=CONF_METHOD,
64 type=ConfigEntryType.STRING,
65 required=True,
66 default_value=METHOD_QQ,
67 value=method_default,
68 options=[
69 ConfigValueOption(METHOD_QQ),
70 ConfigValueOption(METHOD_WX),
71 ],
72 ),
73 ],
74 step_id="user",
75 errors=errors,
76 )
77 method = str(values[CONF_METHOD])
78 method_default = method
79 login_type = QRLoginType.WX if method == METHOD_WX else QRLoginType.QQ
80 credential = await _run_qr_login(session, client, login_type)
81 collected: dict[str, ConfigValueType] = {}
82 _store_credential(collected, credential)
83 try:
84 await session.finish(collected)
85 return
86 except SetupFlowError as err:
87 errors = {"base": err.translation_key or str(err)}
88 finally:
89 await client.close()
90
91
92async def _run_qr_login(
93 session: SetupSession, client: QQClient, login_type: QRLoginType
94) -> Credential:
95 """
96 Show a QR code and wait for the user to scan and confirm the login.
97
98 Mints a fresh QR code and re-emits the progress step whenever the scan window
99 expires, so the displayed code stays valid until the user completes the login.
100
101 :param session: The setup session driving the flow.
102 :param client: The QQ Music API client bound to this flow.
103 :param login_type: Whether to request a QQ or a WeChat QR login.
104 """
105 while True:
106 qr = await client.login.get_qrcode(login_type)
107 try:
108 return await session.progress_until(
109 _poll_qr_login(client, qr),
110 step_id="scan_qr",
111 image=_qr_data_uri(qr),
112 expires_in=_QR_EXPIRES_IN,
113 )
114 except StepExpiredError:
115 # the scan window elapsed (deadline) or the app reported the code expired:
116 # mint a fresh QR and re-emit the progress step in place
117 continue
118
119
120async def _poll_qr_login(client: QQClient, qr: QR) -> Credential:
121 """
122 Poll the QR check endpoint until the login is confirmed, returning the credential.
123
124 Raises StepExpiredError when the app reports the code expired, so the caller
125 refreshes the QR alongside the progress-step deadline; aborts the flow when the
126 user rejects the login in the app.
127
128 :param client: The QQ Music API client bound to this flow.
129 :param qr: The QR object returned by get_qrcode.
130 """
131 while True:
132 result = await client.login.check_qrcode(qr)
133 if result.event == QRCodeLoginEvents.DONE and result.credential:
134 return result.credential
135 if result.event == QRCodeLoginEvents.TIMEOUT:
136 raise StepExpiredError
137 if result.event == QRCodeLoginEvents.REFUSE:
138 raise AbortFlow("login_rejected")
139 # SCAN (awaiting scan) / CONF (scanned, awaiting confirm): keep polling
140 await asyncio.sleep(_QR_POLL_INTERVAL)
141
142
143def _qr_data_uri(qr: QR) -> str:
144 """
145 Build a base64 data-URI from a QR object, honoring its (per-method) mimetype.
146
147 :param qr: The QR object returned by get_qrcode.
148 """
149 encoded = base64.b64encode(bytes(qr.data)).decode("ascii")
150 return f"data:{qr.mimetype};base64,{encoded}"
151