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            # Validar producto no vacío
25            if not venta.get("producto", "").strip():
26                raise ValueError("Producto vacío")
27            
28            # Convertir cantidad (default 1 si vacío)
29            cantidad_raw = venta.get("cantidad", "").strip()
30            cantidad = int(cantidad_raw) if cantidad_raw else 1
31            
32            # Convertir precio
33            precio = float(venta["precio_unitario"])
34            
35            # Calcular total
36            total = cantidad * precio
37            
38            # Registro limpio
39            limpias.append({
40                "fecha": venta["fecha"],
41                "producto": venta["producto"].strip(),
42                "categoria": venta["categoria"].strip(),
43                "cantidad": cantidad,
44                "precio_unitario": precio,
45                "total": round(total, 2),
46                "cliente": venta["cliente"].strip(),
47            })
48        except (ValueError, TypeError) as e:
49            errores.append({
50                "fila": i + 2,  # +2 por header y 0-index
51                "error": str(e),
52                "datos": dict(venta),
53            })
54    
55    return limpias, errores
56
57
58# ═══ ANALYZE: generar métricas ═════════════════════════════════
59def generar_resumen(ventas_limpias):
60    """Genera el resumen ejecutivo a partir de datos limpios."""
61    if not ventas_limpias:
62        return {"error": "No hay datos válidos"}
63    
64    # Métricas generales
65    total_facturado = sum(v["total"] for v in ventas_limpias)
66    num_transacciones = len(ventas_limpias)
67    ticket_medio = total_facturado / num_transacciones
68    
69    # Por categoría
70    por_categoria = defaultdict(lambda: {"total": 0, "transacciones": 0})
71    for v in ventas_limpias:
72        cat = v["categoria"]
73        por_categoria[cat]["total"] += v["total"]
74        por_categoria[cat]["transacciones"] += 1
75    
76    # Top productos
77    productos_total = defaultdict(float)
78    for v in ventas_limpias:
79        productos_total[v["producto"]] += v["total"]
80    
81    top_productos = sorted(productos_total.items(), key=lambda x: x[1], reverse=True)[:3]
82    
83    # Clientes únicos
84    clientes = set(v["cliente"] for v in ventas_limpias)
85    
86    return {
87        "fecha_reporte": datetime.now().strftime("%Y-%m-%d %H:%M"),
88        "total_facturado": round(total_facturado, 2),
89        "num_transacciones": num_transacciones,
90        "ticket_medio": round(ticket_medio, 2),
91        "clientes_unicos": len(clientes),
92        "por_categoria": dict(por_categoria),
93        "top_3_productos": [
94            {"producto": p, "total": round(t, 2)} for p, t in top_productos
95        ],
96    }
97
98
99# ═══ LOAD: escribir resultados ═════════════════════════════════
100def guardar_resultado(resumen, errores, ruta_salida, ruta_errores):
101    """Escribe el resumen y los errores en archivos JSON."""
102    with open(ruta_salida, "w", encoding="utf-8") as f:
103        json.dump(resumen, f, indent=2, ensure_ascii=False)
104    
105    if errores:
106        with open(ruta_errores, "w", encoding="utf-8") as f:
107            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.