"""DDGS class implementation.""" import logging import os from concurrent.futures import ThreadPoolExecutor, wait from math import ceil from random import random, shuffle from types import TracebackType from typing import Any, ClassVar from .base import BaseSearchEngine from .engines import ENGINES from .exceptions import DDGSException, TimeoutException from .http_client import HttpClient from .results import ResultsAggregator from .similarity import SimpleFilterRanker from .utils import _expand_proxy_tb_alias logger = logging.getLogger(__name__) class DDGS: """DDGS | Dux Distributed Global Search. A metasearch library that aggregates results from diverse web search services. Args: proxy: The proxy to use for the search. Defaults to None. timeout: The timeout for the search. Defaults to 5. verify: bool (True to verify, False to skip) or str path to a PEM file. Defaults to True. Attributes: threads: The maximum number of threads per search. Defaults to None (automatic, based on max_results). Raises: DDGSException: If an error occurs during the search. Example: >>> from ddgs import DDGS >>> results = DDGS().search("python") """ threads: ClassVar[int | None] = None def __init__( self, proxy: str | None = None, timeout: int | None = 5, *, verify: bool | str = True, ) -> None: self._proxy = _expand_proxy_tb_alias(proxy) or os.environ.get("DDGS_PROXY") self._timeout = timeout self._verify = verify self._engines_cache: dict[ type[BaseSearchEngine[Any]], BaseSearchEngine[Any] ] = {} # dict[engine_class, engine_instance] def __enter__(self) -> "DDGS": # noqa: PYI034 """Enter the context manager and return the DDGS instance.""" return self def __exit__( self, exc_type: type[BaseException] | None = None, exc_val: BaseException | None = None, exc_tb: TracebackType | None = None, ) -> None: """Exit the context manager.""" def _get_engines( self, category: str, backend: str, ) -> list[BaseSearchEngine[Any]]: """Retrieve a list of search engine instances for a given category and backend. Args: category: The category of search engines (e.g., 'text', 'images', etc.). backend: A single or comma-delimited backends. Defaults to "auto". Returns: A list of initialized search engine instances corresponding to the specified category and backend. Instances are cached for reuse. """ if isinstance(backend, list): # deprecated backend = ",".join(backend) backend_list = [x.strip() for x in backend.split(",")] engine_keys = list(ENGINES[category].keys()) shuffle(engine_keys) if "auto" in backend_list or "all" in backend_list: keys = engine_keys if category == "text": keys = ["wikipedia", "grokipedia"] + [k for k in keys if k not in ("wikipedia", "grokipedia")] else: keys = backend_list engine_classes = [] invalid_keys = [] for key in keys: if engine_class := ENGINES[category].get(key): engine_classes.append(engine_class) else: invalid_keys.append(key) if invalid_keys: logger.warning( "%s - backends do not exist or are disabled. Available: %s", ", ".join(sorted(invalid_keys)), ", ".join(sorted(engine_keys)), ) # Initialize and cache engine instances instances = [] for engine_class in engine_classes: # If already cached, use the cached instance if engine_class in self._engines_cache: instances.append(self._engines_cache[engine_class]) # If not cached, create a new instance else: engine_instance = engine_class(proxy=self._proxy, timeout=self._timeout, verify=self._verify) self._engines_cache[engine_class] = engine_instance instances.append(engine_instance) if not instances: logger.warning("backend is not set. Using 'auto'") return self._get_engines(category, "auto") # sorting by `engine.priority` instances.sort(key=lambda e: (e.priority, random), reverse=True) return instances def _search_sync( # noqa: C901, PLR0912 self, category: str, query: str, keywords: str | None = None, *, region: str = "us-en", safesearch: str = "moderate", timelimit: str | None = None, max_results: int | None = 10, page: int = 1, backend: str = "auto", **kwargs: str, ) -> list[dict[str, Any]]: """Perform a search across engines in the given category. Args: category: The category of search engines (e.g., 'text', 'images', etc.). query: The search query. keywords: Deprecated alias for `query`. region: The region to use for the search (e.g., us-en, uk-en, ru-ru, etc.). safesearch: The safesearch setting (e.g., on, moderate, off). timelimit: The timelimit for the search (e.g., d, w, m, y) or custom date range. max_results: The maximum number of results to return. Defaults to 10. page: The page of results to return. Defaults to 1. backend: A single or comma-delimited backends. Defaults to "auto". **kwargs: Additional keyword arguments to pass to the search engines. Returns: A list of dictionaries containing the search results. """ query = keywords or query if not query: msg = "query is mandatory." raise DDGSException(msg) engines = self._get_engines(category, backend) len_unique_providers = len({engine.provider for engine in engines}) seen_providers: set[str] = set() # Perform search results_aggregator: ResultsAggregator[set[str]] = ResultsAggregator({"href", "image", "url", "embed_url"}) max_workers = min(len_unique_providers, ceil(max_results / 10) + 1) if max_results else len_unique_providers if DDGS.threads: max_workers = min(max_workers, DDGS.threads) futures, err = {}, None with ThreadPoolExecutor(max_workers=max_workers, thread_name_prefix="DDGS") as executor: for i, engine in enumerate([*engines, None], start=1): if engine: if engine.provider in seen_providers: continue future = executor.submit( engine.search, query, region=region, safesearch=safesearch, timelimit=timelimit, page=page, **kwargs, ) futures[future] = engine if len(futures) >= max_workers or i >= max_workers: done, not_done = wait(futures, timeout=self._timeout, return_when="FIRST_EXCEPTION") for f, f_engine in futures.items(): if f in done: try: if r := f.result(): results_aggregator.extend(r) seen_providers.add(f_engine.provider) except Exception as ex: # noqa: BLE001 err = ex logger.info("Error in engine %s: %r", f_engine.name, ex) futures = {f: futures[f] for f in not_done} if max_results and len(results_aggregator) >= max_results: break results = results_aggregator.extract_dicts() # Rank results ranker = SimpleFilterRanker() results = ranker.rank(results, query) if results: return results[:max_results] if max_results else results if "timed out" in f"{err}": raise TimeoutException(err) raise DDGSException(err or "No results found.") def text(self, query: str, **kwargs: Any) -> list[dict[str, Any]]: # noqa: ANN401 """Perform a text search.""" return self._search_sync("text", query, **kwargs) def images(self, query: str, **kwargs: Any) -> list[dict[str, Any]]: # noqa: ANN401 """Perform an image search.""" return self._search_sync("images", query, **kwargs) def news(self, query: str, **kwargs: Any) -> list[dict[str, Any]]: # noqa: ANN401 """Perform a news search.""" return self._search_sync("news", query, **kwargs) def videos(self, query: str, **kwargs: Any) -> list[dict[str, Any]]: # noqa: ANN401 """Perform a video search.""" return self._search_sync("videos", query, **kwargs) def books(self, query: str, **kwargs: Any) -> list[dict[str, Any]]: # noqa: ANN401 """Perform a book search.""" return self._search_sync("books", query, **kwargs) def extract(self, url: str, fmt: str = "text_markdown") -> dict[str, str | bytes]: """Fetch a URL and extract its content. Args: url: The URL to fetch and extract content from. fmt: Output format: "text_markdown", "text_plain", "text_rich", "text" (raw HTML), "content" (raw bytes). Returns: A dictionary with 'url' and 'content' keys. """ client = HttpClient(proxy=self._proxy, timeout=self._timeout, verify=self._verify) resp = client.get(url) if resp.status_code != 200: msg = f"Failed to fetch {url}: HTTP {resp.status_code}" raise DDGSException(msg) if fmt == "text_plain": content = resp.text_plain elif fmt == "text_rich": content = resp.text_rich elif fmt == "text": content = resp.text elif fmt == "content": content = resp.content else: content = resp.text_markdown return {"url": url, "content": content}