Source code for langchain_community.utilities.google_jobs

"""调用谷歌学术搜索的工具。"""
from typing import Any, Dict, Optional, cast

from langchain_core.pydantic_v1 import BaseModel, Extra, SecretStr, root_validator
from langchain_core.utils import convert_to_secret_str, get_from_dict_or_env


class GoogleJobsAPIWrapper(BaseModel):
    """包装器用于SerpApi的Google Scholar API

    您可以通过在以下网址注册来创建SerpApi.com密钥：https://serpapi.com/users/sign_up。
    该包装器使用SerpApi.com的python包：
    https://serpapi.com/integrations/python
    要使用，您应该设置环境变量``SERPAPI_API_KEY``
    与您的API密钥一起，或将`serp_api_key`作为命名参数传递给构造函数。
     示例：
        .. code-block:: python
        from langchain_community.utilities import GoogleJobsAPIWrapper
        google_Jobs = GoogleJobsAPIWrapper()
        google_Jobs.run('langchain')
    """

    serp_search_engine: Any
    serp_api_key: Optional[SecretStr] = None

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

        extra = Extra.forbid

    @root_validator()
    def validate_environment(cls, values: Dict) -> Dict:
        """验证环境中是否存在API密钥和Python包。"""
        values["serp_api_key"] = convert_to_secret_str(
            get_from_dict_or_env(values, "serp_api_key", "SERPAPI_API_KEY")
        )

        try:
            from serpapi import SerpApiClient

        except ImportError:
            raise ImportError(
                "google-search-results is not installed. "
                "Please install it with `pip install google-search-results"
                ">=2.4.2`"
            )
        serp_search_engine = SerpApiClient
        values["serp_search_engine"] = serp_search_engine

        return values

    def run(self, query: str) -> str:
        """通过Serpapi在Google Trends中运行查询。"""

        # set up query
        serpapi_api_key = cast(SecretStr, self.serp_api_key)
        params = {
            "engine": "google_jobs",
            "api_key": serpapi_api_key.get_secret_value(),
            "q": query,
        }

        total_results = []
        client = self.serp_search_engine(params)
        total_results = client.get_dict()["jobs_results"]

        # extract 1 job info:
        res_str = ""
        for i in range(1):
            job = total_results[i]
            res_str += (
                "\n_______________________________________________"
                + f"\nJob Title: {job['title']}\n"
                + f"Company Name: {job['company_name']}\n"
                + f"Location: {job['location']}\n"
                + f"Description: {job['description']}"
                + "\n_______________________________________________\n"
            )

        return res_str + "\n"