#!/usr/bin/env python3 """Verify and query the pinned WebTruffle retail product prices edition. The recipe uses only the Python 3.11+ standard library. It downloads (or reuses) the tagged 2026-08-08 manifest and 238 MB CSV, verifies their exact byte counts and SHA-256 digests while streaming, checks the complete 40-column schema, then makes ONE streaming pass over the file and writes deterministic, bounded analysis products: * coverage-audit.csv * market-sample-{country}-{currency}-{category}-{year}.csv * barcode-price-history.csv * barcode-monthly-summary.csv * barcode-retailer-summary.csv * query-provenance.json Prices stay in their source currencies. Never sum or average values across currencies, and never treat a cumulative community snapshot as a live price feed, a stock signal, or a representative inflation index. SPDX-License-Identifier: MIT """ from __future__ import annotations import argparse import csv import hashlib import json import os import platform import statistics import sys import tempfile from collections import Counter, defaultdict from dataclasses import dataclass from datetime import date, datetime, timezone from decimal import Decimal, InvalidOperation, ROUND_HALF_UP from pathlib import Path from typing import Any, Iterable, Sequence from urllib.request import Request, urlopen if sys.version_info < (3, 11): raise RuntimeError("This recipe requires Python 3.11 or newer.") csv.field_size_limit(4 * 1024 * 1024) REPOSITORY = "webtruffle/retail-product-catalog-prices" REPOSITORY_URL = f"https://github.com/{REPOSITORY}" RELEASE_TAG = "2026-08-08" RELEASE_URL = f"{REPOSITORY_URL}/releases/tag/{RELEASE_TAG}" DATASET_PAGE = "https://www.webtruffle.com/datasets/retail-product-catalog-prices" SOURCE_LICENSE = "https://opendatacommons.org/licenses/odbl/1-0/" SOURCE_ATTRIBUTION = ( "Contains information from Open Food Facts and Open Prices, made available " "under the Open Database License (ODbL) 1.0. Location data includes " "\u00a9 OpenStreetMap contributors, ODbL. Normalized and filtered by " "WebTruffle; not endorsed by Open Food Facts or OpenStreetMap." ) USER_AGENT = "WebTruffle-retail-prices-Python-recipe/1.0 (+https://www.webtruffle.com/)" CHUNK_BYTES = 1024 * 1024 RECIPE_VERSION = "1.0.0" MANIFEST_NAME = "manifest-2026-08-08.json" CSV_NAME = "retail-product-prices-2026-08-08.csv" EXPECTED_HEADERS = ( "id", "source", "source_price_id", "record_type", "product_source", "product_code", "product_name", "quantity", "product_quantity", "product_quantity_unit", "brands", "brands_tags", "categories_tags", "labels_tags", "nutriscore_grade", "environmental_score_grade", "nova_group", "product_url", "price", "currency", "observed_date", "price_is_discounted", "price_without_discount", "discount_type", "price_per", "location_type", "retailer_name", "retailer_brand", "retailer_type", "city", "country", "country_code", "source_created_at", "source_updated_at", "source_url", "source_license", "first_seen_at", "last_seen_at", "content_hash", "change_type", ) if len(EXPECTED_HEADERS) != 40: raise RuntimeError("The pinned retail-prices schema must contain exactly 40 fields.") @dataclass(frozen=True) class ReleaseSpec: tag: str target_date: str generated_at: str schema_version: str manifest_bytes: int manifest_sha256: str csv_bytes: int csv_sha256: str record_count: int product_count: int discounted_count: int observation_date_min: str observation_date_max: str RELEASE = ReleaseSpec( tag="2026-08-08", target_date="2026-08-08", generated_at="2026-08-09T13:32:23Z", schema_version="1.0", manifest_bytes=8_608, manifest_sha256=( "1d9c13c7e8cc506873b6ff0a7e7f3ae2" "8c860ac23f10c52f3001cf6c7e13021e" ), csv_bytes=238_009_759, csv_sha256=( "49e2a5a9944c5e2c905aef968bfa7d4d" "58dc6c9ef6d558c28916cf4a25a29f0e" ), record_count=270_334, product_count=129_150, discounted_count=23_999, observation_date_min="2010-07-08", observation_date_max="2026-08-08", ) DEFAULT_FILTERS = { "country": "France", "currency": "EUR", "category": "en:breakfasts", "sample_year": "2026", "product_code": "3017620422003", } DEFAULT_ORACLES = { "sample_rows": 2_550, "history_rows": 148, "history_distinct_dates": 111, "history_months": 38, "history_retailers": 37, "history_min_price": "2.58", "history_max_price": "5.02", } DISCOUNTED_WITH_REFERENCE_ROWS = 21_702 COVERAGE_SOURCE_GROUPS = 5 SAMPLE_COLUMNS = ( "edition_date", "observed_date", "product_code", "product_name", "brands", "quantity", "price", "currency", "price_is_discounted", "price_without_discount", "discount_type", "retailer_name", "retailer_brand", "retailer_type", "city", "country", "location_type", "source_price_id", "source_url", "source_license", ) HISTORY_COLUMNS = ( "observed_date", "price", "currency", "price_is_discounted", "price_without_discount", "discount_type", "retailer_name", "city", "location_type", "product_name", "source_price_id", "source_url", ) MONTHLY_COLUMNS = ( "month", "observations", "min_price", "median_price", "max_price", "discounted_observations", "distinct_retailers", ) RETAILER_COLUMNS = ( "retailer_name", "observations", "distinct_dates", "min_price", "median_price", "max_price", "first_observed", "last_observed", ) COVERAGE_COLUMNS = ( "product_source", "observation_rows", "unique_products", "rows_with_product_name", "rows_with_brands", "rows_with_quantity", "rows_with_retailer_name", "rows_with_city", "product_name_coverage", ) def fail(message: str) -> None: raise RuntimeError(message) def sha256_and_size(path: Path) -> tuple[str, int]: digest = hashlib.sha256() size = 0 with path.open("rb") as source: for chunk in iter(lambda: source.read(CHUNK_BYTES), b""): digest.update(chunk) size += len(chunk) return digest.hexdigest(), size def download_verified(url: str, destination: Path, expected_bytes: int, expected_sha256: str) -> None: """Stream a release asset to disk, aborting before oversized or altered bytes.""" if destination.exists(): actual_hash, actual_bytes = sha256_and_size(destination) if actual_bytes != expected_bytes: fail(f"{destination.name}: expected {expected_bytes} bytes, found {actual_bytes}.") if actual_hash != expected_sha256: fail(f"{destination.name}: SHA-256 mismatch; refusing to use altered bytes.") return destination.parent.mkdir(parents=True, exist_ok=True) handle, temporary_name = tempfile.mkstemp( prefix=f".{destination.name}.", suffix=".part", dir=destination.parent ) temporary_path = Path(temporary_name) try: digest = hashlib.sha256() received = 0 request = Request(url, headers={"User-Agent": USER_AGENT}) with urlopen(request, timeout=120) as response, os.fdopen(handle, "wb") as target: while True: chunk = response.read(CHUNK_BYTES) if not chunk: break received += len(chunk) if received > expected_bytes: fail(f"{destination.name}: download exceeded the declared {expected_bytes} bytes.") digest.update(chunk) target.write(chunk) if received != expected_bytes: fail(f"{destination.name}: expected {expected_bytes} bytes, downloaded {received}.") if digest.hexdigest() != expected_sha256: fail(f"{destination.name}: downloaded SHA-256 does not match the pinned digest.") os.replace(temporary_path, destination) finally: if temporary_path.exists(): temporary_path.unlink() def parse_args(argv: Sequence[str] | None) -> argparse.Namespace: parser = argparse.ArgumentParser( description=( "Verify the pinned 2026-08-08 WebTruffle retail product prices " "edition and derive bounded, reproducible analysis products." ) ) parser.add_argument( "--data-dir", default="./evidence/retail-product-prices-2026-08-08", help="Directory that stores the pinned manifest, CSV, and derived results.", ) parser.add_argument("--country", default=DEFAULT_FILTERS["country"], help="Market-sample country filter.") parser.add_argument("--currency", default=DEFAULT_FILTERS["currency"], help="Market-sample currency filter.") parser.add_argument( "--category", default=DEFAULT_FILTERS["category"], help="Exact categories_tags member required by the market sample.", ) parser.add_argument( "--sample-year", default=DEFAULT_FILTERS["sample_year"], help="Observed-date year required by the market sample.", ) parser.add_argument( "--product-code", default=DEFAULT_FILTERS["product_code"], help="Barcode whose price history, monthly summary, and retailer summary are derived.", ) return parser.parse_args(argv) def load_manifest(path: Path) -> dict[str, Any]: manifest = json.loads(path.read_text(encoding="utf-8")) checks = ( ("dataset_id", "retail-product-catalog-prices"), ("schema_version", RELEASE.schema_version), ("target_date", RELEASE.target_date), ("generated_at", RELEASE.generated_at), ("record_count", RELEASE.record_count), ("product_count", RELEASE.product_count), ("discounted_count", RELEASE.discounted_count), ("observation_date_min", RELEASE.observation_date_min), ("observation_date_max", RELEASE.observation_date_max), ) for key, expected in checks: if manifest.get(key) != expected: fail(f"manifest {key}: expected {expected!r}, found {manifest.get(key)!r}.") if manifest.get("record_fields") != list(EXPECTED_HEADERS): fail("manifest record_fields does not match the pinned 40-column schema.") csv_entry = manifest.get("files", {}).get("retail-product-prices.csv", {}) if csv_entry.get("bytes") != RELEASE.csv_bytes or csv_entry.get("sha256") != RELEASE.csv_sha256: fail("manifest no longer declares the pinned CSV identity.") return manifest def parse_price(value: str) -> Decimal: try: return Decimal(value) except InvalidOperation as error: raise RuntimeError(f"invalid published price value: {value!r}") from error def parse_categories(value: str) -> list[str]: text = value.strip() if not text: return [] parsed = json.loads(text) if not isinstance(parsed, list) or not all(isinstance(item, str) for item in parsed): raise RuntimeError("categories_tags must decode to a JSON array of strings.") return parsed def slugify(value: str) -> str: return "".join(ch if ch.isalnum() else "-" for ch in value.casefold()).strip("-") def quantize(value: Decimal) -> str: return str(value.quantize(Decimal("0.01"), rounding=ROUND_HALF_UP)) @dataclass class StreamTotals: rows: int = 0 discounted_rows: int = 0 discounted_with_reference: int = 0 seen_ids: set[str] | None = None seen_source_price_ids: set[str] | None = None product_codes: set[str] | None = None coverage: dict[str, Counter] | None = None coverage_products: dict[str, set[str]] | None = None sample_rows: list[dict[str, str]] | None = None history_rows: list[dict[str, str]] | None = None def __post_init__(self) -> None: self.seen_ids = set() self.seen_source_price_ids = set() self.product_codes = set() self.coverage = defaultdict(Counter) self.coverage_products = defaultdict(set) self.sample_rows = [] self.history_rows = [] def stream_analysis(csv_path: Path, args: argparse.Namespace) -> StreamTotals: totals = StreamTotals() sample_year = args.sample_year.strip() category = args.category.strip() country = args.country.strip() currency = args.currency.strip() product_code = args.product_code.strip() with open(csv_path, encoding="utf-8-sig", newline="") as source: reader = csv.DictReader(source) if tuple(reader.fieldnames or ()) != EXPECTED_HEADERS: fail("CSV header does not match the pinned 40-column schema in the declared order.") for row in reader: totals.rows += 1 row_id = row["id"] source_price_id = row["source_price_id"] if row_id in totals.seen_ids: fail(f"duplicate id encountered: {row_id}") if source_price_id in totals.seen_source_price_ids: fail(f"duplicate source_price_id encountered: {source_price_id}") totals.seen_ids.add(row_id) totals.seen_source_price_ids.add(source_price_id) code = row["product_code"].strip() if not code: fail("a retained observation must carry a barcode.") totals.product_codes.add(code) price = parse_price(row["price"]) if price < 0: fail(f"negative published price for source_price_id {source_price_id}.") observed = date.fromisoformat(row["observed_date"]) categories = parse_categories(row["categories_tags"]) discounted = row["price_is_discounted"] == "True" if discounted: totals.discounted_rows += 1 if row["price_without_discount"].strip(): totals.discounted_with_reference += 1 group = row["product_source"].strip() or "unmatched" counters = totals.coverage[group] counters["rows"] += 1 totals.coverage_products[group].add(code) for field in ("product_name", "brands", "quantity", "retailer_name", "city"): if row[field].strip(): counters[field] += 1 if ( row["country"] == country and row["currency"] == currency and observed.year == int(sample_year) and category in categories ): totals.sample_rows.append( { "edition_date": RELEASE.target_date, "observed_date": row["observed_date"], "product_code": code, "product_name": row["product_name"], "brands": row["brands"], "quantity": row["quantity"], "price": row["price"], "currency": row["currency"], "price_is_discounted": row["price_is_discounted"], "price_without_discount": row["price_without_discount"], "discount_type": row["discount_type"], "retailer_name": row["retailer_name"], "retailer_brand": row["retailer_brand"], "retailer_type": row["retailer_type"], "city": row["city"], "country": row["country"], "location_type": row["location_type"], "source_price_id": source_price_id, "source_url": row["source_url"], "source_license": row["source_license"], } ) if code == product_code and row["country"] == country and row["currency"] == currency: totals.history_rows.append( { "observed_date": row["observed_date"], "price": row["price"], "currency": row["currency"], "price_is_discounted": row["price_is_discounted"], "price_without_discount": row["price_without_discount"], "discount_type": row["discount_type"], "retailer_name": row["retailer_name"], "city": row["city"], "location_type": row["location_type"], "product_name": row["product_name"], "source_price_id": source_price_id, "source_url": row["source_url"], } ) return totals def write_csv(path: Path, columns: Sequence[str], rows: Iterable[dict[str, str]]) -> int: count = 0 with open(path, "w", encoding="utf-8", newline="") as target: writer = csv.DictWriter(target, fieldnames=list(columns), lineterminator="\n") writer.writeheader() for row in rows: writer.writerow(row) count += 1 return count def build_outputs(stage: Path, totals: StreamTotals, args: argparse.Namespace) -> list[dict[str, Any]]: outputs: list[dict[str, Any]] = [] coverage_rows = [] for group in sorted(totals.coverage): counters = totals.coverage[group] rows = counters["rows"] coverage_rows.append( { "product_source": group, "observation_rows": str(rows), "unique_products": str(len(totals.coverage_products[group])), "rows_with_product_name": str(counters["product_name"]), "rows_with_brands": str(counters["brands"]), "rows_with_quantity": str(counters["quantity"]), "rows_with_retailer_name": str(counters["retailer_name"]), "rows_with_city": str(counters["city"]), "product_name_coverage": str( (Decimal(counters["product_name"]) / Decimal(rows)).quantize(Decimal("0.0001")) ), } ) if len(coverage_rows) != COVERAGE_SOURCE_GROUPS: fail(f"expected {COVERAGE_SOURCE_GROUPS} coverage groups, found {len(coverage_rows)}.") coverage_path = stage / "coverage-audit.csv" coverage_count = write_csv(coverage_path, COVERAGE_COLUMNS, coverage_rows) outputs.append({"path": "coverage-audit.csv", "rows": coverage_count, "grain": "one product-source group"}) sample_rows = sorted(totals.sample_rows, key=lambda row: (row["observed_date"], row["source_price_id"])) if not sample_rows: fail("the requested market-sample filter matched zero rows; refusing to publish an empty sample.") sample_name = ( "market-sample-" f"{slugify(args.country)}-{slugify(args.currency)}-" f"{slugify(args.category)}-{args.sample_year}.csv" ) sample_count = write_csv(stage / sample_name, SAMPLE_COLUMNS, sample_rows) outputs.append({"path": sample_name, "rows": sample_count, "grain": "one filtered price observation"}) history_rows = sorted(totals.history_rows, key=lambda row: (row["observed_date"], row["source_price_id"])) if not history_rows: fail("the requested barcode history matched zero rows; refusing to publish an empty history.") history_count = write_csv(stage / "barcode-price-history.csv", HISTORY_COLUMNS, history_rows) outputs.append({"path": "barcode-price-history.csv", "rows": history_count, "grain": "one barcode observation"}) monthly: dict[str, list[Decimal]] = {} monthly_discounted: Counter = Counter() monthly_retailers: dict[str, set[str]] = defaultdict(set) for row in history_rows: month = row["observed_date"][:7] monthly.setdefault(month, []).append(Decimal(row["price"])) if row["price_is_discounted"] == "True": monthly_discounted[month] += 1 retailer = row["retailer_name"].strip() if retailer: monthly_retailers[month].add(retailer) monthly_rows = [ { "month": month, "observations": str(len(values)), "min_price": quantize(min(values)), "median_price": quantize(Decimal(statistics.median(values))), "max_price": quantize(max(values)), "discounted_observations": str(monthly_discounted.get(month, 0)), "distinct_retailers": str(len(monthly_retailers.get(month, set()))), } for month, values in sorted(monthly.items()) ] monthly_count = write_csv(stage / "barcode-monthly-summary.csv", MONTHLY_COLUMNS, monthly_rows) outputs.append({"path": "barcode-monthly-summary.csv", "rows": monthly_count, "grain": "one observed month"}) retailer_values: dict[str, list[Decimal]] = defaultdict(list) retailer_dates: dict[str, set[str]] = defaultdict(set) for row in history_rows: retailer = row["retailer_name"].strip() if not retailer: continue retailer_values[retailer].append(Decimal(row["price"])) retailer_dates[retailer].add(row["observed_date"]) retailer_rows = [] for retailer in sorted(retailer_values, key=lambda name: (-len(retailer_values[name]), name)): values = retailer_values[retailer] dates = sorted(retailer_dates[retailer]) retailer_rows.append( { "retailer_name": retailer, "observations": str(len(values)), "distinct_dates": str(len(dates)), "min_price": quantize(min(values)), "median_price": quantize(Decimal(statistics.median(values))), "max_price": quantize(max(values)), "first_observed": dates[0], "last_observed": dates[-1], } ) retailer_count = write_csv(stage / "barcode-retailer-summary.csv", RETAILER_COLUMNS, retailer_rows) outputs.append({"path": "barcode-retailer-summary.csv", "rows": retailer_count, "grain": "one retailer name"}) return outputs def assert_oracles(totals: StreamTotals, args: argparse.Namespace) -> None: if totals.rows != RELEASE.record_count: fail(f"expected {RELEASE.record_count} rows, streamed {totals.rows}.") if len(totals.product_codes) != RELEASE.product_count: fail(f"expected {RELEASE.product_count} unique barcodes, found {len(totals.product_codes)}.") if totals.discounted_rows != RELEASE.discounted_count: fail(f"expected {RELEASE.discounted_count} discounted rows, found {totals.discounted_rows}.") if totals.discounted_with_reference != DISCOUNTED_WITH_REFERENCE_ROWS: fail( "expected " f"{DISCOUNTED_WITH_REFERENCE_ROWS} discounted rows with a reference price, " f"found {totals.discounted_with_reference}." ) defaults_active = all(getattr(args, key) == value for key, value in DEFAULT_FILTERS.items()) if not defaults_active: return oracle_checks = ( (len(totals.sample_rows), DEFAULT_ORACLES["sample_rows"], "market-sample rows"), (len(totals.history_rows), DEFAULT_ORACLES["history_rows"], "barcode history rows"), ( len({row["observed_date"] for row in totals.history_rows}), DEFAULT_ORACLES["history_distinct_dates"], "barcode history distinct dates", ), ) for actual, expected, label in oracle_checks: if actual != expected: fail(f"default-filter oracle failed: {label} expected {expected}, found {actual}.") prices = [Decimal(row["price"]) for row in totals.history_rows] if str(min(prices)) != DEFAULT_ORACLES["history_min_price"]: fail("default-filter oracle failed: history minimum price changed.") if str(max(prices)) != DEFAULT_ORACLES["history_max_price"]: fail("default-filter oracle failed: history maximum price changed.") def recipe_sha256() -> str: try: return sha256_and_size(Path(__file__).resolve())[0] except OSError: return "unavailable" def main(argv: Sequence[str] | None = None) -> int: args = parse_args(argv) data_dir = Path(args.data_dir).expanduser().resolve() data_dir.mkdir(parents=True, exist_ok=True) manifest_path = data_dir / MANIFEST_NAME csv_path = data_dir / CSV_NAME print(f"Verifying pinned release {RELEASE.tag} in {data_dir}") download_verified( f"{REPOSITORY_URL}/releases/download/{RELEASE.tag}/manifest.json", manifest_path, RELEASE.manifest_bytes, RELEASE.manifest_sha256, ) manifest = load_manifest(manifest_path) download_verified( f"{REPOSITORY_URL}/releases/download/{RELEASE.tag}/retail-product-prices.csv", csv_path, RELEASE.csv_bytes, RELEASE.csv_sha256, ) print("Streaming the verified CSV (single pass, bounded memory)...") totals = stream_analysis(csv_path, args) assert_oracles(totals, args) results_dir = data_dir / "results" stage = Path(tempfile.mkdtemp(prefix=".results-stage-", dir=data_dir)) try: outputs = build_outputs(stage, totals, args) manifest_recheck = sha256_and_size(manifest_path) csv_recheck = sha256_and_size(csv_path) if manifest_recheck != (RELEASE.manifest_sha256, RELEASE.manifest_bytes): fail("manifest bytes changed while processing; refusing to promote results.") if csv_recheck != (RELEASE.csv_sha256, RELEASE.csv_bytes): fail("CSV bytes changed while processing; refusing to promote results.") results_dir.mkdir(parents=True, exist_ok=True) stale_receipt = results_dir / "query-provenance.json" if stale_receipt.exists(): stale_receipt.unlink() for output in outputs: staged = stage / output["path"] output_hash, output_bytes = sha256_and_size(staged) output["bytes"] = output_bytes output["sha256"] = output_hash os.replace(staged, results_dir / output["path"]) provenance = { "recipe": {"version": RECIPE_VERSION, "sha256": recipe_sha256()}, "runtime": { "python_implementation": platform.python_implementation(), "python_version": platform.python_version(), "platform": platform.platform(), }, "filters": { "country": args.country, "currency": args.currency, "category": args.category, "sample_year": args.sample_year, "product_code": args.product_code, }, "release": { "repository": REPOSITORY, "repository_url": REPOSITORY_URL, "release_url": RELEASE_URL, "tag": RELEASE.tag, "target_date": RELEASE.target_date, "generated_at": RELEASE.generated_at, "schema_version": RELEASE.schema_version, "record_grain": manifest.get("record_grain"), }, "inputs": { "manifest": { "path": MANIFEST_NAME, "bytes": RELEASE.manifest_bytes, "sha256": RELEASE.manifest_sha256, }, "csv": {"path": CSV_NAME, "bytes": RELEASE.csv_bytes, "sha256": RELEASE.csv_sha256}, }, "manifest_declared": { "record_count": RELEASE.record_count, "product_count": RELEASE.product_count, "location_count": manifest.get("location_count"), "country_count": manifest.get("country_count"), "currency_count": manifest.get("currency_count"), "discounted_count": RELEASE.discounted_count, "observation_date_min": RELEASE.observation_date_min, "observation_date_max": RELEASE.observation_date_max, }, "observed_totals": { "rows": totals.rows, "unique_ids": len(totals.seen_ids), "unique_source_price_ids": len(totals.seen_source_price_ids), "unique_product_codes": len(totals.product_codes), "discounted_rows": totals.discounted_rows, "discounted_rows_with_reference_price": totals.discounted_with_reference, }, "outputs": outputs, "interpretation_notes": [ "Prices are preserved in source currencies; never sum or average across currencies.", "The edition is a cumulative community snapshot, not a live price feed, stock signal, or price index.", "Retailer names are contributor-reported free text, not resolved legal entities.", "A missing product name, brand, or category means the community catalog lacks it, not that the product lacks one.", "Discount flags describe one observed price event, not a current promotion.", "observed_date is the contributor-reported observation date; first_seen_at/last_seen_at are pipeline clocks.", ], "attribution": SOURCE_ATTRIBUTION, "source_license": SOURCE_LICENSE, "dataset_page": DATASET_PAGE, "generated_at": datetime.now(timezone.utc).isoformat(timespec="seconds"), } receipt_path = results_dir / "query-provenance.json" receipt_handle, receipt_name = tempfile.mkstemp(prefix=".provenance-", dir=results_dir) with os.fdopen(receipt_handle, "w", encoding="utf-8") as receipt_file: json.dump(provenance, receipt_file, indent=2, ensure_ascii=False) receipt_file.write("\n") os.replace(receipt_name, receipt_path) finally: for leftover in stage.iterdir(): leftover.unlink() stage.rmdir() print(f"Wrote {len(outputs)} deterministic outputs plus query-provenance.json to {results_dir}") for output in outputs: print(f" {output['path']}: {output['rows']} rows ({output['bytes']} bytes)") return 0 if __name__ == "__main__": sys.exit(main())