137 lines
3.9 KiB
Python
137 lines
3.9 KiB
Python
from __future__ import annotations
|
|
|
|
import logging
|
|
from dataclasses import dataclass
|
|
from pathlib import Path
|
|
from typing import Any, Callable, Protocol
|
|
|
|
from .config import Settings
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
|
|
class GarminClientProtocol(Protocol):
|
|
def login(self, tokenstore: str | None = None) -> Any:
|
|
...
|
|
|
|
def upload_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):
|
|
"""Raised when Garmin login needs user action such as MFA."""
|
|
|
|
|
|
class GarminUploader:
|
|
def __init__(
|
|
self,
|
|
settings: Settings,
|
|
client_factory: Callable[..., GarminClientProtocol] | None = None,
|
|
) -> None:
|
|
self.settings = settings
|
|
self._client_factory = client_factory
|
|
self._client: GarminClientProtocol | None = None
|
|
|
|
def upload(self, fit_path: Path) -> UploadResult:
|
|
client = self._ensure_client()
|
|
try:
|
|
response = client.upload_activity(str(fit_path))
|
|
except Exception as exc:
|
|
if _looks_duplicate_error(exc):
|
|
logger.info("Garmin already has activity for %s", fit_path)
|
|
return UploadResult(
|
|
status="duplicate",
|
|
duplicate=True,
|
|
garmin_activity_id=None,
|
|
raw_response=str(exc),
|
|
)
|
|
raise
|
|
|
|
return UploadResult(
|
|
status="uploaded",
|
|
duplicate=False,
|
|
garmin_activity_id=_extract_activity_id(response),
|
|
raw_response=response,
|
|
)
|
|
|
|
def _ensure_client(self) -> GarminClientProtocol:
|
|
if self._client is not None:
|
|
return self._client
|
|
|
|
factory = self._client_factory or _default_garmin_factory
|
|
client = factory(
|
|
self.settings.garmin_email,
|
|
self.settings.garmin_password,
|
|
prompt_mfa=self._prompt_mfa,
|
|
)
|
|
try:
|
|
client.login(str(self.settings.garmin_tokenstore))
|
|
except RuntimeError:
|
|
raise
|
|
except Exception as exc:
|
|
if "mfa" in str(exc).lower():
|
|
raise GarminUploadBlocked(
|
|
"Garmin MFA is required. Set GARMIN_MFA_CODE for one run."
|
|
) from exc
|
|
raise
|
|
self._client = client
|
|
return client
|
|
|
|
def _prompt_mfa(self) -> str:
|
|
if self.settings.garmin_mfa_code:
|
|
return self.settings.garmin_mfa_code
|
|
raise GarminUploadBlocked(
|
|
"Garmin requested MFA but GARMIN_MFA_CODE is not set."
|
|
)
|
|
|
|
|
|
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_import = response.get("detailedImportResult")
|
|
if isinstance(detailed_import, dict):
|
|
candidates.extend(
|
|
[
|
|
detailed_import.get("uploadId"),
|
|
detailed_import.get("activityId"),
|
|
]
|
|
)
|
|
|
|
for key in ("successes", "success", "importedActivities"):
|
|
items = response.get(key)
|
|
if isinstance(items, list) and items:
|
|
first = items[0]
|
|
if isinstance(first, dict):
|
|
candidates.extend([first.get("activityId"), first.get("id")])
|
|
|
|
for candidate in candidates:
|
|
if candidate is not None:
|
|
return str(candidate)
|
|
return None
|
|
|