|
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,
|
|
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:[email protected]: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 |
