114 lines
3.2 KiB
Python
114 lines
3.2 KiB
Python
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)
|