b48008c16e
Implement GarminUploader adapter around garminconnect library: - Import activities using import_activity() not upload_activity() - Treat duplicate activity responses as terminal success - Raise GarminUploadBlocked for MFA to allow UI code collection - Distinguish auth, transient, and other errors appropriately - Helper functions for duplicate detection and activity ID extraction Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
117 lines
4.1 KiB
Python
117 lines
4.1 KiB
Python
from __future__ import annotations
|
|
|
|
from dataclasses import dataclass
|
|
from pathlib import Path
|
|
from typing import Any, Callable, Protocol
|
|
|
|
|
|
class GarminClientProtocol(Protocol):
|
|
def login(self, tokenstore: str | None = None) -> Any: ...
|
|
def import_activity(self, activity_path: str) -> Any: ...
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class UploadResult:
|
|
status: str
|
|
duplicate: bool
|
|
garmin_activity_id: str | None
|
|
raw_response: Any
|
|
|
|
|
|
class GarminUploadBlocked(RuntimeError):
|
|
pass
|
|
|
|
|
|
class GarminAuthError(RuntimeError):
|
|
pass
|
|
|
|
|
|
class GarminTransientError(RuntimeError):
|
|
pass
|
|
|
|
|
|
class GarminUploader:
|
|
def __init__(
|
|
self,
|
|
*,
|
|
email: str,
|
|
password: str,
|
|
tokenstore: Path,
|
|
client_factory: Callable[..., GarminClientProtocol] | None = None,
|
|
) -> None:
|
|
self.email = email
|
|
self.password = password
|
|
self.tokenstore = tokenstore
|
|
self.client_factory = client_factory
|
|
self._client: GarminClientProtocol | None = None
|
|
self._mfa_code: str | None = None
|
|
|
|
def import_fit(self, fit_path: Path, mfa_code: str | None = None) -> UploadResult:
|
|
self._mfa_code = mfa_code
|
|
try:
|
|
client = self._ensure_client()
|
|
try:
|
|
response = client.import_activity(str(fit_path))
|
|
except Exception as exc:
|
|
if _looks_duplicate_error(exc):
|
|
return UploadResult("duplicate", True, None, str(exc))
|
|
text = str(exc).lower()
|
|
if any(token in text for token in ("timeout", "temporar", "connection", "502", "503", "504")):
|
|
raise GarminTransientError("Garmin import failed transiently") from exc
|
|
raise
|
|
return UploadResult("imported", False, _extract_activity_id(response), response)
|
|
finally:
|
|
self._mfa_code = None
|
|
|
|
def _ensure_client(self) -> GarminClientProtocol:
|
|
if self._client is not None:
|
|
return self._client
|
|
self.tokenstore.mkdir(parents=True, exist_ok=True)
|
|
factory = self.client_factory or _default_garmin_factory
|
|
client = factory(self.email, self.password, prompt_mfa=self._prompt_mfa)
|
|
try:
|
|
client.login(str(self.tokenstore))
|
|
except GarminUploadBlocked:
|
|
raise
|
|
except Exception as exc:
|
|
text = str(exc).lower()
|
|
if "mfa" in text:
|
|
raise GarminUploadBlocked("Garmin MFA is required") from exc
|
|
if any(token in text for token in ("password", "credential", "unauthorized", "401")):
|
|
raise GarminAuthError("Garmin authentication failed") from exc
|
|
if any(token in text for token in ("timeout", "temporar", "connection", "502", "503", "504")):
|
|
raise GarminTransientError("Garmin login failed transiently") from exc
|
|
raise GarminAuthError("Garmin login failed") from exc
|
|
self._client = client
|
|
return client
|
|
|
|
def _prompt_mfa(self) -> str:
|
|
if self._mfa_code:
|
|
return self._mfa_code
|
|
raise GarminUploadBlocked("Garmin requested MFA")
|
|
|
|
|
|
def _default_garmin_factory(*args: Any, **kwargs: Any) -> GarminClientProtocol:
|
|
from garminconnect import Garmin
|
|
|
|
return Garmin(*args, **kwargs)
|
|
|
|
|
|
def _looks_duplicate_error(exc: Exception) -> bool:
|
|
text = str(exc).lower()
|
|
return any(token in text for token in ("duplicate", "already exists", "409"))
|
|
|
|
|
|
def _extract_activity_id(response: Any) -> str | None:
|
|
if not isinstance(response, dict):
|
|
return None
|
|
candidates = [response.get("activityId"), response.get("activity_id"), response.get("id")]
|
|
detailed = response.get("detailedImportResult")
|
|
if isinstance(detailed, dict):
|
|
candidates.extend([detailed.get("uploadId"), detailed.get("activityId")])
|
|
for key in ("successes", "success", "importedActivities"):
|
|
items = response.get(key)
|
|
if isinstance(items, list) and items and isinstance(items[0], dict):
|
|
candidates.extend([items[0].get("activityId"), items[0].get("id")])
|
|
return next((str(value) for value in candidates if value is not None), None)
|