Web Scraper & Data Pipeline

1. Project Overview & Architecture

Welcome to Major Project 4: Web Scraper & Data Pipeline! In this major project, you will build an automated web scraper and data processing pipeline. You will harvest structured information from web pages, transform raw HTML text into clean Python datasets, apply rate-limiting, handle network resilience, and export structured data into standard CSV and JSON formats.

Project Core Objectives

• HTTP Interaction & Rate-Limiting: Request web content politely using custom User-Agent headers and artificial delays (politeness policy) to avoid triggering IP blocks.
• DOM Parsing & Extraction: Navigate complex HTML elements using BeautifulSoup to target, extract, and clean text, tags, and attribute values.
• Data Pipeline & Transformation: Clean raw strings (removing whitespace, parsing numerical values, standardizing dates) into strongly typed dictionaries.
• Multi-Format Export: Output sanitized data into standard flat formats (CSV) and nested hierarchical formats (JSON).

2. Functional Requirements & Feature Specifications

Your Web Scraper & Data Pipeline must implement the following technical specs:

Feature Component	Functional Specification	Python Mechanism
HTTP Request Engine	Fetches web pages with custom request headers and timeout handling.	requests.get(), headers, timeout parameters
HTML Parsing	Parses page HTML and extracts specific tags via CSS selectors or class names.	bs4.BeautifulSoup, select(), find_all()
Data Sanitization Pipeline	Strips whitespace, converts price strings to floats, and filters missing values.	String manipulation, List comprehensions, RegEx (re)
Export Subsystem	Saves cleaned records to target files dynamically based on user preference.	csv.DictWriter, json.dump()

3. Complete Reference Implementation

Study the reference implementation below. It demonstrates a resilient web scraper and data pipeline targeting a sample scraping endpoint (quotes.toscrape.com):

import csv
import json
import time
import logging
from pathlib import Path
import requests
from bs4 import BeautifulSoup

# Setup Logging
logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s - %(levelname)s - %(message)s"
)

# Configuration Constants
BASE_URL = "http://quotes.toscrape.com"
HEADERS = {
    "User-Agent": "DataPipelineBot/1.0 (Educational Scraping Script; contact@example.com)"
}
OUTPUT_DIR = Path("scraped_data")

def fetch_page(url):
    """Fetches HTML content from a URL with robust error handling."""
    try:
        response = requests.get(url, headers=HEADERS, timeout=10)
        response.raise_for_status()
        return response.text
    except requests.exceptions.RequestException as err:
        logging.error(f"Failed to fetch {url}: {err}")
        return None

def parse_quotes(html_content):
    """Parses raw HTML and extracts raw quote records."""
    soup = BeautifulSoup(html_content, "html.parser")
    quote_elements = soup.select("div.quote")
    extracted_records = []

    for element in quote_elements:
        text_el = element.select_one("span.text")
        author_el = element.select_one("small.author")
        tags_els = element.select("div.tags a.tag")

        raw_text = text_el.get_text() if text_el else ""
        raw_author = author_el.get_text() if author_el else ""
        raw_tags = [tag.get_text() for tag in tags_els]

        extracted_records.append({
            "quote": raw_text,
            "author": raw_author,
            "tags": raw_tags
        })

    return extracted_records

def clean_record(record):
    """Data Pipeline Stage: Sanitizes and transforms a single record."""
    cleaned_quote = record["quote"].strip("“”\"' \t\n")
    cleaned_author = record["author"].strip().title()
    cleaned_tags = [tag.strip().lower() for tag in record["tags"]]

    return {
        "quote": cleaned_quote,
        "author": cleaned_author,
        "tag_count": len(cleaned_tags),
        "tags_csv": ", ".join(cleaned_tags)
    }

def export_to_csv(data, filename="quotes.csv"):
    """Exports processed records to a CSV file."""
    if not data:
        print("⚠️ No data to export to CSV.")
        return

    OUTPUT_DIR.mkdir(exist_ok=True)
    file_path = OUTPUT_DIR / filename
    
    fieldnames = ["quote", "author", "tag_count", "tags_csv"]
    try:
        with open(file_path, "w", newline="", encoding="utf-8") as csvfile:
            writer = csv.DictWriter(csvfile, fieldnames=fieldnames)
            writer.writeheader()
            writer.writerows(data)
        print(f"📄 Saved CSV data to '{file_path}'")
    except IOError as err:
        logging.error(f"Failed to write CSV: {err}")

def export_to_json(data, filename="quotes.json"):
    """Exports processed records to a JSON file."""
    if not data:
        print("⚠️ No data to export to JSON.")
        return

    OUTPUT_DIR.mkdir(exist_ok=True)
    file_path = OUTPUT_DIR / filename

    try:
        with open(file_path, "w", encoding="utf-8") as jsonfile:
            json.dump(data, jsonfile, indent=4, ensure_ascii=False)
        print(f"📦 Saved JSON data to '{file_path}'")
    except IOError as err:
        logging.error(f"Failed to write JSON: {err}")

def run_pipeline(max_pages=2):
    """Main execution pipeline: fetch -> parse -> transform -> export."""
    all_cleaned_data = []

    print(f"\n🚀 Starting Data Pipeline (Targeting up to {max_pages} pages)...")
    for page_num in range(1, max_pages + 1):
        target_url = f"{BASE_URL}/page/{page_num}/"
        logging.info(f"Scraping Page {page_num}: {target_url}")

        html = fetch_page(target_url)
        if not html:
            break

        raw_records = parse_quotes(html)
        if not raw_records:
            logging.warning(f"No records found on page {page_num}.")
            break

        # Transform records through pipeline
        for record in raw_records:
            cleaned = clean_record(record)
            all_cleaned_data.append(cleaned)

        # Politeness delay between requests
        time.sleep(1.0)

    print(f"\n✅ Extracted and cleaned {len(all_cleaned_data)} total records.")
    
    # Export results
    export_to_csv(all_cleaned_data)
    export_to_json(all_cleaned_data)

def main():
    """Terminal Menu interface."""
    while True:
        print("\n==================================")
        print("   🕸️ WEB SCRAPER & DATA PIPELINE")
        print("==================================")
        print("1) Run Pipeline (2 Pages)")
        print("2) Custom Page Depth Run")
        print("3) Exit")

        choice = input("\nSelect choice (1-3): ").strip()

        if choice == "1":
            run_pipeline(max_pages=2)
        elif choice == "2":
            try:
                pages = int(input("Enter max pages to scrape (1-5): "))
                if 1 <= pages <= 5:
                    run_pipeline(max_pages=pages)
                else:
                    print("⚠️ Please enter a number between 1 and 5.")
            except ValueError:
                print("⚠️ Invalid number entry.")
        elif choice == "3":
            print("\nExiting Data Pipeline tool. Happy scraping!")
            break
        else:
            print("⚠️ Invalid option.")

if __name__ == "__main__":
    main()

4. Common Project Pitfalls

5. Interactive Project Workspace

Project Workspace

Build & Test: Web Scraper & Data Pipeline

Use the editor below to craft your data scraper. Try adding extra sanitization logic, like stripping special characters or extracting author URL links!

0% complete

Your Project Code

Project Feature Checklist

I included a standard User-Agent header and timeout handling I extracted targeted tags clean using BeautifulSoup selectors I implemented a data cleaning transformation step I exported the final dataset to CSV and JSON formats

6. Key Takeaways & Milestone Achievement

• Scraping pipelines require clear isolation between fetching, parsing, cleaning, and storing data.
• Always follow web politeness guidelines (rate limiting, standard headers, respecting site terms).
• Storing data in both flat (CSV) and semi-structured (JSON) formats ensures high compatibility for downstream data analysis tools.