Python scripts voor bulk adresverwerking

pip install pandas aiohttp python-dotenv tqdm

import asyncio
import aiohttp
from typing import List, Dict, Optional
from urllib.parse import quote
import os
import time

class AsyncGeocoder:
    def __init__(self, api_key: str, concurrency: int = 10):
        self.api_key = api_key
        self.semaphore = asyncio.Semaphore(concurrency)
        self.token: Optional[str] = None
        self.token_expiry: float = 0

    async def get_token(self, session: aiohttp.ClientSession) -> str:
        """Haal een nieuw token op of gebruik cached token"""
        # Vernieuw token als bijna verlopen (binnen 1 minuut)
        if not self.token or time.time() > self.token_expiry - 60:
            async with session.post(
                "https://api.georex.nl/api/token",
                headers={"X-API-Key": self.api_key}
            ) as response:
                data = await response.json()
                self.token = data["token"]
                self.token_expiry = time.time() + data["expires_in"]
        return self.token

    async def geocode_one(self, session: aiohttp.ClientSession, address: str) -> Dict:
        async with self.semaphore:
            token = await self.get_token(session)
            query = quote(address)

            async with session.get(
                f"https://api.georex.nl/geocode?q={query}&token={token}"
            ) as response:
                data = await response.json()

                # GeoJSON response verwerken
                if data.get("features"):
                    feature = data["features"][0]
                    lon, lat = feature["geometry"]["coordinates"]
                    return {
                        "address": address,
                        "found": True,
                        "lat": lat,
                        "lon": lon,
                        **feature["properties"]
                    }
                return {"address": address, "found": False}

    async def geocode_batch(self, addresses: List[str]) -> List[Dict]:
        async with aiohttp.ClientSession() as session:
            tasks = [
                self.geocode_one(session, addr)
                for addr in addresses
            ]
            return await asyncio.gather(*tasks)

# Gebruik
async def main():
    geocoder = AsyncGeocoder(os.getenv("GEOREX_API_KEY"))
    addresses = ["Damrak 1, Amsterdam", "Coolsingel 40, Rotterdam"]
    results = await geocoder.geocode_batch(addresses)
    for r in results:
        if r["found"]:
            print(f"{r['address']} -> {r['lat']}, {r['lon']}")

asyncio.run(main())