diff --git a/huf/ai/tests/test_google_places_tools.py b/huf/ai/tests/test_google_places_tools.py new file mode 100644 index 000000000..104d5720e --- /dev/null +++ b/huf/ai/tests/test_google_places_tools.py @@ -0,0 +1,497 @@ +# Copyright (c) 2025, Tridz Technologies Pvt Ltd and Contributors +# See license.txt + +"""Tests for the Google Places (New) integration tools. + +All HTTP calls and Frappe cache access are mocked — no live API key or +Redis instance is required. +""" + +import json +import unittest +from unittest.mock import MagicMock, patch + +from huf.ai.tools import google_places +from huf.ai.tools._registry import GOOGLE_PLACES_TOOLS + +MODULE = "huf.ai.tools.google_places" + +SAMPLE_PLACE = { + "id": "ChIJN1t_tDeuEmsRUsoyG83frY4", + "displayName": {"text": "Sample Cafe", "languageCode": "en"}, + "formattedAddress": "1 Test St, Sydney", + "primaryType": "cafe", + "types": ["cafe", "food"], + "rating": 4.5, + "userRatingCount": 123, + "priceLevel": "PRICE_LEVEL_MODERATE", + "location": {"latitude": -33.86, "longitude": 151.20}, + "googleMapsUri": "https://maps.google.com/?cid=1", + "businessStatus": "OPERATIONAL", + "currentOpeningHours": {"openNow": True}, +} + + +def _mock_response(payload=None, status_code=200, url="https://places.googleapis.com/v1/x"): + resp = MagicMock() + resp.status_code = status_code + resp.ok = 200 <= status_code < 400 + resp.json.return_value = payload if payload is not None else {} + resp.text = json.dumps(payload or {}) + resp.url = url + resp.headers = {} + return resp + + +def _result(raw): + return json.loads(raw) + + +class TestTextSearch(unittest.TestCase): + @patch(f"{MODULE}.update_last_error") + @patch(f"{MODULE}.require_credential", return_value="test-key") + @patch(f"{MODULE}.requests.post") + def test_builds_full_request_body(self, mock_post, _cred, _err): + mock_post.return_value = _mock_response({"places": [SAMPLE_PLACE], "nextPageToken": "tok-2"}) + + out = _result( + google_places.handle_gplaces_text_search( + query="vegan restaurants", + language_code="en", + region_code="us", + included_type="restaurant", + min_rating="4.0", + price_levels="PRICE_LEVEL_INEXPENSIVE, PRICE_LEVEL_MODERATE", + open_now="true", + rank_preference="relevance", + latitude=-33.86, + longitude=151.20, + radius="5000", + page_size="15", + page_token="tok-1", + ) + ) + + self.assertTrue(out["success"]) + self.assertEqual(out["next_page_token"], "tok-2") + self.assertEqual(len(out["places"]), 1) + place = out["places"][0] + self.assertEqual(place["place_id"], SAMPLE_PLACE["id"]) + self.assertEqual(place["name"], "Sample Cafe") + self.assertEqual(place["latitude"], -33.86) + self.assertTrue(place["open_now"]) + + _, kwargs = mock_post.call_args + self.assertEqual(kwargs["json"]["textQuery"], "vegan restaurants") + body = kwargs["json"] + self.assertEqual(body["languageCode"], "en") + self.assertEqual(body["regionCode"], "us") + self.assertEqual(body["includedType"], "restaurant") + self.assertEqual(body["minRating"], 4.0) + self.assertEqual(body["priceLevels"], ["PRICE_LEVEL_INEXPENSIVE", "PRICE_LEVEL_MODERATE"]) + self.assertTrue(body["openNow"]) + self.assertEqual(body["rankPreference"], "RELEVANCE") + self.assertEqual(body["pageSize"], 15) + self.assertEqual(body["pageToken"], "tok-1") + self.assertEqual( + body["locationBias"]["circle"], + {"center": {"latitude": -33.86, "longitude": 151.2}, "radius": 5000.0}, + ) + self.assertIn("places.id", kwargs["headers"]["X-Goog-FieldMask"]) + self.assertEqual(kwargs["headers"]["X-Goog-Api-Key"], "test-key") + + @patch(f"{MODULE}.update_last_error") + @patch(f"{MODULE}.require_credential", return_value="test-key") + @patch(f"{MODULE}.requests.post") + def test_strict_location_uses_restriction(self, mock_post, _cred, _err): + mock_post.return_value = _mock_response({"places": []}) + + out = _result( + google_places.handle_gplaces_text_search( + query="coffee", latitude=10, longitude=20, strict_location=True + ) + ) + + self.assertTrue(out["success"]) + body = mock_post.call_args.kwargs["json"] + self.assertIn("locationRestriction", body) + self.assertNotIn("locationBias", body) + # default radius applies + self.assertEqual(body["locationRestriction"]["circle"]["radius"], 50000.0) + + @patch(f"{MODULE}.update_last_error") + @patch(f"{MODULE}.require_credential", return_value="test-key") + @patch(f"{MODULE}.requests.post") + def test_validation_errors(self, mock_post, _cred, _err): + bad_calls = [ + {}, # missing query + {"query": " "}, # blank query + {"query": "x", "latitude": 91, "longitude": 0}, # latitude out of range + {"query": "x", "latitude": 0, "longitude": 181}, # longitude out of range + {"query": "x", "latitude": 0, "longitude": 0, "radius": 0}, # radius below min + {"query": "x", "latitude": "abc", "longitude": 0}, # non-numeric latitude + {"query": "x", "rank_preference": "CLOSEST"}, # invalid enum + {"query": "x", "page_size": 0}, # page_size out of range + {"query": "x", "page_size": 21}, + {"query": "x", "min_rating": 5.5}, # rating out of range + ] + for kwargs in bad_calls: + out = _result(google_places.handle_gplaces_text_search(**kwargs)) + self.assertFalse(out["success"], f"expected failure for {kwargs}") + self.assertIn("error", out) + mock_post.assert_not_called() + + +class TestPlaceDetails(unittest.TestCase): + @patch(f"{MODULE}.update_last_error") + @patch(f"{MODULE}.require_credential", return_value="test-key") + @patch(f"{MODULE}.requests.get") + def test_normalizes_details(self, mock_get, _cred, _err): + payload = dict(SAMPLE_PLACE) + payload["internationalPhoneNumber"] = "+61 2 0000 0000" + payload["websiteUri"] = "https://sample.cafe" + payload["regularOpeningHours"] = {"weekdayDescriptions": ["Monday: 9-5"]} + payload["photos"] = [ + { + "name": "places/abc/photos/p1", + "widthPx": 1200, + "heightPx": 800, + "authorAttributions": [{"displayName": "Jane"}], + } + ] + payload["reviews"] = [ + { + "name": f"places/abc/reviews/r{i}", + "authorAttribution": {"displayName": f"Reviewer {i}"}, + "rating": 5, + "text": {"text": "Great", "languageCode": "en"}, + "publishTime": "2024-01-01T00:00:00Z", + "relativePublishTimeDescription": "a month ago", + "googleMapsUri": "https://maps.google.com/review", + } + for i in range(7) + ] + payload["reviewSummary"] = {"text": {"text": "Loved it", "languageCode": "en"}} + payload["editorialSummary"] = {"text": {"text": "A cozy cafe", "languageCode": "en"}} + mock_get.return_value = _mock_response(payload) + + out = _result( + google_places.handle_gplaces_place_details( + place_id="ChIJN1t_tDeuEmsRUsoyG83frY4", language_code="en", region_code="au" + ) + ) + + self.assertTrue(out["success"]) + place = out["place"] + self.assertEqual(place["phone"], "+61 2 0000 0000") + self.assertEqual(place["website"], "https://sample.cafe") + self.assertEqual(place["opening_hours"], ["Monday: 9-5"]) + self.assertEqual( + place["photos"], + [{"name": "places/abc/photos/p1", "width_px": 1200, "height_px": 800, "author": "Jane"}], + ) + # capped at 5 most relevant reviews + self.assertEqual(len(place["reviews"]), 5) + self.assertEqual(place["reviews"][0]["author"], "Reviewer 0") + self.assertEqual(place["reviews"][0]["relative_time"], "a month ago") + self.assertEqual(place["review_summary"], "Loved it") + self.assertEqual(place["editorial_summary"], "A cozy cafe") + + args, kwargs = mock_get.call_args + self.assertIn("/v1/places/ChIJN1t_tDeuEmsRUsoyG83frY4", args[0]) + self.assertIn("reviews", kwargs["headers"]["X-Goog-FieldMask"]) + self.assertEqual(kwargs["params"], {"languageCode": "en", "regionCode": "au"}) + + @patch(f"{MODULE}.requests.get") + def test_requires_place_id(self, mock_get): + out = _result(google_places.handle_gplaces_place_details()) + self.assertFalse(out["success"]) + self.assertIn("place_id", out["error"]) + mock_get.assert_not_called() + + +class TestPlacePhoto(unittest.TestCase): + @patch(f"{MODULE}.update_last_error") + @patch(f"{MODULE}.require_credential", return_value="test-key") + @patch(f"{MODULE}.requests.get") + def test_skip_http_redirect_returns_photo_uri(self, mock_get, _cred, _err): + mock_get.return_value = _mock_response( + {"name": "places/abc/photos/p1", "photoUri": "https://cdn.example.com/photo.jpg"} + ) + + out = _result(google_places.handle_gplaces_place_photo(photo_name="places/abc/photos/p1")) + + self.assertTrue(out["success"]) + self.assertEqual(out["photo_url"], "https://cdn.example.com/photo.jpg") + _, kwargs = mock_get.call_args + self.assertEqual(kwargs["params"]["skipHttpRedirect"], "true") + self.assertEqual(kwargs["params"]["maxHeightPx"], 800) + self.assertEqual(kwargs["params"]["maxWidthPx"], 800) + + @patch(f"{MODULE}.update_last_error") + @patch(f"{MODULE}.require_credential", return_value="test-key") + @patch(f"{MODULE}.requests.get") + def test_redirect_captures_location_header(self, mock_get, _cred, _err): + resp = _mock_response(status_code=302) + resp.headers = {"Location": "https://cdn.example.com/redirected.jpg"} + mock_get.return_value = resp + + out = _result( + google_places.handle_gplaces_place_photo( + photo_name="places/abc/photos/p1", + max_height_px=400, + max_width_px=400, + skip_http_redirect=False, + ) + ) + + self.assertTrue(out["success"]) + self.assertEqual(out["photo_url"], "https://cdn.example.com/redirected.jpg") + _, kwargs = mock_get.call_args + self.assertFalse(kwargs["allow_redirects"]) + self.assertNotIn("skipHttpRedirect", kwargs["params"]) + self.assertEqual(kwargs["params"]["maxHeightPx"], 400) + + @patch(f"{MODULE}.requests.get") + def test_requires_photo_name(self, mock_get): + out = _result(google_places.handle_gplaces_place_photo()) + self.assertFalse(out["success"]) + self.assertIn("photo_name", out["error"]) + mock_get.assert_not_called() + + +class TestAutocomplete(unittest.TestCase): + def _cache(self, get_value=None): + cache = MagicMock() + cache.get_value.return_value = get_value + return cache + + @patch(f"{MODULE}.frappe") + @patch(f"{MODULE}.update_last_error") + @patch(f"{MODULE}.require_credential", return_value="test-key") + @patch(f"{MODULE}.requests.post") + def test_builds_body_and_filters_countries(self, mock_post, _cred, _err, mock_frappe): + cache = self._cache() + mock_frappe.cache.return_value = cache + mock_post.return_value = _mock_response( + { + "suggestions": [ + { + "placePrediction": { + "placeId": "p1", + "text": {"text": "Lisbon, Portugal"}, + "primaryType": "locality", + "types": ["locality", "political"], + "distanceMeters": 1200, + } + }, + { + "placePrediction": { + "placeId": "p2", + "text": {"text": "Portugal"}, + "primaryType": "country", + "types": ["country", "political"], + } + }, + {"queryPrediction": {"text": {"text": "lisbon airport"}}}, + ] + } + ) + + out = _result( + google_places.handle_gplaces_autocomplete( + input="Lisbon", + latitude=38.72, + longitude=-9.14, + radius=10000, + strict_location=True, + origin_latitude=38.70, + origin_longitude=-9.10, + session_token="sess-1", + include_query_predictions=True, + ) + ) + + self.assertTrue(out["success"]) + self.assertFalse(out["cached"]) + # country prediction and query prediction are dropped + self.assertEqual(len(out["suggestions"]), 1) + suggestion = out["suggestions"][0] + self.assertEqual(suggestion["place_id"], "p1") + self.assertEqual(suggestion["text"], "Lisbon, Portugal") + self.assertEqual(suggestion["distance_meters"], 1200) + + body = mock_post.call_args.kwargs["json"] + self.assertEqual(body["input"], "Lisbon") + self.assertEqual( + body["includedPrimaryTypes"], + [ + "locality", + "sublocality", + "administrative_area_level_1", + "administrative_area_level_2", + "neighborhood", + ], + ) + self.assertIn("locationRestriction", body) + self.assertEqual(body["origin"], {"latitude": 38.7, "longitude": -9.1}) + self.assertEqual(body["sessionToken"], "sess-1") + self.assertTrue(body["includeQueryPredictions"]) + + # result cached under a place_suggestions:: key for 24h + cache.set_value.assert_called_once() + cache_key = cache.set_value.call_args.args[0] + self.assertTrue(cache_key.startswith("place_suggestions::lisbon")) + self.assertIn("38.72", cache_key) + self.assertEqual(cache.set_value.call_args.kwargs["expires_in_sec"], 86400) + + @patch(f"{MODULE}.frappe") + @patch(f"{MODULE}.update_last_error") + @patch(f"{MODULE}.require_credential", return_value="test-key") + @patch(f"{MODULE}.requests.post") + def test_cache_hit_skips_api_call(self, mock_post, _cred, _err, mock_frappe): + cached_payload = { + "success": True, + "suggestions": [{"place_id": "p1", "text": "Lisbon", "primary_type": "locality", "types": []}], + "cached": False, + } + mock_frappe.cache.return_value = self._cache(get_value=cached_payload) + + out = _result(google_places.handle_gplaces_autocomplete(input="lisbon")) + + self.assertTrue(out["success"]) + self.assertTrue(out["cached"]) + self.assertEqual(out["suggestions"][0]["place_id"], "p1") + mock_post.assert_not_called() + + @patch(f"{MODULE}.frappe") + @patch(f"{MODULE}.requests.post") + def test_validation_errors(self, mock_post, mock_frappe): + mock_frappe.cache.return_value = self._cache() + + for kwargs in [ + {}, # missing input + {"input": " "}, # blank input + {"input": "x" * 201}, # over 200 chars + {"input": "x", "latitude": -91, "longitude": 0}, # bad latitude + {"input": "x", "latitude": 0, "longitude": 0, "radius": -5}, # bad radius + {"input": "x", "origin_latitude": "north", "origin_longitude": 0}, # bad origin + ]: + out = _result(google_places.handle_gplaces_autocomplete(**kwargs)) + self.assertFalse(out["success"], f"expected failure for {list(kwargs)}") + self.assertIn("error", out) + mock_post.assert_not_called() + + +class TestNearbySearch(unittest.TestCase): + @patch(f"{MODULE}.update_last_error") + @patch(f"{MODULE}.require_credential", return_value="test-key") + @patch(f"{MODULE}.requests.post") + def test_builds_full_request_body(self, mock_post, _cred, _err): + mock_post.return_value = _mock_response({"places": [SAMPLE_PLACE]}) + + out = _result( + google_places.handle_gplaces_nearby_search( + latitude="-33.86", + longitude="151.20", + radius=2000, + included_types="restaurant, cafe", + excluded_types=["bar"], + included_primary_types="restaurant", + max_result_count="20", + language_code="en", + region_code="au", + rank_preference="distance", + ) + ) + + self.assertTrue(out["success"]) + self.assertEqual(out["places"][0]["place_id"], SAMPLE_PLACE["id"]) + + body = mock_post.call_args.kwargs["json"] + self.assertEqual( + body["locationRestriction"]["circle"], + {"center": {"latitude": -33.86, "longitude": 151.2}, "radius": 2000.0}, + ) + self.assertEqual(body["includedTypes"], ["restaurant", "cafe"]) + self.assertEqual(body["excludedTypes"], ["bar"]) + self.assertEqual(body["includedPrimaryTypes"], ["restaurant"]) + self.assertEqual(body["maxResultCount"], 20) + self.assertEqual(body["rankPreference"], "DISTANCE") + self.assertEqual(body["languageCode"], "en") + + @patch(f"{MODULE}.requests.post") + def test_validation_errors(self, mock_post): + for kwargs in [ + {}, # missing coordinates + {"latitude": 10}, # missing longitude + {"latitude": 10, "longitude": 200}, # longitude out of range + {"latitude": 10, "longitude": 20, "radius": 0}, # radius below min + {"latitude": 10, "longitude": 20, "rank_preference": "BEST"}, # invalid enum + {"latitude": 10, "longitude": 20, "max_result_count": 25}, # out of range + ]: + out = _result(google_places.handle_gplaces_nearby_search(**kwargs)) + self.assertFalse(out["success"], f"expected failure for {kwargs}") + self.assertIn("error", out) + mock_post.assert_not_called() + + +class TestErrorEnvelope(unittest.TestCase): + @patch(f"{MODULE}.update_last_error") + @patch(f"{MODULE}.require_credential", return_value="test-key") + @patch(f"{MODULE}.requests.post") + def test_api_error_includes_status_and_truncated_message(self, mock_post, _cred, mock_err): + mock_post.return_value = _mock_response( + {"error": {"message": "API key not valid. " + "x" * 1000}}, status_code=403 + ) + + out = _result(google_places.handle_gplaces_text_search(query="coffee")) + + self.assertFalse(out["success"]) + self.assertIn("403", out["error"]) + self.assertIn("API key not valid", out["error"]) + self.assertLess(len(out["error"]), 600) + mock_err.assert_called_once() + self.assertEqual(mock_err.call_args.args[0], "google_maps") + + @patch(f"{MODULE}.update_last_error") + @patch(f"{MODULE}.require_credential", return_value="test-key") + @patch(f"{MODULE}.requests.post") + def test_network_exception_returns_error_envelope(self, mock_post, _cred, mock_err): + mock_post.side_effect = ConnectionError("dns failure") + + out = _result(google_places.handle_gplaces_text_search(query="coffee")) + + self.assertFalse(out["success"]) + self.assertIn("dns failure", out["error"]) + mock_err.assert_called_once() + + +class TestRegistry(unittest.TestCase): + def test_all_five_tools_registered(self): + expected = { + "gplaces_text_search": "handle_gplaces_text_search", + "gplaces_place_details": "handle_gplaces_place_details", + "gplaces_place_photo": "handle_gplaces_place_photo", + "gplaces_autocomplete": "handle_gplaces_autocomplete", + "gplaces_nearby_search": "handle_gplaces_nearby_search", + } + by_name = {t["tool_name"]: t for t in GOOGLE_PLACES_TOOLS} + self.assertEqual(set(by_name), set(expected)) + for tool_name, handler in expected.items(): + tool = by_name[tool_name] + self.assertEqual(tool["category"], "Google Places") + self.assertEqual(tool["function_path"], f"huf.ai.tools.google_places.{handler}") + # handler actually exists and is importable + self.assertTrue(callable(getattr(google_places, handler))) + + def test_registered_in_all_integration_tools(self): + from huf.ai.tools._registry import ALL_INTEGRATION_TOOLS + + names = {t["tool_name"] for t in ALL_INTEGRATION_TOOLS} + for tool in GOOGLE_PLACES_TOOLS: + self.assertIn(tool["tool_name"], names) + + +if __name__ == "__main__": + unittest.main() diff --git a/huf/ai/tools/_registry.py b/huf/ai/tools/_registry.py index ddcb42beb..37bb63b76 100644 --- a/huf/ai/tools/_registry.py +++ b/huf/ai/tools/_registry.py @@ -597,6 +597,111 @@ def _action(choices): }, ] +GOOGLE_PLACES_TOOLS = [ + { + "tool_name": "gplaces_text_search", + "description": ( + "Search for places with a free-text query using the Google Places API (New). " + "Supports filters (type, rating, price, open now), location bias/restriction, and pagination. " + "Requires a Google Maps/Places API key in Integration Settings or env (GOOGLE_MAPS_API_KEY, PLACE_API_KEY, GOOGLE_PLACES_API_KEY)." + ), + "function_path": "huf.ai.tools.google_places.handle_gplaces_text_search", + "category": "Google Places", + "parameters": [ + _p("query", required=True, description="Free-text search query, e.g. 'vegan restaurants in Lisbon'"), + _p("language_code", description="Response language, e.g. 'en', 'fr'"), + _p("region_code", description="Region bias as CLDR country code, e.g. 'us', 'in'"), + _p("included_type", description="Restrict to a single place type, e.g. 'restaurant', 'museum'"), + _p("min_rating", type="number", description="Minimum average user rating (0.0-5.0)"), + _p("price_levels", description="CSV of PRICE_LEVEL_FREE, PRICE_LEVEL_INEXPENSIVE, PRICE_LEVEL_MODERATE, PRICE_LEVEL_EXPENSIVE, PRICE_LEVEL_VERY_EXPENSIVE"), + _p("open_now", type="boolean", description="Only return places open at query time"), + _p("rank_preference", description="RELEVANCE (default) or DISTANCE (requires location)"), + _p("latitude", type="number", description="Center latitude for location bias/restriction"), + _p("longitude", type="number", description="Center longitude for location bias/restriction"), + _p("radius", type="number", description="Search radius in metres (default 50000, min 1)"), + _p("strict_location", type="boolean", description="With latitude/longitude: hard-restrict to the circle instead of biasing"), + _p("page_size", type="integer", description="Results per page, 1-20 (default 10)"), + _p("page_token", description="next_page_token from a previous response to fetch the next page"), + ], + }, + { + "tool_name": "gplaces_place_details", + "description": ( + "Get full details for a single place by place_id: contact info, opening hours, " + "photos (resource names for gplaces_place_photo), up to 5 reviews, review summary, " + "accessibility/payment/parking options. Requires a Google Places API key." + ), + "function_path": "huf.ai.tools.google_places.handle_gplaces_place_details", + "category": "Google Places", + "parameters": [ + _p("place_id", required=True, description="Google place ID, e.g. 'ChIJN1t_tDeuEmsRUsoyG83frY4'"), + _p("language_code", description="Response language, e.g. 'en', 'fr'"), + _p("region_code", description="Region code as CLDR country code, e.g. 'us'"), + ], + }, + { + "tool_name": "gplaces_place_photo", + "description": ( + "Resolve a photo resource name (from gplaces_place_details photos[].name) to a usable image URL. " + "Requires a Google Places API key." + ), + "function_path": "huf.ai.tools.google_places.handle_gplaces_place_photo", + "category": "Google Places", + "parameters": [ + _p("photo_name", required=True, description="Photo resource name, e.g. 'places/PLACE_ID/photos/PHOTO_ID'"), + _p("max_height_px", type="integer", description="Max image height in pixels (default 800)"), + _p("max_width_px", type="integer", description="Max image width in pixels (default 800)"), + _p("skip_http_redirect", type="boolean", description="Default true: return JSON photoUri. False: follow the media redirect manually"), + ], + }, + { + "tool_name": "gplaces_autocomplete", + "description": ( + "Place autocomplete suggestions for a partial input (cities, districts, neighborhoods by default; " + "country-level results are filtered out). Results are cached for 24h. " + "Supports location bias/restriction and origin-based distance sorting. Requires a Google Places API key." + ), + "function_path": "huf.ai.tools.google_places.handle_gplaces_autocomplete", + "category": "Google Places", + "parameters": [ + _p("input", required=True, description="Partial place name typed by the user (max 200 chars)"), + _p("included_primary_types", description="CSV of primary types (default: locality,sublocality,administrative_area_level_1,administrative_area_level_2,neighborhood)"), + _p("language_code", description="Response language, e.g. 'en', 'fr'"), + _p("region_code", description="Region bias as CLDR country code, e.g. 'us'"), + _p("session_token", description="Session token for billing/session grouping"), + _p("include_query_predictions", type="boolean", description="Also return query (non-place) predictions"), + _p("latitude", type="number", description="Center latitude for location bias/restriction"), + _p("longitude", type="number", description="Center longitude for location bias/restriction"), + _p("radius", type="number", description="Radius in metres (default 50000, min 1)"), + _p("strict_location", type="boolean", description="With latitude/longitude: hard-restrict to the circle instead of biasing"), + _p("origin_latitude", type="number", description="Origin latitude; adds straight-line distance_meters to suggestions"), + _p("origin_longitude", type="number", description="Origin longitude; adds straight-line distance_meters to suggestions"), + ], + }, + { + "tool_name": "gplaces_nearby_search", + "description": ( + "Search for places near a latitude/longitude using the Google Places API (New). " + "Filter by included/excluded types and rank by popularity or distance. Requires a Google Places API key." + ), + "function_path": "huf.ai.tools.google_places.handle_gplaces_nearby_search", + "category": "Google Places", + "parameters": [ + _p("latitude", type="number", required=True, description="Center latitude"), + _p("longitude", type="number", required=True, description="Center longitude"), + _p("radius", type="number", description="Search radius in metres (default 50000, min 1)"), + _p("included_types", description="CSV of place types to include, e.g. 'restaurant,cafe'"), + _p("excluded_types", description="CSV of place types to exclude"), + _p("included_primary_types", description="CSV of primary types to include"), + _p("excluded_primary_types", description="CSV of primary types to exclude"), + _p("max_result_count", type="integer", description="Max results, 1-20 (default 10)"), + _p("language_code", description="Response language, e.g. 'en', 'fr'"), + _p("region_code", description="Region code as CLDR country code, e.g. 'us'"), + _p("rank_preference", description="POPULARITY (default) or DISTANCE"), + ], + }, +] + GOOGLE_DRIVE_TOOLS = [ { "tool_name": "gdrive_list_files", @@ -672,6 +777,7 @@ def _action(choices): + GOOGLE_SHEETS_TOOLS + GOOGLE_CALENDAR_TOOLS + GOOGLE_MAPS_TOOLS + + GOOGLE_PLACES_TOOLS + GOOGLE_DRIVE_TOOLS + GOOGLE_MEET_TOOLS ) diff --git a/huf/ai/tools/credentials.py b/huf/ai/tools/credentials.py index 55e72791d..80d828834 100644 --- a/huf/ai/tools/credentials.py +++ b/huf/ai/tools/credentials.py @@ -105,6 +105,7 @@ def _get_alt_env_names(service: str, key: str) -> list: "GOOGLE_CLIENT_ID", "GOOGLE_CLIENT_SECRET", "GOOGLE_REFRESH_TOKEN", "GOOGLE_MAPS_API_KEY" ], + "google_maps": ["GOOGLE_MAPS_API_KEY", "PLACE_API_KEY", "GOOGLE_PLACES_API_KEY"], "aws": ["AWS_ACCESS_KEY_ID", "AWS_SECRET_ACCESS_KEY", "AWS_DEFAULT_REGION"], "baidu": ["BAIDU_API_KEY"], "brave": ["BRAVE_API_KEY"], diff --git a/huf/ai/tools/google_places.py b/huf/ai/tools/google_places.py new file mode 100644 index 000000000..d820732b3 --- /dev/null +++ b/huf/ai/tools/google_places.py @@ -0,0 +1,566 @@ +import json + +import frappe +import requests + +from huf.ai.tools.credentials import require_credential, update_last_error + +logger = frappe.logger("huf") + +BASE_URL = "https://places.googleapis.com/v1" +SERVICE_NAME = "google_maps" + +# Truncate raw Google error payloads before surfacing them in tool output +_MAX_ERROR_LENGTH = 500 + +# Practical limit for autocomplete input — place names are never this long +_MAX_INPUT_LENGTH = 200 + +# Cache TTL in seconds (24 hours) — place names are geographically stable data +_CACHE_TTL = 86400 + +# Coordinate range constants +_LATITUDE_MIN, _LATITUDE_MAX = -90.0, 90.0 +_LONGITUDE_MIN, _LONGITUDE_MAX = -180.0, 180.0 +_RADIUS_MIN = 1.0 # metres + +_DEFAULT_RADIUS = 50000.0 + +# Non-country place types covering cities, states, districts and neighborhoods. +# Google Places API (New) allows at most 5 types in includedPrimaryTypes. +_DEFAULT_INCLUDED_PRIMARY_TYPES = [ + "locality", + "sublocality", + "administrative_area_level_1", + "administrative_area_level_2", + "neighborhood", +] + +SEARCH_FIELD_MASK = ( + "places.id,places.displayName,places.formattedAddress,places.primaryType," + "places.types,places.rating,places.userRatingCount,places.priceLevel," + "places.location,places.googleMapsUri,places.businessStatus," + "places.currentOpeningHours.openNow,nextPageToken" +) + +DETAILS_FIELD_MASK = ( + "id,displayName,primaryType,types,rating,userRatingCount,priceLevel," + "location,googleMapsUri,businessStatus,formattedAddress," + "internationalPhoneNumber,websiteUri,regularOpeningHours,photos,reviews," + "reviewSummary,editorialSummary,accessibilityOptions,paymentOptions," + "parkingOptions" +) + + +def _key(): + return require_credential(SERVICE_NAME, "api_key") + + +def _as_bool(value): + """Coerce common truthy/falsy representations to bool.""" + if isinstance(value, bool): + return value + if isinstance(value, int | float): + return bool(value) + if isinstance(value, str): + return value.strip().lower() in ("1", "true", "yes", "on") + return bool(value) + + +def _as_float(value): + try: + return float(value) + except (TypeError, ValueError): + return None + + +def _as_int(value): + try: + return int(float(value)) + except (TypeError, ValueError): + return None + + +def _as_csv(value): + """Accept a comma-separated string or a list and return a clean list of strings.""" + if value is None: + return [] + if isinstance(value, str): + return [part.strip() for part in value.split(",") if part.strip()] + if isinstance(value, list | tuple): + return [str(part).strip() for part in value if str(part).strip()] + return [str(value)] + + +def _request_error(resp): + """Build a truncated error message from a non-OK Google API response.""" + try: + message = resp.json().get("error", {}).get("message") or resp.text + except Exception: + message = resp.text + return f"Google Places API Error ({resp.status_code}): {str(message)[:_MAX_ERROR_LENGTH]}" + + +def _post(endpoint, body, field_mask=None): + """POST to the Places API (New). Returns (data, error_message).""" + headers = { + "Content-Type": "application/json", + "X-Goog-Api-Key": _key(), + } + if field_mask: + headers["X-Goog-FieldMask"] = field_mask + resp = requests.post(f"{BASE_URL}/{endpoint}", headers=headers, json=body, timeout=15) + if not resp.ok: + return None, _request_error(resp) + return resp.json(), None + + +def _build_circle(latitude, longitude, radius): + """Validate coordinates and build a circle dict. Returns (circle, error_message).""" + lat = _as_float(latitude) + lng = _as_float(longitude) + rad = _as_float(radius if radius is not None else _DEFAULT_RADIUS) + if lat is None or lng is None: + return None, "latitude and longitude must be numeric" + if rad is None: + return None, "radius must be numeric" + if not (_LATITUDE_MIN <= lat <= _LATITUDE_MAX): + return None, f"latitude must be between {_LATITUDE_MIN} and {_LATITUDE_MAX}" + if not (_LONGITUDE_MIN <= lng <= _LONGITUDE_MAX): + return None, f"longitude must be between {_LONGITUDE_MIN} and {_LONGITUDE_MAX}" + if rad < _RADIUS_MIN: + return None, f"radius must be at least {_RADIUS_MIN} metre" + return {"center": {"latitude": lat, "longitude": lng}, "radius": rad}, None + + +def _normalize_place(place): + location = place.get("location") or {} + opening_hours = place.get("currentOpeningHours") or {} + return { + "place_id": place.get("id", ""), + "name": (place.get("displayName") or {}).get("text", ""), + "address": place.get("formattedAddress", ""), + "primary_type": place.get("primaryType"), + "types": place.get("types", []), + "rating": place.get("rating"), + "user_rating_count": place.get("userRatingCount"), + "price_level": place.get("priceLevel"), + "latitude": location.get("latitude"), + "longitude": location.get("longitude"), + "google_maps_uri": place.get("googleMapsUri"), + "business_status": place.get("businessStatus"), + "open_now": opening_hours.get("openNow"), + } + + +def _is_country_prediction(place_prediction): + """True if an autocomplete placePrediction is a country-level place.""" + types = place_prediction.get("types") + return place_prediction.get("primaryType") == "country" or ( + isinstance(types, list) and "country" in types + ) + + +def _cache_get(cache_key): + try: + return frappe.cache().get_value(cache_key) + except Exception: + # Cache read failure is non-fatal; proceed to live API call + return None + + +def _cache_set(cache_key, value): + try: + frappe.cache().set_value(cache_key, value, expires_in_sec=_CACHE_TTL) + except Exception: + # Cache write failure is non-fatal + pass + + +def handle_gplaces_text_search(**kwargs): + """Search for places with a free-text query using the Google Places API (New).""" + try: + query = (kwargs.get("query") or "").strip() + if not query: + return json.dumps({"success": False, "error": "query is required"}) + + body = {"textQuery": query} + + if kwargs.get("language_code"): + body["languageCode"] = kwargs["language_code"] + if kwargs.get("region_code"): + body["regionCode"] = kwargs["region_code"] + if kwargs.get("included_type"): + body["includedType"] = kwargs["included_type"] + + if kwargs.get("min_rating") is not None: + min_rating = _as_float(kwargs["min_rating"]) + if min_rating is None or not (0.0 <= min_rating <= 5.0): + return json.dumps({"success": False, "error": "min_rating must be a number between 0 and 5"}) + body["minRating"] = min_rating + + price_levels = _as_csv(kwargs.get("price_levels")) + if price_levels: + body["priceLevels"] = price_levels + + if kwargs.get("open_now") is not None: + body["openNow"] = _as_bool(kwargs["open_now"]) + + if kwargs.get("rank_preference"): + rank = str(kwargs["rank_preference"]).upper() + if rank not in ("RELEVANCE", "DISTANCE"): + return json.dumps( + {"success": False, "error": "rank_preference must be RELEVANCE or DISTANCE"} + ) + body["rankPreference"] = rank + + page_size = kwargs.get("page_size") + if page_size is None: + body["pageSize"] = 10 + else: + page_size = _as_int(page_size) + if page_size is None or not (1 <= page_size <= 20): + return json.dumps( + {"success": False, "error": "page_size must be an integer between 1 and 20"} + ) + body["pageSize"] = page_size + + if kwargs.get("page_token"): + body["pageToken"] = kwargs["page_token"] + + if kwargs.get("latitude") is not None or kwargs.get("longitude") is not None: + circle, error = _build_circle( + kwargs.get("latitude"), kwargs.get("longitude"), kwargs.get("radius") + ) + if error: + return json.dumps({"success": False, "error": error}) + key = "locationRestriction" if _as_bool(kwargs.get("strict_location")) else "locationBias" + body[key] = {"circle": circle} + + data, error = _post("places:searchText", body, SEARCH_FIELD_MASK) + if error: + update_last_error(SERVICE_NAME, error) + return json.dumps({"success": False, "error": error}) + + places = [_normalize_place(p) for p in data.get("places", [])] + return json.dumps({"success": True, "places": places, "next_page_token": data.get("nextPageToken")}) + except Exception as e: + logger.warning(f"Google Places Error (Text Search): {e}") + update_last_error(SERVICE_NAME, str(e)) + return json.dumps({"success": False, "error": str(e)}) + + +def handle_gplaces_place_details(**kwargs): + """Fetch full details for a single place by place_id.""" + try: + place_id = (kwargs.get("place_id") or "").strip() + if not place_id: + return json.dumps({"success": False, "error": "place_id is required"}) + + params = {} + if kwargs.get("language_code"): + params["languageCode"] = kwargs["language_code"] + if kwargs.get("region_code"): + params["regionCode"] = kwargs["region_code"] + + resp = requests.get( + f"{BASE_URL}/places/{place_id}", + headers={"X-Goog-Api-Key": _key(), "X-Goog-FieldMask": DETAILS_FIELD_MASK}, + params=params or None, + timeout=15, + ) + if not resp.ok: + error = _request_error(resp) + update_last_error(SERVICE_NAME, error) + return json.dumps({"success": False, "error": error}) + + p = resp.json() + location = p.get("location") or {} + opening_hours = p.get("regularOpeningHours") or {} + photos = [ + { + "name": photo.get("name"), + "width_px": photo.get("widthPx"), + "height_px": photo.get("heightPx"), + "author": ((photo.get("authorAttributions") or [{}])[0]).get("displayName"), + } + for photo in p.get("photos", []) + ] + reviews = [] + for review in p.get("reviews", [])[:5]: + text = review.get("text") or {} + reviews.append( + { + "review_id": review.get("name"), + "author": (review.get("authorAttribution") or {}).get("displayName"), + "rating": review.get("rating"), + "text": text.get("text"), + "language": text.get("languageCode"), + "publish_time": review.get("publishTime"), + "relative_time": review.get("relativePublishTimeDescription"), + "google_maps_uri": review.get("googleMapsUri"), + } + ) + review_summary = (p.get("reviewSummary") or {}).get("text") or {} + editorial_summary = (p.get("editorialSummary") or {}).get("text") or {} + + return json.dumps( + { + "success": True, + "place": { + "place_id": p.get("id", ""), + "name": (p.get("displayName") or {}).get("text", ""), + "address": p.get("formattedAddress", ""), + "primary_type": p.get("primaryType"), + "types": p.get("types", []), + "rating": p.get("rating"), + "user_rating_count": p.get("userRatingCount"), + "price_level": p.get("priceLevel"), + "latitude": location.get("latitude"), + "longitude": location.get("longitude"), + "google_maps_uri": p.get("googleMapsUri"), + "business_status": p.get("businessStatus"), + "phone": p.get("internationalPhoneNumber"), + "website": p.get("websiteUri"), + "opening_hours": opening_hours.get("weekdayDescriptions", []), + "photos": photos, + "reviews": reviews, + "review_summary": review_summary.get("text"), + "editorial_summary": editorial_summary.get("text"), + "accessibility_options": p.get("accessibilityOptions"), + "payment_options": p.get("paymentOptions"), + "parking_options": p.get("parkingOptions"), + }, + } + ) + except Exception as e: + logger.warning(f"Google Places Error (Place Details): {e}") + update_last_error(SERVICE_NAME, str(e)) + return json.dumps({"success": False, "error": str(e)}) + + +def handle_gplaces_place_photo(**kwargs): + """Resolve a Places photo resource name to a usable photo URL.""" + try: + photo_name = (kwargs.get("photo_name") or "").strip() + if not photo_name: + return json.dumps({"success": False, "error": "photo_name is required"}) + + max_height = _as_int(kwargs.get("max_height_px")) or 800 + max_width = _as_int(kwargs.get("max_width_px")) or 800 + skip_redirect = kwargs.get("skip_http_redirect") + skip_redirect = True if skip_redirect is None else _as_bool(skip_redirect) + + url = f"{BASE_URL}/{photo_name}/media" + headers = {"X-Goog-Api-Key": _key()} + params = {"maxHeightPx": max_height, "maxWidthPx": max_width} + + if skip_redirect: + # skipHttpRedirect returns a JSON body with a pre-signed photoUri + params["skipHttpRedirect"] = "true" + resp = requests.get(url, headers=headers, params=params, timeout=15) + if not resp.ok: + error = _request_error(resp) + update_last_error(SERVICE_NAME, error) + return json.dumps({"success": False, "error": error}) + photo_url = resp.json().get("photoUri") + else: + # Manually capture the redirect to the actual image CDN URL + resp = requests.get(url, headers=headers, params=params, timeout=15, allow_redirects=False) + if resp.status_code in (301, 302, 303, 307, 308): + photo_url = resp.headers.get("Location") + elif resp.ok: + photo_url = resp.url + else: + error = _request_error(resp) + update_last_error(SERVICE_NAME, error) + return json.dumps({"success": False, "error": error}) + + if not photo_url: + error = "Google Places API did not return a photo URL" + update_last_error(SERVICE_NAME, error) + return json.dumps({"success": False, "error": error}) + + return json.dumps({"success": True, "photo_url": photo_url}) + except Exception as e: + logger.warning(f"Google Places Error (Place Photo): {e}") + update_last_error(SERVICE_NAME, str(e)) + return json.dumps({"success": False, "error": str(e)}) + + +def handle_gplaces_autocomplete(**kwargs): + """Fetch place autocomplete suggestions, cached in Redis for 24 hours.""" + try: + input_text = kwargs.get("input") + if not input_text or not isinstance(input_text, str): + return json.dumps({"success": False, "error": "input is required"}) + input_text = input_text.strip() + if not input_text: + return json.dumps({"success": False, "error": "input is required"}) + if len(input_text) > _MAX_INPUT_LENGTH: + return json.dumps( + { + "success": False, + "error": f"input exceeds the maximum length of {_MAX_INPUT_LENGTH} characters", + } + ) + + primary_types = _as_csv(kwargs.get("included_primary_types")) or list(_DEFAULT_INCLUDED_PRIMARY_TYPES) + language_code = kwargs.get("language_code") + region_code = kwargs.get("region_code") + + circle = None + if kwargs.get("latitude") is not None or kwargs.get("longitude") is not None: + circle, error = _build_circle( + kwargs.get("latitude"), kwargs.get("longitude"), kwargs.get("radius") + ) + if error: + return json.dumps({"success": False, "error": error}) + + origin = None + if kwargs.get("origin_latitude") is not None or kwargs.get("origin_longitude") is not None: + origin_lat = _as_float(kwargs.get("origin_latitude")) + origin_lng = _as_float(kwargs.get("origin_longitude")) + if origin_lat is None or origin_lng is None: + return json.dumps( + {"success": False, "error": "origin_latitude and origin_longitude must be numeric"} + ) + if not (_LATITUDE_MIN <= origin_lat <= _LATITUDE_MAX): + return json.dumps( + { + "success": False, + "error": f"origin_latitude must be between {_LATITUDE_MIN} and {_LATITUDE_MAX}", + } + ) + if not (_LONGITUDE_MIN <= origin_lng <= _LONGITUDE_MAX): + return json.dumps( + { + "success": False, + "error": f"origin_longitude must be between {_LONGITUDE_MIN} and {_LONGITUDE_MAX}", + } + ) + origin = {"latitude": origin_lat, "longitude": origin_lng} + + # Cache key covers every input that changes the result set + key_parts = [input_text.lower(), ",".join(primary_types)] + if language_code: + key_parts.append(str(language_code)) + if region_code: + key_parts.append(str(region_code)) + if circle: + key_parts.extend( + [str(circle["center"]["latitude"]), str(circle["center"]["longitude"]), str(circle["radius"])] + ) + cache_key = "place_suggestions::" + "::".join(key_parts) + + cached = _cache_get(cache_key) + if cached is not None: + # Re-apply the country filter in case the cache predates it + if isinstance(cached, dict) and isinstance(cached.get("suggestions"), list): + cached = {**cached, "cached": True} + return json.dumps(cached) + return json.dumps({"success": True, "suggestions": [], "cached": True}) + + body = {"input": input_text, "includedPrimaryTypes": primary_types} + if language_code: + body["languageCode"] = language_code + if region_code: + body["regionCode"] = region_code + if kwargs.get("session_token"): + body["sessionToken"] = kwargs["session_token"] + if kwargs.get("include_query_predictions") is not None: + body["includeQueryPredictions"] = _as_bool(kwargs["include_query_predictions"]) + if circle: + key = "locationRestriction" if _as_bool(kwargs.get("strict_location")) else "locationBias" + body[key] = {"circle": circle} + if origin: + body["origin"] = origin + + data, error = _post("places:autocomplete", body) + if error: + update_last_error(SERVICE_NAME, error) + return json.dumps({"success": False, "error": error}) + + # Defensive post-filter: drop country-level results the API may return + # despite includedPrimaryTypes (e.g., due to API changes). + suggestions = [] + for suggestion in data.get("suggestions", []): + prediction = suggestion.get("placePrediction") if isinstance(suggestion, dict) else None + if not isinstance(prediction, dict) or _is_country_prediction(prediction): + continue + item = { + "place_id": prediction.get("placeId"), + "text": (prediction.get("text") or {}).get("text", ""), + "primary_type": prediction.get("primaryType"), + "types": prediction.get("types", []), + } + if prediction.get("distanceMeters") is not None: + item["distance_meters"] = prediction.get("distanceMeters") + suggestions.append(item) + + result = {"success": True, "suggestions": suggestions, "cached": False} + _cache_set(cache_key, result) + return json.dumps(result) + except Exception as e: + logger.warning(f"Google Places Error (Autocomplete): {e}") + update_last_error(SERVICE_NAME, str(e)) + return json.dumps({"success": False, "error": str(e)}) + + +def handle_gplaces_nearby_search(**kwargs): + """Search for places near a location using the Google Places API (New).""" + try: + if kwargs.get("latitude") is None or kwargs.get("longitude") is None: + return json.dumps({"success": False, "error": "latitude and longitude are required"}) + + circle, error = _build_circle(kwargs.get("latitude"), kwargs.get("longitude"), kwargs.get("radius")) + if error: + return json.dumps({"success": False, "error": error}) + + body = {"locationRestriction": {"circle": circle}} + + for kwarg, field in ( + ("included_types", "includedTypes"), + ("excluded_types", "excludedTypes"), + ("included_primary_types", "includedPrimaryTypes"), + ("excluded_primary_types", "excludedPrimaryTypes"), + ): + values = _as_csv(kwargs.get(kwarg)) + if values: + body[field] = values + + max_results = kwargs.get("max_result_count") + if max_results is None: + body["maxResultCount"] = 10 + else: + max_results = _as_int(max_results) + if max_results is None or not (1 <= max_results <= 20): + return json.dumps( + {"success": False, "error": "max_result_count must be an integer between 1 and 20"} + ) + body["maxResultCount"] = max_results + + if kwargs.get("language_code"): + body["languageCode"] = kwargs["language_code"] + if kwargs.get("region_code"): + body["regionCode"] = kwargs["region_code"] + + if kwargs.get("rank_preference"): + rank = str(kwargs["rank_preference"]).upper() + if rank not in ("POPULARITY", "DISTANCE"): + return json.dumps( + {"success": False, "error": "rank_preference must be POPULARITY or DISTANCE"} + ) + body["rankPreference"] = rank + + data, error = _post("places:searchNearby", body, SEARCH_FIELD_MASK) + if error: + update_last_error(SERVICE_NAME, error) + return json.dumps({"success": False, "error": error}) + + places = [_normalize_place(p) for p in data.get("places", [])] + return json.dumps({"success": True, "places": places}) + except Exception as e: + logger.warning(f"Google Places Error (Nearby Search): {e}") + update_last_error(SERVICE_NAME, str(e)) + return json.dumps({"success": False, "error": str(e)})