Coverage for src/local_deep_research/research_library/downloaders/extraction/pipeline.py: 92%

1""" 

2Shared extraction pipeline. 

3 

4Two entry points: 

5 extract_content(html) — HTML string in, clean text out. 

6 fetch_and_extract(url) — URL in, clean text out (static + JS fallback). 

7 

8This is the single source of truth for content extraction in the project. 

9Used by HTMLDownloader, ContentFetcher, FullSearchResults, WaybackSearchEngine, 

10and any other code that needs clean text from a web page. 

11 

12Academic URLs (arXiv, PubMed, bioRxiv, etc.) are automatically routed to 

13specialized downloaders first. Generic HTML runs only when that route permits it. 

14""" 

15 

16from importlib import import_module 

17from typing import Any, Callable, Dict, List, NamedTuple, Optional 

18 

19from bs4 import BeautifulSoup, NavigableString 

20from loguru import logger 

21 

22from local_deep_research.utilities.arxiv import is_arxiv_paper_url 

23 

24from .trafilatura_extractor import TrafilaturaExtractor 

25from .readability_extractor import ReadabilityExtractor 

26from .justext_extractor import JustextExtractor 

27from .newspaper_extractor import NewspaperExtractor 

28from .metadata_extractor import extract_metadata, metadata_to_text 

29from ....security.log_sanitizer import failure_reason_for_log 

30from ....security.ssrf_validator import redact_url_for_log 

31from ....security.egress.policy import PolicyDeniedError 

32from ....utilities.lxml_thread_safety import install_per_thread_html_parsers 

33 

34# Route newspaper4k's and readabilipy's parser-less lxml.html.fromstring calls 

35# onto per-thread parsers. Without this, worker threads extracting different 

36# pages concurrently share one libxml2 xmlDict and corrupt it, aborting the 

37# whole process with "double free or corruption (out)". Installed at import so 

38# it is in place before any worker thread parses. 

39install_per_thread_html_parsers() 

40 

41 

42# Elements a browser lays out on a line of their own, plus the few that are not 

43# laid out at all but still read as a line when a document is flattened 

44# (`title`, `option`). The markup is the only place that boundary exists, so the 

45# flattener marks the end of each one. 

46_BLOCK_LEVEL_TAGS = ( 

47 "address", 

48 "article", 

49 "aside", 

50 "blockquote", 

51 "caption", 

52 "dd", 

53 "details", 

54 "div", 

55 "dl", 

56 "dt", 

57 "fieldset", 

58 "figcaption", 

59 "figure", 

60 "footer", 

61 "h1", 

62 "h2", 

63 "h3", 

64 "h4", 

65 "h5", 

66 "h6", 

67 "header", 

68 "hgroup", 

69 "hr", 

70 "legend", 

71 "li", 

72 "main", 

73 "nav", 

74 "ol", 

75 "option", 

76 "p", 

77 "pre", 

78 "section", 

79 "summary", 

80 "table", 

81 "td", 

82 "th", 

83 "title", 

84 "tr", 

85 "ul", 

86) 

87 

88 

89_END_OF_CHILDREN = object() 

90 

91 

92def _flatten_html(soup: BeautifulSoup) -> str: 

93 """Flatten parsed HTML to text the way a reader sees it. 

94 

95 ``get_text(separator="\n")`` puts a newline between *every* pair of strings, 

96 including the ones inside a sentence, so ``Hello <b>world</b>! See 

97 <a href="#">this</a>.`` arrives as six lines. A line break belongs only at 

98 the start and end of a block-level element and at a ``<br>``, so those are 

99 the only places one is written. 

100 

101 The tree is walked once and left unchanged. Replacing each ``<br>`` in 

102 place scans its siblings on every call, which is quadratic on a page made 

103 of tens of thousands of them. 

104 """ 

105 # The strings get_text() joins: script, style and template text stays out 

106 # exactly as it does there. 

107 wanted = {id(string) for string in soup.strings} 

108 parts: List[str] = [] 

109 children = [iter(soup.contents)] 

110 open_tags: List[Optional[str]] = [None] 

111 while children: 

112 node = next(children[-1], _END_OF_CHILDREN) 

113 if node is _END_OF_CHILDREN: 

114 children.pop() 

115 if open_tags.pop() in _BLOCK_LEVEL_TAGS: 

116 parts.append("\n") 

117 elif isinstance(node, NavigableString): 

118 if id(node) in wanted: 

119 parts.append(node) 

120 elif node.name == "br": 

121 parts.append("\n") 

122 else: 

123 if node.name in _BLOCK_LEVEL_TAGS: 

124 parts.append("\n") 

125 children.append(iter(node.contents)) 

126 open_tags.append(node.name) 

127 

128 lines = [line.strip() for line in "".join(parts).splitlines()] 

129 return "\n".join(line for line in lines if line) 

130 

131 

132# --- Pipeline thresholds --- 

133# Minimum extracted text length to consider extraction successful 

134MIN_CONTENT_LENGTH = 50 

135# If content is shorter than this, enrich with structured metadata 

136# (JSON-LD, OpenGraph) — helps product pages, JS-heavy sites 

137METADATA_ENRICHMENT_THRESHOLD = 1000 

138# Boilerplate penalty per keyword when scoring extraction quality 

139BOILERPLATE_PENALTY = 500 

140# If an extractor discards more than this fraction of the previous 

141# extractor's output, skip it (protects non-English content) 

142SAFETY_DISCARD_RATIO = 0.2 

143 

144# Module-level singleton extractors (avoid re-creating per call) 

145_trafilatura = TrafilaturaExtractor() 

146_readability = ReadabilityExtractor() 

147_justext_en = JustextExtractor(language="English") 

148_newspaper = NewspaperExtractor() 

149 

150# Boilerplate keywords — used to penalize low-quality extraction 

151_BOILERPLATE_KEYWORDS = [ 

152 "cookie", 

153 "sign up", 

154 "newsletter", 

155 "subscribe", 

156 "accept all", 

157 "privacy policy", 

158 "terms of service", 

159] 

160 

161 

162def _run_extractors_parallel( 

163 html: str, url: str 

164) -> tuple[str | None, str | None]: 

165 """Run trafilatura and newspaper4k sequentially. 

166 

167 Both extractors call into lxml's C extension, which is not safe to 

168 share across threads — running them in a ThreadPoolExecutor caused 

169 Fatal Python error: Aborted during pool teardown on Python 3.14 

170 (the workers would deadlock in shutdown's join). Serializing the 

171 calls eliminates the crash; the perf cost is one extra extraction's 

172 worth of CPU per page, which is acceptable. 

173 

174 Correction to that rationale, from a later SIGABRT (core dump showed 

175 ``double free or corruption (out)`` in ``htmlParseDocument``): sharing an 

176 lxml *parser* across threads is in fact supported — lxml serializes it with 

177 a per-parser lock. The real hazard is the libxml2 ``xmlDict`` that threads 

178 adopt from a shared parser and then intern into concurrently. So 

179 serializing here was never the actual fix, and it could not have been: 

180 it only covers two extractors within ONE page, while the corruption comes 

181 from different pages extracted in different threads. The real fix is 

182 per-thread parsers — see ``utilities.lxml_thread_safety``, installed at 

183 this module's import. This sequencing is retained because it is harmless 

184 and the Python 3.14 teardown deadlock above was a separate problem. 

185 

186 The function name is preserved for backwards compatibility with 

187 any callers/tests that import it directly. 

188 """ 

189 try: 

190 trafilatura_content = _trafilatura.extract(html) 

191 except Exception: 

192 logger.debug("Pipeline: trafilatura raised an exception") 

193 trafilatura_content = None 

194 

195 try: 

196 newspaper_content = _newspaper.extract(html, url) 

197 except Exception: 

198 logger.debug("Pipeline: newspaper4k raised an exception") 

199 newspaper_content = None 

200 

201 return trafilatura_content, newspaper_content 

202 

203 

204def _count_boilerplate(text: str) -> int: 

205 """Count boilerplate keyword occurrences in text.""" 

206 if not text: 

207 return 0 

208 lower = text.lower() 

209 return sum(1 for kw in _BOILERPLATE_KEYWORDS if kw in lower) 

210 

211 

212def _quality_score(text: str) -> int: 

213 """Score extraction quality: length minus boilerplate penalty.""" 

214 if not text: 

215 return 0 

216 return len(text) - (_count_boilerplate(text) * BOILERPLATE_PENALTY) 

217 

218 

219def extract_content( 

220 html: str, 

221 language: str = "English", 

222 min_length: int = MIN_CONTENT_LENGTH, 

223 url: str = "", 

224) -> Optional[str]: 

225 """Extract clean text content from HTML. 

226 

227 Pipeline: 

228 1. trafilatura (primary — best benchmarks, multilingual, markdown) 

229 2. newspaper4k (parallel — strong on news/forum pages) 

230 → pick the higher-quality result from steps 1-2 

231 3. readability → justext (fallback if both above fail, with 80% safety) 

232 4. soup.get_text() (last resort) 

233 

234 Args: 

235 html: Raw HTML string. 

236 language: Language for justext stoplist (fallback only). 

237 min_length: Minimum content length to accept. 

238 url: Source URL (improves newspaper4k extraction accuracy). 

239 

240 Returns: 

241 Extracted plain text, or None if content is below min_length. 

242 """ 

243 if not html or not html.strip(): 

244 return None 

245 

246 # Run trafilatura and newspaper4k, pick the better result. newspaper4k is 

247 # strong on news front pages and multi-answer threads where trafilatura 

248 # sometimes extracts less content. 

249 # No time bound here: the extractors run sequentially in-process (see 

250 # _run_extractors_parallel — the ThreadPoolExecutor was removed). 

251 trafilatura_content, newspaper_content = _run_extractors_parallel(html, url) 

252 

253 traf_score = _quality_score(trafilatura_content) 

254 np_score = _quality_score(newspaper_content) 

255 

256 if traf_score >= np_score and trafilatura_content: 

257 content = trafilatura_content 

258 winner = "trafilatura" 

259 elif newspaper_content: 

260 content = newspaper_content 

261 winner = "newspaper4k" 

262 else: 

263 content = trafilatura_content 

264 winner = "trafilatura" 

265 

266 if content and len(content.strip()) >= min_length: 

267 logger.debug( 

268 f"Pipeline: {winner} extracted {len(content)} chars" 

269 + ( 

270 f" (traf={len(trafilatura_content or '')}, " 

271 f"np4k={len(newspaper_content or '')})" 

272 if newspaper_content and trafilatura_content 

273 else "" 

274 ) 

275 ) 

276 else: 

277 # Fallback: readability → justext 

278 logger.debug( 

279 "Pipeline: primary extractors insufficient, using fallback" 

280 ) 

281 

282 soup = BeautifulSoup(html, "html.parser") 

283 for tag_name in [ 

284 "script", 

285 "style", 

286 "iframe", 

287 "noscript", 

288 "svg", 

289 "form", 

290 "button", 

291 "input", 

292 "select", 

293 "textarea", 

294 ]: 

295 for tag in soup.find_all(tag_name): 

296 tag.decompose() 

297 cleaned_html = str(soup) 

298 

299 justext_extractor = ( 

300 _justext_en 

301 if language == "English" 

302 else JustextExtractor(language=language) 

303 ) 

304 

305 content = None 

306 prev_text_len = 0 

307 

308 for extractor in [_readability, justext_extractor]: 

309 result = extractor.extract( 

310 cleaned_html if prev_text_len == 0 else content 

311 ) 

312 if result and result.strip(): 

313 result_len = len(result.strip()) 

314 # Safety: skip if extractor discards >80% of content. 

315 # Compare text lengths (strip HTML tags for fair comparison 

316 # since readability returns HTML but justext returns text). 

317 if ( 

318 prev_text_len > 0 

319 and result_len < prev_text_len * SAFETY_DISCARD_RATIO 

320 ): 

321 logger.debug( 

322 f"Pipeline: {extractor.__class__.__name__} discarded " 

323 f">80% of content — skipping" 

324 ) 

325 continue 

326 content = result 

327 # Store text-equivalent length for fair comparison 

328 if "<" in result: 

329 prev_text_len = len( 

330 BeautifulSoup(result, "html.parser").get_text() 

331 ) 

332 else: 

333 prev_text_len = result_len 

334 logger.debug( 

335 f"Pipeline: {extractor.__class__.__name__} " 

336 f"returned {result_len} chars" 

337 ) 

338 

339 # Strip remaining HTML tags (e.g. readability-only mode) 

340 if content and "<" in content: 

341 content = _flatten_html(BeautifulSoup(content, "html.parser")) 

342 

343 # Last resort 

344 if not content or len(content.strip()) < min_length: 

345 logger.debug("Pipeline: all extractors failed, using get_text()") 

346 content = _flatten_html(soup) 

347 

348 if not content or len(content.strip()) < min_length: 

349 return None 

350 

351 # Enrich with structured metadata when text extraction is thin 

352 # (e.g. product pages, JS-heavy sites) 

353 if len(content.strip()) < METADATA_ENRICHMENT_THRESHOLD: 

354 metadata = extract_metadata(html) 

355 supplement = metadata_to_text(metadata) 

356 if supplement and supplement.strip(): 

357 logger.debug( 

358 f"Pipeline: enriching with {len(supplement)} chars " 

359 f"of structured metadata" 

360 ) 

361 content = content.rstrip() + "\n\n" + supplement 

362 

363 return content 

364 

365 

366def extract_content_with_metadata( 

367 html: str, 

368 language: str = "English", 

369 min_length: int = MIN_CONTENT_LENGTH, 

370) -> Optional[Dict[str, Any]]: 

371 """Extract clean text and page metadata from HTML in a single pass. 

372 

373 Combines content extraction (trafilatura/readability/justext pipeline) 

374 with title and description extraction from HTML meta tags. This avoids 

375 the need for callers to do a separate BeautifulSoup parse for metadata. 

376 

377 Args: 

378 html: Raw HTML string. 

379 language: Language for justext stoplist (fallback only). 

380 min_length: Minimum content length to accept. 

381 

382 Returns: 

383 Dict with keys: content, title, description — or None if content 

384 is below min_length. 

385 """ 

386 if not html or not html.strip(): 

387 return None 

388 

389 # Single parse for metadata (title, description, og:*) 

390 soup = BeautifulSoup(html, "html.parser") 

391 

392 title = None 

393 if soup.title and soup.title.string: 

394 title = soup.title.string.strip() 

395 og_title = soup.find("meta", property="og:title") 

396 if og_title and og_title.get("content"): 

397 title = str(og_title["content"]).strip() 

398 

399 description = None 

400 meta_desc = soup.find("meta", attrs={"name": "description"}) 

401 if meta_desc and meta_desc.get("content"): 

402 description = str(meta_desc["content"]).strip() 

403 og_desc = soup.find("meta", property="og:description") 

404 if og_desc and og_desc.get("content"): 

405 description = str(og_desc["content"]).strip() 

406 

407 # Extract content using the shared pipeline 

408 content = extract_content(html, language=language, min_length=min_length) 

409 if not content: 

410 return None 

411 

412 return { 

413 "title": title, 

414 "description": description, 

415 "content": content, 

416 } 

417 

418 

419class _SpecializedResult(NamedTuple): 

420 content: str | None = None 

421 fallback_allowed: bool = True 

422 skip_reason: str | None = None 

423 

424 

425def _try_specialized_downloader( 

426 url: str, 

427 timeout: int = 30, 

428 allow_private_ips: bool = False, 

429 block_link_local: bool = False, 

430 egress_context: Any = None, 

431) -> _SpecializedResult: 

432 """Try a specialized downloader (arXiv, PubMed, etc.) for the URL. 

433 

434 The result carries extracted text, whether generic HTML fallback is safe, 

435 and the specialist's failure reason for observability. 

436 """ 

437 # Terminal ownership follows the paper identifier, not the host: a 

438 # non-paper arXiv page has no canonical downloader source, so denying it 

439 # the generic fallback would return nothing at all. 

440 arxiv_url = is_arxiv_paper_url(url) 

441 failure = _SpecializedResult( 

442 fallback_allowed=not arxiv_url, 

443 skip_reason="URL classifier is unavailable", 

444 ) 

445 try: 

446 from local_deep_research.content_fetcher.url_classifier import ( 

447 URLClassifier, 

448 URLType, 

449 ) 

450 except ImportError: 

451 return failure 

452 

453 url_type = URLClassifier.classify(url) 

454 arxiv_terminal = arxiv_url or url_type == URLType.ARXIV 

455 failure = _SpecializedResult(fallback_allowed=not arxiv_terminal) 

456 

457 # Only academic URL types have specialized downloaders worth trying. 

458 # HTML, DOI, PDF, INVALID fall through to the generic pipeline. 

459 _SPECIALIZED_TYPES = { 

460 URLType.ARXIV, 

461 URLType.PUBMED, 

462 URLType.PMC, 

463 URLType.SEMANTIC_SCHOLAR, 

464 URLType.BIORXIV, 

465 URLType.MEDRXIV, 

466 } 

467 if url_type not in _SPECIALIZED_TYPES: 

468 return failure 

469 

470 from ..base import BaseDownloader, ContentType 

471 

472 # Map URL type to downloader class (lazy imports to avoid circular deps) 

473 downloader: BaseDownloader | None = None 

474 try: 

475 if url_type == URLType.ARXIV: 

476 package = __package__ 

477 if package is None: 477 ↛ 478line 477 didn't jump to line 478 because the condition on line 477 was never true

478 return failure._replace( 

479 skip_reason="Downloader package is unavailable" 

480 ) 

481 downloaders_package = package.rpartition(".")[0] 

482 arxiv_module = import_module(f"{downloaders_package}.arxiv") 

483 downloader_type = getattr(arxiv_module, "ArxivDownloader", None) 

484 if not isinstance(downloader_type, type) or not issubclass( 484 ↛ 487line 484 didn't jump to line 487 because the condition on line 484 was never true

485 downloader_type, BaseDownloader 

486 ): 

487 return failure._replace( 

488 skip_reason="arXiv downloader is unavailable" 

489 ) 

490 downloader = downloader_type(timeout=timeout) 

491 elif url_type in (URLType.PUBMED, URLType.PMC): 491 ↛ 495line 491 didn't jump to line 495 because the condition on line 491 was always true

492 from ..pubmed import PubMedDownloader 

493 

494 downloader = PubMedDownloader(timeout=timeout) 

495 elif url_type == URLType.SEMANTIC_SCHOLAR: 

496 from ..semantic_scholar import SemanticScholarDownloader 

497 

498 downloader = SemanticScholarDownloader(timeout=timeout) 

499 elif url_type in (URLType.BIORXIV, URLType.MEDRXIV): 

500 from ..biorxiv import BioRxivDownloader 

501 

502 downloader = BioRxivDownloader(timeout=timeout) 

503 except ImportError: 

504 logger.debug( 

505 f"Pipeline: specialized downloader not available for {url_type.value}" 

506 ) 

507 return failure._replace( 

508 skip_reason=f"{url_type.value} downloader is unavailable" 

509 ) 

510 

511 if not downloader: 511 ↛ 512line 511 didn't jump to line 512 because the condition on line 511 was never true

512 return failure._replace( 

513 skip_reason=f"{url_type.value} downloader is unavailable" 

514 ) 

515 

516 skip_reason: str | None = None 

517 try: 

518 session = getattr(downloader, "session", None) 

519 if session is not None: 

520 session.allow_private_ips = allow_private_ips 

521 session.block_link_local = block_link_local 

522 session.egress_context = egress_context 

523 result = downloader.download_with_result(url, ContentType.TEXT) 

524 if getattr(result, "content_is_complete", True) is False: 

525 # This pipeline supplies full-content engines. An abstract 

526 # must not be promoted to a complete paper by those callers. 

527 return failure._replace( 

528 skip_reason="Specialized downloader returned only metadata" 

529 ) 

530 if result.is_success and result.content: 

531 text = result.content.decode("utf-8", errors="replace") 

532 if len(text.strip()) >= MIN_CONTENT_LENGTH: 532 ↛ 542line 532 didn't jump to line 542 because the condition on line 532 was always true

533 logger.debug( 

534 f"Pipeline: specialized downloader ({url_type.value}) " 

535 f"returned {len(text)} chars for " 

536 f"{redact_url_for_log(url)}" 

537 ) 

538 return _SpecializedResult( 

539 content=text, 

540 fallback_allowed=not arxiv_terminal, 

541 ) 

542 skip_reason = ( 

543 result.skip_reason 

544 or "Specialized downloader returned no usable content" 

545 ) 

546 except PolicyDeniedError: 

547 # Propagate egress policy denials so callers enforce per-URL policy. 

548 raise 

549 except Exception as e: 

550 skip_reason = "Specialized downloader failed" 

551 logger.opt(exception=False).debug( 

552 f"Pipeline: specialized downloader failed for " 

553 f"{redact_url_for_log(url)} ({type(e).__name__})" 

554 ) 

555 finally: 

556 try: 

557 downloader.close() 

558 except Exception: # noqa: silent-exception 

559 pass 

560 

561 failure = failure._replace(skip_reason=skip_reason) 

562 if failure.fallback_allowed: 

563 logger.debug( 

564 f"Pipeline: specialized downloader ({url_type.value}) returned " 

565 f"no content for {redact_url_for_log(url)} " 

566 f"({failure_reason_for_log(failure.skip_reason)}), falling back to HTML pipeline" 

567 ) 

568 return failure 

569 

570 

571def _is_cache_hit_allowed( 

572 url: str, 

573 allow_private_ips: bool, 

574 block_link_local: bool, 

575 _cache: object | None = None, 

576) -> bool: 

577 """Whether a cache hit for *url* may be served under the caller's grant. 

578 

579 Defense-in-depth for the shared research cache: a granted call 

580 (``allow_private_ips=True``) may fetch and cache a loopback/private URL, 

581 but a later non-granted call sharing the same cache must not be served 

582 that entry — it must go through the downloader's own SSRF gate (which 

583 will refuse). Path-only library routes (``/library/...``, ``/lib/...``) 

584 are local DB reads, not network fetches, so they are always allowed. 

585 

586 The absolute ``https://library.document/<id>`` alias is NOT exempted: 

587 the pipeline has no library resolver, so that spelling is a network URL 

588 here and must pass the same egress check as any web URL (it fails 

589 ``validate_url`` — unresolvable host — and correctly bypasses the cache 

590 instead of leaking a cached entry to a non-granted caller). 

591 

592 The library exemption uses the fetch tool's strict grammar 

593 (32/64-hex or UUID via ``library_resolver.parse_library_url``), not 

594 the wider citation grammar (``is_safe_document_id`` accepts any 

595 ``[0-9A-Za-z_-]+`` slug): a slug-shaped route can never resolve via 

596 the fetch tool, so exempting it here would let the two surfaces drift. 

597 

598 Fail-closed: when the SSRF validator cannot be imported the hit is 

599 refused (callers fall through to the downloader, which enforces its 

600 own gate). When *_cache* is a ``ResearchContentCache`` successful 

601 verdicts are memoized per-run so repeat hits do not pay a getaddrinfo 

602 per read and stay servable when DNS drops mid-run after a success; 

603 refusals are recomputed so a transient DNS blip never pins the URL 

604 for the rest of the run. 

605 """ 

606 try: 

607 from local_deep_research.advanced_search_system.tools.fetch.library_resolver import ( 

608 parse_library_url, 

609 ) 

610 

611 stripped = url.strip() if isinstance(url, str) else "" 

612 if stripped.startswith("/") and parse_library_url(stripped) is not None: 

613 return True 

614 except Exception: # pragma: no cover - defensive 

615 logger.debug("cache grant check: library route parse failed") 

616 try: 

617 from local_deep_research.security.ssrf_validator import validate_url 

618 except Exception: # pragma: no cover - defensive 

619 logger.debug("cache grant check: validator unavailable, refusing hit") 

620 return False 

621 # Per-run memo: same (url, grant) repeats skip DNS on success. A 

622 # False verdict only skips the cache fast path and is never pinned — 

623 # the downloader re-validates — so a transient DNS failure cannot 

624 # block recovery. 

625 _memo = getattr(_cache, "validated", None) 

626 if callable(_memo): 

627 try: 

628 return bool( 

629 _memo( 

630 url, 

631 allow_private_ips, 

632 block_link_local, 

633 lambda: validate_url( 

634 url, 

635 allow_private_ips=allow_private_ips, 

636 block_link_local=block_link_local, 

637 ), 

638 ) 

639 ) 

640 except Exception: # pragma: no cover - defensive 

641 return False 

642 try: 

643 return bool( 

644 validate_url( 

645 url, 

646 allow_private_ips=allow_private_ips, 

647 block_link_local=block_link_local, 

648 ) 

649 ) 

650 except Exception: # pragma: no cover - defensive 

651 return False 

652 

653 

654def fetch_and_extract( 

655 url: str, 

656 timeout: int = 30, 

657 language: str = "English", 

658 enable_js_rendering: bool = False, 

659 content_cache: Any = None, 

660 allow_private_ips: bool = False, 

661 block_link_local: bool = False, 

662 egress_context: Any = None, 

663) -> Optional[str]: 

664 """Fetch a URL and extract clean text content. 

665 

666 Pipeline: 

667 1. Shared research content cache (when bound — the quick-report 

668 phase, detailed subsections and parallel subagents share one 

669 cache for the whole research run, so repeat URLs are served 

670 without a second network fetch). 

671 2. Specialized downloader (arXiv canonical sources, PubMed API, etc.) 

672 3. Static HTTP fetch → Playwright fallback (if JS needed and 

673 permitted and ``enable_js_rendering`` is True) → trafilatura → 

674 readability → justext 

675 

676 Args: 

677 url: The URL to fetch. 

678 timeout: Request timeout in seconds. 

679 language: Language for justext stoplist. 

680 enable_js_rendering: When True, the HTML pipeline falls back to a 

681 headless browser for pages that need JavaScript. Defaults to 

682 False because the default Docker production image ships without 

683 Chromium. Limited internal benchmark comparisons (dev instances 

684 with Chromium vs Docker without) showed no measurable 

685 research-quality improvement from JS rendering, and most regular 

686 benchmark runs are on Docker without Chromium anyway. The 

687 user-facing toggle is ``web.enable_javascript_rendering``. 

688 content_cache: Optional shared ``ResearchContentCache``. When None, 

689 the cache bound to the current research (if any) is used. 

690 allow_private_ips: Whether the downloader may access private IPs 

691 after SSRF validation. The run policy, when supplied, is also 

692 checked on each redirect and pinned connect address. Cloud 

693 metadata remains blocked regardless. 

694 block_link_local: Keep the whole link-local range blocked even when 

695 ``allow_private_ips`` is True (forwarded to the downloader's 

696 static-fetch ``SafeSession``). No effect without the grant. 

697 egress_context: Optional run policy enforced on every transport hop. 

698 

699 Returns: 

700 Extracted plain text, or None if fetch or extraction failed. 

701 """ 

702 # Shared cache with coalescing: repeat URLs (quick phase vs 

703 # subagents, lead vs subagents) are served without a second network 

704 # fetch, and concurrent same-URL calls collapse to one download via 

705 # ``get_or_fetch``. Only successes are cached; failures retry. 

706 try: 

707 from local_deep_research.utilities.research_content_cache import ( 

708 resolve_cache, 

709 ) 

710 

711 _cache = resolve_cache(content_cache) 

712 except Exception: # pragma: no cover - defensive 

713 _cache = None 

714 

715 def _do_fetch() -> dict[str, Any] | None: 

716 # No cache access in here — get_or_fetch owns coalescing. 

717 try: 

718 specialized = _try_specialized_downloader( 

719 url, 

720 timeout=timeout, 

721 allow_private_ips=allow_private_ips, 

722 block_link_local=block_link_local, 

723 egress_context=egress_context, 

724 ) 

725 except PolicyDeniedError: 

726 # Single-URL fetch: propagate policy denial immediately rather than falling back. 

727 raise 

728 except Exception as e: 

729 logger.opt(exception=False).debug( 

730 f"Pipeline: specialized downloader error for " 

731 f"{redact_url_for_log(url)} ({type(e).__name__})" 

732 ) 

733 specialized = _SpecializedResult( 

734 fallback_allowed=not is_arxiv_paper_url(url) 

735 ) 

736 if specialized.content: 

737 return {"content": specialized.content, "url": url} 

738 if not specialized.fallback_allowed: 

739 logger.warning( 

740 "Pipeline: terminal specialized download failed for {}: {}", 

741 redact_url_for_log(url), 

742 failure_reason_for_log( 

743 specialized.skip_reason or "reason unavailable" 

744 ), 

745 ) 

746 return None 

747 from ..playwright_html import AutoHTMLDownloader 

748 

749 downloader = AutoHTMLDownloader( 

750 timeout=timeout, 

751 language=language, 

752 enable_js_rendering=enable_js_rendering, 

753 allow_private_ips=allow_private_ips, 

754 block_link_local=block_link_local, 

755 ) 

756 downloader.session.egress_context = egress_context 

757 try: 

758 # download() returns extracted text as UTF-8 bytes (not raw 

759 # HTML): AutoHTMLDownloader inherits HTMLDownloader.download() 

760 # which runs _fetch_html() → _extract_content(). 

761 result = downloader.download(url) 

762 if result: 

763 return { 

764 "content": result.decode("utf-8", errors="replace"), 

765 "url": url, 

766 } 

767 return None 

768 except PolicyDeniedError: 

769 # Single-URL fetch: propagate policy denial directly to caller. 

770 raise 

771 except Exception as e: 

772 # renders them with any traceback even when diagnose=False. 

773 logger.opt(exception=False).error( 

774 f"fetch_and_extract failed for {redact_url_for_log(url)} " 

775 f"({type(e).__name__})" 

776 ) 

777 return None 

778 finally: 

779 try: 

780 downloader.close() 

781 except Exception: 

782 logger.debug("Failed to close downloader in fetch_and_extract") 

783 

784 if _cache is None: 

785 outcome = _do_fetch() 

786 if outcome is not None and isinstance(outcome.get("content"), str): 

787 return outcome["content"] or None 

788 return None 

789 if not _is_cache_hit_allowed( 

790 url, allow_private_ips, block_link_local, _cache 

791 ): 

792 # Grant-agnostic cache must not serve private content to a 

793 # non-granted caller: bypass the cache so the downloader's own 

794 # SSRF gate refuses (no hit, no fill). 

795 outcome = _do_fetch() 

796 if outcome is not None and isinstance(outcome.get("content"), str): 796 ↛ 797line 796 didn't jump to line 797 because the condition on line 796 was never true

797 return outcome["content"] or None 

798 return None 

799 try: 

800 cached = _cache.get_or_fetch(url, _do_fetch) 

801 except PolicyDeniedError: 

802 # Single-URL fetch: propagate policy denial directly to caller. 

803 raise 

804 except Exception as e: 

805 # Type name only -- exception texts embed the raw URL and loguru 

806 # renders them with any traceback even when diagnose=False. 

807 logger.opt(exception=False).error( 

808 f"fetch_and_extract failed for {redact_url_for_log(url)} " 

809 f"({type(e).__name__})" 

810 ) 

811 return None 

812 if cached is not None and isinstance(cached.get("content"), str): 

813 return cached["content"] or None 

814 return None 

815 

816 

817def batch_fetch_and_extract( 

818 urls: List[str], 

819 timeout: int = 30, 

820 language: str = "English", 

821 enable_js_rendering: bool = False, 

822 content_cache: Any = None, 

823 allow_private_ips: bool = False, 

824 block_link_local: bool = False, 

825 progress_callback: Optional[Callable[[int, int], None]] = None, 

826 egress_context: Any = None, 

827) -> Dict[str, Optional[str]]: 

828 """Fetch multiple URLs and extract clean text from each. 

829 

830 For each URL: 

831 1. Shared research content cache (single cache for the whole 

832 research run — quick phase, subsections and subagents). 

833 2. Try specialized downloader (arXiv, PubMed, etc.) if URL matches 

834 3. Use generic HTML only when the specialized route permits fallback 

835 

836 Uses a single AutoHTMLDownloader (and thus a single Playwright 

837 browser if JS fallback is triggered) for the generic HTML URLs. 

838 

839 Args: 

840 urls: List of URLs to fetch. 

841 timeout: Request timeout in seconds per URL. 

842 language: Language for justext stoplist. 

843 enable_js_rendering: When True, the HTML pipeline falls back to a 

844 headless browser for pages that need JavaScript. Defaults to 

845 False because the default Docker production image ships without 

846 Chromium. Limited internal benchmark comparisons (dev instances 

847 with Chromium vs Docker without) showed no measurable 

848 research-quality improvement from JS rendering, and most regular 

849 benchmark runs are on Docker without Chromium anyway. The 

850 user-facing toggle is ``web.enable_javascript_rendering``. 

851 content_cache: Optional shared ``ResearchContentCache``. When None, 

852 the cache bound to the current research (if any) is used. 

853 allow_private_ips: Whether the downloader may access private IPs 

854 after SSRF validation. The run policy, when supplied, is also 

855 checked on each redirect and pinned connect address. Cloud 

856 metadata remains blocked regardless. 

857 block_link_local: Keep the whole link-local range blocked even when 

858 ``allow_private_ips`` is True (forwarded to the downloader's 

859 static-fetch ``SafeSession``). No effect without the grant. 

860 egress_context: Optional run policy enforced on every transport hop. 

861 progress_callback: Optional ``(done, total)`` hook invoked once 

862 up front with ``(0, len(urls))`` and after each URL 

863 completes. Lets a long serial fetch report liveness (and 

864 notice cancellation) instead of going silent for minutes. 

865 A regular ``Exception`` raised by the hook is swallowed 

866 (debug-logged) so a flaky UI callback can neither corrupt an 

867 already-fetched result nor abort the remaining batch; 

868 ``BaseException`` — including ``ResearchTerminatedException``, 

869 which derives ``BaseException`` and therefore passes through 

870 the per-URL ``except Exception`` handlers untouched — still 

871 propagates so cancellation takes effect promptly. 

872 

873 Returns: 

874 Dict mapping URL → extracted text (or None if failed). 

875 """ 

876 try: 

877 from local_deep_research.utilities.research_content_cache import ( 

878 resolve_cache, 

879 ) 

880 

881 _cache = resolve_cache(content_cache) 

882 except Exception: # pragma: no cover - defensive 

883 _cache = None 

884 from ..playwright_html import AutoHTMLDownloader 

885 

886 results: Dict[str, Optional[str]] = {} 

887 total = len(urls) 

888 done = 0 

889 

890 def _report_progress() -> None: 

891 if progress_callback is not None and total: 

892 try: 

893 progress_callback(done, total) 

894 except Exception: 

895 # A flaky UI callback must not corrupt already-fetched 

896 # results (specialized loop: the ``except Exception`` 

897 # below would otherwise mistake it for a downloader 

898 # failure and overwrite good content with the HTML 

899 # fallback) nor abort the remaining batch (HTML loop). 

900 # ``except Exception`` deliberately lets ``BaseException`` 

901 # — including ``ResearchTerminatedException`` — propagate 

902 # for prompt cancellation. 

903 logger.debug( 

904 "batch_fetch_and_extract progress_callback failed; " 

905 "continuing batch" 

906 ) 

907 

908 _report_progress() 

909 

910 # Shared-cache fast path first: repeat URLs (quick phase vs subagents, 

911 # lead vs subagents) are served without any network fetch. The fast 

912 # path is grant-aware: a private/loopback URL cached by a granted call 

913 # must not be served to a non-granted caller sharing the same cache. 

914 pending_urls: List[str] = [] 

915 ungranted_urls: List[str] = [] 

916 if _cache is not None: 

917 for url in urls: 

918 if not _is_cache_hit_allowed( 

919 url, allow_private_ips, block_link_local, _cache 

920 ): 

921 ungranted_urls.append(url) 

922 continue 

923 try: 

924 _hit = _cache.get(url) 

925 except Exception: # pragma: no cover - defensive 

926 _hit = None 

927 if _hit is not None and isinstance(_hit.get("content"), str): 

928 results[url] = _hit["content"] or None 

929 done += 1 

930 _report_progress() 

931 else: 

932 pending_urls.append(url) 

933 else: 

934 pending_urls = list(urls) 

935 

936 # Shared holder for the single generic-HTML downloader (one Playwright 

937 # browser when JS fallback is enabled). Lazily created on first HTML 

938 # fallback so specialized-only batches never pay for it. 

939 _holder: Dict[str, Any] = {"downloader": None} 

940 

941 def _get_shared_downloader() -> Any: 

942 if _holder["downloader"] is None: 

943 dl = AutoHTMLDownloader( 

944 timeout=timeout,