mirror of
https://github.com/sstent/FitTrack_GarminSync.git
synced 2026-03-13 16:35:42 +00:00
sstent
ca9d7d9e90
- Ensure code aligns with CentralDB models - Document code alignment with CentralDB models - Remove informal reference documents (data-model.md, DB_API_SPEC.json, GARMINSYNC_SPEC.md) - Run linters and formatters (black, isort, mypy) - Update project configuration files - Add .dockerignore for Docker builds - Perform code formatting and import sorting - Fix type checking issues - Update documentation files - Complete implementation tasks as per spec
92 lines
3.2 KiB
Python
92 lines
3.2 KiB
Python
import json
|
|
import logging
|
|
import os
|
|
import tempfile
|
|
from datetime import datetime
|
|
from typing import Optional, TextIO
|
|
|
|
from garminconnect import Garmin
|
|
from tenacity import (
|
|
retry,
|
|
retry_if_exception_type,
|
|
stop_after_attempt,
|
|
wait_exponential,
|
|
)
|
|
|
|
from ..models.central_db_models import GarminCredentials
|
|
|
|
# Configure debug logging for garminconnect
|
|
logging.basicConfig(level=logging.DEBUG)
|
|
logger = logging.getLogger(__name__)
|
|
|
|
# Define a retry strategy for Garmin login
|
|
GARMIN_LOGIN_RETRY_STRATEGY = retry(
|
|
stop=stop_after_attempt(10), # Increased attempts
|
|
wait=wait_exponential(
|
|
multiplier=1, min=10, max=60
|
|
), # Increased min and max wait times
|
|
retry=retry_if_exception_type(Exception), # Retry on any exception for now
|
|
reraise=True,
|
|
)
|
|
|
|
|
|
class GarminAuthService:
|
|
def __init__(self):
|
|
pass
|
|
|
|
@GARMIN_LOGIN_RETRY_STRATEGY # Apply retry strategy here
|
|
async def _perform_login(self, username: str, password: str) -> Garmin:
|
|
"""Helper to perform the actual garminconnect login with retry."""
|
|
client = Garmin(username, password)
|
|
client.login()
|
|
return client
|
|
|
|
async def initial_login(
|
|
self, username: str, password: str
|
|
) -> Optional[GarminCredentials]:
|
|
"""Performs initial login to Garmin Connect and returns GarminCredentials."""
|
|
try:
|
|
garmin_client = await self._perform_login(
|
|
username, password
|
|
) # Use the retried login helper
|
|
if not garmin_client:
|
|
return None
|
|
|
|
logger.info(f"Successful Garmin login for {username}")
|
|
|
|
with tempfile.TemporaryDirectory() as temp_dir:
|
|
session_file = os.path.join(temp_dir, "garth_session.json")
|
|
garmin_client.garth.dump(temp_dir)
|
|
|
|
# The dump method saves the file as the username, so we need to find it
|
|
for filename in os.listdir(temp_dir):
|
|
if filename.endswith(".json"):
|
|
session_file = os.path.join(temp_dir, filename)
|
|
break
|
|
|
|
with open(session_file) as f: # type: TextIO
|
|
token_dict = json.load(f) # type: ignore
|
|
|
|
# Extract tokens and cookies
|
|
access_token = token_dict.get("access_token", "")
|
|
access_token_secret = token_dict.get("access_token_secret", "")
|
|
token_expiration_date = datetime.fromtimestamp(
|
|
token_dict.get("token_expiration_date", 0)
|
|
)
|
|
|
|
garmin_credentials = GarminCredentials(
|
|
garmin_username=username,
|
|
garmin_password_plaintext=password, # Storing plaintext for re-auth, consider encryption
|
|
access_token=access_token,
|
|
access_token_secret=access_token_secret,
|
|
token_expiration_date=token_expiration_date,
|
|
display_name=garmin_client.display_name,
|
|
full_name=garmin_client.full_name,
|
|
unit_system=garmin_client.unit_system,
|
|
token_dict=token_dict,
|
|
)
|
|
return garmin_credentials
|
|
except Exception as e:
|
|
logger.error(f"Garmin initial login failed for {username}: {e}")
|
|
return None
|