convert.py

#!/usr/bin/env python3
"""Convert a strict customer CSV to JSON (catalog demonstration sample).

Usage:
    python3 convert.py INPUT.csv OUTPUT.json

The input must be UTF-8 (an optional BOM is tolerated) with exactly the
header row customer_id,full_name,city in that order. Every record becomes
{"id": ..., "name": ..., "city": ...} with the name and city whitespace-
trimmed and the identifier preserved exactly, including leading zeroes.
The sample is limited to 20000 records and a 10 MiB input. The output file
is never overwritten, and the whole input is validated before the output
is created. The input is never modified.
"""

import csv
import json
import os
import sys

REQUIRED_HEADERS = ["customer_id", "full_name", "city"]
MAX_RECORDS = 20000
MAX_INPUT_BYTES = 10 * 1024 * 1024  # 10 MiB


class ConversionError(Exception):
    """Routine user-facing error, reported without a traceback."""


def fail(message):
    raise ConversionError(message)


def read_records(input_path):
    """Read and fully validate the input; return the converted records."""
    try:
        input_size = os.path.getsize(input_path)
    except OSError as exc:
        fail(f"cannot read input file: {exc}")
    if input_size > MAX_INPUT_BYTES:
        fail(
            f"input file is {input_size} bytes; this sample is limited to "
            f"{MAX_INPUT_BYTES} bytes (10 MiB)"
        )

    records = []
    try:
        with open(input_path, "r", encoding="utf-8-sig", newline="") as handle:
            reader = csv.reader(handle, strict=True)
            header = next(reader, None)
            if header is None:
                fail(
                    "input file is empty; expected a header row "
                    f"'{','.join(REQUIRED_HEADERS)}'"
                )
            if header != REQUIRED_HEADERS:
                fail(
                    "unexpected header row: this sample requires exactly "
                    f"'{','.join(REQUIRED_HEADERS)}' in that order, got "
                    f"'{','.join(header)}'"
                )
            for line_number, row in enumerate(reader, start=2):
                if len(row) != len(REQUIRED_HEADERS):
                    fail(
                        f"row {line_number}: expected {len(REQUIRED_HEADERS)} "
                        f"fields, got {len(row)}"
                    )
                customer_id, full_name, city = row
                records.append(
                    {"id": customer_id, "name": full_name.strip(), "city": city.strip()}
                )
                if len(records) > MAX_RECORDS:
                    fail(
                        f"row {line_number} exceeds this sample's limit of "
                        f"{MAX_RECORDS} records"
                    )
    except UnicodeDecodeError as exc:
        fail(f"input is not valid UTF-8: {exc}")
    except csv.Error as exc:
        fail(f"input is not valid CSV: {exc}")
    except OSError as exc:
        fail(f"cannot read input file: {exc}")
    return records


def main(argv):
    if len(argv) != 3:
        fail("usage: python3 convert.py INPUT.csv OUTPUT.json")
    input_path, output_path = argv[1], argv[2]

    if not os.path.exists(input_path):
        fail(f"input file not found: {input_path}")
    if os.path.exists(output_path):
        fail(f"output file already exists, refusing to overwrite it: {output_path}")

    records = read_records(input_path)

    try:
        with open(output_path, "x", encoding="utf-8", newline="\n") as handle:
            json.dump(records, handle, ensure_ascii=False, indent=2)
            handle.write("\n")
    except OSError as exc:
        fail(f"cannot write output file: {exc}")
    print(f"Converted {len(records)} record(s).")


if __name__ == "__main__":
    try:
        main(sys.argv)
    except ConversionError as exc:
        print(f"convert.py: error: {exc}", file=sys.stderr)
        sys.exit(1)