"""Client for the EufyLife smart scale cloud API (api.eufylife.com). This is an unofficial, reverse-engineered client. Eufy/Anker do not publish a developer API for their smart scales. The request shapes and field names used here were derived by reading the source of ``m4ary/eufylife-api-hacs`` (MIT-licensed, https://github.com/m4ary/eufylife-api-hacs), a Home Assistant integration that talks to the same undocumented API. This library re-implements just the HTTP calls, with no Home Assistant dependency. Because the API is undocumented, it can change without notice. This client tries to fail loudly rather than silently return partial or stale-looking data — see ``Measurement.missing_fields`` and the raw-response access below. """ from __future__ import annotations import json import time import urllib.error import urllib.request from pathlib import Path from typing import Any from .exceptions import EufyLifeAPIError, EufyLifeAuthError from .models import AuthSession, Measurement API_BASE_URL = "https://api.eufylife.com" # Fixed app-identification values used by the official EufyLife iOS app. # Not secrets — every client of this API (including the app itself) sends # the same values; they identify the calling application, not the user. _CLIENT_ID = "eufy-app" _CLIENT_SECRET = "8FHf22gaTKu7MZXqz5zytw" _USER_AGENT_VERSION = "3.3.7" _TOKEN_REFRESH_MARGIN_SECONDS = 300 class EufyLifeClient: """Minimal client: log in, fetch scale readings. Example: >>> client = EufyLifeClient("me@example.com", "hunter2") >>> for m in client.get_measurements(): ... print(m.timestamp, m.weight_kg, m.muscle_mass_kg) """ def __init__( self, email: str, password: str, *, token_cache_path: str | Path | None = None, ) -> None: self._email = email self._password = password self._token_cache_path = Path(token_cache_path) if token_cache_path else None self._session: AuthSession | None = None # -- authentication ----------------------------------------------------- def login(self) -> AuthSession: """Authenticate and return a new session. Also updates the token cache, if configured.""" body = { "client_id": _CLIENT_ID, "client_secret": _CLIENT_SECRET, "email": self._email, "password": self._password, } data = self._post(f"{API_BASE_URL}/v1/user/v2/email/login", body) if data.get("res_code") != 1: raise EufyLifeAuthError( f"login failed: res_code={data.get('res_code')} message={data.get('message')!r}" ) devices = data.get("devices") or [] customers = data.get("customers") or [] expires_in = data.get("expires_in", 2592000) # API default: 30 days session = AuthSession( access_token=data["access_token"], user_id=data["user_id"], expires_at=time.time() + expires_in, device_id=devices[0].get("id") if devices else None, customer_ids=[c["id"] for c in customers if c.get("id")], ) self._session = session self._save_token_cache(session) return session def _ensure_session(self) -> AuthSession: if self._session is None: self._session = self._load_token_cache() if self._session is None or self._session.is_expired: return self.login() if self._session.expires_at - time.time() <= _TOKEN_REFRESH_MARGIN_SECONDS: return self.login() return self._session # -- data ----------------------------------------------------------- def get_raw_device_data(self, *, after: int | None = None) -> list[dict[str, Any]]: """Fetch raw, unparsed records from ``/v1/device/data``. Use this if you want the exact API response (e.g. to log it for replay tests, or to read a field this library doesn't model yet). ``after`` is a Unix timestamp; when given, only records strictly newer than it are returned. """ session = self._ensure_session() headers = { "Host": "api.eufylife.com", "Accept": "*/*", "Uid": session.user_id, "Accept-Encoding": "gzip, deflate, br", "User-Agent": f"Eufylife-iOS-{_USER_AGENT_VERSION}-281", "Accept-Language": "en-US,en;q=0.9", "Token": session.access_token, } url = f"{API_BASE_URL}/v1/device/data" if after is not None: url += f"?after={after}" raw = self._get(url, headers) if isinstance(raw, list): return raw if isinstance(raw, dict): if raw.get("res_code") not in (1, None): raise EufyLifeAPIError(f"device/data returned an error: {raw}") return raw.get("data", []) raise EufyLifeAPIError(f"unexpected response shape: {type(raw).__name__}") def get_measurements(self, *, after: int | None = None) -> list[Measurement]: """Fetch and parse scale readings, newest last. Records that are missing ``customer_id`` or ``scale_data`` entirely are silently dropped (they carry no usable measurement). Records that parse but are missing individual metrics are kept — check ``Measurement.missing_fields`` rather than assuming every field is populated. """ records = self.get_raw_device_data(after=after) parsed = (Measurement.from_device_record(r) for r in records) measurements = [m for m in parsed if m is not None] measurements.sort(key=lambda m: m.timestamp) return measurements # -- token cache (optional) ----------------------------------------- def _save_token_cache(self, session: AuthSession) -> None: if not self._token_cache_path: return self._token_cache_path.parent.mkdir(parents=True, exist_ok=True) payload = { "access_token": session.access_token, "user_id": session.user_id, "expires_at": session.expires_at, "device_id": session.device_id, "customer_ids": session.customer_ids, } self._token_cache_path.write_text( json.dumps(payload, indent=2), encoding="utf-8" ) def _load_token_cache(self) -> AuthSession | None: if not self._token_cache_path or not self._token_cache_path.exists(): return None try: payload = json.loads(self._token_cache_path.read_text(encoding="utf-8")) return AuthSession(**payload) except (json.JSONDecodeError, TypeError, KeyError): return None # -- HTTP ------------------------------------------------------------- @staticmethod def _headers() -> dict[str, str]: return { "Accept": "*/*", "Accept-Language": "en-US,en;q=0.9", "Accept-Encoding": "gzip, deflate, br", "User-Agent": f"EufyLife-iOS-{_USER_AGENT_VERSION}", "Category": "Health", "Language": "en", "Timezone": "UTC", "Country": "US", "Content-Type": "application/json", } def _post(self, url: str, body: dict[str, Any]) -> dict[str, Any]: req = urllib.request.Request( url, data=json.dumps(body).encode("utf-8"), headers=self._headers(), method="POST", ) return self._send(req) def _get(self, url: str, headers: dict[str, str]) -> Any: req = urllib.request.Request(url, headers=headers, method="GET") return self._send(req) @staticmethod def _send(req: urllib.request.Request) -> Any: try: with urllib.request.urlopen(req, timeout=30) as resp: return json.loads(resp.read().decode("utf-8")) except urllib.error.HTTPError as err: raise EufyLifeAPIError( f"HTTP {err.code} from {req.full_url}: {err.read().decode('utf-8', 'replace')[:500]}" ) from err except urllib.error.URLError as err: raise EufyLifeAPIError(f"network error calling {req.full_url}: {err}") from err