fix(skills/local-places): copy files instead of submodule
Submodules are pain. Just copy the Python code directly.
🦞
This commit is contained in:
Peter Steinberger
314
skills/local-places/src/local_places/google_places.py
Normal file
314
skills/local-places/src/local_places/google_places.py
Normal file
@@ -0,0 +1,314 @@
|
||||
from __future__ import annotations
|
||||
|
||||
import logging
|
||||
import os
|
||||
from typing import Any
|
||||
|
||||
import httpx
|
||||
from fastapi import HTTPException
|
||||
|
||||
from local_places.schemas import (
|
||||
LatLng,
|
||||
LocationResolveRequest,
|
||||
LocationResolveResponse,
|
||||
PlaceDetails,
|
||||
PlaceSummary,
|
||||
ResolvedLocation,
|
||||
SearchRequest,
|
||||
SearchResponse,
|
||||
)
|
||||
|
||||
GOOGLE_PLACES_BASE_URL = os.getenv(
|
||||
"GOOGLE_PLACES_BASE_URL", "https://places.googleapis.com/v1"
|
||||
)
|
||||
logger = logging.getLogger("local_places.google_places")
|
||||
|
||||
_PRICE_LEVEL_TO_ENUM = {
|
||||
0: "PRICE_LEVEL_FREE",
|
||||
1: "PRICE_LEVEL_INEXPENSIVE",
|
||||
2: "PRICE_LEVEL_MODERATE",
|
||||
3: "PRICE_LEVEL_EXPENSIVE",
|
||||
4: "PRICE_LEVEL_VERY_EXPENSIVE",
|
||||
}
|
||||
_ENUM_TO_PRICE_LEVEL = {value: key for key, value in _PRICE_LEVEL_TO_ENUM.items()}
|
||||
|
||||
_SEARCH_FIELD_MASK = (
|
||||
"places.id,"
|
||||
"places.displayName,"
|
||||
"places.formattedAddress,"
|
||||
"places.location,"
|
||||
"places.rating,"
|
||||
"places.priceLevel,"
|
||||
"places.types,"
|
||||
"places.currentOpeningHours,"
|
||||
"nextPageToken"
|
||||
)
|
||||
|
||||
_DETAILS_FIELD_MASK = (
|
||||
"id,"
|
||||
"displayName,"
|
||||
"formattedAddress,"
|
||||
"location,"
|
||||
"rating,"
|
||||
"priceLevel,"
|
||||
"types,"
|
||||
"regularOpeningHours,"
|
||||
"currentOpeningHours,"
|
||||
"nationalPhoneNumber,"
|
||||
"websiteUri"
|
||||
)
|
||||
|
||||
_RESOLVE_FIELD_MASK = (
|
||||
"places.id,"
|
||||
"places.displayName,"
|
||||
"places.formattedAddress,"
|
||||
"places.location,"
|
||||
"places.types"
|
||||
)
|
||||
|
||||
|
||||
class _GoogleResponse:
|
||||
def __init__(self, response: httpx.Response):
|
||||
self.status_code = response.status_code
|
||||
self._response = response
|
||||
|
||||
def json(self) -> dict[str, Any]:
|
||||
return self._response.json()
|
||||
|
||||
@property
|
||||
def text(self) -> str:
|
||||
return self._response.text
|
||||
|
||||
|
||||
def _api_headers(field_mask: str) -> dict[str, str]:
|
||||
api_key = os.getenv("GOOGLE_PLACES_API_KEY")
|
||||
if not api_key:
|
||||
raise HTTPException(
|
||||
status_code=500,
|
||||
detail="GOOGLE_PLACES_API_KEY is not set.",
|
||||
)
|
||||
return {
|
||||
"Content-Type": "application/json",
|
||||
"X-Goog-Api-Key": api_key,
|
||||
"X-Goog-FieldMask": field_mask,
|
||||
}
|
||||
|
||||
|
||||
def _request(
|
||||
method: str, url: str, payload: dict[str, Any] | None, field_mask: str
|
||||
) -> _GoogleResponse:
|
||||
try:
|
||||
with httpx.Client(timeout=10.0) as client:
|
||||
response = client.request(
|
||||
method=method,
|
||||
url=url,
|
||||
headers=_api_headers(field_mask),
|
||||
json=payload,
|
||||
)
|
||||
except httpx.HTTPError as exc:
|
||||
raise HTTPException(status_code=502, detail="Google Places API unavailable.") from exc
|
||||
|
||||
return _GoogleResponse(response)
|
||||
|
||||
|
||||
def _build_text_query(request: SearchRequest) -> str:
|
||||
keyword = request.filters.keyword if request.filters else None
|
||||
if keyword:
|
||||
return f"{request.query} {keyword}".strip()
|
||||
return request.query
|
||||
|
||||
|
||||
def _build_search_body(request: SearchRequest) -> dict[str, Any]:
|
||||
body: dict[str, Any] = {
|
||||
"textQuery": _build_text_query(request),
|
||||
"pageSize": request.limit,
|
||||
}
|
||||
|
||||
if request.page_token:
|
||||
body["pageToken"] = request.page_token
|
||||
|
||||
if request.location_bias:
|
||||
body["locationBias"] = {
|
||||
"circle": {
|
||||
"center": {
|
||||
"latitude": request.location_bias.lat,
|
||||
"longitude": request.location_bias.lng,
|
||||
},
|
||||
"radius": request.location_bias.radius_m,
|
||||
}
|
||||
}
|
||||
|
||||
if request.filters:
|
||||
filters = request.filters
|
||||
if filters.types:
|
||||
body["includedType"] = filters.types[0]
|
||||
if filters.open_now is not None:
|
||||
body["openNow"] = filters.open_now
|
||||
if filters.min_rating is not None:
|
||||
body["minRating"] = filters.min_rating
|
||||
if filters.price_levels:
|
||||
body["priceLevels"] = [
|
||||
_PRICE_LEVEL_TO_ENUM[level] for level in filters.price_levels
|
||||
]
|
||||
|
||||
return body
|
||||
|
||||
|
||||
def _parse_lat_lng(raw: dict[str, Any] | None) -> LatLng | None:
|
||||
if not raw:
|
||||
return None
|
||||
latitude = raw.get("latitude")
|
||||
longitude = raw.get("longitude")
|
||||
if latitude is None or longitude is None:
|
||||
return None
|
||||
return LatLng(lat=latitude, lng=longitude)
|
||||
|
||||
|
||||
def _parse_display_name(raw: dict[str, Any] | None) -> str | None:
|
||||
if not raw:
|
||||
return None
|
||||
return raw.get("text")
|
||||
|
||||
|
||||
def _parse_open_now(raw: dict[str, Any] | None) -> bool | None:
|
||||
if not raw:
|
||||
return None
|
||||
return raw.get("openNow")
|
||||
|
||||
|
||||
def _parse_hours(raw: dict[str, Any] | None) -> list[str] | None:
|
||||
if not raw:
|
||||
return None
|
||||
return raw.get("weekdayDescriptions")
|
||||
|
||||
|
||||
def _parse_price_level(raw: str | None) -> int | None:
|
||||
if not raw:
|
||||
return None
|
||||
return _ENUM_TO_PRICE_LEVEL.get(raw)
|
||||
|
||||
|
||||
def search_places(request: SearchRequest) -> SearchResponse:
|
||||
url = f"{GOOGLE_PLACES_BASE_URL}/places:searchText"
|
||||
response = _request("POST", url, _build_search_body(request), _SEARCH_FIELD_MASK)
|
||||
|
||||
if response.status_code >= 400:
|
||||
logger.error(
|
||||
"Google Places API error %s. response=%s",
|
||||
response.status_code,
|
||||
response.text,
|
||||
)
|
||||
raise HTTPException(
|
||||
status_code=502,
|
||||
detail=f"Google Places API error ({response.status_code}).",
|
||||
)
|
||||
|
||||
try:
|
||||
payload = response.json()
|
||||
except ValueError as exc:
|
||||
logger.error(
|
||||
"Google Places API returned invalid JSON. response=%s",
|
||||
response.text,
|
||||
)
|
||||
raise HTTPException(status_code=502, detail="Invalid Google response.") from exc
|
||||
|
||||
places = payload.get("places", [])
|
||||
results = []
|
||||
for place in places:
|
||||
results.append(
|
||||
PlaceSummary(
|
||||
place_id=place.get("id", ""),
|
||||
name=_parse_display_name(place.get("displayName")),
|
||||
address=place.get("formattedAddress"),
|
||||
location=_parse_lat_lng(place.get("location")),
|
||||
rating=place.get("rating"),
|
||||
price_level=_parse_price_level(place.get("priceLevel")),
|
||||
types=place.get("types"),
|
||||
open_now=_parse_open_now(place.get("currentOpeningHours")),
|
||||
)
|
||||
)
|
||||
|
||||
return SearchResponse(
|
||||
results=results,
|
||||
next_page_token=payload.get("nextPageToken"),
|
||||
)
|
||||
|
||||
|
||||
def get_place_details(place_id: str) -> PlaceDetails:
|
||||
url = f"{GOOGLE_PLACES_BASE_URL}/places/{place_id}"
|
||||
response = _request("GET", url, None, _DETAILS_FIELD_MASK)
|
||||
|
||||
if response.status_code >= 400:
|
||||
logger.error(
|
||||
"Google Places API error %s. response=%s",
|
||||
response.status_code,
|
||||
response.text,
|
||||
)
|
||||
raise HTTPException(
|
||||
status_code=502,
|
||||
detail=f"Google Places API error ({response.status_code}).",
|
||||
)
|
||||
|
||||
try:
|
||||
payload = response.json()
|
||||
except ValueError as exc:
|
||||
logger.error(
|
||||
"Google Places API returned invalid JSON. response=%s",
|
||||
response.text,
|
||||
)
|
||||
raise HTTPException(status_code=502, detail="Invalid Google response.") from exc
|
||||
|
||||
return PlaceDetails(
|
||||
place_id=payload.get("id", place_id),
|
||||
name=_parse_display_name(payload.get("displayName")),
|
||||
address=payload.get("formattedAddress"),
|
||||
location=_parse_lat_lng(payload.get("location")),
|
||||
rating=payload.get("rating"),
|
||||
price_level=_parse_price_level(payload.get("priceLevel")),
|
||||
types=payload.get("types"),
|
||||
phone=payload.get("nationalPhoneNumber"),
|
||||
website=payload.get("websiteUri"),
|
||||
hours=_parse_hours(payload.get("regularOpeningHours")),
|
||||
open_now=_parse_open_now(payload.get("currentOpeningHours")),
|
||||
)
|
||||
|
||||
|
||||
def resolve_locations(request: LocationResolveRequest) -> LocationResolveResponse:
|
||||
url = f"{GOOGLE_PLACES_BASE_URL}/places:searchText"
|
||||
body = {"textQuery": request.location_text, "pageSize": request.limit}
|
||||
response = _request("POST", url, body, _RESOLVE_FIELD_MASK)
|
||||
|
||||
if response.status_code >= 400:
|
||||
logger.error(
|
||||
"Google Places API error %s. response=%s",
|
||||
response.status_code,
|
||||
response.text,
|
||||
)
|
||||
raise HTTPException(
|
||||
status_code=502,
|
||||
detail=f"Google Places API error ({response.status_code}).",
|
||||
)
|
||||
|
||||
try:
|
||||
payload = response.json()
|
||||
except ValueError as exc:
|
||||
logger.error(
|
||||
"Google Places API returned invalid JSON. response=%s",
|
||||
response.text,
|
||||
)
|
||||
raise HTTPException(status_code=502, detail="Invalid Google response.") from exc
|
||||
|
||||
places = payload.get("places", [])
|
||||
results = []
|
||||
for place in places:
|
||||
results.append(
|
||||
ResolvedLocation(
|
||||
place_id=place.get("id", ""),
|
||||
name=_parse_display_name(place.get("displayName")),
|
||||
address=place.get("formattedAddress"),
|
||||
location=_parse_lat_lng(place.get("location")),
|
||||
types=place.get("types"),
|
||||
)
|
||||
)
|
||||
|
||||
return LocationResolveResponse(results=results)
|
||||
Reference in New Issue
Block a user