Source code for langchain_community.utilities.asknews

"""调用AskNews API的工具。"""

from __future__ import annotations

from datetime import datetime, timedelta
from typing import Any, Dict, Optional

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


class AskNewsAPIWrapper(BaseModel):
    """AskNews API的包装器。"""

    asknews_sync: Any  #: :meta private:
    asknews_async: Any  #: :meta private:
    asknews_client_id: Optional[str] = None
    """AskNews API的客户端ID。"""
    asknews_client_secret: Optional[str] = None
    """AskNews API的客户端密钥。"""

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

        extra = Extra.forbid

    @root_validator()
    def validate_environment(cls, values: Dict) -> Dict:
        """验证API凭据和Python包是否存在于环境中。"""

        asknews_client_id = get_from_dict_or_env(
            values, "asknews_client_id", "ASKNEWS_CLIENT_ID"
        )
        asknews_client_secret = get_from_dict_or_env(
            values, "asknews_client_secret", "ASKNEWS_CLIENT_SECRET"
        )

        try:
            import asknews_sdk

        except ImportError:
            raise ImportError(
                "AskNews python package not found. "
                "Please install it with `pip install asknews`."
            )

        an_sync = asknews_sdk.AskNewsSDK(
            client_id=asknews_client_id,
            client_secret=asknews_client_secret,
            scopes=["news"],
        )
        an_async = asknews_sdk.AsyncAskNewsSDK(
            client_id=asknews_client_id,
            client_secret=asknews_client_secret,
            scopes=["news"],
        )

        values["asknews_sync"] = an_sync
        values["asknews_async"] = an_async
        values["asknews_client_id"] = asknews_client_id
        values["asknews_client_secret"] = asknews_client_secret

        return values

    def search_news(
        self, query: str, max_results: int = 10, hours_back: int = 0
    ) -> str:
        """在AskNews API中同步搜索新闻。"""
        if hours_back > 48:
            method = "kw"
            historical = True
            start = int((datetime.now() - timedelta(hours=hours_back)).timestamp())
            stop = int(datetime.now().timestamp())
        else:
            historical = False
            method = "nl"
            start = None
            stop = None

        response = self.asknews_sync.news.search_news(
            query=query,
            n_articles=max_results,
            method=method,
            historical=historical,
            start_timestamp=start,
            end_timestamp=stop,
            return_type="string",
        )
        return response.as_string

    async def asearch_news(
        self, query: str, max_results: int = 10, hours_back: int = 0
    ) -> str:
        """在AskNews API中异步搜索新闻。"""
        if hours_back > 48:
            method = "kw"
            historical = True
            start = int((datetime.now() - timedelta(hours=hours_back)).timestamp())
            stop = int(datetime.now().timestamp())
        else:
            historical = False
            method = "nl"
            start = None
            stop = None

        response = await self.asknews_async.news.search_news(
            query=query,
            n_articles=max_results,
            method=method,
            historical=historical,
            start_timestamp=start,
            end_timestamp=stop,
            return_type="string",
        )
        return response.as_string