mirror of
https://github.com/maziggy/bambuddy.git
synced 2026-10-21 10:15:04 +02:00
Added screenshots
This commit is contained in:
Martin Ziegler
@@ -0,0 +1,256 @@
|
||||
"""
|
||||
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
|
||||
Reference in New Issue
Block a user