"""Build an interactive fuel-price map of France as a self-contained HTML file. Rather than emitting one marker object per station per fuel (which balloons the file to tens of MB), the station data is embedded once as compact JSON and the map is rendered client-side with Leaflet + marker clustering. A fuel selector recolours and filters the points live: green = cheaper, red = pricier, scaled to each fuel's national price distribution. """ from __future__ import annotations import json import math import os import shutil from datetime import datetime from zoneinfo import ZoneInfo from car_data import fetch_cars from fuel_data import FUELS, fetch_stations BASE_DIR = os.path.dirname(os.path.abspath(__file__)) OUTPUT_DIR = os.path.join(BASE_DIR, "output") STATIC_DIR = os.path.join(BASE_DIR, "static") OUTPUT_HTML = os.path.join(OUTPUT_DIR, "index.html") # served at "/" (PWA start_url) CACHE_JSON = os.path.join(OUTPUT_DIR, "stations.json") HISTORY_JSON = os.path.join(OUTPUT_DIR, "price_history.json") # national medians over time PARIS = ZoneInfo("Europe/Paris") SITE_URL = "https://samdnx-prix-carburant.hf.space/" # canonical URL (sitemap / SEO) FUEL_LABELS = list(FUELS) TEMPLATE_PATH = os.path.join(BASE_DIR, "map_template.html") def percentile(sorted_vals: list[float], pct: float) -> float: """Linear-interpolation percentile of an already-sorted list.""" if not sorted_vals: return 0.0 k = (len(sorted_vals) - 1) * pct / 100.0 lo, hi = math.floor(k), math.ceil(k) if lo == hi: return sorted_vals[int(k)] return sorted_vals[lo] * (hi - k) + sorted_vals[hi] * (k - lo) def fuel_stats(stations: list[dict]) -> dict[str, dict]: """Per-fuel count and 5th/50th/95th percentiles used for colour scaling.""" stats = {} for label in FUEL_LABELS: values = sorted( s["prices"][label] for s in stations if s["prices"][label] is not None ) if not values: continue stats[label] = { "count": len(values), "p5": round(percentile(values, 5), 3), "median": round(percentile(values, 50), 3), "p95": round(percentile(values, 95), 3), } return stats def compute_trends(stats: dict[str, dict]) -> dict[str, dict]: """National median per fuel, with the change vs the previous build. A small history of medians is kept in output/price_history.json so the map can show whether each fuel is trending up or down. On hosts with ephemeral storage the history resets on a cold restart (deltas then start at 0). """ today = datetime.now(PARIS).strftime("%Y-%m-%d") medians = {f: stats[f]["median"] for f in stats} history: list[dict] = [] if os.path.exists(HISTORY_JSON): try: with open(HISTORY_JSON, encoding="utf-8") as fh: history = json.load(fh) except Exception: history = [] prev = history[-1]["medians"] if history else {} trends = {f: {"median": m, "delta": round(m - prev[f], 3) if f in prev else 0.0} for f, m in medians.items()} if history and history[-1].get("date") == today: history[-1]["medians"] = medians # same day: keep one entry else: history.append({"date": today, "medians": medians}) history = history[-30:] try: with open(HISTORY_JSON, "w", encoding="utf-8") as fh: json.dump(history, fh) except Exception: pass return trends def compact_records(stations: list[dict]) -> list[list]: """Encode each station as a small positional array to shrink the payload.""" records = [] for s in stations: addr_parts = [s["address"], f"{s['cp']} {s['city']}".strip()] addr = ", ".join(p for p in addr_parts if p) prices = [s["prices"][label] for label in FUEL_LABELS] records.append( [ round(s["lat"], 5), round(s["lon"], 5), s["name"], s["brand"], addr, (s["update"] or "")[:10], prices, s["cp"], s["city"], 1 if s["highway"] else 0, 1 if s.get("open24") else 0, s.get("services") or [], ] ) return records def build_html(stations: list[dict]) -> str: """Render the HTML document from the template and embedded data.""" with open(TEMPLATE_PATH, encoding="utf-8") as fh: template = fh.read() try: cars = fetch_cars() except Exception as exc: # car list is optional; never block the map build print(f" (car list unavailable: {exc})") cars = [] stats = fuel_stats(stations) payload = { "fuels": FUEL_LABELS, "stats": stats, "stations": compact_records(stations), "cars": cars, "trends": compute_trends(stats), } data_json = json.dumps(payload, ensure_ascii=False, separators=(",", ":")) months_fr = ["janv.", "févr.", "mars", "avr.", "mai", "juin", "juil.", "août", "sept.", "oct.", "nov.", "déc."] now = datetime.now(PARIS) stamp = f"{now.day} {months_fr[now.month - 1]} {now.year}, {now:%H:%M}" return ( template.replace("__DATA__", data_json) .replace("__COUNT__", f"{len(stations):,}".replace(",", " ")) .replace("__TIMESTAMP__", stamp) ) def copy_static() -> None: """Copy the PWA assets (manifest, service worker, icons) into output/.""" if not os.path.isdir(STATIC_DIR): return for name in os.listdir(STATIC_DIR): src = os.path.join(STATIC_DIR, name) if os.path.isfile(src): shutil.copy2(src, os.path.join(OUTPUT_DIR, name)) def write_sitemap(lastmod: str) -> None: """Write sitemap.xml with the build date as (freshness signal for crawlers).""" xml = ( '\n' '\n' ' \n' f' {SITE_URL}\n' f' {lastmod}\n' ' daily\n' ' 1.0\n' ' \n' '\n' ) with open(os.path.join(OUTPUT_DIR, "sitemap.xml"), "w", encoding="utf-8") as fh: fh.write(xml) def generate(output_html: str = OUTPUT_HTML) -> str: """Fetch live data, build the map, write the HTML file, return its path.""" os.makedirs(OUTPUT_DIR, exist_ok=True) copy_static() write_sitemap(datetime.now(PARIS).strftime("%Y-%m-%d")) try: # social-share card (needs Pillow + a system font); never block the build import generate_og generate_og.save(os.path.join(OUTPUT_DIR, "og-card.png")) except Exception as exc: print(f" (social card unavailable: {exc})") stations = fetch_stations(cache_path=CACHE_JSON) html_doc = build_html(stations) with open(output_html, "w", encoding="utf-8") as fh: fh.write(html_doc) size_mb = os.path.getsize(output_html) / 1e6 print( f"[{datetime.now(PARIS):%Y-%m-%d %H:%M:%S}] " f"Map generated: {output_html} ({len(stations)} stations, {size_mb:.1f} MB)" ) return output_html if __name__ == "__main__": generate()