Davide Scaini
315 lines
10 KiB
Python
315 lines
10 KiB
Python
"""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
|
|
weight_g = body.get("weight_g")
|
|
if weight_g is not None:
|
|
try:
|
|
weight_g = int(weight_g)
|
|
if weight_g < 0:
|
|
raise ValueError
|
|
except (TypeError, ValueError):
|
|
raise HTTPException(400, "weight_g must be a non-negative integer (grams)")
|
|
|
|
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
|
|
if weight_g is not None:
|
|
item["weight_g"] = weight_g
|
|
|
|
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"])
|
|
if "weight_g" in body:
|
|
w = body["weight_g"]
|
|
if w is None:
|
|
item.pop("weight_g", None)
|
|
else:
|
|
try:
|
|
w = int(w)
|
|
if w < 0:
|
|
raise ValueError
|
|
except (TypeError, ValueError):
|
|
raise HTTPException(400, "weight_g must be a non-negative integer (grams)")
|
|
item["weight_g"] = w
|
|
|
|
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})
|
|
|
|
|
|
# ── Parts ─────────────────────────────────────────────────────────────────────
|
|
|
|
def _find_item(items: list[dict], item_id: str) -> tuple[int, dict]:
|
|
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")
|
|
return idx, items[idx]
|
|
|
|
|
|
@router.post("/api/gear/{item_id}/parts")
|
|
async def part_add(
|
|
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, item = _find_item(items, item_id)
|
|
|
|
body = await request.json()
|
|
name = str(body.get("name", "")).strip()
|
|
if not name:
|
|
raise HTTPException(400, "name is required")
|
|
threshold_km = body.get("threshold_km")
|
|
if threshold_km is not None:
|
|
try:
|
|
threshold_km = float(threshold_km)
|
|
except (TypeError, ValueError):
|
|
raise HTTPException(400, "threshold_km must be a number")
|
|
|
|
part: dict = {"id": str(uuid.uuid4()), "name": name, "replacements": []}
|
|
if threshold_km is not None:
|
|
part["threshold_km"] = threshold_km
|
|
|
|
item = dict(item)
|
|
item.setdefault("parts", [])
|
|
item["parts"] = [*item["parts"], part]
|
|
items[idx] = item
|
|
_save(user_dir, items)
|
|
return JSONResponse({"ok": True, "part": part}, status_code=201)
|
|
|
|
|
|
@router.patch("/api/gear/{item_id}/parts/{part_id}")
|
|
async def part_update(
|
|
item_id: str,
|
|
part_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, item = _find_item(items, item_id)
|
|
|
|
parts = list(item.get("parts", []))
|
|
pidx = next((i for i, p in enumerate(parts) if p["id"] == part_id), None)
|
|
if pidx is None:
|
|
raise HTTPException(404, "Part not found")
|
|
|
|
body = await request.json()
|
|
part = dict(parts[pidx])
|
|
if "name" in body:
|
|
name = str(body["name"]).strip()
|
|
if not name:
|
|
raise HTTPException(400, "name cannot be empty")
|
|
part["name"] = name
|
|
if "threshold_km" in body:
|
|
try:
|
|
part["threshold_km"] = float(body["threshold_km"])
|
|
except (TypeError, ValueError):
|
|
raise HTTPException(400, "threshold_km must be a number")
|
|
|
|
parts[pidx] = part
|
|
item = {**item, "parts": parts}
|
|
items[idx] = item
|
|
_save(user_dir, items)
|
|
return JSONResponse({"ok": True, "part": part})
|
|
|
|
|
|
@router.delete("/api/gear/{item_id}/parts/{part_id}")
|
|
async def part_delete(
|
|
item_id: str,
|
|
part_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)
|
|
idx, item = _find_item(items, item_id)
|
|
|
|
parts = [p for p in item.get("parts", []) if p["id"] != part_id]
|
|
items[idx] = {**item, "parts": parts}
|
|
_save(user_dir, items)
|
|
return JSONResponse({"ok": True})
|
|
|
|
|
|
@router.post("/api/gear/{item_id}/parts/{part_id}/replacements")
|
|
async def replacement_add(
|
|
item_id: str,
|
|
part_id: str,
|
|
request: Request,
|
|
bincio_session: str | None = Cookie(default=None),
|
|
) -> JSONResponse:
|
|
"""Log a replacement event for a part. date defaults to today (UTC)."""
|
|
from datetime import UTC, datetime
|
|
|
|
user = deps._require_user(bincio_session)
|
|
user_dir = deps._get_data_dir() / user.handle
|
|
items = _load(user_dir)
|
|
idx, item = _find_item(items, item_id)
|
|
|
|
parts = list(item.get("parts", []))
|
|
pidx = next((i for i, p in enumerate(parts) if p["id"] == part_id), None)
|
|
if pidx is None:
|
|
raise HTTPException(404, "Part not found")
|
|
|
|
body = await request.json()
|
|
date = str(body.get("date", "")).strip() or datetime.now(UTC).strftime("%Y-%m-%d")
|
|
note = str(body.get("note", "")).strip() or None
|
|
|
|
entry: dict = {"id": str(uuid.uuid4()), "date": date}
|
|
if note:
|
|
entry["note"] = note
|
|
|
|
part = dict(parts[pidx])
|
|
part["replacements"] = [*part.get("replacements", []), entry]
|
|
parts[pidx] = part
|
|
items[idx] = {**item, "parts": parts}
|
|
_save(user_dir, items)
|
|
return JSONResponse({"ok": True, "replacement": entry}, status_code=201)
|
|
|
|
|
|
@router.delete("/api/gear/{item_id}/parts/{part_id}/replacements/{replacement_id}")
|
|
async def replacement_delete(
|
|
item_id: str,
|
|
part_id: str,
|
|
replacement_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)
|
|
idx, item = _find_item(items, item_id)
|
|
|
|
parts = list(item.get("parts", []))
|
|
pidx = next((i for i, p in enumerate(parts) if p["id"] == part_id), None)
|
|
if pidx is None:
|
|
raise HTTPException(404, "Part not found")
|
|
|
|
part = dict(parts[pidx])
|
|
part["replacements"] = [r for r in part.get("replacements", []) if r["id"] != replacement_id]
|
|
parts[pidx] = part
|
|
items[idx] = {**item, "parts": parts}
|
|
_save(user_dir, items)
|
|
return JSONResponse({"ok": True})
|