Jarian Cottingham
645d8820f1
- Wrap RateLimitMiddleware in Middleware() - bare class broke all requests (500) - Clamp/validate num param - ?num=abc no longer returns 500 - Query length cap (500), result counts clamped to 1..20 - Wire lifespan into Starlette so shared httpx client closes on shutdown - Purge stale per-IP rate limit entries to bound memory growth - URL-encode queries in the Playwright scraper path - Lazy lib imports so the package works without optional playwright - Add pyproject.toml (activates ruff/pytest/bandit in CI), README, LICENSE - Remove committed __pycache__, add .gitignore/.dockerignore - Fix SearXNG healthcheck path (/health -> /healthz) and compose docs
80 lines
2.6 KiB
Python
80 lines
2.6 KiB
Python
import asyncio
|
|
import logging
|
|
from typing import Any, Dict, List
|
|
|
|
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]
|