86 lines
2.9 KiB
Python
86 lines
2.9 KiB
Python
"""Geocoder para location_agent.
|
|
|
|
Wrapper sobre el Census Geocoder existente en data_fetchers/census_geocode.py.
|
|
Si falla Census, intenta Nominatim (OpenStreetMap) como fallback gratuito.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import sys
|
|
import time
|
|
from pathlib import Path
|
|
|
|
import requests
|
|
|
|
_PROJECT_ROOT = Path(__file__).resolve().parent.parent.parent
|
|
if str(_PROJECT_ROOT) not in sys.path:
|
|
sys.path.insert(0, str(_PROJECT_ROOT))
|
|
|
|
from data_fetchers.census_geocode import fetch_geocode
|
|
from data_fetchers.base import FetcherError, USER_AGENT, DEFAULT_TIMEOUT
|
|
|
|
NOMINATIM_URL = "https://nominatim.openstreetmap.org/search"
|
|
|
|
|
|
def geocode(address: str) -> dict:
|
|
"""Geocodifica una dirección USA. Census primero, Nominatim como fallback.
|
|
|
|
Returns dict con: address, lat, lon, city, state, zip, county,
|
|
county_fips, state_fips, tract_geoid, source
|
|
Raises ValueError si no se puede geocodificar.
|
|
"""
|
|
# 1. Census Geocoder (preferido — devuelve tract FIPS)
|
|
try:
|
|
result = fetch_geocode(address)
|
|
if result.get("lat") and result.get("lng"):
|
|
return {
|
|
"address": result.get("matched_address", address),
|
|
"lat": float(result["lat"]),
|
|
"lon": float(result["lng"]),
|
|
"city": result.get("city", ""),
|
|
"state": result.get("state", ""),
|
|
"zip": result.get("zip", ""),
|
|
"county": result.get("county_name", ""),
|
|
"county_fips": result.get("county_fips", ""),
|
|
"state_fips": result.get("state_fips", ""),
|
|
"tract_geoid": result.get("tract_geoid", ""),
|
|
"source": "census",
|
|
}
|
|
except (FetcherError, Exception):
|
|
pass
|
|
|
|
# 2. Nominatim fallback
|
|
try:
|
|
params = {
|
|
"q": address,
|
|
"format": "json",
|
|
"addressdetails": 1,
|
|
"limit": 1,
|
|
"countrycodes": "us",
|
|
}
|
|
headers = {"User-Agent": USER_AGENT}
|
|
time.sleep(1)
|
|
r = requests.get(NOMINATIM_URL, params=params, headers=headers, timeout=DEFAULT_TIMEOUT)
|
|
r.raise_for_status()
|
|
results = r.json()
|
|
if results:
|
|
m = results[0]
|
|
addr = m.get("address", {})
|
|
return {
|
|
"address": m.get("display_name", address),
|
|
"lat": float(m["lat"]),
|
|
"lon": float(m["lon"]),
|
|
"city": addr.get("city") or addr.get("town") or addr.get("village", ""),
|
|
"state": addr.get("state", ""),
|
|
"zip": addr.get("postcode", ""),
|
|
"county": addr.get("county", ""),
|
|
"county_fips": "",
|
|
"state_fips": "",
|
|
"tract_geoid": "",
|
|
"source": "nominatim",
|
|
}
|
|
except Exception:
|
|
pass
|
|
|
|
raise ValueError(f"No se pudo geocodificar: {address!r}")
|