Demo release: 14ee5a7eecbd2294
The real Python crawler / pandas ETL behind the results viewer
Sample / portfolio project — demo
These are the current release Python source files for the crawler, pandas
cleaning pipeline and CSV/Excel exporters. The code is embedded statically;
this page loads only local assets. The viewer and downloads are rebuilt from
an archived cleaned 60-book sample from the
books.toscrape.com
sandbox. The original full raw capture is not included; the source project
includes one listing page with detail fixtures for offline testing.
Regenerate locally: python -m scraper.cli --pages 3 --out output → writes CSV, styled Excel and the viewer dataset.
## Viewer QA additions (Oct 2026)
- Sortable headers are keyboard-accessible (Tab + Enter/Space).
- On small screens every column (including Rating/Stock) stays reachable via the horizontal scroll region, and a **Sort by** dropdown offers every sort option without needing to scroll to its header.
- Unknown ratings are shown as "Not rated", never as a zero-star rating.
- Note: categories such as "Default" and "Add a comment" are real category names on books.toscrape.com, not parser errors.
# python-data-scraper
A small, self-contained **web-scraping + ETL pipeline** in Python that crawls a
public book catalogue, cleans and validates the data, and publishes it as a
CSV, a formatted Excel workbook, and an interactive browser-based results
viewer.
> **Sample / portfolio project** by **Hametachnet**. It targets
> [books.toscrape.com](https://books.toscrape.com) — a sandbox site built
> specifically for scraping practice — and uses a clearly fictional demo brand
> ("BookVault"). No real clients, stores or testimonials are implied.
---
## What it does
1. **Fetch** catalogue pages politely (session reuse, descriptive User-Agent,
timeouts, exponential-backoff retries, a delay between requests, and a
`robots.txt` check).
2. **Crawl two levels** — each listing page is parsed for book stubs, then each
book's detail page is fetched to enrich it with its real category, exact
stock count, UPC and description.
3. **Transform** — clean prices (`£51.77` → `51.77`), coerce ratings to ints,
preserve exact availability as nullable integer counts, de-duplicate (by UPC, then title),
and validate rows.
4. **Export** — write `output/current/books.csv`, a styled `output/current/books.xlsx`
(frozen/filtered header, column widths, and a *Summary by category* sheet),
and `output/current/data.js` for the HTML viewer. A complete generation is written
under `output/runs/`; one atomic `current` symlink switch publishes it. Existing
generations remain available. Readers needing a consistent set should resolve
`current` once and read all files from that generation. Requires symlink support.
## Features
- Polite, resilient HTTP layer with retries, backoff and robots awareness.
- **Offline mode** (`--offline`) replays HTML fixtures saved under
`tests/fixtures/`, so the whole pipeline runs with **no network**. Live and
offline runs share the exact same crawl code — offline just swaps in a
fetcher that reads local files.
- Clean pandas transform with de-duplication and validation.
- Formatted Excel output + per-category summary sheet.
- A **self-contained results viewer** (`viewer/index.html`) with live search,
category filter, sortable columns and summary stats. No CDNs, no external
images (inline SVG only) — it works straight from `file://`.
- Tests for the parse and transform layers (run under `pytest` *or* the stdlib
`unittest`).
## Tech stack
`requests` · `beautifulsoup4` · `pandas` · `openpyxl` · standard library
(`argparse`, `unittest`, `urllib.robotparser`). Viewer is vanilla
HTML/CSS/JS — zero dependencies.
## Getting started
```bash
pip install -r requirements.txt
# Live crawl of 3 catalogue pages (default)
python -m scraper.cli --pages 3
# Run with no network, replaying saved fixtures
python -m scraper.cli --offline --pages 1
# Keep only one category
python -m scraper.cli --offline --pages 1 --category Poetry
# Refresh the test fixtures from the live site (writes to tests/fixtures/)
python -m scraper.cli --save-fixtures --pages 1
```
Copy `output/current/data.js` beside a copy of the published demo `index.html`,
then open that copy in a browser. Publication writes only under `--out` (default `./output` in the working
directory); `--save-fixtures` separately writes into the source fixtures directory. The checked-in source `viewer/` is a historical snapshot; the maintained
viewer is `site/demos/python-data-scraper/` in the portfolio repository.
### CLI options
| Flag | Description |
|------|-------------|
| `--pages N` | number of catalogue pages to crawl (default `3`) |
| `--out DIR` | publication directory (default: working-directory `output`; files under `current/`) |
| `--allow-partial` | explicitly permit publication after some listing failures; zero successful listings always aborts |
| `--category NAME` | keep only books in this category (case-insensitive) |
| `--offline` | parse saved fixtures instead of hitting the network |
| `--no-enrich` | skip the detail-page crawl (faster; no category/stock detail) |
| `--delay S` | polite delay between requests in seconds (default `0.4`) |
| `--save-fixtures` | mirror fetched pages into `tests/fixtures/` (online) |
| `--quiet` | log warnings/errors only |
## Running the tests
```bash
python -m unittest discover -s tests # stdlib, no extra install
# or, if pytest is installed:
pytest -q
```
## Project structure
```
python-data-scraper/
├── scraper/
│ ├── fetch.py # polite HTTP session + OfflineFetcher (fixture replay)
│ ├── parse.py # BeautifulSoup parsing of listing & detail pages
│ ├── transform.py # clean / type / dedupe / validate → pandas DataFrame
│ ├── export.py # CSV, formatted Excel, and viewer/data.js
│ ├── pipeline.py # crawl orchestration (shared by live & offline)
│ └── cli.py # argparse entry point
├── tests/
│ ├── test_parse.py
│ ├── test_transform.py
│ └── fixtures/ # saved HTML pages for offline mode & tests
├── viewer/
│ ├── index.html # self-contained results viewer
│ └── data.js # generated dataset (window.SCRAPER_DATA)
├── output/ # current symlink and versioned runs/ directories
├── requirements.txt
└── README.md
```
## Notes on responsible scraping
The target site exists expressly for scraping practice, so the pipeline is a
safe place to demonstrate good manners: a descriptive User-Agent, request
delays, backoff on failure, and a `robots.txt` check before fetching. The same
patterns transfer directly to real-world crawls.
---
*Demo project by Hametachnet.*
Unknown stock quantities stay blank in CSV/Excel and null in JSON, independently
of the listing's `in_stock` status. Summary `total_stock` sums known quantities
(null if none are known or the total exceeds 2**53 - 1); `unknown_stock` reports how many quantities are missing.
The viewer labels incomplete totals. Supported price prefixes are £, £, $, €,
GBP, USD and EUR; arbitrary text prefixes fail validation. Invalid rows are counted
in the CLI cleaning report. Both CSV and spreadsheet text have control and other
XML-forbidden characters removed and formula-leading cells neutralized; spreadsheet
text is additionally truncated to 32,767 UTF-16 units, with warnings. Viewer text
retains the cleaned source text.
Listing fetch/parse failures abort publication by default. The bundled offline
fixtures cover one listing page; requesting additional pages requires matching
fixtures or `--allow-partial`. Detail failures retain listing records with unknown
quantities. No successful listing pages always means failure, regardless of flags.
Robots absence is accepted only for 404/410; unavailable policies fail closed.
Responses are streamed with decompressed limits of 512 KiB for robots and 5 MiB
for HTML and a 30-second response deadline per attempt covering headers and body (socket
timeouts also apply to connection setup).
Retries honor Retry-After up to 60 seconds; longer requests fail closed.
Additional regression checks:
```bash
node tests/viewer_verify.cjs
/workspace/.pwvenv/bin/python tests/browser_verify.py
```
The browser check serves a copy at the required localhost URL and needs a
session that permits sockets and Chrome. It does not contact external hosts.
## Reviewed static release
Stock counts must be whole numbers in 0..2**53 - 1, so JSON and spreadsheet
exports can represent them exactly. Rejected counts and out-of-range totals
are logged and exported as unknown. Totals use Python integer arithmetic.
Description preview removal requires a trailing read-more marker; ordinary
repeated prose is preserved. Malformed listing links retain unlinked records;
listing/pagination failures participate in the CLI partial-publication policy.
Robots matching normalizes Unicode and percent escapes, preserves encoded
reserved octets, and refuses URLs when the total matching budget is exceeded.
Query-bearing fixture URLs have SHA-256 query suffixes for symmetric replay.
Run `python3 build_demo.py` to regenerate the published demo's data, downloads,
and embedded source from these release files and `release_snapshot.json`, the archived cleaned
60-book sample. Only one listing page and its detail fixtures are included;
they do not reproduce the original full raw capture.
The builder verifies decoded source blocks byte-for-byte and stamps the viewer,
source page, data and release manifest with one content-derived release ID.
Run `python3 build_demo.py --check` to verify the source blocks, source/fixture
hashes, and published artifact hashes without writing files.
Cloudflare Pages site headers provide the primary anti-framing control; the
JavaScript frame guard is defense in depth and allows direct visits without JS.
requests>=2.31
beautifulsoup4>=4.12
pandas>=2.0
openpyxl>=3.1
"""python-data-scraper — a self-contained scraping / ETL demo pipeline.
Demo project by Hametachnet. Targets the public scraping sandbox
https://books.toscrape.com which exists specifically for scraping practice.
"""
__version__ = "1.0.0"
"""Command-line entry point for the scraping / ETL pipeline.
Examples
--------
python -m scraper.cli --pages 3 # live crawl, 3 pages
python -m scraper.cli --offline --pages 1 # replay saved fixtures
python -m scraper.cli --category Travel # filter one category
python -m scraper.cli --save-fixtures --pages 1 # refresh test fixtures
"""
from __future__ import annotations
import argparse
import logging
import os
import sys
import shutil
import tempfile
import uuid
from pathlib import Path
from . import __version__
from .export import to_csv, to_excel, to_viewer
from .fetch import FetchConfig, Fetcher, OfflineFetcher
from .pipeline import crawl
from .transform import build_dataframe, category_summary
ROOT = Path(__file__).resolve().parent.parent
FIXTURES_DIR = ROOT / "tests" / "fixtures"
def build_parser() -> argparse.ArgumentParser:
p = argparse.ArgumentParser(
prog="python-data-scraper",
description="Scrape books.toscrape.com and export a cleaned dataset "
"(demo project by Hametachnet).",
)
p.add_argument("--pages", type=int, default=3, help="number of catalogue pages to crawl (default: 3)")
p.add_argument("--out", default="output", help="publication directory (exports under current/) (default: ./output)")
p.add_argument("--allow-partial", action="store_true", help="publish even if some listing pages failed (never a completely failed crawl)")
p.add_argument("--category", default=None, help="keep only books in this category (case-insensitive)")
p.add_argument("--offline", action="store_true", help="parse saved fixtures instead of hitting the network")
p.add_argument("--no-enrich", action="store_true", help="skip detail-page crawl (no category/stock detail)")
p.add_argument("--delay", type=float, default=0.4, help="polite delay between requests in seconds (default: 0.4)")
p.add_argument("--save-fixtures", action="store_true", help="mirror fetched pages into tests/fixtures (online)")
p.add_argument("--quiet", action="store_true", help="only log warnings and errors")
p.add_argument("--version", action="version", version=f"%(prog)s {__version__}")
return p
def run(args: argparse.Namespace) -> int:
logging.basicConfig(
level=logging.WARNING if args.quiet else logging.INFO,
format="%(levelname)s %(message)s",
)
log = logging.getLogger("scraper.cli")
if args.pages <= 0:
raise ValueError("--pages must be positive")
if args.delay < 0:
raise ValueError("--delay must be non-negative")
enrich = not args.no_enrich
source = "offline" if args.offline else "live"
if args.offline:
fetcher = OfflineFetcher(FIXTURES_DIR)
else:
save_dir = FIXTURES_DIR if args.save_fixtures else None
fetcher = Fetcher(FetchConfig(delay=args.delay), save_dir=save_dir)
log.info("Crawling (%s mode), pages=%s, enrich=%s", source, args.pages, enrich)
records = crawl(fetcher, pages=args.pages, enrich=enrich)
log.info("Collected %s raw records", len(records))
if not records.succeeded or (records.failed and not args.allow_partial):
raise RuntimeError(f"Publication aborted: {records.succeeded} listing pages succeeded, {records.failed} failed; use --allow-partial for partial results")
df = build_dataframe(records)
log.info(
"Cleaned to %s rows (dropped %s invalid, %s duplicates)",
len(df),
df.attrs.get("dropped_invalid", 0),
df.attrs.get("dropped_duplicates", 0),
)
if df.empty:
raise RuntimeError("No valid records; publication aborted")
if args.category:
wanted = args.category.strip().lower()
df = df[df["category"].str.lower() == wanted].reset_index(drop=True)
log.info("Filtered to category %r: %s rows", args.category, len(df))
out_dir = Path(args.out)
out_dir.mkdir(parents=True, exist_ok=True)
# All artifacts are immutable within a generation. Only the current link
# switches, so a failed exporter or pointer replacement preserves the old run.
runs = out_dir / "runs"
runs.mkdir(exist_ok=True)
generation = Path(tempfile.mkdtemp(prefix="run-", dir=runs))
# mkdtemp creates the directory 0700; relax it (subject to umask) so a web
# server running as a different user can traverse in and serve the exports.
os.chmod(generation, 0o755)
pointer = out_dir / (".current-" + uuid.uuid4().hex)
committed = False
try:
to_csv(df, generation / "books.csv")
to_excel(df, generation / "books.xlsx")
to_viewer(df, generation / "data.js", source=source)
pointer.symlink_to(Path("runs") / generation.name, target_is_directory=True)
os.replace(pointer, out_dir / "current")
committed = True
finally:
pointer.unlink(missing_ok=True)
if not committed:
shutil.rmtree(generation)
csv_path = out_dir / "current" / "books.csv"
xlsx_path = out_dir / "current" / "books.xlsx"
viewer_path = out_dir / "current" / "data.js"
print(f"\n{'='*60}")
print(f" Done — {len(df)} books across {df['category'].nunique()} categories")
print(f"{'='*60}")
print(f" CSV : {csv_path}")
print(f" Excel : {xlsx_path}")
print(f" Viewer : {viewer_path} (copy beside the static viewer index.html to view)")
summary = category_summary(df)
if not summary.empty:
print("\n Top categories:")
for row in summary.head(5).itertuples(index=False):
print(f" {row.category:<24} {row.books:>3} books avg £{row.avg_price:.2f}")
return 0
def main(argv: list[str] | None = None) -> int:
args = build_parser().parse_args(argv)
try:
return run(args)
except Exception as exc: # pragma: no cover - top-level guard
logging.getLogger("scraper.cli").error("Pipeline failed: %s", exc)
return 1
if __name__ == "__main__":
sys.exit(main())
"""Crawl orchestration: tie fetch + parse together into raw records.
The same :func:`crawl` function drives both live and offline runs — the only
difference is which fetcher object it is handed. That keeps the two code paths
identical and makes ``--offline`` a faithful replay of a live crawl.
"""
from __future__ import annotations
import logging
from .fetch import page_url
from .parse import parse_detail, parse_listing, next_page_path
from urllib.parse import urljoin
log = logging.getLogger(__name__)
class CrawlResult(list):
"""Records plus listing completion counts for publication policy."""
def __init__(self):
super().__init__()
self.succeeded = 0
self.failed = 0
def crawl(fetcher, pages: int = 3, enrich: bool = True) -> list[dict]:
"""Crawl ``pages`` catalogue pages and return raw per-book records.
``fetcher`` must expose a ``.get(url) -> str`` method (both
:class:`~scraper.fetch.Fetcher` and
:class:`~scraper.fetch.OfflineFetcher` do). When ``enrich`` is true each
book's detail page is also fetched for category / stock / UPC.
"""
if pages <= 0:
raise ValueError("pages must be positive")
records = CrawlResult()
url = page_url(1)
for page in range(1, pages + 1):
log.info("Fetching listing page %s: %s", page, url)
try:
html = fetcher.get(url)
except FileNotFoundError:
records.failed += 1
log.warning("Missing listing fixture for page %s", page)
break
except (PermissionError, RuntimeError, ValueError) as exc:
# Disallowed by robots/origin, or failed after retries — skip this
# page rather than aborting the whole crawl.
log.warning("Skipping listing page %s: %s", url, exc)
records.failed += 1
break
# Resolve links against the URL the page was actually served from
# (after any redirects), not the URL we requested.
base = getattr(html, "final_url", url)
try:
stubs = parse_listing(html, base_url=base)
except (ValueError, TypeError, RuntimeError) as exc:
log.warning("Listing parse failed: %s", exc)
records.failed += 1
break
if not stubs:
records.failed += 1
break
records.succeeded += 1
for stub in stubs:
if enrich and stub.get("detail_url"):
try:
detail_html = fetcher.get(stub["detail_url"])
stub.update(parse_detail(detail_html))
except FileNotFoundError:
log.debug("No detail fixture for %s", stub.get("title"))
except (PermissionError, RuntimeError, ValueError) as exc:
# Disallowed or unreachable detail page — keep the listing
# stub without enrichment and move on.
log.warning("Skipping detail page for %s: %s", stub.get("title"), exc)
records.append(stub)
log.info("Page %s: %s books (running total %s)", page, len(stubs), len(records))
try:
next_path = next_page_path(html)
if not next_path or page == pages:
break
url = urljoin(base, next_path)
except (ValueError, TypeError, RuntimeError) as exc:
log.warning("Pagination parse failed: %s", exc)
records.failed += 1
break
return records
"""HTTP fetching with a polite, resilient session.
Responsible scraping notes
--------------------------
* We target https://books.toscrape.com, a sandbox built expressly for
scraping practice, so the load we create is expected and welcome.
* Requests are restricted to that exact HTTPS origin: a link scraped from a
page can never steer the crawler to another host or an internal service.
* ``robots.txt`` is fetched through the same polite session (timeout, retries,
descriptive User-Agent) and checked once per host via :func:`robots_allows`;
if it cannot be retrieved we fail closed and skip the URL.
* Every request carries a descriptive User-Agent, a timeout, exponential
backoff on transient failures, and a small fixed delay between calls so we
never hammer the server. Redirects are followed manually so each hop is
re-validated against the origin and robots policies.
"""
from __future__ import annotations
import hashlib
import logging
import time
import math
import re
import socket
import threading
from datetime import datetime, timezone
from email.utils import parsedate_to_datetime
from dataclasses import dataclass
from pathlib import Path
from urllib.parse import urljoin, urlparse, urlsplit, urlunsplit
import requests
from requests.adapters import HTTPAdapter
from urllib3.connection import HTTPSConnection
from urllib3.connectionpool import HTTPSConnectionPool
class DeadlineAdapter(HTTPAdapter):
"""Interrupt trickling response headers as well as bounded body reads."""
def __init__(self, config):
self.context = threading.local()
self.fetch_config = config
super().__init__()
def init_poolmanager(self, *args, **kwargs):
super().init_poolmanager(*args, **kwargs)
owner = self
class Connection(HTTPSConnection):
def getresponse(self):
remaining = owner.fetch_config.deadline - (time.monotonic() - owner.context.started)
if remaining <= 0:
raise requests.Timeout("HTTP deadline exceeded before headers")
sock = self.sock
def abort():
try:
sock.shutdown(socket.SHUT_RDWR)
except OSError:
pass
timer = threading.Timer(remaining, abort)
timer.daemon = True
timer.start()
try:
return super().getresponse()
finally:
timer.cancel()
class Pool(HTTPSConnectionPool):
ConnectionCls = Connection
self.poolmanager.pool_classes_by_scheme = dict(self.poolmanager.pool_classes_by_scheme)
self.poolmanager.pool_classes_by_scheme["https"] = Pool
def send(self, *args, **kwargs):
self.context.started = time.monotonic()
return super().send(*args, **kwargs)
log = logging.getLogger(__name__)
def _best_effort_shutdown(resp) -> None:
"""Try to unblock a read parked on a trickling socket.
This reaches into urllib3/http.client internals whose layout is not a
public contract, so the access is isolated here and *every* failure is
swallowed. Callers must never depend on it: the deadline is actually
enforced by the socket read timeout, the per-chunk deadline check and
``resp.close()`` — this is only an optimisation to interrupt sooner.
"""
try:
sock = resp.raw._fp.fp.raw._sock
except AttributeError:
return
try:
sock.shutdown(socket.SHUT_RDWR)
except OSError:
pass
DEFAULT_USER_AGENT = (
"python-data-scraper/1.0 (portfolio demo by Hametachnet; "
"+https://books.toscrape.com)"
)
BASE_URL = "https://books.toscrape.com/"
# The crawler is scoped to this one sandbox origin. Only HTTPS requests to this
# host (on the default port) are ever made, which neutralises SSRF via scraped
# links/redirects to internal or arbitrary services.
ALLOWED_HOSTS = frozenset({"books.toscrape.com"})
# HTTP statuses worth retrying; everything else (e.g. 404) fails immediately.
RETRY_STATUSES = frozenset({429, 500, 502, 503, 504})
# Bound the number of redirect hops we will follow for a single request.
MAX_REDIRECTS = 5
# Cap the robots.txt body we parse so a hostile/huge file can't exhaust memory.
MAX_ROBOTS_BYTES = 512 * 1024
MAX_HTML_BYTES = 5 * 1024 * 1024
class FetchedPage(str):
"""The text of a fetched page that also remembers its final URL.
Behaves exactly like :class:`str` (so parsers and tests that compare it to
text keep working), but carries ``final_url`` — the URL the response was
actually served from after any redirects — so callers can resolve relative
links and pagination against the right base.
"""
final_url: str
def __new__(cls, text: str, final_url: str) -> "FetchedPage":
obj = super().__new__(cls, text)
obj.final_url = final_url
return obj
class RobotsPolicy:
"""A small RFC 9309 matcher with longest-match precedence and wildcards.
``urllib.robotparser`` applies rules first-match, so a broad ``Allow: /``
placed before a specific ``Disallow: /private/`` would wrongly permit
``/private/secret``. This parser instead selects the *longest* matching
rule for the applicable user-agent group (``Allow`` winning ties) and
supports ``*`` wildcards and the ``$`` end-of-path anchor.
"""
def __init__(self) -> None:
# Each group is (list of user-agent tokens, list of (is_allow, pattern)).
self._groups: list[tuple[list[str], list[tuple[bool, str]]]] = []
def parse(self, lines) -> None:
groups: list[tuple[list[str], list[tuple[bool, str]]]] = []
agents: list[str] = []
rules: list[tuple[bool, str]] = []
saw_rule = False
def flush():
if agents:
groups.append((list(agents), list(rules)))
for raw in lines:
line = raw.split("#", 1)[0].strip()
if not line or ":" not in line:
continue
field, _, value = line.partition(":")
field = field.strip().lower()
value = value.strip()
if field == "user-agent":
# A user-agent line after any rule starts a fresh group.
if saw_rule:
flush()
agents = []
rules = []
saw_rule = False
agents.append(value.lower())
elif field in ("allow", "disallow"):
saw_rule = True
rules.append((field == "allow", value))
flush()
self._groups = groups
@staticmethod
def _normalize(value: str) -> str:
"""RFC 9309 octets: decode unreserved escapes, retain reserved ones."""
unreserved = "ABCDEFGHIJKLMNOPQRSTUVWXYZabcdefghijklmnopqrstuvwxyz0123456789-._~"
def escape(match):
char = chr(int(match.group()[1:], 16))
return char if char in unreserved else match.group().upper()
value = re.sub(r"%[0-9a-fA-F]{2}", escape, value)
return "".join(ch if ord(ch) < 128 else "".join(f"%{b:02X}" for b in ch.encode("utf-8"))
for ch in value)
@staticmethod
def _matches(pattern: str, path: str) -> bool:
# Dynamic programming has bounded O(pattern * path) work, no regex
# backtracking. can_fetch bounds the total work across all rules.
anchored = pattern.endswith("$")
if anchored:
pattern = pattern[:-1]
previous = [True] + [False] * len(path)
for char in pattern:
current = [previous[0] and char == "*"] + [False] * len(path)
for j, actual in enumerate(path, 1):
current[j] = (previous[j] or current[j-1]) if char == "*" else (previous[j-1] and char == actual)
previous = current
return previous[-1] if anchored else any(previous)
def _applicable_rules(self, user_agent: str) -> list[tuple[bool, str]]:
token = user_agent.split("/")[0].strip().lower()
specific: list[tuple[bool, str]] = []
wildcard: list[tuple[bool, str]] = []
has_specific = False
for agents, rules in self._groups:
for agent in agents:
if agent == "*":
wildcard.extend(rules)
elif agent and token.startswith(agent):
has_specific = True
specific.extend(rules)
return specific if has_specific else wildcard
def can_fetch(self, user_agent: str, url: str) -> bool:
parts = urlsplit(url)
path = parts.path or "/"
if parts.query:
path += "?" + parts.query
path = self._normalize(path)
work = 0
allow_len = -1
disallow_len = -1
for is_allow, pattern in self._applicable_rules(user_agent):
if not pattern: # empty Disallow/Allow imposes no restriction
continue
pattern = self._normalize(pattern)
work += (len(pattern) + 1) * (len(path) + 1)
if work > 1_000_000:
log.warning("Robots matching work budget exceeded; refusing URL")
return False
if self._matches(pattern, path):
if is_allow:
allow_len = max(allow_len, len(pattern))
else:
disallow_len = max(disallow_len, len(pattern))
if disallow_len < 0:
return True
if allow_len < 0:
return False
return allow_len >= disallow_len # tie goes to Allow
def prepare_url(url: str) -> str:
"""Prepare once, rejecting ambiguous encoded paths before policy checks."""
parts = urlsplit(url)
if parts.username is not None or parts.password is not None:
raise ValueError("URL credentials are not allowed")
path = parts.path
# Repeated encodings and encoded path separators can mean different things
# to robots parsers, proxies and servers. Refuse those ambiguous spellings.
if re.search(r"%(?:25|2e|2f|5c)", path, re.I) or "\\" in url:
raise ValueError("Ambiguous encoded URL path")
host = (parts.hostname or "").lower().removesuffix(".")
if parts.scheme != "https" or host not in ALLOWED_HOSTS or parts.port not in (None, 443):
raise ValueError("URL is outside the permitted HTTPS origin")
normalized = urlunsplit(("https", host, path, parts.query, ""))
return requests.Request("GET", normalized).prepare().url
def is_allowed_url(url: str) -> bool:
"""Accept only URLs that can be prepared for the permitted origin."""
try:
prepare_url(url)
return True
except (ValueError, requests.RequestException):
return False
@dataclass
class FetchConfig:
"""Tunable knobs for polite fetching."""
user_agent: str = DEFAULT_USER_AGENT
timeout: float = 15.0
max_retries: int = 3
backoff_factor: float = 0.8 # seconds; doubled each retry
deadline: float = 30.0 # total seconds per HTTP attempt, including body
delay: float = 0.4 # polite pause between requests (matches the CLI default)
class Fetcher:
"""A ``requests.Session`` wrapper that retries, backs off and waits."""
def __init__(self, config: FetchConfig | None = None, save_dir: str | Path | None = None):
self.config = config or FetchConfig()
if (self.config.max_retries < 1 or any(not math.isfinite(x) or x <= 0
for x in (self.config.timeout, self.config.deadline))
or any(not math.isfinite(x) or x < 0 for x in (self.config.delay, self.config.backoff_factor))):
raise ValueError("Invalid fetch limits")
self.session = requests.Session()
self.session.trust_env = False
self.session.mount("https://", DeadlineAdapter(self.config))
self.session.headers.update({"User-Agent": self.config.user_agent})
# Per-origin robots parser, or False when robots.txt could not be read
# (fail closed). ``None`` never appears here, so a missing key means
# "not fetched yet" and a present key is a cached decision.
self._robots: dict[str, RobotsPolicy | bool] = {}
# When set, every fetched page is mirrored to disk so it can later be
# replayed offline. Used to build the test fixtures.
self.save_dir = Path(save_dir) if save_dir else None
# -- robots -----------------------------------------------------------
def _load_robots(self, root: str) -> RobotsPolicy | bool:
"""Fetch and parse ``root``'s robots.txt via the polite session.
Returns a parser on success, ``True`` when the host has no robots.txt
(nothing disallowed), or ``False`` when it could not be retrieved
(treated as "disallow" by the caller — fail closed).
"""
robots_url = urljoin(root + "/", "robots.txt")
try:
resp = self._request(robots_url, limit=MAX_ROBOTS_BYTES, missing_ok=True)
except RuntimeError as exc:
log.warning("Could not establish robots policy for %s: %s", root, exc)
return False
if resp.status_code in (404, 410):
return True
if resp.status_code != 200:
return False
rp = RobotsPolicy()
rp.parse(resp.text.splitlines())
return rp
def robots_allows(self, url: str) -> bool:
"""Return whether ``robots.txt`` permits fetching ``url``.
The decision is cached per origin. If robots.txt cannot be retrieved we
fail closed (return ``False``) and cache that so it is not re-fetched on
every subsequent URL.
"""
url = prepare_url(url)
parts = urlparse(url)
root = f"{parts.scheme}://{parts.netloc}"
if root not in self._robots:
self._robots[root] = self._load_robots(root)
rp = self._robots[root]
if rp is True:
return True
if rp is False:
return False
return rp.can_fetch(self.config.user_agent, url)
# -- fetching ---------------------------------------------------------
def _read_body(self, resp, limit: int, started: float) -> None:
"""Bound decompressed bytes and interrupt a trickling socket at deadline."""
remaining = self.config.deadline - (time.monotonic() - started)
if remaining <= 0:
raise RuntimeError("HTTP deadline exceeded")
expired = threading.Event()
def abort():
expired.set()
# resp.close() is the reliable stop; the socket shutdown is only a
# best-effort nudge to unblock a read already parked on a trickling
# socket. The chunk loop's own deadline check is the real guarantee.
_best_effort_shutdown(resp)
resp.close()
timer = threading.Timer(remaining, abort)
timer.daemon = True
timer.start()
body = bytearray()
try:
for chunk in resp.iter_content(chunk_size=8192):
if expired.is_set() or time.monotonic() - started >= self.config.deadline:
raise RuntimeError("HTTP deadline exceeded")
if len(body) + len(chunk) > limit:
raise RuntimeError("HTTP decoded body size limit exceeded")
body.extend(chunk)
if expired.is_set() or time.monotonic() - started >= self.config.deadline:
raise RuntimeError("HTTP deadline exceeded")
resp._content = bytes(body)
resp.encoding = "utf-8"
finally:
timer.cancel()
def _request(self, url: str, limit: int = MAX_HTML_BYTES, missing_ok: bool = False) -> requests.Response:
"""Send the exact prepared URL, with bounded retries and streamed reads."""
url = prepare_url(url)
last_exc = None
for attempt in range(self.config.max_retries):
resp = None
wait = self.config.backoff_factor * 2 ** attempt
try:
started = time.monotonic()
request = self.session.prepare_request(requests.Request("GET", url))
if request.url != url:
raise RuntimeError("Prepared URL changed after policy check")
# Adapter.send avoids Session.send's redirect prefetch, which
# would otherwise buffer an unbounded redirect body.
resp = self.session.get_adapter(url).send(request, stream=True,
timeout=min(self.config.timeout, self.config.deadline))
if resp.status_code in RETRY_STATUSES:
retry_after = resp.headers.get("Retry-After")
if retry_after:
try:
retry_wait = float(retry_after)
except ValueError:
try:
retry_wait = (parsedate_to_datetime(retry_after) - datetime.now(timezone.utc)).total_seconds()
except (ValueError, TypeError, OverflowError):
retry_wait = 0
# Do not retry earlier than the server asks. If outside
# our retry budget, fail closed instead of shortening it.
if not math.isfinite(retry_wait) or retry_wait > 60:
raise RuntimeError("Retry-After exceeds retry budget")
wait = max(wait, retry_wait)
raise requests.RequestException(f"Transient HTTP {resp.status_code}")
if resp.status_code >= 400 and not (missing_ok and resp.status_code in (404, 410)):
raise RuntimeError(f"Permanent HTTP {resp.status_code} fetching {url}")
if resp.status_code == 200:
self._read_body(resp, limit, started)
return resp
except requests.RequestException as exc:
last_exc = exc
finally:
if resp is not None:
resp.close()
time.sleep(self.config.delay)
if attempt + 1 < self.config.max_retries:
time.sleep(wait)
raise RuntimeError(f"Failed to fetch {url} after {self.config.max_retries} attempts") from last_exc
def get(self, url: str) -> FetchedPage:
"""Fetch ``url`` and return its text, tagged with the final URL.
Only HTTPS requests to the permitted origin are made, redirects are
followed manually (re-checking origin + robots on each hop), and
transient errors are retried with exponential backoff. The returned
:class:`FetchedPage` behaves like the page text but also exposes
``final_url`` so relative links resolve against the real base.
"""
current = prepare_url(url)
for _ in range(MAX_REDIRECTS + 1):
if not is_allowed_url(current):
raise PermissionError(
f"Refusing to fetch non-HTTPS or off-origin URL: {current}"
)
if not self.robots_allows(current):
raise PermissionError(f"robots.txt disallows fetching {current}")
resp = self._request(current)
if resp.is_redirect:
location = resp.headers.get("Location")
if not location:
raise RuntimeError(f"Redirect from {current} had no Location header")
current = prepare_url(urljoin(current, location))
continue
if resp.status_code != 200:
raise RuntimeError(f"Unexpected HTTP {resp.status_code} fetching {current}")
# The sandbox serves UTF-8 but omits a charset hint, so requests
# falls back to ISO-8859-1 and mangles £/’ — force UTF-8.
resp.encoding = "utf-8"
if self.save_dir is not None:
self._mirror(current, resp.text)
return FetchedPage(resp.text, current)
raise RuntimeError(f"Exceeded {MAX_REDIRECTS} redirects starting from {url}")
def _mirror(self, url: str, text: str) -> None:
"""Save a fetched page under ``save_dir`` mirroring its URL path."""
dest = _resolve_within(self.save_dir, url) # type: ignore[arg-type]
dest.parent.mkdir(parents=True, exist_ok=True)
dest.write_text(text, encoding="utf-8")
class OfflineFetcher:
"""Replay saved fixtures with the same interface as :class:`Fetcher`.
URLs are mapped to files under ``fixtures_dir`` using the same
URL-to-path convention the live :class:`Fetcher` uses when mirroring, so
the crawl code is byte-for-byte identical between live and offline runs.
"""
def __init__(self, fixtures_dir: str | Path):
self.fixtures_dir = Path(fixtures_dir)
def robots_allows(self, url: str) -> bool: # fixtures are always allowed
return True
def get(self, url: str) -> FetchedPage:
if not is_allowed_url(url):
raise PermissionError(
f"Refusing to replay non-HTTPS or off-origin URL: {url}"
)
prepared = prepare_url(url)
path = _resolve_within(self.fixtures_dir, prepared)
if not path.exists():
raise FileNotFoundError(
f"No fixture for {url} (expected {path}). "
"Run `python -m scraper.cli --save-fixtures` while online."
)
return FetchedPage(path.read_text(encoding="utf-8"), prepared)
def _url_to_relpath(url: str) -> str:
"""Map a site URL to a stable relative path for mirroring/replay."""
parts = urlparse(url)
path = parts.path.lstrip("/")
if not path or path.endswith("/"):
path += "index.html"
# Never preserve an arbitrary remote filename/extension inside the test
# tree. Forcing a fixed, non-executable ``.html`` suffix means a hostile
# upstream link such as ``/conftest.py`` or ``/test_payload.py`` cannot
# plant a file pytest would import or collect. Mirrored pages are HTML, so
# legitimate fixtures (``*.html``) are unaffected and replay stays symmetric.
if not path.endswith(".html"):
path += ".html"
if parts.query:
digest = hashlib.sha256(parts.query.encode("utf-8")).hexdigest()
path = path[:-5] + ".query-" + digest + ".html"
return path
def _resolve_within(root: str | Path, url: str) -> Path:
"""Resolve ``url``'s mirror path and confirm it stays under ``root``.
Guards against path traversal (``..``) and symlink escapes in a scraped
URL, which could otherwise read or overwrite files outside the fixtures
directory.
"""
root_resolved = Path(root).resolve()
dest = (root_resolved / _url_to_relpath(url)).resolve()
if dest != root_resolved and root_resolved not in dest.parents:
raise PermissionError(f"Refusing path outside fixtures root: {url}")
return dest
def page_url(page: int) -> str:
"""Return the catalogue URL for a 1-based page number."""
if page <= 1:
return urljoin(BASE_URL, "catalogue/page-1.html")
return urljoin(BASE_URL, f"catalogue/page-{page}.html")
"""HTML parsing: turn raw pages into plain dictionaries.
Two levels of the site are parsed:
* **listing pages** (``catalogue/page-N.html``) give the title, price,
star-rating, an in-stock flag, the thumbnail and a link to the detail page;
* **detail pages** enrich each book with its real category (from the
breadcrumb), the exact number in stock, the product description and the UPC.
Parsing is deliberately free of any network or filesystem concerns so it can
be unit-tested against the saved fixtures.
"""
from __future__ import annotations
import logging
import re
from urllib.parse import urljoin
from bs4 import BeautifulSoup
_WORD_TO_INT = {
"zero": 0,
"one": 1,
"two": 2,
"three": 3,
"four": 4,
"five": 5,
}
_STOCK_RE = re.compile(r"\((\d+)\s+available\)")
def _soup(html: str) -> BeautifulSoup:
return BeautifulSoup(html, "html.parser")
def rating_from_classes(classes) -> int | None:
"""Map a ``star-rating Three`` class list to the integer ``3``."""
for cls in classes:
key = cls.strip().lower()
if key in _WORD_TO_INT:
return _WORD_TO_INT[key]
return None
def _safe_join(base, href):
if not href:
return None
try:
return urljoin(base, href)
except (ValueError, TypeError):
logging.getLogger(__name__).warning("Malformed listing URL; retaining unlinked record")
return None
def parse_listing(html: str, base_url: str) -> list[dict]:
"""Parse one catalogue listing page into a list of book stubs."""
soup = _soup(html)
books: list[dict] = []
for pod in soup.select("article.product_pod"):
link = pod.select_one("h3 a")
title = link.get("title") if link else None
href = link.get("href") if link else None
detail_url = _safe_join(base_url, href)
price_el = pod.select_one("p.price_color")
price_text = price_el.get_text(strip=True) if price_el else None
rating_el = pod.select_one("p.star-rating")
rating = rating_from_classes(rating_el.get("class", [])) if rating_el else None
avail_el = pod.select_one("p.instock.availability")
in_stock = bool(avail_el and "in stock" in avail_el.get_text().lower())
thumb_el = pod.select_one("img")
thumb = _safe_join(base_url, thumb_el.get("src")) if thumb_el else None
books.append(
{
"title": title,
"price_text": price_text,
"rating": rating,
"in_stock": in_stock,
"detail_url": detail_url,
"thumbnail": thumb,
}
)
return books
def parse_detail(html: str) -> dict:
"""Parse a book detail page for the enrichment fields."""
soup = _soup(html)
# Category is the breadcrumb entry just before the active (title) one.
category = None
crumbs = soup.select("ul.breadcrumb li a")
if len(crumbs) >= 3: # Home / Books / <Category>
category = crumbs[-1].get_text(strip=True)
stock_count = None
avail = soup.select_one("p.instock.availability")
if avail:
match = _STOCK_RE.search(avail.get_text())
if match:
# Bound before int conversion, including Python's digit limit.
digits = match.group(1).lstrip("0") or "0"
from .transform import MAX_STOCK
if len(digits) <= len(str(MAX_STOCK)) and int(digits) <= MAX_STOCK:
stock_count = int(digits)
else:
logging.getLogger(__name__).warning("Out-of-range parsed stock; treating as unknown")
upc = None
for row in soup.select("table.table-striped tr"):
header = row.select_one("th")
value = row.select_one("td")
if header and value and header.get_text(strip=True) == "UPC":
upc = value.get_text(strip=True)
break
description = None
desc_header = soup.select_one("#product_description")
if desc_header:
sib = desc_header.find_next_sibling("p")
if sib:
description = sib.get_text(strip=True)
return {
"category": category,
"stock_count": stock_count,
"upc": upc,
"description": description,
}
def next_page_path(html: str) -> str | None:
"""Return the relative href of the 'next' pager link, if present."""
soup = _soup(html)
nxt = soup.select_one("li.next a")
return nxt.get("href") if nxt else None
"""Cleaning, typing, de-duplication and validation of parsed records.
Turns the loosely-typed dictionaries coming out of :mod:`scraper.parse` into a
tidy :class:`pandas.DataFrame` with well-defined column types, ready for
export. All steps are pure functions of their input so they are easy to test.
"""
from __future__ import annotations
import logging
import math
import re
import pandas as pd
from .fetch import is_allowed_url
# A price must be the *whole* string (after an optional leading currency
# supported symbol/code or the specific £ encoding repair): an optional sign, digits with optional
# thousands-grouping, and an optional decimal part. Anchoring the match rejects
# malformed values like "12.34oops" or "1e3" instead of silently extracting a
# partial number from them.
PRICE_RE = re.compile(
r"[+-]?(?:\d{1,3}(?:,\d{3})+|\d+)(?:\.\d+)?"
)
_CURRENCY_PREFIX_RE = re.compile(r"^(?:£|£|\$|€|GBP|USD|EUR)\s*")
log = logging.getLogger(__name__)
# Integers must remain exact in JavaScript JSON and spreadsheet exports.
MAX_STOCK = 2**53 - 1
COLUMNS = [
"title",
"category",
"price",
"rating",
"availability",
"in_stock",
"upc",
"thumbnail",
"detail_url",
"description",
]
def clean_price(value) -> float | None:
"""Extract a float from a messy price string such as ``£51.77``.
Returns ``None`` for anything that is not a clean, finite price: partial or
malformed numbers (``"12.34oops"``, ``"1e3"``) and overflowing magnitudes
are rejected rather than silently corrupting the value.
"""
if value is None or isinstance(value, bool):
return None
if isinstance(value, (int, float)):
f = float(value)
return round(f, 2) if math.isfinite(f) else None
# Drop a leading currency supported symbol/code or the specific £ encoding repair, then require the rest to
# be a complete number.
text = _CURRENCY_PREFIX_RE.sub("", str(value).strip())
match = PRICE_RE.fullmatch(text)
if not match:
return None
try:
f = float(match.group().replace(",", ""))
except ValueError:
return None
return round(f, 2) if math.isfinite(f) else None
# A book detail page on the source site embeds a truncated "read more"
# preview immediately followed by the full description (and a trailing
# "...more" marker). The preview repeats the opening of the full text, so the
# raw paragraph doubles its lead-in. Collapse that here.
_READ_MORE_RE = re.compile(r"\s*\.{2,}\s*more\s*$", re.I)
_WS_RE = re.compile(r"\s+")
def _norm_ws(text: str) -> str:
"""Collapse any whitespace run (incl. non-breaking spaces) to one space."""
return _WS_RE.sub(" ", text).strip()
def clean_description(value) -> str:
"""Remove the duplicated truncated preview and trailing read-more marker.
Idempotent: cleaning an already-clean description leaves it unchanged.
"""
if value is None:
return ""
text = str(value).strip()
if not text:
return text
lead = text[:40]
if _READ_MORE_RE.search(text) and len(lead) >= 20:
idx = text.find(lead, 1)
if idx > 0:
preview = text[:idx].rstrip()
cut = preview.rfind(" ") # drop the truncated final word
prefix = preview[:cut] if cut > 0 else preview
# Only collapse when the text after the repeat really continues the
# preview, so a genuine description can't be truncated by accident.
# Compare whitespace-insensitively: the preview often normalises
# non-breaking spaces that the full copy keeps.
if prefix and _norm_ws(text[idx:]).startswith(_norm_ws(prefix)):
text = text[idx:]
return _READ_MORE_RE.sub("", text).strip()
def _clean_url(value) -> str | None:
"""Keep a URL only if it is on the permitted crawl origin, else drop it."""
if value and is_allowed_url(str(value)):
return str(value)
return None
def rating_to_int(value) -> int | None:
"""Coerce a rating to an int in 0..5, or ``None`` if invalid."""
try:
ivalue = int(value)
except (TypeError, ValueError):
return None
return ivalue if 0 <= ivalue <= 5 else None
def availability_to_int(record: dict) -> int | None:
"""Keep unknown quantities nullable; stock status is a separate field."""
count = record.get("stock_count")
if isinstance(count, int) and not isinstance(count, bool):
if 0 <= count <= MAX_STOCK:
return count
elif isinstance(count, float) and math.isfinite(count) and 0 <= count <= MAX_STOCK and count.is_integer():
return int(count)
if count is not None:
log.warning("Invalid/out-of-range stock count; treating as unknown")
return None
def stock_total(values):
"""Sum with Python integers; unsupported export totals remain unknown."""
known = [int(value) for value in values if pd.notna(value)]
if not known:
return None
total = sum(known)
if not 0 <= total <= MAX_STOCK:
log.warning("Stock total exceeds exact export range; treating as unknown")
return None
return total
def build_dataframe(records: list[dict]) -> pd.DataFrame:
"""Build a typed, cleaned, de-duplicated and validated DataFrame."""
rows = []
for rec in records:
rows.append(
{
"title": (rec.get("title") or "").strip(),
"category": (rec.get("category") or "Uncategorised").strip(),
"price": clean_price(rec.get("price_text") or rec.get("price")),
"rating": rating_to_int(rec.get("rating")),
"availability": availability_to_int(rec),
"in_stock": bool(rec.get("in_stock")),
"upc": rec.get("upc"),
"thumbnail": _clean_url(rec.get("thumbnail")),
"detail_url": _clean_url(rec.get("detail_url")),
"description": clean_description(rec.get("description")),
}
)
df = pd.DataFrame(rows, columns=COLUMNS)
# Construct from original integers before pandas can round mixed int/None floats.
df["availability"] = pd.array([row["availability"] for row in rows], dtype="Int64")
df = _validate(df)
df = _dedupe(df)
df = df.sort_values(["category", "title"], kind="stable").reset_index(drop=True)
return df
def _validate(df: pd.DataFrame) -> pd.DataFrame:
"""Drop rows that fail minimal sanity checks and log nothing silently."""
before = len(df)
# A usable record needs a title and a finite, non-negative price.
df = df[df["title"].astype(bool)]
df = df[df["price"].notna() & (df["price"] >= 0) & (df["price"] != float("inf"))]
# Ratings must be within range; coerce out-of-range to <NA>.
df["rating"] = df["rating"].where(df["rating"].between(0, 5))
df["rating"] = df["rating"].astype("Int64")
df["availability"] = df["availability"].astype("Int64")
df.attrs["dropped_invalid"] = before - len(df)
return df
def _dedupe(df: pd.DataFrame) -> pd.DataFrame:
"""Remove duplicate books, keyed by UPC when present, else by title."""
before = len(df)
# Tag keys by their source so a UPC can never collide with a title that
# happens to equal it: a present UPC keys as ("upc", value), otherwise the
# record keys by ("title", value).
has_upc = df["upc"].notna() & df["upc"].astype(bool)
key = [
("upc", upc) if present else ("title", title)
for present, upc, title in zip(has_upc, df["upc"], df["title"])
]
key = pd.Series(key, index=df.index)
df = df.loc[~key.duplicated(keep="first")].reset_index(drop=True)
df.attrs["dropped_duplicates"] = before - len(df)
return df
def category_summary(df: pd.DataFrame) -> pd.DataFrame:
"""Aggregate the catalogue by category for the summary sheet / viewer."""
if df.empty:
return pd.DataFrame(
columns=["category", "books", "avg_price", "avg_rating", "total_stock", "unknown_stock"]
)
grouped = (
df.groupby("category")
.agg(
books=("title", "count"),
avg_price=("price", "mean"),
avg_rating=("rating", "mean"),
total_stock=("availability", stock_total),
unknown_stock=("availability", lambda x: int(x.isna().sum())),
)
.reset_index()
)
grouped["avg_price"] = grouped["avg_price"].round(2)
grouped["avg_rating"] = grouped["avg_rating"].round(2)
return grouped.sort_values("books", ascending=False).reset_index(drop=True)
"""Export a cleaned DataFrame to CSV, formatted Excel and the web viewer.
* :func:`to_csv` — plain UTF-8 CSV.
* :func:`to_excel` — a styled workbook with a frozen, filtered header row, a
per-category summary sheet and sensible column widths.
* :func:`to_viewer` — writes ``viewer/data.js`` embedding the data as JSON so
the self-contained viewer works straight off the ``file://`` protocol.
"""
from __future__ import annotations
import json
import logging
import re
from pathlib import Path
import pandas as pd
from openpyxl.styles import Alignment, Font, PatternFill
from openpyxl.utils import get_column_letter
from .transform import category_summary, stock_total
HEADER_FILL = PatternFill("solid", fgColor="1F2A44")
HEADER_FONT = Font(bold=True, color="FFFFFF")
TITLE_FONT = Font(bold=True, size=14, color="1F2A44")
# Characters that make a spreadsheet treat a cell as a live formula. A scraped
# title/description beginning with one of these must be neutralised before
# export, or opening the CSV/XLSX in Excel/LibreOffice would execute it
# (formula / CSV injection).
_FORMULA_PREFIXES = ("=", "+", "-", "@", "\t", "\r")
# Control characters, surrogates and the non-characters U+FFFE/U+FFFF are
# invalid in XML 1.0 and corrupt CSVs for strict consumers. Strip them from any
# exported text, whatever the target format.
_CONTROL_RE = re.compile(r"[\x00-\x08\x0b\x0c\x0e-\x1f\ud800-\udfff\ufffe\uffff]")
def _sanitize_cell(value):
"""Prefix a formula-leading text cell with an apostrophe so it stays text."""
if isinstance(value, str) and value[:1] in _FORMULA_PREFIXES:
return "'" + value
return value
def _csv_cell(value):
"""Strip invalid control chars, then neutralise formula-leading text."""
if isinstance(value, str):
value = _CONTROL_RE.sub("", value)
return _sanitize_cell(value)
def _excel_value(value, context="value"):
"""Prepare a value for ``openpyxl``: drop pandas NA, neutralise formulas."""
if pd.isna(value): # pd.NA / NaN / None -> blank cell (openpyxl rejects NA)
return None
if isinstance(value, str):
cleaned = _CONTROL_RE.sub("", value)
cleaned = _sanitize_cell(cleaned)
# Excel's limit is measured in UTF-16 code units.
bounded = cleaned.encode("utf-16-le")[:65534].decode("utf-16-le", errors="ignore")
if bounded != value:
logging.getLogger(__name__).warning("Sanitized/truncated spreadsheet text at %s", context)
return bounded
return value
def to_csv(df: pd.DataFrame, path: str | Path) -> Path:
path = Path(path)
path.parent.mkdir(parents=True, exist_ok=True)
df.apply(lambda col: col.map(_csv_cell)).to_csv(path, index=False)
return path
def _style_sheet(ws, df: pd.DataFrame, title: str) -> None:
"""Apply header styling, freeze panes, autofilter and column widths."""
ws.append([title])
ws.merge_cells(start_row=1, start_column=1, end_row=1, end_column=max(len(df.columns), 1))
ws.cell(row=1, column=1).font = TITLE_FONT
ws.append(list(df.columns))
header_row = ws.max_row
for col_idx, _ in enumerate(df.columns, start=1):
cell = ws.cell(row=header_row, column=col_idx)
cell.fill = HEADER_FILL
cell.font = HEADER_FONT
cell.alignment = Alignment(vertical="center")
for row_number, record in enumerate(df.itertuples(index=False), start=3):
ws.append([_excel_value(v, f"{ws.title}!{get_column_letter(i)}{row_number}")
for i, v in enumerate(record, start=1)])
# Freeze the header, add an autofilter over the data block.
ws.freeze_panes = ws.cell(row=header_row + 1, column=1)
last_col = get_column_letter(max(len(df.columns), 1))
ws.auto_filter.ref = f"A{header_row}:{last_col}{ws.max_row}"
# Column widths from the longest cell (title/description capped).
for col_idx, col_name in enumerate(df.columns, start=1):
longest = max(
[len(str(col_name))]
+ [len(str(v)) for v in df[col_name].head(200).tolist()]
)
ws.column_dimensions[get_column_letter(col_idx)].width = min(max(longest + 2, 10), 48)
def to_excel(df: pd.DataFrame, path: str | Path) -> Path:
from openpyxl import Workbook
path = Path(path)
path.parent.mkdir(parents=True, exist_ok=True)
wb = Workbook()
ws = wb.active
ws.title = "Books"
_style_sheet(ws, df, "python-data-scraper — Books (demo)")
summary = category_summary(df)
ws_summary = wb.create_sheet("Summary by category")
_style_sheet(ws_summary, summary, "Summary by category")
wb.save(path)
return path
def _records_for_viewer(df: pd.DataFrame) -> list[dict]:
out = df.copy()
out["rating"] = out["rating"].astype("object").where(out["rating"].notna(), None)
return json.loads(out.to_json(orient="records"))
def to_viewer(df: pd.DataFrame, path: str | Path, source: str = "live") -> Path:
"""Write ``viewer/data.js`` with the dataset and summary embedded as JSON."""
path = Path(path)
path.parent.mkdir(parents=True, exist_ok=True)
summary = category_summary(df)
payload = {
"generatedSource": source,
"count": int(len(df)),
"categories": json.loads(summary.to_json(orient="records")),
"stats": {
"totalBooks": int(len(df)),
"categories": int(df["category"].nunique()) if not df.empty else 0,
"avgPrice": round(float(df["price"].mean()), 2) if not df.empty else 0,
# No known ratings -> null (rendered as "Not rated"), never a
# misleading 0.00-star average.
"avgRating": round(float(df["rating"].dropna().mean()), 2)
if not df.empty and df["rating"].notna().any()
else None,
"totalStock": stock_total(df["availability"]),
"unknownStock": int(df["availability"].isna().sum()),
},
"books": _records_for_viewer(df),
}
data_js = "window.SCRAPER_DATA = " + json.dumps(payload, indent=2, ensure_ascii=False) + ";\n"
path.write_text(data_js, encoding="utf-8")
return path
"""Tests for the HTML parsing layer, driven off the saved fixtures.
Written with :mod:`unittest` so they run under either ``pytest`` or
``python -m unittest`` — pytest is not required.
"""
from __future__ import annotations
import unittest
from pathlib import Path
from scraper.parse import (
next_page_path,
parse_detail,
parse_listing,
rating_from_classes,
)
FIXTURES = Path(__file__).parent / "fixtures"
LISTING = FIXTURES / "catalogue" / "page-1.html"
class RatingTests(unittest.TestCase):
def test_word_ratings(self):
self.assertEqual(rating_from_classes(["star-rating", "Three"]), 3)
self.assertEqual(rating_from_classes(["star-rating", "five"]), 5)
def test_unknown_rating(self):
self.assertIsNone(rating_from_classes(["star-rating", "Eleven"]))
class ListingTests(unittest.TestCase):
@classmethod
def setUpClass(cls):
cls.html = LISTING.read_text(encoding="utf-8")
cls.base = "https://books.toscrape.com/catalogue/page-1.html"
cls.books = parse_listing(cls.html, base_url=cls.base)
def test_twenty_books(self):
self.assertEqual(len(self.books), 20)
def test_fields_present(self):
first = self.books[0]
self.assertTrue(first["title"])
self.assertRegex(first["price_text"], r"\d")
self.assertIn(first["rating"], range(0, 6))
self.assertTrue(first["detail_url"].startswith("https://books.toscrape.com/catalogue/"))
def test_detail_urls_resolved(self):
# Every book should link to a detail page under /catalogue/.
self.assertTrue(all("/catalogue/" in b["detail_url"] for b in self.books))
def test_thumbnails_absolute(self):
self.assertTrue(all(b["thumbnail"].startswith("http") for b in self.books))
class DetailTests(unittest.TestCase):
def test_parse_a_detail_fixture(self):
detail = next((FIXTURES / "catalogue").glob("*/index.html"))
data = parse_detail(detail.read_text(encoding="utf-8"))
self.assertTrue(data["category"])
self.assertIsInstance(data["stock_count"], int)
self.assertTrue(data["upc"])
def test_next_page_link(self):
html = LISTING.read_text(encoding="utf-8")
# page-1 of a multi-page catalogue has a 'next' link.
self.assertEqual(next_page_path(html), "page-2.html")
if __name__ == "__main__":
unittest.main()
"""Tests for the cleaning / typing / validation layer."""
from __future__ import annotations
import unittest
from scraper.transform import (
availability_to_int,
build_dataframe,
category_summary,
clean_price,
rating_to_int,
)
class ScalarCleaningTests(unittest.TestCase):
def test_clean_price_strips_currency(self):
self.assertEqual(clean_price("£51.77"), 51.77)
self.assertEqual(clean_price("£13.99"), 13.99)
self.assertEqual(clean_price(10), 10.0)
self.assertIsNone(clean_price("free"))
self.assertIsNone(clean_price(None))
def test_rating_to_int_bounds(self):
self.assertEqual(rating_to_int(3), 3)
self.assertEqual(rating_to_int("5"), 5)
self.assertIsNone(rating_to_int(9))
self.assertIsNone(rating_to_int("x"))
def test_availability_prefers_stock_count(self):
self.assertEqual(availability_to_int({"stock_count": 22, "in_stock": True}), 22)
self.assertIsNone(availability_to_int({"in_stock": True}))
self.assertIsNone(availability_to_int({"in_stock": False}))
SAMPLE = [
{"title": "Alpha", "price_text": "£10.00", "rating": 4, "in_stock": True,
"stock_count": 5, "category": "Travel", "upc": "aaa"},
{"title": "Beta", "price_text": "£20.00", "rating": 2, "in_stock": True,
"stock_count": 3, "category": "Travel", "upc": "bbb"},
# Duplicate of Alpha by UPC — should be dropped.
{"title": "Alpha (reprint)", "price_text": "£11.00", "rating": 4, "in_stock": True,
"stock_count": 1, "category": "Travel", "upc": "aaa"},
# Invalid: no price — should be dropped.
{"title": "Gamma", "price_text": "n/a", "rating": 9, "in_stock": False,
"category": "Poetry", "upc": "ccc"},
]
class DataFrameTests(unittest.TestCase):
def setUp(self):
self.df = build_dataframe(SAMPLE)
def test_dedup_and_validation(self):
titles = set(self.df["title"])
self.assertIn("Alpha", titles)
self.assertIn("Beta", titles)
self.assertNotIn("Alpha (reprint)", titles) # duplicate UPC
self.assertNotIn("Gamma", titles) # invalid price
self.assertEqual(self.df.attrs["dropped_duplicates"], 1)
self.assertEqual(self.df.attrs["dropped_invalid"], 1)
def test_types(self):
self.assertEqual(self.df["price"].dtype.kind, "f")
self.assertEqual(str(self.df["rating"].dtype), "Int64")
self.assertEqual(self.df["availability"].dtype.kind, "i")
def test_category_summary(self):
summary = category_summary(self.df)
travel = summary[summary["category"] == "Travel"].iloc[0]
self.assertEqual(travel["books"], 2)
self.assertEqual(travel["total_stock"], 8)
self.assertAlmostEqual(travel["avg_price"], 15.0)
if __name__ == "__main__":
unittest.main()