web_django_template/site/directmail/providers/postcard/usps_eddm.py

"""USPS EDDM map link + public GIS stats for carrier-route compose."""

from __future__ import annotations

import logging
import time
from typing import Any

import requests

logger = logging.getLogger(__name__)

USPS_EDDM_MAP_URL = "https://eddm.usps.com/eddm/select-routes.htm"
USPS_EDDM_GIS_URL = (
    "https://gis.usps.com/arcgis/rest/services/EDDM/selectZIP/GPServer/routes/execute"
)

_CACHE_TTL_SECONDS = 1800
_cache: dict[str, tuple[float, dict[str, dict[str, Any]]]] = {}


def zip5(raw: str | None) -> str:
    digits = "".join(ch for ch in str(raw or "") if ch.isdigit())
    return digits[:5] if len(digits) >= 5 else ""


def pcm_code_from_zip_crid(*, zip_code: str, crid_id: str) -> str:
    return f"{zip_code}:{crid_id}"


def zips_from_route_codes(codes: list[str], *, hint: str = "") -> list[str]:
    found: list[str] = []
    z = zip5(hint)
    if z:
        found.append(z)
    for code in codes:
        z = zip5(str(code).split(":", 1)[0])
        if z:
            found.append(z)
    return list(dict.fromkeys(found))


def _as_int(raw: Any) -> int | None:
    if raw in (None, ""):
        return None
    try:
        return int(float(raw))
    except (TypeError, ValueError):
        return None


def _as_float(raw: Any) -> float | None:
    if raw in (None, ""):
        return None
    try:
        return float(raw)
    except (TypeError, ValueError):
        return None


def _stats_from_attrs(attrs: dict[str, Any]) -> dict[str, Any] | None:
    zip_code = zip5(attrs.get("ZIP_CODE"))
    crid = str(attrs.get("CRID_ID") or "").strip().upper()
    if not zip_code or not crid:
        zip_crid = str(attrs.get("ZIP_CRID") or "").strip().upper()
        if len(zip_crid) >= 9:
            zip_code = zip_crid[:5]
            crid = zip_crid[5:]
    if not zip_code or not crid:
        return None
    return {
        "code": pcm_code_from_zip_crid(zip_code=zip_code, crid_id=crid),
        "businesses": _as_int(attrs.get("BUS_CNT")),
        "residential": _as_int(attrs.get("RES_CNT")),
        "usps_total": _as_int(attrs.get("TOT_CNT")),
        "household_size": _as_float(attrs.get("AVG_HH_SIZ")),
        "median_income": _as_int(attrs.get("MED_INCOME")),
        "city_state": str(attrs.get("CITY_STATE") or "").strip(),
    }


def _fetch_one_zip(zip_code: str) -> dict[str, dict[str, Any]]:
    response = requests.get(
        USPS_EDDM_GIS_URL,
        params={"f": "json", "ZIP": zip_code, "Rte_Box": "R", "UserName": "EDDM"},
        timeout=25,
        headers={"Accept": "application/json"},
    )
    response.raise_for_status()
    payload = response.json()
    results = payload.get("results") if isinstance(payload, dict) else None
    if not isinstance(results, list) or not results:
        return {}
    value = results[0].get("value") if isinstance(results[0], dict) else None
    features = value.get("features") if isinstance(value, dict) else None
    if not isinstance(features, list):
        return {}
    out: dict[str, dict[str, Any]] = {}
    for feature in features:
        if not isinstance(feature, dict):
            continue
        attrs = feature.get("attributes")
        if not isinstance(attrs, dict):
            continue
        stats = _stats_from_attrs(attrs)
        if not stats:
            continue
        out[stats["code"]] = stats
    return out


def fetch_eddm_route_stats(zip_codes: list[str]) -> dict[str, dict[str, Any]]:
    """ZIP → PCM-style ``60540:C001`` stats from USPS EDDM GIS (fail open)."""
    out: dict[str, dict[str, Any]] = {}
    now = time.monotonic()
    for zip_code in zip_codes:
        z = zip5(zip_code)
        if not z:
            continue
        cached = _cache.get(z)
        if cached and now - cached[0] < _CACHE_TTL_SECONDS:
            out.update(cached[1])
            continue
        try:
            part = _fetch_one_zip(z)
        except Exception:  # noqa: BLE001 — compose still works without GIS
            logger.exception("USPS EDDM GIS lookup failed for ZIP %s", z)
            part = {}
        _cache[z] = (now, part)
        out.update(part)
    return out


def enrich_crrt_rows(rows: list[dict[str, Any]], *, zip_hint: str = "") -> list[dict[str, Any]]:
    """Attach USPS businesses / size / income onto PCM ZipCRRT rows."""
    codes = [str(row.get("code") or "") for row in rows]
    stats = fetch_eddm_route_stats(zips_from_route_codes(codes, hint=zip_hint))
    enriched = []
    for row in rows:
        extra = stats.get(str(row.get("code") or ""), {})
        merged = {
            **row,
            "businesses": extra.get("businesses"),
            "residential": extra.get("residential"),
            "household_size": extra.get("household_size"),
            "median_income": extra.get("median_income"),
        }
        if extra.get("city_state") and not (row.get("text") or "").strip():
            merged["text"] = extra["city_state"]
        enriched.append(merged)
    return enriched