How to scrape product prices to CSV (and re-run it next week)

<article class="product_pod">
  <h3><a href="catalogue/a-light-in-the-attic_1000/index.html"
         title="A Light in the Attic">A Light in the ...</a></h3>
  <p class="price_color">£51.77</p>
</article>

"""Scrape book titles, prices and URLs from books.toscrape.com.

books.toscrape.com is a public practice shop built for scraping tutorials.
The books are real; the prices are made up for the exercise.

Run:
    python3 tutorial.py
"""

import csv
import time
from urllib.parse import urljoin

from playwright.sync_api import sync_playwright

START_URL = "https://books.toscrape.com/"
MAX_PAGES = 3        # demo cap — there are 50 pages in total, 1,000 books
DELAY_SECONDS = 1.0  # pause between page loads; be polite to the site


def scrape_page(page):
    """Read every book card on the page currently loaded."""
    rows = []
    for card in page.query_selector_all("article.product_pod"):
        link = card.query_selector("h3 a")
        rows.append({
            "title": link.get_attribute("title"),
            "price": card.query_selector("p.price_color").inner_text(),
            "url": urljoin(page.url, link.get_attribute("href")),
        })
    return rows


def main():
    with sync_playwright() as p:
        browser = p.chromium.launch()
        page = browser.new_page()
        page.goto(START_URL)

        rows = []
        pages_done = 0
        while True:
            rows += scrape_page(page)
            pages_done += 1
            if pages_done >= MAX_PAGES:
                break
            next_link = page.query_selector("li.next a")
            if next_link is None:
                break
            next_link.click()
            page.wait_for_load_state()
            time.sleep(DELAY_SECONDS)

        browser.close()

    with open("prices.csv", "w", newline="", encoding="utf-8") as f:
        writer = csv.DictWriter(f, fieldnames=["title", "price", "url"])
        writer.writeheader()
        writer.writerows(rows)

    print(f"Wrote {len(rows)} rows to prices.csv")


if __name__ == "__main__":
    main()

$ python3 tutorial.py
Wrote 60 rows to prices.csv

title,price,url
A Light in the Attic,£51.77,https://books.toscrape.com/catalogue/a-light-in-the-attic_1000/index.html
Tipping the Velvet,£53.74,https://books.toscrape.com/catalogue/tipping-the-velvet_999/index.html

$ diff prices.csv prices-next-week.csv
2c2
< A Light in the Attic,£51.77,https://books.toscrape.com/catalogue/a-light-in-the-attic_1000/index.html
---
> A Light in the Attic,£48.99,https://books.toscrape.com/catalogue/a-light-in-the-attic_1000/index.html

0 8 * * 1 cd /home/you/prices && /home/you/prices/.venv/bin/python tutorial.py >> prices.log 2>&1

"""Scrape books.toscrape.com — generated by Scrapejig on 2026-09-10.

Setup (one time):
    pip3 install playwright
    python3 -m playwright install chromium

Run:
    python3 scrape_books_toscrape_com.py

If pip3 says "externally-managed-environment" (Homebrew, Debian, Ubuntu 23.04+),
create a virtual environment and run all three lines inside it:
    python3 -m venv .venv && source .venv/bin/activate
Every new terminal needs that activate line again before the script will run.

On Windows the launcher is py: py -m pip install playwright, then
py -m playwright install chromium, then py scrape_books_toscrape_com.py.
A venv activates there with .venv/Scripts/activate in place of the source line.

Output: scrape_output.csv and scrape_output.json in the working directory.
This file is yours — edit it freely. The constants below are the usual knobs.
"""

import csv
import json
import sys
import time
from urllib.parse import urljoin

from playwright.sync_api import sync_playwright

START_URL = "https://books.toscrape.com/"
HEADLESS = True
OUTPUT_BASE = "scrape_output"
DELAY_SECONDS = 1.0  # pause between page loads; raise to be gentler
ITEM_SELECTOR = "article.product_pod"
FIELDS = ["title", "price", "url"]
NEXT_SELECTOR = "li.next a"
MAX_PAGES = 3  # set in the panel; change to None to scrape every page


def extract_item(item, base):
    """Pull one row's fields from a single item element.

    base is the URL of the page this item came from. Relative links resolve
    against it, which from page 2 onward is no longer START_URL.
    """
    return {
        "title": attr(item, "h3 a", "title", base),
        "price": text(item, "p.price_color"),
        "url": attr(item, "h3 a", "href", base),
    }


def rendered_text(node):
    """The text a node renders. inner_text() is HTML-only and raises on SVG and other
    non-HTML nodes, where text_content() asks the same question and answers it.
    """
    try:
        return node.inner_text().strip()
    except Exception:
        pass
    try:
        return (node.text_content() or "").strip()
    except Exception:
        return ""


def text(el, selector):
    node = el.query_selector(selector)
    return rendered_text(node) if node else ""


def attr(el, selector, name, base):
    node = el.query_selector(selector)
    try:
        value = (node.get_attribute(name) or "") if node else ""
    except Exception:
        return ""
    if value and name in ("href", "src"):
        value = urljoin(base, value)
    return value


def rows_on(page):
    """Every real item on the page in front of us.

    A row whose every column came back empty means the item selector matched something that is
    not an item — a page section sitting beside the real cards — so it is dropped rather than
    padding the output with blank lines.
    """
    rows = []
    for item in page.query_selector_all(ITEM_SELECTOR):
        row = extract_item(item, page.url)
        if any(row.values()):
            rows.append(row)
    return rows


# The tail every start-page failure shares. A wall served to a fresh profile explains a
# navigation that hangs, a 403, and a document with no items in it equally well.
BLOCKED_ADVICE = (
    "A site will often serve a bot check or a cookie wall to a fresh headless profile "
    "instead of the page. Set HEADLESS = False to see what came back, and if it is a "
    "banner or a login, record a session with the storage_state note in main()."
)


def open_start_page(page):
    """Load the start page and wait for the items, explaining what went wrong before it bites."""
    try:
        # "domcontentloaded", not Playwright's default "load". The default waits for every
        # subresource the page pulls in, so one hanging tracker or ad pixel times the navigation
        # out thirty seconds after the markup — all of it, items included — already arrived.
        # Readiness here is "are the items there", and the wait below asks exactly that.
        response = page.goto(START_URL, wait_until="domcontentloaded")
    except Exception:
        print(f"Could not load {START_URL}: the document never arrived.", file=sys.stderr)
        print(BLOCKED_ADVICE, file=sys.stderr)
        raise
    # goto() raises on a transport failure but not on an HTTP one: 404, 403 and 500 are all
    # successful navigations to a page that says no. Unchecked, they fall through to the wait
    # below, cost its full timeout, and are then reported as something they are not.
    if response is not None and not response.ok:
        print(
            f"{START_URL} answered HTTP {response.status}, so the items never loaded.",
            file=sys.stderr,
        )
        print(
            "A 404 or 410 means START_URL is wrong or the listing has moved — open it in a "
            "normal browser and check. A 401, 403 or 429 is usually a wall rather than a "
            "verdict on the URL:",
            file=sys.stderr,
        )
        print(BLOCKED_ADVICE, file=sys.stderr)
        raise RuntimeError(f"HTTP {response.status} at {START_URL}")
    try:
        page.wait_for_selector(ITEM_SELECTOR)
    except Exception:
        print(f"Nothing matched ITEM_SELECTOR on {START_URL}.", file=sys.stderr)
        print(BLOCKED_ADVICE, file=sys.stderr)
        raise


def scrape(page):
    """Walk every page via the next link, collecting items as we go."""
    open_start_page(page)
    rows = []
    pages_done = 0
    while True:
        rows += rows_on(page)
        pages_done += 1
        if MAX_PAGES is not None and pages_done >= MAX_PAGES:
            break
        next_link = page.query_selector(NEXT_SELECTOR)
        if next_link is None:
            break
        previous_url = page.url
        try:
            next_link.click()
            page.wait_for_load_state()
            page.wait_for_selector(ITEM_SELECTOR)
        except Exception as error:
            # Usually a consent banner over the link: a fresh profile has agreed to nothing, so
            # the overlay is there and it eats the click (see the storage_state note in main()).
            # The pages already walked are still yours; losing them to the last click is not.
            print(
                f"Stopped paginating: {error} — keeping the {len(rows)} rows collected so far.",
                file=sys.stderr,
            )
            break
        time.sleep(DELAY_SECONDS)
        # A next arrow that never disappears would loop forever, so stop if the URL
        # did not move. Site paginates without changing it? Delete this, set MAX_PAGES.
        if page.url == previous_url:
            break
    return rows


def main():
    with sync_playwright() as p:
        browser = p.chromium.launch(headless=HEADLESS)
        page = browser.new_page()
        # Cookie banner or login wall? This browser starts with a clean profile that has
        # agreed to nothing, so plenty of sites — most UK and EU ones — cover the page with a
        # consent overlay that swallows every click, and a login-gated site never shows you
        # the content at all. Fix both the same way: record a session once with
        #     python3 -m playwright codegen --save-storage=auth.json "https://books.toscrape.com/"
        # dismiss the banner in the window that opens (and sign in, if you need to), close it,
        # then replace the new_page() line above with:
        #     page = browser.new_context(storage_state="auth.json").new_page()
        rows = scrape(page)
        browser.close()
    write_output(rows)
    print(f"Scraped {len(rows)} rows -> {OUTPUT_BASE}.csv and {OUTPUT_BASE}.json")


def write_output(rows):
    with open(f"{OUTPUT_BASE}.csv", "w", newline="", encoding="utf-8") as f:
        writer = csv.DictWriter(f, fieldnames=FIELDS)
        writer.writeheader()
        writer.writerows(rows)
    with open(f"{OUTPUT_BASE}.json", "w", encoding="utf-8") as f:
        json.dump(rows, f, indent=2, ensure_ascii=False)


if __name__ == "__main__":
    main()