import asyncio from types import TracebackType from typing import Dict, List, Optional, Type, Union from .webscout_search import WEBS class AsyncWEBS(WEBS): def __init__( self, headers: Optional[Dict[str, str]] = None, proxy: Optional[str] = None, proxies: Union[Dict[str, str], str, None] = None, # deprecated timeout: Optional[int] = 10, ) -> None: """Initialize the AsyncWEBS object. Args: headers (dict, optional): Dictionary of headers for the HTTP client. Defaults to None. proxy (str, optional): proxy for the HTTP client, supports http/https/socks5 protocols. example: "http://user:pass@example.com:3128". Defaults to None. timeout (int, optional): Timeout value for the HTTP client. Defaults to 10. """ super().__init__(headers=headers, proxy=proxy, proxies=proxies, timeout=timeout) self._loop = asyncio.get_running_loop() self._executor = super()._executor async def __aenter__(self) -> "AsyncWEBS": return self async def __aexit__( self, exc_type: Optional[Type[BaseException]], exc_val: Optional[BaseException], exc_tb: Optional[TracebackType], ) -> None: pass async def achat(self, keywords: str, model: str = "gpt-3.5") -> str: """Initiates async chat session with Webscout AI. Args: keywords (str): The initial message or question to send to the AI. model (str): The model to use: "gpt-3.5", "claude-3-haiku". Defaults to "gpt-3.5". Returns: str: The response from the AI. """ result = await self._loop.run_in_executor(self._executor, super().chat, keywords, model) return result async def atext( self, keywords: str, region: str = "wt-wt", safesearch: str = "moderate", timelimit: Optional[str] = None, backend: str = "api", max_results: Optional[int] = None, ) -> List[Dict[str, str]]: """Webscout async text search. Query params: https://duckduckgo.com/params. Args: keywords: keywords for query. region: wt-wt, us-en, uk-en, ru-ru, etc. Defaults to "wt-wt". safesearch: on, moderate, off. Defaults to "moderate". timelimit: d, w, m, y. Defaults to None. backend: api, html, lite. Defaults to api. api - collect data from https://duckduckgo.com, html - collect data from https://html.duckduckgo.com, lite - collect data from https://lite.duckduckgo.com. max_results: max number of results. If None, returns results only from the first response. Defaults to None. Returns: List of dictionaries with search results, or None if there was an error. Raises: DuckDuckGoSearchException: Base exception for duckduckgo_search errors. RatelimitException: Inherits from DuckDuckGoSearchException, raised for exceeding API request rate limits. TimeoutException: Inherits from DuckDuckGoSearchException, raised for API request timeouts. """ result = await self._loop.run_in_executor( self._executor, super().text, keywords, region, safesearch, timelimit, backend, max_results ) return result async def aimages( self, keywords: str, region: str = "wt-wt", safesearch: str = "moderate", timelimit: Optional[str] = None, size: Optional[str] = None, color: Optional[str] = None, type_image: Optional[str] = None, layout: Optional[str] = None, license_image: Optional[str] = None, max_results: Optional[int] = None, ) -> List[Dict[str, str]]: """Webscout async images search. Query params: https://duckduckgo.com/params. Args: keywords: keywords for query. region: wt-wt, us-en, uk-en, ru-ru, etc. Defaults to "wt-wt". safesearch: on, moderate, off. Defaults to "moderate". timelimit: Day, Week, Month, Year. Defaults to None. size: Small, Medium, Large, Wallpaper. Defaults to None. color: color, Monochrome, Red, Orange, Yellow, Green, Blue, Purple, Pink, Brown, Black, Gray, Teal, White. Defaults to None. type_image: photo, clipart, gif, transparent, line. Defaults to None. layout: Square, Tall, Wide. Defaults to None. license_image: any (All Creative Commons), Public (PublicDomain), Share (Free to Share and Use), ShareCommercially (Free to Share and Use Commercially), Modify (Free to Modify, Share, and Use), ModifyCommercially (Free to Modify, Share, and Use Commercially). Defaults to None. max_results: max number of results. If None, returns results only from the first response. Defaults to None. Returns: List of dictionaries with images search results. Raises: DuckDuckGoSearchException: Base exception for duckduckgo_search errors. RatelimitException: Inherits from DuckDuckGoSearchException, raised for exceeding API request rate limits. TimeoutException: Inherits from DuckDuckGoSearchException, raised for API request timeouts. """ result = await self._loop.run_in_executor( self._executor, super().images, keywords, region, safesearch, timelimit, size, color, type_image, layout, license_image, max_results, ) return result async def avideos( self, keywords: str, region: str = "wt-wt", safesearch: str = "moderate", timelimit: Optional[str] = None, resolution: Optional[str] = None, duration: Optional[str] = None, license_videos: Optional[str] = None, max_results: Optional[int] = None, ) -> List[Dict[str, str]]: """Webscout async videos search. Query params: https://duckduckgo.com/params. Args: keywords: keywords for query. region: wt-wt, us-en, uk-en, ru-ru, etc. Defaults to "wt-wt". safesearch: on, moderate, off. Defaults to "moderate". timelimit: d, w, m. Defaults to None. resolution: high, standart. Defaults to None. duration: short, medium, long. Defaults to None. license_videos: creativeCommon, youtube. Defaults to None. max_results: max number of results. If None, returns results only from the first response. Defaults to None. Returns: List of dictionaries with videos search results. Raises: DuckDuckGoSearchException: Base exception for duckduckgo_search errors. RatelimitException: Inherits from DuckDuckGoSearchException, raised for exceeding API request rate limits. TimeoutException: Inherits from DuckDuckGoSearchException, raised for API request timeouts. """ result = await self._loop.run_in_executor( self._executor, super().videos, keywords, region, safesearch, timelimit, resolution, duration, license_videos, max_results, ) return result async def anews( self, keywords: str, region: str = "wt-wt", safesearch: str = "moderate", timelimit: Optional[str] = None, max_results: Optional[int] = None, ) -> List[Dict[str, str]]: """Webscout async news search. Query params: https://duckduckgo.com/params. Args: keywords: keywords for query. region: wt-wt, us-en, uk-en, ru-ru, etc. Defaults to "wt-wt". safesearch: on, moderate, off. Defaults to "moderate". timelimit: d, w, m. Defaults to None. max_results: max number of results. If None, returns results only from the first response. Defaults to None. Returns: List of dictionaries with news search results. Raises: DuckDuckGoSearchException: Base exception for duckduckgo_search errors. RatelimitException: Inherits from DuckDuckGoSearchException, raised for exceeding API request rate limits. TimeoutException: Inherits from DuckDuckGoSearchException, raised for API request timeouts. """ result = await self._loop.run_in_executor( self._executor, super().news, keywords, region, safesearch, timelimit, max_results, ) return result async def aanswers( self, keywords: str, ) -> List[Dict[str, str]]: """Webscout async instant answers. Query params: https://duckduckgo.com/params. Args: keywords: keywords for query, Returns: List of dictionaries with instant answers results. Raises: DuckDuckGoSearchException: Base exception for duckduckgo_search errors. RatelimitException: Inherits from DuckDuckGoSearchException, raised for exceeding API request rate limits. TimeoutException: Inherits from DuckDuckGoSearchException, raised for API request timeouts. """ result = await self._loop.run_in_executor( self._executor, super().answers, keywords, ) return result async def asuggestions( self, keywords: str, region: str = "wt-wt", ) -> List[Dict[str, str]]: """Webscout async suggestions. Query params: https://duckduckgo.com/params. Args: keywords: keywords for query. region: wt-wt, us-en, uk-en, ru-ru, etc. Defaults to "wt-wt". Returns: List of dictionaries with suggestions results. Raises: DuckDuckGoSearchException: Base exception for duckduckgo_search errors. RatelimitException: Inherits from DuckDuckGoSearchException, raised for exceeding API request rate limits. TimeoutException: Inherits from DuckDuckGoSearchException, raised for API request timeouts. """ result = await self._loop.run_in_executor( self._executor, super().suggestions, keywords, region, ) return result async def amaps( self, keywords: str, place: Optional[str] = None, street: Optional[str] = None, city: Optional[str] = None, county: Optional[str] = None, state: Optional[str] = None, country: Optional[str] = None, postalcode: Optional[str] = None, latitude: Optional[str] = None, longitude: Optional[str] = None, radius: int = 0, max_results: Optional[int] = None, ) -> List[Dict[str, str]]: """Webscout async maps search. Query params: https://duckduckgo.com/params. Args: keywords: keywords for query place: if set, the other parameters are not used. Defaults to None. street: house number/street. Defaults to None. city: city of search. Defaults to None. county: county of search. Defaults to None. state: state of search. Defaults to None. country: country of search. Defaults to None. postalcode: postalcode of search. Defaults to None. latitude: geographic coordinate (north-south position). Defaults to None. longitude: geographic coordinate (east-west position); if latitude and longitude are set, the other parameters are not used. Defaults to None. radius: expand the search square by the distance in kilometers. Defaults to 0. max_results: max number of results. If None, returns results only from the first response. Defaults to None. Returns: List of dictionaries with maps search results, or None if there was an error. Raises: DuckDuckGoSearchException: Base exception for duckduckgo_search errors. RatelimitException: Inherits from DuckDuckGoSearchException, raised for exceeding API request rate limits. TimeoutException: Inherits from DuckDuckGoSearchException, raised for API request timeouts. """ result = await self._loop.run_in_executor( self._executor, super().maps, keywords, place, street, city, county, state, country, postalcode, latitude, longitude, radius, max_results, ) return result async def atranslate( self, keywords: Union[List[str], str], from_: Optional[str] = None, to: str = "en", ) -> List[Dict[str, str]]: """Webscout async translate. Args: keywords: string or list of strings to translate. from_: translate from (defaults automatically). Defaults to None. to: what language to translate. Defaults to "en". Returns: List od dictionaries with translated keywords. Raises: DuckDuckGoSearchException: Base exception for duckduckgo_search errors. RatelimitException: Inherits from DuckDuckGoSearchException, raised for exceeding API request rate limits. TimeoutException: Inherits from DuckDuckGoSearchException, raised for API request timeouts. """ result = await self._loop.run_in_executor( self._executor, super().translate, keywords, from_, to, ) return result