GenerateMapboxMarkers/generate_jetpack_map.py

#!/usr/bin/env python3

import argparse
import csv
import json
import os
import statistics
import sys
import time
from pathlib import Path
from urllib.parse import quote

import requests


MAPBOX_GEOCODING_URL = (
    "https://api.mapbox.com/geocoding/v5/mapbox.places"
)


def geocode_address(
    address,
    token,
    country=None,
    language="fr",
    permanent=True,
):
    """
    Return the best Mapbox result for an address.
    """

    url = f"{MAPBOX_GEOCODING_URL}/{quote(address)}.json"

    params = {
        "access_token": token,
        "limit": 1,
        "language": language,
        "autocomplete": "false",
    }

    if country:
        params["country"] = country

    # Use permanent=true only when your Mapbox account permits
    # permanent storage of geocoding results.
    if permanent:
        params["permanent"] = "true"

    response = requests.get(url, params=params, timeout=30)
    response.raise_for_status()

    data = response.json()
    features = data.get("features", [])

    if not features:
        return None

    feature = features[0]
    coordinates = feature.get("center")

    if not coordinates:
        geometry = feature.get("geometry", {})
        coordinates = geometry.get("coordinates")

    if not coordinates or len(coordinates) < 2:
        return None

    longitude = float(coordinates[0])
    latitude = float(coordinates[1])

    return {
        "placeTitle": feature.get("place_name", address),
        "caption": feature.get("place_name", address),
        "id": feature.get("id", ""),
        "longitude": longitude,
        "latitude": latitude,
    }


def read_csv_rows(csv_path):
    """
    Read a CSV with columns:
        name,address
    """

    with open(csv_path, "r", encoding="utf-8-sig", newline="") as file:
        reader = csv.DictReader(file)

        if not reader.fieldnames:
            raise ValueError("The CSV file has no header row.")

        required_columns = {"name", "address"}
        actual_columns = set(reader.fieldnames)

        missing = required_columns - actual_columns
        if missing:
            raise ValueError(
                f"Missing CSV column(s): {', '.join(sorted(missing))}. "
                "Expected columns: name,address"
            )

        rows = []

        for line_number, row in enumerate(reader, start=2):
            name = (row.get("name") or "").strip()
            address = (row.get("address") or "").strip()

            if not name and not address:
                continue

            if not name or not address:
                raise ValueError(
                    f"Line {line_number}: both name and address are required."
                )

            rows.append({
                "name": name,
                "address": address,
            })

        return rows


def create_jetpack_points(rows, token, country=None, language="fr"):
    points = []
    failed = []

    total = len(rows)

    for index, row in enumerate(rows, start=1):
        print(
            f"Geocoding {index}/{total}: "
            f"{row['address']}",
            file=sys.stderr,
        )

        try:
            result = geocode_address(
                address=row["address"],
                token=token,
                country=country,
                language=language,
                permanent=True,
            )
        except requests.RequestException as error:
            print(
                f"  Request failed: {error}",
                file=sys.stderr,
            )
            result = None

        if result is None:
            failed.append(row)
            print("  No geocoding result found.", file=sys.stderr)
            continue

        point = {
            "placeTitle": result["placeTitle"],
            "title": row["name"],
            "caption": result["caption"],
            "id": result["id"],
            "coordinates": {
                "longitude": result["longitude"],
                "latitude": result["latitude"],
            },
        }

        points.append(point)

        # A short delay avoids sending requests too quickly.
        time.sleep(0.1)

    return points, failed


def calculate_map_center(points):
    longitudes = [
        point["coordinates"]["longitude"]
        for point in points
    ]

    latitudes = [
        point["coordinates"]["latitude"]
        for point in points
    ]

    return {
        "lng": statistics.mean(longitudes),
        "lat": statistics.mean(latitudes),
    }


def generate_jetpack_block(
    points,
    zoom=12.125807719593206,
    marker_color="#B50B3E",
):
    if not points:
        raise ValueError("No valid geocoded points were found.")

    map_center = calculate_map_center(points)

    block_attributes = {
        "points": points,
        "zoom": zoom,
        "mapCenter": map_center,
        "markerColor": marker_color,
        "scrollToZoom": True,
    }

    # separators=(",", ":") produces compact JSON.
    attributes_json = json.dumps(
        block_attributes,
        ensure_ascii=False,
        separators=(",", ":"),
    )

    return f"<!-- wp:jetpack/map {attributes_json} -->"


def main():
    parser = argparse.ArgumentParser(
        description="Generate a Jetpack map block from an address CSV."
    )

    parser.add_argument(
        "csv_file",
        help="CSV file containing name,address columns",
    )

    parser.add_argument(
        "-o",
        "--output",
        default="jetpack-map.txt",
        help="Output file; default: jetpack-map.txt",
    )

    parser.add_argument(
        "--token",
        default=os.environ.get("MAPBOX_TOKEN"),
        help="Mapbox access token, or use MAPBOX_TOKEN environment variable",
    )

    parser.add_argument(
        "--country",
        default="fr",
        help="Two-letter country code used to bias results; default: fr",
    )

    parser.add_argument(
        "--language",
        default="fr",
        help="Language for returned place names; default: fr",
    )

    parser.add_argument(
        "--zoom",
        type=float,
        default=12.125807719593206,
        help="Map zoom level",
    )

    parser.add_argument(
        "--marker-color",
        default="#B50B3E",
        help="Jetpack marker color",
    )

    args = parser.parse_args()

    if not args.token:
        parser.error(
            "A Mapbox token is required. Use --token or set MAPBOX_TOKEN."
        )

    rows = read_csv_rows(args.csv_file)

    points, failed = create_jetpack_points(
        rows=rows,
        token=args.token,
        country=args.country,
        language=args.language,
    )

    block = generate_jetpack_block(
        points=points,
        zoom=args.zoom,
        marker_color=args.marker_color,
    )

    output_path = Path(args.output)
    output_path.write_text(block + "\n", encoding="utf-8")

    print(
        f"\nGenerated {len(points)} map point(s) in {output_path}",
        file=sys.stderr,
    )

    if failed:
        failed_path = output_path.with_name(
            output_path.stem + "-failed.csv"
        )

        with open(
            failed_path,
            "w",
            encoding="utf-8",
            newline="",
        ) as file:
            writer = csv.DictWriter(
                file,
                fieldnames=["name", "address"],
            )
            writer.writeheader()
            writer.writerows(failed)

        print(
            f"{len(failed)} address(es) could not be geocoded.",
            file=sys.stderr,
        )
        print(
            f"They were written to {failed_path}",
            file=sys.stderr,
        )


if __name__ == "__main__":
    main()