199 lines
6.3 KiB
Python
199 lines
6.3 KiB
Python
import requests
|
|
from flask import current_app
|
|
from typing import Optional, List, Dict, Any
|
|
|
|
|
|
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) -> 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
|
|
|
|
Returns:
|
|
List of set dictionaries
|
|
"""
|
|
user_hash = self._get_user_hash()
|
|
if not user_hash:
|
|
return []
|
|
|
|
params = {
|
|
'apiKey': self.api_key,
|
|
'userHash': user_hash,
|
|
'params': '{}' # JSON params object
|
|
}
|
|
|
|
# Build search parameters
|
|
search_params = {}
|
|
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
|
|
|
|
params['params'] = str(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', [])
|
|
else:
|
|
current_app.logger.error(f"Brickset search failed: {data.get('message')}")
|
|
return []
|
|
|
|
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")
|
|
|
|
Returns:
|
|
Dictionary with set information or None
|
|
"""
|
|
results = self.search_sets(set_number=set_number)
|
|
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 []
|
|
|
|
def get_instructions(self, set_number: str) -> List[Dict[str, Any]]:
|
|
"""
|
|
Get instruction information for a specific set.
|
|
|
|
Args:
|
|
set_number: The LEGO set number
|
|
|
|
Returns:
|
|
List of instruction dictionaries
|
|
"""
|
|
user_hash = self._get_user_hash()
|
|
if not user_hash:
|
|
return []
|
|
|
|
try:
|
|
response = requests.get(
|
|
f'{self.BASE_URL}/getInstructions',
|
|
params={
|
|
'apiKey': self.api_key,
|
|
'userHash': user_hash,
|
|
'setNumber': set_number
|
|
},
|
|
timeout=10
|
|
)
|
|
response.raise_for_status()
|
|
data = response.json()
|
|
|
|
if data.get('status') == 'success':
|
|
return data.get('instructions', [])
|
|
else:
|
|
return []
|
|
|
|
except Exception as e:
|
|
current_app.logger.error(f'Brickset API instructions error: {str(e)}')
|
|
return []
|