279 lines
10 KiB
Python
279 lines
10 KiB
Python
"""
|
|
abricksetapi.py - Asynchronous Brickset API Client
|
|
This module provides asynchronous interactions with the Brickset API.
|
|
"""
|
|
|
|
import asyncio
|
|
import os
|
|
import aiohttp
|
|
import json
|
|
import math
|
|
import dotenv
|
|
from typing import Dict, Any, Optional
|
|
from datetime import datetime, timedelta
|
|
from async_lru import alru_cache
|
|
from . import throttler
|
|
|
|
Throttler = throttler.GlobalThrottler
|
|
|
|
dotenv.load_dotenv()
|
|
|
|
|
|
class AsyncBricksetAPI:
|
|
"""Asynchronous client for the Brickset API"""
|
|
|
|
def __init__(self, debug: bool = False, logger=None):
|
|
self.username = os.getenv("BRICKSET_USERNAME")
|
|
self.password = os.getenv("BRICKSET_PASS")
|
|
self.api_key = os.getenv("BRICKSET_API_KEY")
|
|
self.debug = debug
|
|
self.headers = {"UserAgent": "studspider/0.1"}
|
|
self.throttler = Throttler(rate_limit=500, period=60)
|
|
self.user_hash = None
|
|
self.logger = logger
|
|
|
|
# API usage tracking
|
|
self.daily_requests_used: Optional[int] = None
|
|
self.daily_requests_limit = 1000
|
|
self.usage_last_checked: Optional[datetime] = None
|
|
self.usage_check_interval = timedelta(hours=1) # Check usage every hour
|
|
|
|
async def get_key_usage_stats(self) -> Dict[str, Any]:
|
|
"""
|
|
Fetch API key usage statistics from Brickset.
|
|
|
|
Returns:
|
|
dict: Usage stats including dateStamp and apiCalls
|
|
"""
|
|
url = "https://brickset.com/api/v3.asmx/getKeyUsageStats"
|
|
params = {"apiKey": self.api_key}
|
|
|
|
async with aiohttp.ClientSession() as session:
|
|
async with session.get(
|
|
url, params=params, headers=self.headers
|
|
) as response:
|
|
if response.status == 200:
|
|
data = await response.read()
|
|
result = json.loads(data.decode("utf-8"))
|
|
|
|
if self.debug:
|
|
print(f"[@DEBUG] API usage stats: {result}")
|
|
if self.logger:
|
|
self.logger.debug(f"API usage stats: {result}")
|
|
|
|
return result
|
|
else:
|
|
error_msg = f"Failed to get usage stats: HTTP {response.status}"
|
|
if self.logger:
|
|
self.logger.error(error_msg)
|
|
raise Exception(error_msg)
|
|
|
|
async def check_and_update_usage(self) -> bool:
|
|
"""
|
|
Check current API usage and update internal counter.
|
|
Automatically called before each request if needed.
|
|
|
|
Returns:
|
|
bool: True if under limit, False if at/over limit
|
|
"""
|
|
now = datetime.now()
|
|
|
|
# Check usage if we haven't checked yet or if it's been over an hour
|
|
should_check = (
|
|
self.daily_requests_used is None
|
|
or self.usage_last_checked is None
|
|
or (now - self.usage_last_checked) > self.usage_check_interval
|
|
)
|
|
|
|
if should_check:
|
|
try:
|
|
stats = await self.get_key_usage_stats()
|
|
# Brickset returns {"status": "success", "apiKeyUsage": [...]}
|
|
usage_array = stats.get("apiKeyUsage", [])
|
|
if usage_array and len(usage_array) > 0:
|
|
today_stats = usage_array[0] # Most recent day
|
|
self.daily_requests_used = today_stats.get("count", 0)
|
|
self.usage_last_checked = now
|
|
|
|
if self.logger:
|
|
self.logger.info(
|
|
f"API usage: {self.daily_requests_used}/{self.daily_requests_limit}"
|
|
)
|
|
|
|
if self.debug:
|
|
print(
|
|
f"[@DEBUG] API usage: {self.daily_requests_used}/{self.daily_requests_limit}"
|
|
)
|
|
except Exception as e:
|
|
if self.logger:
|
|
self.logger.warning(f"Failed to check API usage: {e}")
|
|
# Don't block requests if usage check fails
|
|
return True
|
|
|
|
# Check if we're over the limit
|
|
if self.daily_requests_used is not None:
|
|
if self.daily_requests_used >= self.daily_requests_limit:
|
|
error_msg = f"Daily API limit reached: {self.daily_requests_used}/{self.daily_requests_limit}"
|
|
if self.logger:
|
|
self.logger.error(error_msg)
|
|
if self.debug:
|
|
print(f"[@ERROR] {error_msg}")
|
|
return False
|
|
|
|
# Warn when approaching limit
|
|
if self.daily_requests_used >= self.daily_requests_limit * 0.9:
|
|
warning_msg = f"Approaching daily API limit: {self.daily_requests_used}/{self.daily_requests_limit}"
|
|
if self.logger:
|
|
self.logger.warning(warning_msg)
|
|
if self.debug:
|
|
print(f"[@WARNING] {warning_msg}")
|
|
|
|
return True
|
|
|
|
async def login(self) -> Dict[str, Any]:
|
|
"""Asynchronously authenticate with Brickset API"""
|
|
url = "https://brickset.com/api/v3.asmx/login"
|
|
params = {
|
|
"username": self.username,
|
|
"password": self.password,
|
|
"apiKey": self.api_key,
|
|
}
|
|
|
|
async with aiohttp.ClientSession() as session:
|
|
async with session.get(
|
|
url, params=params, headers=self.headers # type: ignore
|
|
) as response:
|
|
if self.debug:
|
|
print(f"[@DEBUG] Login response status: {response.status}")
|
|
if self.logger:
|
|
self.logger.debug(f"Login response status: {response.status}")
|
|
|
|
data = await response.read()
|
|
result = json.loads(data.decode("utf-8"))
|
|
self.user_hash = result.get("hash")
|
|
|
|
if self.logger:
|
|
self.logger.info("Successfully authenticated with Brickset API")
|
|
return result
|
|
|
|
@alru_cache(maxsize=1000, ttl=604800) # Cache 1000 sets, 7 days TTL
|
|
async def get_set(self, set_number: int) -> Dict[str, Any]:
|
|
"""
|
|
Fetch LEGO set information asynchronously (cached for 7 days)
|
|
|
|
Args:
|
|
set_number (str): LEGO set number
|
|
|
|
Returns:
|
|
Dict containing set information
|
|
|
|
Raises:
|
|
Exception: If daily API limit exceeded or API error
|
|
"""
|
|
# Check API usage before making request
|
|
if not await self.check_and_update_usage():
|
|
raise Exception(
|
|
f"Brickset API daily limit reached ({self.daily_requests_limit} requests/day). "
|
|
"Please try again tomorrow."
|
|
)
|
|
|
|
if not self.user_hash:
|
|
await self.login()
|
|
|
|
url = "https://brickset.com/api/v3.asmx/getSets"
|
|
params = {
|
|
"apiKey": self.api_key,
|
|
"userHash": self.user_hash,
|
|
"params": json.dumps({"setNumber": f"{set_number}-1"}),
|
|
}
|
|
print(f"[@DEBUG] {params}...")
|
|
|
|
async with aiohttp.ClientSession() as session:
|
|
async with session.get(
|
|
url, params=params, headers=self.headers
|
|
) as response:
|
|
if self.debug:
|
|
print(f"[@DEBUG] Get set response status: {response.status}")
|
|
if self.logger:
|
|
self.logger.debug(f"Get set response status: {response.status}")
|
|
|
|
if response.status == 200:
|
|
data = await response.read()
|
|
result = json.loads(data.decode("utf-8"))
|
|
|
|
if not result.get("sets"):
|
|
error_msg = f"No data found for set {set_number}"
|
|
if self.logger:
|
|
self.logger.error(error_msg)
|
|
raise ValueError(error_msg)
|
|
|
|
if self.logger:
|
|
self.logger.info(
|
|
f"Successfully retrieved data for set {set_number}"
|
|
)
|
|
|
|
set_data = result["sets"][0]
|
|
|
|
# Try to get UPC barcode, fall back to EAN if not available
|
|
barcode_data = set_data.get("barcode", {})
|
|
upc_code = barcode_data.get("UPC") or barcode_data.get("EAN")
|
|
|
|
# Get retail price before rounding
|
|
retail_price = set_data["LEGOCom"]["US"]["retailPrice"]
|
|
|
|
our_set = {
|
|
"set_num": set_data["number"],
|
|
"set_name": set_data["name"],
|
|
"set_theme": set_data["theme"],
|
|
"set_image": set_data["image"]["thumbnailURL"],
|
|
"set_barcode": upc_code,
|
|
"set_retailprice": round(retail_price),
|
|
}
|
|
|
|
# Calculate adjusted price (10% markup), then round up
|
|
adjusted_price = retail_price * 1.10
|
|
our_set["set_adjustredprice"] = math.ceil(adjusted_price)
|
|
|
|
# Set PW inventory name with theme
|
|
our_set["pw_name"] = (
|
|
f"{our_set['set_num']} - {our_set['set_theme'].upper()} - {our_set['set_name']}"
|
|
)
|
|
|
|
# Flag for manual price review if needed
|
|
if our_set["set_retailprice"] >= 110:
|
|
our_set["requires_manual_price"] = True
|
|
|
|
return our_set
|
|
|
|
elif response.status == 500:
|
|
error_msg = (
|
|
"Internal Server Error: Please check API key and parameters."
|
|
)
|
|
if self.logger:
|
|
self.logger.error(error_msg)
|
|
raise Exception(error_msg)
|
|
else:
|
|
error_msg = f"Unexpected response: {response.status}"
|
|
if self.logger:
|
|
self.logger.error(error_msg)
|
|
raise Exception(error_msg)
|
|
|
|
async def throttled_get_set(self, set_number: str) -> Dict[str, Any]:
|
|
"""Get set information with rate limiting"""
|
|
async with self.throttler:
|
|
return await self.get_set(set_number)
|
|
|
|
|
|
async def main():
|
|
"""Example usage"""
|
|
client = AsyncBricksetAPI(debug=True)
|
|
try:
|
|
set_info = await client.throttled_get_set("10349")
|
|
print(json.dumps(set_info, indent=2))
|
|
except Exception as e:
|
|
print(f"Error: {e}")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
asyncio.run(main())
|