135 lines
4.4 KiB
Python
135 lines
4.4 KiB
Python
"""HUD Fair Market Rent lookup.
|
|
|
|
API: https://www.huduser.gov/portal/dataset/fmr-api.html
|
|
Requiere API key gratis: https://www.huduser.gov/hudapi/public/register
|
|
|
|
Flow:
|
|
1. GET /fmr/listCounties/{state} -> match county_name -> fips_code
|
|
2. GET /fmr/data/{fips_code}?year=YYYY -> Efficiency / 1BR / 2BR / 3BR / 4BR
|
|
|
|
Si HUD_API_KEY no esta en .env, raise FetcherError (caught por runner, fail-soft).
|
|
|
|
Devuelve dict con:
|
|
year, county, state,
|
|
fmr_efficiency, fmr_1br, fmr_2br, fmr_3br, fmr_4br (en USD/mes)
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import os
|
|
from datetime import datetime
|
|
|
|
import requests
|
|
from dotenv import load_dotenv
|
|
|
|
from .base import FetcherError, DEFAULT_TIMEOUT
|
|
|
|
|
|
HUD_BASE = "https://www.huduser.gov/hudapi/public"
|
|
|
|
|
|
def _normalize_county_name(s: str) -> str:
|
|
"""Normaliza para comparar nombres: lowercase, sin sufijo 'County', sin espacios redundantes."""
|
|
if not s:
|
|
return ""
|
|
s = s.lower().strip()
|
|
if s.endswith(" county"):
|
|
s = s[:-7].strip()
|
|
return " ".join(s.split()) # collapse whitespace
|
|
|
|
|
|
def fetch_fmr(state: str, county_name: str, year: int | None = None) -> dict:
|
|
"""Fetch FMR para un condado USA.
|
|
|
|
state: codigo 2-letras (ej. "FL", "TX")
|
|
county_name: nombre del condado (con o sin "County")
|
|
year: ano del FMR (default: ano actual)
|
|
"""
|
|
# .env ya fue cargado por data_fetchers/__init__.py
|
|
api_key = os.getenv("HUD_API_KEY", "").strip()
|
|
if not api_key:
|
|
raise FetcherError("HUD_API_KEY no esta en .env. Registrate en https://www.huduser.gov/hudapi/public/register")
|
|
|
|
if not state or not county_name:
|
|
raise FetcherError(f"state y county_name son requeridos (got state={state!r}, county={county_name!r})")
|
|
|
|
if year is None:
|
|
year = datetime.now().year
|
|
|
|
headers = {"Authorization": f"Bearer {api_key}"}
|
|
|
|
# 1. listCounties para encontrar el entity_id (fips_code)
|
|
list_url = f"{HUD_BASE}/fmr/listCounties/{state}"
|
|
try:
|
|
r = requests.get(list_url, headers=headers, timeout=DEFAULT_TIMEOUT)
|
|
r.raise_for_status()
|
|
except requests.RequestException as e:
|
|
raise FetcherError(f"listCounties HTTP error: {e}") from e
|
|
|
|
try:
|
|
counties = r.json()
|
|
except ValueError as e:
|
|
raise FetcherError(f"listCounties JSON error: {e}") from e
|
|
|
|
if not isinstance(counties, list):
|
|
raise FetcherError(f"listCounties unexpected format: {type(counties).__name__}")
|
|
|
|
target = _normalize_county_name(county_name)
|
|
matched = None
|
|
for c in counties:
|
|
if _normalize_county_name(c.get("county_name", "")) == target:
|
|
matched = c
|
|
break
|
|
|
|
if not matched:
|
|
sample = [c.get("county_name") for c in counties[:5]]
|
|
raise FetcherError(f"County '{county_name}' not in HUD list for {state}. Primeros 5: {sample}")
|
|
|
|
entity_id = matched.get("fips_code")
|
|
if not entity_id:
|
|
raise FetcherError(f"Match found but no fips_code: {matched}")
|
|
|
|
# 2. FMR data
|
|
fmr_url = f"{HUD_BASE}/fmr/data/{entity_id}"
|
|
try:
|
|
r = requests.get(fmr_url, params={"year": year}, headers=headers, timeout=DEFAULT_TIMEOUT)
|
|
r.raise_for_status()
|
|
except requests.RequestException as e:
|
|
raise FetcherError(f"fmr/data HTTP error: {e}") from e
|
|
|
|
try:
|
|
payload = r.json()
|
|
except ValueError as e:
|
|
raise FetcherError(f"fmr/data JSON error: {e}") from e
|
|
|
|
# payload structure puede variar. Intentamos varias formas conocidas.
|
|
data = payload.get("data", payload) if isinstance(payload, dict) else {}
|
|
|
|
# basicdata puede ser un dict (condado simple) o lista (metro con sub-zonas)
|
|
bd = data.get("basicdata")
|
|
if isinstance(bd, list):
|
|
bd = bd[0] if bd else {}
|
|
if not isinstance(bd, dict):
|
|
bd = data
|
|
|
|
def _g(*keys):
|
|
"""Devuelve el primer valor presente entre las keys provistas."""
|
|
for k in keys:
|
|
v = bd.get(k)
|
|
if v is not None:
|
|
return v
|
|
return None
|
|
|
|
return {
|
|
"year": year,
|
|
"county": matched.get("county_name"),
|
|
"state": state,
|
|
"entity_id": entity_id,
|
|
"fmr_efficiency": _g("Efficiency", "fmr_0"),
|
|
"fmr_1br": _g("One-Bedroom", "fmr_1"),
|
|
"fmr_2br": _g("Two-Bedroom", "fmr_2"),
|
|
"fmr_3br": _g("Three-Bedroom", "fmr_3"),
|
|
"fmr_4br": _g("Four-Bedroom", "fmr_4"),
|
|
"source": "HUD User FMR API",
|
|
}
|