Initial commit - LEGO Instructions Manager v1.5.0
This commit is contained in:
198
app/services/brickset_api.py
Normal file
198
app/services/brickset_api.py
Normal file
@@ -0,0 +1,198 @@
|
||||
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 []
|
||||
Reference in New Issue
Block a user