How Companies Are Scraping the Web for AI Training Data? Technical guide.

The Data Hunger Behind Modern AI: Why Scale Changed Everything

The story of large language model training is fundamentally a story about data. Not compute, not architecture — data. The scaling laws published by research teams in the early 2020s established a clear empirical relationship: given a fixed compute budget, the optimal strategy involves training a smaller model on more tokens rather than a larger model on fewer. This insight cascaded through the industry like a pressure wave, and the result is that training data demand has grown faster than GPU capacity for the past four years.

Consider the progression. The GPT-2 family trained on WebText — roughly 40GB of cleaned Reddit-linked text. GPT-3 consumed 570GB of filtered text spanning Common Crawl, Books, and Wikipedia. By 2023, the LLaMA 2 family trained on 2 trillion tokens. By 2024, LLaMA 3 was trained on over 15 trillion tokens — a more than 7,000× increase in token volume relative to GPT-3’s corpus in just four years. Falcon’s RefinedWeb processing pipeline filtered approximately 5 trillion raw tokens from Common Crawl to produce a 1-trillion-token high-quality subset. Mistral’s training corpus spans multiple trillions of tokens from curated web sources.

This data appetite is not limited to frontier labs. Every company building a domain-specific AI product — a legal reasoning model, a clinical note summarizer, a code completion engine for a proprietary language, a financial document parser — now faces the same fundamental engineering problem: how do you collect the right web data for AI training data at scale, clean it, deduplicate it, and serve it to your training infrastructure efficiently? The global AI training data market was valued at approximately USD 2.4 billion in 2023 and is projected to exceed USD 6.2 billion by 2028, driven almost entirely by the demand for larger, higher-quality corpora to feed the next generation of large language model training runs.

The answer, for the overwhelming majority of these teams, is web scraping. Not purchasing datasets — though that is a supplementary channel — and not purely synthetic data generation, which introduces distributional artifacts. The primary source of AI training data for production models in 2026 remains the public web, accessed through purpose-built data collection pipelines.

This creates a convergence between two disciplines that historically sat in separate parts of the engineering organization: web data engineering and ML infrastructure. Understanding how to architect, operate, and maintain a production web scraping pipeline for AI training data is now a core competency for any team building AI products at scale.

What Makes AI Training Data Different from Ordinary Web Scraping

Before diving into implementation, it is worth being precise about what distinguishes a data collection pipeline for AI training data from a conventional scraping operation. The differences are significant and drive most of the architectural decisions we will discuss.

Volume and throughput requirements are categorically different. A price monitoring pipeline might collect tens of thousands of records per day. An AI training data pipeline needs to process tens of billions of tokens per week to be useful. At 500 tokens per page, reaching 100 billion tokens requires crawling 200 million pages. A pipeline that processes 10,000 pages per day would take over 54 years to reach that target. Production AI training data crawlers operate at 50,000 to 500,000 pages per hour, which requires distributed architecture from day one.

Data quality is measured differently. In a product scraping use case, data quality means accuracy — does the price match the actual listed price? In an training data context, quality means distributional properties, factual density, linguistic naturalness, topical coverage, and the absence of toxic or privacy-violating content. A perfectly accurate copy of a spam page is high-quality in a price scraper but toxic in a language model training corpus. The quality filtering stage of an AI training data pipeline is more computationally expensive than the crawling stage itself.

Deduplication is a first-class concern. The web is massively redundant. Common Crawl analysis shows that after URL-level deduplication, near-duplicate content (same article, different URL, minor wording changes) accounts for 60–70% of remaining content by page count. Feeding a large language model on heavily duplicated training data produces models that regurgitate memorized text, have poor calibration, and perform worse on benchmarks than models trained on the same compute budget with properly deduplicated data. Near-deduplication at scale requires serious infrastructure.

Provenance and compliance metadata are mandatory. Unlike a scraping pipeline that collects live prices for a dashboard, an AI training data pipeline needs to maintain auditable records of where every document came from, when it was collected, what version of the robots.txt was in effect, and whether the domain had opted out of AI training through signals like the noai and noimageai meta tags introduced in 2023. As the EU AI Act’s training data transparency requirements come into full effect in 2026, provenance metadata is not optional — it is a legal requirement for any GPAI model provider operating in European markets.

Multimodal data collection adds image-text pairing requirements. Vision-language models like GPT-4V require paired image-text training pairs where the textual description and the image are aligned in content and quality. Scraping for this type of AI training data involves extracting alt text, captions, surrounding context, and structured image metadata simultaneously — a substantially more complex data collection pipeline than text-only crawling.

Common Crawl: The Backbone of AI Training Data

Before building any proprietary crawling infrastructure, every team working on large language model training needs to thoroughly understand Common Crawl. It is the most significant public resource in the AI training data ecosystem and the primary raw material for most publicly documented training corpora.

Common Crawl is a non-profit organization that has been crawling the web continuously since 2008. Each monthly crawl covers approximately 3.5 billion pages, producing roughly 250 terabytes of compressed WARC (Web ARChive) files per crawl. The total archive now exceeds 250 petabytes. It is freely available on Amazon S3 at s3://commoncrawl/ and accessible for free in the US-East-1 region.

The raw Common Crawl data is not usable directly for large language model training. The standard processing pipeline that transforms raw WARC files into a training-ready corpus involves at minimum: HTML stripping to extract body text, language identification (Common Crawl is multilingual at the page level but not identified), exact and near-deduplication, quality filtering using heuristics or perplexity-based scoring, and tokenization. The C4 dataset — the cleaned English subset derived from Common Crawl that underpins the T5 and Flan-T5 families — reduced 27 billion raw pages to approximately 750GB of training-ready text through these filtering stages. The Colossal Clean Crawled Corpus applies roughly 100× compression to raw crawl data by filtering.

The practical implication for teams building large language model training pipelines: Common Crawl is the right starting point for broad-coverage, general-purpose large language model training data, but it requires substantial engineering investment to clean and filter. For domain-specific AI training data — code, legal text, scientific literature, financial documents — proprietary web scraping pipelines targeting high-quality domain-specific sources will always outperform filtered Common Crawl data for that domain.

The dominant industry pattern in 2026 is a two-tier approach: Common Crawl-derived general web data forms the base layer (60–80% of the total corpus by token count), and proprietary high-quality domain-specific crawling fills the remaining 20–40% with denser, higher-quality signal. This is the architecture documented for Llama 3, Falcon, and most open-weight models that have published training data methodology.

Accessing Common Crawl Programmatically

# Prerequisites:
# python -m venv .commoncrawl-env
# source .commoncrawl-env/bin/activate
# pip install boto3 warcio trafilatura langdetect

import boto3
import gzip
import json
import logging
from io import BytesIO
from typing import Generator, Optional
from warcio.archiveiterator import ArchiveIterator
import trafilatura

logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

# Use anonymous access for Common Crawl — it's free from us-east-1
s3 = boto3.client(
    "s3",
    region_name="us-east-1",
    config=boto3.session.Config(signature_version=boto3.session.UNSIGNED),
)

BUCKET = "commoncrawl"


def fetch_wet_index(crawl_id: str = "CC-MAIN-2025-51") -> list[str]:
    """
    Returns all WET segment paths for a given Common Crawl crawl ID.
    WET files contain extracted plain text — easier for AI training data
    pipelines than raw WARC files which include HTML markup.
    """
    index_path = f"crawl-data/{crawl_id}/wet.paths.gz"
    response = s3.get_object(Bucket=BUCKET, Key=index_path)
    compressed = response["Body"].read()
    paths = gzip.decompress(compressed).decode("utf-8").strip().split("\n")
    logger.info(f"Found {len(paths)} WET segments in {crawl_id}")
    return paths


def stream_wet_segment(s3_path: str) -> Generator[dict, None, None]:
    """
    Streams records from a single Common Crawl WET segment.
    Yields dicts with url, content, and content_length.
    Uses streaming to avoid loading full 200MB+ segments into memory.
    """
    response = s3.get_object(Bucket=BUCKET, Key=s3_path)
    stream = response["Body"]

    with gzip.GzipFile(fileobj=BytesIO(stream.read())) as gz:
        for record in ArchiveIterator(gz):
            if record.rec_type != "conversion":
                continue
            url = record.rec_headers.get_header("WARC-Target-URI")
            try:
                content = record.content_stream().read().decode("utf-8", errors="replace")
            except Exception as e:
                logger.warning(f"Failed to decode content for {url}: {e}")
                continue

            if len(content.strip()) < 200:  # Skip near-empty pages
                continue

            yield {
                "url": url,
                "content": content,
                "content_length": len(content),
                "segment": s3_path,
            }


def quality_filter(record: dict, min_words: int = 100, max_repetition_ratio: float = 0.3) -> bool:
    """
    Basic quality filter for Common Crawl WET records.
    Returns True if the record passes quality thresholds.

    These heuristics are based on the C4 filtering methodology.
    """
    content = record["content"]
    words = content.split()
    word_count = len(words)

    if word_count < min_words:
        return False

    # Detect high repetition (spam/SEO content)
    unique_ratio = len(set(words)) / word_count
    if unique_ratio < (1 - max_repetition_ratio):
        return False

    # C4 heuristic: filter pages where most lines end with punctuation
    # (proxy for template-heavy or listicle spam)
    lines = [l.strip() for l in content.split("\n") if l.strip()]
    if not lines:
        return False

    ending_punct = sum(1 for l in lines if l and l[-1] in ".!?:;,")
    if ending_punct / len(lines) < 0.2:
        return False

    # Filter pages with excessive curly braces (code/JSON templates)
    if content.count("{") > word_count * 0.02:
        return False

    return True


if __name__ == "__main__":
    paths = fetch_wet_index()
    # Process first segment as a test
    count = 0
    for record in stream_wet_segment(paths[0]):
        if quality_filter(record):
            count += 1
            if count <= 3:
                print(f"URL: {record['url']}")
                print(f"Words: {len(record['content'].split())}")
                print("---")
    print(f"Passed quality filter: {count}")

Architecture of a Production AI Training Data Pipeline

A production data collection pipeline for AI training data is not a single monolithic scraper. It is a multi-stage distributed system with distinct components that can be scaled, monitored, and operated independently. Understanding this architecture is the prerequisite for building any serious web scraping for AI infrastructure.

The canonical architecture consists of five stages, each with its own throughput requirements, failure modes, and scaling strategies.

Stage 1: URL Frontier Management — The frontier is the queue of URLs to be crawled, along with metadata about crawl priority, recrawl frequency, domain politeness constraints, and deduplication state. At the scale required for large language model training, a naive in-memory queue fails immediately. Production frontiers use Redis sorted sets or Apache Kafka partitioned by domain hash, allowing hundreds of crawling workers to consume from the same frontier without coordination overhead. The frontier must also maintain a deduplication index — typically a Bloom filter or Redis set of seen URL hashes — to prevent recrawling already-visited pages.

Stage 2: Distributed HTTP Crawling — The crawling tier fetches pages at high throughput. For AI corpus purposes, the majority of pages can be fetched with a fast HTTP client (no JavaScript rendering required) since training corpora emphasize text content. The Scrapy framework with its Twisted async I/O engine remains the dominant choice for this tier, achieving 300–600 pages per second on a single 8-core server against cooperative targets. For JavaScript-rendered content, a Playwright tier runs in parallel at lower throughput but higher JavaScript fidelity.

Stage 3: Content Extraction and Parsing — Raw HTML must be converted to clean text. This stage involves tag stripping, boilerplate removal (navigation menus, footers, cookie banners, advertisements), encoding normalization, and language identification. The trafilatura library (Python) is the current open-source standard for main content extraction, outperforming BeautifulSoup-based approaches on F1 score for identifying the primary article body.

Stage 4: Quality Filtering and Deduplication — The most computationally intensive stage. Exact deduplication uses SHA-256 hashing of normalized document text. Near-deduplication uses MinHash LSH with configurable Jaccard similarity thresholds (typically 0.8 or higher). Quality filtering uses a combination of rule-based heuristics (similar to the C4 methodology), perplexity scoring against a reference language model, and domain-reputation scoring. A well-designed quality filter reduces raw crawl volume by 60–90% while retaining the high-quality subset needed for effective large language model training.

Stage 5: Storage, Indexing, and Dataset Serving — Filtered documents are written to object storage (S3, GCS) in structured formats. The standard is JSONL (one JSON document per line) with fields for text, URL, crawl timestamp, language, quality score, and provenance metadata. For large datasets exceeding 1TB, sharded Parquet files with columnar metadata indexes are preferred because they enable efficient subset sampling by language, domain, quality score, or date range without loading the full corpus.

Stage 1: Building a Production URL Frontier

The URL frontier is the heart of the data collection pipeline. A poorly designed frontier results in crawling the same domains repeatedly, missing high-value targets, and creating politeness violations that result in IP bans. Here is a production-grade frontier implementation backed by Redis.

# Prerequisites:
# python -m venv .frontier-env
# source .frontier-env/bin/activate
# pip install redis aioredis tldextract hashlib
# Redis must be running: docker run -d -p 6379:6379 redis:7-alpine

import asyncio
import hashlib
import time
import tldextract
import logging
from dataclasses import dataclass
from typing import Optional
import redis.asyncio as aioredis

logger = logging.getLogger(__name__)


@dataclass
class FrontierURL:
    url: str
    priority: float  # Higher = crawl sooner
    domain: str
    depth: int = 0
    source: str = "seed"  # seed | extracted | sitemap


class DistributedFrontier:
    """
    Redis-backed URL frontier for distributed AI training data crawling.

    Architecture:
    - Sorted set per domain: {url_hash: priority} — enables politeness delays
    - Global priority queue: {domain: next_fetch_time} — domain-level rate limiting
    - Bloom filter simulation via Redis SET: seen URL hashes for deduplication

    Politeness: respects per-domain crawl delay (default 1s minimum).
    The delay is essential both for ethical crawling and for avoiding IP bans
    that would interrupt your AI training data collection pipeline.
    """

    SEEN_KEY = "frontier:seen"
    DOMAIN_QUEUE_KEY = "frontier:domains"
    DOMAIN_URL_PREFIX = "frontier:urls:"
    DOMAIN_DELAY_PREFIX = "frontier:delay:"
    DEFAULT_CRAWL_DELAY = 1.5  # seconds between requests to the same domain

    def __init__(self, redis_url: str = "redis://localhost:6379"):
        self.redis_url = redis_url
        self._redis: Optional[aioredis.Redis] = None

    async def connect(self):
        self._redis = await aioredis.from_url(self.redis_url, decode_responses=True)
        logger.info("Connected to Redis frontier")

    async def close(self):
        if self._redis:
            await self._redis.aclose()

    def _url_hash(self, url: str) -> str:
        return hashlib.sha256(url.encode()).hexdigest()[:16]

    def _extract_domain(self, url: str) -> str:
        ext = tldextract.extract(url)
        return f"{ext.domain}.{ext.suffix}"

    async def add_url(self, url: str, priority: float = 1.0, depth: int = 0) -> bool:
        """
        Add a URL to the frontier. Returns False if already seen.
        Priority should be higher for domains known to contain dense AI training data
        (e.g., Wikipedia, arXiv, GitHub) and lower for general-purpose pages.
        """
        url_hash = self._url_hash(url)

        # Atomic check-and-set using Redis SETNX (set if not exists)
        is_new = await self._redis.sadd(self.SEEN_KEY, url_hash)
        if not is_new:
            return False  # Already seen

        domain = self._extract_domain(url)
        domain_key = f"{self.DOMAIN_URL_PREFIX}{domain}"

        # Add URL to domain-specific queue with priority score
        await self._redis.zadd(domain_key, {url: priority})

        # Register domain in global queue if not already present
        # Score = earliest next-fetch time (now = can fetch immediately)
        await self._redis.zadd(
            self.DOMAIN_QUEUE_KEY, {domain: time.time()}, nx=True
        )

        return True

    async def add_seed_urls(self, urls: list[str]) -> int:
        """Bulk-add seed URLs with high priority. Returns count of new URLs added."""
        added = 0
        for url in urls:
            if await self.add_url(url, priority=10.0):
                added += 1
        logger.info(f"Added {added}/{len(urls)} new seed URLs to frontier")
        return added

    async def get_next_url(self) -> Optional[FrontierURL]:
        """
        Returns the next URL to crawl, respecting per-domain politeness delays.
        Returns None if all domains are currently rate-limited.
        """
        now = time.time()

        # Get domains that are ready to crawl (next_fetch_time <= now)
        ready_domains = await self._redis.zrangebyscore(
            self.DOMAIN_QUEUE_KEY, "-inf", now, start=0, num=10
        )

        for domain in ready_domains:
            domain_key = f"{self.DOMAIN_URL_PREFIX}{domain}"

            # Get highest-priority URL for this domain (highest score = highest priority)
            urls = await self._redis.zrevrange(domain_key, 0, 0, withscores=True)
            if not urls:
                # Domain queue empty — remove from global queue
                await self._redis.zrem(self.DOMAIN_QUEUE_KEY, domain)
                continue

            url, priority = urls[0]

            # Remove from domain queue
            await self._redis.zrem(domain_key, url)

            # Schedule next crawl for this domain (enforce politeness delay)
            next_fetch = time.time() + self.DEFAULT_CRAWL_DELAY
            await self._redis.zadd(self.DOMAIN_QUEUE_KEY, {domain: next_fetch}, xx=True)

            return FrontierURL(
                url=url,
                priority=priority,
                domain=domain,
            )

        return None  # All domains rate-limited

    async def queue_size(self) -> int:
        """Returns approximate total URLs in frontier (across all domains)."""
        domains = await self._redis.zrange(self.DOMAIN_QUEUE_KEY, 0, -1)
        total = 0
        for domain in domains:
            domain_key = f"{self.DOMAIN_URL_PREFIX}{domain}"
            total += await self._redis.zcard(domain_key)
        return total


# --- Seed URL generator for AI training data targets ---

AI_TRAINING_SEED_DOMAINS = [
    # High-quality text sources for general large language model training
    "https://en.wikipedia.org/wiki/Main_Page",
    "https://arxiv.org/list/cs.AI/2026",
    "https://arxiv.org/list/cs.CL/2026",
    "https://github.com/trending",
    "https://stackoverflow.com/questions?sort=votes",
    "https://news.ycombinator.com/",
    "https://www.reddit.com/r/MachineLearning/hot.json",
    "https://proceedings.mlr.press/",
    "https://aclanthology.org/",
]


async def main():
    frontier = DistributedFrontier()
    await frontier.connect()

    added = await frontier.add_seed_urls(AI_TRAINING_SEED_DOMAINS)
    size = await frontier.queue_size()
    print(f"Frontier initialized: {added} seeds, {size} total URLs")

    # Demonstrate fetching next URL
    next_url = await frontier.get_next_url()
    if next_url:
        print(f"Next URL to crawl: {next_url.url} (priority: {next_url.priority})")

    await frontier.close()


if __name__ == "__main__":
    asyncio.run(main())

Stage 2: Distributed Crawling at Scale with Scrapy-Redis

The crawling tier is where raw throughput is generated. For AI training data at the scale of hundreds of millions of pages, the data collection pipeline must be horizontally scalable — adding more crawling workers should increase throughput linearly with no architectural changes. The Scrapy-Redis pattern achieves this.

# Prerequisites:
# python -m venv .scrapy-ai-env
# source .scrapy-ai-env/bin/activate
# pip install scrapy scrapy-redis trafilatura langdetect tldextract
#
# settings.py additions for distributed mode:
# SCHEDULER = "scrapy_redis.scheduler.Scheduler"
# DUPEFILTER_CLASS = "scrapy_redis.dupefilter.RFPDupeFilter"
# REDIS_URL = "redis://localhost:6379"
# SCHEDULER_PERSIST = True

import scrapy
import hashlib
import json
import langdetect
import trafilatura
from datetime import datetime, timezone
from typing import Optional
from urllib.parse import urlparse


class AITrainingDataSpider(scrapy.Spider):
    """
    Production-grade spider for AI training data collection.
    Designed to run as multiple workers sharing a Redis frontier.
    """

    name = "ai_training_corpus"

    # Minimum content length to consider a page worth extracting
    MIN_CONTENT_CHARS = 500
    # Target languages for this crawl (adjust for multilingual corpora)
    TARGET_LANGUAGES = {"en", "de", "fr", "es", "zh", "ja", "pt", "it"}

    custom_settings = {
        "CONCURRENT_REQUESTS": 64,
        "CONCURRENT_REQUESTS_PER_DOMAIN": 2,  # Politeness per domain
        "DOWNLOAD_DELAY": 0.5,
        "AUTOTHROTTLE_ENABLED": True,
        "AUTOTHROTTLE_START_DELAY": 0.5,
        "AUTOTHROTTLE_TARGET_CONCURRENCY": 32.0,
        "ROBOTSTXT_OBEY": True,   # Always respect robots.txt for AI training data
        "COOKIES_ENABLED": False,  # No session state needed for content crawling
        "RETRY_TIMES": 2,
        "HTTPCACHE_ENABLED": False,
        "DOWNLOADER_MIDDLEWARES": {
            "scrapy.downloadermiddlewares.useragent.UserAgentMiddleware": None,
            "ai_crawler.middlewares.RotatingUserAgentMiddleware": 400,
            "ai_crawler.middlewares.AIOptOutMiddleware": 450,  # Respect noai signals
        },
        "ITEM_PIPELINES": {
            "ai_crawler.pipelines.QualityFilterPipeline": 100,
            "ai_crawler.pipelines.DeduplicationPipeline": 200,
            "ai_crawler.pipelines.JSONLWriterPipeline": 300,
        },
        "DEFAULT_REQUEST_HEADERS": {
            "Accept": "text/html,application/xhtml+xml",
            "Accept-Language": "en-US,en;q=0.9",
            # Identify yourself as an AI training crawler — ethical practice
            # and increasingly a legal requirement in some jurisdictions
            "User-Agent": (
                "DataFlirtBot/1.0 (AI training data collection; "
                "+https://dataflirt.com/bot-policy)"
            ),
        },
    }

    def parse(self, response):
        """
        Main parse method: extract content, check quality, follow links.
        """
        url = response.url
        html = response.text

        # Check for AI training opt-out meta tags
        # Introduced in 2023 and increasingly common on quality sites
        noai_tag = response.css('meta[name="robots"]::attr(content)').get("")
        if "noai" in noai_tag.lower() or "noimageai" in noai_tag.lower():
            self.logger.debug(f"Skipping {url}: noai meta tag present")
            return

        # Extract main content using trafilatura (superior to BS4 for article extraction)
        content = trafilatura.extract(
            html,
            include_comments=False,
            include_tables=True,
            no_fallback=False,
            favor_precision=True,
        )

        if not content or len(content) < self.MIN_CONTENT_CHARS:
            return

        # Language detection — skip pages not in target language set
        try:
            detected_lang = langdetect.detect(content)
        except langdetect.lang_detect_exception.LangDetectException:
            return

        if detected_lang not in self.TARGET_LANGUAGES:
            return

        # Extract title and metadata
        title = response.css("title::text").get("").strip()
        description = response.css('meta[name="description"]::attr(content)').get("")

        yield {
            "url": url,
            "title": title,
            "description": description,
            "content": content,
            "content_hash": hashlib.sha256(content.encode()).hexdigest(),
            "language": detected_lang,
            "domain": urlparse(url).netloc,
            "crawled_at": datetime.now(timezone.utc).isoformat(),
            "content_length": len(content),
            "word_count": len(content.split()),
        }

        # Follow links for broader corpus coverage
        # Restrict to same domain to maintain topical coherence
        domain = urlparse(url).netloc
        for link in response.css("a::attr(href)").getall():
            absolute = response.urljoin(link)
            if urlparse(absolute).netloc == domain:
                yield response.follow(absolute, callback=self.parse)


class AIOptOutMiddleware:
    """
    Middleware that checks X-Robots-Tag response headers for AI training opt-out signals.
    This is the HTTP header equivalent of the meta tag approach.
    """

    OPT_OUT_SIGNALS = {"noai", "noimageai", "noindex"}

    def process_response(self, request, response, spider):
        robots_header = response.headers.get("X-Robots-Tag", b"").decode().lower()
        if any(signal in robots_header for signal in self.OPT_OUT_SIGNALS):
            spider.logger.debug(
                f"Dropping {request.url}: X-Robots-Tag opt-out detected"
            )
            return scrapy.http.Response(url=request.url, status=403)
        return response

Stage 3: Content Extraction and Quality Filtering at Scale

Raw HTML is not AI training data. The extraction stage must produce clean, structured text that a tokenizer can process efficiently. The quality filtering stage must remove noise, spam, and low-value content that would degrade large language model training.

The two most important quality signals for AI training data are perplexity-based scoring and deduplication at the near-duplicate level. Let us implement both.

# Prerequisites:
# pip install datasketch numpy scipy scikit-learn kenlm sentencepiece

import hashlib
import numpy as np
from datasketch import MinHash, MinHashLSH
from collections import defaultdict
from typing import Optional
import re
import unicodedata


class DocumentNormalizer:
    """
    Normalizes document text before deduplication and quality scoring.
    Consistent normalization is critical: two identical documents
    that differ only in whitespace or unicode form would not be
    detected as duplicates without normalization.
    """

    def normalize(self, text: str) -> str:
        # Unicode normalization (NFC form)
        text = unicodedata.normalize("NFC", text)
        # Collapse whitespace
        text = re.sub(r"\s+", " ", text).strip()
        # Lowercase for signature comparison (not for final output)
        text = text.lower()
        return text

    def get_shingles(self, text: str, k: int = 5) -> set[str]:
        """
        Returns k-word shingles for MinHash computation.
        5-word shingles are the standard for document-level
        near-duplicate detection in AI training data pipelines.
        """
        words = text.split()
        if len(words) < k:
            return {text}
        return {" ".join(words[i : i + k]) for i in range(len(words) - k + 1)}


class MinHashDeduplicator:
    """
    LSH-based near-duplicate deduplication for AI training data corpora.

    This is the standard approach for removing near-duplicates before
    large language model training. The Jaccard similarity threshold of 0.8
    means documents that share 80% or more of their 5-gram shingles
    are considered duplicates.

    At billion-page scale, MinHash LSH makes this O(n) rather than O(n²).
    Facebook Research demonstrated that deduplication at this threshold
    improves LLM downstream benchmarks by 3–7% with no other changes.
    """

    def __init__(
        self,
        num_perm: int = 128,  # Number of hash permutations — higher = more accurate
        threshold: float = 0.8,  # Jaccard similarity threshold for duplicate detection
        lsh_bands: int = 16,
        lsh_rows: int = 8,
    ):
        self.num_perm = num_perm
        self.threshold = threshold
        self.lsh = MinHashLSH(threshold=threshold, num_perm=num_perm)
        self.normalizer = DocumentNormalizer()
        self.doc_count = 0
        self.duplicate_count = 0

    def _compute_minhash(self, text: str) -> MinHash:
        """Compute MinHash signature for a document."""
        normalized = self.normalizer.normalize(text)
        shingles = self.normalizer.get_shingles(normalized)
        m = MinHash(num_perm=self.num_perm)
        for shingle in shingles:
            m.update(shingle.encode("utf-8"))
        return m

    def is_duplicate(self, doc_id: str, text: str) -> bool:
        """
        Returns True if this document is a near-duplicate of a
        previously seen document. Side effect: adds non-duplicate
        documents to the LSH index.
        """
        minhash = self._compute_minhash(text)
        self.doc_count += 1

        try:
            result = self.lsh.query(minhash)
        except Exception:
            result = []

        if result:
            self.duplicate_count += 1
            return True

        # Not a duplicate — add to index
        try:
            self.lsh.insert(doc_id, minhash)
        except ValueError:
            pass  # doc_id already in index (race condition in concurrent use)

        return False

    @property
    def duplicate_rate(self) -> float:
        if self.doc_count == 0:
            return 0.0
        return self.duplicate_count / self.doc_count


class HeuristicQualityScorer:
    """
    Rule-based quality scorer for AI training data.
    Implements heuristics from the C4, RefinedWeb, and Dolma methodologies.

    Returns a float from 0.0 (reject) to 1.0 (high quality).
    Documents below 0.5 should be filtered from the training corpus.
    """

    def score(self, text: str) -> dict:
        words = text.split()
        lines = [l.strip() for l in text.split("\n") if l.strip()]
        word_count = len(words)

        if word_count < 50:
            return {"score": 0.0, "reason": "too_short"}

        scores = {}

        # 1. Word count heuristic (log-scale, normalized 0-1)
        scores["length"] = min(1.0, np.log(word_count) / np.log(10000))

        # 2. Average word length (2-12 chars = normal, outliers = spam/code dumps)
        avg_word_len = np.mean([len(w) for w in words])
        scores["word_length"] = 1.0 if 3 <= avg_word_len <= 10 else 0.3

        # 3. Unique word ratio (low = repetitive spam)
        unique_ratio = len(set(w.lower() for w in words)) / word_count
        scores["uniqueness"] = min(1.0, unique_ratio * 2)

        # 4. Punctuation ratio (natural language has ~5-15% punctuation chars)
        punct_chars = sum(1 for c in text if c in ".,;:!?\"'-()")
        punct_ratio = punct_chars / max(len(text), 1)
        scores["punctuation"] = 1.0 if 0.02 <= punct_ratio <= 0.20 else 0.4

        # 5. Digit ratio (very high digit ratio = data dump / boilerplate)
        digit_ratio = sum(c.isdigit() for c in text) / max(len(text), 1)
        scores["digits"] = 1.0 if digit_ratio < 0.15 else max(0.0, 1 - digit_ratio * 5)

        # 6. Line length consistency (extreme variation = boilerplate)
        if len(lines) > 1:
            line_lens = [len(l) for l in lines]
            line_len_std = float(np.std(line_lens))
            scores["line_consistency"] = 1.0 if line_len_std < 200 else 0.5
        else:
            scores["line_consistency"] = 0.8

        # 7. Check for lorem ipsum or placeholder content
        if "lorem ipsum" in text.lower():
            return {"score": 0.0, "reason": "placeholder_content"}

        # 8. Excessive URL density (link farm / SEO spam)
        url_count = text.lower().count("http")
        url_density = url_count / max(word_count, 1)
        scores["url_density"] = 1.0 if url_density < 0.05 else max(0.0, 1 - url_density * 10)

        # Weighted aggregate
        weights = {
            "length": 0.15,
            "word_length": 0.15,
            "uniqueness": 0.25,
            "punctuation": 0.15,
            "digits": 0.10,
            "line_consistency": 0.10,
            "url_density": 0.10,
        }

        final_score = sum(scores[k] * weights[k] for k in weights)

        return {
            "score": round(final_score, 4),
            "components": scores,
            "word_count": word_count,
        }

Stage 4: Deduplication Pipeline Integration with Scrapy

The deduplication and quality scoring components need to be wired into the Scrapy item pipeline to operate inline as documents are extracted.

# pipelines.py

import json
import os
import hashlib
import logging
from pathlib import Path
from typing import Optional

logger = logging.getLogger(__name__)


class QualityFilterPipeline:
    """
    Scrapy pipeline: drops low-quality documents from the training stream.
    Minimum score of 0.50 is a reasonable starting threshold.
    """

    MIN_QUALITY_SCORE = 0.50
    MIN_WORD_COUNT = 100

    def __init__(self):
        self.scorer = HeuristicQualityScorer()
        self.total = 0
        self.passed = 0

    def process_item(self, item, spider):
        self.total += 1
        result = self.scorer.score(item["content"])

        if result["score"] < self.MIN_QUALITY_SCORE:
            spider.logger.debug(
                f"Filtered (quality={result['score']:.2f}): {item['url']}"
            )
            from scrapy.exceptions import DropItem
            raise DropItem(f"Quality score too low: {result['score']:.2f}")

        if result["word_count"] < self.MIN_WORD_COUNT:
            from scrapy.exceptions import DropItem
            raise DropItem(f"Word count too low: {result['word_count']}")

        item["quality_score"] = result["score"]
        item["word_count"] = result["word_count"]
        self.passed += 1
        return item


class DeduplicationPipeline:
    """
    Scrapy pipeline: performs exact and near-duplicate detection.
    Uses MinHash LSH for near-deduplication within the current crawl session.
    Cross-session deduplication should be handled in a post-processing step
    against the full corpus fingerprint database.
    """

    def __init__(self):
        self.deduplicator = MinHashDeduplicator()
        self.seen_hashes = set()  # Exact dedup
        self.total = 0
        self.exact_dupes = 0
        self.near_dupes = 0

    def process_item(self, item, spider):
        from scrapy.exceptions import DropItem
        self.total += 1
        content_hash = item["content_hash"]

        # Exact deduplication first (O(1) hash lookup)
        if content_hash in self.seen_hashes:
            self.exact_dupes += 1
            raise DropItem(f"Exact duplicate: {item['url']}")
        self.seen_hashes.add(content_hash)

        # Near-deduplication via MinHash LSH
        if self.deduplicator.is_duplicate(content_hash, item["content"]):
            self.near_dupes += 1
            raise DropItem(f"Near-duplicate: {item['url']}")

        return item

    def close_spider(self, spider):
        logger.info(
            f"Deduplication summary: {self.total} total, "
            f"{self.exact_dupes} exact dupes, "
            f"{self.near_dupes} near-dupes, "
            f"effective dedup rate: {(self.exact_dupes + self.near_dupes) / max(self.total, 1):.1%}"
        )


class JSONLWriterPipeline:
    """
    Scrapy pipeline: writes clean, quality-filtered AI training data to sharded JSONL files.
    Each shard is 512MB uncompressed — a practical size for distributed training data loading.
    """

    SHARD_SIZE_BYTES = 512 * 1024 * 1024  # 512MB per shard

    def __init__(self, output_dir: str = "./ai_training_data"):
        self.output_dir = Path(output_dir)
        self.output_dir.mkdir(exist_ok=True)
        self.shard_index = 0
        self.current_file = None
        self.current_size = 0
        self.doc_count = 0

    @classmethod
    def from_crawler(cls, crawler):
        return cls(
            output_dir=crawler.settings.get("AI_DATA_OUTPUT_DIR", "./ai_training_data")
        )

    def open_new_shard(self):
        if self.current_file:
            self.current_file.close()
        shard_path = self.output_dir / f"shard_{self.shard_index:06d}.jsonl"
        self.current_file = open(shard_path, "w", encoding="utf-8")
        self.current_size = 0
        logger.info(f"Opened new AI training data shard: {shard_path}")
        self.shard_index += 1

    def open_spider(self, spider):
        self.open_new_shard()

    def process_item(self, item, spider):
        line = json.dumps(
            {
                "text": item["content"],
                "url": item["url"],
                "title": item.get("title", ""),
                "language": item.get("language", "en"),
                "quality_score": item.get("quality_score", 0.0),
                "word_count": item.get("word_count", 0),
                "domain": item.get("domain", ""),
                "crawled_at": item.get("crawled_at", ""),
                "content_hash": item.get("content_hash", ""),
            },
            ensure_ascii=False,
        )
        encoded = (line + "\n").encode("utf-8")
        self.current_file.write(line + "\n")
        self.current_size += len(encoded)
        self.doc_count += 1

        if self.current_size >= self.SHARD_SIZE_BYTES:
            self.open_new_shard()

        return item

    def close_spider(self, spider):
        if self.current_file:
            self.current_file.close()
        logger.info(
            f"AI training data pipeline complete: {self.doc_count} documents "
            f"across {self.shard_index} shards in {self.output_dir}"
        )

LLM-Augmented AI Training Data Collection

The most significant architectural evolution in web scraping for AI infrastructure in 2025–2026 is the deployment of large language models inside the data collection pipeline itself. This creates a virtuous cycle: you use AI to collect better AI training data.

There are three practical deployment patterns for LLMs within an AI training data pipeline.

Pattern 1: Schema-Free Structured Extraction. Instead of writing CSS selectors that break whenever a target website redesigns, you pipe raw HTML into a language model and ask it to extract structured fields. This is particularly valuable for domain-specific AI training data collection where the sources are well-known but structurally heterogeneous — academic paper repositories, legal databases, financial filings, clinical trial records.

Pattern 2: Quality Scoring and Relevance Classification. A language model deployed as a classifier can evaluate whether a document is relevant to a specific domain, whether it contains factually dense content, and whether its writing quality is sufficient for inclusion in a fine-tuning or instruction-tuning dataset. This is more expensive than heuristic scoring but more accurate for nuanced domain relevance judgments.

Pattern 3: Synthetic Data Augmentation. Given a sample of high-quality scraped documents, a language model can generate additional examples that follow the same style, domain, and format but are not derived from any specific source. This is increasingly important for filling distributional gaps in AI training data — covering low-resource languages, rare technical domains, or specific instruction-following formats — without additional crawling.

LLM Extraction with Google GenAI SDK (Gemini 3.1 Flash)

# Prerequisites:
# pip install google-genai trafilatura httpx

import asyncio
import json
import httpx
import trafilatura
from google import genai
from google.genai import types

# Authenticates via GOOGLE_API_KEY environment variable
# export GOOGLE_API_KEY=your_api_key_here
client = genai.Client()


EXTRACTION_SYSTEM_PROMPT = """You are a data extraction engine for an AI training data pipeline.
Your task is to extract structured information from web page content.
Respond ONLY with valid JSON matching the requested schema. No preamble, no explanation."""


async def fetch_page(url: str) -> str | None:
    """Fetch page HTML with a realistic browser user-agent."""
    headers = {
        "User-Agent": (
            "Mozilla/5.0 (Windows NT 10.0; Win64; x64) "
            "AppleWebKit/537.36 (KHTML, like Gecko) "
            "Chrome/124.0.0.0 Safari/537.36"
        ),
        "Accept-Language": "en-US,en;q=0.9",
    }
    try:
        async with httpx.AsyncClient(timeout=15.0, follow_redirects=True) as client_http:
            response = await client_http.get(url, headers=headers)
            response.raise_for_status()
            return response.text
    except Exception as e:
        print(f"Fetch failed for {url}: {e}")
        return None


def extract_main_content(html: str) -> str | None:
    """Use trafilatura for initial content extraction before LLM processing."""
    return trafilatura.extract(
        html,
        include_comments=False,
        include_tables=True,
        favor_precision=True,
    )


async def llm_extract_structured(
    content: str,
    extraction_schema: dict,
    url: str = "",
) -> dict | None:
    """
    Uses Gemini 3.1 Flash to extract structured data from web content.
    Gemini Flash is the preferred model for AI training data pipeline extraction:
    - Large context window (handles full article text)
    - JSON output mode eliminates parsing errors
    - Cost-efficient for high-volume pipeline use
    - Significantly faster than Pro/Ultra variants at acceptable accuracy

    Caveat: Gemini 3.1 Flash was released in early 2026. Verify the exact
    model string via the Google GenAI SDK model list if you encounter 404 errors.
    """
    schema_str = json.dumps(extraction_schema, indent=2)
    prompt = (
        f"Extract the following fields from this web content as JSON.\n"
        f"URL: {url}\n\n"
        f"Required schema:\n{schema_str}\n\n"
        f"Content (first 20,000 chars):\n{content[:20000]}\n\n"
        f"Return ONLY valid JSON matching the schema exactly."
    )

    try:
        response = client.models.generate_content(
            model="gemini-3.1-flash",  # Use flash for cost-efficiency at scale
            contents=[types.Part.from_text(prompt)],
            config=types.GenerateContentConfig(
                system_instruction=EXTRACTION_SYSTEM_PROMPT,
                response_mime_type="application/json",  # Enforces JSON output mode
                temperature=0.1,  # Low temperature for deterministic extraction
                max_output_tokens=2048,
            ),
        )
        return json.loads(response.text)
    except json.JSONDecodeError as e:
        print(f"JSON parse error for {url}: {e}")
        return None
    except Exception as e:
        print(f"Gemini extraction error for {url}: {e}")
        return None


async def llm_quality_score(content: str) -> dict | None:
    """
    Uses Gemini to score content quality for AI training data inclusion.
    Returns a structured quality assessment with reasoning.
    """
    prompt = (
        "Evaluate this text for inclusion in an AI training dataset. "
        "Score each dimension from 0.0 to 1.0.\n\n"
        "Dimensions:\n"
        "- factual_density: proportion of sentences containing specific, verifiable facts\n"
        "- writing_quality: grammar, coherence, vocabulary richness\n"
        "- topical_coherence: is the document focused on a single topic?\n"
        "- information_density: ratio of new information to redundant or filler text\n"
        "- overall: weighted average recommendation for training data inclusion\n\n"
        f"Text (first 3000 chars):\n{content[:3000]}"
    )

    schema = {
        "factual_density": 0.0,
        "writing_quality": 0.0,
        "topical_coherence": 0.0,
        "information_density": 0.0,
        "overall": 0.0,
        "reasoning": "",
    }

    try:
        response = client.models.generate_content(
            model="gemini-3.1-flash",
            contents=[types.Part.from_text(prompt)],
            config=types.GenerateContentConfig(
                response_mime_type="application/json",
                temperature=0.1,
                max_output_tokens=512,
            ),
        )
        return json.loads(response.text)
    except Exception as e:
        print(f"Quality scoring error: {e}")
        return None


# Example: Extracting research paper metadata for academic AI training data
RESEARCH_PAPER_SCHEMA = {
    "title": "string",
    "authors": ["string"],
    "abstract": "string",
    "year": "integer",
    "venue": "string",
    "keywords": ["string"],
    "has_code": "boolean",
    "primary_topic": "string",
}


async def main():
    test_url = "https://arxiv.org/abs/2405.04434"
    html = await fetch_page(test_url)
    if not html:
        return

    content = extract_main_content(html)
    if not content:
        print("Could not extract content")
        return

    print(f"Extracted {len(content.split())} words of content")

    # Structured extraction
    data = await llm_extract_structured(content, RESEARCH_PAPER_SCHEMA, test_url)
    if data:
        print("Extracted structure:")
        print(json.dumps(data, indent=2))

    # Quality scoring
    quality = await llm_quality_score(content)
    if quality:
        print(f"\nQuality assessment: {quality.get('overall', 0):.2f}")
        print(f"Reasoning: {quality.get('reasoning', '')[:200]}")


if __name__ == "__main__":
    asyncio.run(main())

LLM Extraction via Vertex AI (Gemini on Google Cloud)

For teams running their AI training data pipeline on Google Cloud Platform, Vertex AI provides managed access to Gemini models with enterprise SLA, built-in billing controls, and Data Processing Agreements that are essential for GDPR-compliant AI training data collection in the EU.

# Prerequisites:
# pip install google-cloud-aiplatform vertexai
# gcloud auth application-default login
# gcloud config set project YOUR_PROJECT_ID

import json
import asyncio
from typing import Optional
import vertexai
from vertexai.generative_models import GenerativeModel, GenerationConfig, Part

# Initialize Vertex AI — uses application default credentials
# Set your GCP project and region for AI training data processing
VERTEX_PROJECT = "your-gcp-project-id"
VERTEX_LOCATION = "us-central1"  # Choose region closest to your crawling infrastructure

vertexai.init(project=VERTEX_PROJECT, location=VERTEX_LOCATION)


class VertexAIExtractor:
    """
    Vertex AI-based content extractor for AI training data pipelines.
    
    Advantages over direct API access for production AI training data pipelines:
    - Automatic retry with exponential backoff
    - Built-in rate limit management
    - DPA coverage for GDPR Article 28 compliance
    - Native integration with Google Cloud Storage for large-scale pipeline output
    - Grounding with Google Search available for factual verification

    Note: For very high volume AI training data extraction (>10M pages/day),
    use Vertex AI Batch Prediction instead of online serving to reduce cost by ~50%.
    """

    def __init__(self, model_name: str = "gemini-3.1-flash-001"):
        self.model = GenerativeModel(
            model_name=model_name,
            system_instruction=(
                "You are a precision data extraction engine for AI training data curation. "
                "Always respond with valid JSON only. Never include explanatory text outside the JSON structure."
            ),
        )
        self.generation_config = GenerationConfig(
            temperature=0.05,
            max_output_tokens=4096,
            response_mime_type="application/json",
        )

    def extract_training_document(
        self,
        html_content: str,
        url: str,
        extraction_type: str = "article",
    ) -> Optional[dict]:
        """
        Extracts a structured training document from raw HTML.
        Supports multiple extraction types optimized for different AI training data sources.
        """
        prompts = {
            "article": (
                f"Extract the main article from this HTML as a training document.\n"
                f"URL: {url}\n\n"
                f"Return JSON with fields: title, body, summary, keywords (list), "
                f"content_type (article|tutorial|reference|discussion), quality_estimate (0.0-1.0).\n\n"
                f"HTML (first 15000 chars):\n{html_content[:15000]}"
            ),
            "qa_pair": (
                f"If this page contains a question and answer (forum, FAQ, StackOverflow-style), "
                f"extract it as a QA training pair for large language model instruction tuning.\n"
                f"URL: {url}\n\n"
                f"Return JSON with: question, answer, context, quality_estimate (0.0-1.0), "
                f"is_qa_pair (boolean).\n\n"
                f"HTML (first 10000 chars):\n{html_content[:10000]}"
            ),
            "code": (
                f"Extract code examples from this page for code model training data.\n"
                f"URL: {url}\n\n"
                f"Return JSON with: code_blocks (list of {{code, language, description}}), "
                f"programming_languages (list), quality_estimate (0.0-1.0).\n\n"
                f"HTML (first 20000 chars):\n{html_content[:20000]}"
            ),
        }

        prompt = prompts.get(extraction_type, prompts["article"])

        try:
            response = self.model.generate_content(
                [Part.from_text(prompt)],
                generation_config=self.generation_config,
            )
            return json.loads(response.text)
        except json.JSONDecodeError as e:
            print(f"JSON parse error ({extraction_type}): {e}")
            return None
        except Exception as e:
            print(f"Vertex AI error: {e}")
            return None


# Vertex AI Batch Prediction for high-volume AI training data extraction
def create_batch_extraction_job(
    input_gcs_path: str,
    output_gcs_path: str,
    model_name: str = "gemini-3.1-flash-001",
) -> str:
    """
    Creates a Vertex AI batch prediction job for large-scale AI training data extraction.
    Input: GCS path to JSONL file with {url, html_content, extraction_type} records
    Output: GCS path where extraction results are written

    Batch mode reduces cost by approximately 50% vs online serving and is the
    correct approach for processing millions of pages for AI training data.
    """
    from google.cloud import aiplatform

    batch_job = aiplatform.BatchPredictionJob.create(
        job_display_name=f"ai-training-data-extraction-{int(asyncio.get_event_loop().time())}",
        model_name=f"publishers/google/models/{model_name}",
        gcs_source=input_gcs_path,
        gcs_destination_prefix=output_gcs_path,
        dedicated_resources_machine_type="n1-standard-4",
        dedicated_resources_starting_replica_count=5,
        dedicated_resources_max_replica_count=50,
    )
    return batch_job.resource_name


# Example usage
if __name__ == "__main__":
    extractor = VertexAIExtractor()
    sample_html = "<html><body><h1>Introduction to Transformers</h1><p>The transformer architecture...</p></body></html>"
    result = extractor.extract_training_document(sample_html, "https://example.com/transformers", "article")
    if result:
        print(json.dumps(result, indent=2))

LLM Extraction with Anthropic Claude (Opus and Sonnet)

Claude’s long context window and instruction-following accuracy make it particularly effective for two stages of the AI training data pipeline: complex document structure extraction from heterogeneous sources, and nuanced quality assessment of domain-specific content.

# Prerequisites:
# pip install anthropic trafilatura httpx

import anthropic
import json
import asyncio
import httpx
from typing import Optional

# Authenticates via ANTHROPIC_API_KEY environment variable
# export ANTHROPIC_API_KEY=your_api_key_here
client = anthropic.Anthropic()


class ClaudeTrainingDataExtractor:
    """
    Claude-based extractor for AI training data pipelines.

    Model selection guidance:
    - claude-opus-4-6: Best quality for complex, heterogeneous sources where accuracy
      is critical (legal documents, scientific papers, technical manuals). Higher cost.
    - claude-sonnet-4-6: Best balance of quality and throughput for general web
      content extraction in high-volume AI training data pipelines. Recommended default.

    Note: For instruction-tuning dataset generation, Claude Opus produces significantly
    better synthetic QA pairs and reasoning chains than Sonnet. Use Opus for synthetic
    data generation and Sonnet for bulk extraction.
    """

    SYSTEM_PROMPT = (
        "You are a data extraction and quality assessment engine for an AI training data pipeline. "
        "You process web content and produce structured JSON output. "
        "Be precise, complete, and consistent. Never include text outside the JSON structure."
    )

    def __init__(self, model: str = "claude-sonnet-4-6"):
        self.model = model
        self.client = client

    def extract_document(
        self,
        content: str,
        url: str,
        schema: dict,
        max_content_chars: int = 30000,
    ) -> Optional[dict]:
        """
        Extracts structured fields from document content for AI training data collection.
        Claude's document-level understanding handles complex page structures that
        brittle CSS selectors would miss — tables, nested lists, inline citations.
        """
        content_truncated = content[:max_content_chars]
        if len(content) > max_content_chars:
            content_truncated += "\n[CONTENT TRUNCATED]"

        prompt = (
            f"Extract the following JSON schema from this web document.\n\n"
            f"URL: {url}\n"
            f"Schema to extract:\n{json.dumps(schema, indent=2)}\n\n"
            f"Content:\n{content_truncated}\n\n"
            f"Return ONLY the JSON object. Use null for fields not found in the content."
        )

        try:
            message = self.client.messages.create(
                model=self.model,
                max_tokens=4096,
                system=self.SYSTEM_PROMPT,
                messages=[{"role": "user", "content": prompt}],
            )
            raw = message.content[0].text.strip()
            # Strip any accidental markdown code fences
            raw = raw.replace("```json", "").replace("```", "").strip()
            return json.loads(raw)
        except json.JSONDecodeError as e:
            print(f"JSON parse error (Claude extraction): {e}")
            return None
        except anthropic.APIError as e:
            print(f"Anthropic API error: {e}")
            return None

    def generate_qa_pairs(
        self,
        document_content: str,
        n_pairs: int = 5,
        difficulty: str = "medium",
    ) -> list[dict]:
        """
        Uses Claude Opus to generate synthetic QA pairs from a scraped document.
        This is the standard approach for building instruction-tuning AI training data
        from web-scraped content when the raw text alone is insufficient.

        Claude Opus is strongly preferred over Sonnet for this task because:
        - Better reasoning chain generation
        - More nuanced question formulation
        - Higher factual accuracy in synthetic answers

        Caveats:
        - Generated QA pairs should be validated against the source document
        - Do not use for sensitive domains (medical, legal) without expert review
        - Rate limit aggressively — Opus is expensive at scale
        """
        prompt = (
            f"Generate {n_pairs} high-quality question-answer pairs from this document "
            f"for use as AI instruction-tuning training data.\n\n"
            f"Requirements:\n"
            f"- Difficulty: {difficulty} (easy/medium/hard/expert)\n"
            f"- Questions should require reading comprehension, not just keyword matching\n"
            f"- Answers should be complete, accurate, and grounded in the source text\n"
            f"- Avoid yes/no questions\n"
            f"- Include the specific document passage that supports each answer\n\n"
            f"Document:\n{document_content[:25000]}\n\n"
            f"Return a JSON array of objects with fields: "
            f"question, answer, supporting_passage, difficulty, question_type"
        )

        # Use Opus for highest quality synthetic training data generation
        opus_client_message = self.client.messages.create(
            model="claude-opus-4-6",
            max_tokens=8192,
            system=self.SYSTEM_PROMPT,
            messages=[{"role": "user", "content": prompt}],
        )

        try:
            raw = opus_client_message.content[0].text.strip()
            raw = raw.replace("```json", "").replace("```", "").strip()
            result = json.loads(raw)
            return result if isinstance(result, list) else []
        except json.JSONDecodeError:
            return []

    def assess_domain_relevance(
        self,
        content: str,
        target_domain: str,
        reasoning: bool = True,
    ) -> dict:
        """
        Uses Claude Sonnet to assess whether a document is relevant to a
        specific AI training data domain. More nuanced than keyword matching.

        Example target domains: "clinical oncology", "contract law", "GPU architecture",
        "options trading strategies", "Rust programming"
        """
        prompt = (
            f"Assess whether this document is relevant for AI training data "
            f"in the domain: '{target_domain}'\n\n"
            f"Return JSON with:\n"
            f"- relevance_score: float 0.0-1.0\n"
            f"- primary_topic: string (the document's actual main topic)\n"
            f"- domain_overlap: string (how this document overlaps with {target_domain})\n"
            f"- include_in_corpus: boolean (should this be included in the training corpus?)\n"
            f"- reasoning: string (brief explanation)\n\n"
            f"Document (first 3000 chars):\n{content[:3000]}"
        )

        try:
            message = self.client.messages.create(
                model=self.model,
                max_tokens=1024,
                system=self.SYSTEM_PROMPT,
                messages=[{"role": "user", "content": prompt}],
            )
            raw = message.content[0].text.strip().replace("```json", "").replace("```", "").strip()
            return json.loads(raw)
        except Exception as e:
            return {"relevance_score": 0.0, "include_in_corpus": False, "error": str(e)}


# Example: Building a domain-specific training corpus corpus for clinical documentation
if __name__ == "__main__":
    extractor = ClaudeTrainingDataExtractor(model="claude-sonnet-4-6")

    sample_content = """
    Transformer Architecture: Self-Attention and Positional Encoding

    The transformer model, introduced in 'Attention Is All You Need' (Vaswani et al., 2017),
    replaces recurrent layers entirely with self-attention mechanisms. Given query Q, key K,
    and value V matrices, attention is computed as: softmax(QK^T / sqrt(d_k)) * V.
    Multi-head attention applies this in parallel across h attention heads...
    """

    # Assess relevance for a specific AI training data domain
    relevance = extractor.assess_domain_relevance(
        sample_content,
        target_domain="deep learning research and neural architecture papers",
    )
    print("Domain relevance assessment:")
    print(json.dumps(relevance, indent=2))

    # Generate QA pairs for instruction tuning dataset
    qa_pairs = extractor.generate_qa_pairs(sample_content, n_pairs=3, difficulty="expert")
    print(f"\nGenerated {len(qa_pairs)} QA pairs for instruction tuning AI training data:")
    for i, pair in enumerate(qa_pairs, 1):
        print(f"\n{i}. Q: {pair.get('question', 'N/A')}")
        print(f"   A: {pair.get('answer', 'N/A')[:200]}...")

JavaScript in the Wild: Playwright for Dynamic AI Training Data Sources

Not all high-value AI training data sources are static HTML. Academic preprint servers, documentation platforms, interactive code environments, and social platforms all require JavaScript rendering to extract their content. For the headless browser tier of your data collection pipeline, Playwright is the current standard.

// Prerequisites:
// node -v  # require Node.js 18+
// npm init -y
// npm install playwright cheerio
// npx playwright install chromium

const { chromium } = require("playwright");
const cheerio = require("cheerio");
const fs = require("fs");
const path = require("path");
const crypto = require("crypto");

/**
 * Playwright-based AI training data extractor for JavaScript-rendered sources.
 * Designed for documentation sites, GitHub-style code repositories,
 * and academic conference proceedings that require JS execution.
 *
 * Note: This is the high-overhead tier of the AI training data pipeline.
 * Use only for sources that genuinely require JavaScript rendering.
 * The Scrapy HTTP tier should handle 80-90% of your crawl volume.
 */

const AI_TRAINING_UA =
  "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 " +
  "(KHTML, like Gecko) Chrome/124.0.0.0 Safari/537.36 DataFlirtCrawler/1.0";

async function createStealthContext(browser) {
  const context = await browser.newContext({
    userAgent: AI_TRAINING_UA,
    viewport: { width: 1366, height: 768 },
    locale: "en-US",
    timezoneId: "America/New_York",
    // Block unnecessary resources — reduces bandwidth by 60-70% for text-focused crawling
    // Images, fonts, and media are not needed for text AI training data collection
  });

  // Block resource types irrelevant to text AI training data
  await context.route(
    "**/*.{png,jpg,jpeg,gif,svg,ico,woff,woff2,mp4,mp3,pdf}",
    (route) => route.abort()
  );

  return context;
}

async function extractDocumentationPage(page, url) {
  await page.goto(url, {
    waitUntil: "domcontentloaded",
    timeout: 30000,
  });

  // Wait for common documentation content selectors
  try {
    await page.waitForSelector("article, main, .content, .docs-content", {
      timeout: 8000,
    });
  } catch (_) {
    // Some pages have no semantic container — proceed with full body extraction
  }

  const htmlContent = await page.content();
  const $ = cheerio.load(htmlContent);

  // Remove navigation, headers, footers, and boilerplate
  $(
    "nav, header, footer, .sidebar, .navigation, .breadcrumb, .toc, script, style, [aria-hidden=true]"
  ).remove();

  // Check for AI training opt-out signals
  const robotsMeta = $('meta[name="robots"]').attr("content") || "";
  if (
    robotsMeta.toLowerCase().includes("noai") ||
    robotsMeta.toLowerCase().includes("noindex")
  ) {
    return null; // Respect opt-out
  }

  // Extract title
  const title = $("h1").first().text().trim() || $("title").text().trim();

  // Extract main content — prioritize semantic HTML
  let mainContent =
    $("article").text().trim() ||
    $("main").text().trim() ||
    $(".content").text().trim() ||
    $("body").text().trim();

  // Clean up whitespace
  mainContent = mainContent
    .split("\n")
    .map((line) => line.trim())
    .filter((line) => line.length > 0)
    .join("\n")
    .replace(/\n{3,}/g, "\n\n");

  if (mainContent.length < 300) {
    return null; // Too short for AI training data
  }

  // Extract code blocks separately — valuable for code model training data
  const codeBlocks = [];
  $("pre code, .highlight code, code[class]").each((_, el) => {
    const code = $(el).text().trim();
    const langClass = $(el).attr("class") || "";
    const langMatch = langClass.match(/language-(\w+)/);
    const language = langMatch ? langMatch[1] : "unknown";

    if (code.length > 50) {
      codeBlocks.push({
        code,
        language,
        length: code.length,
      });
    }
  });

  return {
    url,
    title,
    content: mainContent,
    word_count: mainContent.split(/\s+/).length,
    code_blocks: codeBlocks,
    has_code: codeBlocks.length > 0,
    content_hash: crypto
      .createHash("sha256")
      .update(mainContent)
      .digest("hex"),
    crawled_at: new Date().toISOString(),
    extraction_method: "playwright",
  };
}

async function crawlDocumentationSite(
  seedUrl,
  outputPath,
  maxPages = 100,
  concurrency = 3
) {
  const browser = await chromium.launch({ headless: true });
  const results = [];
  const visited = new Set();
  const queue = [seedUrl];
  let processed = 0;

  console.log(`Starting AI training data crawl from: ${seedUrl}`);

  while (queue.length > 0 && processed < maxPages) {
    // Process in batches respecting concurrency limit
    const batch = queue.splice(0, concurrency);
    const context = await createStealthContext(browser);

    const batchResults = await Promise.allSettled(
      batch.map(async (url) => {
        if (visited.has(url)) return null;
        visited.add(url);

        const page = await context.newPage();
        try {
          const data = await extractDocumentationPage(page, url);
          processed++;

          if (data && data.word_count > 100) {
            // Collect internal links for continued crawling
            const links = await page.$$eval("a[href]", (anchors) =>
              anchors
                .map((a) => a.href)
                .filter(
                  (href) => href.startsWith("http") && !href.includes("#")
                )
            );

            // Only follow links within same origin for focused AI training data
            const origin = new URL(url).origin;
            links
              .filter((l) => l.startsWith(origin) && !visited.has(l))
              .slice(0, 10)
              .forEach((l) => queue.push(l));

            return data;
          }
          return null;
        } finally {
          await page.close();
        }
      })
    );

    await context.close();

    for (const result of batchResults) {
      if (result.status === "fulfilled" && result.value) {
        results.push(result.value);
        if (results.length % 10 === 0) {
          console.log(
            `Collected ${results.length} documents for AI training data`
          );
        }
      }
    }

    // Politeness delay between batches
    await new Promise((r) => setTimeout(r, 500));
  }

  await browser.close();

  // Write to JSONL for AI training data pipeline
  const output = results
    .map((r) => JSON.stringify(r))
    .join("\n");
  fs.writeFileSync(outputPath, output, "utf-8");
  console.log(
    `AI training data crawl complete: ${results.length} documents → ${outputPath}`
  );
  return results;
}

// Example: Crawl a documentation site for technical AI training data
crawlDocumentationSite(
  "https://docs.python.org/3/",
  "./python_docs_training_data.jsonl",
  50,
  3
).catch(console.error);

PII Detection and Privacy-Safe AI Training Data

One of the most serious risks in web scraping for AI training data is inadvertently including personally identifiable information in the corpus. Research from multiple AI safety teams has demonstrated that large language models trained on corpora containing PII can reproduce that information verbatim at inference time — a phenomenon called memorization. A model that memorizes and regurgitates real email addresses, phone numbers, or social security numbers creates significant legal exposure under GDPR, CCPA, and other data protection frameworks.

Production AI training data pipelines must implement PII detection and redaction as a mandatory pipeline stage.

# Prerequisites:
# pip install presidio-analyzer presidio-anonymizer spacy
# python -m spacy download en_core_web_lg
# pip install presidio-analyzer[transformers]  # For ML-based detection

from presidio_analyzer import AnalyzerEngine
from presidio_anonymizer import AnonymizerEngine
from presidio_anonymizer.entities import OperatorConfig
import re
import logging

logger = logging.getLogger(__name__)


class PIIRedactor:
    """
    PII detection and redaction for AI training data pipelines.
    Uses Microsoft Presidio for entity recognition.

    Critical for GDPR Article 5(1)(b) compliance: data collected for AI
    training data purposes cannot be further processed for identifying
    individuals unless re-identification is explicitly consented to.

    PII entity types detected:
    - PERSON, EMAIL_ADDRESS, PHONE_NUMBER, CREDIT_CARD
    - SSN (US Social Security Numbers)
    - IP_ADDRESS, MEDICAL_LICENSE, URL (when embedded in sensitive context)
    - DATE_TIME (when associated with a person)
    """

    # Entity types to redact from AI training data
    REDACT_ENTITIES = [
        "PERSON",
        "EMAIL_ADDRESS",
        "PHONE_NUMBER",
        "CREDIT_CARD",
        "US_SSN",
        "US_DRIVER_LICENSE",
        "IP_ADDRESS",
        "MEDICAL_LICENSE",
        "NRP",  # Nationality, Religion, Political affiliation
    ]

    def __init__(self):
        self.analyzer = AnalyzerEngine()
        self.anonymizer = AnonymizerEngine()

        # Operator config: replace PII with typed placeholder
        # [PERSON] is more informative for LLM training than [REDACTED]
        self.operators = {
            entity_type: OperatorConfig("replace", {"new_value": f"[{entity_type}]"})
            for entity_type in self.REDACT_ENTITIES
        }

    def redact(self, text: str, language: str = "en") -> dict:
        """
        Detects and redacts PII from document text for AI training data inclusion.
        Returns dict with redacted text and detection metadata.
        """
        if len(text) > 100_000:
            # Process in chunks for very long documents
            return self._redact_chunked(text, language)

        try:
            results = self.analyzer.analyze(
                text=text,
                entities=self.REDACT_ENTITIES,
                language=language,
                score_threshold=0.7,  # High threshold — avoid over-redaction
            )
        except Exception as e:
            logger.warning(f"Presidio analysis error: {e}")
            return {"redacted_text": text, "pii_found": False, "entities": []}

        if not results:
            return {"redacted_text": text, "pii_found": False, "entities": []}

        anonymized = self.anonymizer.anonymize(
            text=text,
            analyzer_results=results,
            operators=self.operators,
        )

        return {
            "redacted_text": anonymized.text,
            "pii_found": True,
            "entity_count": len(results),
            "entity_types": list({r.entity_type for r in results}),
        }

    def _redact_chunked(self, text: str, language: str, chunk_size: int = 50000) -> dict:
        """Process long documents in chunks for memory efficiency."""
        chunks = [text[i:i+chunk_size] for i in range(0, len(text), chunk_size)]
        redacted_chunks = []
        total_entities = []

        for chunk in chunks:
            result = self.redact(chunk, language)
            redacted_chunks.append(result["redacted_text"])
            total_entities.extend(result.get("entity_types", []))

        return {
            "redacted_text": "".join(redacted_chunks),
            "pii_found": bool(total_entities),
            "entity_types": list(set(total_entities)),
        }


class PrivacySafePipeline:
    """
    Combined pipeline for privacy-safe AI training data processing.
    Integrates PII redaction with document quality filtering.
    """

    def __init__(self, pii_threshold: float = 0.05):
        """
        pii_threshold: maximum ratio of redacted characters to total characters.
        Documents exceeding this threshold are dropped entirely — high PII density
        suggests the document is a directory, contact page, or similar non-training
        content that happens to contain useful sentences mixed with personal data.
        """
        self.redactor = PIIRedactor()
        self.pii_threshold = pii_threshold

    def process(self, document: dict) -> dict | None:
        """
        Processes a document through PII redaction for AI training data safety.
        Returns None if the document should be excluded from the training corpus.
        """
        content = document.get("content", "")
        language = document.get("language", "en")

        if not content:
            return None

        result = self.redactor.redact(content, language)
        redacted_text = result["redacted_text"]

        # Check PII density
        if result["pii_found"]:
            original_len = len(content)
            redacted_len = len(redacted_text)
            redaction_ratio = (original_len - redacted_len) / max(original_len, 1)

            if redaction_ratio > self.pii_threshold:
                logger.info(
                    f"Dropping high-PII document ({redaction_ratio:.1%} redacted): "
                    f"{document.get('url', 'unknown')}"
                )
                return None  # Too much PII — exclude from AI training data corpus

        document["content"] = redacted_text
        document["pii_redacted"] = result["pii_found"]
        document["pii_entity_types"] = result.get("entity_types", [])
        return document

Legal Landscape: Navigating AI Training Data Litigation and Regulation

The legal environment for web scraping as a data collection pipeline for training data has undergone seismic shifts since 2023. Any engineering team building AI training data infrastructure in 2026 must understand the current legal framework — not as a peripheral concern, but as a hard architectural constraint.

The US Litigation Wave. Between late 2023 and 2025, a cascade of lawsuits targeting AI companies’ use of web-scraped training data fundamentally changed the industry’s approach to data provenance. Authors, publishers, news organizations, and software developers filed class-action suits arguing that scraping their copyrighted content for commercial training data without licensing constituted infringement. Several cases settled under undisclosed terms. None produced a definitive Supreme Court ruling on whether training data scraping constitutes fair use, leaving the legal status of web scraping for AI training data in US copyright law genuinely uncertain for commercial applications.

The practical response from major AI labs was a shift toward licensing data directly from publishers, investing in opt-in data collection programs, expanding synthetic data generation to reduce dependence on scraped content, and dramatically improving their provenance tracking infrastructure to demonstrate good-faith compliance efforts.

The EU AI Act and Training Data Transparency. The EU AI Act, which came into full effect in August 2026, introduces mandatory transparency requirements for providers of general-purpose AI (GPAI) models above a compute threshold of 10²³ FLOPs. Article 53(d) requires GPAI model providers to publish “a sufficiently detailed summary of the content used for training of the general-purpose AI model, in accordance with a template provided by the AI Office.” This is not a theoretical risk — it is a current compliance requirement for any model deployed in the EU.

Practically, this means every document in your AI training data corpus needs provenance metadata: source URL, crawl date, robots.txt status at crawl time, and opt-out signal status. Without this metadata, you cannot produce the required training data summaries. This is why the JSONL schema in the pipeline code above includes crawled_at, url, domain, and content_hash fields — they are not just operationally useful, they are legally necessary.

Robots.txt and the noai Signal. The noai and noimageai meta tag directives, introduced by a coalition of web publishers in 2023, provide a web-native mechanism for sites to opt out of AI training data collection. Unlike robots.txt (which is legally ambiguous as a terms-of-service mechanism), noai meta tags are an explicit opt-out signal that any court will view as meaningful evidence of a site owner’s intent. Your data collection pipeline’s AIOptOutMiddleware and the noai check in the parser are legally significant, not just ethical best practice.

For full guidance on compliance architecture for AI training data collection in EU and US jurisdictions, DataFlirt’s coverage of web scraping and GDPR compliance and scraping compliance and legal considerations covers the legal framework in operational detail.

Infrastructure Patterns for AI Training Data at Scale

Building a data collection pipeline for data collection pipeline that can sustain petabyte-scale output requires infrastructure patterns that go beyond a single server running Scrapy. The following architectural patterns are the industry standard for high-volume data collection pipeline engineering teams.

Pattern 1: Kubernetes-Orchestrated Distributed Crawling. Multiple Scrapy workers running in Kubernetes pods share a Redis frontier (scrapy-redis pattern). Each pod consumes from the frontier independently, with no inter-worker coordination except through the shared queue. Horizontal scaling is achieved by adjusting the number of pod replicas. The Kubernetes HorizontalPodAutoscaler can automatically scale the crawling fleet based on frontier queue depth — more URLs in queue means more workers.

# kubernetes/ai-training-crawler-hpa.yaml
apiVersion: autoscaling/v2
kind: HorizontalPodAutoscaler
metadata:
  name: ai-training-crawler-hpa
spec:
  scaleTargetRef:
    apiVersion: apps/v1
    kind: Deployment
    name: ai-training-crawler
  minReplicas: 5
  maxReplicas: 200  # Scale to 200 workers during peak AI training data collection
  metrics:
  - type: External
    external:
      metric:
        name: redis_queue_depth
        selector:
          matchLabels:
            queue: "ai-training-frontier"
      target:
        type: AverageValue
        averageValue: "10000"  # Target 10K URLs per worker in queue
---
apiVersion: apps/v1
kind: Deployment
metadata:
  name: ai-training-crawler
spec:
  replicas: 10
  selector:
    matchLabels:
      app: ai-training-crawler
  template:
    metadata:
      labels:
        app: ai-training-crawler
    spec:
      containers:
      - name: scraper
        image: your-registry/ai-training-scraper:latest
        resources:
          requests:
            cpu: "2"
            memory: "4Gi"
          limits:
            cpu: "4"
            memory: "8Gi"
        env:
        - name: REDIS_URL
          valueFrom:
            secretKeyRef:
              name: redis-credentials
              key: url
        - name: OUTPUT_BUCKET
          value: "gs://your-ai-training-data-bucket"
        command:
        - scrapy
        - crawl
        - ai_training_corpus
        - -s
        - REDIS_URL=$(REDIS_URL)

Pattern 2: Apache Spark for Post-Crawl Processing. Once raw crawled content is stored in object storage, large-scale deduplication, quality scoring, and format conversion run most efficiently as Spark jobs. Spark’s distributed processing can deduplicate a 10TB dataset in hours using distributed MinHash computation, a task that would take weeks on a single machine.

# spark_dedup.py — PySpark MinHash deduplication for AI training data at scale
# Prerequisites:
# pip install pyspark datasketch
# spark-submit --master yarn --num-executors 50 spark_dedup.py

from pyspark.sql import SparkSession
from pyspark.sql.functions import col, udf, sha2
from pyspark.sql.types import StringType, ArrayType, FloatType
import hashlib
import json

spark = SparkSession.builder \
    .appName("AITrainingDataDeduplication") \
    .config("spark.executor.memory", "16g") \
    .config("spark.driver.memory", "8g") \
    .getOrCreate()

# Load raw crawled content shards from GCS/S3
# Adjust path for your storage backend
raw_data = spark.read.json("gs://your-bucket/raw_crawl_shards/*.jsonl")

print(f"Total documents before deduplication: {raw_data.count():,}")

# Exact deduplication by content hash (fast, O(n) with hash aggregation)
deduped_exact = raw_data \
    .withColumn("content_hash", sha2(col("text"), 256)) \
    .dropDuplicates(["content_hash"])

exact_count = deduped_exact.count()
print(f"After exact deduplication: {exact_count:,} documents")

# Quality filter pass
@udf(FloatType())
def score_document(text):
    if not text or len(text.split()) < 100:
        return 0.0
    words = text.split()
    unique_ratio = len(set(w.lower() for w in words)) / len(words)
    # Simple quality proxy — production pipelines use full HeuristicQualityScorer
    return float(min(1.0, unique_ratio * 1.5))

quality_filtered = deduped_exact \
    .withColumn("quality_score", score_document(col("text"))) \
    .filter(col("quality_score") >= 0.5)

final_count = quality_filtered.count()
print(f"After quality filtering: {final_count:,} documents")
print(f"Retention rate: {final_count/raw_data.count():.1%}")

# Write final training corpus in sharded Parquet format
quality_filtered \
    .repartition(1000) \
    .write \
    .mode("overwrite") \
    .parquet("gs://your-bucket/ai_training_corpus_final/")

print("AI training data pipeline complete. Corpus written to GCS.")

Pattern 3: Streaming Pipeline with Apache Kafka. For AI training data pipelines that need near-real-time data freshness — particularly for fine-tuning on recent events, financial data, or continuously updated documentation — a Kafka-based streaming architecture allows crawled content to flow through quality filtering and into the training data store within minutes of being published.

For a deeper treatment of the infrastructure patterns used by high-volume data teams, see DataFlirt’s coverage of top scraping infrastructure patterns for enterprise data teams and best queue and job management tools for distributed scraping.

Multimodal AI Training Data: Scraping Image-Text Pairs

Vision-language models require paired image-text AI training data where each image is associated with a meaningful textual description. Web scraping for multimodal web scraping for AI multimodal use is substantially more complex than text-only crawling because it requires extracting, validating, and aligning two modalities simultaneously.

The dominant sources for image-text AI-ready data on the web are alt text on img elements, figure captions in academic papers and technical documentation, surrounding paragraph context for inline images, and explicitly captioned image galleries. Each source has different quality characteristics: alt text is frequently absent or generic (“image.jpg”), figure captions are usually precise but domain-specific, surrounding context is semantically rich but loosely aligned to the image.

# multimodal_extractor.py
# Prerequisites:
# pip install trafilatura pillow httpx beautifulsoup4 lxml

import asyncio
import hashlib
import httpx
from pathlib import Path
from typing import Optional
from bs4 import BeautifulSoup
from urllib.parse import urljoin, urlparse
import json
import re


class MultimodalAITrainingDataExtractor:
    """
    Extracts image-text pairs for multimodal training data from web pages.

    Quality thresholds for inclusion in vision-language model AI training data:
    - Minimum alt text length: 20 characters (generic alt text excluded)
    - Minimum surrounding context: 100 characters
    - Image must be above minimum dimensions (128x128px minimum)
    - Caption must be semantically related to image content (validated by LLM stage)
    """

    MIN_ALT_TEXT_LEN = 20
    MIN_CONTEXT_LEN = 100
    MIN_IMAGE_SIZE = 128  # pixels

    def __init__(self, output_dir: str = "./multimodal_data"):
        self.output_dir = Path(output_dir)
        self.output_dir.mkdir(exist_ok=True)
        (self.output_dir / "images").mkdir(exist_ok=True)

    async def extract_image_text_pairs(self, html: str, page_url: str) -> list[dict]:
        """
        Extracts image-text pairs from a page for multimodal AI training data.
        Returns list of dicts with image_url, alt_text, caption, and context.
        """
        soup = BeautifulSoup(html, "lxml")
        pairs = []

        # Find all figure elements (highest quality image-text alignment)
        for figure in soup.find_all("figure"):
            img = figure.find("img")
            caption = figure.find("figcaption")

            if not img:
                continue

            alt_text = img.get("alt", "").strip()
            caption_text = caption.get_text(strip=True) if caption else ""
            img_src = img.get("src", "") or img.get("data-src", "")

            if not img_src:
                continue

            # Use caption as primary text source, fall back to alt text
            primary_text = caption_text if len(caption_text) >= self.MIN_ALT_TEXT_LEN else alt_text

            if len(primary_text) < self.MIN_ALT_TEXT_LEN:
                continue

            # Extract surrounding paragraph context
            context = self._get_surrounding_context(figure, soup)

            pairs.append({
                "image_url": urljoin(page_url, img_src),
                "alt_text": alt_text,
                "caption": caption_text,
                "context": context,
                "primary_text": primary_text,
                "page_url": page_url,
                "source_type": "figure_caption",
                "pair_id": hashlib.sha256(
                    f"{page_url}{img_src}".encode()
                ).hexdigest()[:16],
            })

        # Also extract standalone images with meaningful alt text
        for img in soup.find_all("img"):
            # Skip images already captured in figure elements
            if img.find_parent("figure"):
                continue

            alt_text = img.get("alt", "").strip()
            if len(alt_text) < self.MIN_ALT_TEXT_LEN:
                continue

            # Skip decorative, icon, or avatar images based on common patterns
            img_src = img.get("src", "") or img.get("data-src", "")
            if not img_src:
                continue
            if any(skip in img_src.lower() for skip in ["icon", "logo", "avatar", "badge", "button"]):
                continue

            context = self._get_surrounding_context(img, soup)
            if len(context) < self.MIN_CONTEXT_LEN:
                continue

            pairs.append({
                "image_url": urljoin(page_url, img_src),
                "alt_text": alt_text,
                "caption": "",
                "context": context,
                "primary_text": alt_text,
                "page_url": page_url,
                "source_type": "alt_text",
                "pair_id": hashlib.sha256(
                    f"{page_url}{img_src}".encode()
                ).hexdigest()[:16],
            })

        return pairs

    def _get_surrounding_context(self, element, soup, max_chars: int = 500) -> str:
        """Extract the paragraph text immediately before and after an element."""
        context_parts = []

        # Get previous sibling text
        prev = element.find_previous("p")
        if prev:
            context_parts.append(prev.get_text(strip=True)[-200:])

        # Get next sibling text
        next_el = element.find_next("p")
        if next_el:
            context_parts.append(next_el.get_text(strip=True)[:200])

        return " ".join(context_parts)[:max_chars]

Synthetic Data: Augmenting Scraped AI Training Data

Web scraping is not the only source of AI training data. Synthetic data generation — using models to generate new training examples — is an increasingly important complement to scraped data, particularly for instruction tuning, RLHF preference data, and low-resource language coverage.

The standard workflow is: scrape high-quality seed documents from the web, use those documents as context for an LLM to generate synthetic instruction-following examples, then filter the synthetic data with a quality model. This pipeline is sometimes called “knowledge distillation” in the AI training data community, though it more precisely resembles semi-supervised data augmentation.

The limitations of synthetic AI training data are well-documented: models trained primarily on synthetic data can develop distributional artifacts (“model collapse”), where the generated data converges toward the style and vocabulary of the generator model rather than reflecting the full diversity of human-generated text. The consensus in 2026 is that synthetic data is most effective as a targeted supplement to scraped data — filling specific distributional gaps — rather than as a primary source.

For practical guidance on building AI training data at the platform level rather than with custom pipelines, DataFlirt’s best scraping platforms for building AI training datasets covers the managed service landscape for teams that need to move faster than a fully custom pipeline allows.

The Anti-Bot Problem in AI Training Data Collection

Collecting AI training data from the web means your crawlers will inevitably encounter bot detection systems. High-value AI training data sources — academic publishers, legal databases, financial data providers, quality journalism — are precisely the sources most likely to have aggressive anti-bot protection, because they are also the sources most targeted by scrapers. Understanding how to navigate this challenge ethically and technically is essential for production AI training data pipelines.

The most common bot detection vectors encountered in AI training data crawling are IP reputation scoring (datacenter IP ranges are immediately suspect), TLS fingerprint analysis (Python’s default HTTP clients produce non-browser TLS signatures), JavaScript challenge pages (Cloudflare’s JS challenge, reCAPTCHA Enterprise), browser fingerprinting (headless Chromium without stealth patches exposes navigator.webdriver), and behavioral analysis (perfectly uniform request intervals are a strong bot signal).

For a technical deep-dive on bypassing each of these detection vectors, DataFlirt’s guide on how to bypass Google CAPTCHA for web scraping and best approaches to scraping dynamic JavaScript sites without getting blocked cover the full evasion stack. For the proxy infrastructure layer, best IP rotation strategies for high-volume scraping projects and best proxy management tools to rotate and manage proxies at scale are essential companion resources.

One important consideration specific to AI training data crawling: many high-value content sources have explicit policies about AI training data scraping that go beyond their general terms of service. Some sites now provide machine-readable AI usage policies at /ai-policy.json (an emerging standard proposed by the Data Provenance Initiative). Checking and honoring these policies is both ethically correct and, increasingly, legally required. Your data collection pipeline should include an AI policy checker that fetches and logs these signals as part of the URL frontier pre-processing stage.

For detailed guidance on the full open-source tooling landscape, DataFlirt’s best free web scraping tools guide and top 10 open-source web scraping tools worth using in 2026 cover the complete framework comparison with production benchmarks.

Storage Architecture for AI Training Data at Petabyte Scale

The output of a serious data collection pipeline for large language model training is measured in terabytes to petabytes. Naive storage choices — flat directories, uncompressed files, single S3 buckets with no partitioning — become operational problems at this scale. The storage architecture for AI training data has specific requirements that differ from standard data warehouse design.

The dominant format in 2026 for AI training data at scale is sharded, compressed JSONL stored in object storage (S3 or GCS), with Parquet as the secondary format for columnar analytics (quality score distributions, language breakdowns, domain coverage statistics). Each JSONL shard should be 512MB–2GB uncompressed before compression — large enough to be efficient to read in parallel, small enough to be downloaded and processed by individual training nodes without memory pressure.

Metadata indexing — tracking which shards contain which domains, languages, quality score ranges, and crawl dates — is essential for dataset composition. Without a metadata index, reproducing a training run’s exact data mixture is effectively impossible. Every major research lab that has published training methodology in 2025–2026 now includes a data provenance statement precisely because this metadata was retained from the beginning of their AI training data pipeline.

For guidance on storage platform selection for AI training data pipelines, DataFlirt’s best cloud storage solutions for managing large scraped datasets and best databases for storing scraped data at scale cover the technical tradeoffs between object storage, columnar databases, and vector databases for downstream retrieval-augmented use cases.

What the Industry Actually Uses: How Real AI Companies Approach Training Data Collection

Understanding the public-domain documentation from major AI labs provides ground truth on how production AI training data pipelines are architected. The following observations are drawn from published papers and technical reports, not from proprietary information.

Common Crawl processing pipelines are the universal starting point. The Falcon team’s RefinedWeb paper is among the most detailed public documentation of a web scraping for AI training data pipeline: they processed 5.17 trillion tokens from 43 monthly Common Crawl dumps, applied URL deduplication, language identification, quality filtering (removing documents shorter than 100 words, those with high symbol ratios, those from adult domains), exact deduplication, near-deduplication using MinHash with 0.87 Jaccard threshold, and machine-generated text removal. The final RefinedWeb corpus was 968 billion tokens — roughly 19% of the raw crawl volume that passed all filters.

Proprietary high-quality source targeting supplements Common Crawl for all major models. GitHub code (licensed under open-source licenses) is a universal inclusion for code capability. Wikipedia in all available languages provides dense, factually verified text. Academic preprint archives provide scientific reasoning chains. Legal databases, when accessible, provide formal argumentation structure. The data collection pipelines for these sources are more targeted and precise than general-purpose crawls — they use structured APIs where available and web scraping where not.