Curso de Python / Proyecto de informes CSV con validación y filas rechazadas

Un informe CSV debe validar encabezados, campos, importes e identificadores duplicados antes de sumar datos. Este proyecto conserva las filas rechazadas en un informe separado y devuelve un estado de error para que un total parcial no se confunda con un procesamiento completo.

Definir entradas y política de rechazo

Cada gasto tiene un id, una categoría y un importe entero no negativo escrito con dígitos decimales. La primera aparición válida aceptada reserva el identificador; una aparición posterior del mismo identificador se rechaza. El programa no modifica la entrada.

Utiliza una carpeta nueva. Si ya existen report.csv o rejected.csv, la ejecución se detiene. Los números del informe de rechazo cuentan registros CSV lógicos, con el encabezado como registro uno. Un campo entre comillas con saltos de línea puede ocupar varias líneas físicas y seguir siendo un único registro.

Preparar los datos de prueba

Guarda expenses-input.csv con este contenido. El identificador duplicado y el importe negativo son intencionados para comprobar los rechazos.

id,category,amount
e1,food,3500
e2,food,8000
e3,transport,2500
e2,food,8000
e4,books,-1

Los tres primeros gastos suman 14000. El cuarto repite e2 y el quinto incumple el formato de importe. No elimines esos casos antes de observar qué informa la herramienta. La lección del formato CSV explica por qué un lector específico es preferible a separar cada línea manualmente por comas.

Ejecutar el programa completo y revisar sus archivos

Guarda el siguiente código Python como csv_report.py en la misma carpeta:

import csv
import re
from collections import defaultdict
from pathlib import Path

source = Path("expenses-input.csv")
report = Path("report.csv")
rejected = Path("rejected.csv")
if report.exists() or rejected.exists():
    raise FileExistsError("Use a new output directory")
totals = defaultdict(int)
seen = set()
errors = []
accepted = 0
with source.open(encoding="utf-8-sig", newline="") as f:
    reader = csv.DictReader(f)
    if reader.fieldnames != ["id", "category", "amount"]:
        raise ValueError("unexpected header")
    for number, row in enumerate(reader, start=2):
        key = (row.get("id") or "").strip()
        category = (row.get("category") or "").strip()
        amount_text = (row.get("amount") or "").strip()
        reason = ""
        if None in row or not key or not category:
            reason = "invalid fields"
        elif key in seen:
            reason = "duplicate id"
        elif not re.fullmatch(r"[0-9]+", amount_text):
            reason = "invalid amount"
        if reason:
            errors.append({"record": number, "reason": reason})
            continue
        seen.add(key)
        totals[category] += int(amount_text)
        accepted += 1
with report.open("x", encoding="utf-8", newline="") as f:
    writer = csv.writer(f)
    writer.writerow(["category", "total"])
    writer.writerows(sorted(totals.items()))
with rejected.open("x", encoding="utf-8", newline="") as f:
    writer = csv.DictWriter(f, fieldnames=["record", "reason"])
    writer.writeheader()
    writer.writerows(errors)
print("accepted:", accepted, "rejected:", len(errors))
print("total:", sum(totals.values()))
raise SystemExit(1 if errors else 0)

Ejecuta el programa con:

python csv_report.py

Debe mostrar accepted: 3 rejected: 2 y total: 14000. El estado de salida es uno porque hubo rechazos, aunque se hayan creado informes. Este es el contenido completo de report.csv:

category,total
food,11500
transport,2500

Y este es rejected.csv:

record,reason
5,duplicate id
6,invalid amount

Los registros cinco y seis corresponden al duplicado y al importe negativo. La suma de comida es 11500 y la de transporte 2500. Reconcilia esos valores con la entrada en lugar de interpretar la existencia del archivo como prueba suficiente de éxito.

Comprobar repeticiones y resultados parciales

1. Repite sin cambiar la carpeta y comprueba que se rechazan las salidas existentes.
2. En otra carpeta nueva, corrige el duplicado y el importe negativo; verifica que el estado final sea cero.
3. Cambia el orden del encabezado para observar la regla estricta de columnas.

La creación de dos archivos no es una transacción atómica. Si falla la segunda escritura, puede quedar el primer informe. Un proceso más amplio puede preparar las salidas en un directorio temporal y distribuirlas solo después de verificarlas y marcar su finalización. No envíes automáticamente una salida parcial al siguiente proceso.

Si las categorías proceden de terceros y el destino es una hoja de cálculo, trata la posible interpretación como fórmulas según la aplicación receptora. La referencia oficial de csv explica lectura y escritura; las pruebas de integración permiten comprobar informes y estados en carpetas temporales.