Hands-on lab — Day 79: Web Scraping Responsibly

"""A polite, cached, parser-based scraper for the fixture catalogue.

This is the reference implementation. ``starter/catalogue_scraper.py`` is the
same file with six numbered exercises hollowed out; the same test suite runs
against both, which is how you know the exercises are load-bearing.

The shape of the module is the lesson in code, in this order:

1. ``RobotsPolicy``  — permission, decided BEFORE any page is fetched.
2. ``ResponseCache`` — a page fetched once is never fetched again.
3. ``fetch_text``    — the only function that is allowed to make a request.
4. ``parse_items``   — pure: HTML in, ``Item`` objects out, no network.
5. ``next_page_url`` — pure: finds the pagination link, or ``None`` at the end.
6. ``scrape_catalogue`` / ``write_csv`` — the loop and the output.

Every boundary — the network session, the clock, the cache directory — is a
parameter rather than a global. That is Day 74's rule, and it is the reason
the tests can run this scraper with a fake clock and count the sleeps it asked
for without ever waiting a real second.
"""

from __future__ import annotations

import csv
import hashlib
import json
import time
import urllib.robotparser
from dataclasses import dataclass
from pathlib import Path
from typing import Callable, Iterable, Sequence
from urllib.parse import urljoin, urlparse

import requests
from bs4 import BeautifulSoup

# Identify yourself honestly, and leave a way to be contacted. A site owner
# who can see who you are and reach you will usually ask you to slow down
# before they block you. `example.com` is reserved by the IETF for
# documentation, so this address belongs to nobody; put a real one in a real
# scraper.
USER_AGENT = "HarbourCatalogueLab/1.0 (course exercise; contact: scraper-owner@example.com)"

DEFAULT_TIMEOUT = 10.0
FALLBACK_DELAY_SECONDS = 1.0


class ScrapeError(Exception):
    """Base class for everything this module refuses to do."""


class DisallowedByRobots(ScrapeError):
    """The path is disallowed by robots.txt for our User-Agent."""


class OffSite(ScrapeError):
    """The URL points somewhere other than the site we were asked to scrape."""


# ---------------------------------------------------------------------------
# 1. Permission
# ---------------------------------------------------------------------------


@dataclass
class RobotsPolicy:
    """A parsed robots.txt, and the two questions worth asking it.

    Built with :meth:`load` so that the fetch of robots.txt itself goes through
    our own session with our own User-Agent. ``RobotFileParser.read()`` would
    also work, but it opens its own connection with urllib's default header,
    which means the site cannot tell it was us.
    """

    parser: urllib.robotparser.RobotFileParser
    user_agent: str

    @classmethod
    def load(
        cls,
        base_url: str,
        *,
        session: requests.Session,
        user_agent: str = USER_AGENT,
        timeout: float = DEFAULT_TIMEOUT,
    ) -> "RobotsPolicy":
        robots_url = urljoin(base_url, "/robots.txt")
        parser = urllib.robotparser.RobotFileParser()
        parser.set_url(robots_url)
        response = session.get(
            robots_url, headers={"User-Agent": user_agent}, timeout=timeout
        )
        if response.status_code == 200:
            parser.parse(response.text.splitlines())
        elif response.status_code in (401, 403):
            # RFC 9309: an authentication or forbidden status means the whole
            # site is off limits. Treat it that way rather than shrugging.
            parser.disallow_all = True
        else:
            # 404 and other 4xx mean "no rules published", which permits all.
            parser.allow_all = True
        return cls(parser=parser, user_agent=user_agent)

    def allows(self, url: str) -> bool:
        """Is this exact URL allowed for our User-Agent?"""
        return self.parser.can_fetch(self.user_agent, url)

    def crawl_delay_seconds(self, default: float = FALLBACK_DELAY_SECONDS) -> float:
        """Seconds to wait between requests, per robots.txt, else ``default``.

        Note that ``RobotFileParser`` only accepts whole-number crawl delays;
        a fractional value in robots.txt is ignored and you get ``None`` here.
        """
        declared = self.parser.crawl_delay(self.user_agent)
        return float(declared) if declared is not None else default


# ---------------------------------------------------------------------------
# 2. The cache
# ---------------------------------------------------------------------------


@dataclass
class ResponseCache:
    """Response bodies stored on disk, keyed by URL.

    The point is not speed, though it is faster. The point is that while you
    are getting the parser right — and you will iterate on the parser twenty
    times — the site is asked for each page exactly once. Re-running your
    program should cost the person who owns that server nothing.
    """

    directory: Path
    hits: int = 0
    misses: int = 0

    def __post_init__(self) -> None:
        self.directory = Path(self.directory)
        self.directory.mkdir(parents=True, exist_ok=True)

    def _path_for(self, url: str) -> Path:
        digest = hashlib.sha256(url.encode("utf-8")).hexdigest()[:32]
        return self.directory / f"{digest}.html"

    def get(self, url: str) -> str | None:
        path = self._path_for(url)
        if path.exists():
            self.hits += 1
            return path.read_text(encoding="utf-8")
        self.misses += 1
        return None

    def put(self, url: str, text: str) -> None:
        self._path_for(url).write_text(text, encoding="utf-8")
        # A human-readable map, so a cache directory is inspectable rather
        # than being 32 hex characters of mystery.
        index_path = self.directory / "index.json"
        index: dict[str, str] = {}
        if index_path.exists():
            index = json.loads(index_path.read_text(encoding="utf-8"))
        index[self._path_for(url).name] = url
        index_path.write_text(json.dumps(index, indent=2, sort_keys=True), encoding="utf-8")


# ---------------------------------------------------------------------------
# 3. The only function allowed to make a request
# ---------------------------------------------------------------------------


def fetch_text(
    url: str,
    *,
    session: requests.Session,
    robots: RobotsPolicy,
    cache: ResponseCache | None = None,
    sleeper: Callable[[float], None] = time.sleep,
    delay_seconds: float | None = None,
    user_agent: str = USER_AGENT,
    timeout: float = DEFAULT_TIMEOUT,
) -> str:
    """Return the body of ``url``, from cache if possible, politely if not.

    Raises :class:`DisallowedByRobots` before any socket is opened if the path
    is refused. The order matters: check, then cache, then sleep, then fetch.
    """
    if not robots.allows(url):
        raise DisallowedByRobots(url)

    if cache is not None:
        cached = cache.get(url)
        if cached is not None:
            return cached

    # Only a real fetch pays the crawl delay. A cache hit costs the site
    # nothing, so waiting for it would be politeness theatre.
    wait = robots.crawl_delay_seconds() if delay_seconds is None else delay_seconds
    if wait > 0:
        sleeper(wait)

    response = session.get(url, headers={"User-Agent": user_agent}, timeout=timeout)
    response.raise_for_status()
    text = response.text
    if cache is not None:
        cache.put(url, text)
    return text


# ---------------------------------------------------------------------------
# 4. Parsing — pure functions, no network
# ---------------------------------------------------------------------------


@dataclass(frozen=True)
class Item:
    """One catalogue row. ``price`` is ``None`` when the page omits the cell."""

    sku: str
    name: str
    price: float | None
    stock: int | None
    source_path: str


def _cell_text(row, selector: str, *, drop: str | None = None) -> str | None:
    """Text of the first cell matching ``selector``, or ``None`` if absent.

    Three habits live in these few lines and all three matter more than they
    look.

    ``select_one`` returns ``None`` rather than raising, so the caller decides
    what a missing cell means — this is the single most common scraping bug,
    and the fix is to never chain ``.text`` onto a lookup you have not checked.

    ``get_text(" ", strip=True)`` collapses the whitespace that HTML authors
    put around text for readability, and BeautifulSoup has already turned
    ``&`` back into ``&`` by this point.

    ``drop`` removes elements you do not want swept up. ``get_text`` returns
    the text of *every* descendant, so a decorative ``<span class="badge">new</span>``
    inside a name cell silently becomes part of the product name unless you
    say otherwise. Naming your exclusions is cheaper than explaining later why
    forty rows are called "Something new".
    """
    cell = row.select_one(selector)
    if cell is None:
        return None
    if drop:
        for extra in cell.select(drop):
            extra.extract()
    return cell.get_text(" ", strip=True)


def parse_items(html: str, *, source_path: str = "") -> list[Item]:
    """Extract every catalogue row from one page.

    ``tr.item`` matches rows whose class *contains* ``item``, which is why the
    ``class="name featured"`` cell below is found by ``td.name`` while the
    obvious regular expression for ``class="name"`` misses it entirely.
    """
    soup = BeautifulSoup(html, "html.parser")
    items: list[Item] = []
    for row in soup.select("table.catalogue tr.item"):
        sku = row.get("data-sku", "")
        name = _cell_text(row, "td.name", drop="span.badge") or ""
        price_text = _cell_text(row, "td.price")
        stock_text = _cell_text(row, "td.stock")
        items.append(
            Item(
                sku=sku,
                name=name,
                price=float(price_text) if price_text else None,
                stock=int(stock_text) if stock_text else None,
                source_path=source_path,
            )
        )
    return items


def next_page_url(html: str, current_url: str) -> str | None:
    """Absolute URL of the next page, or ``None`` when this was the last one.

    The last page carries ``<span class="next disabled">`` instead of a link,
    so ``a.next`` finds nothing and the loop stops. Do not stop on an item
    count or a page number — pages get added.
    """
    soup = BeautifulSoup(html, "html.parser")
    link = soup.select_one("nav.pager a.next")
    if link is None:
        return None
    href = link.get("href")
    if not href:
        return None
    return urljoin(current_url, href)


# ---------------------------------------------------------------------------
# 5. The loop
# ---------------------------------------------------------------------------


def scrape_catalogue(
    start_url: str,
    *,
    session: requests.Session,
    robots: RobotsPolicy,
    cache: ResponseCache | None = None,
    sleeper: Callable[[float], None] = time.sleep,
    delay_seconds: float | None = None,
    max_pages: int = 50,
) -> list[Item]:
    """Follow the pagination from ``start_url`` and return every item found.

    ``max_pages`` and the ``seen`` set are not decoration. A pagination loop
    that points back at itself is common enough that an unbounded crawler will
    eventually hammer somebody's server until they block you.
    """
    items: list[Item] = []
    seen: set[str] = set()
    url: str | None = start_url
    origin = urlparse(start_url)

    while url is not None and len(seen) < max_pages:
        if url in seen:
            break
        if urlparse(url).netloc != origin.netloc:
            raise OffSite(url)
        seen.add(url)
        html = fetch_text(
            url,
            session=session,
            robots=robots,
            cache=cache,
            sleeper=sleeper,
            delay_seconds=delay_seconds,
        )
        items.extend(parse_items(html, source_path=urlparse(url).path))
        url = next_page_url(html, url)

    return items


def follow_links_politely(
    html: str,
    base_url: str,
    *,
    robots: RobotsPolicy,
) -> tuple[list[str], list[str]]:
    """Split every link on a page into (allowed, refused) by robots.txt.

    This is the function the harness uses to prove the ethics were implemented
    rather than described: the disallowed link is present in the HTML, it is
    discovered, and it is put in the refused list instead of being requested.
    """
    soup = BeautifulSoup(html, "html.parser")
    allowed: list[str] = []
    refused: list[str] = []
    for anchor in soup.select("a[href]"):
        target = urljoin(base_url, anchor["href"])
        (allowed if robots.allows(target) else refused).append(target)
    return allowed, refused


# ---------------------------------------------------------------------------
# 6. Output
# ---------------------------------------------------------------------------

CSV_COLUMNS: Sequence[str] = ("sku", "name", "price", "stock", "source_path")


def write_csv(items: Iterable[Item], path: Path) -> Path:
    """Write items to ``path`` as CSV, using Day 65's rules.

    ``newline=""`` is required by the csv module, a missing price becomes an
    empty field rather than the string ``None``, and the file is UTF-8 so the
    ampersand-and-friends survive.
    """
    path = Path(path)
    with path.open("w", encoding="utf-8", newline="") as handle:
        writer = csv.writer(handle)
        writer.writerow(CSV_COLUMNS)
        for item in items:
            writer.writerow(
                [
                    item.sku,
                    item.name,
                    "" if item.price is None else f"{item.price:.2f}",
                    "" if item.stock is None else item.stock,
                    item.source_path,
                ]
            )
    return path

"""The whole lab end to end, against the local fixture server.

Run it from the lab directory::

    PYTHONPATH=examples python3 examples/demo.py catalogue.csv

What you will see, in order:

1. robots.txt fetched and parsed, and the two questions asked of it.
2. Every link on page 1 sorted into allowed and refused — the refused one is
   never requested, and the server's own log at the end proves it.
3. A first scraping run: three pages fetched, twelve items found, one of them
   with no price at all.
4. A second run with the same cache: zero page requests. Only robots.txt is
   re-fetched, because permission is a thing you re-check, not a thing you
   cache.
5. The CSV.
6. The server's access log, which is the honest record of what we did.

Nothing here reaches the internet. The server is on 127.0.0.1 on a port the
operating system picked, and it stops when this script does.
"""

from __future__ import annotations

import shutil
import sys
import tempfile
from pathlib import Path

import requests

from catalogue_scraper import (
    USER_AGENT,
    DisallowedByRobots,
    ResponseCache,
    RobotsPolicy,
    fetch_text,
    follow_links_politely,
    scrape_catalogue,
    write_csv,
)
from fixture_server import serve_fixtures


def main(argv: list[str]) -> int:
    csv_path = Path(argv[1]) if len(argv) > 1 else Path("catalogue.csv")
    cache_dir = Path(tempfile.mkdtemp(prefix="scrape-cache-"))
    recorded_sleeps: list[float] = []

    with serve_fixtures() as site, requests.Session() as session:
        print("fixture server: 127.0.0.1 on an ephemeral port")
        print(f"User-Agent:     {USER_AGENT}")
        print("note:           the crawl delay is recorded rather than slept, so this")
        print("                demo finishes instantly; a real run waits the full")
        print("                delay before every fetch that is not a cache hit.")
        print()

        # 1 -----------------------------------------------------------------
        print("1. Permission")
        robots = RobotsPolicy.load(site.base_url, session=session)
        start = site.url_for("/catalogue/page-1.html")
        private = site.url_for("/private/internal-notes.html")
        print(f"   allowed  /catalogue/page-1.html   -> {robots.allows(start)}")
        print(f"   allowed  /private/internal-notes.html -> {robots.allows(private)}")
        print(f"   crawl delay declared               -> {robots.crawl_delay_seconds()} s")
        greedy = RobotsPolicy.load(site.base_url, session=session, user_agent="GreedyBot")
        print(f"   allowed for GreedyBot              -> {greedy.allows(start)}")
        print()

        # 2 -----------------------------------------------------------------
        print("2. Links on page 1, sorted by permission")
        first_page = fetch_text(
            start,
            session=session,
            robots=robots,
            sleeper=recorded_sleeps.append,
        )
        allowed, refused = follow_links_politely(first_page, start, robots=robots)
        for url in allowed:
            print(f"   allowed: {url.removeprefix(site.base_url)}")
        for url in refused:
            print(f"   REFUSED: {url.removeprefix(site.base_url)}")
        try:
            fetch_text(private, session=session, robots=robots, sleeper=recorded_sleeps.append)
        except DisallowedByRobots as error:
            print(f"   fetching it anyway raises DisallowedByRobots on {str(error).removeprefix(site.base_url)}")
        print()

        # 3 -----------------------------------------------------------------
        print("3. First run — cold cache")
        site.reset()
        cache = ResponseCache(cache_dir)
        items = scrape_catalogue(
            start,
            session=session,
            robots=robots,
            cache=cache,
            sleeper=recorded_sleeps.append,
        )
        print(f"   pages requested: {len(site.paths())} {site.paths()}")
        print(f"   cache misses:    {cache.misses}, hits: {cache.hits}")
        print(f"   sleeps asked for: {recorded_sleeps[-3:]} (seconds, from Crawl-delay)")
        print(f"   items found:     {len(items)}")
        print()
        print("   sku      price     name")
        for item in items:
            price = " (none)" if item.price is None else f"{item.price:>7.2f}"
            print(f"   {item.sku:<8} {price}   {item.name}")
        missing = [item.sku for item in items if item.price is None]
        print(f"   items with no price cell: {missing} — handled, not crashed")
        print()

        # 4 -----------------------------------------------------------------
        print("4. Second run — warm cache")
        site.reset()
        # Permission is re-checked every run. robots.txt is the one thing we
        # deliberately do NOT cache: the site owner may have changed their mind
        # since yesterday, and a cached permission is a stale permission.
        robots_again = RobotsPolicy.load(site.base_url, session=session)
        cache2 = ResponseCache(cache_dir)
        again = scrape_catalogue(
            start,
            session=session,
            robots=robots_again,
            cache=cache2,
            sleeper=recorded_sleeps.append,
        )
        page_requests = [path for path in site.paths() if path != "/robots.txt"]
        print(f"   requests this run:      {site.paths()}")
        print(f"   page requests this run: {len(page_requests)} {page_requests}")
        print(f"   cache hits: {cache2.hits}, misses: {cache2.misses}")
        print(f"   same items as the first run: {again == items}")
        print()

        # 5 -----------------------------------------------------------------
        print("5. Output")
        write_csv(items, csv_path)
        text = csv_path.read_text(encoding="utf-8")
        print(f"   wrote {csv_path} ({len(text.splitlines())} lines including the header)")
        for line in text.splitlines()[:4]:
            print(f"   | {line}")
        print("   ...")
        for line in text.splitlines()[-1:]:
            print(f"   | {line}")
        print()

        # 6 -----------------------------------------------------------------
        print("6. What the server saw, across every run above")
        print(f"   total requests: {len(site.all_paths())}")
        print(f"   requests for /private/internal-notes.html: {site.total_count('/private/internal-notes.html')}")
        for agent in sorted(site.all_user_agents()):
            print(f"   User-Agent seen: {agent}")

    shutil.rmtree(cache_dir, ignore_errors=True)
    return 0


if __name__ == "__main__":
    sys.exit(main(sys.argv))

"""A local fixture web server that counts every request it receives.

Nothing in this lab talks to the internet. Instead we serve the fake catalogue
in ``examples/fixtures/`` over real HTTP from 127.0.0.1, so the scraper you
write is a real scraper making real requests — it just cannot reach anyone
else's machine.

Two details make this useful rather than merely offline:

* The port is **ephemeral**. We bind port 0 and read back whatever the
  operating system assigned. Hard-coding 8000 collides with whatever the
  learner already has running, and a lab that fails because of a port clash
  teaches nothing.
* Every request is **recorded**. ``FixtureServer.requests`` is the server's own
  log, not the client's. That is what lets the test suite prove a negative:
  the path robots.txt disallows was never even asked for. A scraper that
  merely *says* it honours robots.txt cannot fake this.

Standard library only: ``http.server``, ``socketserver``, ``threading``.
"""

from __future__ import annotations

import threading
from contextlib import contextmanager
from dataclasses import dataclass, field
from http.server import SimpleHTTPRequestHandler, ThreadingHTTPServer
from pathlib import Path
from typing import Iterator

FIXTURES = Path(__file__).resolve().parent / "fixtures"


@dataclass
class RequestRecord:
    """One line of the server's access log."""

    method: str
    path: str
    user_agent: str


@dataclass
class FixtureServer:
    """A running server plus the log of everything it was asked for."""

    host: str
    port: int
    requests: list[RequestRecord] = field(default_factory=list)
    history: list[RequestRecord] = field(default_factory=list)

    @property
    def base_url(self) -> str:
        """The root URL of this run, e.g. ``http://127.0.0.1:54321``."""
        return f"http://{self.host}:{self.port}"

    def url_for(self, path: str) -> str:
        """Absolute URL for a site-root-relative path such as ``/robots.txt``."""
        return self.base_url + path

    def paths(self) -> list[str]:
        """Every path requested, in order."""
        return [record.path for record in self.requests]

    def count(self, path: str) -> int:
        """How many times ``path`` was requested. The ethics assertion uses this."""
        return self.paths().count(path)

    def user_agents(self) -> set[str]:
        """The distinct User-Agent strings seen. An honest client sends one."""
        return {record.user_agent for record in self.requests}

    def all_paths(self) -> list[str]:
        """Every path requested since the server started, across all resets."""
        return [record.path for record in self.history] + self.paths()

    def total_count(self, path: str) -> int:
        """Requests for ``path`` since the server started, ignoring resets."""
        return self.all_paths().count(path)

    def all_user_agents(self) -> set[str]:
        """Distinct User-Agent strings seen since the server started."""
        return {record.user_agent for record in self.history} | self.user_agents()

    def reset(self) -> None:
        """Start a fresh log so the next run can be measured on its own.

        The old records move to ``history`` rather than being destroyed, so a
        question like "was the disallowed path *ever* requested?" still has an
        answer at the end of the program.
        """
        self.history.extend(self.requests)
        self.requests.clear()


def _make_handler(record_to: list[RequestRecord], directory: Path):
    class RecordingHandler(SimpleHTTPRequestHandler):
        def __init__(self, *args, **kwargs):
            super().__init__(*args, directory=str(directory), **kwargs)

        def _record(self, method: str) -> None:
            record_to.append(
                RequestRecord(
                    method=method,
                    path=self.path,
                    user_agent=self.headers.get("User-Agent", ""),
                )
            )

        def do_GET(self) -> None:  # noqa: N802 - name fixed by http.server
            self._record("GET")
            super().do_GET()

        def do_HEAD(self) -> None:  # noqa: N802 - name fixed by http.server
            self._record("HEAD")
            super().do_HEAD()

        def log_message(self, fmt: str, *args) -> None:
            """Silence the default stderr access log; we keep our own."""

    return RecordingHandler


@contextmanager
def serve_fixtures(directory: Path | None = None) -> Iterator[FixtureServer]:
    """Serve ``directory`` on 127.0.0.1 and an ephemeral port for the block.

    Usage::

        with serve_fixtures() as site:
            html = requests.get(site.url_for("/catalogue/page-1.html")).text
            assert site.count("/private/internal-notes.html") == 0

    The server is shut down and joined on the way out, including when the body
    raises, so a failing test never leaves a stray listener behind.
    """
    directory = directory or FIXTURES
    records: list[RequestRecord] = []
    handler = _make_handler(records, directory)

    # Port 0 means "any free port"; server_address tells us which one we got.
    httpd = ThreadingHTTPServer(("127.0.0.1", 0), handler)
    host, port = httpd.server_address[0], httpd.server_address[1]

    # A short poll interval keeps shutdown() quick; the default of 0.5 seconds
    # would add half a second to the teardown of every test that starts a
    # server, which is the difference between a suite people run and one they
    # skip.
    thread = threading.Thread(
        target=httpd.serve_forever, kwargs={"poll_interval": 0.02}, name="fixture-server"
    )
    thread.daemon = True
    thread.start()
    try:
        yield FixtureServer(host=host, port=port, requests=records)
    finally:
        httpd.shutdown()
        httpd.server_close()
        thread.join(timeout=5)


if __name__ == "__main__":  # A quick manual check of the server itself.
    import urllib.request

    with serve_fixtures() as site:
        with urllib.request.urlopen(site.url_for("/robots.txt")) as response:
            first_line = response.read().decode().splitlines()[0]
        print("served from 127.0.0.1 on an ephemeral port")
        print("first line of robots.txt:", first_line)
        print("requests recorded:", site.paths())

<!doctype html>
<html lang="en">
<head>
  <meta charset="utf-8">
  <title>Harbour Chandlery — catalogue, page 1</title>
</head>
<body>
  <h1>Harbour Chandlery</h1>
  <p class="note">Page 1 of 3. Prices in GBP.</p>

  <table class="catalogue">
    <thead>
      <tr><th>Item</th><th>Price</th><th>In stock</th></tr>
    </thead>
    <tbody>
      <tr class="item" data-sku="NAV-001">
        <td class="name">Brass Sextant</td>
        <td class="price">42.00</td>
        <td class="stock">7</td>
      </tr>
      <tr class="item" data-sku="NAV-002">
        <td class="name featured">Mariner Astrolabe</td>
        <td class="price">128.50</td>
        <td class="stock">2</td>
      </tr>
      <tr class="item" data-sku="STA-001">
        <td class="name">Ink &amp; Quill Set</td>
        <td class="price">12.75</td>
        <td class="stock">31</td>
      </tr>
      <tr class="item" data-sku="STA-002">
        <td class="name">Vellum Notebook <span class="badge">new</span></td>
        <td class="price">9.99</td>
        <td class="stock">18</td>
      </tr>
    </tbody>
  </table>

  <nav class="pager">
    <span class="prev disabled">Previous</span>
    <a class="next" href="page-2.html">Next</a>
  </nav>

  <footer>
    <p>Staff only: <a class="secret" href="/private/internal-notes.html">supplier notes</a></p>
  </footer>
</body>
</html>

<!doctype html>
<html lang="en">
<head>
  <meta charset="utf-8">
  <title>Harbour Chandlery — catalogue, page 2</title>
</head>
<body>
  <h1>Harbour Chandlery</h1>
  <p class="note">Page 2 of 3. Prices in GBP.</p>

  <table class="catalogue">
    <thead>
      <tr><th>Item</th><th>Price</th><th>In stock</th></tr>
    </thead>
    <tbody>
      <tr class="item" data-sku="NAV-003">
        <td class="name">Brass Compass</td>
        <td class="stock">4</td>
      </tr>
      <tr class="item" data-sku="STA-003">
        <td class="name">Sealing Wax Sticks</td>
        <td class="price">4.20</td>
        <td class="stock">64</td>
      </tr>
      <tr class="item" data-sku="TOO-001">
        <td class="name">Bookbinder's Awl</td>
        <td class="price">6.50</td>
        <td class="stock">12</td>
      </tr>
      <tr class="item" data-sku="TOO-002">
        <td class="name">Bone Folder</td>
        <td class="price">5.25</td>
        <td class="stock">23</td>
      </tr>
    </tbody>
  </table>

  <nav class="pager">
    <a class="prev" href="page-1.html">Previous</a>
    <a class="next" href="page-3.html">Next</a>
  </nav>
</body>
</html>

<!doctype html>
<html lang="en">
<head>
  <meta charset="utf-8">
  <title>Harbour Chandlery — catalogue, page 3</title>
</head>
<body>
  <h1>Harbour Chandlery</h1>
  <p class="note">Page 3 of 3. Prices in GBP.</p>

  <table class="catalogue">
    <thead>
      <tr><th>Item</th><th>Price</th><th>In stock</th></tr>
    </thead>
    <tbody>
      <tr class="item" data-sku="TOO-003">
        <td class="name">
          Linen Thread Spool
        </td>
        <td class="price">3.80</td>
        <td class="stock">45</td>
      </tr>
      <tr class="item" data-sku="MAP-001">
        <td class="name">Coastal Chart Portfolio</td>
        <td class="price">88.00</td>
        <td class="stock">3</td>
      </tr>
      <tr class="item" data-sku="MAP-002">
        <td class="name">Star Atlas</td>
        <td class="price">64.40</td>
        <td class="stock">5</td>
      </tr>
      <tr class="item" data-sku="MAP-003">
        <td class="name">Harbour Plan Set</td>
        <td class="price">27.15</td>
        <td class="stock">9</td>
      </tr>
    </tbody>
  </table>

  <nav class="pager">
    <a class="prev" href="page-2.html">Previous</a>
    <span class="next disabled">Next</span>
  </nav>
</body>
</html>

<!doctype html>
<html lang="en">
<head>
  <meta charset="utf-8">
  <title>Harbour Chandlery — sale, page 1</title>
</head>
<body>
  <h1>Harbour Chandlery — clearance</h1>
  <p class="note">
    The "Next" link below points at a completely different host. Real
    pagination links do this by accident all the time — a template variable
    picks up the wrong base, a mirror is linked from a footer, a redirect
    lands somewhere else. A crawler that follows it has quietly started
    scraping a site it never checked robots.txt for.
  </p>

  <table class="catalogue">
    <tbody>
      <tr class="item" data-sku="CLR-001">
        <td class="name">Frayed Mooring Line</td>
        <td class="price">1.00</td>
        <td class="stock">1</td>
      </tr>
    </tbody>
  </table>

  <nav class="pager">
    <a class="next" href="http://198.51.100.7/catalogue/page-2.html">Next</a>
  </nav>
</body>
</html>

<!doctype html>
<html lang="en">
<head>
  <meta charset="utf-8">
  <title>Harbour Chandlery</title>
</head>
<body>
  <h1>Harbour Chandlery</h1>
  <p>A fixture site with no real existence, served from 127.0.0.1 for this lab.</p>
  <ul>
    <li><a href="/catalogue/page-1.html">Catalogue</a></li>
    <li><a href="/robots.txt">robots.txt</a></li>
  </ul>
</body>
</html>

<!doctype html>
<html lang="en">
<head>
  <meta charset="utf-8">
  <title>Harbour Chandlery — supplier notes</title>
</head>
<body>
  <h1>Supplier notes</h1>
  <p>
    This page sits under /private/, which robots.txt disallows for every
    User-Agent. The fixture server will serve it to anyone who asks — refusing
    to ask is entirely your scraper's job, which is the point of the exercise.
    If a request for this path ever reaches the server during the test run,
    the harness fails.
  </p>
  <table class="catalogue">
    <tbody>
      <tr class="item" data-sku="XXX-999">
        <td class="name">Wholesale margin sheet</td>
        <td class="price">0.00</td>
        <td class="stock">0</td>
      </tr>
    </tbody>
  </table>
</body>
</html>

# robots.txt for the Harbour Chandlery fixture site.
#
# This file is served from the root of the local test server, exactly where a
# real site serves it. Everything below is standard Robots Exclusion Protocol
# (RFC 9309) and is parsed by urllib.robotparser without modification.

User-agent: *
Disallow: /private/
Crawl-delay: 1

# A named rule, so you can see that robotparser matches on the User-Agent
# string you pass to can_fetch. This one is refused everything.
User-agent: GreedyBot
Disallow: /

Sitemap: /sitemap.txt

/catalogue/page-1.html
/catalogue/page-2.html
/catalogue/page-3.html

"""Why a parser beats a regular expression, demonstrated on valid HTML.

Run it::

    python3 examples/regex_vs_parser.py

No server and no network: this reads the three fixture pages straight off the
disk. The regular expression below is not a straw man. It is exactly what a
careful person writes on their first attempt, it is anchored on the real
markup, and it works perfectly on the first row of the first page — which is
the trap. Every failure it produces here comes from HTML that is completely
valid and that a browser renders without complaint.
"""

from __future__ import annotations

import re
from pathlib import Path

from catalogue_scraper import parse_items

FIXTURES = Path(__file__).resolve().parent / "fixtures" / "catalogue"
PAGES = ["page-1.html", "page-2.html", "page-3.html"]

# The obvious regular expression: find a name cell, capture what is inside it.
NAME_RE = re.compile(r'<td class="name">([^<]*)</td>')


def regex_names(html: str) -> list[str]:
    return [match.group(1) for match in NAME_RE.finditer(html)]


def main() -> None:
    regex_found: list[str] = []
    parser_found: list[str] = []

    for page in PAGES:
        html = (FIXTURES / page).read_text(encoding="utf-8")
        regex_found.extend(regex_names(html))
        parser_found.extend(item.name for item in parse_items(html, source_path=page))

    print("regular expression found:", len(regex_found), "names")
    for name in regex_found:
        print(f"  {name!r}")

    print()
    print("BeautifulSoup found:", len(parser_found), "names")
    for name in parser_found:
        print(f"  {name!r}")

    missed = [name for name in parser_found if name not in regex_found]
    print()
    print("missed by the regular expression:", len(missed))
    for name in missed:
        print(f"  {name!r}")

    print()
    print("Three different failures, all on valid HTML:")
    print("  1. class=\"name featured\" — a second class defeats the literal match")
    print("  2. a nested <span> inside the cell — [^<]* stops at the '<'")
    print("  3. &amp; is returned raw; the parser decodes it to '&'")


if __name__ == "__main__":
    main()

lesson_id: D079
day: 79
kind: python-program
languages: [python]
setup_commands:
  - cd labs/sections/programming-with-python/day-079-web-scraping-responsibly
  - python3 -m venv .venv
  - .venv/bin/pip install -r requirements/requirements.txt
run_commands:
  - cat examples/fixtures/robots.txt
  - PYTHONPATH=examples .venv/bin/python3 examples/fixture_server.py
  - PYTHONPATH=examples .venv/bin/python3 examples/regex_vs_parser.py
  - PYTHONPATH=examples .venv/bin/python3 examples/demo.py catalogue.csv
  - head -4 catalogue.csv
  - SCRAPER_MODULE=starter .venv/bin/pytest tests -q
test_commands:
  - bash tests/run_tests.sh
cleanup_commands:
  - rm -f catalogue.csv
  - rm -rf .venv
  - 'git checkout -- starter/  # optional: reset your work'
requires_network: true
requires_api_key: false
estimated_minutes: 30
last_executed: '2026-07-19'
executed_on: 'macOS 26.5.1 (Apple Silicon), Python 3.14.0, beautifulsoup4 4.15.0, requests 2.34.2, pytest 9.1.1, bash 3.2.57 — bash tests/run_tests.sh -> 51 checks, 0 failure(s), exit 0; the pytest suite inside it reports 34 passed against examples/ and 33 failed, 1 passed against the unfinished starter. No test contacts any host other than 127.0.0.1.'

# Dependencies — Day 079 lab

Three packages, all free and open source, all pinned to the exact version this
lab was executed against.

| Pin | Why this lab needs it | Licence | Cost |
| --- | --- | --- | --- |
| `beautifulsoup4==4.15.0` | the HTML parser. Imported as `bs4`, not `beautifulsoup4` — the distribution name and the import name differ, which trips up nearly everyone once. | MIT | free |
| `requests==2.34.2` | the HTTP client from Day 78, used here with a `Session`, a timeout and an explicit `User-Agent` header. | Apache 2.0 | free |
| `pytest==9.1.1` | the test runner from Week 11. | MIT | free |

Everything else is standard library and needs no install:

| Module | Used for |
| --- | --- |
| `urllib.robotparser` | reading and applying `robots.txt` — the whole ethics half of this lab runs on a module that ships with Python |
| `urllib.parse` | `urljoin` for relative links, `urlparse` for the host guard |
| `http.server`, `socketserver`, `threading` | the local fixture server that stands in for a real site |
| `hashlib`, `json`, `pathlib` | the on-disk response cache |
| `csv` | the output, per Day 65 |
| `dataclasses`, `typing` | the `Item` type, per Day 69 |

Note what is **not** here. There is no `lxml`, no `selectolax`, no `scrapy`
and no `playwright`. The lesson's Alternatives section covers all four
honestly, and none of them is installed on this machine, so the lesson
describes them without quoting output it did not produce. `beautifulsoup4`
with the standard library's `html.parser` backend needs no compiler, works
everywhere Python works, and is the right default for a job this size.

## Installation

```bash
python3 -m venv .venv
.venv/bin/pip install -r requirements/requirements.txt
```

`.venv/` is already ignored by the repository. The test runner finds
`.venv/bin/pytest` on its own; you never need to activate anything.

## The one thing that needs the network

Installing those three packages downloads them from the Python Package Index,
which needs an internet connection **once**. After that, **the tests need no
network at all** — that is the entire design of this lab. The fixture server
runs on 127.0.0.1, on a port the operating system assigns, serving HTML files
that ship in `examples/fixtures/`. If you are on a plane, and the packages are
already installed, everything here still passes.

## Verify your setup

```bash
.venv/bin/python3 -c "import bs4, requests; print(bs4.__version__, requests.__version__)"
```

That should print `4.15.0 2.34.2`. If `import bs4` fails with
`ModuleNotFoundError` after a successful `pip install beautifulsoup4`, you
have almost certainly installed into a different interpreter than the one you
are running — see `troubleshooting.md`.

## Windows

Run everything inside WSL, where the commands above work unchanged. Outside
WSL the paths become `.venv\Scripts\pip` and `.venv\Scripts\python`, and
`bash tests/run_tests.sh` needs Git Bash or WSL because the harness is a bash
script.

beautifulsoup4==4.15.0
requests==2.34.2
pytest==9.1.1

"""Day 079 starter — a polite, cached, parser-based scraper.

Six numbered exercises. Everything else in this file is finished and working,
including ``RobotsPolicy.load``, which is left complete on purpose so you have
a worked model of the style the rest of the file expects.

Work in order. After each exercise, run the suite against YOUR module:

    SCRAPER_MODULE=starter .venv/bin/pytest tests -q

The number of failures should fall each time. When it reaches zero, run the
whole harness:

    bash tests/run_tests.sh

The reference answer is in ``examples/catalogue_scraper.py``. Read it after
you have made each exercise pass, not before — the exercises are short, and
the value is entirely in the attempt.

Every boundary here is a parameter, not a global: the ``session``, the
``cache``, the ``sleeper``. That is Day 74's rule. Keep it that way, or the
tests will have to wait real seconds and you will stop running them.
"""

from __future__ import annotations

import csv
import hashlib
import json
import time
import urllib.robotparser
from dataclasses import dataclass
from pathlib import Path
from typing import Callable, Iterable, Sequence
from urllib.parse import urljoin, urlparse

import requests
from bs4 import BeautifulSoup

# Identify yourself honestly, and leave a way to be contacted. `example.com`
# is reserved by the IETF for documentation, so this address belongs to
# nobody; put a real one in a real scraper.
USER_AGENT = "HarbourCatalogueLab/1.0 (course exercise; contact: scraper-owner@example.com)"

DEFAULT_TIMEOUT = 10.0
FALLBACK_DELAY_SECONDS = 1.0


class ScrapeError(Exception):
    """Base class for everything this module refuses to do."""


class DisallowedByRobots(ScrapeError):
    """The path is disallowed by robots.txt for our User-Agent."""


class OffSite(ScrapeError):
    """The URL points somewhere other than the site we were asked to scrape."""


# ===========================================================================
# EXERCISE 1 — permission, before anything else
# ===========================================================================
#
# `RobotsPolicy.load` below is written for you. Implement the two questions
# the rest of the program asks of it.
#
# `allows`: return True when `self.parser.can_fetch(self.user_agent, url)` says
#   so. One line. Note that can_fetch takes the User-Agent string, because
#   robots.txt can carry different rules for different clients — the fixture
#   robots.txt refuses "GreedyBot" everything, and a test checks that.
#
# `crawl_delay_seconds`: ask `self.parser.crawl_delay(self.user_agent)`. It
#   returns None when the site declares no delay; in that case return
#   `default`. Return a float either way.
#
# Tests: test_robots_allows_the_catalogue_and_refuses_the_private_path,
#        test_a_named_user_agent_gets_its_own_rules,
#        test_the_declared_crawl_delay_is_read_from_robots_txt,
#        test_a_missing_crawl_delay_falls_back_to_a_polite_default
# ===========================================================================


@dataclass
class RobotsPolicy:
    """A parsed robots.txt, and the two questions worth asking it."""

    parser: urllib.robotparser.RobotFileParser
    user_agent: str

    @classmethod
    def load(
        cls,
        base_url: str,
        *,
        session: requests.Session,
        user_agent: str = USER_AGENT,
        timeout: float = DEFAULT_TIMEOUT,
    ) -> "RobotsPolicy":
        """WORKED EXAMPLE — read this, then write the two methods below."""
        robots_url = urljoin(base_url, "/robots.txt")
        parser = urllib.robotparser.RobotFileParser()
        parser.set_url(robots_url)
        response = session.get(
            robots_url, headers={"User-Agent": user_agent}, timeout=timeout
        )
        if response.status_code == 200:
            parser.parse(response.text.splitlines())
        elif response.status_code in (401, 403):
            parser.disallow_all = True
        else:
            parser.allow_all = True
        return cls(parser=parser, user_agent=user_agent)

    def allows(self, url: str) -> bool:
        """Is this exact URL allowed for our User-Agent?"""
        raise NotImplementedError(
            "Exercise 1a: return self.parser.can_fetch(self.user_agent, url)"
        )

    def crawl_delay_seconds(self, default: float = FALLBACK_DELAY_SECONDS) -> float:
        """Seconds to wait between requests, per robots.txt, else ``default``."""
        raise NotImplementedError(
            "Exercise 1b: ask self.parser.crawl_delay(self.user_agent); "
            "return float(it) unless it is None, in which case return default"
        )


# ===========================================================================
# EXERCISE 5 — the response cache
# ===========================================================================
#
# `get`: build the path with `self._path_for(url)`. If it exists, increment
#   `self.hits` and return its text (encoding="utf-8"). Otherwise increment
#   `self.misses` and return None. Returning None rather than raising is what
#   lets `fetch_text` treat "not cached" as an ordinary case.
#
# `put`: write the text to `self._path_for(url)`, then add the filename-to-URL
#   pair to `index.json` in the same directory so the cache is readable by a
#   human. Load the existing index if the file is there.
#
# Tests: test_a_second_run_makes_no_page_requests,
#        test_a_cache_hit_does_not_pay_the_crawl_delay,
#        test_the_cache_directory_is_inspectable
# ===========================================================================


@dataclass
class ResponseCache:
    """Response bodies stored on disk, keyed by URL."""

    directory: Path
    hits: int = 0
    misses: int = 0

    def __post_init__(self) -> None:
        self.directory = Path(self.directory)
        self.directory.mkdir(parents=True, exist_ok=True)

    def _path_for(self, url: str) -> Path:
        """WORKED — a stable filename per URL, safe on every filesystem."""
        digest = hashlib.sha256(url.encode("utf-8")).hexdigest()[:32]
        return self.directory / f"{digest}.html"

    def get(self, url: str) -> str | None:
        raise NotImplementedError(
            "Exercise 5a: return the cached text and count a hit, or None and "
            "count a miss"
        )

    def put(self, url: str, text: str) -> None:
        raise NotImplementedError(
            "Exercise 5b: write the body to self._path_for(url), then record "
            "{filename: url} in index.json alongside it"
        )


# ===========================================================================
# fetch_text — WORKED. Read the ORDER of the four steps; it is the lesson.
# ===========================================================================


def fetch_text(
    url: str,
    *,
    session: requests.Session,
    robots: RobotsPolicy,
    cache: ResponseCache | None = None,
    sleeper: Callable[[float], None] = time.sleep,
    delay_seconds: float | None = None,
    user_agent: str = USER_AGENT,
    timeout: float = DEFAULT_TIMEOUT,
) -> str:
    """Return the body of ``url``, from cache if possible, politely if not."""
    # 1. Permission, before a socket is opened.
    if not robots.allows(url):
        raise DisallowedByRobots(url)

    # 2. Cache, so a page is fetched once in its life.
    if cache is not None:
        cached = cache.get(url)
        if cached is not None:
            return cached

    # 3. The crawl delay, paid only by a real fetch.
    wait = robots.crawl_delay_seconds() if delay_seconds is None else delay_seconds
    if wait > 0:
        sleeper(wait)

    # 4. The request itself, with a timeout (Day 78) and an honest User-Agent.
    response = session.get(url, headers={"User-Agent": user_agent}, timeout=timeout)
    response.raise_for_status()
    text = response.text
    if cache is not None:
        cache.put(url, text)
    return text


@dataclass(frozen=True)
class Item:
    """One catalogue row. ``price`` is ``None`` when the page omits the cell."""

    sku: str
    name: str
    price: float | None
    stock: int | None
    source_path: str


# ===========================================================================
# EXERCISE 3 — survive the missing element
# ===========================================================================
#
# This is the four-line function that decides whether your scraper crashes at
# 3 a.m. on row 4,812. Write it before Exercise 2 uses it.
#
#   cell = row.select_one(selector)      # returns None, never raises
#   if cell is None: return None         # the caller decides what that means
#   for extra in cell.select(drop): extra.extract()   # only when drop is given
#   return cell.get_text(" ", strip=True)
#
# Two things `get_text(" ", strip=True)` does for you: it strips the
# whitespace HTML authors leave around text, and it joins nested text with a
# space instead of gluing words together. `drop` exists because get_text
# sweeps up EVERY descendant, including a decorative <span class="badge">.
#
# Tests: test_the_item_with_no_price_cell_parses_instead_of_crashing,
#        test_a_row_with_no_cells_at_all_still_parses,
#        test_whitespace_wrapped_text_is_stripped,
#        test_a_decorative_nested_tag_is_kept_out_of_the_name
# ===========================================================================


def _cell_text(row, selector: str, *, drop: str | None = None) -> str | None:
    """Text of the first cell matching ``selector``, or ``None`` if absent."""
    raise NotImplementedError(
        "Exercise 3: select_one, check for None, optionally extract() the "
        "elements matching `drop`, then get_text(' ', strip=True)"
    )


# ===========================================================================
# EXERCISE 2 — extract the table with CSS selectors
# ===========================================================================
#
#   soup = BeautifulSoup(html, "html.parser")
#   for row in soup.select("table.catalogue tr.item"):
#       sku        = row.get("data-sku", "")
#       name       = _cell_text(row, "td.name", drop="span.badge") or ""
#       price_text = _cell_text(row, "td.price")
#       stock_text = _cell_text(row, "td.stock")
#       ... build an Item, converting price to float and stock to int, but
#           leaving each as None when its text is missing or empty ...
#
# Why `tr.item` and `td.name` rather than a regular expression: a CSS class
# selector matches a class that is PRESENT in the attribute, so it still finds
# `class="name featured"`. The obvious regex for `class="name"` does not, and
# it will not tell you that it missed anything.
#
# Tests: test_the_first_page_yields_four_items_in_document_order,
#        test_a_second_class_on_the_cell_does_not_hide_it,
#        test_an_html_entity_is_decoded_not_returned_raw,
#        test_prices_and_stock_are_numbers_not_strings
# ===========================================================================


def parse_items(html: str, *, source_path: str = "") -> list[Item]:
    """Extract every catalogue row from one page."""
    raise NotImplementedError(
        "Exercise 2: BeautifulSoup(html, 'html.parser').select("
        "'table.catalogue tr.item') and build one Item per row"
    )


# ===========================================================================
# EXERCISE 4 — pagination, to the end and no further
# ===========================================================================
#
# `next_page_url`: find `nav.pager a.next` with select_one. If there is no
#   such link — the last page has a <span class="next disabled"> instead —
#   return None. Otherwise return `urljoin(current_url, link.get("href"))`, so
#   the relative "page-2.html" becomes an absolute URL.
#
# `scrape_catalogue`: loop from `start_url`. On each turn, refuse a URL whose
#   netloc differs from the start URL's (raise OffSite), stop if you have seen
#   this URL before or have hit `max_pages`, fetch it with `fetch_text`,
#   extend the item list with `parse_items(html, source_path=urlparse(url).path)`,
#   and move on to `next_page_url(html, url)`.
#
# Do NOT stop after a fixed number of pages or when the item count reaches
# twelve. Pages get added. Stop when the site says there is no next page.
#
# Tests: test_next_page_url_is_resolved_against_the_current_url,
#        test_the_last_page_reports_no_next_page,
#        test_pagination_visits_every_page_exactly_once,
#        test_the_crawl_finds_every_item_across_all_pages,
#        test_an_off_site_link_is_refused_rather_than_followed
# ===========================================================================


def next_page_url(html: str, current_url: str) -> str | None:
    """Absolute URL of the next page, or ``None`` when this was the last one."""
    raise NotImplementedError(
        "Exercise 4a: select_one('nav.pager a.next'); None when absent; "
        "otherwise urljoin(current_url, href)"
    )


def scrape_catalogue(
    start_url: str,
    *,
    session: requests.Session,
    robots: RobotsPolicy,
    cache: ResponseCache | None = None,
    sleeper: Callable[[float], None] = time.sleep,
    delay_seconds: float | None = None,
    max_pages: int = 50,
) -> list[Item]:
    """Follow the pagination from ``start_url`` and return every item found."""
    raise NotImplementedError(
        "Exercise 4b: the while loop — guard the host, guard against repeats "
        "and max_pages, fetch, parse, then follow next_page_url"
    )


def follow_links_politely(
    html: str,
    base_url: str,
    *,
    robots: RobotsPolicy,
) -> tuple[list[str], list[str]]:
    """WORKED — split every link on a page into (allowed, refused)."""
    soup = BeautifulSoup(html, "html.parser")
    allowed: list[str] = []
    refused: list[str] = []
    for anchor in soup.select("a[href]"):
        target = urljoin(base_url, anchor["href"])
        (allowed if robots.allows(target) else refused).append(target)
    return allowed, refused


# ===========================================================================
# EXERCISE 6 — write the CSV (Day 65)
# ===========================================================================
#
#   open the path with encoding="utf-8" and newline="" — the csv module
#   requires newline="", and forgetting it gives you blank lines on Windows
#   write CSV_COLUMNS as the header row
#   one row per item: sku, name, price formatted to two decimals, stock,
#   source_path — and an EMPTY STRING, not the word "None", where a value is
#   missing. A CSV containing the string None is a CSV that will be loaded as
#   text by whatever reads it next.
#
# Tests: test_the_csv_has_a_header_and_one_row_per_item,
#        test_the_missing_price_becomes_an_empty_field_not_the_word_none,
#        test_the_ampersand_survives_the_round_trip
# ===========================================================================

CSV_COLUMNS: Sequence[str] = ("sku", "name", "price", "stock", "source_path")


def write_csv(items: Iterable[Item], path: Path) -> Path:
    """Write items to ``path`` as CSV and return the path."""
    raise NotImplementedError(
        "Exercise 6: csv.writer into an open(..., newline='', encoding='utf-8'), "
        "header first, empty string for a missing price or stock"
    )

#!/usr/bin/env bash
# Tests for the Day 079 lab. Run from the lab directory:
#   bash tests/run_tests.sh
#
# NOTHING HERE TOUCHES THE INTERNET. Section 1 proves that rather than
# asserting it: the only absolute URLs in the lab's source point at 127.0.0.1
# or at 198.51.100.0/24, the range the IETF reserves for documentation, and
# the one reference to that range exists precisely so a test can watch the
# scraper REFUSE to connect to it. Everything else is served by a local
# fixture server on a port the operating system picks at run time.
#
# The check worth reading is in section 4. The fixture site links a page that
# robots.txt disallows, and the server will serve it to anyone who asks. The
# harness asserts, against the SERVER's own access log, that the path was
# never requested. A scraper that only describes good behaviour cannot pass
# that check — which is the difference between ethics implemented and ethics
# described.
#
# Non-interactive, deterministic, exits 0 only if every check passes.
set -u

export PYTHONDONTWRITEBYTECODE=1

lab_dir="$(cd "$(dirname "${BASH_SOURCE[0]}")/.." && pwd)"
failures=0
checks=0

check() {
  local label="$1" ok="$2"
  checks=$((checks + 1))
  if [ "${ok}" = "yes" ]; then
    echo "  ok: ${label}"
  else
    echo "  FAIL: ${label}"
    failures=$((failures + 1))
  fi
}

# Resolve tools: an explicit override, then this lab's .venv, then PATH.
# Fails loudly with instructions rather than silently skipping.
resolve_tool() {
  local tool="$1" override="$2"
  if [ -n "${override}" ] && [ -x "${override}" ]; then echo "${override}"; return 0; fi
  if [ -x "${lab_dir}/.venv/bin/${tool}" ]; then echo "${lab_dir}/.venv/bin/${tool}"; return 0; fi
  if command -v "${tool}" >/dev/null 2>&1; then command -v "${tool}"; return 0; fi
  return 1
}

pytest_bin="$(resolve_tool pytest "${PYTEST:-}")" || {
  echo "FAIL: pytest not found." >&2
  echo "  Install this lab's dependencies with:" >&2
  echo "    python3 -m venv .venv" >&2
  echo "    .venv/bin/pip install -r requirements/requirements.txt" >&2
  echo "  Or point this suite at an existing pytest: PYTEST=/path/to/pytest bash tests/run_tests.sh" >&2
  exit 1
}

python_bin="$(resolve_tool python3 "${PYTHON:-}")" || {
  echo "FAIL: python3 not found on PATH." >&2
  exit 1
}
if [ -x "${lab_dir}/.venv/bin/python3" ]; then
  python_bin="${lab_dir}/.venv/bin/python3"
fi

echo "Day 079 — Scrape a Site You Are Allowed To"
echo

# --------------------------------------------------------------------------
echo "1. Offline by construction"
# --------------------------------------------------------------------------

# Any absolute http(s) URL beginning with a letter would be a hostname, which
# would mean a DNS lookup. There must be none in the lab's own source.
if grep -rEn 'https?://[A-Za-z]' \
     "${lab_dir}/examples" "${lab_dir}/starter" "${lab_dir}/tests" >/dev/null 2>&1; then
  check "no hostname-bearing URL anywhere in examples/, starter/ or tests/" "no"
  grep -rEn 'https?://[A-Za-z]' "${lab_dir}/examples" "${lab_dir}/starter" "${lab_dir}/tests"
else
  check "no hostname-bearing URL anywhere in examples/, starter/ or tests/" "yes"
fi

# The one absolute URL with a foreign host is the documentation-reserved
# address in the detour fixture, and it exists to be refused.
detour_hits="$(grep -Ec '198\.51\.100\.' "${lab_dir}/examples/fixtures/detour/page-1.html" || true)"
if [ "${detour_hits}" = "1" ]; then
  check "the only off-site link is a documentation-reserved address (198.51.100.0/24)" "yes"
else
  check "the only off-site link is a documentation-reserved address (got ${detour_hits} occurrences)" "no"
fi

# The server binds port 0, not a fixed port, so it never collides.
if grep -q '("127.0.0.1", 0)' "${lab_dir}/examples/fixture_server.py"; then
  check "the fixture server binds port 0 and reads back the assigned port" "yes"
else
  check "the fixture server binds port 0 and reads back the assigned port" "no"
fi

# Two servers started in a row must land on different ports — that is what
# "ephemeral" means, and it is why the lab cannot clash with your own work.
ports="$(cd "${lab_dir}" && PYTHONPATH=examples "${python_bin}" - <<'PY'
from fixture_server import serve_fixtures
with serve_fixtures() as a, serve_fixtures() as b:
    print(a.port, b.port, a.port != b.port)
PY
)"
case "${ports}" in
  *True) check "two fixture servers get two different ephemeral ports ( ${ports% True} )" "yes" ;;
  *) check "two fixture servers get two different ephemeral ports (got '${ports}')" "no" ;;
esac

# --------------------------------------------------------------------------
echo
echo "2. The tools"
# --------------------------------------------------------------------------

version_line="$("${pytest_bin}" --version 2>&1 | head -1)"
case "${version_line}" in
  pytest*) check "pytest --version reports a pytest ( ${version_line} )" "yes" ;;
  *) check "pytest --version reports a pytest ( ${version_line} )" "no" ;;
esac

bs4_version="$("${python_bin}" -c 'import bs4; print(bs4.__version__)' 2>&1)"
case "${bs4_version}" in
  4.*) check "beautifulsoup4 imports as bs4 ( ${bs4_version} )" "yes" ;;
  *) check "beautifulsoup4 imports as bs4 (got: ${bs4_version})" "no" ;;
esac

requests_version="$("${python_bin}" -c 'import requests; print(requests.__version__)' 2>&1)"
case "${requests_version}" in
  2.*) check "requests is importable ( ${requests_version} )" "yes" ;;
  *) check "requests is importable (got: ${requests_version})" "no" ;;
esac

# --------------------------------------------------------------------------
echo
echo "3. The reference suite"
# --------------------------------------------------------------------------

suite_out="$(cd "${lab_dir}" && "${pytest_bin}" tests -q 2>&1)"
suite_exit=$?
if [ "${suite_exit}" -eq 0 ]; then
  check "pytest tests exits 0 against examples/" "yes"
else
  check "pytest tests exits 0 against examples/ (got ${suite_exit})" "no"
  echo "${suite_out}" | tail -30
fi

case "${suite_out}" in
  *"34 passed"*) check "the reference suite reports 34 passed" "yes" ;;
  *) check "the reference suite reports 34 passed (got: $(printf '%s' "${suite_out}" | tail -1))" "no" ;;
esac

# --------------------------------------------------------------------------
echo
echo "4. Ethics implemented, not described"
# --------------------------------------------------------------------------

# The private page IS linked from page 1 and the server WILL serve it. The
# only reason it is never fetched is that the scraper asks robots.txt first.
if grep -q '/private/internal-notes.html' "${lab_dir}/examples/fixtures/catalogue/page-1.html"; then
  check "the fixture really does link the disallowed page from page 1" "yes"
else
  check "the fixture really does link the disallowed page from page 1" "no"
fi

if grep -q 'Disallow: /private/' "${lab_dir}/examples/fixtures/robots.txt"; then
  check "robots.txt really does disallow /private/ for every User-Agent" "yes"
else
  check "robots.txt really does disallow /private/ for every User-Agent" "no"
fi

# Independent evidence: ask the server, not the scraper.
private_report="$(cd "${lab_dir}" && PYTHONPATH=examples "${python_bin}" - <<'PY'
import tempfile
from pathlib import Path

import requests

from catalogue_scraper import ResponseCache, RobotsPolicy, scrape_catalogue
from fixture_server import serve_fixtures

with serve_fixtures() as site, requests.Session() as session:
    robots = RobotsPolicy.load(site.base_url, session=session)
    cache = ResponseCache(Path(tempfile.mkdtemp(prefix="d079-")))
    items = scrape_catalogue(
        site.url_for("/catalogue/page-1.html"),
        session=session,
        robots=robots,
        cache=cache,
        sleeper=lambda seconds: None,
    )
    print("private_requests", site.total_count("/private/internal-notes.html"))
    print("items", len(items))
    print("pages", len([p for p in site.paths() if p.startswith("/catalogue/")]))
    print("agents", len(site.all_user_agents()))
PY
)"
case "${private_report}" in
  *"private_requests 0"*) check "the disallowed path was requested ZERO times, per the server's own log" "yes" ;;
  *) check "the disallowed path was requested ZERO times (report: ${private_report})" "no" ;;
esac
case "${private_report}" in
  *"pages 3"*) check "pagination followed all 3 pages and stopped" "yes" ;;
  *) check "pagination followed all 3 pages and stopped" "no" ;;
esac
case "${private_report}" in
  *"items 12"*) check "the crawl found all 12 items" "yes" ;;
  *) check "the crawl found all 12 items" "no" ;;
esac
case "${private_report}" in
  *"agents 1"*) check "exactly one User-Agent string was sent — the client identified itself" "yes" ;;
  *) check "exactly one User-Agent string was sent" "no" ;;
esac

# The named-agent rules in robots.txt are honoured too.
greedy="$(cd "${lab_dir}" && PYTHONPATH=examples "${python_bin}" - <<'PY'
import requests
from catalogue_scraper import RobotsPolicy
from fixture_server import serve_fixtures

with serve_fixtures() as site, requests.Session() as session:
    polite = RobotsPolicy.load(site.base_url, session=session)
    greedy = RobotsPolicy.load(site.base_url, session=session, user_agent="GreedyBot")
    target = site.url_for("/catalogue/page-1.html")
    print("polite", polite.allows(target), "greedy", greedy.allows(target),
          "delay", polite.crawl_delay_seconds())
PY
)"
case "${greedy}" in
  "polite True greedy False delay 1.0") check "robots.txt rules are per-User-Agent ( ${greedy} )" "yes" ;;
  *) check "robots.txt rules are per-User-Agent (got: ${greedy})" "no" ;;
esac

# --------------------------------------------------------------------------
echo
echo "5. The messy realities"
# --------------------------------------------------------------------------

messy="$(cd "${lab_dir}" && PYTHONPATH=examples "${python_bin}" - <<'PY'
from pathlib import Path
from catalogue_scraper import parse_items

base = Path("examples/fixtures/catalogue")
one = parse_items(base.joinpath("page-1.html").read_text(encoding="utf-8"))
two = parse_items(base.joinpath("page-2.html").read_text(encoding="utf-8"))
three = parse_items(base.joinpath("page-3.html").read_text(encoding="utf-8"))
by_sku = {item.sku: item for item in one + two + three}
print("missing_price", by_sku["NAV-003"].price, "rows_on_page_2", len(two))
print("entity", repr(by_sku["STA-001"].name))
print("nested", repr(by_sku["STA-002"].name))
print("second_class", repr(by_sku["NAV-002"].name))
print("whitespace", repr(by_sku["TOO-003"].name))
PY
)"
case "${messy}" in
  *"missing_price None rows_on_page_2 4"*)
    check "the item with no price cell yields price=None and does not shorten the page" "yes" ;;
  *) check "the item with no price cell yields price=None (got: ${messy})" "no" ;;
esac
case "${messy}" in
  *"entity 'Ink & Quill Set'"*) check "the HTML entity &amp; is decoded to '&'" "yes" ;;
  *) check "the HTML entity &amp; is decoded to '&'" "no" ;;
esac
case "${messy}" in
  *"nested 'Vellum Notebook'"*) check "the decorative nested tag is kept out of the name" "yes" ;;
  *) check "the decorative nested tag is kept out of the name" "no" ;;
esac
case "${messy}" in
  *"second_class 'Mariner Astrolabe'"*) check "a cell with two classes is still found by td.name" "yes" ;;
  *) check "a cell with two classes is still found by td.name" "no" ;;
esac
case "${messy}" in
  *"whitespace 'Linen Thread Spool'"*) check "whitespace-wrapped text is stripped" "yes" ;;
  *) check "whitespace-wrapped text is stripped" "no" ;;
esac

# --------------------------------------------------------------------------
echo
echo "6. The regular expression fails where the parser does not"
# --------------------------------------------------------------------------

regex_out="$(cd "${lab_dir}" && PYTHONPATH=examples "${python_bin}" examples/regex_vs_parser.py 2>&1)"
regex_exit=$?
if [ "${regex_exit}" -eq 0 ]; then
  check "examples/regex_vs_parser.py exits 0" "yes"
else
  check "examples/regex_vs_parser.py exits 0 (got ${regex_exit})" "no"
fi
case "${regex_out}" in
  *"regular expression found: 10 names"*) check "the naive regex finds only 10 of the 12 names" "yes" ;;
  *) check "the naive regex finds only 10 of the 12 names" "no" ;;
esac
case "${regex_out}" in
  *"BeautifulSoup found: 12 names"*) check "the parser finds all 12" "yes" ;;
  *) check "the parser finds all 12" "no" ;;
esac
case "${regex_out}" in
  *"'Ink &amp; Quill Set'"*) check "the regex hands back an undecoded HTML entity" "yes" ;;
  *) check "the regex hands back an undecoded HTML entity" "no" ;;
esac

# --------------------------------------------------------------------------
echo
echo "7. The cache means a re-run costs the site nothing"
# --------------------------------------------------------------------------

cache_report="$(cd "${lab_dir}" && PYTHONPATH=examples "${python_bin}" - <<'PY'
import shutil, tempfile
from pathlib import Path

import requests

from catalogue_scraper import ResponseCache, RobotsPolicy, scrape_catalogue
from fixture_server import serve_fixtures

cache_dir = Path(tempfile.mkdtemp(prefix="d079-cache-"))
with serve_fixtures() as site, requests.Session() as session:
    robots = RobotsPolicy.load(site.base_url, session=session)
    start = site.url_for("/catalogue/page-1.html")
    first = scrape_catalogue(start, session=session, robots=robots,
                             cache=ResponseCache(cache_dir), sleeper=lambda s: None)
    site.reset()
    waits = []
    warm = ResponseCache(cache_dir)
    second = scrape_catalogue(start, session=session, robots=robots,
                              cache=warm, sleeper=waits.append)
    pages = [p for p in site.paths() if p != "/robots.txt"]
    print("second_run_page_requests", len(pages))
    print("hits", warm.hits, "misses", warm.misses)
    print("identical", first == second)
    print("sleeps_on_cache_hits", len(waits))
    print("cache_files", len(list(cache_dir.glob("*.html"))))
shutil.rmtree(cache_dir, ignore_errors=True)
PY
)"
case "${cache_report}" in
  *"second_run_page_requests 0"*) check "a second run makes ZERO page requests" "yes" ;;
  *) check "a second run makes ZERO page requests (report: ${cache_report})" "no" ;;
esac
case "${cache_report}" in
  *"hits 3 misses 0"*) check "all three pages came from the cache" "yes" ;;
  *) check "all three pages came from the cache" "no" ;;
esac
case "${cache_report}" in
  *"identical True"*) check "the cached run produces byte-identical items" "yes" ;;
  *) check "the cached run produces byte-identical items" "no" ;;
esac
case "${cache_report}" in
  *"sleeps_on_cache_hits 0"*) check "a cache hit does not pay the crawl delay" "yes" ;;
  *) check "a cache hit does not pay the crawl delay" "no" ;;
esac
case "${cache_report}" in
  *"cache_files 3"*) check "the cache holds one file per page fetched" "yes" ;;
  *) check "the cache holds one file per page fetched" "no" ;;
esac

# --------------------------------------------------------------------------
echo
echo "8. End to end: the demo and its CSV"
# --------------------------------------------------------------------------

work="$(mktemp -d "${TMPDIR:-/tmp}/d079-demo.XXXXXX")"
demo_out="$(cd "${lab_dir}" && PYTHONPATH=examples "${python_bin}" examples/demo.py "${work}/catalogue.csv" 2>&1)"
demo_exit=$?
if [ "${demo_exit}" -eq 0 ]; then
  check "examples/demo.py exits 0" "yes"
else
  check "examples/demo.py exits 0 (got ${demo_exit})" "no"
  echo "${demo_out}" | tail -20
fi
case "${demo_out}" in
  *"requests for /private/internal-notes.html: 0"*)
    check "the demo's closing server log shows zero requests for the disallowed path" "yes" ;;
  *) check "the demo's closing server log shows zero requests for the disallowed path" "no" ;;
esac
case "${demo_out}" in
  *"page requests this run: 0"*) check "the demo's second run makes no page requests" "yes" ;;
  *) check "the demo's second run makes no page requests" "no" ;;
esac

if [ -f "${work}/catalogue.csv" ]; then
  csv_lines="$(wc -l < "${work}/catalogue.csv" | tr -d ' ')"
  if [ "${csv_lines}" = "13" ]; then
    check "the CSV has 13 lines: a header and 12 items" "yes"
  else
    check "the CSV has 13 lines (got ${csv_lines})" "no"
  fi
  # The csv module's default dialect ends lines with CRLF, as RFC 4180 asks.
  # That is correct, and it is also why this comparison strips the carriage
  # return rather than pretending the byte is not there.
  header="$(head -1 "${work}/catalogue.csv" | tr -d '\r')"
  if [ "${header}" = "sku,name,price,stock,source_path" ]; then
    check "the CSV header names the five columns" "yes"
  else
    check "the CSV header names the five columns (got '${header}')" "no"
  fi
  if head -1 "${work}/catalogue.csv" | grep -q $'\r'; then
    check "the CSV uses RFC 4180 CRLF line endings, as the csv module defaults to" "yes"
  else
    check "the CSV uses RFC 4180 CRLF line endings" "no"
  fi
  if grep -q '^NAV-003,Brass Compass,,4,' "${work}/catalogue.csv"; then
    check "the price-less item writes an EMPTY field, not the word None" "yes"
  else
    check "the price-less item writes an EMPTY field, not the word None" "no"
  fi
  if grep -q 'Ink & Quill Set' "${work}/catalogue.csv"; then
    check "the ampersand survives HTML, parsing and CSV" "yes"
  else
    check "the ampersand survives HTML, parsing and CSV" "no"
  fi
  if grep -q 'None' "${work}/catalogue.csv"; then
    check "the CSV contains no literal 'None'" "no"
  else
    check "the CSV contains no literal 'None'" "yes"
  fi
else
  check "examples/demo.py wrote the CSV" "no"
fi
rm -rf "${work}"

# --------------------------------------------------------------------------
echo
echo "9. The starter's exercises are load-bearing"
# --------------------------------------------------------------------------

starter_out="$(cd "${lab_dir}" && SCRAPER_MODULE=starter "${pytest_bin}" tests -q 2>&1)"
starter_exit=$?
if [ "${starter_exit}" -ne 0 ]; then
  check "the same suite FAILS against the unfinished starter (exit ${starter_exit})" "yes"
else
  check "the same suite FAILS against the unfinished starter — it did not, so the exercises are vacuous" "no"
fi
case "${starter_out}" in
  *"NotImplementedError"*) check "the starter fails on its own NotImplementedError, not on an import error" "yes" ;;
  *) check "the starter fails on its own NotImplementedError, not on an import error" "no" ;;
esac
case "${starter_out}" in
  *"1 passed"*) check "exactly one test passes on the starter — the worked RobotsPolicy.load" "yes" ;;
  *) check "exactly one test passes on the starter ( $(printf '%s' "${starter_out}" | tail -1) )" "no" ;;
esac

for exercise in "Exercise 1a" "Exercise 1b" "Exercise 2" "Exercise 3" "Exercise 4a" "Exercise 4b" "Exercise 5a" "Exercise 5b" "Exercise 6"; do
  if grep -q "${exercise}" "${lab_dir}/starter/catalogue_scraper.py"; then
    check "the starter names ${exercise}" "yes"
  else
    check "the starter names ${exercise}" "no"
  fi
done

echo
echo "${checks} checks, ${failures} failure(s)."
[ "${failures}" -eq 0 ]

"""The reference suite for Day 079.

It runs against ``examples/catalogue_scraper.py`` by default and against
``starter/catalogue_scraper.py`` when ``SCRAPER_MODULE=starter`` is set. The
starter run is expected to FAIL until the six exercises are finished — that is
how ``tests/run_tests.sh`` proves the exercises are load-bearing rather than
decorative.

Every test in this file is offline. The only socket opened points at
127.0.0.1, at a port the operating system chose, serving files that ship with
this lab. Nothing resolves a hostname and nothing leaves the machine.

The check that matters most is ``test_the_disallowed_path_is_never_requested``.
It does not ask the scraper what it did; it asks the *server* what it was
asked for. A scraper that describes good behaviour in a docstring cannot pass
it.
"""

from __future__ import annotations

import importlib
import os
import re
import sys
from pathlib import Path

import pytest
import requests

LAB_DIR = Path(__file__).resolve().parent.parent
WHICH = os.environ.get("SCRAPER_MODULE", "examples")
sys.path.insert(0, str(LAB_DIR / "examples"))
sys.path.insert(0, str(LAB_DIR / WHICH))

scraper = importlib.import_module("catalogue_scraper")
from fixture_server import serve_fixtures  # noqa: E402

EXPECTED_ITEM_COUNT = 12
EXPECTED_PAGES = [
    "/catalogue/page-1.html",
    "/catalogue/page-2.html",
    "/catalogue/page-3.html",
]
PRIVATE_PATH = "/private/internal-notes.html"


class RecordingSleeper:
    """A stand-in for ``time.sleep`` that records instead of waiting.

    Day 74's boundary rule, applied to the clock: the scraper takes a
    ``sleeper`` parameter, so the suite can assert that it asked to wait one
    second between fetches without the suite taking three seconds to run.
    """

    def __init__(self) -> None:
        self.waits: list[float] = []

    def __call__(self, seconds: float) -> None:
        self.waits.append(seconds)


@pytest.fixture
def site():
    with serve_fixtures() as server:
        yield server


@pytest.fixture
def session():
    with requests.Session() as s:
        yield s


@pytest.fixture
def robots(site, session):
    return scraper.RobotsPolicy.load(site.base_url, session=session)


# ---------------------------------------------------------------------------
# Exercise 1 — robots.txt is consulted before anything is fetched
# ---------------------------------------------------------------------------


def test_robots_txt_is_fetched_with_our_own_user_agent(site, session):
    scraper.RobotsPolicy.load(site.base_url, session=session)
    assert site.paths() == ["/robots.txt"]
    assert site.user_agents() == {scraper.USER_AGENT}


def test_robots_allows_the_catalogue_and_refuses_the_private_path(site, robots):
    assert robots.allows(site.url_for("/catalogue/page-1.html")) is True
    assert robots.allows(site.url_for(PRIVATE_PATH)) is False


def test_a_named_user_agent_gets_its_own_rules(site, session):
    greedy = scraper.RobotsPolicy.load(
        site.base_url, session=session, user_agent="GreedyBot"
    )
    assert greedy.allows(site.url_for("/catalogue/page-1.html")) is False


def test_the_declared_crawl_delay_is_read_from_robots_txt(robots):
    assert robots.crawl_delay_seconds() == 1.0


def test_a_missing_crawl_delay_falls_back_to_a_polite_default(site, session, tmp_path):
    bare = tmp_path / "site"
    (bare / "catalogue").mkdir(parents=True)
    (bare / "robots.txt").write_text("User-agent: *\nDisallow: /nowhere/\n")
    with serve_fixtures(bare) as plain_site:
        policy = scraper.RobotsPolicy.load(plain_site.base_url, session=session)
        assert policy.crawl_delay_seconds() == pytest.approx(1.0)
        assert policy.crawl_delay_seconds(default=2.5) == pytest.approx(2.5)


def test_fetching_a_disallowed_url_raises_before_any_request(site, session, robots):
    site.reset()
    with pytest.raises(scraper.DisallowedByRobots):
        scraper.fetch_text(
            site.url_for(PRIVATE_PATH),
            session=session,
            robots=robots,
            sleeper=RecordingSleeper(),
        )
    assert site.paths() == []


def test_the_disallowed_path_is_never_requested(site, session, robots, tmp_path):
    """The check that proves the ethics were implemented, not described.

    The private page is linked from page 1, the server will happily serve it,
    and the crawl walks right past it. The assertion reads the SERVER's log.
    """
    site.reset()
    cache = scraper.ResponseCache(tmp_path / "cache")
    scraper.scrape_catalogue(
        site.url_for("/catalogue/page-1.html"),
        session=session,
        robots=robots,
        cache=cache,
        sleeper=RecordingSleeper(),
    )
    first_page = (LAB_DIR / "examples/fixtures/catalogue/page-1.html").read_text()
    assert PRIVATE_PATH in first_page, "the fixture must actually link the private page"
    assert site.count(PRIVATE_PATH) == 0
    assert site.total_count(PRIVATE_PATH) == 0


def test_links_are_sorted_into_allowed_and_refused(site, session, robots):
    start = site.url_for("/catalogue/page-1.html")
    html = scraper.fetch_text(
        start, session=session, robots=robots, sleeper=RecordingSleeper()
    )
    allowed, refused = scraper.follow_links_politely(html, start, robots=robots)
    assert refused == [site.url_for(PRIVATE_PATH)]
    assert site.url_for("/catalogue/page-2.html") in allowed


def test_the_site_publishes_a_sitemap_that_beats_guessing(site, session, robots):
    """The alternatives-first rule, in miniature.

    robots.txt names a sitemap. Reading it gives you every page directly,
    with no pagination logic to get wrong and no pages missed when the site
    adds a fourth one. Look for this before you write a crawl loop.
    """
    sitemap_url = site.url_for("/sitemap.txt")
    body = scraper.fetch_text(
        sitemap_url, session=session, robots=robots, sleeper=RecordingSleeper()
    )
    listed = [line.strip() for line in body.splitlines() if line.strip()]
    assert listed == EXPECTED_PAGES


# ---------------------------------------------------------------------------
# Exercise 2 — CSS selectors extract the table
# ---------------------------------------------------------------------------


def page_html(name: str) -> str:
    return (LAB_DIR / "examples/fixtures/catalogue" / name).read_text(encoding="utf-8")


def test_the_first_page_yields_four_items_in_document_order():
    items = scraper.parse_items(page_html("page-1.html"), source_path="/p1")
    assert [item.sku for item in items] == ["NAV-001", "NAV-002", "STA-001", "STA-002"]


def test_a_second_class_on_the_cell_does_not_hide_it():
    items = scraper.parse_items(page_html("page-1.html"))
    astrolabe = next(item for item in items if item.sku == "NAV-002")
    assert astrolabe.name == "Mariner Astrolabe"
    assert astrolabe.price == pytest.approx(128.50)


def test_an_html_entity_is_decoded_not_returned_raw():
    items = scraper.parse_items(page_html("page-1.html"))
    quills = next(item for item in items if item.sku == "STA-001")
    assert quills.name == "Ink & Quill Set"
    assert "&" not in quills.name


def test_a_decorative_nested_tag_is_kept_out_of_the_name():
    items = scraper.parse_items(page_html("page-1.html"))
    notebook = next(item for item in items if item.sku == "STA-002")
    assert notebook.name == "Vellum Notebook"


def test_whitespace_wrapped_text_is_stripped():
    items = scraper.parse_items(page_html("page-3.html"))
    thread = next(item for item in items if item.sku == "TOO-003")
    assert thread.name == "Linen Thread Spool"


def test_prices_and_stock_are_numbers_not_strings():
    items = scraper.parse_items(page_html("page-1.html"))
    sextant = items[0]
    assert isinstance(sextant.price, float)
    assert isinstance(sextant.stock, int)
    assert sextant.price == pytest.approx(42.00)
    assert sextant.stock == 7


# ---------------------------------------------------------------------------
# Exercise 3 — survive the missing element
# ---------------------------------------------------------------------------


def test_the_item_with_no_price_cell_parses_instead_of_crashing():
    items = scraper.parse_items(page_html("page-2.html"), source_path="/p2")
    compass = next(item for item in items if item.sku == "NAV-003")
    assert compass.name == "Brass Compass"
    assert compass.price is None
    assert compass.stock == 4


def test_the_missing_price_does_not_shorten_the_page():
    items = scraper.parse_items(page_html("page-2.html"))
    assert len(items) == 4


def test_a_row_with_no_cells_at_all_still_parses():
    html = '<table class="catalogue"><tr class="item" data-sku="X-1"></tr></table>'
    items = scraper.parse_items(html)
    assert len(items) == 1
    assert items[0].sku == "X-1"
    assert items[0].name == ""
    assert items[0].price is None


# ---------------------------------------------------------------------------
# Exercise 4 — pagination to the end
# ---------------------------------------------------------------------------


def test_next_page_url_is_resolved_against_the_current_url(site):
    current = site.url_for("/catalogue/page-1.html")
    assert scraper.next_page_url(page_html("page-1.html"), current) == site.url_for(
        "/catalogue/page-2.html"
    )


def test_the_last_page_reports_no_next_page(site):
    current = site.url_for("/catalogue/page-3.html")
    assert scraper.next_page_url(page_html("page-3.html"), current) is None


def test_pagination_visits_every_page_exactly_once(site, session, robots, tmp_path):
    site.reset()
    scraper.scrape_catalogue(
        site.url_for("/catalogue/page-1.html"),
        session=session,
        robots=robots,
        cache=scraper.ResponseCache(tmp_path / "cache"),
        sleeper=RecordingSleeper(),
    )
    assert site.paths() == EXPECTED_PAGES


def test_the_crawl_finds_every_item_across_all_pages(site, session, robots, tmp_path):
    items = scraper.scrape_catalogue(
        site.url_for("/catalogue/page-1.html"),
        session=session,
        robots=robots,
        cache=scraper.ResponseCache(tmp_path / "cache"),
        sleeper=RecordingSleeper(),
    )
    assert len(items) == EXPECTED_ITEM_COUNT
    assert len({item.sku for item in items}) == EXPECTED_ITEM_COUNT
    assert sum(1 for item in items if item.price is None) == 1


def test_the_crawl_waits_the_declared_delay_before_each_real_fetch(
    site, session, robots, tmp_path
):
    sleeper = RecordingSleeper()
    scraper.scrape_catalogue(
        site.url_for("/catalogue/page-1.html"),
        session=session,
        robots=robots,
        cache=scraper.ResponseCache(tmp_path / "cache"),
        sleeper=sleeper,
    )
    assert sleeper.waits == [1.0, 1.0, 1.0]


def test_an_off_site_link_is_refused_rather_than_followed(site, session, robots):
    """The detour fixture's Next link points at a host reserved for docs.

    It must be refused before any connection is attempted — the assertion on
    the server log shows the crawl stopped after the one page it was allowed.
    """
    site.reset()
    with pytest.raises(scraper.OffSite):
        scraper.scrape_catalogue(
            site.url_for("/detour/page-1.html"),
            session=session,
            robots=robots,
            sleeper=RecordingSleeper(),
        )
    assert site.paths() == ["/detour/page-1.html"]


# ---------------------------------------------------------------------------
# Exercise 5 — the cache
# ---------------------------------------------------------------------------


def test_a_second_run_makes_no_page_requests(site, session, robots, tmp_path):
    cache_dir = tmp_path / "cache"
    start = site.url_for("/catalogue/page-1.html")
    first = scraper.scrape_catalogue(
        start,
        session=session,
        robots=robots,
        cache=scraper.ResponseCache(cache_dir),
        sleeper=RecordingSleeper(),
    )
    site.reset()
    warm = scraper.ResponseCache(cache_dir)
    second = scraper.scrape_catalogue(
        start, session=session, robots=robots, cache=warm, sleeper=RecordingSleeper()
    )
    assert site.paths() == []
    assert warm.hits == 3
    assert warm.misses == 0
    assert second == first


def test_a_cache_hit_does_not_pay_the_crawl_delay(site, session, robots, tmp_path):
    cache_dir = tmp_path / "cache"
    start = site.url_for("/catalogue/page-1.html")
    scraper.scrape_catalogue(
        start,
        session=session,
        robots=robots,
        cache=scraper.ResponseCache(cache_dir),
        sleeper=RecordingSleeper(),
    )
    sleeper = RecordingSleeper()
    scraper.scrape_catalogue(
        start,
        session=session,
        robots=robots,
        cache=scraper.ResponseCache(cache_dir),
        sleeper=sleeper,
    )
    assert sleeper.waits == []


def test_the_cache_directory_is_inspectable(site, session, robots, tmp_path):
    cache_dir = tmp_path / "cache"
    scraper.scrape_catalogue(
        site.url_for("/catalogue/page-1.html"),
        session=session,
        robots=robots,
        cache=scraper.ResponseCache(cache_dir),
        sleeper=RecordingSleeper(),
    )
    index = cache_dir / "index.json"
    assert index.exists()
    import json

    mapping = json.loads(index.read_text(encoding="utf-8"))
    assert sorted(mapping.values()) == [site.url_for(path) for path in EXPECTED_PAGES]


def test_scraping_without_a_cache_still_works(site, session, robots):
    site.reset()
    items = scraper.scrape_catalogue(
        site.url_for("/catalogue/page-1.html"),
        session=session,
        robots=robots,
        cache=None,
        sleeper=RecordingSleeper(),
    )
    assert len(items) == EXPECTED_ITEM_COUNT
    assert site.paths() == EXPECTED_PAGES


# ---------------------------------------------------------------------------
# Exercise 6 — CSV output
# ---------------------------------------------------------------------------


def test_the_csv_has_a_header_and_one_row_per_item(site, session, robots, tmp_path):
    items = scraper.scrape_catalogue(
        site.url_for("/catalogue/page-1.html"),
        session=session,
        robots=robots,
        cache=scraper.ResponseCache(tmp_path / "cache"),
        sleeper=RecordingSleeper(),
    )
    out = scraper.write_csv(items, tmp_path / "catalogue.csv")
    lines = out.read_text(encoding="utf-8").splitlines()
    assert lines[0] == "sku,name,price,stock,source_path"
    assert len(lines) == EXPECTED_ITEM_COUNT + 1


def test_the_missing_price_becomes_an_empty_field_not_the_word_none(
    site, session, robots, tmp_path
):
    items = scraper.scrape_catalogue(
        site.url_for("/catalogue/page-1.html"),
        session=session,
        robots=robots,
        cache=scraper.ResponseCache(tmp_path / "cache"),
        sleeper=RecordingSleeper(),
    )
    out = scraper.write_csv(items, tmp_path / "catalogue.csv")
    import csv as csv_module

    with out.open(newline="", encoding="utf-8") as handle:
        rows = list(csv_module.DictReader(handle))
    compass = next(row for row in rows if row["sku"] == "NAV-003")
    assert compass["price"] == ""
    assert "None" not in out.read_text(encoding="utf-8")


def test_the_ampersand_survives_the_round_trip(site, session, robots, tmp_path):
    items = scraper.scrape_catalogue(
        site.url_for("/catalogue/page-1.html"),
        session=session,
        robots=robots,
        cache=scraper.ResponseCache(tmp_path / "cache"),
        sleeper=RecordingSleeper(),
    )
    out = scraper.write_csv(items, tmp_path / "catalogue.csv")
    import csv as csv_module

    with out.open(newline="", encoding="utf-8") as handle:
        rows = list(csv_module.DictReader(handle))
    quills = next(row for row in rows if row["sku"] == "STA-001")
    assert quills["name"] == "Ink & Quill Set"


# ---------------------------------------------------------------------------
# The parser earns its place: a regular expression cannot do this
# ---------------------------------------------------------------------------

NAIVE_NAME_RE = re.compile(r'<td class="name">([^<]*)</td>')


def test_the_naive_regex_misses_rows_the_parser_finds():
    html = page_html("page-1.html")
    regex_names = [match.group(1) for match in NAIVE_NAME_RE.finditer(html)]
    parser_names = [item.name for item in scraper.parse_items(html)]
    assert len(parser_names) == 4
    assert len(regex_names) == 2
    assert "Mariner Astrolabe" in parser_names
    assert "Mariner Astrolabe" not in regex_names


def test_the_naive_regex_returns_an_undecoded_entity():
    html = page_html("page-1.html")
    regex_names = [match.group(1) for match in NAIVE_NAME_RE.finditer(html)]
    assert "Ink &amp; Quill Set" in regex_names
    parser_names = [item.name for item in scraper.parse_items(html)]
    assert "Ink & Quill Set" in parser_names
    assert "Ink &amp; Quill Set" not in parser_names


def test_the_naive_regex_returns_unstripped_whitespace():
    html = page_html("page-3.html")
    regex_names = [match.group(1) for match in NAIVE_NAME_RE.finditer(html)]
    assert any(name != name.strip() for name in regex_names)
    parser_names = [item.name for item in scraper.parse_items(html)]
    assert all(name == name.strip() for name in parser_names)