brut/bincio-activity
1
0
Fork 0
Code Issues Pull Requests Actions 3 Packages Projects Releases Wiki Activity

feat: gear registry — manage bikes/shoes per athlete, set per activity

Browse Source 
- New /api/gear CRUD endpoints (gear.json per user)
- Gear tab in AthleteView (owner-only): add, edit, retire items
- EditDrawer gear field becomes a dropdown when registry has items
- Strava API sync now resolves gear_id → name, adds to registry automatically
- Strava ZIP import reads Gear column from activities.csv
- POST /api/strava/import-gear for one-time backfill from stored originals
This commit is contained in:
Davide Scaini
2026-05-24 12:33:41 +02:00
parent aca9f79b46
commit e553e08663
9 changed files with 576 additions and 10 deletions
Download Patch File Download Diff File Expand all files Collapse all files
bincio/extract/ingest.py
+38
View File
@@ -100,18 +100,22 @@ def strava_sync_iter(
- ``"done"`` — final summary; keys: imported, skipped, error_count, errors
- ``"error"`` — fatal error before processing started; key: message
"""
import contextlib
import time
import uuid
from bincio.extract.strava_api import (
StravaError,
ensure_fresh,
fetch_activities,
fetch_gear,
fetch_streams,
save_token,
strava_meta_to_partial,
strava_to_parsed,
)
from bincio.extract.writer import make_activity_id
from bincio.serve.routers.gear import _load as _gear_load, _save as _gear_save
if not client_id or not client_secret:
yield {"type": "error", "message": "Strava not configured"}
@@ -137,6 +141,35 @@ def strava_sync_iter(
skipped = 0
errors: list[str] = []
# Cache: strava gear_id → gear name (avoid duplicate API calls within one sync)
_gear_name_cache: dict[str, str] = {}
def _resolve_gear(gear_id: str) -> str:
"""Return gear name for a Strava gear_id, adding to registry if new."""
if gear_id in _gear_name_cache:
return _gear_name_cache[gear_id]
# Check registry first
registry = _gear_load(data_dir)
existing = next((g for g in registry if g.get("strava_id") == gear_id), None)
if existing:
name = existing["name"]
_gear_name_cache[gear_id] = name
return name
# Fetch from Strava
details = fetch_gear(token["access_token"], gear_id)
name = details.get("name") or ""
if not name:
_gear_name_cache[gear_id] = ""
return ""
# Determine type from Strava: primary_type "A" = bike, "B" = shoe
gear_type = "shoes" if details.get("primary_type") == "B" else "bike"
# Add to registry
new_item: dict = {"id": str(uuid.uuid4()), "name": name, "type": gear_type, "retired": False, "strava_id": gear_id}
registry.append(new_item)
_gear_save(data_dir, registry)
_gear_name_cache[gear_id] = name
return name
for n, meta in enumerate(activities, 1):
name = meta.get("name", "Untitled")
try:
@@ -146,6 +179,11 @@ def strava_sync_iter(
yield {"type": "progress", "n": n, "total": total, "name": name, "status": "skipped"}
continue
streams = fetch_streams(token["access_token"], meta["id"])
# Resolve gear name before converting
gear_id = meta.get("gear_id") or ""
if gear_id:
with contextlib.suppress(Exception):
meta["_gear_name"] = _resolve_gear(gear_id)
if originals_dir is not None:
orig_path = originals_dir / f"{activity_id}.json"
orig_path.write_text(
bincio/extract/strava_api.py
+10
View File
@@ -150,6 +150,15 @@ def fetch_streams(access_token: str, activity_id: int) -> dict:
return result if isinstance(result, dict) else {}
def fetch_gear(access_token: str, gear_id: str) -> dict:
"""Fetch gear details for a single gear item. Returns {} on error."""
try:
result = _api_get(f"{_API_BASE}/gear/{gear_id}", access_token)
return result if isinstance(result, dict) else {}
except StravaError:
return {}
# ── Model conversion ───────────────────────────────────────────────────────────
def strava_meta_to_partial(meta: dict) -> ParsedActivity:
@@ -215,4 +224,5 @@ def strava_to_parsed(meta: dict, streams: dict) -> ParsedActivity:
description=meta.get("description") or None,
strava_id=str(meta["id"]),
privacy="unlisted" if is_private else "public",
gear=meta.get("_gear_name") or None,
)
bincio/extract/strava_zip.py
+2
View File
@@ -115,6 +115,8 @@ def strava_zip_iter(
parsed.description = meta_row["Activity Description"].strip()
if not parsed.strava_id and meta_row.get("Activity ID"):
parsed.strava_id = meta_row["Activity ID"].strip()
if not parsed.gear and meta_row.get("Gear"):
parsed.gear = meta_row["Gear"].strip()
if originals_dir is not None:
import shutil
bincio/serve/routers/gear.py
+134
View File
@@ -0,0 +1,134 @@
"""Gear registry endpoints (/api/gear)."""
from __future__ import annotations
import json
import uuid
from pathlib import Path
from fastapi import APIRouter, Cookie, HTTPException, Request
from fastapi.responses import JSONResponse
from bincio.serve import deps
router = APIRouter()
_GEAR_TYPES = {"bike", "shoes", "skis", "other"}
def _gear_path(user_dir: Path) -> Path:
return user_dir / "gear.json"
def _load(user_dir: Path) -> list[dict]:
p = _gear_path(user_dir)
if not p.exists():
return []
try:
data = json.loads(p.read_text(encoding="utf-8"))
return data.get("items", [])
except (OSError, json.JSONDecodeError):
return []
def _save(user_dir: Path, items: list[dict]) -> None:
_gear_path(user_dir).write_text(
json.dumps({"items": items}, indent=2, ensure_ascii=False),
encoding="utf-8",
)
@router.get("/api/gear")
async def gear_list(bincio_session: str | None = Cookie(default=None)) -> JSONResponse:
user = deps._require_user(bincio_session)
items = _load(deps._get_data_dir() / user.handle)
return JSONResponse({"items": items})
@router.post("/api/gear")
async def gear_add(
request: Request,
bincio_session: str | None = Cookie(default=None),
) -> JSONResponse:
user = deps._require_user(bincio_session)
body = await request.json()
name = str(body.get("name", "")).strip()
if not name:
raise HTTPException(400, "name is required")
gear_type = str(body.get("type", "other")).strip()
if gear_type not in _GEAR_TYPES:
raise HTTPException(400, f"type must be one of: {', '.join(sorted(_GEAR_TYPES))}")
strava_id = str(body.get("strava_id", "")).strip() or None
user_dir = deps._get_data_dir() / user.handle
items = _load(user_dir)
# Deduplicate by strava_id if provided
if strava_id and any(i.get("strava_id") == strava_id for i in items):
existing = next(i for i in items if i.get("strava_id") == strava_id)
return JSONResponse({"ok": True, "item": existing, "created": False})
item: dict = {
"id": str(uuid.uuid4()),
"name": name,
"type": gear_type,
"retired": False,
}
if strava_id:
item["strava_id"] = strava_id
items.append(item)
_save(user_dir, items)
return JSONResponse({"ok": True, "item": item, "created": True}, status_code=201)
@router.patch("/api/gear/{item_id}")
async def gear_update(
item_id: str,
request: Request,
bincio_session: str | None = Cookie(default=None),
) -> JSONResponse:
user = deps._require_user(bincio_session)
user_dir = deps._get_data_dir() / user.handle
items = _load(user_dir)
idx = next((i for i, g in enumerate(items) if g["id"] == item_id), None)
if idx is None:
raise HTTPException(404, "Gear item not found")
body = await request.json()
item = dict(items[idx])
if "name" in body:
name = str(body["name"]).strip()
if not name:
raise HTTPException(400, "name cannot be empty")
item["name"] = name
if "type" in body:
gear_type = str(body["type"]).strip()
if gear_type not in _GEAR_TYPES:
raise HTTPException(400, f"type must be one of: {', '.join(sorted(_GEAR_TYPES))}")
item["type"] = gear_type
if "retired" in body:
item["retired"] = bool(body["retired"])
items[idx] = item
_save(user_dir, items)
return JSONResponse({"ok": True, "item": item})
@router.delete("/api/gear/{item_id}")
async def gear_delete(
item_id: str,
bincio_session: str | None = Cookie(default=None),
) -> JSONResponse:
user = deps._require_user(bincio_session)
user_dir = deps._get_data_dir() / user.handle
items = _load(user_dir)
before = len(items)
items = [g for g in items if g["id"] != item_id]
if len(items) == before:
raise HTTPException(404, "Gear item not found")
_save(user_dir, items)
return JSONResponse({"ok": True})
bincio/serve/routers/strava.py
+99
View File
@@ -171,6 +171,105 @@ async def serve_strava_sync_stream(bincio_session: Optional[str] = Cookie(defaul
)
@router.post("/api/strava/import-gear")
async def serve_strava_import_gear(bincio_session: Optional[str] = Cookie(default=None)) -> JSONResponse:
"""One-time backfill: scan stored Strava originals for gear_ids, fetch names, populate gear registry."""
user = deps._require_user(bincio_session)
cid, csec = deps._strava_creds(user.handle)
if not cid or not csec:
raise HTTPException(400, "Strava not configured on this server")
dd = deps._get_data_dir() / user.handle
originals_dir = dd / "originals" / "strava"
if not originals_dir.exists():
return JSONResponse({"ok": True, "gear_added": 0, "activities_updated": 0, "message": "No stored originals found"})
import contextlib
import uuid
from bincio.extract.strava_api import StravaError, ensure_fresh, fetch_gear
from bincio.render.merge import merge_one
from bincio.serve.routers.gear import _load as _gear_load, _save as _gear_save
try:
token = ensure_fresh(dd, cid, csec)
except StravaError as e:
raise HTTPException(502, str(e))
registry = _gear_load(dd)
known_strava_ids = {g.get("strava_id") for g in registry if g.get("strava_id")}
# Collect all unique gear_ids from originals
gear_id_to_activities: dict[str, list[str]] = {}
for orig_path in originals_dir.glob("*.json"):
try:
data = json.loads(orig_path.read_text(encoding="utf-8"))
gear_id = (data.get("meta") or {}).get("gear_id") or ""
if gear_id:
gear_id_to_activities.setdefault(gear_id, []).append(orig_path.stem)
except (OSError, json.JSONDecodeError):
continue
gear_added = 0
activities_updated = 0
for gear_id, activity_ids in gear_id_to_activities.items():
if gear_id in known_strava_ids:
gear_name = next(g["name"] for g in registry if g.get("strava_id") == gear_id)
else:
details = fetch_gear(token["access_token"], gear_id)
gear_name = details.get("name") or ""
if not gear_name:
continue
gear_type = "shoes" if details.get("primary_type") == "B" else "bike"
new_item: dict = {"id": str(uuid.uuid4()), "name": gear_name, "type": gear_type, "retired": False, "strava_id": gear_id}
registry.append(new_item)
known_strava_ids.add(gear_id)
gear_added += 1
# Backfill: write sidecar for each activity that has no gear set yet
import yaml as _yaml
edits_dir = dd / "edits"
edits_dir.mkdir(exist_ok=True)
for activity_id in activity_ids:
activity_json = dd / "activities" / f"{activity_id}.json"
if not activity_json.exists():
continue
try:
act = json.loads(activity_json.read_text(encoding="utf-8"))
if act.get("gear"):
continue # already has gear
except (OSError, json.JSONDecodeError):
continue
sidecar = edits_dir / f"{activity_id}.md"
fm: dict = {}
body = ""
if sidecar.exists():
try:
text = sidecar.read_text(encoding="utf-8")
import re as _re
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 # sidecar already sets gear
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(dd, activity_id)
activities_updated += 1
_gear_save(dd, registry)
tasks._trigger_rebuild(user.handle)
return JSONResponse({"ok": True, "gear_added": gear_added, "activities_updated": activities_updated})
@router.post("/api/strava/sync")
async def serve_strava_sync(bincio_session: Optional[str] = Cookie(default=None)) -> JSONResponse:
user = deps._require_user(bincio_session)
bincio/serve/server.py
+2
View File
@@ -23,6 +23,7 @@ from bincio.serve.routers import (
download,
feed,
garmin,
gear,
ideas,
me,
ogimage,
@@ -69,5 +70,6 @@ for _router in [
garmin.router,
ideas.router,
ogimage.router,
gear.router,
]:
app.include_router(_router)