jessikitty
f7c20de5ac
LIM configured no logging at all, so the instruction fetcher had no trail to follow. Adds a rotating file handler on the root logger plus two owner-only admin pages. - Admin > Logs: filter by level, source and text; tracebacks kept with their error; auto-refresh, raw download, rotated-file browsing and rotate-now - Admin > Logs > Brickset API calls: every outbound call recorded with method, params, caller, duration, result count and error, plus daily quota accounting (only getSets counts) and a comparison against getKeyUsageStats - New owner concept (services/ownership.py): a single account, distinct from is_admin, since these pages expose other admins activity. Defaults to the first admin; flask owner show/set to change - Services now log through module loggers; retention setting and pruning for call records - New CLI: flask owner show/set, flask logs path/prune - New table brickset_api_calls, created automatically at boot; LOG_* env vars, /data/logs in Docker
350 lines
13 KiB
Python
350 lines
13 KiB
Python
"""
|
|
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 []
|