Why Your Scrapers Are Flying Blind (And How to Fix That)

from pydantic import BaseModel, ValidationError, Field
from typing import Optional

class ProductData(BaseModel):
    name: str = Field(..., min_length=1, description="Product name must not be empty")
    price: float = Field(..., gt=0, description="Price must be a positive number")
    currency: str = Field(..., max_length=3, min_length=3, description="Currency must be a 3-letter code")
    availability: Optional[str] = "In Stock" # Optional field

def process_scraped_item(item_dict: dict):
    try:
        # Attempt to validate the scraped data
        validated_item = ProductData(**item_dict)
        print(f"✅ Validated Product: {validated_item.name}, Price: {validated_item.price}")
        # Here you would typically save validated_item to your database
        return validated_item
    except ValidationError as e:
        print(f"❌ Data Validation Error for item: {item_dict.get('name', 'N/A')}")
        print(e.json()) # Log the detailed validation error
        # Trigger alert here (e.g., send to Slack, Sentry)
        return None

# --- Simulating Scraped Data ---
good_data = {
    "name": "Super Widget",
    "price": 29.99,
    "currency": "USD"
}

bad_price_data = {
    "name": "Broken Gadget",
    "price": "twenty dollars", # Incorrect type
    "currency": "USD"
}

missing_name_data = {
    "name": "", # Empty string, fails min_length
    "price": 10.50,
    "currency": "EUR"
}

# Process the data
process_scraped_item(good_data)
process_scraped_item(bad_price_data)
process_scraped_item(missing_name_data)

import pandas as pd
from typing import List, Dict

def monitor_data_density(scraped_records: List[Dict], threshold: float = 0.10):
    """
    Calculates null percentage for each field and checks against a threshold.
    Args:
        scraped_records: A list of dictionaries, where each dict is a scraped item.
        threshold: The maximum acceptable percentage of nulls for any given field (e.g., 0.10 for 10%).
    Returns:
        True if all fields are below the null threshold, False otherwise.
    """
    if not scraped_records:
        print("No records to monitor.")
        return True

    df = pd.DataFrame(scraped_records)
    null_counts = df.isnull().sum()
    total_records = len(df)
    null_percentages = (null_counts / total_records) * 100

    print(f"\n--- Data Density Report ({total_records} records) ---")
    all_healthy = True
    for field, percentage in null_percentages.items():
        if percentage > (threshold * 100):
            print(f"❌ Field '{field}': {percentage:.2f}% null (Exceeds {threshold*100:.0f}% threshold!)")
            all_healthy = False
        elif percentage > 0:
            print(f"⚠️ Field '{field}': {percentage:.2f}% null")
        else:
            print(f"✅ Field '{field}': {percentage:.2f}% null")
    
    if not all_healthy:
        print(f"🚨 ALERT: Some fields exceed the null threshold of {threshold*100:.0f}%!")
        # Trigger alert here
    else:
        print("All fields are within acceptable null limits.")
    
    return all_healthy

# --- Simulating Scraped Data ---
daily_crawl_data = [
    {"name": "Item A", "price": 10.00, "description": "Good item"},
    {"name": "Item B", "price": 12.50, "description": "Another item"},
    {"name": "Item C", "price": None, "description": "Price missing!"},
    {"name": "Item D", "price": 15.00, "description": None}, # Description missing
    {"name": "Item E", "price": 20.00, "description": "New item"},
    {"name": "Item F", "price": None, "description": "Also missing price!"},
    {"name": "Item G", "price": 25.00, "description": "Final item"},
]

# Simulate a good day (e.g., 100 records, 2% null)
healthy_data = daily_crawl_data * 10 
monitor_data_density(healthy_data, threshold=0.05) # 5% threshold

# Simulate a bad day (e.g., 100 records, 30% null on price)
bad_data_day = [
    {"name": f"Product {i}", "price": (None if i % 3 == 0 else float(i)), "description": f"Desc {i}"}
    for i in range(100)
]
monitor_data_density(bad_data_day, threshold=0.10) # 10% threshold

from playwright.sync_api import sync_playwright

def scrape_with_screenshot(url: str, selector: str, screenshot_path: str = "failure_screenshot.png"):
    """
    Attempts to scrape a page and takes a screenshot if a specific selector is not found.
    """
    with sync_playwright() as p:
        browser = p.chromium.launch()
        page = browser.new_page()
        
        try:
            page.goto(url, wait_until='domcontentloaded')
            # Check if the expected data element exists
            if page.locator(selector).count() > 0:
                print(f"✅ Selector '{selector}' found. Data likely present.")
                # Extract data here...
                return page.locator(selector).inner_text()
            else:
                print(f"❌ Selector '{selector}' NOT found on the page. Taking screenshot.")
                page.screenshot(path=screenshot_path)
                print(f"Screenshot saved to {screenshot_path}")
                return None
        except Exception as e:
            print(f"An error occurred: {e}")
            page.screenshot(path=f"error_{screenshot_path}")
            print(f"Error screenshot saved to error_{screenshot_path}")
            return None
        finally:
            browser.close()

# --- Usage ---
# Example 1: Scrape a real site, expecting a title
# scrape_with_screenshot("https://www.scrapingbee.com/", "h1")

# Example 2: Simulate a failure (selector not found)
# You might need to adjust the URL or selector to reliably trigger a failure
# For demonstration, let's use a simple page and a non-existent selector
# Or better yet, a page known to sometimes show CAPTCHA or no data
# For now, let's use a valid page but an unlikely selector to trigger the screenshot logic
scrape_with_screenshot(
    "https://www.example.com", 
    "#non-existent-data-element-id", 
    "example_failure.png"
)

# You could extend this to check for specific anti-bot messages and screenshot those