Scraping Discipline & Extraction Framework Specification (scrapespec)

scrapespec (typically imported as import scrapespec as ss) is a lightweight, zero-dependency data discipline, normalizer stacking, multi-extractor candidate execution, and diagnostic provenance engine designed specifically for modern web scraping pipelines.

Unlike traditional contract-validation libraries (such as Pydantic, Attrs, or Marshmallow) whose fail-fast design drops entire scraped records when encountering minor anomalies or unexpected DOM variations, scrapespec adopts a discipline-first, multi-extractor, extraction-agnostic architecture to maximize usable data yield while preserving strict auditability and data integrity.

🌟 Key Features

• Namespace Ergonomics: Designed to be imported as import scrapespec as ss, exposing ss.Schema, ss.Field, ss.OutlierPolicy, semantic types, and normalizers directly under one cohesive namespace.
• Separation of Extraction vs. Discipline: Decouples extraction mechanisms (BeautifulSoup, Playwright locators, Scrapy selectors, regex, direct dict lookups, or heuristic inference engines) from data normalization, type coercion, and boundary disciplines.
• Multi-Extractor Execution (extractors=[...]): Enables multi-perspective extraction hooks across layout variations and engines to produce candidate profiles without premature single-winner discard.
• Cascading Normalization Stacks: Composable schema pre-processors stacked deterministically with field-specific transformers (e.g. whitespace collapsing, Arabic/Eastern digit translation, currency/percentage stripping).
• Modern Schema Configuration: Full support for ss.ConfigDict, standard Python type annotations (int, float, str, Optional, Union), and parameterized semantic type factories (ss.Number, ss.PositiveNumber, ss.Percentage, ss.Duration, ss.Text).
• Declarative Outlier & Fallback Policies: Granular handling of boundary violations (WARN, CLAMP, DEFAULT, QUARANTINE, ERROR) and missing fields (USE_DEFAULT, NULLIFY, ERROR) to eliminate unhandled crawl terminations.
• Audit & Diagnostic Provenance: Automatic per-field metadata tracking (raw, source, confidence, outlier, issues, formatted) for total audit transparency.
• Zero Dependencies: Pure standard library implementation with zero external runtime dependencies.

📦 Installation

# Using uv (recommended)
uv add scrapespec

# Using pip
pip install scrapespec

🚀 Quickstart: Idiomatic Namespace Usage

The recommended way to use scrapespec is via the namespace import:

import scrapespec as ss
from bs4 import BeautifulSoup

# Define a robust scraping schema
class FreelancerProfile(ss.Schema):
    model_config: ss.ConfigDict = {
        "normalizers": [
            ss.strip_text,
            ss.collapse_whitespace,
        ],
        "default_outlier_policy": ss.OutlierPolicy.WARN,
        "arabic_digit_translation": True,
    }

    # Multi-extractor pipeline with semantic positive integer bounds
    completed_projects: ss.PositiveNumber = ss.Field(
        max=500,
        on_outlier=ss.OutlierPolicy.CLAMP,
        normalizers=[
            lambda val: val.replace("مشروع", "").strip() if isinstance(val, str) else val
        ],
        extractors=[
            lambda soup: soup.select_one(".completed-projects"),
            lambda soup: soup.find("div", {"data-metric": "completed"}),
            lambda raw_dict: raw_dict.get("projects_count"),
        ],
        default=0,
    )

    # Percentage discipline with automatic symbol stripping
    completion_rate: ss.Percentage = ss.Field(
        on_outlier=ss.OutlierPolicy.WARN,
        normalizers=[ss.strip_percentage],
        extractors=[
            lambda soup: soup.select_one(".completion-rate"),
            lambda raw_dict: raw_dict.get("completion_rate"),
        ],
        default=0.0,
    )

    # Duration with boundary clamping (minutes)
    response_time_minutes: ss.Duration = ss.Field(
        unit="minutes",
        min=1.0,
        max=10080.0,  # Max 7 days
        on_outlier=ss.OutlierPolicy.CLAMP,
        fallback=ss.FallbackPolicy.USE_DEFAULT,
        extractors=[
            lambda soup: soup.select_one(".response-time"),
        ],
        default=1440.0,
    )

    # Record-level invariant & cross-field coherence rule
    @ss.coherence
    def check_valid_completion(record: dict) -> ss.CoherenceReport:
        report = ss.CoherenceReport()
        if record.get("completed_projects", 0) > 0 and record.get("completion_rate", 0) == 0:
            report.add_issue(
                field="completion_rate",
                rule="non_zero_projects_require_rate",
                severity=ss.MetricQuality.SUSPECT,
                message="Profile has completed projects but 0% completion rate",
            )
        return report

Parsing Records & Auditing Provenance

html_doc = """
<div class="profile">
    <div class="completed-projects">  ٢٥ مشروع  </div>
    <div class="completion-rate">98%</div>
    <div class="response-time">30 دقيقة</div>
</div>
"""
soup = BeautifulSoup(html_doc, "html.parser")

# Parse using the schema
results = FreelancerProfile.parse(soup)

# Inspect the primary parsed record
primary = results[0]
print(primary.data)
# {
#     'completed_projects': 25,
#     'completion_rate': 98.0,
#     'response_time_minutes': 30.0
# }

# Audit provenance and field-level metadata
for field_name, meta in primary.meta.items():
    print(f"[{field_name}] raw: '{meta.raw}' -> value: {meta.value} (outlier={meta.outlier}, issues={meta.issues})")

🏛️ Architectural Pipeline

Raw Source Context (DOM Tree, Page Locator, Dict, or Inference Engine)
   │
   ▼
[Stage 1: Multi-Extractor Execution] (`extractors=[ext1, ext2, ext3]`)
   │  (Executes all extractors across fields to produce extraction candidate profiles)
   │
   ▼
[Stage 2: Schema Normalizer Stack] (`model_config["normalizers"]`)
   │  (e.g., whitespace collapsing, unicode cleanup, entity unescaping)
   │
   ▼
[Stage 3: Local Field Normalizers] (`ss.Field(normalizers=[...])`)
   │  (e.g., Arabic-to-ASCII digit conversion, currency regex, unit stripping)
   │
   ▼
[Stage 4: Semantic Type Coercion & Discipline] (`projects: ss.PositiveNumber = ss.Field(max=60)`)
   │  (Safe type casting, min/max constraint evaluation)
   │
   ▼
[Stage 5: Outlier Policy Evaluation] (`on_outlier`)
   │  (WARN, CLAMP, DEFAULT, ERROR, QUARANTINE)
   │
   ▼
[Stage 6: Record-Level Invariants & Coherence Rules] (`@ss.coherence`)
   │  (Cross-field validation, record quality classification per profile outcome)
   │
   ▼
Array of Extracted Profiles + FieldMeta Provenance Reports (`profiles: list[ss.ParsedResult]`)

⚙️ Core Components Reference

1. ss.Schema

Base class for declaring extraction specifications. Provides .parse(source) which safely runs all configured extractors, normalizers, type disciplines, outlier policies, and coherence validations.

2. ss.Field(...)

The primary field declaration construct:

• extractors: Sequence of callables taking the source object and returning raw candidates or values.
• normalizers: Sequence of functions applied sequentially to field values.
• min / max: Boundary constraints for numeric and duration types.
• on_outlier: ss.OutlierPolicy (WARN, CLAMP, DEFAULT, QUARANTINE, ERROR).
• fallback: ss.FallbackPolicy (USE_DEFAULT, NULLIFY, ERROR).
• default: Default value when extraction produces null or empty results.
• confidence: Confidence score (0.0 to 1.0) assigned to extraction candidates.

3. Semantic Types

• ss.Number(min=..., max=...): General numeric factory.
• ss.PositiveNumber: Non-negative integer (min=0).
• ss.PositiveFloat: Non-negative float (min=0.0).
• ss.Percentage: Percentage float constrained between 0.0 and 100.0.
• ss.Duration(unit="seconds"|"minutes"|"hours"|"days"): Safe duration parser with unit normalization.
• ss.Text(min_len=..., max_len=...): Text discipline with length constraints.

4. Policies (ss.OutlierPolicy & ss.FallbackPolicy)

• ss.OutlierPolicy.WARN: Retain the outlier value but flag meta.outlier = True and log issue.
• ss.OutlierPolicy.CLAMP: Clamp numeric value to boundary minimum or maximum.
• ss.OutlierPolicy.DEFAULT: Replace value with field default if out of bounds.
• ss.OutlierPolicy.QUARANTINE: Nullify field value and mark record as quarantined.
• ss.OutlierPolicy.ERROR: Raise ValueError on outlier (for zero-tolerance fields).
• ss.FallbackPolicy.USE_DEFAULT: Use declared default when field is missing.
• ss.FallbackPolicy.NULLIFY: Yield None when extraction yields empty.
• ss.FallbackPolicy.ERROR: Raise error when field cannot be extracted.

5. Built-in Normalizers

• ss.strip_text(val): Strips leading/trailing whitespace.
• ss.collapse_whitespace(val): Collapses repeated spaces, tabs, and newlines.
• ss.translate_arabic_digits(val): Converts Eastern Arabic numerals (٠١٢٣٤٥٦٧٨٩) to standard ASCII numerals (0123456789).
• ss.strip_percentage(val): Strips % and Arabic percent signs, returning numeric strings.

🛠️ Low-Level Interoperability: parse_item

For legacy pipelines or runtime dynamic schemas, scrapespec provides ss.parse_item:

import scrapespec as ss

raw_data = {"price": "  1,250.50 $  ", "rating": "6.5"}

clean_price = ss.strip_text(raw_data["price"]).replace(",", "").replace("$", "").strip()
price_val = ss.Number(min=0.0, max=10000.0).coerce(clean_price)
print(price_val)  # 1250.5

📄 License

MIT License. Designed and engineered for high-throughput web scraping operations.