feat: Strava export dry-run preview (new vs duplicate counts + confirm import) and prefer existing Garmin data over Strava on dedup

backend/app/api/upload.py

@@ -3,12 +3,13 @@ import zipfile
 from pathlib import Path
 from fastapi import APIRouter, Depends, UploadFile, File, HTTPException, BackgroundTasks
 from sqlalchemy.ext.asyncio import AsyncSession
+from pydantic import BaseModel
 
 from app.core.database import get_db
 from app.core.security import get_current_user
 from app.core.config import settings
 from app.models.user import User
-from app.workers.tasks import process_activity_file, process_garmin_health_zip
+from app.workers.tasks import process_activity_file, process_garmin_health_zip, analyze_strava_export
 
 router = APIRouter()
 
@@ -187,13 +188,39 @@ async def upload_garmin_export(
     }
 
 
+def _collect_activity_files(extracted: list[Path]) -> list[Path]:
+    """From extracted archive members, gunzip any .gz wrappers (Strava ships most
+    activities as <id>.fit.gz / .gpx.gz / .tcx.gz) and return the .fit/.gpx/.tcx
+    paths ready to parse. Without this, the gzipped majority is silently skipped."""
+    files: list[Path] = []
+    for path in extracted:
+        if path.suffix.lower() == ".gz":
+            inner = _gunzip(path)
+            if inner is None:
+                continue
+            path = inner
+        if path.suffix.lower() in (".fit", ".gpx", ".tcx"):
+            files.append(path)
+    return files
+
+
+class StravaConfirmIn(BaseModel):
+    token: str
+
+
 @router.post("/strava-export")
 async def upload_strava_export(
     file: UploadFile = File(...),
+    dry_run: bool = False,
     db: AsyncSession = Depends(get_db),
     current_user: User = Depends(get_current_user),
 ):
-    """Upload a Strava bulk export ZIP (contains activities/ folder with GPX/FIT files)."""
+    """Upload a Strava bulk export ZIP (activities/ folder with GPX/FIT/TCX files).
+
+    With `dry_run=true` nothing is imported: the archive is extracted and a single
+    analysis task reports how many activities are new vs. already present (Garmin
+    data is preferred), returning a `token` to confirm the real import with.
+    Otherwise activities import straight away (existing data is kept on a match)."""
     if not file.filename.endswith(".zip"):
         raise HTTPException(status_code=400, detail="Please upload a .zip Strava export")
 
@@ -202,7 +229,6 @@ async def upload_strava_export(
 
     extract_dir = dest_dir / f"strava_{dest.stem}"
 
-    task_ids = []
     try:
         with zipfile.ZipFile(dest) as zf:
             extracted = _safe_extract(zf, extract_dir)
@@ -210,27 +236,25 @@ async def upload_strava_export(
         dest.unlink(missing_ok=True)
         raise HTTPException(status_code=400, detail="Uploaded file is not a valid ZIP archive")
 
-    for path in extracted:
-        # Strava compresses most exported activities as <id>.fit.gz / .gpx.gz /
-        # .tcx.gz. Decompress those to their underlying file first, otherwise the
-        # gzipped majority of the archive is silently skipped.
-        if path.suffix.lower() == ".gz":
-            inner = _gunzip(path)
-            if inner is None:
-                continue
-            path = inner
-
-        suffix = path.suffix.lower()
-        if suffix in (".fit", ".gpx", ".tcx"):
-            task = process_activity_file.delay(str(path), current_user.id, suffix[1:])
-            task_ids.append(task.id)
-
-    if not task_ids:
+    files = _collect_activity_files(extracted)
+    if not files:
         raise HTTPException(
             status_code=400,
             detail="No activity files (.fit, .gpx or .tcx) found in this Strava archive",
         )
 
+    if dry_run:
+        # Preview only — classify new/duplicate without writing. The extracted
+        # files are kept so the confirm step can import them without re-uploading.
+        task = analyze_strava_export.delay([str(p) for p in files], current_user.id)
+        return {"status": "analyzing", "task_id": task.id, "token": dest.stem,
+                "activity_files": len(files)}
+
+    task_ids = [
+        process_activity_file.delay(str(p), current_user.id, p.suffix.lower()[1:],
+                                    prefer_existing=True).id
+        for p in files
+    ]
     return {
         "status": "queued",
         "activity_tasks": len(task_ids),
@@ -238,6 +262,32 @@ async def upload_strava_export(
     }
 
 
+@router.post("/strava-export/confirm")
+async def confirm_strava_export(
+    body: StravaConfirmIn,
+    current_user: User = Depends(get_current_user),
+):
+    """Import the activities from a previously previewed (dry-run) Strava export.
+    `token` is the value returned by the dry-run upload; the already-extracted
+    files are imported with Garmin data preferred over Strava duplicates."""
+    token = _safe_name(body.token)
+    extract_dir = Path(settings.file_store_path) / str(current_user.id) / "exports" / f"strava_{token}"
+    if not extract_dir.is_dir():
+        raise HTTPException(status_code=404, detail="Preview not found — please upload and preview again")
+
+    files = [p for p in extract_dir.rglob("*")
+             if p.is_file() and p.suffix.lower() in (".fit", ".gpx", ".tcx")]
+    if not files:
+        raise HTTPException(status_code=400, detail="No activity files found for this preview")
+
+    task_ids = [
+        process_activity_file.delay(str(p), current_user.id, p.suffix.lower()[1:],
+                                    prefer_existing=True).id
+        for p in files
+    ]
+    return {"status": "queued", "activity_tasks": len(task_ids)}
+
+
 @router.get("/task/{task_id}")
 async def check_task_status(
     task_id: str,

backend/app/workers/tasks.py

@@ -90,11 +90,14 @@ def _apply_garmin_summary(parsed: dict, summary: dict):
 
 @celery_app.task(bind=True, name="process_activity_file")
 def process_activity_file(self, file_path: str, user_id: int, source_type: str,
-                          garmin_activity_id: str = None, summary: dict = None):
+                          garmin_activity_id: str = None, summary: dict = None,
+                          prefer_existing: bool = False):
     """Parse a FIT/GPX file. Routes wellness files to health parser.
 
     `summary` (optional, from Garmin Connect sync) carries Garmin's corrected
-    distance/moving/elapsed values which override the raw FIT figures."""
+    distance/moving/elapsed values which override the raw FIT figures.
+    `prefer_existing` (Strava-export imports) keeps an existing same-time
+    activity instead of adding a duplicate — Garmin data wins."""
 
     if is_wellness_file(file_path):
         parse_wellness_fit.delay(file_path, user_id)
@@ -134,20 +137,46 @@ def process_activity_file(self, file_path: str, user_id: int, source_type: str,
             db, user_id, parsed,
             source_file=file_path,
             garmin_activity_id=garmin_activity_id,
+            prefer_existing=prefer_existing,
         )
 
 
+def _find_existing_activity(db, user_id: int, sport_type: str, start_time,
+                            prefer_existing: bool = False):
+    """Find an already-imported activity that matches `start_time` (±60s) for this
+    user. Normally also requires the same sport_type. When `prefer_existing` is
+    set (Strava-export imports), the sport_type filter is dropped so a Strava
+    file is treated as a duplicate of an existing activity — typically the Garmin
+    original — even when the two sources labelled the sport differently. This is
+    what makes Garmin data win over Strava re-imports."""
+    from app.models.user import Activity
+    from sqlalchemy import select
+    from datetime import timedelta
+
+    conds = [
+        Activity.user_id == user_id,
+        Activity.start_time >= start_time - timedelta(seconds=60),
+        Activity.start_time <= start_time + timedelta(seconds=60),
+    ]
+    if not prefer_existing:
+        conds.append(Activity.sport_type == sport_type)
+    return db.execute(select(Activity).where(*conds)).scalars().first()
+
+
 def persist_activity(db, user_id: int, parsed: dict, *, source_file: str = None,
-                     garmin_activity_id: str = None, strava_activity_id: str = None) -> dict:
+                     garmin_activity_id: str = None, strava_activity_id: str = None,
+                     prefer_existing: bool = False) -> dict:
     """Insert a parsed activity (+ data points, laps) and dispatch the
     PR/route/segment follow-up tasks. Shared by the file-upload path
     (process_activity_file) and the Strava API sync, so both get identical
     dedup, HR-zone and downstream behaviour. `parsed` is the fit_parser-shaped
-    dict. Returns {"activity_id", "status"}."""
+    dict. `prefer_existing` skips this activity if any existing one matches by
+    time alone (used by Strava imports to keep the Garmin original).
+    Returns {"activity_id", "status"}."""
     from app.models.user import Activity, ActivityDataPoint, ActivityLap, User as UserModel
     from app.services.fit_parser import calculate_hr_zones
     from sqlalchemy import select
-    from datetime import datetime, timedelta
+    from datetime import datetime
 
     start_time = datetime.fromisoformat(parsed["start_time"])
 
@@ -159,16 +188,11 @@ def persist_activity(db, user_id: int, parsed: dict, *, source_file: str = None,
         if existing:
             return {"activity_id": existing.id, "status": "duplicate"}
 
-    # Deduplicate across sources: same user + sport_type + start_time within ±60s.
-    # This also collapses an activity synced from both Garmin and Strava into one.
-    existing = db.execute(
-        select(Activity).where(
-            Activity.user_id == user_id,
-            Activity.sport_type == parsed["sport_type"],
-            Activity.start_time >= start_time - timedelta(seconds=60),
-            Activity.start_time <= start_time + timedelta(seconds=60),
-        )
-    ).scalars().first()
+    # Deduplicate across sources (see _find_existing_activity). This collapses an
+    # activity present from both Garmin and Strava into the one already stored.
+    existing = _find_existing_activity(
+        db, user_id, parsed["sport_type"], start_time, prefer_existing=prefer_existing,
+    )
 
     if existing:
         # Stamp the external id so future syncs skip straight to the fast path.
@@ -287,6 +311,63 @@ def persist_activity(db, user_id: int, parsed: dict, *, source_file: str = None,
     return {"activity_id": activity_id, "status": "ok"}
 
 
+@celery_app.task(name="analyze_strava_export")
+def analyze_strava_export(file_paths: list, user_id: int):
+    """Dry-run for a Strava bulk export: parse each activity file and classify it
+    as new / duplicate / unreadable against existing activities, WITHOUT writing
+    anything. Uses the same Garmin-priority dedup the real import uses, so the
+    preview matches what the import would actually do."""
+    from app.services.fit_parser import parse_fit_file, parse_gpx_file, parse_tcx_file
+    from app.core.database import SyncSessionLocal
+    from datetime import datetime
+
+    new = duplicate = unreadable = 0
+    new_samples = []
+    dup_samples = []
+
+    with SyncSessionLocal() as db:
+        for fp in file_paths:
+            try:
+                low = fp.lower()
+                if low.endswith(".fit"):
+                    parsed = parse_fit_file(fp)
+                elif low.endswith(".tcx"):
+                    parsed = parse_tcx_file(fp)
+                else:
+                    parsed = parse_gpx_file(fp)
+                start_str = parsed.get("start_time")
+                if not start_str:
+                    unreadable += 1
+                    continue
+                start_time = datetime.fromisoformat(start_str)
+            except Exception:
+                unreadable += 1
+                continue
+
+            existing = _find_existing_activity(
+                db, user_id, parsed.get("sport_type"), start_time, prefer_existing=True,
+            )
+            label = parsed.get("name") or start_str
+            if existing:
+                duplicate += 1
+                if len(dup_samples) < 8:
+                    dup_samples.append(label)
+            else:
+                new += 1
+                if len(new_samples) < 8:
+                    new_samples.append(label)
+
+    return {
+        "status": "ok",
+        "total": new + duplicate + unreadable,
+        "new": new,
+        "duplicate": duplicate,
+        "unreadable": unreadable,
+        "new_samples": new_samples,
+        "duplicate_samples": dup_samples,
+    }
+
+
 @celery_app.task(name="parse_wellness_fit")
 def parse_wellness_fit(file_path: str, user_id: int):
     """Parse a Garmin wellness FIT file and upsert into health_metrics."""