A parser and Zephyr nottification bot for park.io's auctions

check_parkio.py

#!/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)