161 lines
5.8 KiB
Python
161 lines
5.8 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 dotenv
|
|
from typing import Dict, Any
|
|
from asyncio_throttle.throttler import Throttler
|
|
|
|
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": "Edderkop/0.1"}
|
|
self.throttler = Throttler(rate_limit=500, period=60)
|
|
self.user_hash = None
|
|
self.logger = logger
|
|
|
|
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
|
|
|
|
async def get_set(self, set_number: int) -> Dict[str, Any]:
|
|
"""
|
|
Fetch LEGO set information asynchronously
|
|
|
|
Args:
|
|
set_number (str): LEGO set number
|
|
|
|
Returns:
|
|
Dict containing set information
|
|
"""
|
|
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]
|
|
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": set_data["barcode"]["UPC"],
|
|
"set_retailprice": round(
|
|
set_data["LEGOCom"]["US"]["retailPrice"]
|
|
),
|
|
}
|
|
|
|
# Calculate adjusted price (10% markup)
|
|
retail_price = our_set["set_retailprice"]
|
|
our_set["set_adjustredprice"] = retail_price + (
|
|
(10 / 100) * retail_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())
|