scraper.py

#!/usr/bin/env python3
"""
playwright_js_scraper.py
========================
Async scraper for JavaScript-rendered pages with infinite scroll support.

Target (demo): https://quotes.toscrape.com/js/
- Page content is injected by JavaScript — requests/BeautifulSoup returns empty
- Infinite scroll loads new quotes as you reach the bottom
- Author detail pages are scraped for bio + birthdate

Usage:
    pip install -r requirements.txt
    playwright install chromium

    python scraper.py                          # scrape all pages, save to output/
    python scraper.py --max-pages 5            # limit to 5 scroll pages
    python scraper.py --out results.json       # custom output file
    python scraper.py --headless false         # watch the browser (debug mode)

Output:
    output/quotes.csv   — flat CSV, one row per quote
    output/quotes.json  — structured JSON with nested author detail
"""

import asyncio
import argparse
import csv
import json
import logging
import sys
import time
from datetime import datetime
from pathlib import Path

from playwright.async_api import async_playwright, Page, TimeoutError as PWTimeout

# ── Logging ────────────────────────────────────────────────────────────────────
logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s  %(levelname)-7s  %(message)s",
    datefmt="%H:%M:%S",
)
log = logging.getLogger("js-scraper")

# ── Config ─────────────────────────────────────────────────────────────────────
BASE_URL        = "https://quotes.toscrape.com/js/"
AUTHOR_BASE_URL = "https://quotes.toscrape.com"
OUTPUT_DIR      = Path("output")
DEFAULT_MAX_PAGES = 0           # 0 = scrape until no more content
SCROLL_PAUSE_MS   = 1200        # ms to wait after each scroll before checking DOM
PAGE_TIMEOUT_MS   = 15_000      # max wait for element to appear
USER_AGENT        = (
    "Mozilla/5.0 (Windows NT 10.0; Win64; x64) "
    "AppleWebKit/537.36 (KHTML, like Gecko) "
    "Chrome/124.0.0.0 Safari/537.36"
)


# ── Data extraction helpers ────────────────────────────────────────────────────

async def extract_quotes_from_page(page: Page) -> list[dict]:
    """Parse all quote cards currently visible in the DOM."""
    return await page.evaluate("""
        () => {
            const cards = document.querySelectorAll('.quote');
            return Array.from(cards).map(card => ({
                text: card.querySelector('.text')?.innerText?.trim() ?? '',
                author: card.querySelector('.author')?.innerText?.trim() ?? '',
                author_url: card.querySelector('a')?.getAttribute('href') ?? '',
                tags: Array.from(card.querySelectorAll('.tag')).map(t => t.innerText.trim()),
            }));
        }
    """)


async def scrape_author_detail(page: Page, author_url: str) -> dict:
    """
    Visit an author's detail page and extract bio + birth info.
    Returns empty dict if the page fails or author_url is blank.
    """
    if not author_url:
        return {}
    full_url = f"{AUTHOR_BASE_URL}{author_url}"
    try:
        await page.goto(full_url, wait_until="domcontentloaded", timeout=PAGE_TIMEOUT_MS)
        await page.wait_for_selector(".author-details", timeout=PAGE_TIMEOUT_MS)
    except PWTimeout:
        log.warning("Timeout fetching author page: %s", full_url)
        return {}

    return await page.evaluate("""
        () => ({
            born_date:  document.querySelector('.author-born-date')?.innerText?.trim() ?? '',
            born_place: document.querySelector('.author-born-location')?.innerText?.trim() ?? '',
            description: document.querySelector('.author-description')?.innerText?.trim() ?? '',
        })
    """)


# ── Infinite scroll logic ──────────────────────────────────────────────────────

async def scroll_until_loaded(page: Page, max_pages: int) -> list[dict]:
    """
    Scroll to the bottom of the page repeatedly, collecting quotes.
    The JS site paginates via a 'Next' button rather than true infinite scroll,
    so we click 'Next' until it disappears or we hit max_pages.
    """
    all_quotes: list[dict] = []
    seen_texts: set[str] = set()
    page_num = 1

    while True:
        await page.wait_for_selector(".quote", timeout=PAGE_TIMEOUT_MS)
        quotes = await extract_quotes_from_page(page)

        new = [q for q in quotes if q["text"] not in seen_texts]
        for q in new:
            seen_texts.add(q["text"])
        all_quotes.extend(new)

        log.info("Page %d — %d quotes found (+%d new, %d total)",
                 page_num, len(quotes), len(new), len(all_quotes))

        if max_pages and page_num >= max_pages:
            log.info("Reached max_pages=%d, stopping.", max_pages)
            break

        # Click 'Next' if it exists
        next_btn = page.locator("li.next > a")
        if await next_btn.count() == 0:
            log.info("No 'Next' button — all pages scraped.")
            break

        await next_btn.click()
        await asyncio.sleep(SCROLL_PAUSE_MS / 1000)
        page_num += 1

    return all_quotes


# ── Output writers ─────────────────────────────────────────────────────────────

def save_csv(records: list[dict], path: Path) -> None:
    path.parent.mkdir(parents=True, exist_ok=True)
    if not records:
        log.warning("No records to write.")
        return
    fieldnames = list(records[0].keys())
    with open(path, "w", newline="", encoding="utf-8") as f:
        writer = csv.DictWriter(f, fieldnames=fieldnames)
        writer.writeheader()
        for row in records:
            # flatten tags list to pipe-separated string for CSV
            flat = {**row, "tags": " | ".join(row.get("tags", []))}
            writer.writerow(flat)
    log.info("CSV saved → %s (%d rows)", path, len(records))


def save_json(records: list[dict], path: Path) -> None:
    path.parent.mkdir(parents=True, exist_ok=True)
    with open(path, "w", encoding="utf-8") as f:
        json.dump(records, f, ensure_ascii=False, indent=2)
    log.info("JSON saved → %s (%d records)", path, len(records))


# ── Main pipeline ──────────────────────────────────────────────────────────────

async def run(args: argparse.Namespace) -> None:
    headless = args.headless.lower() != "false"
    max_pages = args.max_pages

    out_stem = Path(args.out).stem if args.out else "quotes"
    csv_path  = OUTPUT_DIR / f"{out_stem}.csv"
    json_path = OUTPUT_DIR / f"{out_stem}.json"

    async with async_playwright() as pw:
        browser = await pw.chromium.launch(headless=headless)
        context = await browser.new_context(
            user_agent=USER_AGENT,
            viewport={"width": 1280, "height": 800},
        )

        # Main scraping page
        page = await context.new_page()
        log.info("Navigating to %s", BASE_URL)
        await page.goto(BASE_URL, wait_until="networkidle", timeout=30_000)

        quotes = await scroll_until_loaded(page, max_pages)
        log.info("Collected %d quotes total.", len(quotes))

        # Author detail enrichment (deduplicated by author name)
        author_page  = await context.new_page()
        author_cache: dict[str, dict] = {}

        log.info("Enriching author details...")
        for q in quotes:
            name = q["author"]
            if name not in author_cache:
                detail = await scrape_author_detail(author_page, q["author_url"])
                author_cache[name] = detail
                if detail:
                    log.info("  ✓ %s", name)
            q["author_detail"] = author_cache[name]

        await browser.close()

    # Flatten for CSV (author_detail is nested — expand inline)
    flat_records = []
    for q in quotes:
        detail = q.pop("author_detail", {})
        flat_records.append({
            **q,
            "born_date":    detail.get("born_date", ""),
            "born_place":   detail.get("born_place", ""),
            "bio_snippet":  detail.get("description", "")[:200],
        })

    save_csv(flat_records, csv_path)
    save_json(quotes, json_path)

    print(f"\nDone. {len(quotes)} quotes scraped.")
    print(f"  CSV  → {csv_path}")
    print(f"  JSON → {json_path}")


# ── CLI ────────────────────────────────────────────────────────────────────────

def parse_args() -> argparse.Namespace:
    p = argparse.ArgumentParser(
        description="Async Playwright scraper for JS-rendered quote pages."
    )
    p.add_argument("--max-pages", type=int, default=DEFAULT_MAX_PAGES,
                   help="Max pagination pages to scrape (0 = all, default: 0)")
    p.add_argument("--out", type=str, default="",
                   help="Output filename stem (default: quotes)")
    p.add_argument("--headless", type=str, default="true",
                   help="Run browser headless (default: true). Set 'false' to watch.")
    return p.parse_args()


if __name__ == "__main__":
    args = parse_args()
    asyncio.run(run(args))