Coverage for src/local_deep_research/research_library/downloaders/extraction/readability_extractor.py: 100%
104 statements
« prev ^ index » next coverage.py v7.16.2, created at 2026-10-22 21:20 +0000
1"""
2Mozilla Readability-based content extractor.
4Uses readabilipy (Python wrapper around Readability.js) to extract the
5main article content from a page, stripping navigation, sidebars, and
6other non-article elements at the DOM level.
8Uses Node.js for full Readability.js support when available;
9readabilipy falls back to pure-Python mode automatically.
10"""
12import threading
13from typing import Any, Callable, Optional
15from loguru import logger
17from .base import BaseExtractor
19# Readability.js does not run in-process: readabilipy shells out to a Node
20# subprocess (ExtractArticle.js) and exposes no timeout, so a slow or wedged
21# node blocks the calling thread for as long as it takes. That is unbounded
22# for a research run, and under pytest-timeout it costs a whole xdist worker
23# rather than a single test (#6209). Bound the Node path instead and fall
24# back to readabilipy's pure-Python parser, which is lower quality but does
25# not leave a process behind.
26#
27# Seconds to allow the Node extraction. At <= 0 the pure-Python parser is
28# used and the availability probe below is skipped too, so no node process is
29# started at all. The test suite sets 0 (tests/conftest.py) so unit tests
30# never shell out.
31NODE_TIMEOUT_SECONDS = 20.0
33# How many node calls may be IN FLIGHT at once. This is a resource bound, not
34# a health signal. A research run extracts several pages concurrently (one
35# worker thread per page — see lxml_thread_safety), and node is healthy on
36# the overwhelming majority of them, so this has to sit above the normal
37# fan-out: every caller turned away here is silently downgraded to the
38# lower-quality pure-Python parser. What it buys is a ceiling on a single
39# burst — if node wedges while these are all running, at most this many
40# calls strand at once.
41NODE_MAX_CONCURRENT_CALLS = 8
43# How many STRANDED calls — timed out, and still blocked on their node
44# process — may pile up before we stop reaching for node at all. This is the
45# health signal, and it is deliberately much tighter than the concurrency
46# bound above. Gate on threads that are STILL RUNNING rather than on a count
47# of past timeouts: the budget is wall-clock, so under concurrent extraction
48# a merely BUSY node can blow it on every thread at once. Those threads
49# finish shortly afterwards and drop out of the list; a genuinely wedged
50# node's threads never do. Counting live threads therefore measures the thing
51# we actually care about — stranded process/thread pairs — and recovers on
52# its own once they clear, instead of latching off permanently the first time
53# the machine is busy.
54NODE_MAX_STRANDED_CALLS = 3
56_node_state_lock = threading.Lock()
57# Threads whose node call overran the budget and that we stopped waiting for.
58_stranded_calls: list[threading.Thread] = []
59# Node calls admitted but not yet resolved; see _acquire_node_slot.
60_node_calls_in_flight = 0
62_have_node_lock = threading.Lock()
63_have_node_cache: Optional[bool] = None
66def _node_is_available() -> bool:
67 """Whether readabilipy can actually run Readability.js on this machine.
69 Probed once and cached, because the probe is expensive and unbounded:
70 readabilipy's ``have_node()`` shells out to ``node -v`` with no timeout of
71 its own and, on a first run with no bundled ``node_modules``, goes on to
72 run ``npm install`` under a process-wide ``os.chdir``. Running that per
73 page is exactly the cost this module exists to avoid.
75 Caching is therefore a trade-off, not a free win: the answer *can* change,
76 because ``have_node()`` returns False on a transient failure (a busy
77 machine, a failed or half-finished ``npm install``) as readily as on a
78 genuinely node-less host. A False cached that way disables Readability.js
79 for the rest of the process — until a restart, there is no re-probe. We
80 accept that over an unbounded shell-out on every page.
82 Getting False right matters beyond saving the probe: the published
83 runtime image installs Node only in the build stage, so the shipped
84 container has no ``node``. There, ``use_readability=True`` is downgraded
85 to ``False`` inside readabilipy and the "node worker" is really a
86 pure-Python html5lib parse. Bounding *that* is worse than useless — an
87 overrun abandons the parse and re-runs the identical work on the caller's
88 thread, doubling both CPU and peak memory for no timeout benefit. When
89 node is absent we therefore take the pure-Python path directly, with no
90 worker thread and no budget.
92 A probe that raises is treated as "node may be present", which keeps the
93 bounded path (and therefore the timeout protection) rather than silently
94 dropping it.
95 """
96 global _have_node_cache
97 with _have_node_lock:
98 if _have_node_cache is None:
99 try:
100 from readabilipy.simple_json import have_node
102 _have_node_cache = bool(have_node())
103 except Exception:
104 logger.debug(
105 "readabilipy node probe failed; assuming node may be "
106 "present and keeping the bounded path"
107 )
108 _have_node_cache = True
109 return _have_node_cache
112def _stranded_call_count() -> int:
113 """Timed-out calls whose thread is still blocked on its node process."""
114 with _node_state_lock:
115 _stranded_calls[:] = [t for t in _stranded_calls if t.is_alive()]
116 return len(_stranded_calls)
119def _acquire_node_slot() -> bool:
120 """Reserve one live node slot, if both bounds allow it.
122 Prune, count and reserve happen under a single hold of the lock. Checking
123 a count and then starting a thread as two steps is a check-then-act race:
124 sixteen concurrent callers would each read "0 stranded" and each start a
125 worker, so a node that wedges while they run strands sixteen threads.
126 Reserving here makes both ceilings hold no matter how wide the caller
127 fan-out is.
129 The two bounds are separate on purpose and compose:
131 * ``NODE_MAX_CONCURRENT_CALLS`` caps calls that are merely *running*, so
132 healthy concurrency is not throttled down to the stranded budget, and
133 one burst against a node that wedges mid-flight can strand at most that
134 many calls;
135 * ``NODE_MAX_STRANDED_CALLS`` caps calls that already timed out and are
136 still stuck, and once it is reached nothing new is admitted until they
137 clear.
138 """
139 global _node_calls_in_flight
140 with _node_state_lock:
141 _stranded_calls[:] = [t for t in _stranded_calls if t.is_alive()]
142 if len(_stranded_calls) >= NODE_MAX_STRANDED_CALLS:
143 return False
144 if _node_calls_in_flight >= NODE_MAX_CONCURRENT_CALLS:
145 return False
146 _node_calls_in_flight += 1
147 return True
150def _release_node_slot(
151 stranded_worker: Optional[threading.Thread], settled: list
152) -> None:
153 """Give the slot back, or hand it over to the stranded-thread list.
155 ``settled`` is a one-element list (``[False]``) owned by the caller and
156 shared across every exit path of a single call. Checking and setting
157 ``settled[0]`` happens under the same lock hold as the release itself,
158 which is what makes this idempotent: a call's slot settles exactly once
159 no matter how many of the caller's exit paths try to release it.
161 That matters because a caller can legitimately reach for this more than
162 once for the same call — the ordinary release after ``join()`` returns,
163 and a ``finally`` backstop below it for exits nobody asked for. Those
164 two used to be "release, then note it happened" as two separate
165 statements, which left a window: an asynchronous exception (e.g. a
166 ``KeyboardInterrupt`` delivered to the main thread) landing between them
167 made the backstop believe the slot was never settled and release it
168 again — decrementing the live count twice (drifting the effective
169 concurrency bound up by one admission) and, when the call was stranded,
170 appending the same still-alive thread to ``_stranded_calls`` twice.
171 Moving the check-and-set inside this lock hold closes that window: the
172 second call in such a race sees ``settled[0]`` already ``True`` and is a
173 no-op.
175 Both the settle and the stranded-list handoff happen under one hold of
176 the lock so the live count never dips and lets an extra caller through
177 in between.
179 The count is clamped at zero. It should never go negative — every
180 admitted call settles exactly once — but a release against an
181 already-zero counter would otherwise buy admissions *above* the bound,
182 which is the wrong way to fail.
183 """
184 global _node_calls_in_flight
185 with _node_state_lock:
186 if settled[0]:
187 return
188 settled[0] = True
189 _node_calls_in_flight = max(0, _node_calls_in_flight - 1)
190 if stranded_worker is None:
191 return
192 _stranded_calls.append(stranded_worker)
193 stranded = sum(1 for t in _stranded_calls if t.is_alive())
194 if stranded >= NODE_MAX_STRANDED_CALLS:
195 logger.warning(
196 f"{stranded} Readability.js calls are still stuck on node — "
197 "using pure-Python extraction until they clear"
198 )
201def _reset_node_breaker() -> None:
202 """Test hook: forget the stranded calls and the cached node probe.
204 It deliberately leaves ``_node_calls_in_flight`` alone. That counter is
205 owned by the calls themselves, each of which releases it in a ``finally``;
206 zeroing it underneath a call that is still running made the eventual
207 release drive it negative, which handed out slots above the bound. A test
208 that strands workers must drain them, not reset the count out from under
209 them.
210 """
211 global _have_node_cache
212 with _node_state_lock:
213 _stranded_calls.clear()
214 with _have_node_lock:
215 _have_node_cache = None
218def _extract_json(
219 simple_json_from_html_string: Callable[..., Any],
220 html: str,
221 *,
222 node_available: bool,
223) -> Any:
224 """Run readabilipy, keeping the Node subprocess on a leash.
226 Only the Node path is bounded, and only when node is actually installed
227 (``node_available``); otherwise readabilipy's pure-Python parser runs
228 inline on this thread, because there is no subprocess to leash and
229 abandoning an in-process parse only duplicates the work.
231 The Node call runs on a daemon thread: if it overruns the budget we stop
232 waiting and take the pure-Python result instead. ``daemon=True`` keeps
233 the abandoned thread from blocking interpreter shutdown, but it does not
234 kill anything — a timed-out call leaves three things behind:
236 * the stray ``node`` process (readabilipy owns the handle and gives us no
237 way to signal it), which is reparented and outlives this interpreter;
238 * the thread itself, which is not parked but blocked inside
239 ``subprocess.run`` and can only exit when that process does;
240 * readabilipy's ``NamedTemporaryFile(delete=False)`` copy of the full
241 page, which it unlinks only after node returns *successfully*. On a
242 ``subprocess.CalledProcessError`` — node exits non-zero rather than
243 merely running long — readabilipy re-raises without unlinking either
244 the input or the output temp file (``readabilipy/simple_json.py``,
245 around the ``except subprocess.CalledProcessError`` block), so a node
246 crash leaks both regardless of whether the call was one we timed out
247 on.
249 ``daemon=True`` bounds none of those; it only stops them delaying exit.
250 Bounding *how many* accumulate is what ``NODE_MAX_STRANDED_CALLS``,
251 ``NODE_MAX_CONCURRENT_CALLS`` and :func:`_acquire_node_slot` are for.
253 A reserved slot settles — is released, or handed to the stranded list —
254 exactly once, on every exit, including one taken by an exception that is
255 not ours, such as a ``KeyboardInterrupt`` delivered to the main thread
256 while it is parked in ``join()``. That path hands the slot *over* rather
257 than releasing it: the worker is still running, so its node process is
258 still outstanding and must keep being counted. Leaking the reservation
259 instead was a silent, permanent loss of capacity — three interrupted
260 joins and node was off for the life of the process, with nothing logged.
261 The "exactly once" is enforced by :func:`_release_node_slot` itself: it
262 takes a one-element ``settled`` list shared across every exit path below
263 and checks-and-sets it under the same lock hold as the release, so a
264 second call for the same reservation — e.g. this function's ``finally``
265 backstop running after the ordinary release already did — is a no-op
266 rather than a double release.
268 Note the budget is wall-clock, not CPU: under concurrent extraction the
269 node calls contend, so a busy machine falls back more often. That is
270 working as intended — the fallback still returns content — and it no
271 longer disables node, because only calls that are still running count.
272 """
273 timeout = NODE_TIMEOUT_SECONDS
274 if timeout <= 0 or not node_available:
275 return simple_json_from_html_string(html, use_readability=False)
277 if not _acquire_node_slot():
278 return simple_json_from_html_string(html, use_readability=False)
280 outcome: dict[str, Any] = {}
282 def _run() -> None:
283 try:
284 outcome["article"] = simple_json_from_html_string(
285 html, use_readability=True
286 )
287 except Exception as exc: # re-raised on the calling thread
288 # Deliberately not BaseException: re-raising a SystemExit or a
289 # KeyboardInterrupt raised on *this* thread would unwind an
290 # unrelated caller.
291 outcome["error"] = exc
293 worker: Optional[threading.Thread] = None
294 # Shared with _release_node_slot, which checks-and-sets settled[0] under
295 # its own lock hold so the slot below settles exactly once no matter how
296 # many of the paths below try to release it — see its docstring.
297 settled = [False]
298 try:
299 try:
300 worker = threading.Thread(
301 target=_run, name="readability-node", daemon=True
302 )
303 worker.start()
304 except RuntimeError:
305 # Thread exhaustion or interpreter shutdown. Extraction still has
306 # a working answer available, so fall back rather than fail the
307 # page. Nothing was started, so the slot is simply given back.
308 _release_node_slot(None, settled)
309 logger.warning(
310 "could not start the Readability.js worker thread — "
311 "falling back to pure-Python extraction"
312 )
313 return simple_json_from_html_string(html, use_readability=False)
315 worker.join(timeout)
316 overran = worker.is_alive()
317 _release_node_slot(worker if overran else None, settled)
318 finally:
319 # A no-op whenever one of the releases above already ran — see
320 # _release_node_slot. It only does real work when this finally is
321 # reached WITHOUT either of them having run: an interrupt or an
322 # asynchronous exception unwinding through the constructor, start(),
323 # or join() before the ordinary release got a chance to run. A
324 # worker that is still running at that point is stranded, not
325 # finished, so hand the slot over instead of releasing it.
326 _release_node_slot(
327 worker if worker is not None and worker.is_alive() else None,
328 settled,
329 )
331 if overran:
332 logger.warning(
333 f"Readability.js (node) exceeded {timeout}s — "
334 "falling back to pure-Python extraction"
335 )
336 return simple_json_from_html_string(html, use_readability=False)
338 if "error" in outcome:
339 raise outcome["error"]
340 return outcome.get("article")
343class ReadabilityExtractor(BaseExtractor):
344 """Extract content using Mozilla Readability.js via readabilipy.
346 Returns cleaned HTML (not plain text) so downstream extractors
347 like justext can still detect headings and structure.
348 """
350 def extract(self, html: str) -> Optional[str]:
351 if not html or not html.strip():
352 return None
354 try:
355 from readabilipy import simple_json_from_html_string
356 except ImportError:
357 logger.warning(
358 "readabilipy not installed — skipping Readability extraction"
359 )
360 return None
362 try:
363 article = _extract_json(
364 simple_json_from_html_string,
365 html,
366 # The probe itself shells out (see _node_is_available), so
367 # don't pay for it when the Node path is switched off anyway.
368 node_available=NODE_TIMEOUT_SECONDS > 0
369 and _node_is_available(),
370 )
371 except Exception:
372 logger.exception("readabilipy extraction failed")
373 return None
375 if not article:
376 return None
378 # Return HTML content only — preserves headings and structure
379 # so downstream extractors (justext) can parse them properly.
380 # Plain-text fallbacks are intentionally skipped: they would
381 # break justext (which expects HTML) and the pipeline has its
382 # own last-resort get_text() path.
383 content = article.get("content")
384 if content and isinstance(content, str) and content.strip():
385 return content
387 return None