Template
Add EDDM size, income, businesses and USPS map link (#14)
## Summary Closes #13. - Mirror Monica: join USPS EDDM GIS stats onto the carrier-route picker (household size, median income, business count). - Add Open USPS EDDM map link next to Find routes. - GIS miss stays a dash and does not block Find routes. PCM households and cost unchanged. Monica: ai_ml_operations/monica_site#17 ## Test plan - [ ] Find routes for a ZIP; table shows Size / Income / Businesses columns - [ ] Missing GIS values render as — - [ ] Open USPS EDDM map opens eddm.usps.com - [ ] Quote and 1–2 route send still work - [ ] `CarrierRoutePostcardTests` pass Reviewed-on: #14
This commit was merged in pull request #14.
This commit is contained in:
1 parent
f2eb4b7152
commit
7ff52c776b
6 files changed
+254
-3
No files matched your search
@@ -0,0 +1,151 @@
|
||||
"""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
|
||||
Reference in new issue
Block a user