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 MyWhooshDeviceConflictError(MyWhooshAuthError): 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._owns_http = http_client is None self.http = http_client or httpx.AsyncClient(timeout=30.0) self.token = token_store.load() async def aclose(self) -> None: if self._owns_http: await self.http.aclose() async def __aenter__(self) -> "MyWhooshClient": return self async def __aexit__(self, *exc_info: object) -> None: await self.aclose() async def login(self, email: str, password: str) -> None: payload = { "Username": email, "Password": password, "Platform": "Android", "Action": 1001, "CorrelationId": str(uuid.uuid4()), "DeviceId": self.token_store.get_or_create_device_id(), "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 not isinstance(body, dict): raise MyWhooshIntegrationError("MyWhoosh login response is not a JSON object") if body.get("Success") is not True or not body.get("AccessToken"): message = str(body.get("Message") or "MyWhoosh login failed") if "another device" in message.lower(): raise MyWhooshDeviceConflictError(message) raise MyWhooshAuthError(message) 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): if self.token is None: raise MyWhooshIntegrationError("no token after ensure_authenticated") 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 == 403: raise MyWhooshAuthError(f"MyWhoosh returned HTTP {response.status_code}") if response.status_code != 401: 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 MyWhooshIntegrationError("unreachable state in _authenticated_post") async def list_activities( self, email: str, password: str, max_pages: int | None = None ) -> list[MyWhooshActivity]: activities: list[MyWhooshActivity] = [] page = 1 total_pages = 1 while page <= total_pages and (max_pages is None or page <= max_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: activity = self._normalize_activity(row) if activity is not None: activities.append(activity) page += 1 return activities def _normalize_activity(self, row: object) -> MyWhooshActivity | None: 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 activity_id is None or activity_id == "" or activity_file_id is None or activity_file_id == "": return None raw_started = row.get("startDatetime") started_at: datetime | None = None if raw_started: try: parsed = datetime.fromisoformat(str(raw_started).replace("Z", "+00:00")) except ValueError: return None if parsed.tzinfo is None: started_at = parsed.replace(tzinfo=timezone.utc) else: started_at = parsed.astimezone(timezone.utc) 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}") try: body = response.json() except ValueError as exc: raise MyWhooshIntegrationError("MyWhoosh download response returned invalid JSON") from exc if not isinstance(body, dict): raise MyWhooshIntegrationError("MyWhoosh download response is not a JSON object") url = body.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