Edit Activity

"""FastAPI edit server — serves the activity edit UI and writes sidecar .md files.""" from __future__ import annotations import json 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 the Astro dev server (and any local origin) to call the write API app.add_middleware( CORSMiddleware, allow_origins=["*"], allow_methods=["*"], allow_headers=["*"], ) 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() 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() 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 = payload.get("hide_stats") or [] if hide: lines.append(f"hide_stats: [{', '.join(str(s) for s in 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() 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", ) # Patch athlete.json in-place (preserves power_curve, updated_at, etc.) data = json.loads(athlete_path.read_text(encoding="utf-8")) data.update(overrides) athlete_path.write_text(json.dumps(data, indent=2, ensure_ascii=False)) # Re-merge so _merged/athlete.json symlink stays valid 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 {} @app.delete("/api/activity/{activity_id}/images/{filename}") async def delete_image(activity_id: str, filename: str) -> JSONResponse: dd = _get_data_dir() target = dd / "edits" / "images" / activity_id / filename 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})