arneandClaude Opus 5.5
6466f002e0
Python client for the Baby Tracker sync server, reverse engineered from the Android app (4.53), and a HACS-installable Home Assistant integration built on it. Each baby gets a log calendar and summary sensors, and actions add entries that sync back to the group. Reading has been run against the real server. Uploading and running inside real Home Assistant have not. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
372 lines
15 KiB
Python
372 lines
15 KiB
Python
"""HTTP client for the Baby Tracker sync server.
|
|
|
|
Reconstructed from com.nighp.babytracker_android.sync.SyncImple (app 4.53).
|
|
See docs/API.md for the protocol and for what is inferred rather than observed.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
import uuid
|
|
from datetime import datetime, timezone
|
|
from pathlib import Path
|
|
from typing import Any, Iterable, Iterator, Mapping, MutableMapping, Optional, Union
|
|
from urllib.parse import quote
|
|
|
|
import requests
|
|
|
|
from .models import Device, OpCode, Transaction
|
|
|
|
DEFAULT_BASE_URL = "https://prodapp.babytrackers.com"
|
|
PAGE_SIZE = 500
|
|
|
|
PHOTO_BOUNDARY = "__BABY_TRACKER_BOUNDARY____Photo__"
|
|
SEED_BOUNDARY = "__BABY_TRACKER_BOUNDARY____SEED__"
|
|
|
|
# The app sends zeros: App_Info's constructor sets both and nothing changes them.
|
|
_APP_INFO = {"AccountType": 0, "AppType": 0}
|
|
_TOKEN_APP_TYPE = 5
|
|
|
|
PathLike = Union[str, Path]
|
|
|
|
|
|
class BabyTrackerError(Exception):
|
|
"""The server returned an unexpected status."""
|
|
|
|
def __init__(self, status: int, message: str = ""):
|
|
super().__init__(f"HTTP {status}: {message}" if message else f"HTTP {status}")
|
|
self.status = status
|
|
self.message = message
|
|
|
|
|
|
class AuthenticationError(BabyTrackerError):
|
|
"""Bad credentials, or a session that could not be renewed (400 or 401)."""
|
|
|
|
|
|
class AccountStateError(BabyTrackerError):
|
|
"""Login returned 307. The app turns sync off; the cause is not documented."""
|
|
|
|
|
|
class ConflictError(BabyTrackerError):
|
|
"""409, for example when creating a group that already exists."""
|
|
|
|
|
|
class NotFoundError(BabyTrackerError):
|
|
"""404: no such seed or picture."""
|
|
|
|
|
|
class MissingSeedError(BabyTrackerError):
|
|
"""424: a group was finalized before its seed was uploaded."""
|
|
|
|
|
|
class AccountChangedError(Exception):
|
|
"""A renewed session belongs to another account than the first login.
|
|
|
|
The app turns sync off in this case, as merging would mix two accounts.
|
|
"""
|
|
|
|
|
|
_ERRORS_BY_STATUS = {
|
|
400: AuthenticationError,
|
|
401: AuthenticationError,
|
|
307: AccountStateError,
|
|
404: NotFoundError,
|
|
409: ConflictError,
|
|
424: MissingSeedError,
|
|
}
|
|
|
|
|
|
class BabyTrackerClient:
|
|
"""A sync client, which the server sees as one more device on the account.
|
|
|
|
Reuse the same ``device_uuid`` between runs: it identifies this client's
|
|
transaction log. A new value is a new device on the account.
|
|
"""
|
|
|
|
def __init__(
|
|
self,
|
|
email: str,
|
|
password: str,
|
|
device_uuid: Optional[str] = None,
|
|
device_name: str = "Python client",
|
|
device_os_info: str = "Python",
|
|
base_url: str = DEFAULT_BASE_URL,
|
|
timeout: float = 60,
|
|
):
|
|
self.email = email
|
|
self.password = password
|
|
self.device_uuid = device_uuid or str(uuid.uuid4())
|
|
self.device_name = device_name
|
|
self.device_os_info = device_os_info
|
|
self.base_url = base_url.rstrip("/")
|
|
self.timeout = timeout
|
|
self.account_id: Optional[int] = None
|
|
self._session = requests.Session()
|
|
self._next_sync_id: Optional[int] = None
|
|
|
|
# ------------------------------------------------------------------
|
|
# Transport
|
|
# ------------------------------------------------------------------
|
|
|
|
def _send(self, method: str, path: str, **kwargs: Any) -> requests.Response:
|
|
# HttpURLConnection follows redirects for GET only. A 307 on POST reaches
|
|
# the app's error handling, so it must not be followed here either.
|
|
return self._session.request(
|
|
method,
|
|
self.base_url + path,
|
|
timeout=self.timeout,
|
|
allow_redirects=method == "GET",
|
|
**kwargs,
|
|
)
|
|
|
|
def _send_json(self, method: str, path: str, body: Mapping[str, Any]) -> requests.Response:
|
|
return self._send(
|
|
method,
|
|
path,
|
|
data=json.dumps(body, ensure_ascii=False).encode("utf-8"),
|
|
headers={"Content-Type": "application/json", "charset": "utf-8"},
|
|
)
|
|
|
|
@staticmethod
|
|
def _check(response: requests.Response, *expected: int) -> requests.Response:
|
|
if response.status_code in expected:
|
|
return response
|
|
error = _ERRORS_BY_STATUS.get(response.status_code, BabyTrackerError)
|
|
raise error(response.status_code, response.text.strip())
|
|
|
|
def _authed(self, send: Any, *expected: int) -> requests.Response:
|
|
"""Run a request, logging in again once if the session has expired."""
|
|
response = send()
|
|
if response.status_code == 401:
|
|
self.login()
|
|
response = send()
|
|
return self._check(response, *expected)
|
|
|
|
def _user_info(self) -> dict[str, Any]:
|
|
return {
|
|
"AppInfo": _APP_INFO,
|
|
"Device": {
|
|
"DeviceName": self.device_name,
|
|
"DeviceOSInfo": self.device_os_info,
|
|
"DeviceUUID": self.device_uuid,
|
|
},
|
|
"EmailAddress": self.email,
|
|
"Password": self.password,
|
|
}
|
|
|
|
def _remember_account(self, response: requests.Response) -> int:
|
|
account_id = response.json().get("AccountID")
|
|
if not account_id:
|
|
raise BabyTrackerError(response.status_code, "response has no AccountID")
|
|
if self.account_id is not None and account_id != self.account_id:
|
|
raise AccountChangedError(f"logged in to account {account_id}, expected {self.account_id}")
|
|
self.account_id = account_id
|
|
return account_id
|
|
|
|
# ------------------------------------------------------------------
|
|
# Session
|
|
# ------------------------------------------------------------------
|
|
|
|
def login(self) -> int:
|
|
"""Start a session and return the account ID.
|
|
|
|
The request carries this device's details, so expect it to be added to
|
|
the account's devices.
|
|
"""
|
|
response = self._check(self._send_json("POST", "/session", self._user_info()), 200)
|
|
return self._remember_account(response)
|
|
|
|
def logout(self) -> None:
|
|
self._authed(lambda: self._send("DELETE", "/session"), 200)
|
|
|
|
def forgot_password(self, language: str = "en") -> None:
|
|
"""Ask the server to send a password reset email."""
|
|
body = {"EmailAddress": self.email, "Language": language, "AppInfo": _APP_INFO}
|
|
self._check(self._send_json("POST", "/account/forgot_password", body), 200)
|
|
|
|
def register_push_token(self, token: str) -> None:
|
|
"""Register a Firebase token for "new data" push notifications."""
|
|
body = {"TokenNative": token, "TokenAppType": _TOKEN_APP_TYPE}
|
|
self._authed(lambda: self._send_json("POST", "/device_token", body), 200, 201)
|
|
|
|
# ------------------------------------------------------------------
|
|
# Reading the account
|
|
# ------------------------------------------------------------------
|
|
|
|
def get_devices(self) -> list[Device]:
|
|
response = self._authed(lambda: self._send("GET", "/account/device"), 200)
|
|
return [Device.from_json(item) for item in response.json()]
|
|
|
|
def get_transactions(self, device_uuid: str, after_sync_id: int = 0) -> list[Transaction]:
|
|
"""Fetch one page of a device's log, starting after ``after_sync_id``."""
|
|
path = f"/account/transaction/{quote(device_uuid, safe='')}/{int(after_sync_id)}"
|
|
response = self._authed(lambda: self._send("GET", path), 200)
|
|
return [Transaction.from_json(item) for item in response.json() or []]
|
|
|
|
def iter_transactions(self, device_uuid: str, after_sync_id: int = 0) -> Iterator[Transaction]:
|
|
"""Iterate over a device's log, fetching pages until a short one."""
|
|
cursor = after_sync_id
|
|
while True:
|
|
page = self.get_transactions(device_uuid, cursor)
|
|
for item in page:
|
|
if item.sync_id > cursor:
|
|
yield item
|
|
newest = max((item.sync_id for item in page), default=cursor)
|
|
# Also stop if a full page did not advance, rather than loop forever.
|
|
if len(page) < PAGE_SIZE or newest <= cursor:
|
|
return
|
|
cursor = newest
|
|
|
|
def pull(
|
|
self, cursors: MutableMapping[str, int], include_own: bool = False
|
|
) -> Iterator[tuple[str, Transaction]]:
|
|
"""Yield ``(device_uuid, transaction)`` for everything not yet seen.
|
|
|
|
``cursors`` maps device UUIDs to the last SyncID handled and is updated
|
|
after each item is consumed, so it can be saved and passed to a later
|
|
call. This device's own log is skipped, as in the app, unless
|
|
``include_own`` is set. A client rebuilding its state from the seed needs
|
|
it, because the seed does not contain what this device uploaded later.
|
|
"""
|
|
for device in self.get_devices():
|
|
if device.uuid == self.device_uuid and not include_own:
|
|
continue
|
|
cursor = cursors.get(device.uuid, 0)
|
|
if cursor >= device.last_sync_id:
|
|
continue
|
|
for item in self.iter_transactions(device.uuid, cursor):
|
|
yield device.uuid, item
|
|
cursors[device.uuid] = item.sync_id
|
|
|
|
def download_seed(self, dest_path: PathLike) -> Path:
|
|
"""Download the account's seed zip. See :mod:`babytracker_api.seed`."""
|
|
return self._download("/account/seed", dest_path)
|
|
|
|
def download_picture(self, file_name: str, dest_path: PathLike) -> Path:
|
|
"""Download a picture.
|
|
|
|
``file_name`` is a server file name. For a ``Picture`` record with
|
|
``fileName`` X these are ``X.jpg`` (``X.mp4`` for a video) and
|
|
``X__thumbnail.jpg``.
|
|
"""
|
|
return self._download(f"/account/picture/{quote(file_name, safe='')}", dest_path)
|
|
|
|
def _download(self, path: str, dest_path: PathLike) -> Path:
|
|
dest = Path(dest_path)
|
|
response = self._authed(lambda: self._send("GET", path, stream=True), 200)
|
|
with dest.open("wb") as target:
|
|
for chunk in response.iter_content(chunk_size=65536):
|
|
target.write(chunk)
|
|
return dest
|
|
|
|
# ------------------------------------------------------------------
|
|
# Writing to the account
|
|
# ------------------------------------------------------------------
|
|
|
|
def upload_transaction(self, transaction: Transaction) -> None:
|
|
self._authed(lambda: self._send_json("POST", "/account/transaction", transaction.to_json()), 201)
|
|
|
|
def next_sync_id(self) -> int:
|
|
"""Return the next SyncID for this device's log.
|
|
|
|
The app counts these locally. This client resumes from what the server
|
|
reports, then counts locally, so only one process per ``device_uuid``
|
|
may upload at a time.
|
|
"""
|
|
if self._next_sync_id is None:
|
|
last = next((d.last_sync_id for d in self.get_devices() if d.uuid == self.device_uuid), 0)
|
|
self._next_sync_id = last + 1
|
|
return self._next_sync_id
|
|
|
|
def _push(self, op: OpCode, obj: Mapping[str, Any]) -> Transaction:
|
|
transaction = Transaction.create(self.next_sync_id(), op, obj)
|
|
self.upload_transaction(transaction)
|
|
self._next_sync_id = transaction.sync_id + 1
|
|
return transaction
|
|
|
|
def add(self, obj: Mapping[str, Any]) -> Transaction:
|
|
"""Upload a new record, as built by the helpers in ``models``."""
|
|
return self._push(OpCode.INSERT, obj)
|
|
|
|
def update(self, obj: Mapping[str, Any]) -> Transaction:
|
|
"""Upload a changed record. Other devices replace their copy with it."""
|
|
return self._push(OpCode.UPDATE, {**obj, "timestamp": datetime.now(timezone.utc)})
|
|
|
|
def delete(self, obj: Mapping[str, Any]) -> Transaction:
|
|
"""Upload a deletion. The whole record is sent, not just its ID."""
|
|
return self._push(OpCode.DELETE, {**obj, "timestamp": datetime.now(timezone.utc)})
|
|
|
|
def upload_picture(self, path: PathLike, staging: bool = False) -> None:
|
|
"""Upload a picture. ``staging`` is for pictures of an unfinished group."""
|
|
url = "/staging_account/picture" if staging else "/account/picture"
|
|
self._upload_file(url, Path(path), "picture", "image/jpeg", PHOTO_BOUNDARY)
|
|
|
|
def upload_clone_data(self, path: PathLike, file_name: Optional[str] = None) -> None:
|
|
"""Upload a backup zip. The app sends one when a device joins a group.
|
|
|
|
The app names the file after the device ID and carries on joining if
|
|
this upload fails. The app has no call that downloads this data again.
|
|
"""
|
|
source = Path(path)
|
|
self._upload_file(
|
|
"/account/clonedata",
|
|
source,
|
|
"clonedata",
|
|
"application/zip",
|
|
PHOTO_BOUNDARY,
|
|
file_name=file_name or source.name,
|
|
)
|
|
|
|
def _upload_file(
|
|
self,
|
|
path: str,
|
|
source: Path,
|
|
field: str,
|
|
content_type: str,
|
|
boundary: str,
|
|
file_name: Optional[str] = None,
|
|
) -> None:
|
|
head = (
|
|
f"\r\n--{boundary}\r\n"
|
|
f'Content-Disposition: form-data; name="{field}"; filename="{file_name or source.name}"\r\n'
|
|
f"Content-Type: {content_type}\r\n\r\n"
|
|
).encode("utf-8")
|
|
tail = f"\r\n--{boundary}--\r\n".encode("utf-8")
|
|
body = head + source.read_bytes() + tail
|
|
headers = {"Content-Type": f"multipart/form-data; boundary={boundary}"}
|
|
self._authed(lambda: self._send("POST", path, data=body, headers=headers), 201)
|
|
|
|
# ------------------------------------------------------------------
|
|
# Group management
|
|
# ------------------------------------------------------------------
|
|
|
|
def create_group(self, seed_zip: PathLike, pictures: Iterable[PathLike] = ()) -> int:
|
|
"""Create an account from a seed zip holding the app's database.
|
|
|
|
Returns the new account ID. Raises :class:`ConflictError` if the email
|
|
is already in use.
|
|
"""
|
|
self._check(self._send_json("POST", "/staging_account", self._user_info()), 200)
|
|
return self._finish_group(seed_zip, pictures)
|
|
|
|
def reset_group(self, seed_zip: PathLike, pictures: Iterable[PathLike] = (), *, confirm: bool = False) -> int:
|
|
"""Replace the whole account with a seed zip.
|
|
|
|
Judging by the endpoint and the app's flow, this discards the account's
|
|
server-side data and pushes its other devices out of sync, so
|
|
``confirm=True`` is required. It has not been tried against the server.
|
|
"""
|
|
if not confirm:
|
|
raise ValueError("reset_group replaces all account data; pass confirm=True")
|
|
self._check(self._send_json("POST", "/account_reset", self._user_info()), 200)
|
|
return self._finish_group(seed_zip, pictures)
|
|
|
|
def _finish_group(self, seed_zip: PathLike, pictures: Iterable[PathLike]) -> int:
|
|
self._upload_file("/staging_account/seed", Path(seed_zip), "seed", "application/zip", SEED_BOUNDARY, "seed.zip")
|
|
for picture in pictures:
|
|
self.upload_picture(picture, staging=True)
|
|
response = self._check(self._send("PUT", "/staging_account"), 200)
|
|
self.account_id = None
|
|
self._next_sync_id = None
|
|
return self._remember_account(response)
|