Convertir una lista JSON a CSV y comprobar campos ausentes
Convierte los datos a una tabla diferenciando claves ausentes, valores null y cadenas vacías, y crea un informe separado de incidencias.
Contenido revisado 2026.09.19Incluye archivos de ejemplo
Ver el índice
La traducción se ha realizado con IA. Comprueba el código, las unidades y los valores junto con el original. La revisión por hablantes nativos de cada idioma aún no se ha completado. English
Para quién esPrincipiantes de Python que quieren ver un archivo JSON como una tabla y localizar dónde faltan datos
Preparación
Ten preparados Python 3.12 o posterior y una terminal. No se requieren paquetes adicionales.
Extrae el ZIP y abre una terminal en la carpeta que contiene example.py y sample.json.
Empieza únicamente con los datos sintéticos de items incluidos. Conserva los archivos originales por separado.
01¿Por qué distinguir campos ausentes de valores vacíos?
En JSON, una clave ausente, un valor null y una cadena vacía son estados diferentes. Como todos pueden aparecer como celdas en blanco en CSV, este ejemplo registra el estado original en un archivo field_issues.csv separado. No rellena valores de sustitución arbitrarios.
Las columnas esperadas son item, quantity y unit. Los valores 0, false y un único espacio no se consideran ausentes. El código comprueba primero si existe la clave y después comprueba el estado de su valor.
Hay 4 registros. Al segundo le falta la clave unit y al tercero le falta la clave quantity. En el cuarto, quantity es null y unit es una cadena vacía. Los nombres de item y las units son valores de entrada reales, por lo que se mantienen iguales en las versiones traducidas.
Las rutas relativas del comando se resuelven desde la carpeta actual de la terminal. Antes de ejecutarlo, ve a la carpeta que contiene example.py. La carpeta de salida todavía no debe existir y su carpeta padre ya debe existir. Si outputs ya existe, elige un nombre nuevo.
Se crea una carpeta nueva después de validar la entrada. El código no vacía ni sobrescribe una carpeta de salida existente y no modifica el JSON de entrada.
Los recuentos de incidencias se refieren a celdas con problemas, no al número de filas. La cuarta fila tiene dos incidencias, por lo que el total de incidencias notificadas es 4. Comprueba primero el CSV en un editor de texto. Los datos externos arbitrarios pueden ser interpretados automáticamente por el software de hojas de cálculo, y este código no sanea formula strings.
05Código completo y flujo de procesamiento
Solo se permiten objects dentro de un JSON array. Se rechazan nested objects o arrays, columns desconocidas, duplicate JSON keys, NaN e Infinity. El código lee los decimal numbers como Decimal para evitar conversiones innecesarias a binary floating point. No se procesan archivos de entrada mayores de 5 MiB.
example.py
#!/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())
06Ejecutar las pruebas incluidas
bash
python -B -X utf8 -m unittest -v test_example
Las pruebas usan temporary folders para comprobar la conversión normal, la distinción entre valores ausentes y vacíos, entradas no válidas y la protección de una salida existente. Comprueba que el resultado final de los 22 tests sea OK. Superar las pruebas no significa que se admitan todos los formatos de datos del mundo real.
07Encontrar la causa mediante el mensaje de error
Mensaje
Qué comprobar
E_INPUT
Comprueba JSON syntax, column names, value types y input size.
E_OUTPUT_EXISTS
Especifica un nuevo output folder name.
E_READ
Comprueba input path y read permissions.
E_WRITE
Comprueba write permissions y el estado del disk. No utilices una output incompleta.
08Alcance y limitaciones
Este ejemplo está pensado para archivos pequeños y mantiene en memoria todo el contenido de entrada y salida. No aplana nested JSON, no convierte fechas, no infiere columns automáticamente ni rellena missing values. Al corregir errores, trabaja con una copia de práctica del archivo original.
Si se produce un disk error mientras se escribe la salida, pueden quedar solo algunos archivos. Comprueba tanto el success message como el contenido de los tres archivos. Este ejemplo no verifica el comportamiento de conversión automática del software de hojas de cálculo al abrir archivos CSV.
Registro de ejecución y verificación
2026-09-19 · Windows 11 · CPython 3.12.14 · Biblioteca estándar
Se confirmó que 22 tests pasaron en una revisión independiente
Se confirmó que el código del artículo coincide con el código descargable
Se confirmaron 4 sample rows y 4 reported issues
Se confirmó el rechazo de una output folder existente y la conservación del input original
Límites de la verificación
El autor original realizó una verificación separada en Linux con CPython 3.13.5.
No se han verificado todos los datos del mundo real ni la interpretación automática del software de hojas de cálculo.
El código de ejemplo, los nombres de archivo y las claves de entrada se mantienen como en el original. Consulta también los comandos y los pasos de comprobación del texto traducido.
Material de práctica original · Guarda los archivos originales por separado antes de ejecutarlo.
Fuentes de referencia
Las explicaciones y los ejemplos son de elaboración propia. Puedes consultar los comportamientos y conceptos relacionados en las siguientes fuentes oficiales.
Recorre las subcarpetas y registra en una tabla las rutas, extensiones, tamaños y fechas de modificación de los archivos. Empieza con 4 archivos de ejemplo pequeños, manteniendo intactos los originales y los resultados existentes.
Combina en orden archivos CSV con la misma estructura de columnas y añade una columna source_file. Usa conjuntos de datos pequeños para comprobar nombres de elementos con comas, columnas faltantes y salidas ya existentes.
Compara los nombres antiguos y nuevos en una tabla y comprueba conflictos. Solo un --apply explícito crea copias en una carpeta nueva, manteniendo los originales.