Extracting Structured Data from Any Web Page with Python and Claude

pip install unweb anthropic pydantic

import os
import json
import asyncio
from typing import Optional
from pydantic import BaseModel, ValidationError

import unweb
import anthropic

unweb_client = unweb.Client(api_key=os.environ["UNWEB_API_KEY"])
claude = anthropic.Anthropic(api_key=os.environ["ANTHROPIC_API_KEY"])

def fetch_page(url: str, quality_threshold: int = 35) -> Optional[str]:
    """
    Fetch a URL and return clean Markdown.
    Returns None if the page is below quality threshold (login wall, error, etc).
    """
    result = unweb_client.convert.url(url)

    if result.quality_score < quality_threshold:
        print(f"Skipped {url} — quality score {result.quality_score} below threshold")
        return None

    # Trim to a reasonable token budget; most content is in the first ~6000 chars
    return result.markdown[:6000]

class ProductData(BaseModel):
    name: str
    price: Optional[float] = None           # None if not found
    currency: Optional[str] = None          # "USD", "EUR", etc.
    rating: Optional[float] = None          # 0.0–5.0
    review_count: Optional[int] = None
    availability: Optional[str] = None      # "In stock", "Ships in 3 days", etc.
    description: Optional[str] = None       # First 300 chars of product description
    brand: Optional[str] = None


class ArticleData(BaseModel):
    title: str
    author: Optional[str] = None
    published_date: Optional[str] = None    # ISO 8601 if possible, verbatim otherwise
    summary: str                            # 2–3 sentence summary
    word_count_estimate: Optional[int] = None
    primary_topic: Optional[str] = None     # Single topic label

def extract_structured_data(markdown: str, schema: type[BaseModel]) -> dict:
    """
    Extract structured data from Markdown using Claude.
    Returns a dict matching the schema, or raises ValidationError.
    """
    # Build schema description from Pydantic model
    fields = []
    for field_name, field_info in schema.model_fields.items():
        annotation = field_info.annotation
        description = field_info.description or ""
        fields.append(f"- {field_name} ({annotation}): {description}")
    schema_desc = "\n".join(fields)

    system_prompt = (
        "You are a data extraction assistant. Extract structured data from web page "
        "content provided as Markdown. Return only valid JSON matching the requested schema. "
        "Use null for fields you cannot find. Do not hallucinate data."
    )

    user_prompt = f"""Extract data from this web page content:

---
{markdown}
---

Return a JSON object with these fields:
{schema_desc}

Respond with only the JSON object. No explanation, no markdown fences."""

    message = claude.messages.create(
        model="claude-haiku-4-5-20251001",   # fast and cheap for extraction
        max_tokens=1024,
        system=system_prompt,
        messages=[{"role": "user", "content": user_prompt}]
    )

    raw_json = message.content[0].text.strip()

    # Strip markdown fences if the model added them anyway
    if raw_json.startswith("```"):
        raw_json = raw_json.split("```")[1]
        if raw_json.startswith("json"):
            raw_json = raw_json[4:]

    data = json.loads(raw_json)
    validated = schema(**data)  # Raises ValidationError if schema doesn't match
    return validated.model_dump()

def extract_from_url(url: str, schema: type[BaseModel]) -> Optional[dict]:
    """Fetch a URL and extract structured data in one call."""
    markdown = fetch_page(url)
    if markdown is None:
        return None

    try:
        return extract_structured_data(markdown, schema)
    except (json.JSONDecodeError, ValidationError) as e:
        print(f"Extraction failed for {url}: {e}")
        return None


# Single URL
product = extract_from_url("https://example.com/products/widget", ProductData)
if product:
    print(f"{product['name']}: ${product['price']} ({product['review_count']} reviews)")

# Article
article = extract_from_url("https://example.com/blog/post-123", ArticleData)
if article:
    print(f"{article['title']} — {article['published_date']}")
    print(article['summary'])

async def fetch_page_async(url: str, quality_threshold: int = 35) -> tuple[str, Optional[str]]:
    """Async version of fetch_page — returns (url, markdown_or_None)."""
    result = await unweb_client.convert.url_async(url)
    if result.quality_score < quality_threshold:
        return url, None
    return url, result.markdown[:6000]


async def batch_extract(urls: list[str], schema: type[BaseModel], concurrency: int = 5) -> list[dict]:
    """
    Fetch all URLs concurrently, then extract structured data from each.
    Skips failed fetches and extraction errors.
    """
    sem = asyncio.Semaphore(concurrency)

    async def fetch_with_sem(url):
        async with sem:
            return await fetch_page_async(url)

    # Fetch all pages concurrently
    fetch_tasks = [fetch_with_sem(url) for url in urls]
    fetch_results = await asyncio.gather(*fetch_tasks)

    results = []
    for url, markdown in fetch_results:
        if markdown is None:
            continue
        try:
            data = extract_structured_data(markdown, schema)
            data["_source_url"] = url
            results.append(data)
        except Exception as e:
            print(f"Failed {url}: {e}")

    return results


# Usage
product_urls = [
    "https://shop.example.com/products/item-1",
    "https://shop.example.com/products/item-2",
    "https://shop.example.com/products/item-3",
]

products = asyncio.run(batch_extract(product_urls, ProductData))
print(f"Extracted {len(products)} products from {len(product_urls)} URLs")

def extract_with_fallback(url: str, schema: type[BaseModel]) -> dict:
    """
    Extract with explicit error categorization.
    Returns a result dict with 'status' field: 'ok', 'low_quality', 'parse_error', 'validation_error'.
    """
    markdown = fetch_page(url)
    if markdown is None:
        return {"status": "low_quality", "url": url}

    try:
        data = extract_structured_data(markdown, schema)
        return {"status": "ok", "url": url, **data}
    except json.JSONDecodeError as e:
        return {"status": "parse_error", "url": url, "error": str(e)}
    except ValidationError as e:
        return {"status": "validation_error", "url": url, "error": str(e)}