¿Cómo construir un proyecto completo de automatización en Python? | Automatización con Python

Un proyecto completo de automatización combina descarga de datos, procesamiento y envío de reportes en un solo flujo coordinado y mantenible.

Estructura del proyecto

Antes de escribir código, organiza tu carpeta de trabajo. Una estructura clara evita errores y facilita el mantenimiento.

reporte_ventas/
├── main.py
├── config.py
├── descargador.py
├── procesador.py
├── reporteador.py
├── logs/
│   └── ejecucion.log
├── datos/
│   └── ventas_raw.csv
├── reportes/
│   └── reporte_final.xlsx
└── .env

Cada archivo tiene una sola responsabilidad. main.py orquesta todo. Los demás módulos hacen una tarea específica.

El archivo de configuración

config.py centraliza todas las variables del proyecto. Nunca escribas contraseñas ni rutas directamente en el código principal.

# config.py
import os
from dotenv import load_dotenv

load_dotenv()  # Carga el archivo .env

# Rutas absolutas
BASE_DIR = os.path.dirname(os.path.abspath(__file__))
DIR_DATOS = os.path.join(BASE_DIR, "datos")
DIR_REPORTES = os.path.join(BASE_DIR, "reportes")
DIR_LOGS = os.path.join(BASE_DIR, "logs")

# Credenciales de correo
CORREO_ORIGEN = os.getenv("CORREO_ORIGEN")
CORREO_CLAVE = os.getenv("CORREO_CLAVE")
CORREO_DESTINO = os.getenv("CORREO_DESTINO")

# URL de datos
URL_VENTAS = os.getenv("URL_VENTAS", "https://api.empresa.com.mx/ventas")

El archivo .env guarda los valores reales. Agrégalo a tu .gitignore para no exponerlo en repositorios.

CORREO_ORIGEN=reportes@miempresa.com.mx
CORREO_CLAVE=clave_secreta_123
CORREO_DESTINO=gerencia@miempresa.com.mx
URL_VENTAS=https://api.femsa.com/ventas

Módulo 1: Descarga de datos

descargador.py obtiene los datos desde una fuente externa y los guarda localmente.

# descargador.py
import requests
import os
import logging
from config import URL_VENTAS, DIR_DATOS

def descargar_ventas():
    ruta_archivo = os.path.join(DIR_DATOS, "ventas_raw.csv")
    logging.info("Iniciando descarga de datos de ventas...")

    try:
        respuesta = requests.get(URL_VENTAS, timeout=30)
        respuesta.raise_for_status()

        with open(ruta_archivo, "wb") as f:
            f.write(respuesta.content)

        logging.info(f"Archivo guardado en: {ruta_archivo}")
        return ruta_archivo

    except requests.exceptions.Timeout:
        logging.error("La conexión tardó demasiado. Verifica tu red.")
        raise
    except requests.exceptions.HTTPError as e:
        logging.error(f"Error HTTP al descargar datos: {e}")
        raise

Siempre usa timeout en tus peticiones. Sin él, el script puede quedarse colgado indefinidamente.

Módulo 2: Procesamiento de datos

procesador.py limpia, filtra y resume los datos descargados.

# procesador.py
import pandas as pd
import os
import logging
from config import DIR_DATOS, DIR_REPORTES

def procesar_ventas(ruta_raw):
    logging.info("Procesando datos de ventas...")

    df = pd.read_csv(ruta_raw)

    # Limpieza básica
    df.columns = df.columns.str.strip().str.lower()
    df.dropna(subset=["monto", "sucursal"], inplace=True)
    df["monto"] = pd.to_numeric(df["monto"], errors="coerce")
    df.dropna(subset=["monto"], inplace=True)

    # Resumen por sucursal
    resumen = df.groupby("sucursal")["monto"].agg(
        total="sum",
        promedio="mean",
        transacciones="count"
    ).reset_index()

    resumen["total"] = resumen["total"].round(2)
    resumen["promedio"] = resumen["promedio"].round(2)
    resumen = resumen.sort_values("total", ascending=False)

    ruta_reporte = os.path.join(DIR_REPORTES, "reporte_final.xlsx")
    resumen.to_excel(ruta_reporte, index=False)

    logging.info(f"Reporte guardado en: {ruta_reporte}")
    return ruta_reporte, resumen

El módulo regresa tanto la ruta del archivo como el DataFrame. Así main.py puede usar los datos para el correo sin leer el archivo de nuevo.

Módulo 3: Envío del reporte por correo

reporteador.py construye y envía el correo con el reporte adjunto.

# reporteador.py
import smtplib
import logging
from email.mime.multipart import MIMEMultipart
from email.mime.base import MIMEBase
from email.mime.text import MIMEText
from email import encoders
from config import CORREO_ORIGEN, CORREO_CLAVE, CORREO_DESTINO

def enviar_reporte(ruta_reporte, resumen):
    logging.info("Preparando envío del reporte por correo...")

    top3 = resumen.head(3)
    filas_html = ""
    for _, fila in top3.iterrows():
        total_fmt = f"${fila['total']:,.0f}"
        promedio_fmt = f"${fila['promedio']:,.0f}"
        filas_html += f"<tr><td>{fila['sucursal']}</td><td>{total_fmt}</td><td>{promedio_fmt}</td></tr>"

    cuerpo_html = f"""
    <h2>Reporte de Ventas por Sucursal</h2>
    <p>Las 3 sucursales con mayor venta del periodo:</p>
    <table border=\"1\" cellpadding=\"5\">
      <tr><th>Sucursal</th><th>Total</th><th>Promedio</th></tr>
      {filas_html}
    </table>
    <p>Consulta el archivo adjunto para el detalle completo.</p>
    """

    msg = MIMEMultipart()
    msg["From"] = CORREO_ORIGEN
    msg["To"] = CORREO_DESTINO
    msg["Subject"] = "Reporte automático de ventas"
    msg.attach(MIMEText(cuerpo_html, "html"))

    with open(ruta_reporte, "rb") as f:
        adjunto = MIMEBase("application", "octet-stream")
        adjunto.set_payload(f.read())
    encoders.encode_base64(adjunto)
    adjunto.add_header("Content-Disposition", f"attachment; filename=reporte_final.xlsx")
    msg.attach(adjunto)

    with smtplib.SMTP_SSL("smtp.gmail.com", 465) as servidor:
        servidor.login(CORREO_ORIGEN, CORREO_CLAVE)
        servidor.sendmail(CORREO_ORIGEN, CORREO_DESTINO, msg.as_string())

    logging.info("Correo enviado correctamente.")

El orquestador principal

main.py conecta todos los módulos en orden. Maneja errores globales y registra cada etapa.

# main.py
import logging
import os
import sys
from logging.handlers import RotatingFileHandler
from config import DIR_LOGS
from descargador import descargar_ventas
from procesador import procesar_ventas
from reporteador import enviar_reporte

# Configuración del log
os.makedirs(DIR_LOGS, exist_ok=True)
log_path = os.path.join(DIR_LOGS, "ejecucion.log")

handler = RotatingFileHandler(log_path, maxBytes=1_000_000, backupCount=3)
logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s [%(levelname)s] %(message)s",
    handlers=[handler, logging.StreamHandler(sys.stdout)]
)

def main():
    logging.info("=== Inicio del proceso de reporte ===")
    try:
        ruta_raw = descargar_ventas()
        ruta_reporte, resumen = procesar_ventas(ruta_raw)
        enviar_reporte(ruta_reporte, resumen)
        logging.info("=== Proceso completado exitosamente ===")
    except Exception as e:
        logging.critical(f"El proceso falló: {e}")
        sys.exit(1)

if __name__ == "__main__":
    main()

Usa sys.exit(1) cuando el proceso falla. Así cron o el Programador de tareas detecta el error y puede notificarte.

Errores comunes

Error 1: Importaciones circulares. Si procesador.py importa de reporteador.py y viceversa, Python lanza un error. Mantén un flujo de una sola dirección: main.py → módulos especializados.

Error 2: Rutas relativas en producción. Usar "datos/ventas.csv" funciona en tu máquina pero falla en cron. Siempre construye rutas con os.path.join(BASE_DIR, ...) como se muestra en config.py.

Error 3: No crear carpetas antes de escribir archivos. Si datos/ o reportes/ no existen, Python lanza FileNotFoundError. Agrega os.makedirs(DIR_DATOS, exist_ok=True) al inicio de main.py para cada carpeta necesaria.

Error 4: Credenciales en el código fuente. Subir config.py con contraseñas a GitHub expone tus datos. Usa siempre .env con python-dotenv y verifica que .env esté en .gitignore.

Tabla de responsabilidades de cada módulo

Archivo	Responsabilidad	Depende de
`config.py`	Variables globales y rutas	`.env`, `os`, `dotenv`
`descargador.py`	Obtener datos externos	`config.py`, `requests`
`procesador.py`	Limpiar y resumir datos	`config.py`, `pandas`
`reporteador.py`	Construir y enviar correo	`config.py`, `smtplib`
`main.py`	Orquestar el flujo completo	Todos los módulos

Cómo agendar el proyecto

Una vez que el proyecto funciona manualmente, agrégalo a cron (Linux/macOS) o al Programador de tareas (Windows).

Ejemplo para ejecutar cada lunes a las 7:00 AM:

0 7 * * 1 /usr/bin/python3 /home/usuario/reporte_ventas/main.py >> /home/usuario/reporte_ventas/logs/cron.log 2>&1

Usa la ruta absoluta de Python (which python3) y la ruta completa de main.py.

Próximos pasos

Este proyecto base escala fácilmente. Estas son las mejoras más valiosas para aplicar después:

Agrega notificaciones por Slack o WhatsApp Business cuando el proceso falla, usando webhooks.
Usa una base de datos SQLite para guardar el historial de reportes y detectar tendencias.
Dockeriza el proyecto para ejecutarlo en cualquier servidor sin instalar dependencias manualmente.
Conecta con la API del SAT o del IMSS para enriquecer los datos con información fiscal o de nómina.
Agrega pruebas automatizadas con pytest para verificar que cada módulo funciona antes de desplegarlo.

Con estas mejoras, tu proyecto pasa de un script a una herramienta profesional lista para entornos empresariales como los de Bimbo, Liverpool o FEMSA.