from __future__ import annotations import asyncio import base64 import hashlib import hmac import logging import re from typing import Any, Dict, Optional, Tuple from urllib.parse import unquote, urlencode import httpx from app.config import get_settings from app.utils.http_utils import SharedAsyncClient _logger = logging.getLogger(__name__) _settings = get_settings() PRICE_LEVEL_MAP: dict[str, int] = { "PRICE_LEVEL_UNSPECIFIED": 0, "PRICE_LEVEL_FREE": 0, "PRICE_LEVEL_INEXPENSIVE": 1, "PRICE_LEVEL_MODERATE": 2, "PRICE_LEVEL_EXPENSIVE": 3, "PRICE_LEVEL_VERY_EXPENSIVE": 4, } DEFAULT_SEARCH_FIELD_MASK = ( "places.id,places.displayName,places.formattedAddress," "places.location,places.rating,places.userRatingCount," "places.priceLevel,places.types,places.businessStatus," "places.photos,places.plusCode,places.shortFormattedAddress," "places.regularOpeningHours,places.websiteUri," "places.internationalPhoneNumber,places.nationalPhoneNumber," "places.googleMapsUri,places.iconMaskBaseUri" ) DEFAULT_DETAILS_FIELD_MASK = ( "id,displayName,formattedAddress,location,rating,userRatingCount," "priceLevel,types,businessStatus,photos,plusCode,shortFormattedAddress," "regularOpeningHours,currentOpeningHours,websiteUri," "internationalPhoneNumber,nationalPhoneNumber,googleMapsUri," "iconMaskBaseUri,editorialSummary" ) DEFAULT_AUTOCOMPLETE_FIELD_MASK = ( "suggestions.placePrediction.text.text," "suggestions.placePrediction.placeId," "suggestions.placePrediction.types," "suggestions.placePrediction.distanceMeters," "suggestions.queryPrediction.text.text" ) STATIC_MAP_BASE_URL = "https://maps.googleapis.com/maps/api/staticmap" STATIC_MAP_FORMAT_CONTENT_TYPES: dict[str, str] = { "png": "image/png", "png8": "image/png", "png32": "image/png", "gif": "image/gif", "jpg": "image/jpeg", "jpg-baseline": "image/jpeg", } _STYLE_OPERATION_KEYS = ( "hue", "lightness", "saturation", "gamma", "invert_lightness", "visibility", "color", "weight", ) class GoogleMapsService: def __init__(self) -> None: self._base_url: str = _settings.google_maps_base_url self._places_base_url: str = _settings.google_maps_places_base_url self._timeout: int = _settings.google_maps_timeout self._max_retries: int = _settings.google_maps_max_retries self._http = SharedAsyncClient(timeout=self._timeout) # ----------------------------------------------------------------------- # HTTP client management # ----------------------------------------------------------------------- async def _get_client(self) -> httpx.AsyncClient: """Return the shared connection-pooled AsyncClient, creating it lazily.""" return await self._http.get() async def close(self) -> None: """Close the shared AsyncClient and release pooled connections.""" await self._http.close() # ----------------------------------------------------------------------- # Geocoding — still uses the legacy Google Geocoding API # ----------------------------------------------------------------------- async def geocode(self, address: str, region: Optional[str] = None, language: Optional[str] = None, bounds: Optional[str] = None, api_key: Optional[str] = None) -> Dict[str, Any]: if not api_key or not api_key.strip(): return {"success": False, "error": "Google API key is missing. Provide a valid API key via the X-Goog-Api-Key header."} params: Dict[str, Any] = { "address": address, "key": api_key, } if region: params["region"] = region if language: params["language"] = language if bounds: params["bounds"] = bounds return await self._call_api("/geocode/json", params) async def reverse_geocode(self, latlng: str, language: Optional[str] = None, result_type: Optional[str] = None, location_type: Optional[str] = None, api_key: Optional[str] = None) -> Dict[str, Any]: if not api_key or not api_key.strip(): return {"success": False, "error": "Google API key is missing. Provide a valid API key via the X-Goog-Api-Key header."} params: Dict[str, Any] = { "latlng": latlng, "key": api_key, } if language: params["language"] = language if result_type: params["result_type"] = result_type if location_type: params["location_type"] = location_type return await self._call_api("/geocode/json", params) # ----------------------------------------------------------------------- # Places API (New) — places.googleapis.com/v1 # ----------------------------------------------------------------------- def _price_levels_from_range(self, min_price: Optional[int], max_price: Optional[int]) -> list[str]: levels = ["PRICE_LEVEL_FREE", "PRICE_LEVEL_INEXPENSIVE", "PRICE_LEVEL_MODERATE", "PRICE_LEVEL_EXPENSIVE", "PRICE_LEVEL_VERY_EXPENSIVE"] lo = max(0, min_price if min_price is not None else 0) hi = min(4, max_price if max_price is not None else 4) if lo > hi: lo, hi = hi, lo return [levels[i] for i in range(lo, hi + 1)] async def places_search(self, query: str, region: Optional[str] = None, language: Optional[str] = None, min_price: Optional[int] = None, max_price: Optional[int] = None, open_now: Optional[bool] = None, type_filter: Optional[str] = None, radius: Optional[int] = None, page_token: Optional[str] = None, page_size: Optional[int] = None, min_rating: Optional[float] = None, api_key: Optional[str] = None) -> Dict[str, Any]: body: Dict[str, Any] = { "textQuery": query, } if page_token: body["pageToken"] = page_token else: body["maxResultCount"] = page_size or 10 if region: body["regionCode"] = region.upper() if language: body["languageCode"] = language if type_filter: body["includedType"] = type_filter if min_price is not None or max_price is not None: body["priceLevels"] = self._price_levels_from_range(min_price, max_price) if open_now is not None: body["openNow"] = open_now if min_rating is not None: body["minRating"] = min_rating return await self._call_places_api("POST", "/places:searchText", body=body, field_mask=DEFAULT_SEARCH_FIELD_MASK, api_key=api_key) async def places_nearby(self, location: str, radius: int = 1000, keyword: Optional[str] = None, language: Optional[str] = None, min_price: Optional[int] = None, max_price: Optional[int] = None, open_now: Optional[bool] = None, type_filter: Optional[str] = None, page_token: Optional[str] = None, page_size: Optional[int] = None, rank_preference: Optional[str] = None, api_key: Optional[str] = None) -> Dict[str, Any]: parts = location.split(",") lat = float(parts[0].strip()) lng = float(parts[1].strip()) body: Dict[str, Any] = { "locationRestriction": { "circle": { "center": {"latitude": lat, "longitude": lng}, "radius": float(radius), } }, } if page_token: body["pageToken"] = page_token else: body["maxResultCount"] = page_size or 10 if type_filter: body["includedTypes"] = type_filter.split(",") if keyword: body["includedPrimaryTypes"] = [keyword] if language: body["languageCode"] = language if min_price is not None or max_price is not None: body["priceLevels"] = self._price_levels_from_range(min_price, max_price) if open_now is not None: body["openNow"] = open_now if rank_preference: body["rankPreference"] = rank_preference.upper() return await self._call_places_api("POST", "/places:searchNearby", body=body, field_mask=DEFAULT_SEARCH_FIELD_MASK, api_key=api_key) async def place_autocomplete(self, input: str, offset: Optional[int] = None, origin: Optional[str] = None, location: Optional[str] = None, radius: Optional[int] = None, language: Optional[str] = None, types: Optional[str] = None, components: Optional[str] = None, strictbounds: Optional[bool] = None, sessiontoken: Optional[str] = None, api_key: Optional[str] = None) -> Dict[str, Any]: body: Dict[str, Any] = { "input": input, } if language: body["languageCode"] = language if offset is not None: body["inputOffset"] = offset if location: parts = location.split(",") body["locationBias"] = { "circle": { "center": {"latitude": float(parts[0].strip()), "longitude": float(parts[1].strip())}, "radius": float(radius or 50000), } } if origin: parts = origin.split(",") body["origin"] = {"latitude": float(parts[0].strip()), "longitude": float(parts[1].strip())} if types: mapped = [] for t in types.replace("(", "").replace(")", "").split("|"): t = t.strip() if t == "cities": t = "locality" mapped.append(t) body["includedPrimaryTypes"] = mapped if components: parts_list = [c.split(":") for c in components.split("|") if ":" in c] for pair in parts_list: key, val = pair[0].strip(), pair[1].strip() if key == "country": body["regionCode"] = val.upper() if sessiontoken: body["sessionToken"] = sessiontoken field_mask = "suggestions.placePrediction.text.text,suggestions.placePrediction.placeId,suggestions.placePrediction.types,suggestions.placePrediction.distanceMeters,suggestions.placePrediction.structuredFormat.mainText.text,suggestions.placePrediction.structuredFormat.secondaryText.text" return await self._call_places_api("POST", "/places:autocomplete", body=body, field_mask=field_mask, api_key=api_key) async def query_autocomplete(self, input: str, offset: Optional[int] = None, location: Optional[str] = None, radius: Optional[int] = None, language: Optional[str] = None, api_key: Optional[str] = None) -> Dict[str, Any]: body: Dict[str, Any] = { "input": input, "includeQueryPredictions": True, } if language: body["languageCode"] = language if offset is not None: body["inputOffset"] = offset if location: parts = location.split(",") body["locationBias"] = { "circle": { "center": {"latitude": float(parts[0].strip()), "longitude": float(parts[1].strip())}, "radius": float(radius or 50000), } } field_mask = "suggestions.placePrediction.text.text,suggestions.placePrediction.placeId,suggestions.placePrediction.types,suggestions.placePrediction.distanceMeters,suggestions.queryPrediction.text.text" return await self._call_places_api("POST", "/places:autocomplete", body=body, field_mask=field_mask, api_key=api_key) async def place_details(self, place_id: str, region: Optional[str] = None, language: Optional[str] = None, fields: Optional[str] = None, api_key: Optional[str] = None) -> Dict[str, Any]: path = f"/places/{place_id}" field_mask = fields or DEFAULT_DETAILS_FIELD_MASK params = {} if language: params["languageCode"] = language return await self._call_places_api("GET", path, params=params, field_mask=field_mask, api_key=api_key) # ----------------------------------------------------------------------- # Google Maps Static API — image generation # ----------------------------------------------------------------------- @staticmethod def _build_marker_param(marker: Dict[str, Any]) -> str: """Serialize a marker group dict into a Google `markers` parameter string.""" style = marker.get("style") or {} parts: list[str] = [] for key in ("size", "color", "label", "icon", "anchor", "scale"): value = style.get(key) if value is not None: parts.append(f"{key}:{value}") parts.extend(marker.get("locations", [])) return "|".join(parts) @staticmethod def _build_path_param(path: Dict[str, Any]) -> str: """Serialize a path dict into a Google `path` parameter string.""" style = path.get("style") or {} parts: list[str] = [] if style.get("color"): parts.append(f"color:{style['color']}") if style.get("weight") is not None: parts.append(f"weight:{style['weight']}") if style.get("fill"): parts.append(f"fill:{style['fill']}") if style.get("geodesic"): parts.append("geodesic:true") if path.get("encoded_polyline"): parts.append(f"enc:{path['encoded_polyline']}") else: parts.extend(path.get("points", [])) return "|".join(parts) @staticmethod def _build_style_param(rule: Dict[str, Any]) -> str: """Serialize a style rule dict into a Google `style` parameter string.""" parts: list[str] = [] if rule.get("feature"): parts.append(f"feature:{rule['feature']}") if rule.get("element"): parts.append(f"element:{rule['element']}") for key in _STYLE_OPERATION_KEYS: value = rule.get(key) if value is None: continue if isinstance(value, bool): parts.append(f"{key}:{str(value).lower()}") else: parts.append(f"{key}:{value}") return "|".join(parts) @staticmethod def _sign_static_map_url(url: str, signing_secret: str) -> str: """Apply a Google Maps URL digital signature (HMAC-SHA1) to a request URL.""" # Strip the protocol scheme and host, keeping only the path + query. path_and_query = url.split("://", 1)[1].split("/", 1)[1] # Remove any existing signature parameter before signing. path_and_query = re.sub(r"&signature=[^&]*", "", path_and_query) decoded = unquote(path_and_query) key = base64.urlsafe_b64decode(signing_secret + "=" * (-len(signing_secret) % 4)) signature = base64.urlsafe_b64encode( hmac.new(key, decoded.encode("utf-8"), hashlib.sha1).digest() ).rstrip(b"=").decode("ascii") return f"{url}&signature={signature}" def build_static_map_url(self, params: Dict[str, Any], api_key: str, signature_secret: Optional[str] = None) -> Optional[str]: """Build a Google Static Maps URL from validated params. Returns ``None`` when the API key is missing. ``markers`` / ``paths`` / ``styles`` entries may be either structured dicts (as produced by the request schema) or pre-formatted strings (raw GET query values). """ if not api_key or not api_key.strip(): return None query: list[Tuple[str, str]] = [] for key in ("center", "zoom", "size", "scale", "format", "maptype", "language", "region", "map_id"): value = params.get(key) if value is not None: query.append((key, str(value))) for marker in params.get("markers", []): query.append(("markers", marker if isinstance(marker, str) else self._build_marker_param(marker))) for path in params.get("paths", []): query.append(("path", path if isinstance(path, str) else self._build_path_param(path))) visible = params.get("visible") if visible: if isinstance(visible, str): visible = [v for v in visible.split("|") if v.strip()] if visible: query.append(("visible", "|".join(visible))) for rule in params.get("styles", []): query.append(("style", rule if isinstance(rule, str) else self._build_style_param(rule))) query.append(("key", api_key)) url = f"{STATIC_MAP_BASE_URL}?{urlencode(query)}" if signature_secret: url = self._sign_static_map_url(url, signature_secret) return url @staticmethod def _format_static_map_error(status_code: int, response: httpx.Response) -> str: """Extract a clean, human-readable error message from a Static Maps error response.""" body = response.text.strip() if status_code == 401: return "Unauthorized. The API key may be missing or restricted." if status_code == 403: if "api key" in body.lower(): return "Invalid Google API key. Please provide a valid API key via the X-Goog-Api-Key header." return "Access forbidden. The API key may not have the required APIs enabled or the request is unsigned." if status_code == 429: return "API rate limit exceeded. Please wait and retry." if body: return body[:500] if len(body) > 500 else body return f"Maps Static API error (HTTP {status_code})." async def static_map(self, params: Dict[str, Any], api_key: str) -> Dict[str, Any]: """Fetch a static map image from the Google Maps Static API. ``params`` should be the validated request payload (as a dict). Returns a dict with ``success``, and on success ``content`` (image bytes), ``content_type``, ``url`` and optionally ``warning``. """ signature_secret = params.pop("signature_secret", None) if isinstance(params, dict) else None url = self.build_static_map_url(params, api_key, signature_secret) if url is None: return {"success": False, "error": "Google API key is missing. Provide a valid API key via the X-Goog-Api-Key header."} last_error: Optional[str] = None client = await self._get_client() for attempt in range(1 + self._max_retries): try: response = await client.get(url) if response.status_code == 200: return { "success": True, "content": response.content, "content_type": response.headers.get("content-type", "image/png"), "url": url, "warning": response.headers.get("X-Staticmap-API-Warning") or None, "error": None, } last_error = self._format_static_map_error(response.status_code, response) _logger.warning("Maps Static API error on static map: HTTP %s -> %s", response.status_code, last_error) if response.status_code == 429: if attempt < self._max_retries: await asyncio.sleep(2.0 ** (attempt + 1)) continue if 400 <= response.status_code < 500: break except httpx.TimeoutException: last_error = "Request timed out" _logger.warning("Maps Static API timeout (attempt %d/%d)", attempt + 1, 1 + self._max_retries) except httpx.RequestError as e: last_error = f"Request failed: {e}" _logger.warning("Maps Static API request error: %s (attempt %d/%d)", e, attempt + 1, 1 + self._max_retries) except Exception as e: last_error = f"Unexpected error: {e}" _logger.error("Maps Static API unexpected error: %s", last_error) break if attempt < self._max_retries: await asyncio.sleep(1.0 * (attempt + 1)) return {"success": False, "error": last_error or "Unknown error"} # ----------------------------------------------------------------------- # Internal — Legacy API (for Geocoding) # ----------------------------------------------------------------------- @staticmethod def _format_legacy_api_error(response: httpx.Response) -> str: """Extract a clean error message from a legacy Geocoding API error response.""" try: body = response.json() msg = body.get("error_message", "") or body.get("status", "") if msg: if "invalid" in msg.lower() or "denied" in msg.lower(): return "Invalid Google API key. Please provide a valid API key via the X-Goog-Api-Key header." return msg.rstrip(".") + "." except Exception: pass return f"Geocoding API error (HTTP {response.status_code})." async def _call_api(self, path: str, params: Dict[str, Any]) -> Dict[str, Any]: url = f"{self._base_url}{path}" last_error: Optional[str] = None client = await self._get_client() for attempt in range(1 + self._max_retries): try: response = await client.get(url, params=params) response.raise_for_status() data: Dict[str, Any] = response.json() return self._normalize_response(data) except httpx.TimeoutException: last_error = "Request timed out" _logger.warning("Google Maps API timeout on %s (attempt %d/%d)", path, attempt + 1, 1 + self._max_retries) except httpx.HTTPStatusError as e: last_error = self._format_legacy_api_error(e.response) _logger.warning("Google Maps API HTTP error on %s: %s (attempt %d/%d)", path, last_error, attempt + 1, 1 + self._max_retries) if 400 <= e.response.status_code < 500: break except httpx.RequestError as e: last_error = f"Request failed: {e}" _logger.warning("Google Maps API request error on %s: %s (attempt %d/%d)", path, last_error, attempt + 1, 1 + self._max_retries) except Exception as e: last_error = f"Unexpected error: {e}" _logger.error("Google Maps API unexpected error on %s: %s", path, last_error) break if attempt < self._max_retries: await asyncio.sleep(1.0 * (attempt + 1)) return {"success": False, "error": last_error or "Unknown error"} def _normalize_response(self, data: Dict[str, Any]) -> Dict[str, Any]: status: str = data.get("status", "") error_message: str = data.get("error_message", "") if status == "OK" or status == "ZERO_RESULTS": return { "success": True, "status": status, "data": data, "error": None, } if status == "OVER_QUERY_LIMIT": return {"success": False, "status": status, "error": "API quota exceeded. Please wait and retry."} if status == "REQUEST_DENIED": return {"success": False, "status": status, "error": f"Request denied: {error_message}"} if status == "INVALID_REQUEST": return {"success": False, "status": status, "error": f"Invalid request: {error_message}"} if status == "NOT_FOUND": return {"success": False, "status": status, "error": "The specified place was not found."} return {"success": False, "status": status, "error": error_message or f"Unknown status: {status}"} # ----------------------------------------------------------------------- # Internal — Places API (New) # ----------------------------------------------------------------------- @staticmethod def _format_places_api_error(status_code: int, response: httpx.Response) -> str: """Extract a clean, human-readable error message from a Places API error response.""" try: error_body = response.json() err = error_body.get("error", {}) msg = err.get("message", "") or "" api_status = err.get("status", "") # Build a clean message without dumping raw JSON if "API key not valid" in msg or "API_KEY_INVALID" in api_status: return "Invalid Google API key. Please provide a valid API key via the X-Goog-Api-Key header." if "API key expired" in msg or "API_KEY_EXPIRED" in api_status: return "Google API key has expired. Please renew your API key." if status_code == 403: return "Access forbidden. The API key may not have the required APIs enabled." if status_code == 404: return "The requested resource was not found." if status_code == 429: return "API rate limit exceeded. Please wait and retry." # Return just the human-readable message, not the raw JSON if msg: return msg.rstrip(".") + "." except Exception: pass return f"Google Places API error (HTTP {status_code})." async def _call_places_api(self, method: str, path: str, body: Optional[Dict[str, Any]] = None, params: Optional[Dict[str, Any]] = None, field_mask: Optional[str] = None, api_key: Optional[str] = None) -> Dict[str, Any]: if not api_key or not api_key.strip(): return {"success": False, "error": "Google API key is missing. Provide a valid API key via the X-Goog-Api-Key header."} url = f"{self._places_base_url}{path}" headers: Dict[str, str] = { "X-Goog-Api-Key": api_key, "Content-Type": "application/json", } if field_mask: headers["X-Goog-FieldMask"] = field_mask last_error: Optional[str] = None client = await self._get_client() for attempt in range(1 + self._max_retries): try: if method == "POST": response = await client.post(url, json=body, headers=headers) else: response = await client.get(url, params=params, headers=headers) if response.status_code == 200: data: Dict[str, Any] = response.json() return {"success": True, "data": data, "error": None} last_error = self._format_places_api_error(response.status_code, response) _logger.warning( "Places API error on %s: HTTP %s -> %s", path, response.status_code, last_error, ) if response.status_code == 429: backoff = 2.0 ** (attempt + 1) _logger.warning("Places API rate limited on %s, backing off %.1fs", path, backoff) if attempt < self._max_retries: await asyncio.sleep(backoff) continue if 400 <= response.status_code < 500: break except httpx.TimeoutException: last_error = "Request timed out" _logger.warning("Places API timeout on %s (attempt %d/%d)", path, attempt + 1, 1 + self._max_retries) except httpx.RequestError as e: last_error = f"Request failed: {e}" _logger.warning("Places API request error on %s: %s (attempt %d/%d)", path, last_error, attempt + 1, 1 + self._max_retries) except Exception as e: last_error = f"Unexpected error: {e}" _logger.error("Places API unexpected error on %s: %s", path, last_error) break if attempt < self._max_retries: await asyncio.sleep(1.0 * (attempt + 1)) return {"success": False, "error": last_error or "Unknown error"} async def place_photo(self, photo_reference: str, max_width_px: int, max_height_px: int, api_key: Optional[str]) -> Dict[str, Any]: """Fetch a place photo's image bytes from the New Places API.""" if not api_key or not api_key.strip(): return {"success": False, "error": "Google API key is missing. Provide a valid API key via the X-Goog-Api-Key header."} url = f"{self._places_base_url}/{photo_reference}/media" headers: Dict[str, str] = {"X-Goog-Api-Key": api_key} params: Dict[str, Any] = {"maxWidthPx": max_width_px, "maxHeightPx": max_height_px} last_error: Optional[str] = None client = await self._get_client() for attempt in range(1 + self._max_retries): try: response = await client.get(url, headers=headers, params=params) if response.status_code == 200: return { "success": True, "content": response.content, "content_type": response.headers.get("content-type", "image/jpeg"), "error": None, } last_error = self._format_places_api_error(response.status_code, response) _logger.warning("Places API photo error: HTTP %s -> %s", response.status_code, last_error) if response.status_code == 429: if attempt < self._max_retries: await asyncio.sleep(2.0 ** (attempt + 1)) continue if 400 <= response.status_code < 500: break except httpx.TimeoutException: last_error = "Request timed out" _logger.warning("Places API photo timeout (attempt %d/%d)", attempt + 1, 1 + self._max_retries) except httpx.RequestError as e: last_error = f"Request failed: {e}" _logger.warning("Places API photo request error: %s (attempt %d/%d)", e, attempt + 1, 1 + self._max_retries) except Exception as e: last_error = f"Unexpected error: {e}" _logger.error("Places API photo unexpected error: %s", last_error) break if attempt < self._max_retries: await asyncio.sleep(1.0 * (attempt + 1)) return {"success": False, "error": last_error or "Unknown error"}