bambuddy/backend/app/services/bambu_cloud.py

"""
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