Initial version
This commit is contained in:
@@ -0,0 +1,329 @@
|
||||
#!/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()
|
||||
|
||||
Reference in New Issue
Block a user