如何使用地理编码API批量验证10,000个地址

pip install requests pandas tqdm

#!/usr/bin/env python3
"""
bulk_geocode.py, Validate a CSV of addresses using the MapAtlas Geocoding API.

Input CSV must have either:
  - An 'address' column (full address string), or
  - 'street', 'city', and 'country' columns (will be concatenated)

Outputs a new CSV with added columns:
  geocoded_label, geocoded_lat, geocoded_lng, confidence, match_type, status
"""

import csv
import time
import logging
import requests
import pandas as pd
from tqdm import tqdm
from pathlib import Path

# ── Configuration ──────────────────────────────────────────────────────────────
API_KEY         = 'YOUR_API_KEY'
API_BASE        = 'https://api.mapatlas.eu/geocoding/v1/search'
INPUT_CSV       = 'addresses.csv'
OUTPUT_CSV      = 'addresses_validated.csv'
RATE_LIMIT_RPS  = 5          # Requests per second (stay within your plan limits)
RETRY_ATTEMPTS  = 3          # Retries on transient errors
RETRY_DELAY_S   = 2.0        # Seconds between retries

# Confidence thresholds
ACCEPT_THRESHOLD = 0.85
REVIEW_THRESHOLD = 0.60

logging.basicConfig(level=logging.INFO, format='%(levelname)s %(message)s')
log = logging.getLogger(__name__)

# ── Geocoding function ─────────────────────────────────────────────────────────
def geocode_address(address_str: str) -> dict:
    """
    Geocode a single address string. Returns a dict with result fields.
    Retries on network errors and 429 rate-limit responses.
    """
    params = {
        'text': address_str,
        'key':  API_KEY,
        'size': 1,
    }

    for attempt in range(RETRY_ATTEMPTS):
        try:
            resp = requests.get(API_BASE, params=params, timeout=10)

            if resp.status_code == 429:
                # Rate limited, wait and retry
                wait = float(resp.headers.get('Retry-After', RETRY_DELAY_S * (attempt + 1)))
                log.warning(f'Rate limited. Waiting {wait:.1f}s before retry {attempt + 1}.')
                time.sleep(wait)
                continue

            resp.raise_for_status()
            data = resp.json()

            if not data.get('features'):
                return {'geocoded_label': '', 'geocoded_lat': None, 'geocoded_lng': None,
                        'confidence': 0.0, 'match_type': 'no_match', 'status': 'reject'}

            feature    = data['features'][0]
            props      = feature['properties']
            coords     = feature['geometry']['coordinates']  # [lng, lat]
            confidence = round(float(props.get('confidence', 0.0)), 4)
            match_type = props.get('match_type', 'unknown')

            if confidence >= ACCEPT_THRESHOLD:
                status = 'accept'
            elif confidence >= REVIEW_THRESHOLD:
                status = 'review'
            else:
                status = 'reject'

            return {
                'geocoded_label': props.get('label', ''),
                'geocoded_lat':   round(coords[1], 6),
                'geocoded_lng':   round(coords[0], 6),
                'confidence':     confidence,
                'match_type':     match_type,
                'status':         status,
            }

        except requests.exceptions.RequestException as exc:
            log.warning(f'Network error on attempt {attempt + 1}: {exc}')
            if attempt < RETRY_ATTEMPTS - 1:
                time.sleep(RETRY_DELAY_S * (attempt + 1))

    # All retries exhausted
    return {'geocoded_label': '', 'geocoded_lat': None, 'geocoded_lng': None,
            'confidence': 0.0, 'match_type': 'error', 'status': 'error'}


# ── Main processing loop ───────────────────────────────────────────────────────
def main():
    df = pd.read_csv(INPUT_CSV, dtype=str).fillna('')

    # Build address string from available columns
    if 'address' in df.columns:
        df['_query'] = df['address']
    elif all(c in df.columns for c in ['street', 'city', 'country']):
        df['_query'] = df['street'] + ', ' + df['city'] + ', ' + df['country']
    else:
        raise ValueError("CSV must have 'address' or 'street'+'city'+'country' columns.")

    results = []
    sleep_interval = 1.0 / RATE_LIMIT_RPS

    for query in tqdm(df['_query'], desc='Geocoding', unit='addr'):
        result = geocode_address(query.strip())
        results.append(result)
        time.sleep(sleep_interval)

    results_df = pd.DataFrame(results)
    output_df  = pd.concat([df.drop(columns=['_query']), results_df], axis=1)
    output_df.to_csv(OUTPUT_CSV, index=False, quoting=csv.QUOTE_NONNUMERIC)

    # Summary
    total   = len(output_df)
    accept  = (output_df['status'] == 'accept').sum()
    review  = (output_df['status'] == 'review').sum()
    reject  = (output_df['status'] == 'reject').sum()
    errors  = (output_df['status'] == 'error').sum()

    log.info(f'\n── Results ────────────────────────────────')
    log.info(f'Total processed : {total:,}')
    log.info(f'Accept (≥{ACCEPT_THRESHOLD}) : {accept:,}  ({accept/total:.1%})')
    log.info(f'Review           : {review:,}  ({review/total:.1%})')
    log.info(f'Reject           : {reject:,}  ({reject/total:.1%})')
    log.info(f'Errors           : {errors:,}  ({errors/total:.1%})')
    log.info(f'Output written to {OUTPUT_CSV}')


if __name__ == '__main__':
    main()

python bulk_geocode.py

address, ..., geocoded_label, geocoded_lat, geocoded_lng, confidence, match_type, status

# After running main(), split into acceptance tiers:
df = pd.read_csv('addresses_validated.csv')

df[df['status'] == 'accept'].to_csv('addresses_clean.csv',  index=False)
df[df['status'] == 'review'].to_csv('addresses_review.csv', index=False)
df[df['status'] == 'reject'].to_csv('addresses_reject.csv', index=False)

print(f"Clean:  {len(df[df['status']=='accept']):,} addresses ready for use")
print(f"Review: {len(df[df['status']=='review']):,} addresses for manual check")
print(f"Reject: {len(df[df['status']=='reject']):,} addresses to discard or fix")

from concurrent.futures import ThreadPoolExecutor, as_completed

def main_concurrent(max_workers=10):
    df = pd.read_csv(INPUT_CSV, dtype=str).fillna('')
    # ... (same setup as before) ...

    results = [None] * len(df)

    with ThreadPoolExecutor(max_workers=max_workers) as executor:
        future_to_idx = {
            executor.submit(geocode_address, row['_query'].strip()): idx
            for idx, row in df.iterrows()
        }
        for future in tqdm(as_completed(future_to_idx), total=len(df), desc='Geocoding'):
            idx = future_to_idx[future]
            results[idx] = future.result()

    # ... (same output writing as before) ...