"""DuckDuckGo search tool implementation."""
import logging
import random
import time
from typing import Any, Dict, List
from ddgs import DDGS
from ..constants import PROXY_URL
def search_ddgs(query: str, max_results: int = 5, retries: int = 3) -> list[dict[str, Any]]:
"""Execute DuckDuckGo search.
This function performs web searches using DuckDuckGo with retry logic,
rate limiting handling, and proxy support.
Args:
query: The search query string.
max_results: Maximum number of results to return (default: 5).
retries: Number of retry attempts on failure (default: 3).
Returns:
A list of search result dictionaries containing:
- title: The result title
- url: The result URL
- snippet: The result body/snippet
Raises:
Exception: On TLS errors or when retries are exhausted.
"""
base_delay = 5
fatal_tls_errors = [
"tls handshake failed",
"unexpected eof",
"client error (connect)",
"ssl",
]
rate_limit_errors = [
"ratelimit",
"429",
"202",
]
network_errors = [
"timeout",
"connection",
"connect error",
]
for attempt in range(retries):
try:
if attempt > 0:
delay = base_delay * (2**attempt) + random.uniform(1, 3)
logging.info(f"Retry wait {delay:.1f} seconds")
time.sleep(delay)
else:
initial_delay = random.uniform(1, 2)
logging.info(f"Initial delay {initial_delay:.1f} seconds")
time.sleep(initial_delay)
logging.info(f"Search query ({attempt + 1}/{retries}): {query}")
ddgs = DDGS(proxy=PROXY_URL)
search_results = list(ddgs.text(query, max_results=max_results))
if not search_results:
logging.info("Search successful but no results")
return []
results = []
for r in search_results:
results.append({
"title": r.get("title", ""),
"url": r.get("href", ""),
"snippet": r.get("body", ""),
})
logging.info(f"Search successful, results: {len(results)}")
return results
except Exception as e:
error_msg = str(e)
error_lower = error_msg.lower()
logging.error(f"Search error: {error_msg}")
if any(k in error_lower for k in fatal_tls_errors):
logging.error("TLS layer failure, possibly blocked")
raise
if any(k in error_lower for k in rate_limit_errors):
if attempt < retries - 1:
wait_time = 15 + random.uniform(5, 10)
logging.warning(f"Rate limit hit, waiting {wait_time:.1f} seconds before retry")
time.sleep(wait_time)
continue
else:
raise Exception("Rate limit still active after multiple retries, please try again later")
retryable = any(k in error_lower for k in network_errors)
if not retryable or attempt == retries - 1:
raise
return []
