"""Google Maps Geocoding + Place Search for restaurant location lookup.""" from __future__ import annotations import logging import os import httpx logger = logging.getLogger(__name__) def _api_key() -> str: return os.environ["GOOGLE_MAPS_API_KEY"] def geocode_restaurant(name: str, address: str | None = None, region: str | None = None) -> dict | None: """Look up restaurant coordinates using Google Maps. Tries Places Text Search first (more accurate for restaurant names), falls back to Geocoding API. Returns dict with: latitude, longitude, formatted_address, google_place_id or None if not found. """ query = name if address: query += f" {address}" elif region: query += f" {region}" # Try Places Text Search (better for business names) result = _places_text_search(query) if result: return result # Fallback: Geocoding API return _geocode(query) def _places_text_search(query: str) -> dict | None: """Search for a place using Google Places Text Search API.""" try: r = httpx.get( "https://maps.googleapis.com/maps/api/place/textsearch/json", params={ "query": query, "key": _api_key(), "language": "ko", "type": "restaurant", }, timeout=10, ) r.raise_for_status() data = r.json() if data.get("status") == "OK" and data.get("results"): place = data["results"][0] loc = place["geometry"]["location"] result = { "latitude": loc["lat"], "longitude": loc["lng"], "formatted_address": place.get("formatted_address", ""), "google_place_id": place.get("place_id", ""), "business_status": place.get("business_status"), "rating": place.get("rating"), "rating_count": place.get("user_ratings_total"), } # Fetch phone/website from Place Details place_id = place.get("place_id") if place_id: details = _place_details(place_id) if details: result.update(details) return result except Exception as e: logger.warning("Places text search failed for '%s': %s", query, e) return None def _place_details(place_id: str) -> dict | None: """Fetch phone and website from Google Place Details API.""" try: r = httpx.get( "https://maps.googleapis.com/maps/api/place/details/json", params={ "place_id": place_id, "key": _api_key(), "language": "ko", "fields": "formatted_phone_number,website", }, timeout=10, ) r.raise_for_status() data = r.json() if data.get("status") == "OK" and data.get("result"): res = data["result"] return { "phone": res.get("formatted_phone_number"), "website": res.get("website"), } except Exception as e: logger.warning("Place details failed for '%s': %s", place_id, e) return None def _geocode(query: str) -> dict | None: """Geocode an address string.""" try: r = httpx.get( "https://maps.googleapis.com/maps/api/geocode/json", params={ "address": query, "key": _api_key(), "language": "ko", }, timeout=10, ) r.raise_for_status() data = r.json() if data.get("status") == "OK" and data.get("results"): result = data["results"][0] loc = result["geometry"]["location"] return { "latitude": loc["lat"], "longitude": loc["lng"], "formatted_address": result.get("formatted_address", ""), "google_place_id": "", } except Exception as e: logger.warning("Geocoding failed for '%s': %s", query, e) return None