1from __future__ import annotations
  2
  3import base64
  4import hashlib
  5import hmac
  6import logging
  7import math
  8import secrets
  9import string
 10import time
 11from dataclasses import dataclass
 12
 13import aiohttp
 14from aiohttp import ContentTypeError, FormData
 15from pyrate_limiter import BucketFullException, Duration, Limiter, Rate
 16
 17from roborock import HomeDataSchedule
 18from roborock.data import HomeData, HomeDataRoom, HomeDataScene, ProductResponse, RRiot, UserData
 19from roborock.exceptions import (
 20    RoborockAccountDoesNotExist,
 21    RoborockException,
 22    RoborockInvalidCode,
 23    RoborockInvalidCredentials,
 24    RoborockInvalidEmail,
 25    RoborockInvalidUserAgreement,
 26    RoborockMissingParameters,
 27    RoborockNoResponseFromBaseURL,
 28    RoborockNoUserAgreement,
 29    RoborockRateLimit,
 30    RoborockTooFrequentCodeRequests,
 31)
 32
 33_LOGGER = logging.getLogger(__name__)
 34BASE_URLS = [
 35    "https://usiot.roborock.com",
 36    "https://euiot.roborock.com",
 37    "https://cniot.roborock.com",
 38    "https://ruiot.roborock.com",
 39]
 40
 41
 42@dataclass
 43class IotLoginInfo:
 44    """Information about the login to the iot server."""
 45
 46    base_url: str
 47    country_code: str
 48    country: str
 49
 50
 51class RoborockApiClient:
 52    _LOGIN_RATES = [
 53        Rate(1, Duration.SECOND),
 54        Rate(3, Duration.MINUTE),
 55        Rate(10, Duration.HOUR),
 56        Rate(20, Duration.DAY),
 57    ]
 58    _HOME_DATA_RATES = [
 59        Rate(1, Duration.SECOND),
 60        Rate(3, Duration.MINUTE),
 61        Rate(5, Duration.HOUR),
 62        Rate(40, Duration.DAY),
 63    ]
 64
 65    _login_limiter = Limiter(_LOGIN_RATES, max_delay=1000)
 66    _home_data_limiter = Limiter(_HOME_DATA_RATES)
 67
 68    def __init__(
 69        self, username: str, base_url: str | None = None, session: aiohttp.ClientSession | None = None
 70    ) -> None:
 71        """Sample API Client."""
 72        self._username = username
 73        self._base_url = base_url
 74        self._device_identifier = secrets.token_urlsafe(16)
 75        self.session = session
 76        self._iot_login_info: IotLoginInfo | None = None
 77        self._base_urls = BASE_URLS if base_url is None else [base_url]
 78
 79    async def _get_iot_login_info(self) -> IotLoginInfo:
 80        if self._iot_login_info is None:
 81            for iot_url in self._base_urls:
 82                url_request = PreparedRequest(iot_url, self.session)
 83                response = await url_request.request(
 84                    "post",
 85                    "/api/v1/getUrlByEmail",
 86                    params={"email": self._username, "needtwostepauth": "false"},
 87                )
 88                if response is None:
 89                    continue
 90                response_code = response.get("code")
 91                if response_code != 200:
 92                    if response_code == 2003:
 93                        raise RoborockInvalidEmail("Your email was incorrectly formatted.")
 94                    elif response_code == 1001:
 95                        raise RoborockMissingParameters(
 96                            "You are missing parameters for this request, are you sure you entered your username?"
 97                        )
 98                    else:
 99                        raise RoborockException(f"{response.get('msg')} - response code: {response_code}")
100                country_code = response["data"]["countrycode"]
101                country = response["data"]["country"]
102                if country_code is not None or country is not None:
103                    self._iot_login_info = IotLoginInfo(
104                        base_url=response["data"]["url"],
105                        country=country,
106                        country_code=country_code,
107                    )
108                    _LOGGER.debug("Country determined to be %s and code is %s", country, country_code)
109                    return self._iot_login_info
110            raise RoborockNoResponseFromBaseURL(
111                "No account was found for any base url we tried. Either your email is incorrect or we do not have a"
112                " record of the roborock server your device is on."
113            )
114        return self._iot_login_info
115
116    @property
117    async def base_url(self):
118        if self._base_url is not None:
119            return self._base_url
120        return (await self._get_iot_login_info()).base_url
121
122    @property
123    async def country(self):
124        return (await self._get_iot_login_info()).country
125
126    @property
127    async def country_code(self):
128        return (await self._get_iot_login_info()).country_code
129
130    def _get_header_client_id(self):
131        md5 = hashlib.md5()
132        md5.update(self._username.encode())
133        md5.update(self._device_identifier.encode())
134        return base64.b64encode(md5.digest()).decode()
135
136    async def nc_prepare(self, user_data: UserData, timezone: str) -> dict:
137        """This gets a few critical parameters for adding a device to your account."""
138        if (
139            user_data.rriot is None
140            or user_data.rriot.r is None
141            or user_data.rriot.u is None
142            or user_data.rriot.r.a is None
143        ):
144            raise RoborockException("Your userdata is missing critical attributes.")
145        base_url = user_data.rriot.r.a
146        prepare_request = PreparedRequest(base_url, self.session)
147        hid = await self._get_home_id(user_data)
148
149        data = FormData()
150        data.add_field("hid", hid)
151        data.add_field("tzid", timezone)
152
153        prepare_response = await prepare_request.request(
154            "post",
155            "/nc/prepare",
156            headers={
157                "Authorization": _get_hawk_authentication(
158                    user_data.rriot, "/nc/prepare", {"hid": hid, "tzid": timezone}
159                ),
160            },
161            data=data,
162        )
163
164        if prepare_response is None:
165            raise RoborockException("prepare_response is None")
166        if not prepare_response.get("success"):
167            raise RoborockException(f"{prepare_response.get('msg')} - response code: {prepare_response.get('code')}")
168
169        return prepare_response["result"]
170
171    async def add_device(self, user_data: UserData, s: str, t: str) -> dict:
172        """This will add a new device to your account
173        it is recommended to only use this during a pairing cycle with a device.
174        Please see here: https://github.com/Python-roborock/Roborockmitmproxy/blob/main/handshake_protocol.md
175        """
176        if (
177            user_data.rriot is None
178            or user_data.rriot.r is None
179            or user_data.rriot.u is None
180            or user_data.rriot.r.a is None
181        ):
182            raise RoborockException("Your userdata is missing critical attributes.")
183        base_url = user_data.rriot.r.a
184        add_device_request = PreparedRequest(base_url, self.session)
185
186        add_device_response = await add_device_request.request(
187            "GET",
188            "/user/devices/newadd",
189            headers={
190                "Authorization": _get_hawk_authentication(
191                    user_data.rriot, "/user/devices/newadd", params={"s": s, "t": t}
192                ),
193            },
194            params={"s": s, "t": t},
195        )
196
197        if add_device_response is None:
198            raise RoborockException("add_device is None")
199        if not add_device_response.get("success"):
200            raise RoborockException(
201                f"{add_device_response.get('msg')} - response code: {add_device_response.get('code')}"
202            )
203
204        return add_device_response["result"]
205
206    async def request_code(self) -> None:
207        try:
208            await self._login_limiter.try_acquire_async("login")
209        except BucketFullException as ex:
210            _LOGGER.info(ex.meta_info)
211            raise RoborockRateLimit("Reached maximum requests for login. Please try again later.") from ex
212        base_url = await self.base_url
213        header_clientid = self._get_header_client_id()
214        code_request = PreparedRequest(base_url, self.session, {"header_clientid": header_clientid})
215
216        code_response = await code_request.request(
217            "post",
218            "/api/v1/sendEmailCode",
219            params={
220                "username": self._username,
221                "type": "auth",
222            },
223        )
224        if code_response is None:
225            raise RoborockException("Failed to get a response from send email code")
226        response_code = code_response.get("code")
227        if response_code != 200:
228            _LOGGER.info("Request code failed for %s with the following context: %s", self._username, code_response)
229            if response_code == 2008:
230                raise RoborockAccountDoesNotExist("Account does not exist - check your login and try again.")
231            elif response_code == 9002:
232                raise RoborockTooFrequentCodeRequests("You have attempted to request too many codes. Try again later")
233            else:
234                raise RoborockException(f"{code_response.get('msg')} - response code: {code_response.get('code')}")
235
236    async def request_code_v4(self) -> None:
237        """Request a code using the v4 endpoint."""
238        if await self.country_code is None or await self.country is None:
239            _LOGGER.info("No country code or country found, trying old version of request code.")
240            return await self.request_code()
241        try:
242            await self._login_limiter.try_acquire_async("login")
243        except BucketFullException as ex:
244            _LOGGER.info(ex.meta_info)
245            raise RoborockRateLimit("Reached maximum requests for login. Please try again later.") from ex
246        base_url = await self.base_url
247        header_clientid = self._get_header_client_id()
248        code_request = PreparedRequest(
249            base_url,
250            self.session,
251            {
252                "header_clientid": header_clientid,
253                "Content-Type": "application/x-www-form-urlencoded",
254                "header_clientlang": "en",
255            },
256        )
257
258        code_response = await code_request.request(
259            "post",
260            "/api/v4/email/code/send",
261            data={"email": self._username, "type": "login", "platform": ""},
262        )
263        if code_response is None:
264            raise RoborockException("Failed to get a response from send email code")
265        response_code = code_response.get("code")
266        if response_code != 200:
267            _LOGGER.info("Request code failed for %s with the following context: %s", self._username, code_response)
268            if response_code == 2008:
269                raise RoborockAccountDoesNotExist("Account does not exist - check your login and try again.")
270            elif response_code == 9002:
271                raise RoborockTooFrequentCodeRequests("You have attempted to request too many codes. Try again later")
272            elif response_code == 3030 and len(self._base_urls) > 1:
273                self._base_urls = self._base_urls[1:]
274                self._iot_login_info = None
275                return await self.request_code_v4()
276            else:
277                raise RoborockException(f"{code_response.get('msg')} - response code: {code_response.get('code')}")
278
279    async def _sign_key_v3(self, s: str) -> str:
280        """Sign a randomly generated string."""
281        base_url = await self.base_url
282        header_clientid = self._get_header_client_id()
283        code_request = PreparedRequest(base_url, self.session, {"header_clientid": header_clientid})
284
285        code_response = await code_request.request(
286            "post",
287            "/api/v3/key/sign",
288            params={"s": s},
289        )
290
291        if not code_response or "data" not in code_response or "k" not in code_response["data"]:
292            raise RoborockException("Failed to get a response from sign key")
293        response_code = code_response.get("code")
294
295        if response_code != 200:
296            _LOGGER.info("Request code failed for %s with the following context: %s", self._username, code_response)
297            raise RoborockException(f"{code_response.get('msg')} - response code: {code_response.get('code')}")
298
299        return code_response["data"]["k"]
300
301    async def code_login_v4(
302        self, code: int | str, country: str | None = None, country_code: int | None = None
303    ) -> UserData:
304        """
305        Login via code authentication.
306        :param code: The code from the email.
307        :param country: The two-character representation of the country, i.e. "US"
308        :param country_code: the country phone number code i.e. 1 for US.
309        """
310        base_url = await self.base_url
311        if country is None:
312            country = await self.country
313        if country_code is None:
314            country_code = await self.country_code
315        if country_code is None or country is None:
316            _LOGGER.info("No country code or country found, trying old version of code login.")
317            return await self.code_login(code)
318        header_clientid = self._get_header_client_id()
319        x_mercy_ks = "".join(secrets.choice(string.ascii_letters + string.digits) for _ in range(16))
320        x_mercy_k = await self._sign_key_v3(x_mercy_ks)
321        login_request = PreparedRequest(
322            base_url,
323            self.session,
324            {
325                "header_clientid": header_clientid,
326                "x-mercy-ks": x_mercy_ks,
327                "x-mercy-k": x_mercy_k,
328                "Content-Type": "application/x-www-form-urlencoded",
329                "header_clientlang": "en",
330                "header_appversion": "4.54.02",
331                "header_phonesystem": "iOS",
332                "header_phonemodel": "iPhone16,1",
333            },
334        )
335        login_response = await login_request.request(
336            "post",
337            "/api/v4/auth/email/login/code",
338            data={
339                "country": country,
340                "countryCode": country_code,
341                "email": self._username,
342                "code": code,
343                # Major and minor version are the user agreement version, we will need to see if this needs to be
344                # dynamic https://usiot.roborock.com/api/v3/app/agreement/latest?country=US
345                "majorVersion": 14,
346                "minorVersion": 0,
347            },
348        )
349        if login_response is None:
350            raise RoborockException("Login request response is None")
351        response_code = login_response.get("code")
352        if response_code != 200:
353            _LOGGER.info("Login failed for %s with the following context: %s", self._username, login_response)
354            if response_code == 2018:
355                raise RoborockInvalidCode("Invalid code - check your code and try again.")
356            if response_code == 3009:
357                raise RoborockNoUserAgreement("You must accept the user agreement in the Roborock app to continue.")
358            if response_code == 3006:
359                raise RoborockInvalidUserAgreement(
360                    "User agreement must be accepted again - or you are attempting to use the Mi Home app account."
361                )
362            if response_code == 3039:
363                raise RoborockAccountDoesNotExist(
364                    "This account does not exist - please ensure that you selected the right region and email."
365                )
366            raise RoborockException(f"{login_response.get('msg')} - response code: {response_code}")
367        user_data = login_response.get("data")
368        if not isinstance(user_data, dict):
369            raise RoborockException("Got unexpected data type for user_data")
370        return UserData.from_dict(user_data)
371
372    async def pass_login(self, password: str) -> UserData:
373        try:
374            await self._login_limiter.try_acquire_async("login")
375        except BucketFullException as ex:
376            _LOGGER.info(ex.meta_info)
377            raise RoborockRateLimit("Reached maximum requests for login. Please try again later.") from ex
378        base_url = await self.base_url
379        header_clientid = self._get_header_client_id()
380
381        login_request = PreparedRequest(base_url, self.session, {"header_clientid": header_clientid})
382        login_response = await login_request.request(
383            "post",
384            "/api/v1/login",
385            params={
386                "username": self._username,
387                "password": password,
388                "needtwostepauth": "false",
389            },
390        )
391        if login_response is None:
392            raise RoborockException("Login response is none")
393        if login_response.get("code") != 200:
394            _LOGGER.info("Login failed for %s with the following context: %s", self._username, login_response)
395            raise RoborockException(f"{login_response.get('msg')} - response code: {login_response.get('code')}")
396        user_data = login_response.get("data")
397        if not isinstance(user_data, dict):
398            raise RoborockException("Got unexpected data type for user_data")
399        return UserData.from_dict(user_data)
400
401    async def pass_login_v3(self, password: str) -> UserData:
402        """Seemingly it follows the format below, but password is encrypted in some manner.
403        # login_response = await login_request.request(
404        #     "post",
405        #     "/api/v3/auth/email/login",
406        #     params={
407        #         "email": self._username,
408        #         "password": password,
409        #         "twoStep": 1,
410        #         "version": 0
411        #     },
412        # )
413        """
414        raise NotImplementedError("Pass_login_v3 has not yet been implemented")
415
416    async def code_login(self, code: int | str) -> UserData:
417        base_url = await self.base_url
418        header_clientid = self._get_header_client_id()
419
420        login_request = PreparedRequest(base_url, self.session, {"header_clientid": header_clientid})
421        login_response = await login_request.request(
422            "post",
423            "/api/v1/loginWithCode",
424            params={
425                "username": self._username,
426                "verifycode": code,
427                "verifycodetype": "AUTH_EMAIL_CODE",
428            },
429        )
430        if login_response is None:
431            raise RoborockException("Login request response is None")
432        response_code = login_response.get("code")
433        if response_code != 200:
434            _LOGGER.info("Login failed for %s with the following context: %s", self._username, login_response)
435            if response_code == 2018:
436                raise RoborockInvalidCode("Invalid code - check your code and try again.")
437            if response_code == 3009:
438                raise RoborockNoUserAgreement("You must accept the user agreement in the Roborock app to continue.")
439            if response_code == 3006:
440                raise RoborockInvalidUserAgreement(
441                    "User agreement must be accepted again - or you are attempting to use the Mi Home app account."
442                )
443            raise RoborockException(f"{login_response.get('msg')} - response code: {response_code}")
444        user_data = login_response.get("data")
445        if not isinstance(user_data, dict):
446            raise RoborockException("Got unexpected data type for user_data")
447        return UserData.from_dict(user_data)
448
449    async def _get_home_id(self, user_data: UserData):
450        base_url = await self.base_url
451        header_clientid = self._get_header_client_id()
452        home_id_request = PreparedRequest(base_url, self.session, {"header_clientid": header_clientid})
453        home_id_response = await home_id_request.request(
454            "get",
455            "/api/v1/getHomeDetail",
456            headers={"Authorization": user_data.token},
457        )
458        if home_id_response is None:
459            raise RoborockException("home_id_response is None")
460        if home_id_response.get("code") != 200:
461            _LOGGER.info("Get Home Id failed with the following context: %s", home_id_response)
462            if home_id_response.get("code") == 2010:
463                raise RoborockInvalidCredentials(
464                    f"Invalid credentials ({home_id_response.get('msg')}) - check your login and try again."
465                )
466            raise RoborockException(f"{home_id_response.get('msg')} - response code: {home_id_response.get('code')}")
467
468        return home_id_response["data"]["rrHomeId"]
469
470    async def get_home_data(self, user_data: UserData) -> HomeData:
471        try:
472            self._home_data_limiter.try_acquire("home_data")
473        except BucketFullException as ex:
474            _LOGGER.info(ex.meta_info)
475            raise RoborockRateLimit("Reached maximum requests for home data. Please try again later.") from ex
476        rriot = user_data.rriot
477        if rriot is None:
478            raise RoborockException("rriot is none")
479        home_id = await self._get_home_id(user_data)
480        if rriot.r.a is None:
481            raise RoborockException("Missing field 'a' in rriot reference")
482        home_request = PreparedRequest(
483            rriot.r.a,
484            self.session,
485            {
486                "Authorization": _get_hawk_authentication(rriot, f"/user/homes/{str(home_id)}"),
487            },
488        )
489        home_response = await home_request.request("get", "/user/homes/" + str(home_id))
490        if not home_response.get("success"):
491            raise RoborockException(home_response)
492        home_data = home_response.get("result")
493        if isinstance(home_data, dict):
494            return HomeData.from_dict(home_data)
495        else:
496            raise RoborockException("home_response result was an unexpected type")
497
498    async def get_home_data_v2(self, user_data: UserData) -> HomeData:
499        """This is the same as get_home_data, but uses a different endpoint and includes non-robotic vacuums."""
500        try:
501            self._home_data_limiter.try_acquire("home_data")
502        except BucketFullException as ex:
503            _LOGGER.info(ex.meta_info)
504            raise RoborockRateLimit("Reached maximum requests for home data. Please try again later.") from ex
505        rriot = user_data.rriot
506        if rriot is None:
507            raise RoborockException("rriot is none")
508        home_id = await self._get_home_id(user_data)
509        if rriot.r.a is None:
510            raise RoborockException("Missing field 'a' in rriot reference")
511        home_request = PreparedRequest(
512            rriot.r.a,
513            self.session,
514            {
515                "Authorization": _get_hawk_authentication(rriot, "/v2/user/homes/" + str(home_id)),
516            },
517        )
518        home_response = await home_request.request("get", "/v2/user/homes/" + str(home_id))
519        if not home_response.get("success"):
520            raise RoborockException(home_response)
521        home_data = home_response.get("result")
522        if isinstance(home_data, dict):
523            return HomeData.from_dict(home_data)
524        else:
525            raise RoborockException("home_response result was an unexpected type")
526
527    async def get_home_data_v3(self, user_data: UserData) -> HomeData:
528        """This is the same as get_home_data, but uses a different endpoint and includes non-robotic vacuums."""
529        try:
530            self._home_data_limiter.try_acquire("home_data")
531        except BucketFullException as ex:
532            _LOGGER.info(ex.meta_info)
533            raise RoborockRateLimit("Reached maximum requests for home data. Please try again later.") from ex
534        rriot = user_data.rriot
535        home_id = await self._get_home_id(user_data)
536        if rriot.r.a is None:
537            raise RoborockException("Missing field 'a' in rriot reference")
538        home_request = PreparedRequest(
539            rriot.r.a,
540            self.session,
541            {
542                "Authorization": _get_hawk_authentication(rriot, "/v3/user/homes/" + str(home_id)),
543            },
544        )
545        home_response = await home_request.request("get", "/v3/user/homes/" + str(home_id))
546        if not home_response.get("success"):
547            raise RoborockException(home_response)
548        home_data = home_response.get("result")
549        if isinstance(home_data, dict):
550            return HomeData.from_dict(home_data)
551        raise RoborockException(f"home_response result was an unexpected type: {home_data}")
552
553    async def get_rooms(self, user_data: UserData, home_id: int | None = None) -> list[HomeDataRoom]:
554        rriot = user_data.rriot
555        if rriot is None:
556            raise RoborockException("rriot is none")
557        if home_id is None:
558            home_id = await self._get_home_id(user_data)
559        if rriot.r.a is None:
560            raise RoborockException("Missing field 'a' in rriot reference")
561        room_request = PreparedRequest(
562            rriot.r.a,
563            self.session,
564            {
565                "Authorization": _get_hawk_authentication(rriot, "/v2/user/homes/" + str(home_id)),
566            },
567        )
568        room_response = await room_request.request("get", f"/user/homes/{str(home_id)}/rooms" + str(home_id))
569        if not room_response.get("success"):
570            raise RoborockException(room_response)
571        rooms = room_response.get("result")
572        if isinstance(rooms, list):
573            output_list = []
574            for room in rooms:
575                output_list.append(HomeDataRoom.from_dict(room))
576            return output_list
577        else:
578            raise RoborockException("home_response result was an unexpected type")
579
580    async def get_scenes(self, user_data: UserData, device_id: str) -> list[HomeDataScene]:
581        rriot = user_data.rriot
582        if rriot is None:
583            raise RoborockException("rriot is none")
584        if rriot.r.a is None:
585            raise RoborockException("Missing field 'a' in rriot reference")
586        scenes_request = PreparedRequest(
587            rriot.r.a,
588            self.session,
589            {
590                "Authorization": _get_hawk_authentication(rriot, f"/user/scene/device/{str(device_id)}"),
591            },
592        )
593        scenes_response = await scenes_request.request("get", f"/user/scene/device/{str(device_id)}")
594        if not scenes_response.get("success"):
595            raise RoborockException(scenes_response)
596        scenes = scenes_response.get("result")
597        if isinstance(scenes, list):
598            return [HomeDataScene.from_dict(scene) for scene in scenes]
599        else:
600            raise RoborockException("scene_response result was an unexpected type")
601
602    async def execute_scene(self, user_data: UserData, scene_id: int) -> None:
603        rriot = user_data.rriot
604        if rriot is None:
605            raise RoborockException("rriot is none")
606        if rriot.r.a is None:
607            raise RoborockException("Missing field 'a' in rriot reference")
608        execute_scene_request = PreparedRequest(
609            rriot.r.a,
610            self.session,
611            {
612                "Authorization": _get_hawk_authentication(rriot, f"/user/scene/{str(scene_id)}/execute"),
613            },
614        )
615        execute_scene_response = await execute_scene_request.request("POST", f"/user/scene/{str(scene_id)}/execute")
616        if not execute_scene_response.get("success"):
617            raise RoborockException(execute_scene_response)
618
619    async def get_schedules(self, user_data: UserData, device_id: str) -> list[HomeDataSchedule]:
620        rriot = user_data.rriot
621        if rriot is None:
622            raise RoborockException("rriot is none")
623        if rriot.r.a is None:
624            raise RoborockException("Missing field 'a' in rriot reference")
625        schedules_request = PreparedRequest(
626            rriot.r.a,
627            self.session,
628            {
629                "Authorization": _get_hawk_authentication(rriot, f"/user/devices/{device_id}/jobs"),
630            },
631        )
632        schedules_response = await schedules_request.request("get", f"/user/devices/{str(device_id)}/jobs")
633        if not schedules_response.get("success"):
634            raise RoborockException(schedules_response)
635        schedules = schedules_response.get("result")
636        if isinstance(schedules, list):
637            return [HomeDataSchedule.from_dict(schedule) for schedule in schedules]
638        else:
639            raise RoborockException(f"schedule_response result was an unexpected type: {schedules}")
640
641    async def get_products(self, user_data: UserData) -> ProductResponse:
642        """Gets all products and their schemas, good for determining status codes and model numbers."""
643        base_url = await self.base_url
644        header_clientid = self._get_header_client_id()
645        product_request = PreparedRequest(base_url, self.session, {"header_clientid": header_clientid})
646        product_response = await product_request.request(
647            "get",
648            "/api/v4/product",
649            headers={"Authorization": user_data.token},
650        )
651        if product_response is None:
652            raise RoborockException("home_id_response is None")
653        if product_response.get("code") != 200:
654            raise RoborockException(f"{product_response.get('msg')} - response code: {product_response.get('code')}")
655        result = product_response.get("data")
656        if isinstance(result, dict):
657            return ProductResponse.from_dict(result)
658        raise RoborockException("product result was an unexpected type")
659
660    async def download_code(self, user_data: UserData, product_id: int):
661        base_url = await self.base_url
662        header_clientid = self._get_header_client_id()
663        product_request = PreparedRequest(base_url, self.session, {"header_clientid": header_clientid})
664        request = {"apilevel": 99999, "productids": [product_id], "type": 2}
665        response = await product_request.request(
666            "post",
667            "/api/v1/appplugin",
668            json=request,
669            headers={"Authorization": user_data.token, "Content-Type": "application/json"},
670        )
671        return response["data"][0]["url"]
672
673    async def download_category_code(self, user_data: UserData):
674        base_url = await self.base_url
675        header_clientid = self._get_header_client_id()
676        product_request = PreparedRequest(base_url, self.session, {"header_clientid": header_clientid})
677        response = await product_request.request(
678            "get",
679            "api/v1/plugins?apiLevel=99999&type=2",
680            headers={
681                "Authorization": user_data.token,
682            },
683        )
684        return {r["category"]: r["url"] for r in response["data"]["categoryPluginList"]}
685
686
687class PreparedRequest:
688    def __init__(
689        self, base_url: str, session: aiohttp.ClientSession | None = None, base_headers: dict | None = None
690    ) -> None:
691        self.base_url = base_url
692        self.base_headers = base_headers or {}
693        self.session = session
694
695    async def request(self, method: str, url: str, params=None, data=None, headers=None, json=None) -> dict:
696        _url = "/".join(s.strip("/") for s in [self.base_url, url])
697        _headers = {**self.base_headers, **(headers or {})}
698        close_session = self.session is None
699        session = self.session if self.session is not None else aiohttp.ClientSession()
700        try:
701            async with session.request(method, _url, params=params, data=data, headers=_headers, json=json) as resp:
702                return await resp.json()
703        except ContentTypeError as err:
704            """If we get an error, lets log everything for debugging."""
705            try:
706                resp_json = await resp.json(content_type=None)
707                _LOGGER.info("Resp: %s", resp_json)
708            except ContentTypeError as err_2:
709                _LOGGER.info(err_2)
710            resp_raw = await resp.read()
711            _LOGGER.info("Resp raw: %s", resp_raw)
712            # Still raise the err so that it's clear it failed.
713            raise err
714        finally:
715            if close_session:
716                await session.close()
717
718
719def _process_extra_hawk_values(values: dict | None) -> str:
720    if values is None:
721        return ""
722    else:
723        sorted_keys = sorted(values.keys())
724        result = []
725        for key in sorted_keys:
726            value = values.get(key)
727            result.append(f"{key}={value}")
728        return hashlib.md5("&".join(result).encode()).hexdigest()
729
730
731def _get_hawk_authentication(rriot: RRiot, url: str, formdata: dict | None = None, params: dict | None = None) -> str:
732    timestamp = math.floor(time.time())
733    nonce = secrets.token_urlsafe(6)
734    formdata_str = _process_extra_hawk_values(formdata)
735    params_str = _process_extra_hawk_values(params)
736
737    prestr = ":".join(
738        [
739            rriot.u,
740            rriot.s,
741            nonce,
742            str(timestamp),
743            hashlib.md5(url.encode()).hexdigest(),
744            params_str,
745            formdata_str,
746        ]
747    )
748    mac = base64.b64encode(hmac.new(rriot.h.encode(), prestr.encode(), hashlib.sha256).digest()).decode()
749    return f'Hawk id="{rriot.u}",s="{rriot.s}",ts="{timestamp}",nonce="{nonce}",mac="{mac}"'
750
751
752class UserWebApiClient:
753    """Wrapper around RoborockApiClient to provide information for a specific user.
754
755    This binds a RoborockApiClient to a specific user context with the
756    provided UserData. This allows for easier access to user-specific data,
757    to avoid needing to pass UserData around and mock out the web API.
758    """
759
760    def __init__(self, web_api: RoborockApiClient, user_data: UserData) -> None:
761        """Initialize the wrapper with the API client and user data."""
762        self._web_api = web_api
763        self._user_data = user_data
764
765    async def get_home_data(self) -> HomeData:
766        """Fetch home data using the API client."""
767        return await self._web_api.get_home_data_v3(self._user_data)
768
769    async def get_routines(self, device_id: str) -> list[HomeDataScene]:
770        """Fetch routines (scenes) for a specific device."""
771        return await self._web_api.get_scenes(self._user_data, device_id)
772
773    async def execute_routine(self, scene_id: int) -> None:
774        """Execute a specific routine (scene) by its ID."""
775        await self._web_api.execute_scene(self._user_data, scene_id)