Backend (FastAPI + Oracle ADB), Frontend (Next.js), daemon worker. Features: channel/video/restaurant management, semantic search, Google OAuth, user reviews. Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
98 lines
2.8 KiB
Python
98 lines
2.8 KiB
Python
"""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"]
|
|
return {
|
|
"latitude": loc["lat"],
|
|
"longitude": loc["lng"],
|
|
"formatted_address": place.get("formatted_address", ""),
|
|
"google_place_id": place.get("place_id", ""),
|
|
}
|
|
except Exception as e:
|
|
logger.warning("Places text search failed for '%s': %s", query, 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
|