build_map.py

"""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 <lastmod> (freshness signal for crawlers)."""
    xml = (
        '<?xml version="1.0" encoding="UTF-8"?>\n'
        '<urlset xmlns="http://www.sitemaps.org/schemas/sitemap/0.9">\n'
        '  <url>\n'
        f'    <loc>{SITE_URL}</loc>\n'
        f'    <lastmod>{lastmod}</lastmod>\n'
        '    <changefreq>daily</changefreq>\n'
        '    <priority>1.0</priority>\n'
        '  </url>\n'
        '</urlset>\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()