"""Garmin Connect endpoints (/api/garmin/*).""" from __future__ import annotations import json from datetime import UTC from fastapi import APIRouter, Cookie, HTTPException, Request from fastapi.responses import JSONResponse, StreamingResponse from bincio.serve import deps, tasks router = APIRouter() def _garmin_user_message(exc: Exception) -> str: """Return a human-friendly error message for common Garmin login failures.""" msg = str(exc) fallback = ( " In the meantime, you can export your activities from Garmin Connect " "(garmin.com → Activities → Export) or Garmin Express as FIT files " "and upload them directly." ) if "429" in msg or "rate limit" in msg.lower(): return ( "Garmin is rate-limiting this server's IP address (HTTP 429). " "Wait a few hours and try again." + fallback ) if "403" in msg: return ( "Cloudflare is blocking the login request (HTTP 403). " "This is a known upstream issue — try again later or update garminconnect " "(uv sync --extra garmin)." + fallback ) if "GARMIN Authentication Application" in msg or "unexpected title" in msg.lower(): return ( "Garmin's login page returned a CAPTCHA or MFA challenge that " "cannot be completed automatically. Try again later, or disable " "two-factor authentication on your Garmin account." + fallback ) return f"Login failed: {exc}" + fallback @router.get("/api/garmin/status") async def garmin_status(bincio_session: str | None = Cookie(default=None)) -> JSONResponse: """Return whether Garmin credentials are stored for the current user.""" user = deps._require_user(bincio_session) dd = deps._get_data_dir() / user.handle from bincio.extract.garmin_api import has_credentials from bincio.extract.garmin_sync import _load_sync_state connected = has_credentials(dd) last_sync = None if connected: state = _load_sync_state(dd) last_sync = state.get("last_sync_at") return JSONResponse({"connected": connected, "last_sync": last_sync}) @router.post("/api/garmin/connect") async def garmin_connect( request: Request, bincio_session: str | None = Cookie(default=None), ) -> JSONResponse: """Test Garmin login with the supplied credentials and save them on success.""" user = deps._require_user(bincio_session) body = await request.json() email = (body.get("email") or "").strip() password = body.get("password") or "" if not email or not password: raise HTTPException(400, "email and password are required") data_dir = deps._get_data_dir() user_dir = data_dir / user.handle from bincio.extract.garmin_api import GarminError, test_login try: info = test_login(data_dir, user_dir, email, password) except GarminError as exc: raise HTTPException(400, _garmin_user_message(exc)) return JSONResponse({"ok": True, **info}) @router.post("/api/garmin/disconnect") async def garmin_disconnect(bincio_session: str | None = Cookie(default=None)) -> JSONResponse: """Remove stored Garmin credentials and session for the current user.""" user = deps._require_user(bincio_session) dd = deps._get_data_dir() / user.handle from bincio.extract.garmin_api import delete_credentials delete_credentials(dd) return JSONResponse({"ok": True}) @router.get("/api/garmin/sync/stream") async def garmin_sync_stream(bincio_session: str | None = Cookie(default=None)) -> StreamingResponse: """SSE endpoint — streams per-activity Garmin sync progress.""" user = deps._require_user(bincio_session) data_dir = deps._get_data_dir() user_dir = data_dir / user.handle from bincio.extract.garmin_api import GarminError, has_credentials if not has_credentials(user_dir): raise HTTPException(400, "No Garmin credentials stored — connect first") from bincio.extract.garmin_sync import garmin_sync_iter def event_stream(): try: for event in garmin_sync_iter(data_dir, user_dir): if event["type"] == "done": tasks._trigger_rebuild(user.handle) yield f"data: {json.dumps(event)}\n\n" except GarminError as exc: yield f"data: {json.dumps({'type': 'error', 'message': _garmin_user_message(exc)})}\n\n" except Exception as exc: yield f"data: {json.dumps({'type': 'error', 'message': _garmin_user_message(exc)})}\n\n" return StreamingResponse( event_stream(), media_type="text/event-stream", headers={"Cache-Control": "no-cache", "X-Accel-Buffering": "no"}, ) @router.post("/api/garmin/import-gear") async def garmin_import_gear(bincio_session: str | None = Cookie(default=None)) -> JSONResponse: """One-time backfill: fetch gear registry from Garmin and match to existing activities by timestamp.""" import contextlib import re import uuid import yaml from bincio.extract.garmin_api import GarminError, get_client, has_credentials from bincio.render.merge import merge_one from bincio.serve.routers.gear import _load as _gear_load from bincio.serve.routers.gear import _save as _gear_save user = deps._require_user(bincio_session) data_dir = deps._get_data_dir() user_dir = data_dir / user.handle if not has_credentials(user_dir): raise HTTPException(400, "No Garmin credentials stored — connect first") try: client = get_client(data_dir, user_dir) except GarminError as e: raise HTTPException(502, str(e)) # Fetch gear list try: prof = client.connectapi("/userprofile-service/socialProfile") profile_id = prof.get("profileId") if isinstance(prof, dict) else None if not profile_id: raise HTTPException(502, "Could not read Garmin profile ID") garmin_gear = client.get_gear(profile_id) except GarminError as e: raise HTTPException(502, str(e)) if not isinstance(garmin_gear, list) or not garmin_gear: return JSONResponse({"ok": True, "gear_added": 0, "activities_updated": 0}) # Build / update gear registry registry = _gear_load(user_dir) known = {g.get("garmin_id") for g in registry if g.get("garmin_id")} uuid_to_name: dict[str, str] = {} gear_added = 0 for g in garmin_gear: guuid = g.get("uuid") or "" name = (g.get("customMakeModel") or g.get("displayName") or f"{g.get('gearMakeName','')} {g.get('gearModelName','')}".strip()) if not name or not guuid: continue uuid_to_name[guuid] = name if guuid not in known: gear_type = g.get("gearTypeName", "").lower() if gear_type not in ("bike", "shoes", "skis"): gear_type = "other" retired = g.get("gearStatusName") == "retired" registry.append({"id": str(uuid.uuid4()), "name": name, "type": gear_type, "retired": retired, "garmin_id": guuid}) known.add(guuid) gear_added += 1 else: for item in registry: if item.get("garmin_id") == guuid: item["name"] = name _gear_save(user_dir, registry) # Build timestamp → activity_id map from the user's index shards from datetime import datetime ts_to_id: dict[int, str] = {} merged_dir = user_dir / "_merged" shard_dirs = [merged_dir] if merged_dir.exists() else [user_dir] for shard_dir in shard_dirs: for shard_path in sorted(shard_dir.glob("index*.json")): try: idx = json.loads(shard_path.read_text(encoding="utf-8")) for a in idx.get("activities", []): started = a.get("started_at") or "" if started and a.get("id"): dt = datetime.fromisoformat(started.replace("Z", "+00:00")) ts_to_id[int(dt.astimezone(UTC).timestamp())] = a["id"] except (OSError, json.JSONDecodeError, KeyError): continue # For each gear, fetch its activities and match by timestamp edits_dir = user_dir / "edits" edits_dir.mkdir(exist_ok=True) activities_updated = 0 for guuid, gear_name in uuid_to_name.items(): try: gear_acts = client.get_gear_activities(guuid, limit=10000) except Exception: continue if not isinstance(gear_acts, list): continue for ga in gear_acts: gmt = ga.get("startTimeGMT") or "" if not gmt: continue try: from datetime import datetime dt = datetime.strptime(gmt, "%Y-%m-%d %H:%M:%S").replace(tzinfo=UTC) ts = int(dt.timestamp()) except ValueError: continue # Match within ±60 s act_id = None for delta in range(0, 61): act_id = ts_to_id.get(ts + delta) or ts_to_id.get(ts - delta) if act_id: break if not act_id: continue # Skip if activity already has gear set act_json = user_dir / "activities" / f"{act_id}.json" if act_json.exists(): try: if json.loads(act_json.read_text(encoding="utf-8")).get("gear"): continue except (OSError, json.JSONDecodeError): pass sidecar = edits_dir / f"{act_id}.md" fm, body = {}, "" if sidecar.exists(): try: text = sidecar.read_text(encoding="utf-8") parts = re.split(r"^---[ \t]*$", text, maxsplit=2, flags=re.MULTILINE) if len(parts) >= 3: fm = yaml.safe_load(parts[1]) or {} body = parts[2].strip() except Exception: pass if fm.get("gear"): continue fm["gear"] = gear_name fm_text = yaml.safe_dump(fm, default_flow_style=False, allow_unicode=True).strip() content = f"---\n{fm_text}\n---\n" if body: content += f"\n{body}\n" sidecar.write_text(content, encoding="utf-8") with contextlib.suppress(Exception): merge_one(user_dir, act_id) activities_updated += 1 tasks._trigger_rebuild(user.handle) return JSONResponse({"ok": True, "gear_added": gear_added, "activities_updated": activities_updated})