Add files via upload

snipe.py

@@ -0,0 +1,175 @@
+import asyncio
+import re
+from concurrent.futures import ThreadPoolExecutor
+from timeit import default_timer
+import time
+import requests
+from datetime import datetime, timezone
+
+c = requests.get("https://api.hypixel.net/skyblock/auctions?page=0")
+resp = c.json()
+now = resp['lastUpdated']
+toppage = resp['totalPages']
+
+results = []
+prices = {}
+
+# stuff to remove
+REFORGES = [" ✦", "⚚ ", " ✪", "✪", "Stiff ", "Lucky ", "Jerry's ", "Dirty ", "Fabled ", "Suspicious ", "Gilded ", "Warped ", "Withered ", "Bulky ", "Stellar ", "Heated ", "Ambered ", "Fruitful ", "Magnetic ", "Fleet ", "Mithraic ", "Auspicious ", "Refined ", "Headstrong ", "Precise ", "Spiritual ", "Moil ", "Blessed ", "Toil ", "Bountiful ", "Candied ", "Submerged ", "Reinforced ", "Cubic ", "Warped ", "Undead ", "Ridiculous ", "Necrotic ", "Spiked ", "Jaded ", "Loving ", "Perfect ", "Renowned ", "Giant ", "Empowered ", "Ancient ", "Sweet ", "Silky ", "Bloody ", "Shaded ", "Gentle ", "Odd ", "Fast ", "Fair ", "Epic ", "Sharp ", "Heroic ", "Spicy ", "Legendary ", "Deadly ", "Fine ", "Grand ", "Hasty ", "Neat ", "Rapid ", "Unreal ", "Awkward ", "Rich ", "Clean ", "Fierce ", "Heavy ", "Light ", "Mythic ", "Pure ", "Smart ", "Titanic ", "Wise ", "Bizarre ", "Itchy ", "Ominous ", "Pleasant ", "Pretty ", "Shiny ", "Simple ", "Strange ", "Vivid ", "Godly ", "Demonic ", "Forceful ", "Hurtful ", "Keen ", "Strong ", "Superior ", "Unpleasant ", "Zealous "]
+
+# Constant for the discord webhook you want the messages sent to; you need to change this
+DISCORD_WEBHOOK = "CHANGEME"
+
+# Constant for the lowest priced item you want to be shown to you; feel free to change this
+LOWEST_PRICE = 5
+
+# Constant for the lowest profit in a fli that you want to be shown to you; feel free to change this
+LOWEST_PROFIT = 1000000 # Minimum 1 million coins profit to show
+
+# Constant for the lowest percent difference you want to be shown to you; feel free to change this
+LOWEST_PERCENT_MARGIN = 1/2
+
+START_TIME = default_timer()
+
+def fetch(session, page):
+    global toppage
+    base_url = "https://api.hypixel.net/skyblock/auctions?page="
+    with session.get(base_url + page) as response:
+        # puts response in a dict
+        data = response.json()
+        toppage = data['totalPages']
+        if data['success']:
+            toppage = data['totalPages']
+            for auction in data['auctions']:
+                if not auction['claimed'] and auction['bin'] == True and not "Furniture" in auction["item_lore"]: # if the auction isn't a) claimed and is b) BIN
+                    # removes level if it's a pet, also 
+                    index = re.sub("\[[^\]]*\]", "", auction['item_name']) + auction['tier']
+                    # removes reforges and other yucky characters
+                    for reforge in REFORGES: index = index.replace(reforge, "")
+                    # if the current item already has a price in the prices map, the price is updated
+                    if index in prices:
+                        if prices[index][0] > auction['starting_bid']:
+                            prices[index][1] = prices[index][0]
+                            prices[index][0] = auction['starting_bid']
+                        elif prices[index][1] > auction['starting_bid']:
+                            prices[index][1] = auction['starting_bid']
+                    # otherwise, it's added to the prices map
+                    else:
+                        prices[index] = [auction['starting_bid'], float("inf")]
+
+                    # if the auction fits in some parameters
+                    if prices[index][1] > LOWEST_PRICE and prices[index][0]/prices[index][1] < LOWEST_PERCENT_MARGIN and auction['start']+60000 > now:
+                        results.append([auction['uuid'], auction['item_name'], auction['starting_bid'], index])
+        return data
+
+async def get_data_asynchronous():
+    # puts all the page strings
+    pages = [str(x) for x in range(toppage)]
+    with ThreadPoolExecutor(max_workers=10) as executor:
+        with requests.Session() as session:
+            loop = asyncio.get_event_loop()
+            START_TIME = default_timer()
+            tasks = [
+                loop.run_in_executor(
+                    executor,
+                    fetch,
+                    *(session, page) # Allows us to pass in multiple arguments to `fetch`
+                )
+                # runs for every page
+                for page in pages if int(page) < toppage
+            ]
+            for response in await asyncio.gather(*tasks):
+                pass
+def send(price, name, profit, uuid, test1, test2):
+    url = DISCORD_WEBHOOK
+    data = {
+    "embeds": [
+        {
+       "title": "Snipe Incoming!", #feel free to change this
+        "description": "Found a cool snipe from the BINs for ya!", #feel free to change this
+       "color": 3388452,
+       "timestamp": str(datetime.now(timezone.utc)),
+       "footer": {
+         "icon_url": "https://samzy.dev/files/img/7d97481b1fe66f4b51db90da7e794d9f.webp", #feel free to change this
+         "text": "Made by SamzyDev#1205", #feel free to change this
+          },
+          "fields": [
+            {
+              "name": "Item Name",
+              "value": "```" + str(name) + "```"
+            },
+            {
+              "name": "Auction",
+              "value": "```/viewauction " + str(uuid) + "```"
+            },
+            {
+             "name": "Price",
+             "value": "```{:,}```".format(price)
+           },
+           {
+             "name": "Potential Profit",
+             "value": "```{:,}```".format(int(profit))
+           },
+           {
+             "name": "Second Lowest BIN - Lowest BIN Price",
+             "value": "```" + str("{:,}").format(test1) + " - " + str("{:,}").format(test2) + "```"
+           }
+         ]
+      }
+    ]
+    }
+    result = requests.post(url, json=data)
+    if 200 <= result.status_code < 300:
+      print(f"Message Sent!")
+    else:
+       print(f"Not sent with {result.status_code}, response:\n{result.json()}") #most likey a rate limit
+       time.sleep(5) #wait 5 seconds to send it
+       result2 = requests.post(url, json=data)
+       if 200 <= result2.status_code < 300:
+        print(f"Message Sent!")
+       else:
+        print(f"Not sent with {result2.status_code}, response:\n{result2.json()}")
+
+def main():
+    # Resets variables
+    global results, prices, START_TIME
+    START_TIME = default_timer()
+    results = []
+    prices = {}
+
+    loop = asyncio.new_event_loop()
+    asyncio.set_event_loop(loop)
+    future = asyncio.ensure_future(get_data_asynchronous())
+    loop.run_until_complete(future)
+
+    # Makes sure all the results are still up to date
+    if len(results): results = [[entry, prices[entry[3]][1]] for entry in results if (entry[2] > LOWEST_PRICE and prices[entry[3]][1] != float('inf') and prices[entry[3]][0] == entry[2] and prices[entry[3]][0]/prices[entry[3]][1] < LOWEST_PERCENT_MARGIN)]
+
+    if len(results): # if there's results to print
+        done = default_timer() - START_TIME
+        for result in results:
+            if (int(result[1]) - int(result[0][2]) > LOWEST_PROFIT): #More than 1M Profit
+                send(result[0][2], result[0][1], int(result[1]) - int(result[0][2]), result[0][0], result[1], result[0][2])
+                print("\nLooking for auctions...")
+
+print("Looking for auctions...")
+main()
+
+def dostuff():
+    global now, toppage
+
+    # if 60 seconds have passed since the last update
+    if time.time()*1000 > now + 60000:
+        prevnow = now
+        now = float('inf')
+        c = requests.get("https://api.hypixel.net/skyblock/auctions?page=0").json()
+        if c['lastUpdated'] != prevnow:
+            now = c['lastUpdated']
+            toppage = c['totalPages']
+            main()
+        else:
+            now = prevnow
+    time.sleep(0.25)
+while True:
+    dostuff()
+