#!/usr/bin/env python3
"""r4 - read a product/listing page into a clean CSV that carries its own provenance.
Read-date of every fact and figure in this file and its docs: 2026-11-10Z.
WHAT IT DOES
Input a local file (HTML, or browser_collect-style JSON, or CSV), or - with
the --i-have-checked-robots-and-am-authorized flag - an http(s) URL.
Output CSV on stdout (or --out FILE) with exactly these columns:
title, price_raw, price_cents, currency, url, read_date, source_hash
read_date is stamped when the row is READ, not when this file was written,
in ISO 8601 UTC with a trailing Z.
source_hash is sha256 over the RAW BYTES of the source row the row came from
(the exact HTML slice for HTML input; json.dumps(..., sort_keys=
True).encode("utf-8") for JSON input). Re-read the same page and
the same hash comes back; change one character of the row and it
does not. That is what makes a stale list provably stale.
THE THREE RULES IT ENFORCES (these are the reason the recipe exists)
1. robots.txt is checked before any live fetch. The check is a pure function
over the robots text (parse_robots / RobotsPolicy.allowed) with a real
User-agent group match and longest-path-match with Allow winning a tie, so
it can be - and is - unit-tested without a network.
2. It REFUSES a live URL unless the caller passes
--i-have-checked-robots-and-am-authorized, and the error says so.
Refusal happens before any socket is opened.
3. It never stores more than --max-rows rows (default 500). The count it
dropped is printed, never hidden.
OFFLINE BY DEFAULT
With a file path there is no network code path at all: nothing in urllib is
imported at call time for the offline branch, and tests/monkeypatches
urllib.request.urlopen to prove no fetch happens.
python3 scrape_listings.py --input fixtures/listings.html --format html
python3 scrape_listings.py --input fixtures/listings.json --top 5 --out rows.csv
Exit codes: 0 ok | 2 usage, or live refused without the flag | 3 robots.txt denied
4 input could not be read | 5 rows were read but none could be kept
"""
from __future__ import annotations
import argparse
import csv
import hashlib
import json
import re
import sys
import time
from datetime import datetime, timezone
from decimal import Decimal, InvalidOperation
from html.parser import HTMLParser
from typing import Callable, Iterable, List, Optional, Sequence, Tuple
from urllib.parse import urljoin, urlsplit
# Read-date stamp of this build; also what --version prints.
READ_DATE = "2026-11-10"
USER_AGENT = "back-office-autopilot-r4/1.0 (listing reader)"
COLUMNS = ("title", "price_raw", "price_cents", "currency", "url",
"read_date", "source_hash")
#: The one separator a raw row's fields are joined with before hashing.
FIELD_SEP = "\x1f"
# --------------------------------------------------------------------------- #
# money: strings -> integer cents, never floats, never a guess
# --------------------------------------------------------------------------- #
CURRENCY_SIGNS = {
"$": "USD", "US$": "USD", "US$ ": "USD", "CA$": "CAD", "C$": "CAD",
"A$": "AUD", "AU$": "AUD", "\u20ac": "EUR", "\u00a3": "GBP",
"\u00a5": "JPY", "Kr": "SEK", "z\u0142": "PLN",
}
ISO_CODES = ("USD", "CAD", "EUR", "GBP", "JPY", "AUD", "CHF", "SEK", "PLN",
"NZD", "SGD", "INR", "MXN", "BRL")
def parse_price(raw: str,
default_currency: str = "USD") -> Optional[Tuple[str, int]]:
"""'CA$1,234.56' -> ('CAD', 123456). None when there is no number to trust.
Currency comes from a sign or an ISO code IN THE STRING; only when the string
carries neither does default_currency apply. Separators are resolved by which
one comes last, so '1.234,56' (European) and '1,234.56' (Anglo) both work and
'1,50' is read as 1.50, not 150. A figure with more than two decimal places
(a per-unit rate) is not silently rounded: it is refused here and reported by
the caller, because a fabricated cent on a provenance row is worse than a gap.
"""
if raw is None:
return None
s = str(raw).strip()
if not s:
return None
currency = ""
# a 3-letter ISO code anywhere: "1 234,56 EUR"
m = re.search(r"\b(%s)\b" % "|".join(ISO_CODES), s, flags=re.IGNORECASE)
if m:
currency = m.group(1).upper()
else:
for sign, code in sorted(CURRENCY_SIGNS.items(),
key=lambda kv: -len(kv[0])):
if sign in s:
currency = code
break
if not currency:
currency = default_currency
# strip everything that is not a digit or a separator
digits = re.sub(r"[^0-9.,]", "", s)
if not digits or not re.search(r"\d", digits):
return None
last_comma = digits.rfind(",")
last_dot = digits.rfind(".")
if last_comma >= 0 and last_dot >= 0:
if last_comma > last_dot: # 1.234,56
digits = digits.replace(".", "").replace(",", ".")
else: # 1,234.56
digits = digits.replace(",", "")
elif last_comma >= 0:
tail = len(digits) - last_comma - 1
if tail == 3 and re.match(r"^\d{1,3}(,\d{3})+$", digits):
digits = digits.replace(",", "") # 1,234 = one thousand two hundred
else:
digits = digits.replace(",", ".") # 1,50 = one fifty
# a bare '.' with 3 digits after it stays a decimal point ('6.500' -> 6.5)
if not re.match(r"^\d+(\.\d+)?$", digits):
return None
try:
value = Decimal(digits)
except InvalidOperation:
return None
if value < 0:
return None
if value == 0:
# A 0.00 listing is a real observation but a useless price; keep it out
# of a file whose every row is meant to be quotable.
return None
if value != value.quantize(Decimal("0.01")):
return None
return currency, int((value * 100).to_integral_value())
def iso_utc(now: Optional[float] = None) -> str:
"""ISO 8601 in UTC with a trailing Z. `now` is epoch seconds (injectable)."""
dt = (datetime.fromtimestamp(now, tz=timezone.utc) if now is not None
else datetime.now(timezone.utc))
return dt.strftime("%Y-%m-%dT%H:%M:%SZ")
# --------------------------------------------------------------------------- #
# robots.txt: a pure function over text
# --------------------------------------------------------------------------- #
class RobotsPolicy:
"""Parsed robots.txt. `allowed(url, user_agent)` is a pure lookup.
Matching follows the RFC 9309 shape: pick the most specific User-agent group
that names us (a group whose name is a substring of our UA beats the '*'
group), then the LONGEST path rule that matches wins, and an Allow rule wins
a tie against a Disallow rule of the same length. An empty `Disallow:` means
"everything allowed" for that group. No group matching us means no rule
applies to us, which is allowed.
"""
def __init__(self, groups: dict, crawl_delays: dict, text: str = ""):
self.groups = groups # ua-name -> [(path, allow), ...]
self.crawl_delays = crawl_delays # ua-name -> seconds (float)
self.text = text
def group_for(self, user_agent: str) -> Tuple[str, list]:
ua = user_agent.lower()
specific = [name for name in self.groups
if name != "*" and name and name in ua]
if specific:
best = max(specific, key=len)
return best, self.groups[best]
if "*" in self.groups:
return "*", self.groups["*"]
return "", []
def crawl_delay(self, user_agent: str) -> Optional[float]:
name, _ = self.group_for(user_agent)
return self.crawl_delays.get(name)
def allowed(self, url: str, user_agent: str = USER_AGENT) -> bool:
path = urlsplit(url).path or "/"
query = urlsplit(url).query
target = path + ("?" + query if query else "")
name, rules = self.group_for(user_agent)
best_len, best_allow = -1, True
for rule_path, allow in rules:
if rule_path == "":
continue
if target.startswith(rule_path) or path.startswith(rule_path):
n = len(rule_path)
if n > best_len or (n == best_len and allow):
best_len, best_allow = n, allow
return best_allow
def why(self, url: str, user_agent: str = USER_AGENT) -> str:
name, rules = self.group_for(user_agent)
if not rules:
return "no rules for user-agent %r in robots.txt" % user_agent
path = urlsplit(url).path or "/"
hits = [(len(p), p, a) for p, a in rules if p and path.startswith(p)]
if not hits:
return "group %r: no rule matches %s" % (name, path)
n = max(h[0] for h in hits)
winner = [h for h in hits if h[0] == n]
allow = any(h[2] for h in winner)
return ("group %r: %s '%s' (length %d) decides %s for %s"
% (name, "Allow" if allow else "Disallow",
winner[0][1], n, "ALLOW" if allow else "DENY", path))
def parse_robots(text: str) -> RobotsPolicy:
"""Parse robots.txt text. Tolerant of CRLF, comments, missing colons, junk.
Anything that is not a rule is ignored rather than fatal - a robots.txt that
also lists a Sitemap: or a Crawl-delay: is the normal case, not an error.
"""
groups: dict = {}
delays: dict = {}
current: List[str] = []
last_was_ua = False
for line in (text or "").replace("\r\n", "\n").replace("\r", "\n").split("\n"):
line = line.split("#", 1)[0].strip()
if not line:
last_was_ua = False
continue
if ":" not in line:
continue
field, _, value = line.partition(":")
field = field.strip().lower()
value = value.strip()
if field == "user-agent":
# Consecutive User-agent lines name ONE group ("User-agent: a" then
# "User-agent: b" both mean the rules below). A User-agent line that
# comes after any rule starts a NEW group - without this, the rules
# of the group below would be attached to '*' as well, and a site
# that bans one bot would look like it bans everyone (this was one
# of the two bugs the r4 suite caught on 2026-11-10).
if not last_was_ua:
current = []
current.append(value.lower())
last_was_ua = True
elif field in ("disallow", "allow"):
if not current:
current = ["*"]
allow = field == "allow"
for name in current:
groups.setdefault(name, []).append((value, allow))
if value == "" and field == "disallow":
# 'Disallow:' with no value = allow everything for this group.
for name in current:
groups.setdefault(name, []).append(("", True))
elif field == "crawl-delay":
try:
for name in current:
delays[name] = float(value)
except ValueError:
pass
# sitemap/Host/other fields: not our business here
if field not in ("user-agent", "disallow", "allow", "crawl-delay"):
current = [] if field in ("host",) else current
return RobotsPolicy(groups, delays, text or "")
def robots_allowed(robots_text: str, url: str,
user_agent: str = USER_AGENT) -> Tuple[bool, str]:
"""Pure: (allowed, reason). Empty/absent robots text means nothing was said."""
if not (robots_text or "").strip():
return True, "robots.txt was empty or absent: no rule was stated"
policy = parse_robots(robots_text)
return policy.allowed(url, user_agent), policy.why(url, user_agent)
# --------------------------------------------------------------------------- #
# the rate limiter: 1.0 s per request, unit-testable with an injected clock
# --------------------------------------------------------------------------- #
class RateLimiter:
"""At most one request per `min_interval_s`. clock and sleep are injectable.
delay_for(now) is the pure part: how long a request arriving at `now` must
wait. acquire() is the part that actually sleeps. Tests drive the clock by
hand, so the 1.0 s floor is asserted, not hoped for.
"""
def __init__(self, min_interval_s: float = 1.0,
clock: Callable[[], float] = time.monotonic,
sleep: Callable[[float], None] = time.sleep):
if min_interval_s < 0:
raise ValueError("min_interval_s cannot be negative")
self.min_interval_s = float(min_interval_s)
self.clock = clock
self.sleep = sleep
self.last_request_at: Optional[float] = None
self.waited_total = 0.0
self.requests = 0
def delay_for(self, now: float) -> float:
if self.last_request_at is None:
return 0.0
gap = now - self.last_request_at
return self.min_interval_s - gap if gap < self.min_interval_s else 0.0
def acquire(self, url: str = "") -> float:
"""Wait the required gap, record the request, return the seconds waited."""
now = self.clock()
wait = self.delay_for(now)
if wait > 0:
self.sleep(wait)
self.waited_total += wait
self.last_request_at = self.clock()
self.requests += 1
return wait
# --------------------------------------------------------------------------- #
# extraction: one RawRow per source row, with its raw bytes
# --------------------------------------------------------------------------- #
class RawRow:
__slots__ = ("title", "price_raw", "url", "raw_bytes")
def __init__(self, title: str, price_raw: str, url: str, raw: bytes):
self.title = title
self.price_raw = price_raw
self.url = url
self.raw_bytes = raw
def source_hash(self) -> str:
return hashlib.sha256(self.raw_bytes).hexdigest()
def _clean(text: str) -> str:
return re.sub(r"\s+", " ", (text or "").replace("\xa0", " ")).strip()
class _ListingHTMLParser(HTMLParser):
"""Collect items of one class, and keep the exact source slice of each.
Structure it expects (documented in SETUP.md, and in the shipped fixture):
title_class / price_class are class-name fragments; if a class is not found
in an item, the item's anchor text / first money-looking text is used, and
the row still carries the raw bytes it was read from.
"""
def __init__(self, item_class="listing", title_class="title",
price_class="price", item_tag="div", base_url="",
text="", max_rows=500):
super().__init__(convert_charrefs=True)
self.item_class, self.title_class = item_class, title_class
self.price_class = price_class
self.item_tag, self.base_url, self.text = item_tag, base_url, text
self.max_rows = max_rows
self.lines = text.splitlines(keepends=True)
self.prefixes = [0] * (len(self.lines) + 1)
for i, ln in enumerate(self.lines):
self.prefixes[i + 1] = self.prefixes[i] + len(ln)
self.rows: List[RawRow] = []
self.stopped_early = False
self._depth = 0 # nesting of item elements (0 = outside)
self._field_depth: dict = {}
self._buf: dict = {}
self._item_start = -1
self._href = ""
# -- offsets -- #
def _abs(self) -> int:
line, col = self.getpos()
return self.prefixes[line - 1] + col
#: HTML void elements never get an end tag, so they must not take depth.
VOID = frozenset(("area", "base", "br", "col", "embed", "hr", "img",
"input", "link", "meta", "param", "source", "track",
"wbr"))
def _raw_slice(self, endtag_abs: int) -> bytes:
"""THIS item's own source bytes: from its '<' to the '>' of its .
This is what source_hash is taken over, so the hash belongs to the row
and to nothing else - not to the page, not to the start tag alone.
"""
idx = self.text.find(">", endtag_abs)
end = len(self.text) if idx < 0 else idx + 1
return self.text[self._item_start:end].encode("utf-8")
@staticmethod
def _has_class(attrs: Sequence[Tuple[Optional[str], Optional[str]]],
name: str) -> bool:
for key, value in attrs:
if key == "class" and value and name in value.split():
return True
return False
@staticmethod
def _attr(attrs, name):
for key, value in attrs:
if key == name:
return value or ""
return ""
# -- handlers -- #
def handle_starttag(self, tag, attrs):
if self.item_class and tag == self.item_tag and \
self._has_class(attrs, self.item_class):
if self._depth == 0:
self._depth = 1
self._item_start = self._abs()
self._buf = {"title": "", "price": "", "href": ""}
self._field_depth = {}
return
self._depth += 1 # nested item: folded into its parent
return
if self._depth == 0:
return
void = tag in self.VOID
if not void:
self._depth += 1
if not self._href and tag == "a":
href = self._attr(attrs, "href")
if href:
self._href = href
if tag == "img":
src = self._attr(attrs, "src")
if not self._href and src:
self._href = src
if not void:
if self.title_class and self._has_class(attrs, self.title_class):
self._field_depth.setdefault("title", self._depth)
if self.price_class and self._has_class(attrs, self.price_class):
self._field_depth.setdefault("price", self._depth)
def handle_startendtag(self, tag, attrs):
self.handle_starttag(tag, attrs)
if tag not in self.VOID and self._depth:
self.handle_endtag(tag)
def handle_data(self, data):
if self._depth == 0:
return
for field, depth in self._field_depth.items():
if self._depth >= depth:
self._buf[field] += data
def handle_endtag(self, tag):
if self._depth == 0:
return
if self._depth == 1 and self.item_class and tag == self.item_tag:
raw = self._raw_slice(self._abs())
title = _clean(self._buf.get("title", ""))
price = _clean(self._buf.get("price", ""))
href = self._href
url = urljoin(self.base_url, href) if href else ""
if not title:
title = _clean(href)
if title or price:
if len(self.rows) < self.max_rows:
self.rows.append(RawRow(title, price, url, raw))
else:
self.stopped_early = True
self._depth = 0
self._buf = {}
self._href = ""
return
for field in list(self._field_depth):
# this field's element just closed: stop collecting it, keep its text
if self._field_depth[field] == self._depth:
del self._field_depth[field]
self._depth -= 1
def read_html_bytes(data: bytes, *, base_url: str = "", item_class="listing",
title_class="title", price_class="price", item_tag="div",
max_rows: int = 500) -> Tuple[List[RawRow], bool]:
text = data.decode("utf-8", errors="replace")
p = _ListingHTMLParser(item_class=item_class, title_class=title_class,
price_class=price_class, item_tag=item_tag,
base_url=base_url, text=text, max_rows=max_rows)
p.feed(text)
p.close()
return p.rows, p.stopped_early
def read_json_bytes(data: bytes, *, title_field="title", price_field="price",
url_fields=("url", "href", "link"),
max_rows: int = 500) -> Tuple[List[RawRow], bool]:
"""browser_collect-style JSON: [{href, fields:{title, price}}, ...] or flat."""
try:
payload = json.loads(data.decode("utf-8", errors="replace"))
except (json.JSONDecodeError, UnicodeDecodeError) as exc:
raise ValueError("JSON input could not be parsed: %s" % exc)
if isinstance(payload, dict):
for key in ("items", "rows", "data", "results"):
if isinstance(payload.get(key), list):
payload = payload[key]
break
if not isinstance(payload, list):
raise ValueError("JSON input is not a list of rows (nor an object with "
"an items/rows/data/results list)")
rows, stopped = [], False
for item in payload:
if not isinstance(item, dict):
continue
raw = json.dumps(item, sort_keys=True, ensure_ascii=False).encode("utf-8")
def pick(*names):
for name in names:
if name in item and item[name] is not None:
return str(item[name])
fields = item.get("fields")
if isinstance(fields, dict) and name in fields \
and fields[name] is not None:
return str(fields[name])
return ""
title = pick(title_field)
price = pick(price_field)
url = pick(*url_fields)
if not title and "text" in item:
title = _clean(str(item["text"]))[:200]
if not (title or price):
continue
if len(rows) < max_rows:
rows.append(RawRow(_clean(title), _clean(price), url.strip(), raw))
else:
stopped = True
return rows, stopped
def read_csv_bytes(data: bytes, *, title_field="title", price_field="price",
url_field="url", max_rows: int = 500
) -> Tuple[List[RawRow], bool]:
text = data.decode("utf-8", errors="replace")
reader = csv.DictReader(text.splitlines())
rows, stopped = [], False
for item in reader:
raw = json.dumps({k: v for k, v in item.items() if k is not None},
sort_keys=True, ensure_ascii=False).encode("utf-8")
title = _clean((item.get(title_field) or "").strip())
price = _clean((item.get(price_field) or "").strip())
url = (item.get(url_field) or "").strip()
if not (title or price):
continue
if len(rows) < max_rows:
rows.append(RawRow(title, price, url, raw))
else:
stopped = True
return rows, stopped
# --------------------------------------------------------------------------- #
# raw rows -> provenance rows
# --------------------------------------------------------------------------- #
def to_output_rows(raw_rows: Iterable[RawRow], *, default_currency: str = "USD",
now: Optional[float] = None) -> Tuple[List[dict], List[str]]:
"""Every kept row has all seven columns non-empty; rejects are named, not dropped."""
out, rejected = [], []
stamp = iso_utc(now)
for row in raw_rows:
parsed = parse_price(row.price_raw, default_currency)
if parsed is None:
rejected.append("no usable price in %r (title %r)"
% (row.price_raw, row.title[:60]))
continue
currency, cents = parsed
if not row.title:
rejected.append("empty title for url %r" % row.url)
continue
if not row.url:
rejected.append("no url on %r" % row.title[:60])
continue
out.append({
"title": row.title,
"price_raw": row.price_raw,
"price_cents": str(cents),
"currency": currency,
"url": row.url,
"read_date": stamp,
"source_hash": row.source_hash(),
})
return out, rejected
def write_csv(rows: List[dict], path: str) -> int:
if path in ("", "-"):
writer = csv.DictWriter(sys.stdout, fieldnames=list(COLUMNS))
writer.writeheader()
for row in rows:
writer.writerow(row)
return len(rows)
with open(path, "w", newline="", encoding="utf-8") as fh:
writer = csv.DictWriter(fh, fieldnames=list(COLUMNS))
writer.writeheader()
for row in rows:
writer.writerow(row)
return len(rows)
# --------------------------------------------------------------------------- #
# the live path (never taken by tests; the flag is the gate)
# --------------------------------------------------------------------------- #
LIVE_FLAG = "--i-have-checked-robots-and-am-authorized"
REFUSAL = (
"ERROR: refusing to fetch %s live.\n"
" A live fetch touches another machine, and it is your judgement - not this\n"
" script's - that the site permits it. Before you may pass the flag:\n"
" 1. read %s/robots.txt and confirm this path is not disallowed;\n"
" 2. read that site's Terms of Service about automated access;\n"
" 3. keep a request interval of at least %ss (this script enforces it);\n"
" 4. take nothing that identifies a person.\n"
" When all four are true for THIS site, re-run with %s.\n"
" Reading a local file instead needs no flag: --input some/file.html"
% ("%(url)s", "%(host)s", 1.0, LIVE_FLAG))
def host_root(url: str) -> str:
parts = urlsplit(url)
return "%s://%s" % (parts.scheme, parts.netloc)
def live_fetch(url: str, *, limiter: RateLimiter, opener=None,
user_agent: str = USER_AGENT,
clock: Callable[[], float] = time.monotonic) -> Tuple[bytes, str]:
"""robots first, then the page, rate-limited. Returns (bytes, robots_text)."""
if opener is None: # pragma: no cover
import urllib.request
def opener(target, ua):
req = urllib.request.Request(target, headers={"User-Agent": ua})
with urllib.request.urlopen(req, timeout=30) as resp:
return resp.read()
root = host_root(url)
try:
robots_text = opener(root + "/robots.txt", user_agent).decode(
"utf-8", errors="replace")
except Exception as exc: # noqa: BLE001
robots_text = ""
print("note: robots.txt at %s/robots.txt could not be read (%s); the "
"ToS and your own eyes are then the only check you have."
% (root, exc), file=sys.stderr)
allowed, reason = robots_allowed(robots_text, url, user_agent)
if not allowed:
raise PermissionError("robots.txt: %s -> %s" % (url, reason))
delay = limiter.min_interval_s
policy = parse_robots(robots_text)
cd = policy.crawl_delay(user_agent)
if cd and cd > delay:
delay = cd
limiter.min_interval_s = delay
limiter.acquire(url)
print("robots: %s\nfetch: %s (Crawl-delay floor %.2fs)"
% (reason, url, delay), file=sys.stderr)
return opener(url, user_agent), robots_text
# --------------------------------------------------------------------------- #
# CLI
# --------------------------------------------------------------------------- #
def build_parser() -> argparse.ArgumentParser:
p = argparse.ArgumentParser(
prog="scrape_listings.py",
description=("Read listings into CSV with per-row provenance "
"(price_cents, read_date, sha256 of the raw row bytes)."),
epilog=("A live http(s) --input is refused unless you pass %s. "
"Local files need no flag and make no request." % LIVE_FLAG))
p.add_argument("-i", "--input", required=True,
help="file path, '-', or an http(s) URL (live: flag required)")
p.add_argument("-f", "--format", choices=("html", "json", "csv"),
default=None, help="input format (default: guessed)")
p.add_argument("-o", "--out", default="-",
help="CSV destination (default: stdout)")
p.add_argument("--max-rows", type=int, default=500,
help="hard cap on rows stored (default 500)")
p.add_argument("--limit", type=float, default=1.0,
help="min seconds between live requests (default 1.0, floor)")
p.add_argument("--ua", default=USER_AGENT, help="User-Agent sent live")
p.add_argument("--base-url", default="",
help="join relative hrefs against this (a local file's URL)")
p.add_argument("--default-currency", default="USD",
help="only used when a price carries no sign or code (USD)")
p.add_argument("--item-class", default="listing")
p.add_argument("--title-class", default="title")
p.add_argument("--price-class", default="price")
p.add_argument("--item-tag", default="div")
p.add_argument("--title-field", default="title")
p.add_argument("--price-field", default="price")
p.add_argument("--url-field", default="url")
p.add_argument(LIVE_FLAG, dest="authorized", action="store_true",
help="assert you checked robots.txt AND the site's ToS")
p.add_argument("--version", action="version",
version="r4 scrape_listings 1.0 (facts read %s)" % READ_DATE)
return p
def guess_format(path: str, data: bytes) -> str:
low = path.lower().rsplit("?", 1)[0]
if low.endswith((".html", ".htm")):
return "html"
if low.endswith(".json"):
return "json"
if low.endswith(".csv") or low.endswith(".tsv"):
return "csv"
head = data[:400].lstrip()
if head[:1] in (b"[", b"{"):
return "json"
if b" int:
args = build_parser().parse_args(argv)
live = bool(re.match(r"^https?://", args.input, flags=re.IGNORECASE))
# ---- the gate, ahead of any network code ---- #
if live and not args.authorized:
sys.stderr.write(REFUSAL % {"url": args.input,
"host": host_root(args.input)})
return 2
limiter = RateLimiter(min_interval_s=max(1.0, args.limit))
robots_text = ""
if args.input == "-":
data = sys.stdin.buffer.read()
base = args.base_url
elif live:
try:
data, robots_text = live_fetch(args.input, limiter=limiter,
user_agent=args.user_agent)
except PermissionError as exc:
print("ERROR: %s" % exc, file=sys.stderr)
return 3
except Exception as exc: # noqa: BLE001
print("ERROR: live fetch failed: %s" % exc, file=sys.stderr)
return 4
base = args.base_url or args.input
else:
try:
with open(args.input, "rb") as fh:
data = fh.read()
except OSError as exc:
print("ERROR: input could not be read: %s" % exc, file=sys.stderr)
return 4
base = args.base_url
fmt = args.format or guess_format(args.input, data)
try:
if fmt == "html":
raw_rows, stopped = read_html_bytes(
data, base_url=base, item_class=args.item_class,
title_class=args.title_class, price_class=args.price_class,
item_tag=args.item_tag, max_rows=args.max_rows)
elif fmt == "json":
raw_rows, stopped = read_json_bytes(
data, title_field=args.title_field, price_field=args.price_field,
url_fields=(args.url_field, "href", "link"),
max_rows=args.max_rows)
else:
raw_rows, stopped = read_csv_bytes(
data, title_field=args.title_field, price_field=args.price_field,
url_field=args.url_field, max_rows=args.max_rows)
except ValueError as exc:
print("ERROR: %s" % exc, file=sys.stderr)
return 4
rows, rejected = to_output_rows(raw_rows,
default_currency=args.default_currency)
if not rows:
print("ERROR: %d source row(s) read, none could be kept. Reasons:"
% len(raw_rows), file=sys.stderr)
for line in rejected[:20]:
print(" - %s" % line, file=sys.stderr)
return 5
write_csv(rows, args.out)
print("\nread %d source row(s) -> %d provenance row(s) [%s], %d rejected, "
"cap %d%s" % (len(raw_rows), len(rows), fmt, len(rejected),
args.max_rows,
"; CAP HIT, more were not stored" if stopped else ""),
file=sys.stderr)
for line in rejected[:10]:
print(" rejected: %s" % line, file=sys.stderr)
return 0
if __name__ == "__main__":
raise SystemExit(main())