116 lines
2.7 KiB
Python
Executable File
116 lines
2.7 KiB
Python
Executable File
#! /usr/bin/env python3
|
|
|
|
from argparse import ArgumentParser
|
|
from config import conf
|
|
from datetime import date
|
|
from models import Price, IntegrityError
|
|
|
|
import requests
|
|
import json
|
|
import webapp
|
|
|
|
|
|
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_price():
|
|
today = date.today()
|
|
# Get price returns a float (price per L)
|
|
# We store an int (price per m^3)
|
|
price = int(get_today_price_from_internet() * 1000)
|
|
try:
|
|
Price.create(date=today, price=price)
|
|
except IntegrityError:
|
|
print("Price for today already in database, not writing")
|
|
return price
|
|
|
|
|
|
# ----------------------------------------------------------------------
|
|
# 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_price()
|
|
|
|
|
|
@subcommand([argument("n", default=42, nargs="?", type=int)])
|
|
def show(args):
|
|
for x in get_prices(args.n):
|
|
print(x.date, x.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()
|
|
|
|
|
|
# ----------------------------------------------------------------------
|
|
# Main
|
|
|
|
|
|
def main():
|
|
args = cli.parse_args()
|
|
if args.subcommand is None:
|
|
cli.print_help()
|
|
else:
|
|
args.func(args)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|