import asyncio import logging from typing import List, Dict, Any, Optional from lib.playwright_manager import PlaywrightManager from lib.rate_limiter import RateLimiter logger = logging.getLogger(__name__) DDG_URL = "https://duckduckgo.com/html/" class DuckDuckGoSearch: """ Search DuckDuckGo using Playwright headless browser. Falls back to lightweight HTTP mode if available. """ def __init__(self, min_interval: float = 3.0): self.rate_limiter = RateLimiter(min_interval_seconds=min_interval) async def search( self, query: str, num_results: int = 10 ) -> List[Dict[str, Any]]: await self.rate_limiter.acquire() page = None try: page = await PlaywrightManager.get_page() logger.info(f"Navigating to DuckDuckGo search: {query}") await page.goto(DDG_URL, wait_until="domcontentloaded", timeout=30000) await page.fill('input[name="q"]', query) await page.keyboard.press("Enter") await page.wait_for_selector( "#rweb-results .result", state="attached", timeout=15000 ) await asyncio.sleep(1) results = await self._parse_page(page, num_results) logger.info( f"DuckDuckGo search returned {len(results)} results for: {query}" ) return results except Exception as e: logger.error(f"DuckDuckGo search failed for '{query}': {e}") raise RuntimeError(f"DuckDuckGo search failed: {e}") from e finally: if page: await PlaywrightManager.close_page(page) async def _parse_page( self, page, max_results: int ) -> List[Dict[str, Any]]: results = await page.evaluate( """() => { const items = document.querySelectorAll('#rweb-results .result'); const results = []; for (const item of items) { const a = item.querySelector('.result__a'); const h2 = item.querySelector('.result__title'); const snippetEl = item.querySelector('.result__snippet'); if (!a || !h2) continue; const url = a.href || ''; const title = h2.textContent?.trim() || ''; const snippet = snippetEl?.textContent?.trim() || ''; if (title) { results.push({ title, url, snippet }); } } return results; }""" ) return results[:max_results]