From 4400b6c96517b9d6fc59f34f1b6e8a3583b8d02a Mon Sep 17 00:00:00 2001 From: Aliaksandr Nazaruk Date: Wed, 29 Jul 2026 14:55:08 +0200 Subject: [PATCH] Detect and repair extraction schemas that no longer fit the page A generated schema is validated once, against the snapshot it was generated from, and then applied for months while the site keeps changing. When a selector stops matching, extraction does not fail: it returns fewer records, or records with a silently empty column, and the caller is told nothing. The second shape is the dangerous one - the record count is unchanged, so nothing downstream can see it. The diagnostic for this already exists in the library (_validate_schema); it just never runs after generation. This adds serve-time health scoring: health_threshold score every extraction, report anything below it on_degraded callback for the report; defaults to a logger warning heal_llm_config optional: regenerate the schema from the page that degraded it max_regenerations bound on that (default 1) The score is computed from the records already extracted - no second pass over the DOM and no model call, so the check costs nothing. Off by default: without health_threshold the behaviour is byte-identical to before, which is pinned by a test. Regeneration is deliberately conservative. The replacement is scored by the same measure that raised the alarm and adopted only if it beats it, because 'the selectors match something' is not the same as 'this extractor is healthy again' - and the current schema's field names are carried into the request, since everything downstream is keyed on them. A failed repair keeps the old schema and never breaks the crawl. 12 tests: both rot shapes, threshold behaviour, the off-by-default guarantee, and the regeneration policy with the generator stubbed (adopt / refuse / bound / provider failure). Co-Authored-By: Claude Fable 5 --- crawl4ai/extraction_strategy.py | 151 ++++++++++++++++++++++- tests/test_schema_health.py | 204 ++++++++++++++++++++++++++++++++ 2 files changed, 354 insertions(+), 1 deletion(-) create mode 100644 tests/test_schema_health.py diff --git a/crawl4ai/extraction_strategy.py b/crawl4ai/extraction_strategy.py index ed039890e..e0f7011c0 100644 --- a/crawl4ai/extraction_strategy.py +++ b/crawl4ai/extraction_strategy.py @@ -3,6 +3,7 @@ from typing import Any, List, Dict, Optional, Tuple, Pattern, Union from concurrent.futures import ThreadPoolExecutor, as_completed import json +import sys import time from enum import IntFlag, auto @@ -1080,10 +1081,158 @@ def __init__(self, schema: Dict[str, Any], **kwargs): Args: schema (Dict[str, Any]): The schema defining the extraction rules. + health_threshold (float, optional): Enables serve-time health checking. A + schema is generated and validated against one snapshot of a page, but it + is then applied for months while the site keeps changing. When a selector + stops matching, extraction does not fail - it returns fewer records, or + records with silently empty fields, and the caller is told nothing. With + this set, every extraction is scored (see `_health_report`) and anything + below the threshold is reported through `on_degraded`. Costs nothing: + the score is computed from the records already extracted, with no second + pass over the DOM. Defaults to None (off, no behaviour change). + on_degraded (Callable[[dict], None], optional): Called with the health report + when a extraction scores below `health_threshold`. Defaults to a warning + on the strategy's logger, or stderr when there is none. + heal_llm_config (LLMConfig, optional): When set, a degraded extraction also + triggers ONE regeneration of the schema from the page that degraded it, + through `generate_schema`. The replacement is adopted only if it actually + scores better on that same page, so a failed repair leaves the old schema + in place rather than swapping one broken extractor for another. The field + names of the current schema are carried into the request, because + everything downstream is keyed on them. Requires `health_threshold`. + max_regenerations (int): How many times this strategy may regenerate over its + lifetime. Defaults to 1 - a site that is permanently unreadable must not + bill for a model call per page. """ super().__init__(**kwargs) self.schema = schema self.verbose = kwargs.get("verbose", False) + self.health_threshold = kwargs.get("health_threshold", None) + self.on_degraded = kwargs.get("on_degraded", None) + self.heal_llm_config = kwargs.get("heal_llm_config", None) + self.max_regenerations = kwargs.get("max_regenerations", 1) + self._regenerations = 0 + + def _health_report(self, results: List[Dict[str, Any]]) -> Dict[str, Any]: + """Score an extraction that has already happened. No DOM work, no model call. + + `coverage` is the fraction of the schema's declared fields that came back + populated in at least one record; a schema whose baseSelector matched nothing + scores 0. Both are the shapes site changes actually take: a renamed container + empties the result, a renamed field empties one column and leaves the record + count untouched - which is the one nothing downstream can see. + """ + declared = [f.get("name") for f in self.schema.get("fields", []) + if f.get("name")] + declared += [f.get("name") for f in self.schema.get("baseFields", []) + if f.get("name")] + populated = {name for r in results for name in declared + if r.get(name) not in (None, "", [], {})} + coverage = (len(populated) / len(declared)) if declared else 1.0 + return { + "records": len(results), + "declared_fields": len(declared), + "populated_fields": len(populated), + "coverage": 0.0 if not results else coverage, + "empty_fields": sorted(set(declared) - populated), + "healthy": bool(results) and coverage >= (self.health_threshold or 0.0), + } + + def _check_health(self, results: List[Dict[str, Any]], url: str = "", + html_content: str = "") -> List[Dict[str, Any]]: + """Report an extraction that no longer fits the page it is being applied to, and - + when a heal config is present - regenerate the schema from that same page. + + Returns the results to serve: the original ones, or the ones a healed schema + produced. + """ + if self.health_threshold is None: + return results + report = self._health_report(results) + if report["healthy"]: + return results + report["url"] = url + report["schema_name"] = self.schema.get("name", "") + healed = self._try_heal(report, html_content, url) + if healed is not None: + return healed + if self.on_degraded is not None: + self.on_degraded(report) + return results + message = ( + f"Extraction schema '{report['schema_name']}' looks stale on {url or 'page'}: " + f"{report['records']} record(s), coverage {report['coverage']:.0%} " + f"(threshold {self.health_threshold:.0%})" + + (f", always-empty field(s): {', '.join(report['empty_fields'])}" + if report["empty_fields"] else "") + + " - the page has probably changed; regenerate the schema." + ) + logger = getattr(self, "logger", None) + if logger is not None: + logger.warning(message=message, tag="EXTRACT") + else: + print(f"[EXTRACT] {message}", file=sys.stderr) + return results + + def _try_heal(self, report: Dict[str, Any], html_content: str, + url: str) -> Optional[List[Dict[str, Any]]]: + """Regenerate the schema from the page that broke it. None when not attempted or + when the replacement did not turn out better. + + The check after regeneration is the point: `generate_schema` validates a candidate + against the HTML it was given, but 'the selectors match something' is not the same + as 'this extractor is healthy again', and adopting a replacement on the strength of + the generator's own opinion is how a stale extractor becomes a differently-stale + one. So the candidate is scored by the same measure that raised the alarm, and it + has to beat it. + """ + if (self.heal_llm_config is None or not html_content + or self._regenerations >= self.max_regenerations): + return None + self._regenerations += 1 + schema_type = "XPATH" if "XPath" in type(self).__name__ else "CSS" + # Carry the field names over: everything downstream is keyed on them, so a repair + # that renames a column is not a repair. + target = {f["name"]: "..." for f in self.schema.get("fields", []) + if f.get("name")} + try: + candidate = JsonElementExtractionStrategy.generate_schema( + html=html_content, schema_type=schema_type, + target_json_example=json.dumps(target), + llm_config=self.heal_llm_config) + probe = type(self)(schema=candidate) + probe_results = probe.extract(url=url, html_content=html_content) + probe.schema, probe.health_threshold = candidate, self.health_threshold + new_report = probe._health_report(probe_results) + except Exception as e: # noqa: BLE001 - a failed repair must not fail the crawl + self._report_heal(report, None, f"regeneration failed: {e}", url) + return None + if new_report["coverage"] <= report["coverage"] or not new_report["healthy"]: + self._report_heal(report, new_report, "regenerated schema is no better - " + "keeping the current one", url) + return None + self.schema = candidate + self._report_heal(report, new_report, "schema regenerated and adopted", url) + return probe_results + + def _report_heal(self, before: Dict[str, Any], after: Optional[Dict[str, Any]], + outcome: str, url: str) -> None: + payload = dict(before) + payload["heal"] = {"outcome": outcome, "attempt": self._regenerations, + "coverage_before": before["coverage"], + "coverage_after": after["coverage"] if after else None} + if self.on_degraded is not None: + self.on_degraded(payload) + return + message = (f"Extraction schema '{before.get('schema_name', '')}' on " + f"{url or 'page'}: {outcome} (coverage " + f"{before['coverage']:.0%}" + + (f" -> {after['coverage']:.0%}" if after else "") + ")") + logger = getattr(self, "logger", None) + if logger is not None: + logger.warning(message=message, tag="EXTRACT") + else: + print(f"[EXTRACT] {message}", file=sys.stderr) def extract( self, url: str, html_content: str, *q, **kwargs @@ -1128,7 +1277,7 @@ def extract( if item: results.append(item) - return results + return self._check_health(results, url, html_content) @abstractmethod def _parse_html(self, html_content: str): diff --git a/tests/test_schema_health.py b/tests/test_schema_health.py new file mode 100644 index 000000000..6105e8ad8 --- /dev/null +++ b/tests/test_schema_health.py @@ -0,0 +1,204 @@ +"""Serve-time health checking for generated extraction schemas. + +A schema is validated once, against the snapshot it was generated from, and then applied +for months while the site keeps changing. When a selector stops matching, extraction does +not raise: it returns fewer records, or records with a silently empty column. These tests +pin both shapes of that failure and the report that now names them. +""" +import sys +from pathlib import Path + +import pytest + +sys.path.insert(0, str(Path(__file__).resolve().parents[1])) + +from crawl4ai.extraction_strategy import ( # noqa: E402 + JsonCssExtractionStrategy, + JsonElementExtractionStrategy, +) + +SCHEMA = { + "name": "Products", + "baseSelector": "article.product", + "fields": [ + {"name": "title", "selector": "h3", "type": "text"}, + {"name": "price", "selector": "p.price", "type": "text"}, + ], +} + +PAGE = """ + +

First

10

+

Second

20

+ +""" + +# The two shapes a redesign takes. +PAGE_CONTAINER_RENAMED = PAGE.replace('class="product"', 'class="product-card"') +PAGE_FIELD_RENAMED = PAGE.replace('class="price"', 'class="product-price"') + + +def test_off_by_default_and_silent(): + """No threshold: behaviour is exactly as before, including on a broken page.""" + strategy = JsonCssExtractionStrategy(schema=SCHEMA) + assert len(strategy.extract(url="", html_content=PAGE)) == 2 + assert strategy.extract(url="", html_content=PAGE_CONTAINER_RENAMED) == [] + + +def test_healthy_page_reports_nothing(): + seen = [] + strategy = JsonCssExtractionStrategy( + schema=SCHEMA, health_threshold=0.9, on_degraded=seen.append) + results = strategy.extract(url="http://x/", html_content=PAGE) + assert len(results) == 2 + assert seen == [] + + +def test_container_renamed_is_reported(): + """The obvious rot: nothing matches, the result is empty.""" + seen = [] + strategy = JsonCssExtractionStrategy( + schema=SCHEMA, health_threshold=0.9, on_degraded=seen.append) + results = strategy.extract(url="http://x/", html_content=PAGE_CONTAINER_RENAMED) + assert results == [] + assert len(seen) == 1 + assert seen[0]["records"] == 0 + assert seen[0]["coverage"] == 0.0 + assert seen[0]["url"] == "http://x/" + + +def test_field_renamed_is_reported_although_record_count_is_unchanged(): + """The dangerous rot: the same number of records, one column silently empty.""" + seen = [] + strategy = JsonCssExtractionStrategy( + schema=SCHEMA, health_threshold=0.9, on_degraded=seen.append) + results = strategy.extract(url="http://x/", html_content=PAGE_FIELD_RENAMED) + assert len(results) == 2, "the count is unchanged - this is why it goes unnoticed" + assert all("price" not in r or not r["price"] for r in results) + assert len(seen) == 1 + assert seen[0]["empty_fields"] == ["price"] + assert seen[0]["coverage"] == pytest.approx(0.5) + + +def test_threshold_is_respected(): + """Half the fields populated passes at 0.5 and fails at 0.6.""" + lenient, strict = [], [] + JsonCssExtractionStrategy(schema=SCHEMA, health_threshold=0.5, + on_degraded=lenient.append + ).extract(url="", html_content=PAGE_FIELD_RENAMED) + JsonCssExtractionStrategy(schema=SCHEMA, health_threshold=0.6, + on_degraded=strict.append + ).extract(url="", html_content=PAGE_FIELD_RENAMED) + assert lenient == [] + assert len(strict) == 1 + + +def test_default_reporter_warns_without_a_callback(capsys): + JsonCssExtractionStrategy(schema=SCHEMA, health_threshold=0.9).extract( + url="http://x/", html_content=PAGE_FIELD_RENAMED) + err = capsys.readouterr().err + assert "looks stale" in err + assert "price" in err + + +# --- regeneration ----------------------------------------------------------------- +# The generator is stubbed: these pin the POLICY around regeneration - when it fires, +# what it accepts, what it refuses, how often - which is what can go wrong without a +# model being involved. + +REPAIRED = { + "name": "Products", + "baseSelector": "article.product", + "fields": [ + {"name": "title", "selector": "h3", "type": "text"}, + {"name": "price", "selector": "p.product-price", "type": "text"}, + ], +} +USELESS = dict(REPAIRED, fields=[{"name": "title", "selector": "h3", "type": "text"}, + {"name": "price", "selector": "p.gone", "type": "text"}]) + + +@pytest.fixture +def stub_generator(monkeypatch): + calls = [] + + def make(schema_to_return): + def fake(**kwargs): + calls.append(kwargs) + return schema_to_return + monkeypatch.setattr(JsonElementExtractionStrategy, "generate_schema", + staticmethod(fake)) + return calls + return make + + +def test_heal_adopts_a_schema_that_actually_fixes_the_page(stub_generator): + calls = stub_generator(REPAIRED) + seen = [] + strategy = JsonCssExtractionStrategy( + schema=dict(SCHEMA), health_threshold=0.9, on_degraded=seen.append, + heal_llm_config=object()) + results = strategy.extract(url="http://x/", html_content=PAGE_FIELD_RENAMED) + assert len(calls) == 1, "one regeneration, from the page that degraded" + assert all(r.get("price") for r in results), "the served records are the healed ones" + assert strategy.schema == REPAIRED, "and the strategy keeps the new schema" + assert seen[0]["heal"]["outcome"] == "schema regenerated and adopted" + assert seen[0]["heal"]["coverage_before"] == pytest.approx(0.5) + assert seen[0]["heal"]["coverage_after"] == pytest.approx(1.0) + + +def test_heal_refuses_a_replacement_that_is_no_better(stub_generator): + stub_generator(USELESS) + seen = [] + original = dict(SCHEMA) + strategy = JsonCssExtractionStrategy( + schema=dict(SCHEMA), health_threshold=0.9, on_degraded=seen.append, + heal_llm_config=object()) + strategy.extract(url="http://x/", html_content=PAGE_FIELD_RENAMED) + assert strategy.schema == original, "a repair that does not repair is discarded" + assert "no better" in seen[0]["heal"]["outcome"] + + +def test_heal_carries_the_field_names_into_the_request(stub_generator): + calls = stub_generator(REPAIRED) + JsonCssExtractionStrategy( + schema=dict(SCHEMA), health_threshold=0.9, heal_llm_config=object() + ).extract(url="http://x/", html_content=PAGE_FIELD_RENAMED) + example = calls[0]["target_json_example"] + assert "title" in example and "price" in example, ( + "downstream is keyed on the field names - a repair may not rename a column") + + +def test_heal_is_bounded(stub_generator): + calls = stub_generator(USELESS) + strategy = JsonCssExtractionStrategy( + schema=dict(SCHEMA), health_threshold=0.9, heal_llm_config=object(), + max_regenerations=1) + for _ in range(5): + strategy.extract(url="http://x/", html_content=PAGE_FIELD_RENAMED) + assert len(calls) == 1, "a permanently unreadable site must not bill per page" + + +def test_heal_failure_does_not_break_the_crawl(monkeypatch): + def boom(**kwargs): + raise RuntimeError("provider is down") + monkeypatch.setattr(JsonElementExtractionStrategy, "generate_schema", + staticmethod(boom)) + seen = [] + strategy = JsonCssExtractionStrategy( + schema=dict(SCHEMA), health_threshold=0.9, on_degraded=seen.append, + heal_llm_config=object()) + results = strategy.extract(url="http://x/", html_content=PAGE_FIELD_RENAMED) + assert len(results) == 2, "the original records are still served" + assert "regeneration failed" in seen[0]["heal"]["outcome"] + + +def test_no_heal_config_means_no_generation(stub_generator): + calls = stub_generator(REPAIRED) + JsonCssExtractionStrategy(schema=dict(SCHEMA), health_threshold=0.9).extract( + url="http://x/", html_content=PAGE_FIELD_RENAMED) + assert calls == [], "reporting stays available without opting into model calls" + + +if __name__ == "__main__": + raise SystemExit(pytest.main([__file__, "-v"]))