60e0706736
- Hourly data exporter (InfluxDB → CSV) for prices, P1, irradiance. - LP-based 24h-foresight oracle dispatch with SoC-consistent state engine. - Reverse-engineered thuisbatterijgids.nl formula (matches their quotes to within €0.50 across three battery configs). - Catalog scraper for the 52 batteries on thuisbatterijgids.net via their /wp-json REST endpoint. - Web app (Flask) that ranks every catalog battery by honest payback and contrasts with the store's quote, deployable via the included Procfile.
79 lines
2.6 KiB
Python
79 lines
2.6 KiB
Python
#!/usr/bin/env python3
|
|
"""Pull thuisbatterijgids.net's full battery catalogue into a CSV.
|
|
|
|
The site exposes a JSON REST endpoint at /wp-json/tbg/v1/batteries — no
|
|
browser, no scraping, just paginated JSON. Output goes to data/raw/.
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import argparse
|
|
import csv
|
|
import json
|
|
import urllib.request
|
|
from pathlib import Path
|
|
|
|
API = "https://thuisbatterijgids.net/wp-json/tbg/v1/batteries"
|
|
HEADERS = {"User-Agent": "Mozilla/5.0 (research; pluginbattery)"}
|
|
|
|
FIELDS = [
|
|
"brand", "title", "capacity_kwh", "power_w",
|
|
"price_eur", "price_per_kwh", "installation", "review_score",
|
|
"recommended", "url", "id",
|
|
]
|
|
|
|
|
|
def fetch_page(page: int, per_page: int = 100) -> dict:
|
|
url = f"{API}?per_page={per_page}&page={page}"
|
|
req = urllib.request.Request(url, headers=HEADERS)
|
|
with urllib.request.urlopen(req, timeout=30) as r:
|
|
return json.load(r)
|
|
|
|
|
|
def main() -> None:
|
|
ap = argparse.ArgumentParser(description=__doc__)
|
|
ap.add_argument("--out", default="data/raw/thuisbatterijgids_catalog.csv",
|
|
help="CSV output path (a parallel .json with all fields is also written)")
|
|
args = ap.parse_args()
|
|
|
|
all_batteries = []
|
|
page = 1
|
|
while True:
|
|
data = fetch_page(page)
|
|
all_batteries.extend(data["batteries"])
|
|
print(f" page {page}/{data['total_pages']}: +{len(data['batteries'])} batteries "
|
|
f"(running total {len(all_batteries)})")
|
|
if page >= data["total_pages"]:
|
|
break
|
|
page += 1
|
|
|
|
csv_path = Path(args.out)
|
|
csv_path.parent.mkdir(parents=True, exist_ok=True)
|
|
with csv_path.open("w", newline="") as f:
|
|
writer = csv.DictWriter(f, fieldnames=FIELDS)
|
|
writer.writeheader()
|
|
for b in all_batteries:
|
|
writer.writerow({
|
|
"brand": b.get("brand", ""),
|
|
"title": b.get("title", ""),
|
|
"capacity_kwh": b.get("capacity"),
|
|
"power_w": b.get("power_continuous"),
|
|
"price_eur": b.get("price"),
|
|
"price_per_kwh": b.get("price_per_kwh"),
|
|
"installation": b.get("installation"),
|
|
"review_score": b.get("review_score"),
|
|
"recommended": b.get("recommended"),
|
|
"url": b.get("url", ""),
|
|
"id": b.get("id"),
|
|
})
|
|
|
|
json_path = csv_path.with_suffix(".json")
|
|
with json_path.open("w") as f:
|
|
json.dump(all_batteries, f, indent=2, ensure_ascii=False)
|
|
|
|
print(f"\nWrote {len(all_batteries)} batteries:")
|
|
print(f" CSV → {csv_path}")
|
|
print(f" JSON → {json_path}")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|