Maintain .pricedb file for ledger3 for gbp/usd to zar

update-pricedb.py

#!/usr/bin/env python3

from html.parser import HTMLParser
import datetime
import itertools
import os.path
import re
import urllib.request

symbols = {
    "GBP": "P",
    "ZAR": "R",
    "USD": "$",
}


def fetch(currency, price_currency, year):
    url = f"https://www.poundsterlinglive.com/history/{currency}-{price_currency}-{year}"
    return urllib.request.urlopen(url).read().decode("utf-8")


class ExchangeHistoryHTMLParser(HTMLParser):
    @staticmethod
    def parse(data, currency, price_currency):
        p = ExchangeHistoryHTMLParser()
        p.feed(data)
        pricedb = {}
        for row in p.rows:
            price = row["Mid Rate"]
            if float(price) == 0:
                continue
            symbol, price_symbol = symbols[currency], symbols[price_currency]
            date = re.search(r'\((\d+/\d+/\d+)\)', row["Date"]).group(1)
            date = datetime.datetime.strptime(date, "%d/%m/%Y").date()
            k = (date, symbol)
            assert k not in pricedb, k
            pricedb[k] = price_symbol + price
        return pricedb

    def __init__(self):
        super().__init__()
        self.rows = []
        self.row = {}
        self.title = ""

    def handle_starttag(self, tag, attrs):
        if tag == "tr":
            self.row = {}
        if tag == "td":
            self.title = dict(attrs).get("data-title")

    def handle_endtag(self, tag):
        if tag == "tr" and self.row:
            self.rows.append(self.row)
        if tag == "td":
            self.title = None

    def handle_data(self, data):
        if self.title:
            self.row[self.title] = self.row.get(self.title, "") + data


def pricedb_unmarshal(data):
    pricedb = {}
    for line in data.splitlines():
        assert line.startswith("P "), line
        _, date, symbol, price = line.split()
        date = datetime.date(*map(int, date.split("/")))
        k = (date, symbol)
        assert k not in pricedb, k
        pricedb[k] = price
    return pricedb


def pricedb_marshal(pricedb):
    lines = []
    for (date, symbol), price in sorted(pricedb.items()):
        lines.append(f"P {date:%Y/%m/%d} {symbol} {price}")
    return "\n".join(lines)


def pricedb_read():
    pricedb_path = os.path.expanduser("~/.pricedb")
    if not os.path.exists(pricedb_path):
        return {}
    with open(pricedb_path) as f:
        return pricedb_unmarshal(f.read())


def pricedb_write(pricedb):
    pricedb_path = os.path.expanduser("~/.pricedb")
    data = pricedb_marshal(pricedb)
    with open(pricedb_path, "w") as f:
        f.write(data)


def main():
    pricedb = pricedb_read()

    price_currency = "ZAR"
    fetch_currencies = ["GBP", "USD"]
    fetch_years = range(2018, datetime.date.today().year + 1)
    for currency, year in itertools.product(fetch_currencies, fetch_years):
        if (datetime.date(year, 12, 31), symbols[currency]) in pricedb:
            print(f"skipping {currency} {year}")
            continue
        print(f"fetching {currency} {year}...")
        page = fetch(currency, price_currency, year)
        page_pricedb = ExchangeHistoryHTMLParser.parse(page, currency, price_currency)
        for k, v in page_pricedb.items():
            (date, symbol), price = k, v
            old_price = pricedb.get(k)

            if old_price is None:
                print(f"P {date:%Y/%m/%d} {symbol} {price}")
            elif price != old_price:
                print(f"CHANGE {date:%Y/%m/%d} {symbol} {old_price} -> {price}")

            pricedb[k] = v

    pricedb_write(pricedb)


if __name__ == "__main__":
    main()