163 lines
4.0 KiB
Python
Executable File
163 lines
4.0 KiB
Python
Executable File
#! /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
|
|
}
|
|
}
|
|
}
|
|
"""
|
|
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), None)
|
|
price_1000 = next((x["price"] for x in zones if x["min_quantity"] == 1000), 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()
|