Build an AI Web Scraper with Python and Claude

pip install httpx anthropic beautifulsoup4

export EZAI_API_KEY="sk-your-key-here"

import httpx, json, os
from bs4 import BeautifulSoup
import anthropic

def clean_html(raw_html: str) -> str:
    """Strip scripts, styles, nav, and footer — keep the meat."""
    soup = BeautifulSoup(raw_html, "html.parser")
    for tag in soup.find_all(["script", "style", "nav", "footer", "header"]):
        tag.decompose()
    text = soup.get_text(separator="\n", strip=True)
    # Collapse blank lines, cap at 12k chars to stay under token limits
    lines = [l for l in text.splitlines() if l.strip()]
    return "\n".join(lines)[:12000]

def scrape_with_ai(url: str, schema: dict) -> dict:
    # 1. Fetch the page
    resp = httpx.get(url, headers={"User-Agent": "Mozilla/5.0"}, follow_redirects=True)
    resp.raise_for_status()
    cleaned = clean_html(resp.text)

    # 2. Send to Claude via EzAI
    client = anthropic.Anthropic(
        api_key=os.environ["EZAI_API_KEY"],
        base_url="https://ezaiapi.com",
    )

    message = client.messages.create(
        model="claude-sonnet-4-5",
        max_tokens=2048,
        messages=[{
            "role": "user",
            "content": f"""Extract data from this webpage content.

Return ONLY valid JSON matching this schema:
{json.dumps(schema, indent=2)}

Webpage content:
---
{cleaned}
---

Respond with the JSON object only. No markdown, no explanation."""
        }],
    )

    return json.loads(message.content[0].text)

product_schema = {
    "products": [{
        "name": "string",
        "price": "number (USD)",
        "rating": "number (0-5)",
        "review_count": "integer",
        "in_stock": "boolean",
        "description": "string (first 100 chars)"
    }]
}

data = scrape_with_ai(
    "https://example.com/electronics/keyboards",
    product_schema
)

for p in data["products"]:
    print(f"{p['name']}: ${p['price']} ({p['rating']}★, {p['review_count']} reviews)")

Keychron Q1 Pro: $169.99 (4.7★, 2341 reviews)
HHKB Professional Hybrid: $289.00 (4.8★, 876 reviews)
Logitech MX Keys S: $109.99 (4.5★, 5210 reviews)

import anthropic, json, os

def chunk_text(text: str, chunk_size=10000, overlap=500) -> list:
    chunks = []
    start = 0
    while start < len(text):
        end = start + chunk_size
        chunks.append(text[start:end])
        start = end - overlap
    return chunks

def scrape_large_page(url: str, schema: dict) -> dict:
    resp = httpx.get(url, headers={"User-Agent": "Mozilla/5.0"})
    cleaned = clean_html(resp.text)
    chunks = chunk_text(cleaned)

    client = anthropic.Anthropic(
        api_key=os.environ["EZAI_API_KEY"],
        base_url="https://ezaiapi.com",
    )

    all_items = []
    for i, chunk in enumerate(chunks):
        msg = client.messages.create(
            model="claude-sonnet-4-5",
            max_tokens=2048,
            messages=[{
                "role": "user",
                "content": f"""Extract items from chunk {i+1}/{len(chunks)}.
Return JSON matching: {json.dumps(schema)}
Content:
---
{chunk}
---
JSON only."""
            }],
        )
        result = json.loads(msg.content[0].text)
        # Merge arrays from each chunk
        for key, val in result.items():
            if isinstance(val, list):
                all_items.extend(val)

    # Deduplicate by name
    seen = set()
    unique = []
    for item in all_items:
        key = item.get("name", str(item))
        if key not in seen:
            seen.add(key)
            unique.append(item)

    return {"items": unique}

import time

def scrape_with_retry(url: str, schema: dict, max_retries=3) -> dict:
    for attempt in range(max_retries):
        try:
            result = scrape_with_ai(url, schema)
            # Validate: ensure all schema keys exist
            for key in schema:
                if key not in result:
                    raise ValueError(f"Missing key: {key}")
            return result

        except json.JSONDecodeError:
            print(f"Attempt {attempt+1}: Invalid JSON, retrying...")
        except anthropic.RateLimitError:
            wait = 2 ** attempt
            print(f"Rate limited, waiting {wait}s...")
            time.sleep(wait)
        except httpx.HTTPStatusError as e:
            print(f"HTTP {e.response.status_code} for {url}")
            if e.response.status_code >= 500:
                time.sleep(1)
                continue
            raise

    raise RuntimeError(f"Failed after {max_retries} attempts: {url}")

#!/usr/bin/env python3
"""AI-powered job listing scraper using Claude via EzAI."""

import httpx, json, os
from bs4 import BeautifulSoup
import anthropic

client = anthropic.Anthropic(
    api_key=os.environ["EZAI_API_KEY"],
    base_url="https://ezaiapi.com",
)

JOB_SCHEMA = {
    "jobs": [{
        "title": "string",
        "department": "string",
        "location": "string",
        "type": "full-time | part-time | contract",
        "salary_range": "string or null",
        "apply_url": "string or null"
    }]
}

def scrape_jobs(url: str) -> list:
    resp = httpx.get(url, headers={"User-Agent": "Mozilla/5.0"})
    soup = BeautifulSoup(resp.text, "html.parser")
    for t in soup.find_all(["script", "style", "nav", "footer"]):
        t.decompose()
    content = soup.get_text("\n", strip=True)[:12000]

    msg = client.messages.create(
        model="claude-sonnet-4-5",
        max_tokens=4096,
        messages=[{
            "role": "user",
            "content": f"""Extract all job listings.
Return JSON: {json.dumps(JOB_SCHEMA)}
Content:
---
{content}
---"""
        }],
    )
    return json.loads(msg.content[0].text)["jobs"]

if __name__ == "__main__":
    jobs = scrape_jobs("https://example.com/careers")
    for j in jobs:
        print(f"📌 {j['title']} — {j['location']} ({j['type']})")
        if j.get("salary_range"):
            print(f"   💰 {j['salary_range']}")