Proyecto: script que procesa un CSV de ventas

1# generar_datos.py — Crea el CSV de ventas con datos sucios
2import csv
3import os
4
5os.makedirs("data", exist_ok=True)
6
7ventas = [
8    ["2024-03-15", "Laptop Pro 15", "electrónica", "2", "1299.99", "Ana García"],
9    ["2024-03-15", "Monitor 4K", "electrónica", "1", "449.00", "Luis Pérez"],
10    ["2024-03-15", "Camiseta Algodón", "ropa", "5", "24.99", "María López"],
11    ["2024-03-15", "Teclado Mecánico", "electrónica", "3", "89.99", "Pedro Ruiz"],
12    ["2024-03-15", "Pantalón Slim", "ropa", "2", "59.99", "Sara Torres"],
13    ["2024-03-15", "Auriculares BT", "electrónica", "4", "N/A", "Jorge Díaz"],  # precio inválido
14    ["2024-03-15", "Zapatillas Run", "calzado", "1", "129.99", "Ana García"],
15    ["2024-03-15", "", "ropa", "3", "34.99", "Carlos Ruiz"],  # producto vacío
16    ["2024-03-15", "Mochila Pro", "accesorios", "2", "79.99", "Luis Pérez"],
17    ["2024-03-15", "Portátil Air", "electrónica", "1", "999.00", "María López"],
18    ["2024-03-15", "Jersey Lana", "ropa", "4", "45.50", "Pedro Ruiz"],
19    ["2024-03-15", "Ratón Gaming", "electrónica", "", "49.99", "Sara Torres"],  # cantidad vacía
20    ["2024-03-15", "Botas Montaña", "calzado", "1", "189.00", "Jorge Díaz"],
21    ["2024-03-15", "USB-C Hub", "electrónica", "6", "35.99", "Ana García"],
22    ["2024-03-15", "Sudadera Tech", "ropa", "2", "69.99", "Carlos Ruiz"],
23]
24
25with open("data/ventas_dia.csv", "w", encoding="utf-8", newline="") as f:
26    writer = csv.writer(f)
27    writer.writerow(["fecha", "producto", "categoria", "cantidad", "precio_unitario", "cliente"])
28    writer.writerows(ventas)
29
30print(f"[OK] CSV generado: data/ventas_dia.csv ({len(ventas)} filas)")

Paso 1: Generamos el CSV con datos sucios (precios inválidos, campos vacíos).

1# pipeline_ventas.py — Tu primer script de limpieza y análisis
2import csv
3import json
4from datetime import datetime
5from collections import defaultdict
6
7
8# ═══ EXTRACT: leer datos ═══════════════════════════════════════
9def extraer_ventas(ruta_csv):
10    """Lee el CSV y devuelve una lista de diccionarios (datos crudos)."""
11    with open(ruta_csv, "r", encoding="utf-8") as f:
12        reader = csv.DictReader(f)
13        return list(reader)
14
15
16# ═══ TRANSFORM: limpiar y calcular ═════════════════════════════
17def transformar_ventas(ventas_raw):
18    """Limpia datos sucios y calcula totales. Devuelve (limpias, errores)."""
19    limpias = []
20    errores = []
21    
22    for i, venta in enumerate(ventas_raw):
23        try:
24            # .get() da None si la columna existe pero viene vacía en esa fila,
25            # y None no tiene .strip(). El "or ''" lo convierte en texto vacío.
26            if not (venta.get("producto") or "").strip():
27                raise ValueError("Producto vacío")
28            
29            # Convertir cantidad (default 1 si vacío)
30            cantidad_raw = (venta.get("cantidad") or "").strip()
31            cantidad = int(cantidad_raw) if cantidad_raw else 1
32            
33            # Convertir precio. .get() en vez de venta["precio_unitario"] para
34            # que una columna ausente no lance KeyError; si viene vacía o None,
35            # float() falla con ValueError/TypeError y la fila va a errores.
36            precio = float(venta.get("precio_unitario"))
37            
38            # Calcular total
39            total = cantidad * precio
40            
41            # Registro limpio
42            limpias.append({
43                "fecha": (venta.get("fecha") or "").strip(),
44                "producto": (venta.get("producto") or "").strip(),
45                "categoria": (venta.get("categoria") or "").strip(),
46                "cantidad": cantidad,
47                "precio_unitario": precio,
48                "total": round(total, 2),
49                "cliente": (venta.get("cliente") or "").strip(),
50            })
51        except (ValueError, TypeError, AttributeError, KeyError) as e:
52            errores.append({
53                "fila": i + 2,  # +2 por header y 0-index
54                "error": str(e),
55                "datos": dict(venta),
56            })
57    
58    return limpias, errores
59
60
61# ═══ ANALYZE: generar métricas ═════════════════════════════════
62def generar_resumen(ventas_limpias):
63    """Genera el resumen ejecutivo a partir de datos limpios."""
64    if not ventas_limpias:
65        return {"error": "No hay datos válidos"}
66    
67    # Métricas generales
68    total_facturado = sum(v["total"] for v in ventas_limpias)
69    num_transacciones = len(ventas_limpias)
70    ticket_medio = total_facturado / num_transacciones
71    
72    # Por categoría
73    por_categoria = defaultdict(lambda: {"total": 0, "transacciones": 0})
74    for v in ventas_limpias:
75        cat = v["categoria"]
76        por_categoria[cat]["total"] += v["total"]
77        por_categoria[cat]["transacciones"] += 1
78    
79    # Top productos
80    productos_total = defaultdict(float)
81    for v in ventas_limpias:
82        productos_total[v["producto"]] += v["total"]
83    
84    top_productos = sorted(productos_total.items(), key=lambda x: x[1], reverse=True)[:3]
85    
86    # Clientes únicos
87    clientes = set(v["cliente"] for v in ventas_limpias)
88    
89    return {
90        "fecha_reporte": datetime.now().strftime("%Y-%m-%d %H:%M"),
91        "total_facturado": round(total_facturado, 2),
92        "num_transacciones": num_transacciones,
93        "ticket_medio": round(ticket_medio, 2),
94        "clientes_unicos": len(clientes),
95        "por_categoria": dict(por_categoria),
96        "top_3_productos": [
97            {"producto": p, "total": round(t, 2)} for p, t in top_productos
98        ],
99    }
100
101
102# ═══ LOAD: escribir resultados ═════════════════════════════════
103def guardar_resultado(resumen, errores, ruta_salida, ruta_errores):
104    """Escribe el resumen y los errores en archivos JSON."""
105    with open(ruta_salida, "w", encoding="utf-8") as f:
106        json.dump(resumen, f, indent=2, ensure_ascii=False)
107    
108    if errores:
109        with open(ruta_errores, "w", encoding="utf-8") as f:
110            json.dump(errores, f, indent=2, ensure_ascii=False)

El pipeline completo: extraer, transformar, analizar y cargar.

1# ═══ MAIN: orquestar el pipeline ══════════════════════════════
2def main():
3    """Punto de entrada del pipeline."""
4    print("=" * 50)
5    print("  PIPELINE DE VENTAS DIARIAS")
6    print("=" * 50)
7    
8    # Configuración
9    ruta_entrada = "data/ventas_dia.csv"
10    ruta_salida = "data/resumen_dia.json"
11    ruta_errores = "data/errores_dia.json"
12    
13    # 1. EXTRACT
14    print("\n[1/4] Extrayendo datos...")
15    ventas_raw = extraer_ventas(ruta_entrada)
16    print(f"      Leídos: {len(ventas_raw)} registros")
17    
18    # 2. TRANSFORM
19    print("[2/4] Limpiando y transformando...")
20    ventas_limpias, errores = transformar_ventas(ventas_raw)
21    tasa_exito = len(ventas_limpias) / len(ventas_raw) * 100
22    print(f"      Válidos: {len(ventas_limpias)} ({tasa_exito:.0f}%)")
23    print(f"      Errores: {len(errores)}")
24    
25    # 3. ANALYZE
26    print("[3/4] Generando resumen ejecutivo...")
27    resumen = generar_resumen(ventas_limpias)
28    print(f"      Total facturado: {resumen['total_facturado']}€")
29    
30    # 4. LOAD
31    print("[4/4] Guardando resultados...")
32    guardar_resultado(resumen, errores, ruta_salida, ruta_errores)
33    print(f"      Resumen: {ruta_salida}")
34    if errores:
35        print(f"      Errores: {ruta_errores}")
36    
37    # Reporte final
38    print("\n" + "=" * 50)
39    print("  RESUMEN EJECUTIVO")
40    print("=" * 50)
41    print(f"  Facturación total: {resumen['total_facturado']}€")
42    print(f"  Ticket medio:      {resumen['ticket_medio']}€")
43    print(f"  Transacciones:     {resumen['num_transacciones']}")
44    print(f"  Clientes únicos:   {resumen['clientes_unicos']}")
45    print(f"\n  Top 3 productos:")
46    for item in resumen["top_3_productos"]:
47        print(f"    • {item['producto']}: {item['total']}€")
48    print("=" * 50)
49
50
51if __name__ == "__main__":
52    main()

El main orquesta todo: extrae → transforma → analiza → guarda.

1python generar_datos.py
2python pipeline_ventas.py
3
4# Verifica los archivos generados:
5# data/resumen_dia.json  ← resumen ejecutivo
6# data/errores_dia.json  ← registros que fallaron

Primero genera los datos, luego ejecuta el pipeline.