""" 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 import logging import time from typing import Any, Dict, List, Optional import requests from flask import current_app, has_request_context, request from flask_login import current_user logger = logging.getLogger(__name__) class BricksetAPI: """Service for interacting with the Brickset API v3.""" BASE_URL = 'https://brickset.com/api/v3.asmx' def __init__(self, context: Optional[str] = None): 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 # Recorded against every call so the log shows which part of LIM # was responsible. Falls back to the endpoint name in a request. self.context = context or self._default_context() @staticmethod def _default_context() -> str: if has_request_context() and request.endpoint: return request.endpoint.split('.')[-1][:40] return 'background' def _log_call(self, method, params=None, ok=True, http_status=None, result_count=None, started=None, message=None): """Record one call. Never raises.""" duration = None if started is not None: duration = int((time.monotonic() - started) * 1000) if ok: logger.info('Brickset %s (%s) -> %s result(s) in %sms', method, params or '-', '?' if result_count is None else result_count, duration) else: logger.warning('Brickset %s (%s) failed: %s', method, params or '-', message) try: from app.models.api_call import BricksetApiCall user_id = None if has_request_context() and getattr(current_user, 'is_authenticated', False): user_id = current_user.id BricksetApiCall.record( method=method, params=params, ok=ok, http_status=http_status, result_count=result_count, duration_ms=duration, message=message, context=self.context, user_id=user_id) except Exception: logger.debug('Could not record Brickset call', exc_info=True) @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 started = time.monotonic() 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') self._log_call('login', self.username, ok=True, http_status=response.status_code, started=started) return self._user_hash self._log_call('login', self.username, ok=False, http_status=response.status_code, started=started, message=str(data.get('message'))) return None except Exception as e: self._log_call('login', self.username, ok=False, started=started, message=f'{type(e).__name__}: {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), } summary = json.dumps(search_params, separators=(',', ':')) started = time.monotonic() 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': sets = data.get('sets', []) self._log_call('getSets', summary, ok=True, http_status=response.status_code, result_count=len(sets), started=started) return sets message = data.get('message', 'unknown error') self._log_call('getSets', summary, ok=False, http_status=response.status_code, started=started, message=message) if 'limit exceeded' in message.lower(): raise RuntimeError('Brickset API limit exceeded') return [] except RuntimeError: raise except Exception as e: self._log_call('getSets', summary, ok=False, started=started, message=f'{type(e).__name__}: {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 [] started = time.monotonic() 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', []) names = [theme.get('theme') for theme in themes if theme.get('theme')] self._log_call('getThemes', ok=True, http_status=response.status_code, result_count=len(names), started=started) return names self._log_call('getThemes', ok=False, http_status=response.status_code, started=started, message=str(data.get('message'))) return [] except Exception as e: self._log_call('getThemes', ok=False, started=started, message=f'{type(e).__name__}: {e}') return [] # -- instructions ------------------------------------------------------- def _instructions_request(self, method: str, params: Dict[str, Any]) -> List[Dict[str, Any]]: summary = ', '.join(f'{k}={v}' for k, v in params.items()) params = dict(params, apiKey=self.api_key) started = time.monotonic() 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': instructions = data.get('instructions', []) or [] self._log_call(method, summary, ok=True, http_status=response.status_code, result_count=len(instructions), started=started) return instructions self._log_call(method, summary, ok=False, http_status=response.status_code, started=started, message=str(data.get('message'))) return [] except Exception as e: self._log_call(method, summary, ok=False, started=started, message=f'{type(e).__name__}: {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 [] started = time.monotonic() 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': usage = data.get('apiKeyUsage', []) or [] self._log_call('getKeyUsageStats', ok=True, http_status=response.status_code, result_count=len(usage), started=started) return usage self._log_call('getKeyUsageStats', ok=False, http_status=response.status_code, started=started, message=str(data.get('message'))) return [] except Exception as e: self._log_call('getKeyUsageStats', ok=False, started=started, message=f'{type(e).__name__}: {e}') return []