missing file

bincio/edit/ops.py

@@ -0,0 +1,140 @@
+"""Pure write operations used by both the single-user edit server and the
+multi-user serve server.
+
+No FastAPI, no globals — all context is passed as explicit arguments.
+"""
+
+from __future__ import annotations
+
+import json
+import time
+from pathlib import Path
+from typing import Any
+
+SPORTS = ["cycling", "running", "hiking", "walking", "swimming", "skiing", "other"]
+STAT_PANELS = ["elevation", "speed", "heart_rate", "cadence", "power"]
+
+
+def apply_sidecar_edit(activity_id: str, payload: dict[str, Any], data_dir: Path) -> None:
+    """Write a sidecar .md file and trigger merge_all().
+
+    Args:
+        activity_id: Validated activity ID (caller must validate).
+        payload:     Dict with optional keys: title, sport, gear, description,
+                     highlight, private, hide_stats.
+        data_dir:    Per-user data directory (contains activities/, edits/).
+    """
+    edits_dir = data_dir / "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"] in SPORTS 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")
+
+    from bincio.render.merge import merge_all
+    merge_all(data_dir)
+
+
+def run_strava_sync(data_dir: Path, client_id: str, client_secret: str) -> dict[str, Any]:
+    """Fetch new Strava activities and write them into data_dir.
+
+    Args:
+        data_dir:      Per-user data directory.
+        client_id:     Strava OAuth client ID.
+        client_secret: Strava OAuth client secret.
+
+    Returns:
+        Dict with keys: ok, imported, skipped, error_count, errors.
+
+    Raises:
+        RuntimeError: If Strava credentials are missing or API calls fail.
+    """
+    if not client_id or not client_secret:
+        raise RuntimeError("Strava not configured (missing client_id or client_secret)")
+
+    from bincio.extract.strava_api import (
+        StravaError,
+        ensure_fresh,
+        fetch_activities,
+        fetch_streams,
+        save_token,
+        strava_meta_to_partial,
+        strava_to_parsed,
+    )
+
+    try:
+        token = ensure_fresh(data_dir, client_id, client_secret)
+    except StravaError as e:
+        raise RuntimeError(str(e)) from e
+
+    after: int | None = token.get("last_sync_at")
+    try:
+        activities = fetch_activities(token["access_token"], after=after)
+    except StravaError as e:
+        raise RuntimeError(str(e)) from e
+
+    from bincio.extract.metrics import compute
+    from bincio.extract.writer import build_summary, make_activity_id, write_activity, write_index
+    from bincio.render.merge import merge_all
+
+    index_path = data_dir / "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", {})
+    summaries: dict[str, dict] = {s["id"]: s for s in index_data.get("activities", [])}
+
+    imported = 0
+    skipped = 0
+    errors: list[str] = []
+
+    for meta in activities:
+        try:
+            activity_id = make_activity_id(strava_meta_to_partial(meta))
+            if (data_dir / "activities" / f"{activity_id}.json").exists():
+                skipped += 1
+                continue
+            streams = fetch_streams(token["access_token"], meta["id"])
+            parsed = strava_to_parsed(meta, streams)
+            metrics = compute(parsed)
+            write_activity(parsed, metrics, data_dir, privacy="public", rdp_epsilon=0.0001)
+            summaries[activity_id] = build_summary(parsed, metrics, activity_id, "public")
+            imported += 1
+        except Exception as exc:
+            errors.append(f"{meta.get('id')}: {type(exc).__name__}")
+
+    if imported:
+        write_index(list(summaries.values()), data_dir, owner)
+        merge_all(data_dir)
+
+    token["last_sync_at"] = int(time.time())
+    save_token(data_dir, token)
+
+    return {
+        "ok": True,
+        "imported": imported,
+        "skipped": skipped,
+        "error_count": len(errors),
+        "errors": errors[:5],
+    }