import os, json, requests, psycopg2
AUTOCOMPLETE = "https://autocomplete.search.hereapi.com/v1/autocomplete"
LOOKUP = "https://lookup.search.hereapi.com/v1/lookup"
MATRIX = "https://matrix.router.hereapi.com/v8/matrix"
ROUTER = "https://router.hereapi.com/v8/routes"
API_KEY = os.environ["HERE_API_KEY"]
K = 10 # shortlist size — bounds the matrix, and the bill
COORD_DP = 3 # ~110 m. Ranking is stable within a block.
class EntitlementError(Exception): ...
# ── 1. Autocomplete ───────────────────────────────────────────────────────
def suggest(query: str, at: str, country: str = "USA") -> list[dict]:
"""
Debouncing lives in the CLIENT. This endpoint bills per request.
Calling it per keystroke from a backend moves the problem, not solves it.
"""
if len(query) < 3:
return []
r = requests.get(AUTOCOMPLETE, params={
"q": query, "at": at, "in": f"countryCode:{country}",
"limit": 5, "apiKey": API_KEY,
}, timeout=5)
if r.status_code == 403:
raise EntitlementError("Autocomplete not entitled")
r.raise_for_status()
return [
{"label": it["address"]["label"],
"here_id": it["id"], # ← the key. Never re-geocode the label.
"highlights": it.get("highlights")}
for it in r.json().get("items", [])
]
# ── 2. Resolve the selection ──────────────────────────────────────────────
def resolve(here_id: str) -> dict:
"""
The user handed you a HERE ID. Use it.
Re-geocoding the label string is a wasted call and a worse match.
"""
r = requests.get(LOOKUP, params={"id": here_id, "apiKey": API_KEY}, timeout=5)
r.raise_for_status()
it = r.json()
# `access` is the road-network entry point. Route to this, not `position`.
return (it.get("access") or [it["position"]])[0]
# ── 3. Shortlist. Zero API calls. ─────────────────────────────────────────
def shortlist(conn, lat: float, lng: float, open_now: bool = True) -> list[dict]:
"""
<-> is the KNN operator. Sub-millisecond against the GIST index.
Business filters (open_now, has_pharmacy) run here, for free,
BEFORE anything is sent to HERE.
"""
with conn.cursor() as cur:
cur.execute("""
SELECT id, name, ST_Y(geom::geometry), ST_X(geom::geometry)
FROM stores
WHERE (%s = false OR (now()::time BETWEEN opens_at AND closes_at))
ORDER BY geom <-> ST_SetSRID(ST_MakePoint(%s,%s),4326)::geography
LIMIT %s
""", (open_now, lng, lat, K)) # MakePoint takes (lng, lat)
return [{"id": r[0], "name": r[1], "lat": r[2], "lng": r[3]}
for r in cur.fetchall()]
# ── 4. Rank by drive time. ONE matrix call. ───────────────────────────────
def rank(origin: dict, stores: list[dict]) -> list[dict]:
"""
1 origin × 10 destinations = ONE request.
Ten routing calls would be ten. That ratio is the cost structure.
"""
body = {
"origins": [{"lat": origin["lat"], "lng": origin["lng"]}],
"destinations": [{"lat": s["lat"], "lng": s["lng"]} for s in stores],
"regionDefinition": {"type": "circle",
"center": {"lat": origin["lat"], "lng": origin["lng"]},
"radius": 100_000},
"matrixAttributes": ["travelTimes", "distances"],
}
r = requests.post(MATRIX, params={"apiKey": API_KEY, "async": "false"},
json=body, timeout=20)
if r.status_code == 403:
raise EntitlementError("Matrix Routing not entitled")
r.raise_for_status()
m = r.json()["matrix"]
n = m["numDestinations"]
errs = m.get("errorCodes") or [0] * n
out = []
for j, s in enumerate(stores):
idx = 0 * n + j # flat, row-major. origin index is 0.
if errs[idx]: # 1=disconnected 2=nomatch 3=restriction
continue
out.append({**s,
"travel_time_s": m["travelTimes"][idx],
"distance_m": m["distances"][idx]})
return sorted(out, key=lambda s: s["travel_time_s"])
# ── 5. Directions. ON CLICK ONLY. ─────────────────────────────────────────
def directions(origin: dict, store: dict) -> dict:
r = requests.get(ROUTER, params={
"transportMode": "car",
"origin": f"{origin['lat']},{origin['lng']}",
"destination": f"{store['lat']},{store['lng']}",
"return": "summary,polyline", # explicit. never the default.
"apiKey": API_KEY,
}, timeout=15)
r.raise_for_status()
body = r.json()
if not body.get("routes"): # HTTP 200 ≠ route found
raise ValueError(f"No route: {body.get('notice')}")
s = body["routes"][0]["sections"][0]
return {"duration_s": s["summary"]["duration"],
"length_m": s["summary"]["length"],
"polyline": s["polyline"]}
# ── Orchestration with the ranking cache ──────────────────────────────────
def nearest_stores(conn, here_id: str) -> list[dict]:
coord = resolve(here_id)
key = f"{coord['lat']:.{COORD_DP}f},{coord['lng']:.{COORD_DP}f}"
with conn.cursor() as cur:
cur.execute("""SELECT ranked FROM ranking_cache
WHERE coord_key = %s AND computed_at > now() - interval '7 days'""",
(key,))
hit = cur.fetchone()
if hit:
return hit[0] # zero HERE calls
ranked = rank(coord, shortlist(conn, coord["lat"], coord["lng"]))
with conn.cursor() as cur:
cur.execute("""INSERT INTO ranking_cache (coord_key, ranked)
VALUES (%s,%s)
ON CONFLICT (coord_key) DO UPDATE
SET ranked = EXCLUDED.ranked, computed_at = now()""",
(key, json.dumps(ranked)))
conn.commit()
return ranked
