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.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) 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.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})