mywhoosh2garmin/app/mywhoosh/client.py

from __future__ import annotations

import uuid
from datetime import datetime, timezone

import httpx

from app.mywhoosh.models import MyWhooshActivity, MyWhooshToken
from app.mywhoosh.tokenstore import MyWhooshTokenStore

LOGIN_URL = "https://services.mywhoosh.com/http-service/api/login"
ACTIVITIES_BASE = "https://service14.mywhoosh.com/v2/"


class MyWhooshError(RuntimeError):
    pass


class MyWhooshAuthError(MyWhooshError):
    pass


class MyWhooshTransientError(MyWhooshError):
    pass


class MyWhooshIntegrationError(MyWhooshError):
    pass


class MyWhooshClient:
    def __init__(self, token_store: MyWhooshTokenStore, http_client: httpx.AsyncClient | None = None) -> None:
        self.token_store = token_store
        self.http = http_client or httpx.AsyncClient(timeout=30.0)
        self.token = token_store.load()

    async def login(self, email: str, password: str) -> None:
        payload = {
            "Username": email,
            "Password": password,
            "Platform": "Android",
            "Action": 1001,
            "CorrelationId": str(uuid.uuid4()),
            "DeviceId": str(uuid.uuid4()),
            "Authorization": "",
        }
        try:
            response = await self.http.post(LOGIN_URL, json=payload)
        except httpx.TransportError as exc:
            raise MyWhooshTransientError("MyWhoosh login request failed") from exc
        if response.status_code >= 500:
            raise MyWhooshTransientError(f"MyWhoosh login returned HTTP {response.status_code}")
        if response.status_code >= 400:
            raise MyWhooshAuthError(f"MyWhoosh login returned HTTP {response.status_code}")
        try:
            body = response.json()
        except ValueError as exc:
            raise MyWhooshIntegrationError("MyWhoosh login returned invalid JSON") from exc
        if body.get("Success") is not True or not body.get("AccessToken"):
            raise MyWhooshAuthError(str(body.get("Message") or "MyWhoosh login failed"))
        self.token = MyWhooshToken(
            access_token=str(body["AccessToken"]),
            refresh_token=str(body["RefreshToken"]) if body.get("RefreshToken") else None,
            whoosh_id=str(body["WhooshId"]) if body.get("WhooshId") else None,
        )
        self.token_store.save(self.token)

    async def ensure_authenticated(self, email: str, password: str) -> None:
        if self.token is None:
            await self.login(email, password)

    async def _authenticated_post(self, url: str, payload: dict, email: str, password: str) -> httpx.Response:
        await self.ensure_authenticated(email, password)
        for attempt in range(2):
            assert self.token is not None
            try:
                response = await self.http.post(
                    url,
                    json=payload,
                    headers={"Authorization": f"Bearer {self.token.access_token}"},
                )
            except httpx.TransportError as exc:
                raise MyWhooshTransientError("MyWhoosh request failed") from exc
            if response.status_code not in {401, 403}:
                if response.status_code >= 500:
                    raise MyWhooshTransientError(f"MyWhoosh returned HTTP {response.status_code}")
                return response
            if attempt == 0:
                self.token_store.clear()
                self.token = None
                await self.login(email, password)
                continue
            raise MyWhooshAuthError("MyWhoosh session rejected after reauthentication")
        raise AssertionError("unreachable")

    async def list_activities(self, email: str, password: str) -> list[MyWhooshActivity]:
        activities: list[MyWhooshActivity] = []
        page = 1
        total_pages = 1
        while page <= total_pages:
            response = await self._authenticated_post(
                ACTIVITIES_BASE + "rider/profile/activities",
                {"sortDate": "DESC", "page": page},
                email,
                password,
            )
            if response.status_code >= 400:
                raise MyWhooshIntegrationError(f"MyWhoosh activities returned HTTP {response.status_code}")
            try:
                body = response.json()
            except ValueError as exc:
                raise MyWhooshIntegrationError("MyWhoosh activities returned invalid JSON") from exc
            try:
                data = body["data"]
                total_pages = int(data["totalPages"])
                results = data["results"]
            except (KeyError, TypeError, ValueError) as exc:
                raise MyWhooshIntegrationError("MyWhoosh activities response has unexpected shape") from exc
            if not isinstance(results, list):
                raise MyWhooshIntegrationError("MyWhoosh activities response has unexpected shape")
            for row in results:
                activities.append(self._normalize_activity(row))
            page += 1
        return activities

    def _normalize_activity(self, row: object) -> MyWhooshActivity:
        if not isinstance(row, dict):
            raise MyWhooshIntegrationError("MyWhoosh activity row is not an object")
        activity_id = row.get("id")
        activity_file_id = row.get("activityFileId")
        if not activity_id or not activity_file_id:
            raise MyWhooshIntegrationError("MyWhoosh activity row missing stable id or activityFileId")
        raw_started = row.get("startDatetime")
        started_at: datetime | None = None
        if raw_started:
            try:
                started_at = datetime.fromisoformat(str(raw_started).replace("Z", "+00:00")).astimezone(
                    timezone.utc
                )
            except ValueError as exc:
                raise MyWhooshIntegrationError("MyWhoosh activity row has invalid startDatetime") from exc
        return MyWhooshActivity(
            id=str(activity_id),
            title=str(row.get("title") or ""),
            activity_file_id=str(activity_file_id),
            started_at=started_at,
        )

    async def download_fit(self, activity_file_id: str, email: str, password: str) -> bytes:
        response = await self._authenticated_post(
            ACTIVITIES_BASE + "rider/profile/download-activity-file",
            {"fileId": activity_file_id},
            email,
            password,
        )
        if response.status_code >= 400:
            raise MyWhooshIntegrationError(f"download metadata returned HTTP {response.status_code}")
        url = response.json().get("data")
        if not isinstance(url, str) or not url:
            raise MyWhooshIntegrationError("MyWhoosh download response has no URL")
        try:
            fit_response = await self.http.get(url)
        except httpx.TransportError as exc:
            raise MyWhooshTransientError("FIT download failed") from exc
        if fit_response.status_code >= 500:
            raise MyWhooshTransientError(f"FIT host returned HTTP {fit_response.status_code}")
        if fit_response.status_code >= 400:
            raise MyWhooshIntegrationError(f"FIT host returned HTTP {fit_response.status_code}")
        return fit_response.content