fioul/fioul.py

#! /usr/bin/env python3

import json
from argparse import ArgumentParser
from datetime import date

import requests

import webapp
from config import conf
from models import IntegrityError, Price


def get_today_prices_from_internet():
    query = """query {
      getProducts(group: 1, zipcode: 1184) {
        tax {
          value
        }
        pricesZone {
          price
          min_quantity
          max_quantity
        }
      }
    }
    """
    response = requests.post(
        url="https://www.bretagne-multi-energies.fr/graphql",
        json={"query": query},
        headers={"content-type": "application/json"},
    )
    json_data = json.loads(response.text)["data"]
    json_data = json_data["getProducts"][0]
    tax = float(json_data["tax"]["value"])
    zones = json_data["pricesZone"]
    price_500 = next(
        (x["price"] for x in zones if x["min_quantity"] <= 500 < x["max_quantity"]),
        None,
    )
    price_1000 = next(
        (x["price"] for x in zones if x["min_quantity"] <= 1000 < x["max_quantity"]),
        None,
    )
    # apply tax
    price_500 = price_500 * (100 + tax) / 100.0
    price_1000 = price_1000 * (100 + tax) / 100.0
    # Return an int, for db storage
    price_500 = int(price_500 * 1000)
    price_1000 = int(price_1000 * 1000)
    return price_500, price_1000


def get_today_price_from_internet():
    query = """query {
        getProductGroups(category: 1, zipcode: 1184) {
            name
            default_price_tax
        }
    }
    """
    response = requests.post(
        url="https://www.bretagne-multi-energies.fr/graphql",
        json={"query": query},
        headers={"content-type": "application/json"},
    )
    json_data = json.loads(response.text)
    products = json_data["data"]["getProductGroups"]
    product = next(
        filter(lambda x: x["name"] == "Fioul Domestique Standard", products), None
    )
    return product["default_price_tax"]


def get_prices(n=10):
    query = Price.select().order_by(Price.date.desc()).limit(n)
    return query


def store_current_prices():
    today = date.today()
    # Get price returns a float (price per L)
    # We store an int (price per m^3)
    price500, price1000 = get_today_prices_from_internet()
    try:
        Price.create(date=today, price=price1000, price500=price500)
    except IntegrityError:
        print("Price for today already in database, not writing")
    return price500, price1000


# ----------------------------------------------------------------------
# Argument parsing
# use a decorator to simplify argparse usage, as suggested by
# https://mike.depalatis.net/blog/simplifying-argparse.html

cli = ArgumentParser(description="Balises")
subparsers = cli.add_subparsers(dest="subcommand")


def subcommand(args=[], parent=subparsers):
    def decorator(func):
        parser = parent.add_parser(func.__name__, description=func.__doc__)
        for arg in args:
            parser.add_argument(*arg[0], **arg[1])
        parser.set_defaults(func=func)

    return decorator


def argument(*name_or_flags, **kwargs):
    return ([*name_or_flags], kwargs)


@subcommand()
def update(args):
    store_current_prices()


@subcommand([argument("n", default=42, nargs="?", type=int)])
def show(args):
    for x in get_prices(args.n):
        print(x.date, x.price, x.price500)


@subcommand()
def show_current(args):
    current_price = get_today_prices_from_internet()
    print(current_price)


@subcommand()
def serve(args):
    webapp.run(
        host=conf["server"].get("host", "localhost"),
        port=conf["server"].getint("port", 9980),
        debug=conf["server"].getboolean("debug", False),
        reloader=conf["server"].getboolean("reloader", False),
    )


@subcommand()
def initdb(args):
    from models import init

    init()


@subcommand()
def migratedb(args):
    from models import migrate_001

    migrate_001()


# ----------------------------------------------------------------------
# Main


def main():
    args = cli.parse_args()
    if args.subcommand is None:
        cli.print_help()
    else:
        args.func(args)


if __name__ == "__main__":
    main()