Quick summary
Summarize this blog with AI
A useful automation has one clear input, a validated transformation, and an observable output. This tutorial fetches a small HTML price table, converts integer prices into a pandas pivot, previews a CSV destination, and creates the output without replacing an existing file. It never moves arbitrary files or submits a remote form.
Tested on macOS with Python 3.12.14, Beautiful Soup 4.15.0, Requests 2.34.2, and pandas 2.2.3/3.0.6. Network tests use a local fixture. File publication uses a same-directory hard link, tested on the local filesystem; unsupported filesystems raise an error. Use this as a controlled workflow, with permission for any real source you later configure.
1. Fetch only the configured bounded page
import requests
from urllib.parse import urlsplit
def fetch_html(url, max_bytes=65536):
parsed = urlsplit(url)
local = parsed.scheme == "http" and parsed.hostname == "127.0.0.1"
if (parsed.scheme != "https" and not local) or not parsed.hostname:
raise ValueError("use HTTPS or the local fixture server")
if parsed.username is not None or parsed.password is not None:
raise ValueError("credentials must not be embedded in the URL")
if type(max_bytes) is not int or max_bytes < 1:
raise ValueError("max_bytes must be a positive integer")
with requests.Session() as session:
session.trust_env = False
with session.get(url, timeout=(5, 15), stream=True,
allow_redirects=False,
headers={"Accept-Encoding": "identity"}) as response:
response.raise_for_status()
if response.status_code != 200:
raise ValueError("expected HTTP 200 without redirects")
media = response.headers.get("Content-Type", "").split(";", 1)[0].strip().lower()
if media != "text/html":
raise ValueError("expected HTML")
if response.headers.get("Content-Encoding", "identity").lower() != "identity":
raise ValueError("compressed content is not accepted")
content = bytearray()
for chunk in response.iter_content(chunk_size=4096):
if len(content) + len(chunk) > max_bytes:
raise ValueError("HTML exceeds byte limit")
content.extend(chunk)
return content.decode("utf-8")
This helper has explicit connect/read timeouts, status checks, a 64 KiB byte cap, no redirects, and UTF-8 decoding. It does not retry, execute JavaScript, follow discovered URLs, or use environment credentials. Its timeout is a socket-inactivity limit rather than a whole-job deadline. Keep targets application-owned: accepting arbitrary external URLs would require a separate SSRF and access-control design.
2. Validate the source rows before reshaping
from bs4 import BeautifulSoup
import pandas as pd
import re
PRICES = """<table id="prices">
<tr data-sku="book" data-store="east"><td class="cents">1200</td></tr>
<tr data-sku="book" data-store="west"><td class="cents">1300</td></tr>
<tr data-sku="game" data-store="east"><td class="cents">2500</td></tr>
</table>"""
def price_frame(html):
soup = BeautifulSoup(html, "html.parser")
table = soup.select_one("table#prices")
if table is None:
raise ValueError("price table missing")
nodes = table.select("tr")
if not 1 <= len(nodes) <= 100:
raise ValueError("expected one to 100 price rows")
rows, seen = [], set()
for node in nodes:
sku = node.get("data-sku", "")
store = node.get("data-store", "")
cell = node.select_one("td.cents")
raw = cell.get_text(strip=True) if cell else ""
if not re.fullmatch(r"[a-z0-9_-]{1,30}", sku) or store not in {"east", "west"}:
raise ValueError("invalid SKU or store")
if not re.fullmatch(r"[0-9]{1,8}", raw):
raise ValueError("price must be integer cents")
if (sku, store) in seen:
raise ValueError("duplicate SKU/store price")
seen.add((sku, store))
rows.append({"sku": sku, "store": store, "cents": int(raw)})
return pd.DataFrame(rows)
prices = price_frame(PRICES)
wide = prices.pivot(index="sku", columns="store", values="cents")
wide = wide.reindex(columns=["east", "west"]).sort_index().astype("Int64")
print(wide.to_csv(index=True, na_rep="UNKNOWN", lineterminator="\n"), end="")
Expected output:
sku,east,west
book,1200,1300
game,2500,UNKNOWN
pivot requires unique SKU/store cells; the parser rejects duplicates instead of choosing an unexplained average or maximum. Missing west/game remains unknown, not zero. Prices are integer cents in one agreed currency. A human-formatted value such as $12.00 is rejected because accepting it would require a separate locale/currency parser.
3. Preview and create one output atomically
The caller supplies a trusted destination path in an existing directory. The function defaults to a dry run. It never overwrites an existing destination; a repeated scheduled run needs an explicit filename or reconciliation policy.
from pathlib import Path
from tempfile import NamedTemporaryFile, TemporaryDirectory
import os
def publish_csv(frame, destination, dry_run=True):
destination = Path(destination)
if not destination.parent.is_dir():
raise FileNotFoundError("destination directory must already exist")
if destination.exists() or destination.is_symlink():
raise FileExistsError("destination already exists")
if dry_run:
return "would create " + destination.name
temporary = None
try:
with NamedTemporaryFile(mode="w", encoding="utf-8", newline="",
dir=destination.parent, prefix=".prices-", delete=False) as handle:
temporary = Path(handle.name)
frame.to_csv(handle, index=True, na_rep="UNKNOWN", lineterminator="\n")
handle.flush()
os.fsync(handle.fileno())
os.link(temporary, destination)
return "created " + destination.name
finally:
if temporary is not None:
temporary.unlink(missing_ok=True)
with TemporaryDirectory() as directory:
destination = Path(directory) / "prices.csv"
print(publish_csv(wide, destination))
print(destination.exists())
print(publish_csv(wide, destination, dry_run=False))
print(destination.read_text(encoding="utf-8").splitlines())
try:
publish_csv(wide, destination, dry_run=False)
except FileExistsError:
print("existing output preserved")
Expected output:
would create prices.csv
False
created prices.csv
['sku,east,west', 'book,1200,1300', 'game,2500,UNKNOWN']
existing output preserved
The complete temporary file is linked into place only after writing succeeds. Unlike a prior exists check alone, os.link fails if another writer creates the destination in the meantime. The staging file is removed on either success or failure. This relies on hard-link support and a caller-controlled directory; it is not a promise of directory-metadata durability after sudden power loss, or protection against a malicious local directory owner.
4. Run collection through output as one job
def run_collection(url, destination, dry_run=True):
rows = price_frame(fetch_html(url))
pivot = rows.pivot(index="sku", columns="store", values="cents")
pivot = pivot.reindex(columns=["east", "west"]).sort_index().astype("Int64")
return publish_csv(pivot, destination, dry_run=dry_run)
Retrieval or validation failure propagates before publication. A dry run still fetches and validates the source, but creates no output file. Supply only a permitted configured URL; the parser is demonstrated above on owned HTML, and the complete run was tested against a loopback fixture serving that same HTML. No third-party site, paid API, or authenticated workflow was executed.
5. Schedule only after defining failure and repetition behavior
A scheduler should run the pinned environment, use an absolute configuration/output path, prevent unintended overlapping runs, and record success only after publication succeeds. Do not suppress an HTTP error or layout change and publish an empty report. Choose whether a repeated run is a new dated artifact, an explicitly reviewed replacement, or an idempotent no-op based on content identity; this example deliberately rejects collisions.
Input values here are strictly constrained SKU labels, known store names, and integer amounts. If you later export arbitrary scraped text to a spreadsheet, handle formula-like text according to the target application's rules; CSV quote escaping alone is not a universal spreadsheet-formula defense.
For more parser detail, see the complete Beautiful Soup guide. References: Requests timeouts, pandas pivot, os.link, and temporary files.
