mywhoosh2garmin/app/garmin/uploader.py

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 GarminError(RuntimeError):
    pass


class GarminUploadBlocked(GarminError):
    pass


class GarminAuthError(GarminError):
    pass


class GarminTransientError(GarminError):
    pass


class GarminImportRejected(GarminError):
    pass


_TRANSIENT_ERROR_TOKENS = (
    "timeout",
    "temporar",
    "connection",
    "429",
    "too many",
    "rate limit",
    "500",
    "502",
    "503",
    "504",
)


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 _TRANSIENT_ERROR_TOKENS):
                    raise GarminTransientError("Garmin import failed transiently") from exc
                if any(token in text for token in ("password", "credential", "unauthorized", "401")):
                    raise GarminAuthError("Garmin import failed: authentication rejected") from exc
                raise
            _raise_if_import_rejected(response)
            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, mode=0o700)
        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 _TRANSIENT_ERROR_TOKENS):
                raise GarminTransientError("Garmin login failed transiently") from exc
            raise GarminTransientError("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:
    status_code = getattr(getattr(exc, "response", None), "status_code", None)
    if status_code == 409:
        return True
    text = str(exc).lower()
    return any(token in text for token in ("duplicate", "already exists", "409"))


def _raise_if_import_rejected(response: Any) -> None:
    if not isinstance(response, dict):
        return
    detailed = response.get("detailedImportResult")
    if not isinstance(detailed, dict):
        return
    failures = detailed.get("failures")
    successes = detailed.get("successes")
    if isinstance(failures, list) and failures and not successes:
        raise GarminImportRejected(f"Garmin rejected the import ({len(failures)} failure(s))")


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)