Coverage for src/local_deep_research/research_library/downloaders/extraction/pipeline.py: 89%
204 statements
« prev ^ index » next coverage.py v7.16.0, created at 2026-09-06 15:42 +0000
1"""
2Shared extraction pipeline.
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).
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.
12Academic URLs (arXiv, PubMed, bioRxiv, etc.) are automatically routed to
13specialized downloaders first, with the generic HTML pipeline as fallback.
14"""
16from typing import Any, Dict, List, Optional
18from bs4 import BeautifulSoup
19from loguru import logger
21from .trafilatura_extractor import TrafilaturaExtractor
22from .readability_extractor import ReadabilityExtractor
23from .justext_extractor import JustextExtractor
24from .newspaper_extractor import NewspaperExtractor
25from .metadata_extractor import extract_metadata, metadata_to_text
26from ....utilities.lxml_thread_safety import install_per_thread_html_parsers
28# Route newspaper4k's and readabilipy's parser-less lxml.html.fromstring calls
29# onto per-thread parsers. Without this, worker threads extracting different
30# pages concurrently share one libxml2 xmlDict and corrupt it, aborting the
31# whole process with "double free or corruption (out)". Installed at import so
32# it is in place before any worker thread parses.
33install_per_thread_html_parsers()
36# --- Pipeline thresholds ---
37# Minimum extracted text length to consider extraction successful
38MIN_CONTENT_LENGTH = 50
39# If content is shorter than this, enrich with structured metadata
40# (JSON-LD, OpenGraph) — helps product pages, JS-heavy sites
41METADATA_ENRICHMENT_THRESHOLD = 1000
42# Boilerplate penalty per keyword when scoring extraction quality
43BOILERPLATE_PENALTY = 500
44# If an extractor discards more than this fraction of the previous
45# extractor's output, skip it (protects non-English content)
46SAFETY_DISCARD_RATIO = 0.2
48# Module-level singleton extractors (avoid re-creating per call)
49_trafilatura = TrafilaturaExtractor()
50_readability = ReadabilityExtractor()
51_justext_en = JustextExtractor(language="English")
52_newspaper = NewspaperExtractor()
54# Boilerplate keywords — used to penalize low-quality extraction
55_BOILERPLATE_KEYWORDS = [
56 "cookie",
57 "sign up",
58 "newsletter",
59 "subscribe",
60 "accept all",
61 "privacy policy",
62 "terms of service",
63]
66def _run_extractors_parallel(
67 html: str, url: str
68) -> tuple[str | None, str | None]:
69 """Run trafilatura and newspaper4k sequentially.
71 Both extractors call into lxml's C extension, which is not safe to
72 share across threads — running them in a ThreadPoolExecutor caused
73 Fatal Python error: Aborted during pool teardown on Python 3.14
74 (the workers would deadlock in shutdown's join). Serializing the
75 calls eliminates the crash; the perf cost is one extra extraction's
76 worth of CPU per page, which is acceptable.
78 Correction to that rationale, from a later SIGABRT (core dump showed
79 ``double free or corruption (out)`` in ``htmlParseDocument``): sharing an
80 lxml *parser* across threads is in fact supported — lxml serializes it with
81 a per-parser lock. The real hazard is the libxml2 ``xmlDict`` that threads
82 adopt from a shared parser and then intern into concurrently. So
83 serializing here was never the actual fix, and it could not have been:
84 it only covers two extractors within ONE page, while the corruption comes
85 from different pages extracted in different threads. The real fix is
86 per-thread parsers — see ``utilities.lxml_thread_safety``, installed at
87 this module's import. This sequencing is retained because it is harmless
88 and the Python 3.14 teardown deadlock above was a separate problem.
90 The function name is preserved for backwards compatibility with
91 any callers/tests that import it directly.
92 """
93 try:
94 trafilatura_content = _trafilatura.extract(html)
95 except Exception:
96 logger.debug("Pipeline: trafilatura raised an exception")
97 trafilatura_content = None
99 try:
100 newspaper_content = _newspaper.extract(html, url)
101 except Exception:
102 logger.debug("Pipeline: newspaper4k raised an exception")
103 newspaper_content = None
105 return trafilatura_content, newspaper_content
108def _count_boilerplate(text: str) -> int:
109 """Count boilerplate keyword occurrences in text."""
110 if not text:
111 return 0
112 lower = text.lower()
113 return sum(1 for kw in _BOILERPLATE_KEYWORDS if kw in lower)
116def _quality_score(text: str) -> int:
117 """Score extraction quality: length minus boilerplate penalty."""
118 if not text:
119 return 0
120 return len(text) - (_count_boilerplate(text) * BOILERPLATE_PENALTY)
123def extract_content(
124 html: str,
125 language: str = "English",
126 min_length: int = MIN_CONTENT_LENGTH,
127 url: str = "",
128) -> Optional[str]:
129 """Extract clean text content from HTML.
131 Pipeline:
132 1. trafilatura (primary — best benchmarks, multilingual, markdown)
133 2. newspaper4k (parallel — strong on news/forum pages)
134 → pick the higher-quality result from steps 1-2
135 3. readability → justext (fallback if both above fail, with 80% safety)
136 4. soup.get_text() (last resort)
138 Args:
139 html: Raw HTML string.
140 language: Language for justext stoplist (fallback only).
141 min_length: Minimum content length to accept.
142 url: Source URL (improves newspaper4k extraction accuracy).
144 Returns:
145 Extracted plain text, or None if content is below min_length.
146 """
147 if not html or not html.strip():
148 return None
150 # Run trafilatura and newspaper4k in parallel, pick the better result.
151 # newspaper4k is strong on news front pages and multi-answer threads
152 # where trafilatura sometimes extracts less content.
153 # 5s timeout per extractor — covers P95 of pages, cuts off outliers.
154 trafilatura_content, newspaper_content = _run_extractors_parallel(html, url)
156 traf_score = _quality_score(trafilatura_content)
157 np_score = _quality_score(newspaper_content)
159 if traf_score >= np_score and trafilatura_content:
160 content = trafilatura_content
161 winner = "trafilatura"
162 elif newspaper_content:
163 content = newspaper_content
164 winner = "newspaper4k"
165 else:
166 content = trafilatura_content
167 winner = "trafilatura"
169 if content and len(content.strip()) >= min_length:
170 logger.debug(
171 f"Pipeline: {winner} extracted {len(content)} chars"
172 + (
173 f" (traf={len(trafilatura_content or '')}, "
174 f"np4k={len(newspaper_content or '')})"
175 if newspaper_content and trafilatura_content
176 else ""
177 )
178 )
179 else:
180 # Fallback: readability → justext
181 logger.debug(
182 "Pipeline: primary extractors insufficient, using fallback"
183 )
185 soup = BeautifulSoup(html, "html.parser")
186 for tag_name in [
187 "script",
188 "style",
189 "iframe",
190 "noscript",
191 "svg",
192 "form",
193 "button",
194 "input",
195 "select",
196 "textarea",
197 ]:
198 for tag in soup.find_all(tag_name):
199 tag.decompose()
200 cleaned_html = str(soup)
202 justext_extractor = (
203 _justext_en
204 if language == "English"
205 else JustextExtractor(language=language)
206 )
208 content = None
209 prev_text_len = 0
211 for extractor in [_readability, justext_extractor]:
212 result = extractor.extract(
213 cleaned_html if prev_text_len == 0 else content
214 )
215 if result and result.strip():
216 result_len = len(result.strip())
217 # Safety: skip if extractor discards >80% of content.
218 # Compare text lengths (strip HTML tags for fair comparison
219 # since readability returns HTML but justext returns text).
220 if (
221 prev_text_len > 0
222 and result_len < prev_text_len * SAFETY_DISCARD_RATIO
223 ):
224 logger.debug(
225 f"Pipeline: {extractor.__class__.__name__} discarded "
226 f">80% of content — skipping"
227 )
228 continue
229 content = result
230 # Store text-equivalent length for fair comparison
231 if "<" in result:
232 prev_text_len = len(
233 BeautifulSoup(result, "html.parser").get_text()
234 )
235 else:
236 prev_text_len = result_len
237 logger.debug(
238 f"Pipeline: {extractor.__class__.__name__} "
239 f"returned {result_len} chars"
240 )
242 # Strip remaining HTML tags (e.g. readability-only mode)
243 if content and "<" in content:
244 content = BeautifulSoup(content, "html.parser").get_text(
245 separator="\n", strip=True
246 )
248 # Last resort
249 if not content or len(content.strip()) < min_length:
250 logger.debug("Pipeline: all extractors failed, using get_text()")
251 content = soup.get_text(separator="\n", strip=True)
253 if not content or len(content.strip()) < min_length:
254 return None
256 # Enrich with structured metadata when text extraction is thin
257 # (e.g. product pages, JS-heavy sites)
258 if len(content.strip()) < METADATA_ENRICHMENT_THRESHOLD:
259 metadata = extract_metadata(html)
260 supplement = metadata_to_text(metadata)
261 if supplement and supplement.strip():
262 logger.debug(
263 f"Pipeline: enriching with {len(supplement)} chars "
264 f"of structured metadata"
265 )
266 content = content.rstrip() + "\n\n" + supplement
268 return content
271def extract_content_with_metadata(
272 html: str,
273 language: str = "English",
274 min_length: int = MIN_CONTENT_LENGTH,
275) -> Optional[Dict[str, Any]]:
276 """Extract clean text and page metadata from HTML in a single pass.
278 Combines content extraction (trafilatura/readability/justext pipeline)
279 with title and description extraction from HTML meta tags. This avoids
280 the need for callers to do a separate BeautifulSoup parse for metadata.
282 Args:
283 html: Raw HTML string.
284 language: Language for justext stoplist (fallback only).
285 min_length: Minimum content length to accept.
287 Returns:
288 Dict with keys: content, title, description — or None if content
289 is below min_length.
290 """
291 if not html or not html.strip():
292 return None
294 # Single parse for metadata (title, description, og:*)
295 soup = BeautifulSoup(html, "html.parser")
297 title = None
298 if soup.title and soup.title.string:
299 title = soup.title.string.strip()
300 og_title = soup.find("meta", property="og:title")
301 if og_title and og_title.get("content"):
302 title = str(og_title["content"]).strip()
304 description = None
305 meta_desc = soup.find("meta", attrs={"name": "description"})
306 if meta_desc and meta_desc.get("content"):
307 description = str(meta_desc["content"]).strip()
308 og_desc = soup.find("meta", property="og:description")
309 if og_desc and og_desc.get("content"):
310 description = str(og_desc["content"]).strip()
312 # Extract content using the shared pipeline
313 content = extract_content(html, language=language, min_length=min_length)
314 if not content:
315 return None
317 return {
318 "title": title,
319 "description": description,
320 "content": content,
321 }
324def _try_specialized_downloader(url: str, timeout: int = 30) -> Optional[str]:
325 """Try a specialized downloader (arXiv, PubMed, etc.) for the URL.
327 Returns extracted text if a specialized downloader handles this URL
328 and succeeds, or None to signal "fall back to generic HTML pipeline".
329 """
330 try:
331 from local_deep_research.content_fetcher.url_classifier import (
332 URLClassifier,
333 URLType,
334 )
335 except ImportError:
336 return None
338 url_type = URLClassifier.classify(url)
340 # Only academic URL types have specialized downloaders worth trying.
341 # HTML, DOI, PDF, INVALID fall through to the generic pipeline.
342 _SPECIALIZED_TYPES = {
343 URLType.ARXIV,
344 URLType.PUBMED,
345 URLType.PMC,
346 URLType.SEMANTIC_SCHOLAR,
347 URLType.BIORXIV,
348 URLType.MEDRXIV,
349 }
350 if url_type not in _SPECIALIZED_TYPES:
351 return None
353 # Map URL type to downloader class (lazy imports to avoid circular deps)
354 downloader = None
355 try:
356 if url_type == URLType.ARXIV: 356 ↛ 360line 356 didn't jump to line 360 because the condition on line 356 was always true
357 from ..arxiv import ArxivDownloader
359 downloader = ArxivDownloader(timeout=timeout)
360 elif url_type in (URLType.PUBMED, URLType.PMC):
361 from ..pubmed import PubMedDownloader
363 downloader = PubMedDownloader(timeout=timeout)
364 elif url_type == URLType.SEMANTIC_SCHOLAR:
365 from ..semantic_scholar import SemanticScholarDownloader
367 downloader = SemanticScholarDownloader(timeout=timeout)
368 elif url_type in (URLType.BIORXIV, URLType.MEDRXIV):
369 from ..biorxiv import BioRxivDownloader
371 downloader = BioRxivDownloader(timeout=timeout)
372 except ImportError:
373 logger.debug(
374 f"Pipeline: specialized downloader not available for {url_type.value}"
375 )
376 return None
378 if not downloader: 378 ↛ 379line 378 didn't jump to line 379 because the condition on line 378 was never true
379 return None
381 try:
382 from ..base import ContentType
384 result = downloader.download_with_result(url, ContentType.TEXT)
385 if result.is_success and result.content:
386 text = result.content.decode("utf-8", errors="replace")
387 if len(text.strip()) >= MIN_CONTENT_LENGTH: 387 ↛ 399line 387 didn't jump to line 399 because the condition on line 387 was always true
388 logger.debug(
389 f"Pipeline: specialized downloader ({url_type.value}) "
390 f"returned {len(text)} chars for {url}"
391 )
392 return text
393 except Exception:
394 logger.debug(
395 f"Pipeline: specialized downloader failed for {url}",
396 exc_info=True,
397 )
398 finally:
399 try:
400 downloader.close()
401 except Exception: # noqa: silent-exception
402 pass
404 # Specialized downloader didn't produce content — fall back to HTML
405 logger.debug(
406 f"Pipeline: specialized downloader ({url_type.value}) returned "
407 f"no content for {url}, falling back to HTML pipeline"
408 )
409 return None
412def fetch_and_extract(
413 url: str,
414 timeout: int = 30,
415 language: str = "English",
416 enable_js_rendering: bool = False,
417) -> Optional[str]:
418 """Fetch a URL and extract clean text content.
420 Pipeline:
421 1. Specialized downloader (arXiv PDF, PubMed API, etc.) if URL matches
422 2. Static HTTP fetch → Playwright fallback (if JS needed and
423 ``enable_js_rendering`` is True) → trafilatura → readability →
424 justext
426 Args:
427 url: The URL to fetch.
428 timeout: Request timeout in seconds.
429 language: Language for justext stoplist.
430 enable_js_rendering: When True, the HTML pipeline falls back to a
431 headless browser for pages that need JavaScript. Defaults to
432 False because the default Docker production image ships without
433 Chromium. Limited internal benchmark comparisons (dev instances
434 with Chromium vs Docker without) showed no measurable
435 research-quality improvement from JS rendering, and most regular
436 benchmark runs are on Docker without Chromium anyway. The
437 user-facing toggle is ``web.enable_javascript_rendering``.
439 Returns:
440 Extracted plain text, or None if fetch or extraction failed.
441 """
442 # Try specialized downloader first (arXiv, PubMed, etc.)
443 specialized = _try_specialized_downloader(url, timeout=timeout)
444 if specialized:
445 return specialized
447 # Generic HTML pipeline
448 from ..playwright_html import AutoHTMLDownloader
450 downloader = AutoHTMLDownloader(
451 timeout=timeout,
452 language=language,
453 enable_js_rendering=enable_js_rendering,
454 )
455 try:
456 # download() returns extracted text as UTF-8 bytes (not raw HTML):
457 # AutoHTMLDownloader inherits HTMLDownloader.download() which runs
458 # _fetch_html() → _extract_content() → the full extraction pipeline.
459 result = downloader.download(url)
460 if result:
461 return result.decode("utf-8", errors="replace")
462 return None
463 except Exception:
464 logger.exception(f"fetch_and_extract failed for {url}")
465 return None
466 finally:
467 try:
468 downloader.close()
469 except Exception:
470 logger.debug("Failed to close downloader in fetch_and_extract")
473def batch_fetch_and_extract(
474 urls: List[str],
475 timeout: int = 30,
476 language: str = "English",
477 enable_js_rendering: bool = False,
478) -> Dict[str, Optional[str]]:
479 """Fetch multiple URLs and extract clean text from each.
481 For each URL:
482 1. Try specialized downloader (arXiv, PubMed, etc.) if URL matches
483 2. Fall back to generic HTML pipeline (AutoHTMLDownloader)
485 Uses a single AutoHTMLDownloader (and thus a single Playwright
486 browser if JS fallback is triggered) for the generic HTML URLs.
488 Args:
489 urls: List of URLs to fetch.
490 timeout: Request timeout in seconds per URL.
491 language: Language for justext stoplist.
492 enable_js_rendering: When True, the HTML pipeline falls back to a
493 headless browser for pages that need JavaScript. Defaults to
494 False because the default Docker production image ships without
495 Chromium. Limited internal benchmark comparisons (dev instances
496 with Chromium vs Docker without) showed no measurable
497 research-quality improvement from JS rendering, and most regular
498 benchmark runs are on Docker without Chromium anyway. The
499 user-facing toggle is ``web.enable_javascript_rendering``.
501 Returns:
502 Dict mapping URL → extracted text (or None if failed).
503 """
504 from ..playwright_html import AutoHTMLDownloader
506 results: Dict[str, Optional[str]] = {}
508 # Try specialized downloaders first — collect URLs that need HTML fallback
509 html_urls: List[str] = []
510 for url in urls:
511 try:
512 specialized = _try_specialized_downloader(url, timeout=timeout)
513 if specialized:
514 results[url] = specialized
515 continue
516 except Exception:
517 logger.debug(
518 f"Pipeline: specialized downloader error for {url}",
519 exc_info=True,
520 )
521 html_urls.append(url)
523 # Generic HTML pipeline for remaining URLs
524 if html_urls:
525 downloader = AutoHTMLDownloader(
526 timeout=timeout,
527 language=language,
528 enable_js_rendering=enable_js_rendering,
529 )
530 try:
531 for url in html_urls:
532 try:
533 data = downloader.download(url)
534 if data: 534 ↛ 537line 534 didn't jump to line 537 because the condition on line 534 was always true
535 results[url] = data.decode("utf-8", errors="replace")
536 else:
537 results[url] = None
538 except Exception:
539 logger.exception(
540 f"batch_fetch_and_extract failed for {url}"
541 )
542 results[url] = None
543 finally:
544 try:
545 downloader.close()
546 except Exception:
547 logger.debug(
548 "Failed to close downloader in batch_fetch_and_extract"
549 )
551 return results