private_health/bridge/server.py

#!/usr/bin/env python3
"""Minimal bridge: Life Dashboard Companion webhook -> Nextcloud Health v3.

Scope: only forwards the Health Connect types that are both sent by the
Companion app and already supported by Health v3: weight and steps (Daily
Values, one upsert per calendar day), and pulse/heart rate (a Measurement).
Which metrics a given Health instance actually knows is detected at startup
rather than assumed, since shipped releases can lag behind the documented
API. Sleep duration isn't a Daily Value, Measurement, or journal metric in
Health v3 yet at all, despite being documented, so it can't be mapped here
either. Body fat, blood glucose, blood pressure, oxygen saturation and
temperature aren't sent by the Companion app at all yet. Hydration and
exercise exist in both apps but with incompatible shapes (a volume/duration
vs. a closed set of event options) and would need lossy guessing, so they
are skipped for now too.

Run:
    NC_WEBHOOK_SECRET=... python server.py
"""
import json
import logging
import os
import uuid
from datetime import datetime
from hashlib import sha256
from hmac import compare_digest, new as hmac_new
from pathlib import Path

import requests
from flask import Flask, request

logging.basicConfig(level=logging.INFO, format="%(asctime)s %(levelname)s %(message)s")
log = logging.getLogger("bridge")

CREDENTIALS_FILE = Path(__file__).parent / "credentials.json"
try:
    WEBHOOK_SECRET: str = os.environ["NC_WEBHOOK_SECRET"]
except KeyError:
    raise SystemExit("NC_WEBHOOK_SECRET must be set (the Companion app's HMAC signing secret)")

credentials = json.loads(CREDENTIALS_FILE.read_text())
NC_SERVER = credentials["server"]
NC_AUTH = (credentials["login_name"], credentials["app_password"])
HEALTH_API = f"{NC_SERVER}/ocs/v2.php/apps/health/api/v2"
DAILY_VALUES_URL = f"{HEALTH_API}/daily-values"
MEASUREMENTS_URL = f"{HEALTH_API}/measurements"
CONFIGURATION_URL = f"{HEALTH_API}/configuration"
# Identifies this bridge in Nextcloud's server logs and admin UI instead of the
# default "python-requests/x.y.z".
USER_AGENT = "PrivateHealthBridge"
# Some instances (cold PHP-FPM, slow DB) need more than a few seconds to answer.
NC_TIMEOUT_SECONDS = float(os.environ.get("NC_TIMEOUT_SECONDS", "30"))
# Every metric this bridge can forward; new users only have "stress" enabled
# by default, so writes would otherwise fail with an "unsupported metric" error.
WANTED_METRICS = ("weight", "steps", "pulse")
# Filled at startup from the server's actual configuration response, since a
# Health instance may not know a metric yet at all or may just have it
# disabled for this user.
supported_metrics: set[str] = set()


def discover_supported_metrics() -> None:
    resp = requests.get(
        CONFIGURATION_URL,
        auth=NC_AUTH,
        headers={"OCS-APIRequest": "true", "Accept": "application/json", "User-Agent": USER_AGENT},
        params={"format": "json"},
        timeout=NC_TIMEOUT_SECONDS,
    )
    resp.raise_for_status()
    current = resp.json()["ocs"]["data"]["metrics"]

    unknown = [m for m in WANTED_METRICS if m not in current]
    if unknown:
        log.warning("This Health instance doesn't know these metrics yet, skipping them: %s", unknown)

    disabled = [m for m in WANTED_METRICS if m in current and not current[m]["enabled"]]
    if disabled:
        log.info("Enabling Health metrics not yet active for this user: %s", disabled)
        resp = requests.put(
            CONFIGURATION_URL,
            json={"metrics": {m: {"enabled": True} for m in disabled}},
            auth=NC_AUTH,
            headers={
                "OCS-APIRequest": "true",
                "Accept": "application/json",
                "User-Agent": USER_AGENT,
            },
            params={"format": "json"},
            timeout=NC_TIMEOUT_SECONDS,
        )
        if not resp.ok:
            log.error("Nextcloud rejected configuration update: %s -> %s", resp.status_code, resp.text)
        resp.raise_for_status()

    supported_metrics.update(m for m in WANTED_METRICS if m not in unknown)

app = Flask(__name__)


def verify_signature(raw_body: bytes, header_value: str | None) -> bool:
    if not header_value or not header_value.startswith("sha256="):
        return False
    expected = "sha256=" + hmac_new(WEBHOOK_SECRET.encode(), raw_body, sha256).hexdigest()
    return compare_digest(expected, header_value)


def as_operation_id(record_uuid: str | None) -> str | None:
    if not record_uuid:
        return None
    try:
        uuid.UUID(record_uuid)
        return record_uuid
    except ValueError:
        return None


def deterministic_operation_id(*parts: str) -> str:
    """Stable id for records that carry no HC uuid (e.g. bucketed windows)."""
    return str(uuid.uuid5(uuid.NAMESPACE_URL, "|".join(parts)))


def parse_rfc3339(value: str) -> datetime:
    return datetime.fromisoformat(value.replace("Z", "+00:00"))


def local_date(iso_timestamp: str) -> str:
    # Approximation: uses the UTC calendar date since we don't know the
    # Nextcloud user's timezone here; fine for anything but readings taken
    # right around midnight.
    return parse_rfc3339(iso_timestamp).date().isoformat()


# Canonical unit Health expects per Daily Value metric (src/metrics.ts's
# getMetricUnits(); "null" is rejected as "Unsupported unit or numeric value").
DAILY_VALUE_UNITS = {"weight": "kg", "steps": "steps"}


def build_daily_value(metric_key: str, date: str, numeric_value: float) -> dict:
    return {"kind": "daily_value", "metricKey": metric_key, "date": date, "numericValue": numeric_value}


# Canonical unit Health expects per Measurement metric (src/metrics.ts's
# getMetricUnits()).
MEASUREMENT_UNITS = {"pulse": "bpm"}


def build_measurement(
    metric_key: str, numeric_value: float, recorded_at: str, operation_id: str | None
) -> dict:
    payload = {
        "kind": "measurement",
        "metricKey": metric_key,
        "numericValue": numeric_value,
        "values": None,
        "unit": MEASUREMENT_UNITS[metric_key],
        # "api" isn't a valid Measurement context (only manual/checkin/checkout);
        # unlike Entries and Daily Values there's no dedicated automation value.
        "context": "manual",
        "source": "api",
        "recordedAt": recorded_at,
        "note": None,
    }
    if operation_id:
        payload["operationId"] = operation_id
    return payload


def extract_entries(body: dict) -> list[dict]:
    entries = []

    if "weight" in supported_metrics:
        for record in body.get("weight", []):
            entries.append(build_daily_value("weight", local_date(record["time"]), record["kilograms"]))

    if "steps" in supported_metrics:
        # Health Connect's own aggregate, deduplicated across sources (phone,
        # watch, ...) -- summing the raw per-window records ourselves would
        # double-count. See docs/webhook.md#daily-totals.
        for totals in body.get("daily_totals", []):
            if "steps" in totals:
                entries.append(build_daily_value("steps", totals["date"], totals["steps"]))

    if "pulse" in supported_metrics:
        for record in body.get("heart_rate", []):
            if "bucket_start" in record:
                # Bucketed resolution: no single sample, so use the window's
                # average as the representative pulse value.
                entries.append(
                    build_measurement(
                        "pulse",
                        record["avg"],
                        record["bucket_start"],
                        deterministic_operation_id("pulse", record["bucket_start"]),
                    )
                )
            else:
                entries.append(
                    build_measurement(
                        "pulse",
                        record["bpm"],
                        record["time"],
                        as_operation_id(record.get("uuid")),
                    )
                )

    return entries


def forward_entry(entry: dict) -> None:
    if entry["kind"] == "daily_value":
        resp = requests.put(
            f"{DAILY_VALUES_URL}/{entry['metricKey']}/{entry['date']}",
            json={"numericValue": entry["numericValue"], "unit": DAILY_VALUE_UNITS[entry["metricKey"]]},
            auth=NC_AUTH,
            headers={
                "OCS-APIRequest": "true",
                "Accept": "application/json",
                "User-Agent": USER_AGENT,
            },
            params={"format": "json"},
            timeout=NC_TIMEOUT_SECONDS,
        )
    else:
        payload = {k: v for k, v in entry.items() if k != "kind"}
        resp = requests.post(
            MEASUREMENTS_URL,
            json=payload,
            auth=NC_AUTH,
            headers={
                "OCS-APIRequest": "true",
                "Accept": "application/json",
                "User-Agent": USER_AGENT,
            },
            params={"format": "json"},
            timeout=NC_TIMEOUT_SECONDS,
        )
    if not resp.ok:
        # Nextcloud's error body has the actual validation reason; the plain
        # status code alone isn't enough to debug a 400.
        log.error("Nextcloud rejected %s: %s %s -> %s", entry["kind"], resp.status_code, entry, resp.text)
    resp.raise_for_status()


@app.post("/webhook/health-connect")
def health_connect_webhook():
    raw_body = request.get_data()
    if not verify_signature(raw_body, request.headers.get("X-Signature")):
        log.warning("Rejected webhook: invalid or missing signature")
        return "invalid signature", 401

    body = request.get_json(force=True)
    entries = extract_entries(body)
    # Map Companion's payload field names to Health's metric keys so this log
    # line uses the same vocabulary as "supported metrics" below.
    present_types = [
        metric_key
        for payload_key, metric_key in (("weight", "weight"), ("daily_totals", "steps"), ("heart_rate", "pulse"))
        if body.get(payload_key)
    ]
    log.info(
        "Payload contains %s; forwarding %d entr(y/ies) (supported metrics: %s)",
        present_types,
        len(entries),
        sorted(supported_metrics),
    )

    failures = 0
    ok_counts: dict[str, int] = {}
    failure_counts: dict[str, int] = {}
    for entry in entries:
        metric_key = entry["metricKey"]
        try:
            forward_entry(entry)
        except requests.RequestException:
            log.exception("Failed to forward %s entry", metric_key)
            failures += 1
            failure_counts[metric_key] = failure_counts.get(metric_key, 0) + 1
        else:
            ok_counts[metric_key] = ok_counts.get(metric_key, 0) + 1

    # The per-entry log lines above are easy to miss in a busy log; this
    # summary makes it obvious which metrics actually made it through.
    log.info("Forwarded ok: %s, failed: %s", ok_counts, failure_counts)

    if failures:
        # Non-2xx makes the Companion app retry the whole payload later.
        return f"{failures} record(s) failed to forward", 502

    return "ok", 200


if __name__ == "__main__":
    discover_supported_metrics()
    app.run(host="0.0.0.0", port=8080)