"""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", }