fuel_data.py

"""Fetch and normalise French fuel-price data from the open-data portal.

Data source: https://public.opendatasoft.com/explore/dataset/prix_des_carburants_j_7
The dataset holds the prices reported by every petrol station in France over the
last 7 days, refreshed continuously upstream.
"""
from __future__ import annotations

import json
import os
from datetime import datetime, timezone

import requests

DATASET = "prix_des_carburants_j_7"
EXPORT_URL = (
    "https://public.opendatasoft.com/api/explore/v2.1/"
    f"catalog/datasets/{DATASET}/exports/json"
)

# Human label -> price field in the dataset (ordered: most common fuel first).
FUELS: dict[str, str] = {
    "Gazole": "price_gazole",
    "SP95": "price_sp95",
    "SP98": "price_sp98",
    "E10": "price_e10",
    "E85": "price_e85",
    "GPLc": "price_gplc",
}

# Plausible price band (€/L). Real road fuels in France sit well inside this;
# anything outside is a data-entry error and is dropped.
PRICE_MIN, PRICE_MAX = 0.4, 4.0


def fetch_raw(timeout: int = 180) -> list[dict]:
    """Download every record from the dataset's bulk export endpoint."""
    resp = requests.get(EXPORT_URL, params={"timezone": "Europe/Paris"}, timeout=timeout)
    resp.raise_for_status()
    return resp.json()


def normalise(raw: list[dict]) -> list[dict]:
    """Keep geolocated stations that report at least one price; tidy the fields."""
    stations = []
    for r in raw:
        geo = r.get("geo_point") or {}
        lat, lon = geo.get("lat"), geo.get("lon")
        if lat is None or lon is None:
            continue

        # Some stations report nonsense prices (e.g. 0.002 €/L); keep only
        # plausible values so they don't skew the cheapest/cost figures.
        prices = {
            label: (v if isinstance(v, (int, float)) and PRICE_MIN <= v <= PRICE_MAX else None)
            for label, field in FUELS.items()
            for v in (r.get(field),)
        }
        if not any(v is not None for v in prices.values()):
            continue

        stations.append(
            {
                "id": r.get("id"),
                "name": (r.get("name") or "").strip() or "Station",
                "brand": (r.get("brand") or "").strip(),
                "address": (r.get("address") or "").strip(),
                "city": (r.get("city") or "").strip(),
                "cp": (r.get("cp") or "").strip(),
                "lat": lat,
                "lon": lon,
                "prices": prices,
                "update": r.get("update"),
                "highway": r.get("pop") == "A",  # A = autoroute, R = route
                "open24": is_24h(r.get("automate_24_24")),
                "services": clean_services(r.get("services")),
            }
        )
    return stations


def is_24h(value) -> bool:
    """Whether the station has 24/7 automated payment (field is bool or Oui/Non)."""
    if value is True:
        return True
    return isinstance(value, str) and value.strip().lower() in ("oui", "yes", "true", "1")


def clean_services(value) -> list[str]:
    """Normalise the `services` field (list, or {'service': [...]}) to a string list."""
    if isinstance(value, dict):
        value = value.get("service")
    if not isinstance(value, list):
        return []
    return [str(s).strip() for s in value if str(s).strip()]


def fetch_stations(cache_path: str | None = None) -> list[dict]:
    """Fetch + normalise the live dataset, optionally caching the result to disk."""
    stations = normalise(fetch_raw())
    if cache_path:
        os.makedirs(os.path.dirname(cache_path) or ".", exist_ok=True)
        payload = {
            "fetched_at": datetime.now(timezone.utc).isoformat(),
            "count": len(stations),
            "stations": stations,
        }
        with open(cache_path, "w", encoding="utf-8") as fh:
            json.dump(payload, fh, ensure_ascii=False)
    return stations


if __name__ == "__main__":
    data = fetch_stations()
    print(f"Fetched {len(data)} geolocated stations.")
    sample = data[0]
    print("Example:", json.dumps(sample, ensure_ascii=False, indent=2))