Convertir una lista JSON a CSV y comprobar los campos ausentes
Convierta los datos en una tabla distinguiendo entre claves ausentes, valores null y cadenas vacías, y genere un informe de incidencias por separado.
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. 한국어
Para quién esPrincipiantes en Python que quieran ver un archivo JSON como una tabla y localizar los datos que faltan
Preparación
Prepare Python 3.12 o posterior y una terminal. No se necesitan paquetes adicionales.
Descomprima el ZIP y abra una terminal en la carpeta que contiene example.py y sample.json.
Para empezar, utilice únicamente los datos de artículos ficticios incluidos. Conserve los archivos originales por separado.
01Por qué distinguir entre campos ausentes y valores vacíos
En JSON, la ausencia de una clave, un valor null y una cadena vacía son estados distintos. Como todos pueden verse como celdas en blanco en un CSV, este ejemplo registra el estado original en un archivo field_issues.csv independiente. No rellena los datos con valores arbitrarios.
Las columnas esperadas son item, quantity y unit. Los valores 0, false y un único espacio no se consideran ausentes. Primero se comprueba si existe la clave y después se revisa el estado de su valor.
Hay 4 registros. En el segundo falta la clave unit y en el tercero falta la clave quantity. En el cuarto, quantity es null y unit es una cadena vacía. Los nombres de los artículos y las unidades son valores reales de entrada, por lo que se muestran sin cambios también en las versiones traducidas.
Las rutas relativas del comando se resuelven a partir de la carpeta actual de la terminal. Antes de ejecutarlo, vaya a la carpeta que contiene example.py. La carpeta de salida todavía no debe existir, y su carpeta superior debe existir. Si outputs ya existe, elija otro nombre.
La carpeta nueva se crea una vez validada 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 corresponden al número de celdas con problemas, no al número de filas. La cuarta fila tiene dos problemas, por lo que se notifican 4 incidencias en total. Compruebe primero el CSV en un editor de texto. Los datos externos arbitrarios pueden ser interpretados automáticamente por una hoja de cálculo, y este código no sanea las cadenas de fórmulas.
05Código completo y flujo de procesamiento
Solo se admiten objetos dentro de un array JSON. Se rechazan los objetos o arrays anidados, las columnas desconocidas, las claves JSON duplicadas y los valores NaN e Infinity. Los números decimales se leen como Decimal para evitar conversiones innecesarias a coma flotante binaria. No se procesan archivos de entrada de más 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 utilizan carpetas temporales para comprobar la conversión normal, la distinción entre datos ausentes y valores vacíos, las entradas no válidas y la protección de las salidas existentes. Compruebe que el resultado final de las 22 pruebas sea OK. Superar las pruebas no significa que se admitan todos los formatos de datos reales.
07Identificar la causa a partir del mensaje de error
Mensaje
Qué comprobar
E_INPUT
Compruebe el formato JSON, los nombres de las columnas, los tipos de los valores y el tamaño de la entrada.
E_OUTPUT_EXISTS
Indique un nombre nuevo para la carpeta de salida.
E_READ
Compruebe la ruta de entrada y los permisos de lectura.
E_WRITE
Compruebe los permisos de escritura y el estado del disco. No utilice salidas incompletas.
08Alcance y limitaciones
Este ejemplo está pensado para archivos pequeños y mantiene en memoria toda la entrada y el contenido de la salida. No aplana JSON anidado, no convierte fechas, no infiere columnas automáticamente ni rellena valores ausentes. Al corregir errores, trabaje con una copia del archivo original destinada a la práctica.
Si se produce un error de disco al escribir la salida, puede que solo queden algunos archivos. Compruebe tanto el mensaje de éxito como el contenido de los tres archivos. Este ejemplo no verifica el comportamiento de conversión automática de las 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
22 pruebas superadas en una revisión independiente
Confirmada la coincidencia entre el código del artículo y el código descargable
Confirmadas las 4 filas de muestra y las 4 incidencias notificadas
Confirmados el rechazo de una carpeta de salida existente y la conservación de la entrada original
Límites de la verificación
El autor original realizó una verificación por separado en Linux con CPython 3.13.5.
No se han verificado todos los datos reales ni la interpretación automática de las 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 también las subcarpetas y registra la ruta, extensión, tamaño y fecha de modificación de cada archivo. Empieza con 4 archivos de ejemplo pequeños y conserva los originales y los resultados existentes.
Combina en orden archivos CSV con la misma estructura de columnas y añade la columna source_file. Comprueba con datos pequeños los nombres de artículos con comas, las columnas ausentes y las salidas ya existentes.
Revisa en una tabla los nombres actuales y los nuevos, y comprueba posibles conflictos. Solo al indicar --apply se crean copias en una carpeta nueva, conservando los originales.