Detect (and optionally repair) extraction schemas that no longer fit the page

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):

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 = """
+<html><body>
+  <article class="product"><h3>First</h3><p class="price">10</p></article>
+  <article class="product"><h3>Second</h3><p class="price">20</p></article>
+</body></html>
+"""
+
+# 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"]))