Created
October 14, 2016 16:18
-
-
Save vulpicastor/076212631e619ae3a6378e0c9e768344 to your computer and use it in GitHub Desktop.
A parser and Zephyr nottification bot for park.io's auctions
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
#!/usr/bin/env python3 | |
import requests | |
import time | |
from datetime import datetime | |
import pytz | |
from zwrite import zwrite | |
PARKIO_DOMAIN = "https://app.park.io" | |
AUCTION_PATH = "/auctions.json" | |
AUCTION_KEYS_DATETIME = "created", "close_date" | |
AUCTION_KEYS_INT = "num_bids", "price" | |
AUCTION_KEYS = ("name",) + AUCTION_KEYS_INT + AUCTION_KEYS_DATETIME | |
AUCTION_PRETTY_FORMAT = """ | |
bids: {num_bids} | |
price: {price} | |
created: {created:%F %R %Z} | |
close: {close_date:%F %R %Z} | |
""" | |
AUCTION_STR_FORMAT = "{name:<16}{num_bids:4}{price:6} {created:%F %R} {close_date:%F %R}" | |
AUCTION_STR_HEADER = "{0:<16}{1:4}{2:>6} {3:<20}{4:<16}".format("domain", "bids", "price", "created (UTC)", "close date (UTC)") | |
LIST_STR_FORMAT = "{aid:>6} {auction!s}" | |
LIST_STR_HEADER = "{0:<6} {1}".format("id", AUCTION_STR_HEADER) | |
PARKIO_TIMEZONE = pytz.timezone('US/Eastern') | |
TIMEOUT = 10 | |
UPDATE_INTERVAL = 15 * 60 | |
DEFAULT_ZWRITE_ARGS = {"classname": "parkio-auto", | |
"sender": "parkio_bot", | |
"zsig": "park.io auction monitor bot", | |
"opcode": "auto"} | |
class ParkIoAuction(dict): | |
def __init__(self, auction_dict): | |
self["name"] = auction_dict["name"] | |
for k in AUCTION_KEYS_INT: | |
self[k] = int(auction_dict[k]) | |
for k in AUCTION_KEYS_DATETIME: | |
self[k] = self.strptime_parkio(auction_dict[k]) | |
def __repr__(self): | |
repr_dict = {"name": self["name"]} | |
for k in AUCTION_KEYS_INT: | |
repr_dict[k] = str(self[k]) | |
for k in AUCTION_KEYS_DATETIME: | |
# For park.io API compatibility, force into US/Eastern but claims UTC | |
repr_dict[k] = self[k].astimezone(PARKIO_TIMEZONE).strftime("%Y-%d-%mUTC%H:%M:%S%f")[:-3] | |
return "{0}({1!r})".format(self.__class__.__name__, repr_dict) | |
def __str__(self): | |
return AUCTION_STR_FORMAT.format(**self) | |
def update(self, auction_dict): | |
if self["name"] != auction_dict["name"]: | |
raise ValueError("Auction name mismatch") | |
updated = False | |
for k in AUCTION_KEYS_INT: | |
if self[k] != int(auction_dict[k]): | |
self[k] = int(auction_dict[k]) | |
updated = True | |
for k in AUCTION_KEYS_DATETIME: | |
if self[k] != self.strptime_parkio(auction_dict[k]): | |
self[k] = self.strptime_parkio(auction_dict[k]) | |
updated = True | |
return updated | |
@classmethod | |
def strptime_parkio(cls, date_string): | |
naive_datetime = datetime.strptime(date_string + "000", "%Y-%d-%m%Z%H:%M:%S%f") | |
# park.io timestamps claim UTC. Reality begs to defer. | |
return PARKIO_TIMEZONE.localize(naive_datetime).astimezone(pytz.utc) | |
class ParkIoAuctionList(dict): | |
def __init__(self, auction_url=PARKIO_DOMAIN+AUCTION_PATH): | |
self.url = auction_url | |
r = requests.get(self.url, timeout=TIMEOUT) | |
r.raise_for_status() | |
self.deleted = {} | |
for a in r.json()["auctions"]: | |
self[a["id"]] = ParkIoAuction(a) | |
def __str__(self): | |
return self.str_help(self.keys()) | |
def str_help(self, keys): | |
lines = [LIST_STR_FORMAT.format(aid=k, auction=self[k]) for k in keys] | |
lines.sort() | |
return "\n".join(lines) | |
def update(self): | |
r = requests.get(self.url, timeout=TIMEOUT) | |
r.raise_for_status() | |
updated = set() | |
exist = set() | |
self.deleted = {} | |
for a in r.json()["auctions"]: | |
aid = a["id"] | |
exist.add(aid) | |
if aid in self: | |
if self[aid].update(a): | |
updated.add(aid) | |
# Handle new auctions | |
else: | |
self[aid] = ParkIoAuction(a) | |
updated.add(aid) | |
# Clean up deleted auctions | |
for k in set(self.keys()): | |
if k not in exist: | |
self.deleted[k] = self[k] | |
del self[k] | |
return updated | |
def zwrite_auction(auction): | |
return zwrite(AUCTION_PRETTY_FORMAT.format(**auction), instance=auction["name"], **DEFAULT_ZWRITE_ARGS) | |
def zwrite_auction_list(auction_list, auction_keys=None): | |
if auction_keys == None: | |
auctions = str(auction_list) | |
else: | |
auctions = auction_list.str_help(auction_keys) | |
return zwrite(LIST_STR_HEADER + "\n" + auctions, instance="auctions", **DEFAULT_ZWRITE_ARGS) | |
def log_spew(spew): | |
print(spew) | |
return zwrite(spew, instance="spew", **DEFAULT_ZWRITE_ARGS) | |
def update_auctions(auction_list): | |
updated = auction_list.update() | |
if updated: | |
zwrite_auction_list(auction_list, updated) | |
for k in updated: | |
zwrite_auction(auction_list[k]) | |
if auction_list.deleted: | |
for k in auction_list.deleted: | |
zwrite("auction {} has disappeared.".format(k), instance=auction_list.deleted[k]["name"], **DEFAULT_ZWRITE_ARGS) | |
if __name__ == "__main__": | |
auction_list = ParkIoAuctionList() | |
zwrite_auction_list(auction_list) | |
for k in auction_list: | |
zwrite_auction(auction_list[k]) | |
while True: | |
try: | |
update_auctions(auction_list) | |
except (requests.exceptions.ConnectionError, | |
requests.exceptions.HTTPError, | |
requests.exceptions.Timeout) as err: | |
log_spew(str(err)) | |
time.sleep(UPDATE_INTERVAL) |
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment