216 lines
9.0 KiB
Python
216 lines
9.0 KiB
Python
"""Durable sync requests; shared filesystem locks serialize each user's session."""
|
|
|
|
import json
|
|
import logging
|
|
import time
|
|
from datetime import UTC, datetime, timedelta
|
|
from pathlib import Path
|
|
|
|
import click
|
|
from cryptography.fernet import Fernet
|
|
from filelock import FileLock, Timeout
|
|
from flask import current_app
|
|
from garminconnect import Garmin, GarminConnectAuthenticationError
|
|
|
|
from garmin_sync.service import SyncError, account_key, private_directory, sync_activities
|
|
|
|
from .extensions import db
|
|
from .models import GarminConnection, User
|
|
from .services import import_cache
|
|
|
|
|
|
def cipher():
|
|
key = current_app.config.get("TOKEN_ENCRYPTION_KEY")
|
|
if not key:
|
|
raise SyncError("Garmin connections are not configured on this server.")
|
|
return Fernet(key)
|
|
|
|
|
|
def session_lock(user_id):
|
|
directory = Path(current_app.instance_path) / "sync-locks"
|
|
private_directory(directory)
|
|
return FileLock(str(directory / f"{user_id}.lock"), timeout=0)
|
|
|
|
|
|
def connect(user, payload):
|
|
if len(payload) > 65536:
|
|
raise SyncError("Token file must be smaller than 64 KB.")
|
|
try:
|
|
data = json.loads(payload)
|
|
keys = ("di_token", "di_refresh_token", "di_client_id")
|
|
if not isinstance(data, dict) or not all(
|
|
isinstance(data.get(k), str) and data[k] for k in keys
|
|
):
|
|
raise ValueError()
|
|
except (ValueError, TypeError):
|
|
raise SyncError("Export a session with garminconnect 0.3.17.") from None
|
|
encryption = cipher()
|
|
with session_lock(user.id):
|
|
api = Garmin()
|
|
api.login(json.dumps({k: data[k] for k in keys}))
|
|
key = account_key(api, "garmin.com")
|
|
other = db.session.scalar(db.select(User).filter_by(local_account_key=key))
|
|
if other and other.id != user.id:
|
|
raise SyncError("This Garmin account is already connected.")
|
|
connection = db.session.scalar(db.select(GarminConnection).filter_by(user_id=user.id))
|
|
if connection and connection.account_key != key:
|
|
raise SyncError("Disconnect the existing Garmin account before replacing it.")
|
|
if not connection:
|
|
connection = GarminConnection(user_id=user.id, account_key=key)
|
|
db.session.add(connection)
|
|
connection.tokens = encryption.encrypt(api.client.dumps().encode()).decode()
|
|
connection.pending = True
|
|
connection.next_sync = datetime.now(UTC)
|
|
connection.failures = 0
|
|
connection.message = "Session connected. Sync queued."
|
|
user.local_account_key = key
|
|
db.session.commit()
|
|
|
|
|
|
def run_connection(connection_id):
|
|
connection = db.session.get(GarminConnection, connection_id)
|
|
if not connection:
|
|
return False
|
|
try:
|
|
with session_lock(connection.user_id):
|
|
db.session.refresh(connection)
|
|
now = datetime.now(UTC)
|
|
due = connection.next_sync.replace(tzinfo=UTC) <= now
|
|
if not connection.pending and not (connection.scheduled and due):
|
|
return False
|
|
api = None
|
|
try:
|
|
api = Garmin()
|
|
api.login(cipher().decrypt(connection.tokens.encode()).decode())
|
|
if account_key(api, "garmin.com") != connection.account_key:
|
|
raise GarminConnectAuthenticationError("Account mismatch")
|
|
# Persist rotated sessions even if the subsequent download fails.
|
|
connection.tokens = cipher().encrypt(api.client.dumps().encode()).decode()
|
|
db.session.commit()
|
|
try:
|
|
result = sync_activities(
|
|
api,
|
|
Path(current_app.config["DATA_ROOT"]) / connection.account_key,
|
|
100,
|
|
None,
|
|
)
|
|
finally:
|
|
connection.tokens = cipher().encrypt(api.client.dumps().encode()).decode()
|
|
db.session.commit()
|
|
user = db.session.get(User, connection.user_id)
|
|
count = import_cache(user, current_app.config["DATA_ROOT"])
|
|
if result["failed"]:
|
|
raise SyncError("Some activities failed; sync will retry.")
|
|
db.session.commit()
|
|
if connection.health_enabled:
|
|
from .recovery import sync_health
|
|
|
|
try:
|
|
health = sync_health(api, user)
|
|
connection.health_message = health["message"]
|
|
except Exception: # noqa: BLE001 -- optional provider data cannot block activities
|
|
db.session.rollback()
|
|
connection.health_message = "Daily recovery unavailable; will retry. Activity sync completed."
|
|
finally:
|
|
connection.tokens = cipher().encrypt(api.client.dumps().encode()).decode()
|
|
db.session.commit()
|
|
connection.last_success = now
|
|
connection.failures = 0
|
|
connection.message = f"Sync complete. {count} new activities imported."
|
|
connection.next_sync = now + timedelta(hours=1)
|
|
except Exception as error: # noqa: BLE001 -- do not expose provider credentials
|
|
db.session.rollback()
|
|
connection = db.session.get(GarminConnection, connection_id)
|
|
connection.failures += 1
|
|
connection.pending = False
|
|
if isinstance(error, GarminConnectAuthenticationError):
|
|
connection.scheduled = False
|
|
connection.message = (
|
|
"Session expired. Upload a fresh Garmin session to reconnect."
|
|
)
|
|
else:
|
|
connection.message = (
|
|
"Sync failed. Retry later or reconnect your Garmin session."
|
|
)
|
|
connection.next_sync = now + timedelta(
|
|
minutes=min(360, 5 * 2 ** min(connection.failures, 6))
|
|
)
|
|
connection.pending = False
|
|
db.session.commit()
|
|
return True
|
|
except Timeout:
|
|
db.session.rollback()
|
|
return False
|
|
|
|
|
|
def run_due():
|
|
ids = list(
|
|
db.session.scalars(
|
|
db.select(GarminConnection.id).where(
|
|
db.or_(
|
|
GarminConnection.pending.is_(True),
|
|
db.and_(
|
|
GarminConnection.scheduled.is_(True),
|
|
GarminConnection.next_sync <= datetime.now(UTC),
|
|
),
|
|
)
|
|
)
|
|
)
|
|
)
|
|
for connection_id in ids:
|
|
run_connection(connection_id)
|
|
return len(ids)
|
|
|
|
|
|
def register_cli(app):
|
|
@app.cli.command("backfill-health")
|
|
@click.option("--email", required=True)
|
|
@click.option("--days", default=90, type=click.IntRange(1, 365))
|
|
def health_backfill(email, days):
|
|
from .recovery import sync_health
|
|
|
|
logging.getLogger("garminconnect").disabled = True
|
|
user = db.session.scalar(db.select(User).filter_by(email=email.strip().lower()))
|
|
connection = db.session.scalar(db.select(GarminConnection).filter_by(
|
|
user_id=user.id)) if user else None
|
|
if not connection:
|
|
raise click.ClickException("No connected account found.")
|
|
try:
|
|
with session_lock(user.id):
|
|
api = Garmin()
|
|
authenticated = False
|
|
try:
|
|
api.login(cipher().decrypt(connection.tokens.encode()).decode())
|
|
authenticated = True
|
|
result = sync_health(api, user, days, days)
|
|
connection.health_enabled = True
|
|
connection.health_message = result["message"]
|
|
click.echo(result["message"])
|
|
finally:
|
|
if authenticated:
|
|
connection.tokens = cipher().encrypt(api.client.dumps().encode()).decode()
|
|
db.session.commit()
|
|
except Timeout:
|
|
raise click.ClickException("A sync is running; retry after it finishes.") from None
|
|
except Exception: # noqa: BLE001 -- suppress provider credentials at the operator boundary
|
|
raise click.ClickException("Daily recovery import failed; retry or reconnect.") from None
|
|
|
|
@app.cli.command("sync-worker")
|
|
@click.option("--once", is_flag=True, help="Process currently due syncs and exit.")
|
|
def worker(once):
|
|
logging.getLogger("garminconnect").disabled = True
|
|
while True:
|
|
try:
|
|
count = run_due()
|
|
click.echo(f"Sync worker heartbeat: {count} due connections.")
|
|
except Exception: # noqa: BLE001 -- suppress provider credentials at boundary
|
|
db.session.rollback()
|
|
click.echo("Sync worker iteration failed; retrying without logging credentials.")
|
|
if once:
|
|
raise click.ClickException("Sync worker failed.") from None
|
|
finally:
|
|
db.session.remove()
|
|
if once:
|
|
break
|
|
time.sleep(30)
|