lego-instructions-manager/app/services/brickset_api.py

"""
Brickset API v3 client.

https://brickset.com/article/52664/api-version-3-documentation

Two things worth knowing about the API, both of which shape this file:

  * getSets takes its search terms as a JSON string. Python's str() on a dict
    produces single quotes, which is not JSON — hence json.dumps below.
  * getInstructions takes a Brickset setID; getInstructions2 takes a set
    number like "10497-1". They are different methods and passing a set
    number to the first one quietly returns nothing.

Only getSets counts against the daily key quota.

Created by: Jess Rogerson (yelling commands at Claude.AI)
"""

import json
from typing import Any, Dict, List, Optional

import requests
from flask import current_app


class BricksetAPI:
    """Service for interacting with the Brickset API v3."""

    BASE_URL = 'https://brickset.com/api/v3.asmx'

    def __init__(self):
        self.api_key = current_app.config.get('BRICKSET_API_KEY')
        self.username = current_app.config.get('BRICKSET_USERNAME')
        self.password = current_app.config.get('BRICKSET_PASSWORD')
        self._user_hash = None

    @staticmethod
    def is_configured() -> bool:
        """Check if Brickset API is properly configured."""
        return bool(
            current_app.config.get('BRICKSET_API_KEY') and
            current_app.config.get('BRICKSET_USERNAME') and
            current_app.config.get('BRICKSET_PASSWORD')
        )

    def _get_user_hash(self) -> Optional[str]:
        """Authenticate and get user hash token."""
        if self._user_hash:
            return self._user_hash

        if not all([self.api_key, self.username, self.password]):
            current_app.logger.warning('Brickset API credentials not configured')
            return None

        try:
            response = requests.get(
                f'{self.BASE_URL}/login',
                params={
                    'apiKey': self.api_key,
                    'username': self.username,
                    'password': self.password
                },
                timeout=10
            )
            response.raise_for_status()
            data = response.json()

            if data.get('status') == 'success':
                self._user_hash = data.get('hash')
                return self._user_hash
            else:
                current_app.logger.error(f"Brickset login failed: {data.get('message')}")
                return None

        except Exception as e:
            current_app.logger.error(f'Brickset API authentication error: {str(e)}')
            return None

    def search_sets(self,
                    set_number: Optional[str] = None,
                    theme: Optional[str] = None,
                    year: Optional[int] = None,
                    query: Optional[str] = None,
                    page_size: Optional[int] = None) -> List[Dict[str, Any]]:
        """
        Search for LEGO sets using various parameters.

        Args:
            set_number: Specific set number to search for
            theme: Theme name to filter by
            year: Year released
            query: General search query
            page_size: Number of records to return (max 500)

        Returns:
            List of set dictionaries
        """
        user_hash = self._get_user_hash()
        if not user_hash:
            return []

        # Build search parameters
        search_params: Dict[str, Any] = {}
        if set_number:
            search_params['setNumber'] = set_number
        if theme:
            search_params['theme'] = theme
        if year:
            search_params['year'] = year
        if query:
            search_params['query'] = query
        if page_size:
            search_params['pageSize'] = page_size

        if not search_params:
            return []

        params = {
            'apiKey': self.api_key,
            'userHash': user_hash,
            # Must be a JSON string — str(dict) gives single quotes and is
            # rejected with "Parameter error".
            'params': json.dumps(search_params),
        }

        try:
            response = requests.get(
                f'{self.BASE_URL}/getSets',
                params=params,
                timeout=15
            )
            response.raise_for_status()
            data = response.json()

            if data.get('status') == 'success':
                return data.get('sets', [])

            message = data.get('message', 'unknown error')
            current_app.logger.error(f'Brickset search failed: {message}')
            if 'limit exceeded' in message.lower():
                raise RuntimeError('Brickset API limit exceeded')
            return []

        except RuntimeError:
            raise
        except Exception as e:
            current_app.logger.error(f'Brickset API search error: {str(e)}')
            return []

    def get_set_by_number(self, set_number: str) -> Optional[Dict[str, Any]]:
        """
        Get detailed information for a specific set by its number.

        Args:
            set_number: The LEGO set number (e.g., "10497" or "10497-1")

        Returns:
            Dictionary with set information or None
        """
        results = self.search_sets(set_number=set_number)
        if not results and '-' not in set_number:
            # Brickset stores numbers with a variant suffix.
            results = self.search_sets(set_number=f'{set_number}-1')
        return results[0] if results else None

    def get_themes(self) -> List[str]:
        """
        Get list of all available LEGO themes.

        Returns:
            List of theme names
        """
        user_hash = self._get_user_hash()
        if not user_hash:
            return []

        try:
            response = requests.get(
                f'{self.BASE_URL}/getThemes',
                params={
                    'apiKey': self.api_key,
                    'userHash': user_hash
                },
                timeout=10
            )
            response.raise_for_status()
            data = response.json()

            if data.get('status') == 'success':
                themes = data.get('themes', [])
                return [theme.get('theme') for theme in themes if theme.get('theme')]
            else:
                return []

        except Exception as e:
            current_app.logger.error(f'Brickset API themes error: {str(e)}')
            return []

    # -- instructions -------------------------------------------------------

    def _instructions_request(self, method: str,
                              params: Dict[str, Any]) -> List[Dict[str, Any]]:
        params = dict(params, apiKey=self.api_key)
        try:
            response = requests.get(f'{self.BASE_URL}/{method}',
                                    params=params, timeout=20)
            response.raise_for_status()
            data = response.json()

            if data.get('status') == 'success':
                return data.get('instructions', []) or []

            current_app.logger.error(
                f"Brickset {method} failed: {data.get('message')}")
            return []

        except Exception as e:
            current_app.logger.error(f'Brickset {method} error: {str(e)}')
            return []

    def get_instructions(self, set_number: str) -> List[Dict[str, Any]]:
        """
        Instructions for a set number, e.g. "10497-1" (getInstructions2).

        Returns a list of {'URL': ..., 'description': ...} dictionaries.
        Does not need a user hash and does not count against the key quota.
        """
        if not self.api_key or not set_number:
            return []
        return self._instructions_request('getInstructions2',
                                          {'setNumber': set_number})

    def get_instructions_by_id(self, set_id: int) -> List[Dict[str, Any]]:
        """Instructions for an internal Brickset set ID (getInstructions)."""
        if not self.api_key or not set_id:
            return []
        return self._instructions_request('getInstructions', {'setID': set_id})

    # -- quota --------------------------------------------------------------

    def get_key_usage(self) -> List[Dict[str, Any]]:
        """
        Daily getSets usage for the last 30 days.

        Returns a list of {'dateStamp': ..., 'count': ...} dictionaries.
        """
        if not self.api_key:
            return []
        try:
            response = requests.get(f'{self.BASE_URL}/getKeyUsageStats',
                                    params={'apiKey': self.api_key},
                                    timeout=10)
            response.raise_for_status()
            data = response.json()
            if data.get('status') == 'success':
                return data.get('apiKeyUsage', []) or []
            return []
        except Exception as e:
            current_app.logger.error(f'Brickset key usage error: {str(e)}')
            return []