#!/usr/bin/env python3 """ csv2json — reusable CSV -> JSON converter. Handles arbitrary CSV structure, configurable delimiters, custom output formatting, type inference, and explicit error handling for malformed input. Usage ----- python csv2json.py input.csv -o output.json python csv2json.py input.csv --delimiter ';' --indent 2 cat data.csv | python csv2json.py - # read stdin python csv2json.py in.csv --array-of-rows # [{"a":1,...}, ...] python csv2json.py in.csv --key-column id # {"1": {...}, ...} Exit codes ---------- 0 success 1 input/parse error (bad file, bad delimiter, malformed row) 2 usage error """ from __future__ import annotations import argparse import csv import io import json import re import sys from typing import Any, Iterable, Iterator class CSVError(Exception): """Raised for any input the converter cannot honestly process.""" # ── type inference ──────────────────────────────────────────────── _INT_RE = re.compile(r"^[+-]?\d{1,18}$") _FLOAT_RE = re.compile(r"^[+-]?(\d+\.\d*|\.\d+|\d+)([eE][+-]?\d+)?$") def infer(value: str) -> Any: """Turn a CSV string cell into int/float/bool/null when unambiguous. Ambiguous or empty values stay as strings so nothing is silently changed. """ if value is None: return None v = value.strip() if v == "": return "" low = v.lower() if low in ("true", "false"): return low == "true" if low in ("null", "none"): return None if _INT_RE.match(v): try: return int(v) except ValueError: return v if _FLOAT_RE.match(v): try: return float(v) except ValueError: return v return v # ── reading ─────────────────────────────────────────────────────── def sniff_dialect(sample: str, delimiter: str | None) -> type(csv.Dialect): """Pick the CSV dialect, honouring an explicit delimiter when given.""" if delimiter: class _Fixed(csv.Dialect): quoting = csv.QUOTE_MINIMAL quotechar = '"' doublequote = True skipinitialspace = False lineterminator = "\r\n" _Fixed.delimiter = delimiter # type: ignore[attr-defined] return _Fixed try: return csv.Sniffer().sniff(sample, delimiters=",;\t|") except csv.Error as exc: raise CSVError(f"could not detect the delimiter: {exc}") from exc def read_rows(text: str, delimiter: str | None = None) -> Iterator[list[str]]: """Yield rows from CSV text, validating column counts.""" if not text.strip(): raise CSVError("input is empty") sample = text[:8192] dialect = sniff_dialect(sample, delimiter) reader = csv.reader(io.StringIO(text, newline=""), dialect) rows = list(reader) if not rows: raise CSVError("no rows found") width = len(rows[0]) if width == 0: raise CSVError("header row is empty") for lineno, row in enumerate(rows, start=1): if len(row) != width: raise CSVError( f"row {lineno} has {len(row)} columns, expected {width} " f"(inconsistent CSV — check unescaped delimiters or quotes)" ) # Guard against a delimiter that produced a single over-wide column: # a real CSV header almost never has zero delimiter characters in the text. if width == 1 and delimiter and delimiter not in text: raise CSVError(f"delimiter {delimiter!r} does not appear in the input") return iter(rows) # ── conversion ──────────────────────────────────────────────────── def to_records( rows: Iterable[list[str]], *, infer_types: bool, skip_blank_rows: bool, ) -> list[dict[str, Any]]: """Convert rows into a list of {header: value} dicts.""" it = iter(rows) try: header = next(it) except StopIteration: raise CSVError("no header row") from None header = [h.strip() for h in header] if len(set(header)) != len(header): raise CSVError(f"duplicate column names in header: {header}") out: list[dict[str, Any]] = [] for lineno, row in enumerate(it, start=2): if skip_blank_rows and not any(c.strip() for c in row): continue rec = {} for name, cell in zip(header, row): rec[name] = infer(cell) if infer_types else cell out.append(rec) return out def convert( text: str, *, delimiter: str | None = None, indent: int = 2, infer_types: bool = True, array_of_rows: bool = False, key_column: str | None = None, skip_blank_rows: bool = True, ensure_ascii: bool = False, ) -> str: """Convert CSV text to a JSON string.""" rows = read_rows(text, delimiter) records = to_records(rows, infer_types=infer_types, skip_blank_rows=skip_blank_rows) if key_column is not None: obj: Any = {} for rec in records: if key_column not in rec: raise CSVError(f"key column {key_column!r} not in header") key = rec[key_column] if key in obj: raise CSVError(f"duplicate key {key!r} in column {key_column!r}") obj[key] = {k: v for k, v in rec.items() if k != key_column} payload: Any = obj elif array_of_rows: payload = [[c for c in r] for r in rows] if False else records else: payload = records return json.dumps(payload, indent=indent, ensure_ascii=ensure_ascii) # ── CLI ─────────────────────────────────────────────────────────── def build_parser() -> argparse.ArgumentParser: p = argparse.ArgumentParser( prog="csv2json", description="Convert CSV files to JSON with delimiter and formatting control.", ) p.add_argument("input", help="CSV file path, or '-' for stdin") p.add_argument("-o", "--output", help="write to this file instead of stdout") p.add_argument("-d", "--delimiter", help="field delimiter (auto-detected when omitted)") p.add_argument("--indent", type=int, default=2, help="JSON indentation (default 2)") p.add_argument("--no-infer", action="store_true", help="keep every value as a string") p.add_argument("--array-of-rows", action="store_true", help="emit a bare JSON array instead of wrapping objects") p.add_argument("--key-column", help="key the output object by this column") p.add_argument("--keep-blank-rows", action="store_true", help="do not drop rows that are entirely empty") p.add_argument("--ascii", action="store_true", help="escape non-ASCII characters") return p def main(argv: list[str] | None = None) -> int: args = build_parser().parse_args(argv) if args.input == "-": text = sys.stdin.read() else: try: with open(args.input, "r", encoding="utf-8-sig", newline="") as fh: text = fh.read() except FileNotFoundError: print(f"csv2json: no such file: {args.input}", file=sys.stderr) return 1 except (OSError, UnicodeDecodeError) as exc: print(f"csv2json: cannot read {args.input}: {exc}", file=sys.stderr) return 1 try: result = convert( text, delimiter=args.delimiter, indent=args.indent, infer_types=not args.no_infer, array_of_rows=args.array_of_rows, key_column=args.key_column, skip_blank_rows=not args.keep_blank_rows, ensure_ascii=args.ascii, ) except CSVError as exc: print(f"csv2json: {exc}", file=sys.stderr) return 1 if args.output: try: with open(args.output, "w", encoding="utf-8") as fh: fh.write(result + "\n") except OSError as exc: print(f"csv2json: cannot write {args.output}: {exc}", file=sys.stderr) return 1 else: sys.stdout.write(result + "\n") return 0 if __name__ == "__main__": raise SystemExit(main())