473 lines
16 KiB
Python
473 lines
16 KiB
Python
"""
|
|
Zepp/Huami API Client
|
|
|
|
Provides authentication and data fetching from the Zepp/Huami cloud API
|
|
for Amazfit and Mi Band devices.
|
|
|
|
Based on the Huami API technical documentation:
|
|
- Authentication: account.huami.com
|
|
- Data: api-mifit-de2.zepp.com (EU GDPR compliant)
|
|
"""
|
|
|
|
import base64
|
|
import datetime
|
|
import hashlib
|
|
import json
|
|
import logging
|
|
import uuid
|
|
from typing import Any, Dict, List, Optional, Tuple
|
|
|
|
import requests
|
|
|
|
from backend.core.config import settings
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
# --- Constants ---
|
|
AUTH_URL = "https://account.huami.com/v2/client/login"
|
|
API_BASE_EU = "https://api-mifit-de2.zepp.com"
|
|
API_BASE_GLOBAL = "https://api-mifit.huami.com"
|
|
|
|
# App identification (mimics official Zepp app)
|
|
APP_NAME = "com.huami.watch.hmwatchmanager"
|
|
CLIENT_ID = "HuaMi"
|
|
APP_VERSION = "6.9.3"
|
|
|
|
|
|
class ZeppApiError(Exception):
|
|
"""Custom exception for Zepp API errors."""
|
|
pass
|
|
|
|
|
|
class ZeppApiClient:
|
|
"""
|
|
Client for interacting with the Zepp/Huami API.
|
|
|
|
Handles authentication, token management, and data fetching.
|
|
"""
|
|
|
|
def __init__(self, email: str = None, password: str = None):
|
|
self.email = email
|
|
self.password = password
|
|
self.app_token: Optional[str] = None
|
|
self.login_token: Optional[str] = None
|
|
self.user_id: Optional[str] = None
|
|
self.token_expiry: Optional[datetime.datetime] = None
|
|
self.api_base = API_BASE_EU # Default to EU for GDPR
|
|
self._session = requests.Session()
|
|
self._session.headers.update({
|
|
"User-Agent": f"Zepp/{APP_VERSION} (Android)",
|
|
"Accept": "application/json",
|
|
})
|
|
|
|
def authenticate(self, email: str = None, password: str = None) -> Dict[str, Any]:
|
|
"""
|
|
Authenticate with the Huami/Zepp API.
|
|
|
|
Args:
|
|
email: Zepp account email
|
|
password: Zepp account password
|
|
|
|
Returns:
|
|
Dict containing user_id, app_token, and other session info
|
|
"""
|
|
email = email or self.email
|
|
password = password or self.password
|
|
|
|
if not email or not password:
|
|
raise ZeppApiError("Email and password are required")
|
|
|
|
# Generate unique device ID
|
|
device_id = uuid.uuid4().hex[:16].upper()
|
|
|
|
# Prepare authentication request
|
|
auth_data = {
|
|
"grant_type": "access_token",
|
|
"client_id": CLIENT_ID,
|
|
"app_name": APP_NAME,
|
|
"app_version": APP_VERSION,
|
|
"third_name": "huami",
|
|
"device_id": device_id,
|
|
"device_model": "android_phone",
|
|
"allow_registration": "false",
|
|
"dn": "api-user.huami.com,api-mifit-de2.zepp.com,app-analytics.huami.com",
|
|
"country_code": "DE",
|
|
"lang": "en",
|
|
"state": "REDIRECTION",
|
|
"redirect_uri": "https://s3-us-west-2.amazonws.com/hm-registration/successs498.html",
|
|
}
|
|
|
|
# Add credentials
|
|
auth_data["email"] = email
|
|
auth_data["password"] = password
|
|
|
|
headers = {
|
|
"Content-Type": "application/x-www-form-urlencoded",
|
|
}
|
|
|
|
try:
|
|
response = self._session.post(
|
|
AUTH_URL,
|
|
data=auth_data,
|
|
headers=headers,
|
|
timeout=30
|
|
)
|
|
response.raise_for_status()
|
|
|
|
result = response.json()
|
|
|
|
if "error_code" in result:
|
|
error_msg = result.get("message", "Authentication failed")
|
|
raise ZeppApiError(f"Auth error: {error_msg}")
|
|
|
|
# Extract tokens
|
|
token_info = result.get("token_info", {})
|
|
self.app_token = token_info.get("app_token")
|
|
self.login_token = token_info.get("login_token")
|
|
self.user_id = token_info.get("user_id")
|
|
|
|
if not self.app_token or not self.user_id:
|
|
raise ZeppApiError("Invalid authentication response: missing tokens")
|
|
|
|
# Store credentials for refresh
|
|
self.email = email
|
|
self.password = password
|
|
|
|
# Set token expiry (typically 30 days, be conservative with 7)
|
|
self.token_expiry = datetime.datetime.now() + datetime.timedelta(days=7)
|
|
|
|
logger.info(f"Successfully authenticated Zepp user: {self.user_id}")
|
|
|
|
return {
|
|
"user_id": self.user_id,
|
|
"app_token": self.app_token,
|
|
"login_token": self.login_token,
|
|
}
|
|
|
|
except requests.exceptions.RequestException as e:
|
|
logger.error(f"Zepp authentication request failed: {e}")
|
|
raise ZeppApiError(f"Network error during authentication: {e}")
|
|
|
|
def _ensure_authenticated(self):
|
|
"""Ensure we have valid tokens, re-authenticate if needed."""
|
|
if not self.app_token or not self.user_id:
|
|
if self.email and self.password:
|
|
self.authenticate()
|
|
else:
|
|
raise ZeppApiError("Not authenticated. Call authenticate() first.")
|
|
|
|
# Check token expiry
|
|
if self.token_expiry and datetime.datetime.now() > self.token_expiry:
|
|
logger.info("Token expired, re-authenticating...")
|
|
self.authenticate()
|
|
|
|
def _make_request(self, endpoint: str, params: Dict = None, method: str = "GET") -> Dict:
|
|
"""
|
|
Make an authenticated request to the Zepp API.
|
|
"""
|
|
self._ensure_authenticated()
|
|
|
|
url = f"{self.api_base}{endpoint}"
|
|
|
|
headers = {
|
|
"apptoken": self.app_token,
|
|
"Accept": "application/json",
|
|
}
|
|
|
|
try:
|
|
if method == "GET":
|
|
response = self._session.get(url, params=params, headers=headers, timeout=30)
|
|
else:
|
|
response = self._session.post(url, data=params, headers=headers, timeout=30)
|
|
|
|
response.raise_for_status()
|
|
return response.json()
|
|
|
|
except requests.exceptions.RequestException as e:
|
|
logger.error(f"Zepp API request failed: {e}")
|
|
raise ZeppApiError(f"API request failed: {e}")
|
|
|
|
def fetch_band_data(
|
|
self,
|
|
from_date: datetime.date,
|
|
to_date: datetime.date,
|
|
query_type: str = "summary"
|
|
) -> Dict[str, Any]:
|
|
"""
|
|
Fetch daily health data (steps, sleep, heart rate).
|
|
|
|
Args:
|
|
from_date: Start date
|
|
to_date: End date
|
|
query_type: "summary" for daily totals, "detail" for minute-by-minute
|
|
|
|
Returns:
|
|
Dict containing decoded health data
|
|
"""
|
|
params = {
|
|
"query_type": query_type,
|
|
"from_date": from_date.strftime("%Y-%m-%d"),
|
|
"to_date": to_date.strftime("%Y-%m-%d"),
|
|
"userid": self.user_id,
|
|
"device_type": "android_phone",
|
|
}
|
|
|
|
response = self._make_request("/v1/data/band_data.json", params)
|
|
|
|
if response.get("code") != 1:
|
|
raise ZeppApiError(f"band_data request failed: {response.get('message')}")
|
|
|
|
# Decode the Base64 summary data
|
|
data_list = response.get("data", [])
|
|
decoded_data = []
|
|
|
|
for entry in data_list:
|
|
decoded_entry = self._decode_band_data_entry(entry)
|
|
if decoded_entry:
|
|
decoded_data.append(decoded_entry)
|
|
|
|
return {"data": decoded_data, "raw_response": response}
|
|
|
|
def _decode_band_data_entry(self, entry: Dict) -> Optional[Dict]:
|
|
"""
|
|
Decode a single band_data entry with Base64 encoded fields.
|
|
"""
|
|
try:
|
|
date_str = entry.get("date_time", entry.get("date"))
|
|
if not date_str:
|
|
return None
|
|
|
|
result = {
|
|
"date": date_str,
|
|
"summary": {},
|
|
}
|
|
|
|
# Decode summary field (Base64 JSON)
|
|
summary_b64 = entry.get("summary")
|
|
if summary_b64:
|
|
try:
|
|
summary_json = base64.b64decode(summary_b64).decode("utf-8")
|
|
result["summary"] = json.loads(summary_json)
|
|
except Exception as e:
|
|
logger.warning(f"Failed to decode summary: {e}")
|
|
|
|
# Extract steps data
|
|
stp = result["summary"].get("stp", {})
|
|
result["steps"] = stp.get("ttl", 0)
|
|
result["distance"] = stp.get("dis", 0)
|
|
result["calories"] = stp.get("cal", 0)
|
|
|
|
# Extract sleep data
|
|
slp = result["summary"].get("slp", {})
|
|
if slp:
|
|
result["sleep"] = {
|
|
"start_time": slp.get("st"),
|
|
"end_time": slp.get("ed"),
|
|
"deep_sleep_minutes": slp.get("dp", 0),
|
|
"light_sleep_minutes": slp.get("lt", 0),
|
|
"rem_sleep_minutes": slp.get("rm", 0),
|
|
"awake_minutes": slp.get("wk", 0),
|
|
}
|
|
# Calculate total
|
|
result["sleep"]["total_minutes"] = (
|
|
result["sleep"]["deep_sleep_minutes"] +
|
|
result["sleep"]["light_sleep_minutes"] +
|
|
result["sleep"]["rem_sleep_minutes"]
|
|
)
|
|
|
|
# Decode heart rate blob if present
|
|
hr_blob = entry.get("data_hr")
|
|
if hr_blob:
|
|
hr_values = self._decode_heart_rate_blob(hr_blob)
|
|
if hr_values:
|
|
valid_hr = [v for v in hr_values if 30 < v < 220]
|
|
if valid_hr:
|
|
result["heart_rate"] = {
|
|
"avg": sum(valid_hr) // len(valid_hr),
|
|
"min": min(valid_hr),
|
|
"max": max(valid_hr),
|
|
"resting": min(valid_hr), # Approximation
|
|
}
|
|
|
|
# Decode SpO2 if present
|
|
spo2_blob = entry.get("data_spo2")
|
|
if spo2_blob:
|
|
spo2_values = self._decode_spo2_blob(spo2_blob)
|
|
if spo2_values:
|
|
valid_spo2 = [v for v in spo2_values if 70 < v <= 100]
|
|
if valid_spo2:
|
|
result["spo2"] = {
|
|
"avg": sum(valid_spo2) / len(valid_spo2),
|
|
"min": min(valid_spo2),
|
|
"max": max(valid_spo2),
|
|
}
|
|
|
|
return result
|
|
|
|
except Exception as e:
|
|
logger.error(f"Error decoding band_data entry: {e}")
|
|
return None
|
|
|
|
def _decode_heart_rate_blob(self, b64_data: str) -> List[int]:
|
|
"""
|
|
Decode Base64-encoded heart rate binary data.
|
|
Format: sequence of 1-byte or 2-byte integers.
|
|
"""
|
|
try:
|
|
raw = base64.b64decode(b64_data)
|
|
# Try single-byte format first (most common)
|
|
values = list(raw)
|
|
# Filter out invalid values (0, 255 are typically "no reading")
|
|
return [v for v in values if v > 0 and v < 255]
|
|
except Exception as e:
|
|
logger.warning(f"Failed to decode HR blob: {e}")
|
|
return []
|
|
|
|
def _decode_spo2_blob(self, b64_data: str) -> List[float]:
|
|
"""
|
|
Decode Base64-encoded SpO2 binary data.
|
|
"""
|
|
try:
|
|
raw = base64.b64decode(b64_data)
|
|
# SpO2 is typically stored as single bytes (percentage)
|
|
values = list(raw)
|
|
return [float(v) for v in values if v > 0]
|
|
except Exception as e:
|
|
logger.warning(f"Failed to decode SpO2 blob: {e}")
|
|
return []
|
|
|
|
def fetch_workout_history(self, days: int = 30) -> List[Dict]:
|
|
"""
|
|
Fetch workout/sport session history.
|
|
|
|
Args:
|
|
days: Number of days of history to fetch
|
|
|
|
Returns:
|
|
List of workout sessions
|
|
"""
|
|
from_date = datetime.date.today() - datetime.timedelta(days=days)
|
|
|
|
params = {
|
|
"from_date": from_date.strftime("%Y-%m-%d"),
|
|
"to_date": datetime.date.today().strftime("%Y-%m-%d"),
|
|
"userid": self.user_id,
|
|
}
|
|
|
|
try:
|
|
response = self._make_request("/v1/sport/run/history.json", params)
|
|
|
|
if response.get("code") != 1:
|
|
logger.warning(f"Workout history request failed: {response}")
|
|
return []
|
|
|
|
return response.get("data", {}).get("summary", [])
|
|
|
|
except ZeppApiError as e:
|
|
logger.warning(f"Failed to fetch workout history: {e}")
|
|
return []
|
|
|
|
def fetch_workout_detail(self, track_id: str) -> Optional[Dict]:
|
|
"""
|
|
Fetch detailed workout data including GPS track.
|
|
|
|
Args:
|
|
track_id: The workout track ID
|
|
|
|
Returns:
|
|
Dict with workout details and GPS data
|
|
"""
|
|
params = {
|
|
"trackid": track_id,
|
|
"source": "run.mifit.huami.com",
|
|
}
|
|
|
|
try:
|
|
response = self._make_request("/v1/sport/run/detail.json", params)
|
|
|
|
if response.get("code") != 1:
|
|
return None
|
|
|
|
data = response.get("data", {})
|
|
|
|
# Parse GPS coordinates if present
|
|
gps_str = data.get("longitude_latitude", "")
|
|
if gps_str:
|
|
data["gps_track"] = self._parse_gps_string(gps_str)
|
|
|
|
return data
|
|
|
|
except ZeppApiError as e:
|
|
logger.warning(f"Failed to fetch workout detail {track_id}: {e}")
|
|
return None
|
|
|
|
def _parse_gps_string(self, gps_str: str) -> List[Tuple[float, float]]:
|
|
"""
|
|
Parse Zepp's proprietary GPS format.
|
|
Format: lat,lon;lat,lon;lat,lon...
|
|
"""
|
|
points = []
|
|
try:
|
|
for point_str in gps_str.split(";"):
|
|
if "," in point_str:
|
|
parts = point_str.split(",")
|
|
if len(parts) >= 2:
|
|
lat = float(parts[0])
|
|
lon = float(parts[1])
|
|
points.append((lat, lon))
|
|
except Exception as e:
|
|
logger.warning(f"Failed to parse GPS string: {e}")
|
|
return points
|
|
|
|
|
|
# --- Convenience Functions ---
|
|
|
|
def create_zepp_client(email: str, password: str) -> ZeppApiClient:
|
|
"""Create and authenticate a Zepp API client."""
|
|
client = ZeppApiClient(email, password)
|
|
client.authenticate()
|
|
return client
|
|
|
|
|
|
def sync_zepp_data(
|
|
client: ZeppApiClient,
|
|
days: int = 30
|
|
) -> Dict[str, Any]:
|
|
"""
|
|
Sync all available data from Zepp.
|
|
|
|
Args:
|
|
client: Authenticated ZeppApiClient
|
|
days: Number of days to sync
|
|
|
|
Returns:
|
|
Dict with synced data counts
|
|
"""
|
|
from_date = datetime.date.today() - datetime.timedelta(days=days)
|
|
to_date = datetime.date.today()
|
|
|
|
results = {
|
|
"daily_metrics": [],
|
|
"workouts": [],
|
|
"days_synced": 0,
|
|
"workouts_synced": 0,
|
|
}
|
|
|
|
# Fetch band data (daily metrics)
|
|
try:
|
|
band_data = client.fetch_band_data(from_date, to_date)
|
|
results["daily_metrics"] = band_data.get("data", [])
|
|
results["days_synced"] = len(results["daily_metrics"])
|
|
except ZeppApiError as e:
|
|
logger.error(f"Failed to fetch band data: {e}")
|
|
|
|
# Fetch workouts
|
|
try:
|
|
workouts = client.fetch_workout_history(days)
|
|
results["workouts"] = workouts
|
|
results["workouts_synced"] = len(workouts)
|
|
except ZeppApiError as e:
|
|
logger.error(f"Failed to fetch workouts: {e}")
|
|
|
|
return results
|