Edit Activity

"""FastAPI edit server — serves the activity edit UI and writes sidecar .md files.""" from __future__ import annotations import json import re import shutil from pathlib import Path from typing import Any from fastapi import FastAPI, File, HTTPException, UploadFile from fastapi.middleware.cors import CORSMiddleware from fastapi.responses import HTMLResponse, JSONResponse, RedirectResponse # Populated by the CLI before uvicorn starts data_dir: Path | None = None site_url: str = "http://localhost:4321" app = FastAPI(title="BincioActivity Edit Server", docs_url=None, redoc_url=None) # Allow localhost origins only — this server is never meant to be public app.add_middleware( CORSMiddleware, allow_origin_regex=r"https?://localhost(:\d+)?", allow_methods=["GET", "POST", "DELETE"], allow_headers=["Content-Type"], ) _VALID_ACTIVITY_ID = re.compile(r'^[a-zA-Z0-9][a-zA-Z0-9\-]{0,250}$') def _check_id(activity_id: str) -> str: """Reject activity IDs that contain path traversal sequences.""" if not _VALID_ACTIVITY_ID.match(activity_id): raise HTTPException(400, "Invalid activity ID") return activity_id SPORTS = ["cycling", "running", "hiking", "walking", "swimming", "skiing", "other"] STAT_PANELS = ["elevation", "speed", "heart_rate", "cadence", "power"] # ── HTML UI ─────────────────────────────────────────────────────────────────── _HTML = """\ Edit Activity

""" # ── Routes ──────────────────────────────────────────────────────────────────── def _get_data_dir() -> Path: if data_dir is None: raise HTTPException(500, "Edit server not configured (data_dir is None)") return data_dir @app.get("/") async def root() -> RedirectResponse: return RedirectResponse(url=site_url) @app.get("/edit/{activity_id}", response_class=HTMLResponse) async def edit_page(activity_id: str) -> str: sport_opts = "\n".join( f'' for s in SPORTS ) stat_cbs = "\n".join( f' {s.replace("_", " ").capitalize()}' for s in STAT_PANELS ) html = ( _HTML .replace("__SITE_URL__", site_url) .replace("__SPORT_OPTIONS__", sport_opts) .replace("__STAT_CHECKBOXES__", stat_cbs) ) return html @app.get("/api/activity/{activity_id}") async def get_activity(activity_id: str) -> JSONResponse: dd = _get_data_dir() _check_id(activity_id) json_path = dd / "activities" / f"{activity_id}.json" if not json_path.exists(): raise HTTPException(404, f"Activity {activity_id!r} not found") detail: dict[str, Any] = json.loads(json_path.read_text(encoding="utf-8")) # Read existing sidecar if any — these are the "user" values shown in the form from bincio.render.merge import parse_sidecar sidecar_path = dd / "edits" / f"{activity_id}.md" fm: dict = {} body = "" if sidecar_path.exists(): fm, body = parse_sidecar(sidecar_path) # Existing uploaded images for this activity images_dir = dd / "edits" / "images" / activity_id images = sorted(p.name for p in images_dir.iterdir() if p.is_file()) if images_dir.exists() else [] return JSONResponse({ "id": activity_id, "started_at": detail.get("started_at", ""), "title": fm.get("title", detail.get("title", "")), "sport": fm.get("sport", detail.get("sport", "other")), "gear": fm.get("gear", detail.get("gear") or ""), "description": body or fm.get("description") or detail.get("description") or "", "highlight": fm.get("highlight", detail.get("custom", {}).get("highlight", False)), "private": fm.get("private", detail.get("privacy") == "private"), "hide_stats": fm.get("hide_stats", detail.get("custom", {}).get("hide_stats", [])), "images": images, }) @app.post("/api/activity/{activity_id}") async def save_activity(activity_id: str, payload: dict[str, Any]) -> JSONResponse: dd = _get_data_dir() _check_id(activity_id) if not (dd / "activities" / f"{activity_id}.json").exists(): raise HTTPException(404, f"Activity {activity_id!r} not found") edits_dir = dd / "edits" edits_dir.mkdir(exist_ok=True) sidecar_path = edits_dir / f"{activity_id}.md" lines: list[str] = [] if payload.get("title"): lines.append(f"title: {json.dumps(payload['title'])}") if payload.get("sport") and payload["sport"] != "other": lines.append(f"sport: {payload['sport']}") if payload.get("gear"): lines.append(f"gear: {json.dumps(payload['gear'])}") if payload.get("highlight"): lines.append("highlight: true") if payload.get("private"): lines.append("private: true") hide = [s for s in (payload.get("hide_stats") or []) if s in STAT_PANELS] if hide: lines.append(f"hide_stats: [{', '.join(hide)}]") description = (payload.get("description") or "").strip() content = "---\n" + "\n".join(lines) + "\n---\n" if description: content += "\n" + description + "\n" sidecar_path.write_text(content, encoding="utf-8") # Re-merge so the Astro dev server immediately serves updated data from bincio.render.merge import merge_all merge_all(dd) return JSONResponse({"ok": True, "sidecar": str(sidecar_path)}) @app.post("/api/activity/{activity_id}/images") async def upload_image(activity_id: str, file: UploadFile = File(...)) -> JSONResponse: dd = _get_data_dir() _check_id(activity_id) if not (dd / "activities" / f"{activity_id}.json").exists(): raise HTTPException(404, f"Activity {activity_id!r} not found") if not file.filename: raise HTTPException(400, "No filename") images_dir = dd / "edits" / "images" / activity_id images_dir.mkdir(parents=True, exist_ok=True) dest = images_dir / Path(file.filename).name dest.write_bytes(await file.read()) return JSONResponse({"ok": True, "filename": dest.name}) @app.get("/api/athlete") async def get_athlete() -> JSONResponse: dd = _get_data_dir() athlete_path = dd / "athlete.json" if not athlete_path.exists(): raise HTTPException(404, "athlete.json not found — run bincio extract first") data = json.loads(athlete_path.read_text(encoding="utf-8")) # Layer edits/athlete.yaml overrides on top overrides = _read_athlete_edits(dd) for key in ("max_hr", "ftp_w", "hr_zones", "power_zones", "seasons", "gear"): if key in overrides: data[key] = overrides[key] return JSONResponse({ "max_hr": data.get("max_hr"), "ftp_w": data.get("ftp_w"), "hr_zones": data.get("hr_zones"), "power_zones": data.get("power_zones"), "seasons": data.get("seasons", []), "gear": data.get("gear", {}), }) @app.post("/api/athlete") async def save_athlete(payload: dict[str, Any]) -> JSONResponse: dd = _get_data_dir() athlete_path = dd / "athlete.json" if not athlete_path.exists(): raise HTTPException(404, "athlete.json not found — run bincio extract first") # Write edits/athlete.yaml with validated fields edits_dir = dd / "edits" edits_dir.mkdir(exist_ok=True) overrides: dict[str, Any] = {} if payload.get("max_hr") is not None: overrides["max_hr"] = int(payload["max_hr"]) if payload.get("ftp_w") is not None: overrides["ftp_w"] = int(payload["ftp_w"]) if payload.get("hr_zones") is not None: overrides["hr_zones"] = [[int(lo), int(hi)] for lo, hi in payload["hr_zones"]] if payload.get("power_zones") is not None: overrides["power_zones"] = [[int(lo), int(hi)] for lo, hi in payload["power_zones"]] if payload.get("seasons") is not None: overrides["seasons"] = [ {"name": str(s["name"]), "start": str(s["start"]), "end": str(s["end"])} for s in payload["seasons"] ] if payload.get("gear") is not None: overrides["gear"] = payload["gear"] import yaml (edits_dir / "athlete.yaml").write_text( yaml.dump(overrides, allow_unicode=True, default_flow_style=False), encoding="utf-8", ) # Re-merge — merge_all() applies edits/athlete.yaml on top of athlete.json from bincio.render.merge import merge_all merge_all(dd) return JSONResponse({"ok": True}) def _read_athlete_edits(data_dir: Path) -> dict: path = data_dir / "edits" / "athlete.yaml" if not path.exists(): return {} try: import yaml return yaml.safe_load(path.read_text(encoding="utf-8")) or {} except Exception: return {} _SUPPORTED_SUFFIXES = {".fit", ".gpx", ".tcx", ".fit.gz", ".gpx.gz", ".tcx.gz"} def _file_suffix(name: str) -> str: """Return the effective suffix, including .gz double-extension.""" p = Path(name.lower()) if p.suffix == ".gz": return p.stem.rsplit(".", 1)[-1].join([".", ".gz"]) if "." in p.stem else ".gz" return p.suffix @app.post("/api/upload") async def upload_activity(file: UploadFile = File(...)) -> JSONResponse: """Accept a FIT/GPX/TCX file, extract it, update index.json, and re-merge.""" dd = _get_data_dir() name = Path(file.filename or "upload.fit").name # strip any path components suffix = _file_suffix(name) if suffix not in _SUPPORTED_SUFFIXES: raise HTTPException(400, f"Unsupported file type '{Path(name).suffix}'. Expected FIT, GPX, or TCX.") staging = dd / "_uploads" staging.mkdir(exist_ok=True) staged = staging / name staged.write_bytes(await file.read()) try: from bincio.extract.metrics import compute from bincio.extract.parsers.factory import parse_file from bincio.extract.writer import build_summary, make_activity_id, write_activity, write_index activity = parse_file(staged) metrics = compute(activity) activity_id = make_activity_id(activity) existing_json = dd / "activities" / f"{activity_id}.json" if existing_json.exists(): raise HTTPException(409, f"Activity already exists: {activity_id}") write_activity(activity, metrics, dd, privacy="public", rdp_epsilon=0.0001) summary = build_summary(activity, metrics, activity_id, "public") # Read current index to preserve owner + existing summaries index_path = dd / "index.json" if index_path.exists(): index_data = json.loads(index_path.read_text(encoding="utf-8")) else: index_data = {"owner": {"handle": "unknown"}, "activities": []} owner = index_data.get("owner", {}) existing = {s["id"]: s for s in index_data.get("activities", [])} existing[activity_id] = summary write_index(list(existing.values()), dd, owner) from bincio.render.merge import merge_all merge_all(dd) except HTTPException: raise except Exception as exc: raise HTTPException(422, str(exc)) finally: staged.unlink(missing_ok=True) return JSONResponse({"ok": True, "id": activity_id}) @app.delete("/api/activity/{activity_id}/images/{filename}") async def delete_image(activity_id: str, filename: str) -> JSONResponse: dd = _get_data_dir() _check_id(activity_id) safe_name = Path(filename).name # strip any path traversal if not safe_name: raise HTTPException(400, "Invalid filename") target = dd / "edits" / "images" / activity_id / safe_name if target.exists() and target.is_file(): target.unlink() # Remove empty parent dir if not any(target.parent.iterdir()): shutil.rmtree(target.parent) return JSONResponse({"ok": True})