JSONのリストをCSVに変換し、欠落フィールドを確認する

[
  {"item": "가상부품-A", "quantity": 3, "unit": "개"},
  {"item": "가상부품-B", "quantity": 5},
  {"item": "가상부품-C", "unit": "개"},
  {"item": "가상부품-D", "quantity": null, "unit": ""}
]

python --version
python example.py sample.json --output-dir outputs

python example.py sample.json --output-dir outputs_second

OK: rows=4; missing_key_cells=2; null_cells=1; empty_string_cells=1; issue_cells=4

#!/usr/bin/env python3
"""Convert a small local JSON array to CSV and a field-issue report.

Target: Python 3.12+, standard library only. No network access.
All output goes to a NEW directory; existing directories/files are refused.
"""

from __future__ import annotations

import argparse
import csv
import io
import json
import sys
from decimal import Decimal
from pathlib import Path

DEFAULT_FIELDS = ("item", "quantity", "unit")
MAX_INPUT_BYTES = 5 * 1024 * 1024
ISSUE_FIELDS = ("row_number", "field", "issue")


class InputError(ValueError):
    """The input does not satisfy this example's data contract."""


def unique_object(pairs: list[tuple[str, object]]) -> dict[str, object]:
    result: dict[str, object] = {}
    for key, value in pairs:
        if key in result:
            raise InputError("Duplicate key in a JSON object.")
        result[key] = value
    return result


def reject_constant(token: str) -> object:
    raise InputError(f"Non-standard JSON number: {token}.")


def csv_bytes(fields: tuple[str, ...], rows: list[dict]) -> bytes:
    buffer = io.StringIO(newline="")
    writer = csv.DictWriter(buffer, fieldnames=fields, lineterminator="\r\n")
    writer.writeheader()
    writer.writerows(rows)
    return buffer.getvalue().encode("utf-8")


def prepare_outputs(source: Path, fields: tuple[str, ...]) -> tuple[dict, dict]:
    if not fields or len(fields) != len(set(fields)):
        raise InputError("--fields must contain at least one unique field name.")
    if any(not field.strip() for field in fields):
        raise InputError("Field names must not be empty or whitespace-only.")

    # Read-only, bounded input. Do not modify the source, even on failure.
    with source.open("rb") as stream:
        raw = stream.read(MAX_INPUT_BYTES + 1)
    if len(raw) > MAX_INPUT_BYTES:
        raise InputError("Input exceeds the example's 5 MiB limit.")
    records = json.loads(
        raw.decode("utf-8-sig"),
        object_pairs_hook=unique_object,
        parse_float=Decimal,
        parse_constant=reject_constant,
    )
    if not isinstance(records, list):
        raise InputError("Top-level JSON value must be an array.")

    rows: list[dict[str, str]] = []
    issues: list[dict[str, object]] = []
    counts = {"missing_key": 0, "null": 0, "empty_string": 0}
    missing_rows: set[int] = set()
    allowed = set(fields)
    for row_number, record in enumerate(records, start=1):
        if not isinstance(record, dict):
            raise InputError(f"Record {row_number} must be an object.")
        if set(record) - allowed:
            raise InputError(
                f"Record {row_number} has unlisted keys; include them in --fields."
            )
        row: dict[str, str] = {}
        for field in fields:
            issue = None
            if field not in record:
                issue = "missing_key"
                missing_rows.add(row_number)
                value = None
            else:
                value = record[field]
                if value is None:
                    issue = "null"
                elif value == "":
                    issue = "empty_string"
            if issue is not None:
                counts[issue] += 1
                issues.append({"row_number": row_number, "field": field, "issue": issue})
                row[field] = ""
            elif isinstance(value, str):
                row[field] = value
            elif isinstance(value, bool):
                row[field] = "true" if value else "false"
            elif isinstance(value, (int, Decimal)):
                row[field] = str(value)
            else:
                raise InputError(
                    f"Record {row_number} contains an unsupported nested value."
                )
        rows.append(row)

    summary = {
        "status": "complete",
        "columns": list(fields),
        "input_rows": len(records),
        "output_rows": len(rows),
        "missing_key_cells": counts["missing_key"],
        "rows_with_missing_keys": len(missing_rows),
        "null_cells": counts["null"],
        "empty_string_cells": counts["empty_string"],
        "reported_issue_cells": len(issues),
    }
    # Render and UTF-8-encode EVERYTHING before creating the output directory.
    # summary.json is written last and serves as a completion record.
    outputs = {
        "converted.csv": csv_bytes(fields, rows),
        "field_issues.csv": csv_bytes(ISSUE_FIELDS, issues),
        "summary.json": (json.dumps(summary, ensure_ascii=False, indent=2) + "\n").encode("utf-8"),
    }
    return outputs, summary


def main(argv: list[str] | None = None) -> int:
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument("input", type=Path, help="Local UTF-8 JSON file")
    parser.add_argument("--output-dir", type=Path, required=True, help="NEW directory; parent must exist")
    parser.add_argument("--fields", nargs="+", default=list(DEFAULT_FIELDS), help="Expected CSV columns, in order")
    args = parser.parse_args(argv)
    if sys.version_info < (3, 12):
        print("E_VERSION: Python 3.12 or newer is required.", file=sys.stderr)
        return 2
    try:
        outputs, summary = prepare_outputs(args.input, tuple(args.fields))
    except json.JSONDecodeError as exc:
        print(f"E_INPUT: Invalid JSON at line {exc.lineno}, column {exc.colno}.", file=sys.stderr)
        return 2
    except (ValueError, ArithmeticError, RecursionError) as exc:
        print(f"E_INPUT: {exc}", file=sys.stderr)
        return 2
    except OSError as exc:
        print(f"E_READ: Cannot read input ({exc.__class__.__name__}).", file=sys.stderr)
        return 4

    try:
        args.output_dir.mkdir(exist_ok=False)
    except FileExistsError:
        print("E_OUTPUT_EXISTS: Output path already exists; choose a new directory.", file=sys.stderr)
        return 3
    except OSError as exc:
        print(f"E_WRITE: Cannot create output directory ({exc.__class__.__name__}).", file=sys.stderr)
        return 4
    try:
        for name, payload in outputs.items():
            with (args.output_dir / name).open("xb") as stream:
                stream.write(payload)
    except OSError as exc:
        print(
            f"E_WRITE: Incomplete NEW output directory ({exc.__class__.__name__}); "
            "do not use its results. Inspect it and choose a new directory.",
            file=sys.stderr,
        )
        return 4
    print(
        f"OK: rows={summary['output_rows']}; "
        f"missing_key_cells={summary['missing_key_cells']}; "
        f"null_cells={summary['null_cells']}; "
        f"empty_string_cells={summary['empty_string_cells']}; "
        f"issue_cells={summary['reported_issue_cells']}"
    )
    return 0


if __name__ == "__main__":
    raise SystemExit(main())

python -B -X utf8 -m unittest -v test_example