Coverage for src/local_deep_research/web_search_engines/engines/search_engine

1import enum

2import json

3import time

4from typing import Any, Dict, List, Optional

6import requests

7from langchain_core.language_models import BaseLLM

9from ...security import redact_url_for_log

10from ...security.safe_requests import safe_get

12from ..search_engine_base import BaseSearchEngine, Exposure, Sensitivity

13from ...security.secure_logging import logger

16@enum.unique

17class SafeSearchSetting(enum.IntEnum):

18 """

19 Acceptable settings for safe search.

20 """

22 OFF = 0

23 MODERATE = 1

24 STRICT = 2

27class SearXNGSearchEngine(BaseSearchEngine):

28 """

29 SearXNG search engine implementation that requires an instance URL provided via

30 environment variable or configuration. Designed for ethical usage with proper

31 rate limiting and single-instance approach.

32 """

34 # Mark as public search engine

35 is_public = True

36 egress_sensitivity = Sensitivity.NON_SENSITIVE

37 egress_exposure = Exposure.EXPOSING

38 # Mark as generic search engine (general web search)

39 is_generic = True

40 # The egress engine-selection gate uses the static is_public flag above —

41 # SearXNG always queries the internet regardless of where it's hosted, so

42 # a localhost instance_url does NOT reclassify it as private (the PDP's

43 # URL override is fail-up only and never relaxes a public nature).

44 url_setting = "search.engine.web.searxng.default_params.instance_url"

46 @staticmethod

47 def _normalize_list(value):

48 """Ensure *value* is a ``list[str]`` or ``None``.

50 Settings saved via the web UI may arrive as raw JSON strings

51 (e.g. ``'[\\r\\n "general"\\r\\n]'``) instead of parsed lists.

52 This helper decodes such strings so that ``",".join()`` later

53 works on list items rather than individual characters (issue #1030).

54 """

55 if value is None:

56 return None

57 if isinstance(value, list):

58 return value

59 if isinstance(value, str):

60 stripped = value.strip()

61 if stripped:

62 try:

63 parsed = json.loads(stripped)

64 if isinstance(parsed, list):

65 return [str(item) for item in parsed]

66 except (json.JSONDecodeError, ValueError, RecursionError):

67 pass

68 # Comma-separated fallback

69 return [

70 item.strip() for item in stripped.split(",") if item.strip()

71 ]

72 return None

74 def _is_valid_search_result(self, url: str) -> bool:

75 """

76 Check if a parsed result is a valid search result vs an error page.

78 When SearXNG's backend engines fail or get rate-limited, it returns

79 error/stats pages that shouldn't be treated as search results.

81 Returns False for:

82 - Relative URLs (don't start with http:// or https://, case-insensitive)

83 - URLs pointing to the SearXNG instance itself (catches /stats, /preferences, etc.)

84 """

85 # Must have an absolute URL (case-insensitive scheme check)

86 if not url or not url.lower().startswith(("http://", "https://")):

87 return False

89 # Reject URLs pointing back to the SearXNG instance itself

90 # This catches all internal pages like /stats?engine=, /preferences, /about

91 if url.startswith(self.instance_url):

92 return False

94 return True

96 def __init__(

97 self,

98 max_results: int = 15,

99 instance_url: str = "http://localhost:8080",

100 categories: Optional[List[str]] = None,

101 engines: Optional[List[str]] = None,

102 language: str = "en",

103 safe_search: str = SafeSearchSetting.OFF.name,

104 time_range: Optional[str] = None,

105 delay_between_requests: float = 0.0,

106 llm: Optional[BaseLLM] = None,

107 max_filtered_results: Optional[int] = None,

108 include_full_content: bool = True,

109 settings_snapshot: Optional[Dict[str, Any]] = None,

110 **kwargs,

111 ): # API key is actually the instance URL

112 """

113 Initialize the SearXNG search engine with ethical usage patterns.

114

115 Args:

116 max_results: Maximum number of search results

117 instance_url: URL of your SearXNG instance (preferably self-hosted)

118 categories: List of SearXNG categories to search in (general, images, videos, news, etc.)

119 engines: List of engines to use (google, bing, duckduckgo, etc.)

120 language: Language code for search results

121 safe_search: Safe search level (0=off, 1=moderate, 2=strict)

122 time_range: Time range for results (day, week, month, year)

123 delay_between_requests: Seconds to wait between requests

124 llm: Language model for relevance filtering

125 max_filtered_results: Maximum number of results to keep after filtering

126 include_full_content: Whether to include full webpage content in results

127 """

128

129 # Initialize the BaseSearchEngine with LLM, max_filtered_results, and max_results

130 super().__init__(

131 llm=llm,

132 max_filtered_results=max_filtered_results,

133 max_results=max_results,

134 include_full_content=include_full_content,

135 settings_snapshot=settings_snapshot,

136 **kwargs, # Pass through all other kwargs including search_snippets_only

137 )

138

139 # Validate and normalize the instance URL if provided

140 self.instance_url = instance_url.rstrip("/")

141 logger.info(

142 f"SearXNG initialized with instance URL: {redact_url_for_log(self.instance_url)}"

143 )

144 try:

145 # Make sure it's accessible.

146 # allow_private_ips=True since SearXNG is typically self-hosted on local network

147 response = safe_get(

148 self.instance_url, timeout=5, allow_private_ips=True

149 )

150 if response.status_code == 200:

151 logger.info("SearXNG instance is accessible.")

152 self.is_available = True

153 else:

154 self.is_available = False

155 logger.error(

156 f"Failed to access SearXNG instance at {redact_url_for_log(self.instance_url)}. Status code: {response.status_code}"

157 )

158 except (requests.RequestException, ValueError) as e:

159 self.is_available = False

160 safe_msg = self._scrub_error(e)

161 logger.exception(

162 f"Error while trying to access SearXNG instance at {redact_url_for_log(self.instance_url)} ({type(e).__name__}): {safe_msg}"

163 )

164

165 # Add debug logging for all parameters

166 logger.info(

167 f"SearXNG init params: max_results={max_results}, language={language}, "

168 f"max_filtered_results={max_filtered_results}, is_available={self.is_available}"

169 )

170

171 self.max_results = max_results

172 self.categories = self._normalize_list(categories) or ["general"]

173 self.engines = self._normalize_list(engines)

174 self.language = language

175 try:

176 # Handle both string names and integer values

177 if isinstance(safe_search, int) or (

178 isinstance(safe_search, str) and str(safe_search).isdigit()

179 ):

180 self.safe_search = SafeSearchSetting(int(safe_search))

181 else:

182 self.safe_search = SafeSearchSetting[safe_search]

183 except (ValueError, KeyError) as e:

184 safe_msg = self._scrub_error(e)

185 logger.exception(

186 f"'{safe_search}' is not a valid safe search setting ({type(e).__name__}). Disabling safe search: {safe_msg}"

187 )

188 self.safe_search = SafeSearchSetting.OFF

189 self.time_range = time_range

190

191 self.delay_between_requests = float(delay_between_requests)

192

193 if self.is_available:

194 self.search_url = f"{self.instance_url}/search"

195 logger.info(

196 f"SearXNG engine initialized with instance: {redact_url_for_log(self.instance_url)}"

197 )

198 logger.info(

199 f"Rate limiting set to {self.delay_between_requests} seconds between requests"

200 )

201

202 self._init_full_search(

203 web_search=self,

204 language=language,

205 max_results=max_results,

206 region="wt-wt",

207 time_period="y",

208 safe_search=self.safe_search.value,

209 )

210

211 self.last_request_time: float = 0.0

212

213 def _respect_rate_limit(self):

214 """Apply self-imposed rate limiting between requests"""

215 current_time = time.time()

216 time_since_last_request = current_time - self.last_request_time

217

218 if time_since_last_request < self.delay_between_requests:

219 wait_time = self.delay_between_requests - time_since_last_request

220 logger.info(f"Rate limiting: waiting {wait_time:.2f} seconds")

221 time.sleep(wait_time)

222

223 self.last_request_time = time.time()

224

225 def _get_search_results(self, query: str) -> List[Dict[str, Any]]:

226 """

227 Get search results from SearXNG with ethical rate limiting.

228

229 Args:

230 query: The search query

231

232 Returns:

233 List of search results from SearXNG

234 """

235 if not self.is_available:

236 logger.error(

237 "SearXNG engine is disabled (no instance URL provided) - cannot run search"

238 )

239 return []

240

241 logger.info(f"SearXNG running search for query: {query}")

242

243 try:

244 self._respect_rate_limit()

245

246 initial_headers = {

247 "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/120.0.0.0 Safari/537.36",

248 "Accept": "text/html,application/xhtml+xml,application/xml;q=0.9,image/webp,*/*;q=0.8",

249 "Accept-Language": "en-US,en;q=0.9",

250 }

251

252 try:

253 initial_response = safe_get(

254 self.instance_url,

255 headers=initial_headers,

256 timeout=10,

257 allow_private_ips=True,

258 )

259 cookies = initial_response.cookies

260 except Exception as e:

261 safe_msg = self._scrub_error(e)

262 logger.exception(

263 f"Failed to get initial cookies ({type(e).__name__}): {safe_msg}"

264 )

265 cookies = None

266

267 params = {

268 "q": query,

269 "categories": ",".join(self.categories),

270 "language": self.language,

271 "format": "html", # Use HTML format instead of JSON

272 "pageno": 1,

273 "safesearch": self.safe_search.value,

274 "count": self.max_results,

275 }

276

277 if self.engines: 277 ↛ 278line 277 didn't jump to line 278 because the condition on line 277 was never true

278 params["engines"] = ",".join(self.engines)

279

280 if self.time_range: 280 ↛ 281line 280 didn't jump to line 281 because the condition on line 280 was never true

281 params["time_range"] = self.time_range

282

283 # Browser-like headers

284 headers = {

285 "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/120.0.0.0 Safari/537.36",

286 "Accept": "text/html,application/xhtml+xml,application/xml;q=0.9,image/webp,*/*;q=0.8",

287 "Accept-Language": "en-US,en;q=0.9",

288 "Referer": self.instance_url + "/",

289 "Connection": "keep-alive",

290 "Upgrade-Insecure-Requests": "1",

291 }

292

293 logger.info(

294 f"Sending request to SearXNG instance at {redact_url_for_log(self.instance_url)}"

295 )

296 response = safe_get(

297 self.search_url,

298 params=params,

299 headers=headers,

300 cookies=cookies,

301 timeout=15,

302 allow_private_ips=True,

303 )

304

305 if response.status_code == 200:

306 try:

307 from bs4 import BeautifulSoup

308

309 soup = BeautifulSoup(response.text, "html.parser")

310 results = []

311

312 result_elements = soup.select(".result-item")

313

314 if not result_elements:

315 result_elements = soup.select(".result")

316

317 if not result_elements:

318 result_elements = soup.select("article")

319

320 if not result_elements:

321 logger.debug(

322 f"Classes found in HTML: {[c['class'] for c in soup.select('[class]') if 'class' in c.attrs][:10]}"

323 )

324 result_elements = soup.select('div[id^="result"]')

325

326 logger.info(

327 f"Found {len(result_elements)} search result elements"

328 )

329

330 for idx, result_element in enumerate(result_elements):

331 if idx >= self.max_results: 331 ↛ 332line 331 didn't jump to line 332 because the condition on line 331 was never true

332 break

333

334 title_element = (

335 result_element.select_one(".result-title")

336 or result_element.select_one(".title")

337 or result_element.select_one("h3")

338 or result_element.select_one("a[href]")

339 )

340

341 url_element = (

342 result_element.select_one(".result-url")

343 or result_element.select_one(".url")

344 or result_element.select_one("a[href]")

345 )

346

347 content_element = (

348 result_element.select_one(".result-content")

349 or result_element.select_one(".content")

350 or result_element.select_one(".snippet")

351 or result_element.select_one("p")

352 )

353

354 title = (

355 title_element.get_text(" ", strip=True)

356 if title_element

357 else ""

358 )

359

360 url = ""

361 if url_element and url_element.has_attr("href"):

362 url = self._clean_result_url(url_element["href"])

363 elif url_element: 363 ↛ 366line 363 didn't jump to line 366 because the condition on line 363 was always true

364 url = url_element.get_text(strip=True)

365

366 content = (

367 content_element.get_text(" ", strip=True)

368 if content_element

369 else ""

370 )

371

372 if (

373 not url

374 and title_element

375 and title_element.has_attr("href")

376 ):

377 url = self._clean_result_url(title_element["href"])

378

379 logger.debug(

380 f"Extracted result {idx}: title={title[:30]}..., url={redact_url_for_log(url)}, content={content[:30]}..."

381 )

382

383 # Add to results only if it's a valid search result

384 # (not an error page or internal SearXNG page)

385 if self._is_valid_search_result(url):

386 results.append(

387 {

388 "title": title,

389 "url": url,

390 "content": content,

391 "engine": "searxng",

392 "category": "general",

393 }

394 )

395 else:

396 # Check if this is a backend engine failure

397 if url and "/stats?engine=" in url: 397 ↛ 407line 397 didn't jump to line 407 because the condition on line 397 was always true

398 try:

399 engine_name = url.split("/stats?engine=")[

400 1

401 ].split("&")[0]

402 logger.warning(

403 f"SearXNG backend engine failed or rate-limited: {engine_name}"

404 )

405 except (IndexError, AttributeError):

406 pass # Couldn't parse engine name

407 logger.debug(

408 f"Filtered invalid SearXNG result: title={title!r}, url={redact_url_for_log(url)}"

409 )

410

411 if results:

412 logger.info(

413 f"SearXNG returned {len(results)} valid results from HTML parsing"

414 )

415 else:

416 logger.warning(

417 f"SearXNG returned no valid results for query: {query}. "

418 "This may indicate SearXNG backend engine issues or rate limiting."

419 )

420 return results

421

422 except ImportError as e:

423 safe_msg = self._scrub_error(e)

424 logger.exception(

425 f"BeautifulSoup not available for HTML parsing ({type(e).__name__}): {safe_msg}"

426 )

427 return []

428 except Exception as e:

429 safe_msg = self._scrub_error(e)

430 logger.exception(

431 f"Error parsing HTML results ({type(e).__name__}): {safe_msg}"

432 )

433 return []

434 else:

435 logger.error(

436 f"SearXNG returned status code {response.status_code}"

437 )

438 return []

439

440 except Exception as e:

441 safe_msg = self._scrub_error(e)

442 logger.exception(

443 f"Error getting SearXNG results ({type(e).__name__}): {safe_msg}"

444 )

445 return []

446

447 def _get_previews(self, query: str) -> List[Dict[str, Any]]:

448 """

449 Get preview information for SearXNG search results.

450

451 Args:

452 query: The search query

453

454 Returns:

455 List of preview dictionaries

456 """

457 if not self.is_available:

458 logger.warning(

459 "SearXNG engine is disabled (no instance URL provided)"

460 )

461 return []

462

463 logger.info(f"Getting SearXNG previews for query: {query}")

464

465 results = self._get_search_results(query)

466

467 if not results:

468 logger.warning(f"No SearXNG results found for query: {query}")

469 return []

470

471 previews = []

472 for i, result in enumerate(results):

473 title = result.get("title", "")

474 url = self._clean_result_url(result.get("url"))

475 content = result.get("content", "")

476

477 preview = {

478 "id": url or f"searxng-result-{i}",

479 "title": title,

480 "link": url,

481 "snippet": content,

482 "engine": result.get("engine", ""),

483 "category": result.get("category", ""),

484 }

485

486 previews.append(preview)

487

488 return previews

489

490 def _get_full_content(

491 self, relevant_items: List[Dict[str, Any]]

492 ) -> List[Dict[str, Any]]:

493 """

494 Get full content for the relevant search results.

495

496 Args:

497 relevant_items: List of relevant preview dictionaries

498

499 Returns:

500 List of result dictionaries with full content

501 """

502 if not self.is_available:

503 return relevant_items

504

505 if not hasattr(self, "full_search"): 505 ↛ 506line 505 didn't jump to line 506 because the condition on line 505 was never true

506 return relevant_items

507

508 logger.info("Retrieving full webpage content")

509

510 try:

511 return self.full_search._get_full_content(relevant_items)

512

513 except Exception as e:

514 safe_msg = self._scrub_error(e)

515 logger.exception(

516 f"Error retrieving full content ({type(e).__name__}): {safe_msg}"

517 )

518 return relevant_items

519

520 def invoke(self, query: str) -> List[Dict[str, Any]]:

521 """Compatibility method for LangChain tools"""

522 return self.run(query)

523

524 def results(

525 self, query: str, max_results: Optional[int] = None

526 ) -> List[Dict[str, Any]]:

527 """

528 Get search results in a format compatible with other search engines.

529

530 Args:

531 query: The search query

532 max_results: Optional override for maximum results

533

534 Returns:

535 List of search result dictionaries

536 """

537 if not self.is_available:

538 return []

539

540 original_max_results = self.max_results

541

542 try:

543 if max_results is not None:

544 self.max_results = max_results

545

546 results = self._get_search_results(query)

547

548 formatted_results = []

549 for result in results:

550 formatted_results.append(

551 {

552 "title": result.get("title", ""),

553 "link": self._clean_result_url(result.get("url")),

554 "snippet": result.get("content", ""),

555 }

556 )

557

558 return formatted_results

559

560 finally:

561 self.max_results = original_max_results

562

563 @staticmethod

564 def get_self_hosting_instructions() -> str:

565 """

566 Get instructions for self-hosting a SearXNG instance.

567

568 Returns:

569 String with installation instructions

570 """

571 return """

572# SearXNG Self-Hosting Instructions

573

574The most ethical way to use SearXNG is to host your own instance. Here's how:

575

576## Using Docker (easiest method)

577

5781. Install Docker if you don't have it already

5792. Run these commands:

580

581```bash

582# Pull the SearXNG Docker image

583docker pull searxng/searxng

584

585# Run SearXNG (will be available at http://localhost:8080)

586docker run -d -p 8080:8080 --name searxng searxng/searxng

587```

588

589## Using Docker Compose (recommended for production)

590

5911. Create a file named `docker-compose.yml` with the following content:

592

593```yaml

594version: '3'

595services:

596 searxng:

597 container_name: searxng

598 image: searxng/searxng

599 ports:

600 - "8080:8080"

601 volumes:

602 - ./searxng:/etc/searxng

603 environment:

604 - SEARXNG_BASE_URL=http://localhost:8080/

605 restart: unless-stopped

606```

607

6082. Run with Docker Compose:

609

610```bash

611docker-compose up -d

612```

613

614For more detailed instructions and configuration options, visit:

615https://searxng.github.io/searxng/admin/installation.html

616"""

617

618 def run(

619 self, query: str, research_context: Dict[str, Any] | None = None

620 ) -> List[Dict[str, Any]]:

621 """

622 Override BaseSearchEngine run method to add SearXNG-specific error handling.

623 """

624 if not self.is_available:

625 logger.error(

626 "SearXNG run method called but engine is not available (missing instance URL)"

627 )

628 return []

629

630 logger.info(

631 f"SearXNG instance URL: {redact_url_for_log(self.instance_url)}"

632 )

633

634 try:

635 # Call the parent class's run method

636 results = super().run(query, research_context=research_context)

637 logger.info(f"SearXNG search completed with {len(results)} results")

638 return results

639 except Exception as e:

640 safe_msg = self._scrub_error(e)

641 logger.exception(

642 f"Error in SearXNG run method ({type(e).__name__}): {safe_msg}"

643 )

644 # Return empty results on error

645 return []

Coverage for src/local_deep_research/web_search_engines/engines/search_engine_searxng.py: 95%

228 statements