116 lines
3.4 KiB
Python
116 lines
3.4 KiB
Python
#!/usr/bin/env python3
|
|
"""Minimal bridge: Life Dashboard Companion webhook -> Nextcloud Health v3.
|
|
|
|
MVP scope: only the `weight` array of the Health Connect payload is forwarded,
|
|
since that is the one OpenScale-relevant metric already built into Health v3.
|
|
Everything else in the payload is ignored for now.
|
|
|
|
Run:
|
|
NC_WEBHOOK_SECRET=... python server.py
|
|
"""
|
|
import json
|
|
import logging
|
|
import os
|
|
import uuid
|
|
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"])
|
|
ENTRIES_URL = f"{NC_SERVER}/ocs/v2.php/apps/health/api/v2/entries"
|
|
# Identifies this bridge in Nextcloud's server logs and admin UI instead of the
|
|
# default "python-requests/x.y.z".
|
|
USER_AGENT = "PrivateHealthBridge"
|
|
|
|
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 forward_weight_entry(record: dict) -> None:
|
|
payload = {
|
|
"metricKey": "weight",
|
|
"numericValue": record["kilograms"],
|
|
"optionValue": None,
|
|
"context": "api",
|
|
"source": "api",
|
|
"recordedAt": record["time"],
|
|
"note": None,
|
|
}
|
|
operation_id = as_operation_id(record.get("uuid"))
|
|
if operation_id:
|
|
payload["operationId"] = operation_id
|
|
|
|
resp = requests.post(
|
|
ENTRIES_URL,
|
|
json=payload,
|
|
auth=NC_AUTH,
|
|
headers={
|
|
"OCS-APIRequest": "true",
|
|
"Accept": "application/json",
|
|
"User-Agent": USER_AGENT,
|
|
},
|
|
params={"format": "json"},
|
|
timeout=10,
|
|
)
|
|
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)
|
|
weight_records = body.get("weight", [])
|
|
log.info("Received %d weight record(s)", len(weight_records))
|
|
|
|
failures = 0
|
|
for record in weight_records:
|
|
try:
|
|
forward_weight_entry(record)
|
|
except requests.RequestException:
|
|
log.exception("Failed to forward weight record %s", record.get("uuid"))
|
|
failures += 1
|
|
|
|
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__":
|
|
app.run(host="0.0.0.0", port=8080)
|