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

Add activity file downloads with per-activity download_disabled flag

Browse Source 
New endpoint: GET /api/activity/{id}/download/{bas|original|gpx}
- bas: streams the BAS detail JSON as an attachment
- original: streams the original FIT or GPX file from originals/
- gpx: generates a GPX from the timeseries (always available when GPS exists)

download_disabled flag stored in sidecar (edits/{id}.md), propagated to
the merged BAS detail JSON. When set, only the owner can download.

Backend: ops.py writes flag to sidecar; merge.py propagates it to detail
JSON; download.py implements the endpoint; server.py registers the router.
Frontend: EditDrawer gets a "No download" toggle button; ActivityDetail
shows a Download section (hidden when disabled and viewer is not the owner).
This commit is contained in:
Davide Scaini
2026-05-15 18:35:40 +02:00
parent fe437626e6
commit c465e518e5
8 changed files with 227 additions and 3 deletions
Download Patch File Download Diff File Expand all files Collapse all files
bincio/edit/ops.py
+2
View File
@@ -50,6 +50,8 @@ def apply_sidecar_edit(activity_id: str, payload: dict[str, Any], data_dir: Path
hide = [s for s in (payload.get("hide_stats") or []) if s in STAT_PANELS]
if hide:
lines.append(f"hide_stats: [{', '.join(hide)}]")
if payload.get("download_disabled"):
lines.append("download_disabled: true")
description = (payload.get("description") or "").strip()
bincio/render/merge.py
+2
View File
@@ -97,6 +97,8 @@ def apply_sidecar(detail: dict, fm: dict, body: str) -> dict:
d["privacy"] = "unlisted" if fm["private"] else detail.get("privacy", "public")
if "hide_stats" in fm:
d["custom"]["hide_stats"] = [str(s) for s in (fm["hide_stats"] or [])]
if "download_disabled" in fm:
d["download_disabled"] = True
return d
bincio/serve/models.py
+1
View File
@@ -56,6 +56,7 @@ class ActivityEditRequest(BaseModel):
private: bool | None = Field(default=None, description="Hide from public feed")
highlight: bool | None = Field(default=None, description="Mark as favorite")
gear: str | None = Field(default=None, description="Gear used")
download_disabled: bool | None = Field(default=None, description="Prevent others from downloading files")
class ActivityEditResponse(BaseModel):
bincio/serve/routers/download.py
+173
View File
@@ -0,0 +1,173 @@
"""Activity file download endpoint.
GET /api/activity/{activity_id}/download/{fmt}
fmt: bas | original | gpx
Permission:
- If activity.download_disabled is true: only the owner (authenticated) may download.
- Otherwise: no auth required — anyone who can see the activity can download.
"""
from __future__ import annotations
import json
from datetime import datetime, timedelta, timezone
from pathlib import Path
from typing import Optional
from fastapi import APIRouter, Cookie, HTTPException
from fastapi.responses import FileResponse, Response
from bincio.serve import deps
router = APIRouter()
def _find_activity(activity_id: str) -> tuple[str, Path] | None:
"""Return (handle, detail_path) for whichever user owns this activity."""
data_dir = deps._get_data_dir()
for user_dir in sorted(data_dir.iterdir()):
if not user_dir.is_dir() or user_dir.name.startswith(("_", ".")):
continue
for base in (user_dir / "_merged" / "activities", user_dir / "activities"):
p = base / f"{activity_id}.json"
if p.exists():
return user_dir.name, p
return None
def _check_download_permission(
detail: dict, handle: str, bincio_session: Optional[str]
) -> None:
if not detail.get("download_disabled"):
return
try:
user = deps._require_user(bincio_session)
except HTTPException:
raise HTTPException(403, "Downloads are disabled for this activity")
if user.handle != handle:
raise HTTPException(403, "Downloads are disabled for this activity")
def _generate_gpx(detail: dict, ts: dict) -> str:
t_vals = ts.get("t") or []
lat_vals = ts.get("lat") or []
lon_vals = ts.get("lon") or []
ele_vals = ts.get("elevation_m") or []
hr_vals = ts.get("hr_bpm") or []
title = (detail.get("title") or "Activity").replace("&", "&amp;").replace("<", "&lt;").replace(">", "&gt;")
started = detail.get("started_at") or "1970-01-01T00:00:00+00:00"
try:
t0 = datetime.fromisoformat(started)
except ValueError:
t0 = datetime(1970, 1, 1, tzinfo=timezone.utc)
lines = [
'<?xml version="1.0" encoding="UTF-8"?>',
'<gpx version="1.1" creator="bincio"'
' xmlns="http://www.topografix.com/GPX/1/1"'
' xmlns:gpxtpx="http://www.garmin.com/xmlschemas/TrackPointExtension/v1">',
f' <trk><name>{title}</name><trkseg>',
]
for i, t in enumerate(t_vals):
lat = lat_vals[i] if i < len(lat_vals) else None
lon = lon_vals[i] if i < len(lon_vals) else None
if lat is None or lon is None:
continue
ele = ele_vals[i] if i < len(ele_vals) else None
hr = hr_vals[i] if i < len(hr_vals) else None
ts_str = (t0 + timedelta(seconds=t)).strftime("%Y-%m-%dT%H:%M:%SZ")
trkpt = f' <trkpt lat="{lat}" lon="{lon}">'
if ele is not None:
trkpt += f"<ele>{ele}</ele>"
trkpt += f"<time>{ts_str}</time>"
if hr is not None:
trkpt += (
f"<extensions><gpxtpx:TrackPointExtension>"
f"<gpxtpx:hr>{hr}</gpxtpx:hr>"
f"</gpxtpx:TrackPointExtension></extensions>"
)
trkpt += "</trkpt>"
lines.append(trkpt)
lines += [" </trkseg></trk>", "</gpx>"]
return "\n".join(lines)
@router.get("/api/activity/{activity_id}/download/{fmt}")
async def download_activity(
activity_id: str,
fmt: str,
bincio_session: Optional[str] = Cookie(default=None),
) -> Response:
deps._check_id(activity_id)
if fmt not in ("bas", "original", "gpx"):
raise HTTPException(400, "fmt must be bas, original, or gpx")
result = _find_activity(activity_id)
if result is None:
raise HTTPException(404, "Activity not found")
handle, detail_path = result
detail = json.loads(detail_path.read_text(encoding="utf-8"))
_check_download_permission(detail, handle, bincio_session)
if fmt == "bas":
return FileResponse(
detail_path,
media_type="application/json",
filename=f"{activity_id}.json",
headers={"Content-Disposition": f'attachment; filename="{activity_id}.json"'},
)
if fmt == "original":
source = detail.get("source") or ""
source_file = detail.get("source_file") or ""
if source not in ("fit_file", "gpx_file") or not source_file:
raise HTTPException(404, "No original file available for this activity")
safe_name = Path(source_file).name # strip any directory traversal
orig_path = deps._get_data_dir() / handle / "originals" / safe_name
if not orig_path.exists():
raise HTTPException(404, "Original file not found on disk")
media_type = "application/octet-stream"
if safe_name.endswith(".fit"):
media_type = "application/vnd.ant.fit"
elif safe_name.endswith(".gpx"):
media_type = "application/gpx+xml"
return FileResponse(
orig_path,
media_type=media_type,
filename=safe_name,
headers={"Content-Disposition": f'attachment; filename="{safe_name}"'},
)
# fmt == "gpx"
data_dir = deps._get_data_dir()
ts_path: Path | None = None
for base in (
data_dir / handle / "_merged" / "activities",
data_dir / handle / "activities",
):
p = base / f"{activity_id}.timeseries.json"
if p.exists():
ts_path = p
break
if ts_path is None:
raise HTTPException(404, "No GPS data available for this activity")
ts = json.loads(ts_path.read_text(encoding="utf-8"))
lat_vals = ts.get("lat") or []
if not any(v is not None for v in lat_vals):
raise HTTPException(404, "No GPS data available for this activity")
gpx_content = _generate_gpx(detail, ts)
raw_title = detail.get("title") or activity_id
safe_title = "".join(c for c in raw_title if c.isalnum() or c in " -_")[:50].strip()
filename = f"{safe_title or activity_id}.gpx"
return Response(
content=gpx_content,
media_type="application/gpx+xml",
headers={"Content-Disposition": f'attachment; filename="{filename}"'},
)
bincio/serve/server.py
+2
View File
@@ -20,6 +20,7 @@ from bincio.serve.routers import (
activities,
admin,
auth,
download,
feed,
garmin,
ideas,
@@ -60,6 +61,7 @@ for _router in [
me.router,
admin.router,
activities.router,
download.router,
uploads.router,
segments.router,
strava.router,