Source code for langchain_community.utilities.google_places_api

"""调用Google Places API 的链。"""

import logging
from typing import Any, Dict, Optional

from langchain_core._api.deprecation import deprecated
from langchain_core.pydantic_v1 import BaseModel, Extra, root_validator
from langchain_core.utils import get_from_dict_or_env


@deprecated(
    since="0.0.33",
    removal="0.3.0",
    alternative_import="langchain_google_community.GooglePlacesAPIWrapper",
)
class GooglePlacesAPIWrapper(BaseModel):
    """封装了Google Places API。

    要使用，您应该安装``googlemaps`` python包，
      **一个用于google地图平台的API密钥** ，
     并设置环境变量''GPLACES_API_KEY''
     为您的API密钥，或将'gplaces_api_key'
     作为构造函数的一个命名参数传递。

    默认情况下，这将返回输入查询的所有结果。
     您可以使用top_k_results参数来限制结果的数量。

    示例:
        .. code-block:: python


            from langchain_community.utilities import GooglePlacesAPIWrapper
            gplaceapi = GooglePlacesAPIWrapper()"""

    gplaces_api_key: Optional[str] = None
    google_map_client: Any  #: :meta private:
    top_k_results: Optional[int] = None

    class Config:
        """此pydantic对象的配置。"""

        extra = Extra.forbid
        arbitrary_types_allowed = True

    @root_validator()
    def validate_environment(cls, values: Dict) -> Dict:
        """验证API密钥是否在您的环境变量中。"""
        gplaces_api_key = get_from_dict_or_env(
            values, "gplaces_api_key", "GPLACES_API_KEY"
        )
        values["gplaces_api_key"] = gplaces_api_key
        try:
            import googlemaps

            values["google_map_client"] = googlemaps.Client(gplaces_api_key)
        except ImportError:
            raise ImportError(
                "Could not import googlemaps python package. "
                "Please install it with `pip install googlemaps`."
            )
        return values

    def run(self, query: str) -> str:
        """运行地点搜索并获取存在的匹配的k个地点。"""
        search_results = self.google_map_client.places(query)["results"]
        num_to_return = len(search_results)

        places = []

        if num_to_return == 0:
            return "Google Places did not find any places that match the description"

        num_to_return = (
            num_to_return
            if self.top_k_results is None
            else min(num_to_return, self.top_k_results)
        )

        for i in range(num_to_return):
            result = search_results[i]
            details = self.fetch_place_details(result["place_id"])

            if details is not None:
                places.append(details)

        return "\n".join([f"{i+1}. {item}" for i, item in enumerate(places)])

    def fetch_place_details(self, place_id: str) -> Optional[str]:
        try:
            place_details = self.google_map_client.place(place_id)
            place_details["place_id"] = place_id
            formatted_details = self.format_place_details(place_details)
            return formatted_details
        except Exception as e:
            logging.error(f"An Error occurred while fetching place details: {e}")
            return None

    def format_place_details(self, place_details: Dict[str, Any]) -> Optional[str]:
        try:
            name = place_details.get("result", {}).get("name", "Unknown")
            address = place_details.get("result", {}).get(
                "formatted_address", "Unknown"
            )
            phone_number = place_details.get("result", {}).get(
                "formatted_phone_number", "Unknown"
            )
            website = place_details.get("result", {}).get("website", "Unknown")
            place_id = place_details.get("result", {}).get("place_id", "Unknown")

            formatted_details = (
                f"{name}\nAddress: {address}\n"
                f"Google place ID: {place_id}\n"
                f"Phone: {phone_number}\nWebsite: {website}\n\n"
            )
            return formatted_details
        except Exception as e:
            logging.error(f"An error occurred while formatting place details: {e}")
            return None