Best Offer Aggregator API across Opensea, Looksrare, X2Y2, Sudoswap, NFTX

lambda_function.py

import json
import threading
from decimal import Decimal
from time import time

import boto3
import cloudscraper
import requests
from bs4 import BeautifulSoup


class UpdateDB:
    def __init__(self, collection, test=False):
        self.best_offers = {
            "collection": collection,
            "timestamp": int(time()),
            "opensea": None,
            "looksrare": None,
            "x2y2": None,
            "sudoswap": None,
            "nftx": None,
        }
        if not test:
            cached_data = self.check_db(collection)
            if cached_data and int(cached_data["timestamp"]) + 900 > time():
                self.best_offers = cached_data
                return
        opensea_thread = threading.Thread(target=self.get_opensea, args=(collection,))
        looksrare_thread = threading.Thread(
            target=self.get_looksrare, args=(collection,)
        )
        x2y2_thread = threading.Thread(target=self.get_x2y2, args=(collection,))
        sudoswap_thread = threading.Thread(target=self.get_sudoswap, args=(collection,))
        nftx_thread = threading.Thread(target=self.get_nftx, args=(collection,))
        opensea_thread.start()
        looksrare_thread.start()
        x2y2_thread.start()
        sudoswap_thread.start()
        nftx_thread.start()
        opensea_thread.join()
        x2y2_thread.join()
        looksrare_thread.join()
        sudoswap_thread.join()
        nftx_thread.join()
        if not test:
            self.update_db()
        else:
            print(self.best_offers)

    def return_best_offers(self):
        self.best_offers["timestamp"] = int(self.best_offers["timestamp"])
        self.best_offers["opensea"] = float(
            self.best_offers["opensea"] if self.best_offers["opensea"] else 0
        )
        self.best_offers["looksrare"] = float(
            self.best_offers["looksrare"] if self.best_offers["looksrare"] else 0
        )
        self.best_offers["x2y2"] = float(
            self.best_offers["x2y2"] if self.best_offers["x2y2"] else 0
        )
        self.best_offers["sudoswap"] = float(
            self.best_offers["sudoswap"] if self.best_offers["sudoswap"] else 0
        )
        self.best_offers["nftx"] = float(
            self.best_offers["nftx"] if self.best_offers["nftx"] else 0
        )
        return {
            "statusCode": 200,
            "headers": {"Content-Type": "application/json"},
            "body": json.dumps(self.best_offers),
        }

    def check_db(self, collection):
        client_dynamo = boto3.resource("dynamodb")
        table_dynamo = client_dynamo.Table("best_offers")
        try:
            res = table_dynamo.get_item(Key={"collection": collection})["Item"]
            return res
        except Exception:
            return False

    def update_db(self):
        client_dynamo = boto3.resource("dynamodb")
        table_dynamo = client_dynamo.Table("best_offers")
        try:
            table_dynamo.put_item(Item=self.best_offers)
        except Exception:
            raise

    def get_opensea(self, collection):
        try:
            url = f"https://api.opensea.io/api/v1/asset_contract/{collection}"
            headers = {"X-API-KEY": "2972994f68bb4dfc9f68c944f473c329"}
            collection_name = requests.get(url, headers=headers).json()["collection"][
                "slug"
            ]
            scraper = cloudscraper.create_scraper()
            html_res = scraper.get(
                f"https://opensea.io/collection/{collection_name}"
            ).text
            soup = BeautifulSoup(html_res, "html.parser")
            element = soup.find_all(
                "span",
                class_="sc-1xf18x6-0 sc-1aqfqq9-0 haVRLx cUPmoY styledPhoenixText",
            )[4]
            self.best_offers["opensea"] = Decimal(element.text)
        except Exception:
            self.best_offers["opensea"] = None

    def get_looksrare(self, collection):
        url = f"https://api.looksrare.org/api/v1/orders?isOrderAsk=false&collection={collection}&status%5B%5D=VALID&sort=PRICE_DESC"
        headers = {
            "Accept": "application/json",
            "X-API-KEY": "ajZV70CYg4ask6yHVtfIldXQ",
        }
        try:
            response = requests.get(url, headers=headers, timeout=5)
            self.best_offers["looksrare"] = Decimal(
                "%.3f" % (float(response.json()["data"][0]["price"]) / 1e18)
            )
        except Exception:
            self.best_offers["looksrare"] = None

    def get_x2y2(self, collection):
        headers = {
            "Accept": "application/json",
            "X-API-KEY": "a5926278-bd6c-449a-8b67-366d03a61eb4",
        }
        params = {
            "contract": collection,
            "sort": "price",
        }
        try:
            self.best_offers["x2y2"] = Decimal(
                "%.3f"
                % (
                    float(
                        requests.get(
                            "https://api.x2y2.org/v1/offers",
                            headers=headers,
                            params=params,
                            timeout=5,
                        ).json()["data"][-1]["price"]
                    )
                    / 1e18
                )
            )
        except Exception:
            self.best_offers["x2y2"] = None

    def get_sudoswap(self, collection):
        client_dynamo = boto3.resource("dynamodb")
        table_dynamo = client_dynamo.Table("sudoswap")
        try:
            res = table_dynamo.get_item(Key={"address": collection.lower()})["Item"]
            self.best_offers["sudoswap"] = Decimal(
                "%.3f" % (float(res["sell_quote"]) / 1e18)
            )
        except Exception:
            self.best_offers["sudoswap"] = None

    def get_nftx(self, collection):
        url = f"http://api.spicyest.com/nftx_sell_price?address={collection}"
        headers = {"Accept": "*/*", "X-API-Key": "eyJhbGciOiJIUzI1NiIsInR5cCI6IkpXVCJ9"}
        try:
            response = requests.get(url, headers=headers, timeout=5)
            self.best_offers["nftx"] = Decimal(
                "%.3f" % (float(response.json()["price"]))
            )
        except Exception:
            self.best_offers["nftx"] = None


def lambda_handler(event, context):
    return UpdateDB(event["queryStringParameters"]["collection"]).return_best_offers()