mirror of
https://github.com/maziggy/bambuddy.git
synced 2026-09-30 11:12:35 +02:00
257 lines
8.2 KiB
Python
257 lines
8.2 KiB
Python
"""
|
|
Bambu Lab Cloud API Service
|
|
|
|
Handles authentication and profile management with Bambu Lab's cloud services.
|
|
"""
|
|
|
|
import httpx
|
|
import json
|
|
import logging
|
|
from typing import Optional
|
|
from pathlib import Path
|
|
from datetime import datetime, timedelta
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
BAMBU_API_BASE = "https://api.bambulab.com"
|
|
BAMBU_API_BASE_CN = "https://api.bambulab.cn"
|
|
|
|
|
|
class BambuCloudError(Exception):
|
|
"""Base exception for Bambu Cloud errors."""
|
|
pass
|
|
|
|
|
|
class BambuCloudAuthError(BambuCloudError):
|
|
"""Authentication related errors."""
|
|
pass
|
|
|
|
|
|
class BambuCloudService:
|
|
"""Service for interacting with Bambu Lab Cloud API."""
|
|
|
|
def __init__(self, region: str = "global"):
|
|
self.base_url = BAMBU_API_BASE if region == "global" else BAMBU_API_BASE_CN
|
|
self.access_token: Optional[str] = None
|
|
self.refresh_token: Optional[str] = None
|
|
self.token_expiry: Optional[datetime] = None
|
|
self._client = httpx.AsyncClient(timeout=30.0)
|
|
|
|
@property
|
|
def is_authenticated(self) -> bool:
|
|
"""Check if we have a valid token."""
|
|
if not self.access_token:
|
|
return False
|
|
if self.token_expiry and datetime.now() > self.token_expiry:
|
|
return False
|
|
return True
|
|
|
|
def _get_headers(self) -> dict:
|
|
"""Get headers for authenticated requests."""
|
|
headers = {
|
|
"Content-Type": "application/json",
|
|
"User-Agent": "BambuTrack/1.0",
|
|
}
|
|
if self.access_token:
|
|
headers["Authorization"] = f"Bearer {self.access_token}"
|
|
return headers
|
|
|
|
async def login_request(self, email: str, password: str) -> dict:
|
|
"""
|
|
Initiate login - this will trigger a verification code email.
|
|
|
|
Returns dict with login status and whether verification is needed.
|
|
"""
|
|
try:
|
|
response = await self._client.post(
|
|
f"{self.base_url}/v1/user-service/user/login",
|
|
headers={"Content-Type": "application/json"},
|
|
json={
|
|
"account": email,
|
|
"password": password,
|
|
}
|
|
)
|
|
|
|
data = response.json()
|
|
|
|
if response.status_code == 200:
|
|
# Check if we need verification code
|
|
# Bambu API returns loginType or may require tfaKey
|
|
if data.get("loginType") == "verifyCode" or "tfaKey" in data:
|
|
return {
|
|
"success": False,
|
|
"needs_verification": True,
|
|
"message": "Verification code sent to email"
|
|
}
|
|
|
|
# Direct login success (rare, usually needs 2FA)
|
|
if "accessToken" in data:
|
|
self._set_tokens(data)
|
|
return {
|
|
"success": True,
|
|
"needs_verification": False,
|
|
"message": "Login successful"
|
|
}
|
|
|
|
# Handle specific error codes
|
|
error_msg = data.get("message") or data.get("error") or "Login failed"
|
|
return {
|
|
"success": False,
|
|
"needs_verification": False,
|
|
"message": error_msg
|
|
}
|
|
|
|
except Exception as e:
|
|
logger.error(f"Login request failed: {e}")
|
|
raise BambuCloudAuthError(f"Login request failed: {e}")
|
|
|
|
async def verify_code(self, email: str, code: str) -> dict:
|
|
"""
|
|
Complete login with verification code.
|
|
"""
|
|
try:
|
|
response = await self._client.post(
|
|
f"{self.base_url}/v1/user-service/user/login",
|
|
headers={"Content-Type": "application/json"},
|
|
json={
|
|
"account": email,
|
|
"code": code,
|
|
}
|
|
)
|
|
|
|
data = response.json()
|
|
|
|
if response.status_code == 200 and "accessToken" in data:
|
|
self._set_tokens(data)
|
|
return {
|
|
"success": True,
|
|
"message": "Login successful"
|
|
}
|
|
|
|
return {
|
|
"success": False,
|
|
"message": data.get("message", "Verification failed")
|
|
}
|
|
|
|
except Exception as e:
|
|
logger.error(f"Verification failed: {e}")
|
|
raise BambuCloudAuthError(f"Verification failed: {e}")
|
|
|
|
def _set_tokens(self, data: dict):
|
|
"""Set tokens from login response."""
|
|
self.access_token = data.get("accessToken")
|
|
self.refresh_token = data.get("refreshToken")
|
|
# Token typically valid for ~3 months, but we'll refresh more often
|
|
self.token_expiry = datetime.now() + timedelta(days=30)
|
|
|
|
def set_token(self, access_token: str):
|
|
"""Set access token directly (for stored tokens)."""
|
|
self.access_token = access_token
|
|
self.token_expiry = datetime.now() + timedelta(days=30)
|
|
|
|
def logout(self):
|
|
"""Clear authentication state."""
|
|
self.access_token = None
|
|
self.refresh_token = None
|
|
self.token_expiry = None
|
|
|
|
async def get_user_profile(self) -> dict:
|
|
"""Get user profile information."""
|
|
if not self.is_authenticated:
|
|
raise BambuCloudAuthError("Not authenticated")
|
|
|
|
try:
|
|
response = await self._client.get(
|
|
f"{self.base_url}/v1/design-user-service/my/preference",
|
|
headers=self._get_headers()
|
|
)
|
|
|
|
if response.status_code == 200:
|
|
return response.json()
|
|
|
|
raise BambuCloudError(f"Failed to get profile: {response.status_code}")
|
|
|
|
except httpx.RequestError as e:
|
|
raise BambuCloudError(f"Request failed: {e}")
|
|
|
|
async def get_slicer_settings(self, version: str = "01.09.00.00") -> dict:
|
|
"""
|
|
Get all slicer settings (filament, printer, process presets).
|
|
|
|
Args:
|
|
version: Slicer version string
|
|
"""
|
|
if not self.is_authenticated:
|
|
raise BambuCloudAuthError("Not authenticated")
|
|
|
|
try:
|
|
response = await self._client.get(
|
|
f"{self.base_url}/v1/iot-service/api/slicer/setting",
|
|
headers=self._get_headers(),
|
|
params={"version": version}
|
|
)
|
|
|
|
data = response.json()
|
|
|
|
if response.status_code == 200:
|
|
return data
|
|
|
|
raise BambuCloudError(f"Failed to get settings: {response.status_code}")
|
|
|
|
except httpx.RequestError as e:
|
|
raise BambuCloudError(f"Request failed: {e}")
|
|
|
|
async def get_setting_detail(self, setting_id: str) -> dict:
|
|
"""Get detailed information for a specific setting/preset."""
|
|
if not self.is_authenticated:
|
|
raise BambuCloudAuthError("Not authenticated")
|
|
|
|
try:
|
|
response = await self._client.get(
|
|
f"{self.base_url}/v1/iot-service/api/slicer/setting/{setting_id}",
|
|
headers=self._get_headers()
|
|
)
|
|
|
|
if response.status_code == 200:
|
|
return response.json()
|
|
|
|
raise BambuCloudError(f"Failed to get setting detail: {response.status_code}")
|
|
|
|
except httpx.RequestError as e:
|
|
raise BambuCloudError(f"Request failed: {e}")
|
|
|
|
async def get_devices(self) -> dict:
|
|
"""Get list of bound devices."""
|
|
if not self.is_authenticated:
|
|
raise BambuCloudAuthError("Not authenticated")
|
|
|
|
try:
|
|
response = await self._client.get(
|
|
f"{self.base_url}/v1/iot-service/api/user/bind",
|
|
headers=self._get_headers()
|
|
)
|
|
|
|
if response.status_code == 200:
|
|
return response.json()
|
|
|
|
raise BambuCloudError(f"Failed to get devices: {response.status_code}")
|
|
|
|
except httpx.RequestError as e:
|
|
raise BambuCloudError(f"Request failed: {e}")
|
|
|
|
async def close(self):
|
|
"""Close the HTTP client."""
|
|
await self._client.aclose()
|
|
|
|
|
|
# Singleton instance
|
|
_cloud_service: Optional[BambuCloudService] = None
|
|
|
|
|
|
def get_cloud_service() -> BambuCloudService:
|
|
"""Get the singleton cloud service instance."""
|
|
global _cloud_service
|
|
if _cloud_service is None:
|
|
_cloud_service = BambuCloudService()
|
|
return _cloud_service
|