complete project restructure
This commit is contained in:
C committed
1 parent
a560273c3f
commit
a51c19ea53
11 files changed
+995
-1026
No files matched your search
@@ -0,0 +1,116 @@
|
||||
import os
|
||||
import requests
|
||||
import json
|
||||
import dotenv
|
||||
from typing import Dict, Any
|
||||
|
||||
dotenv.load_dotenv()
|
||||
|
||||
|
||||
class BricksetAPI:
|
||||
"""Class for interacting with the Brickset API"""
|
||||
|
||||
def __init__(self, debug: bool = False):
|
||||
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.user_hash = None
|
||||
|
||||
def login(self) -> Dict[str, Any]:
|
||||
"""Login to Brickset API and get user hash"""
|
||||
url = "https://brickset.com/api/v3.asmx/login"
|
||||
params = {
|
||||
"username": self.username,
|
||||
"password": self.password,
|
||||
"apiKey": self.api_key,
|
||||
}
|
||||
|
||||
response = requests.get(url, params=params, headers=self.headers)
|
||||
if self.debug:
|
||||
print(f"[@DEBUG] Login response: {response.status_code}")
|
||||
print(f"[@DEBUG] Response content: {response.text}")
|
||||
|
||||
try:
|
||||
data = response.json()
|
||||
self.user_hash = data.get("hash")
|
||||
if not self.user_hash:
|
||||
raise ValueError("No hash received in login response")
|
||||
return data
|
||||
except json.JSONDecodeError as e:
|
||||
print(f"[@ERROR] Failed to decode JSON response: {e}")
|
||||
print(f"[@DEBUG] Raw response: {response.text}")
|
||||
raise
|
||||
|
||||
def get_set(self, set_number: str) -> Dict[str, Any]:
|
||||
"""Get LEGO set information by set number"""
|
||||
if not self.user_hash:
|
||||
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"}),
|
||||
}
|
||||
|
||||
response = requests.get(url, params=params, headers=self.headers)
|
||||
|
||||
if self.debug:
|
||||
print(f"[@DEBUG] Get set response: {response.status_code}")
|
||||
print(f"[@DEBUG] Response content: {response.text}")
|
||||
|
||||
if response.status_code == 200:
|
||||
try:
|
||||
data = response.json()
|
||||
if not data.get("sets"):
|
||||
raise ValueError(f"No set data found for {set_number}")
|
||||
|
||||
set_data = data["sets"][0]
|
||||
our_set = {
|
||||
"set_num": set_data["number"],
|
||||
"set_name": set_data["name"],
|
||||
"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 Playwell inventory name
|
||||
if our_set["theme"] == "Duplo":
|
||||
our_set["pw_name"] = (
|
||||
f"{our_set['set_num']} - DUPLO {our_set['set_name']}"
|
||||
)
|
||||
else:
|
||||
our_set["pw_name"] = f"{our_set['set_num']} - {our_set['set_name']}"
|
||||
|
||||
return our_set
|
||||
|
||||
except json.JSONDecodeError as e:
|
||||
# print(f"[@ERROR] Failed to decode JSON response: {e}")
|
||||
# print(f"[@DEBUG] Raw response: {response.text}")
|
||||
raise
|
||||
except KeyError as e:
|
||||
# print(f"[@ERROR] Missing expected field in response: {e}")
|
||||
# print(f"[@DEBUG] Response data: {data}")
|
||||
raise
|
||||
elif response.status_code == 500:
|
||||
raise Exception(
|
||||
"Internal Server Error: Please check API key and parameters."
|
||||
)
|
||||
else:
|
||||
raise Exception(f"Unexpected response: {response.status_code}")
|
||||
|
||||
|
||||
def example():
|
||||
client = BricksetAPI(debug=True)
|
||||
try:
|
||||
set_info = client.get_set("10931")
|
||||
print(set_info)
|
||||
except Exception as e:
|
||||
print(f"Error: {e}")
|
||||
@@ -0,0 +1,113 @@
|
||||
import asyncio
|
||||
import aiohttp.web
|
||||
import uuid
|
||||
import os
|
||||
from base64 import b64decode
|
||||
import dotenv
|
||||
|
||||
dotenv.load_dotenv()
|
||||
SQUARE_LEGO_CAT_ID = os.environ.get("SQUARE_LEGO_CAT_ID")
|
||||
|
||||
from asyncio_throttle.throttler import Throttler
|
||||
|
||||
import asquare as sq
|
||||
import abricksetapi as alego
|
||||
|
||||
|
||||
def valid_uuid(val):
|
||||
try:
|
||||
uuid.UUID(str(val))
|
||||
return True
|
||||
except ValueError:
|
||||
return False
|
||||
|
||||
|
||||
async def fullsync():
|
||||
"""'
|
||||
full_db = await sq.acat2id(SQUARE_LEGO_CAT_ID)
|
||||
for l_set in full_db:
|
||||
l_setinfo = await alego.throttled_get_set(l_set)
|
||||
await sq.update_itm(l_set, l_setinfo)"""
|
||||
|
||||
|
||||
async def stop():
|
||||
pass
|
||||
|
||||
|
||||
async def priceonly():
|
||||
pass
|
||||
|
||||
|
||||
async def handle_post(request):
|
||||
|
||||
valid_cmd = ["RUNFULLSYNC", "STOP", "RUNPRICEONLY"]
|
||||
|
||||
async def handle_cmd(cmd: str):
|
||||
if cmd not in valid_cmd:
|
||||
return False
|
||||
else:
|
||||
return True
|
||||
|
||||
idp_key = request.headers.get("Idempotency-Key")
|
||||
user_agent = request.headers.get("User-Agent")
|
||||
|
||||
if not idp_key and not valid_uuid(idp_key) and user_agent != "Kopling/0.1":
|
||||
# 412 Precondition Failed
|
||||
# The server does not meet one of the preconditions that the requester put on the request header fields.
|
||||
return aiohttp.web.Response(status=412, text="412 Headers rejected!")
|
||||
|
||||
else:
|
||||
data = await request.post()
|
||||
reqdata = data.get(b64decode("CMD"))
|
||||
|
||||
if await handle_cmd(reqdata):
|
||||
match reqdata:
|
||||
case "RUNFULLSYNC":
|
||||
print("Running full sync...")
|
||||
await fullsync()
|
||||
return True
|
||||
case "STOP":
|
||||
await stop()
|
||||
return True
|
||||
case "RUNPRICEONLY":
|
||||
await priceonly()
|
||||
return True
|
||||
else:
|
||||
return aiohttp.web.Response(status=400, text="400 Bad Request.")
|
||||
result = {"message": f"studspider/0.1 received command: {reqdata}"}
|
||||
return aiohttp.web.json_response(status=200, data=result)
|
||||
|
||||
|
||||
app = aiohttp.web.Application()
|
||||
app.router.add_post("/", handle_post)
|
||||
|
||||
import subprocess
|
||||
|
||||
|
||||
def generate_toilet_banner(text, font="smblock", filter_effect=None, colors=None):
|
||||
command = ["toilet", "-f", font]
|
||||
if filter_effect:
|
||||
command.extend(["-F", filter_effect])
|
||||
if colors:
|
||||
# Toilet uses a specific format for color arguments, e.g., --gay or --metal
|
||||
# For general colors, you might need to use specific filters or direct color codes.
|
||||
# This example focuses on filters that inherently use colors.
|
||||
pass # More complex color handling would be needed for arbitrary colors
|
||||
|
||||
command.append(text)
|
||||
|
||||
try:
|
||||
result = subprocess.run(command, capture_output=True, text=True, check=True)
|
||||
return result.stdout
|
||||
except subprocess.CalledProcessError as e:
|
||||
return f"Error executing toilet: {e}\n{e.stderr}"
|
||||
|
||||
|
||||
async def main():
|
||||
throttler = Throttler(rate_limit=500, period=60) # 500 requests per minute
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
print(generate_toilet_banner("studspider v0.1"))
|
||||
print("Built for PLAY-WELL TEKNOLOGIES by contact@traps.moe\n\n")
|
||||
aiohttp.web.run_app(app, port=7432)
|
||||
Reference in new issue
Block a user