146 lines
6.1 KiB
Python
146 lines
6.1 KiB
Python
"""Run with `uv run python -m garmin_sync --help`. Never accepts passwords as flags."""
|
||
|
||
import argparse
|
||
import getpass
|
||
import json
|
||
import logging
|
||
import os
|
||
import re
|
||
import sys
|
||
import warnings
|
||
from pathlib import Path
|
||
|
||
from filelock import FileLock, Timeout
|
||
from garminconnect import (
|
||
Garmin,
|
||
GarminConnectAuthenticationError,
|
||
GarminConnectConnectionError,
|
||
GarminConnectTooManyRequestsError,
|
||
)
|
||
|
||
from .service import (
|
||
SyncError,
|
||
account_key,
|
||
atomic_write,
|
||
private_directory,
|
||
sync_activities,
|
||
write_json,
|
||
)
|
||
|
||
ROOT = Path(__file__).resolve().parents[1]
|
||
|
||
|
||
def parser():
|
||
cli = argparse.ArgumentParser(description="Read-only Garmin completed-activity sync.")
|
||
cli.add_argument("--profile", default="default", help="Local account slot (default: default)")
|
||
cli.add_argument("--domain", choices=["garmin.com", "garmin.cn"], default="garmin.com")
|
||
commands = cli.add_subparsers(dest="command", required=True)
|
||
commands.add_parser("login", help="Interactive local Garmin login and MFA; save session")
|
||
commands.add_parser("verify", help="Validate the saved session without credentials")
|
||
imported = commands.add_parser("import-tokens", help="Validate and copy a local token export")
|
||
imported.add_argument("file", type=Path)
|
||
sync = commands.add_parser("sync", help="Download and decode recent completed activities")
|
||
sync.add_argument("--limit", type=int, default=5, help="Recent activities to examine (1–1000)")
|
||
sync.add_argument("--type", choices=["all", "running"], default="all")
|
||
return cli
|
||
|
||
|
||
def run(args):
|
||
if not re.fullmatch(r"[a-zA-Z0-9_-]{1,64}", args.profile):
|
||
raise SyncError("Profile must be 1–64 letters, numbers, underscores or hyphens.")
|
||
if args.command == "sync" and not 1 <= args.limit <= 1000:
|
||
raise SyncError("Limit must be between 1 and 1000.")
|
||
# An inherited global token store must never select another account.
|
||
os.environ.pop("GARMINTOKENS", None)
|
||
store = ROOT / "instance" / "garmin" / args.profile
|
||
private_directory(store)
|
||
token_path = store / "garmin_tokens.json"
|
||
with FileLock(str(store / ".session.lock"), timeout=0):
|
||
api = Garmin(is_cn=args.domain == "garmin.cn")
|
||
if args.command == "login":
|
||
if not sys.stdin.isatty():
|
||
raise SyncError(
|
||
"Run login in your own interactive terminal; do not paste credentials into chat."
|
||
)
|
||
email = input("Garmin email: ").strip()
|
||
with warnings.catch_warnings():
|
||
warnings.simplefilter("error", getpass.GetPassWarning)
|
||
password = getpass.getpass("Garmin password (hidden): ")
|
||
api = Garmin(
|
||
email=email,
|
||
password=password,
|
||
is_cn=args.domain == "garmin.cn",
|
||
prompt_mfa=lambda: getpass.getpass("Garmin MFA code (hidden): "),
|
||
)
|
||
password = None
|
||
print("Authenticating with Garmin...", flush=True)
|
||
api.login() # Deliberately do not reuse a previous account's token file.
|
||
elif args.command == "import-tokens":
|
||
if args.file.stat().st_size > 65536:
|
||
raise SyncError("Token file is too large.")
|
||
payload = json.loads(args.file.read_text(encoding="utf-8-sig"))
|
||
if not isinstance(payload, dict) or not all(
|
||
isinstance(payload.get(key), str) and payload[key]
|
||
for key in ("di_token", "di_refresh_token", "di_client_id")
|
||
):
|
||
raise SyncError("Unsupported token format; export with garminconnect 0.3.17.")
|
||
api.login(
|
||
json.dumps(
|
||
{key: payload[key] for key in ("di_token", "di_refresh_token", "di_client_id")}
|
||
)
|
||
)
|
||
else:
|
||
if not token_path.is_file():
|
||
raise SyncError("No saved session. Run login or import-tokens first.")
|
||
api.login(str(token_path))
|
||
|
||
key = account_key(api, args.domain)
|
||
# Writes refreshed state for inline imports/login, plus every successful resume.
|
||
atomic_write(token_path, api.client.dumps().encode())
|
||
print(f"Garmin session verified. Local account: {key}", flush=True)
|
||
if args.command != "sync":
|
||
print(f"Session saved privately in instance/garmin/{args.profile}/garmin_tokens.json")
|
||
return 0
|
||
destination = ROOT / "data" / "garmin" / key
|
||
try:
|
||
result = sync_activities(
|
||
api, destination, args.limit, None if args.type == "all" else args.type
|
||
)
|
||
finally:
|
||
# Preserve rotations even if a later download fails.
|
||
atomic_write(token_path, api.client.dumps().encode())
|
||
write_json(destination / "last-sync.json", result)
|
||
print(json.dumps(result, indent=2))
|
||
print(f"Private activity files: {destination}")
|
||
return 1 if result["failed"] else 0
|
||
|
||
|
||
def main():
|
||
# Third-party exception/log strings can contain credentials and request URLs.
|
||
logging.disable(logging.CRITICAL)
|
||
args = parser().parse_args()
|
||
try:
|
||
return run(args)
|
||
except SyncError as error:
|
||
print(f"Error: {error}", file=sys.stderr)
|
||
except GarminConnectTooManyRequestsError:
|
||
print("Garmin rate limit reached. Wait before retrying.", file=sys.stderr)
|
||
except GarminConnectAuthenticationError:
|
||
print("Garmin rejected the session/login. Run login locally to reconnect.", file=sys.stderr)
|
||
except GarminConnectConnectionError:
|
||
print("Garmin connection failed. Retry later; no credentials were logged.", file=sys.stderr)
|
||
except Timeout:
|
||
print("Another command is using this profile. Wait for it to finish.", file=sys.stderr)
|
||
except (KeyboardInterrupt, EOFError):
|
||
print("Cancelled.", file=sys.stderr)
|
||
except Exception as error: # noqa: BLE001 -- CLI boundary must not expose provider secrets.
|
||
print(
|
||
f"Operation failed ({type(error).__name__}); details suppressed to protect credentials.",
|
||
file=sys.stderr,
|
||
)
|
||
return 1
|
||
|
||
|
||
if __name__ == "__main__":
|
||
raise SystemExit(main())
|