Programación y Automatización

Se trata de un artículo extenso sobre programación y automatización, que incluye preguntas de examen, scripts de ejemplo, componentes clave, herramientas de monitoreo y mucho más.
¡Revisa los encabezados!

En resumen

La automatización mediante programación reduce la carga de los procesos de TI, mejora la eficiencia y minimiza errores. Para lograrlo se utilizan scripts, lenguajes de programación y herramientas de automatización de manera estratégica.

Descripción técnica compacta

Las tareas recurrentes como backups, gestión de usuarios o deployments pueden automatizarse mediante scripts (por ejemplo, Bash, PowerShell, Python). A esto se suman la programación de tareas (Cron, Task Scheduler), monitoreo, logging y verificaciones de seguridad.

También existen herramientas específicas de cada fabricante que utilizan sus propios componentes de scripts (por ejemplo, VMware PowerCLI). Frecuentemente también es necesario analizar código existente y adaptarlo de manera específica para integrar la automatización en sistemas en funcionamiento.

Necesitarás editar archivos que creas o que ya existan. En ocasiones hay que reemplazar numerosos “formatos” u otros elementos mediante variables, o buscar y eliminar un enlace. Aquí es donde pequeños y simples scripts en Python, Bash o PowerShell resultan útiles.

Puntos clave relevantes para el examen

Seleccionar lenguajes de programación de forma estratégica

La elección del lenguaje de programación correcto es crucial para el éxito de la automatización. Bash es adecuado para tareas simples del sistema bajo Linux, mientras que Python es preferible para lógica compleja y soluciones multiplataforma. PowerShell es la opción recomendada para entornos Windows con integraciones API profundas.

La automatización reduce intervenciones manuales y errores

Los procesos automatizados minimizan las fuentes de error humano y garantizan resultados consistentes. Las tareas recurrentes como backups, creación de usuarios o deployments se ejecutan de forma confiable y trazable.

Lenguajes de script para tareas del sistema y lógica de procesos

Los lenguajes de script permiten automatizar tareas del sistema (gestión de archivos, control de procesos) y lógica de procesos compleja (condicionales, bucles, manejo de errores). Ofrecen desarrollo rápido y opciones de adaptación sencillas.

Herramientas de plataforma: Bash (Linux), PowerShell (Windows)

Bash es el estándar en Linux para shell scripting y automatización del sistema. PowerShell proporciona procesamiento basado en objetos e integración profunda en sistemas Windows y APIs. Ambas son relevantes para administradores de sistemas.

Supervisión mediante logs, códigos de salida y triggers

Los procesos automatizados requieren supervisión. El logging documenta todas las acciones, los códigos de salida indican éxito o fallo, y los triggers permiten automatización dirigida por eventos. Esto proporciona transparencia y seguridad contra fallos.

Procesos sensibles solo con logging y verificación de roles

Para automatizaciones sensibles (por ejemplo, derechos de usuario, cambios del sistema) son imprescindibles el logging y la verificación de roles. Solo usuarios autorizados pueden ejecutar scripts críticos, y todas las acciones deben registrarse.

Menor esfuerzo en mantenimiento y administración del sistema

La automatización reduce considerablemente el esfuerzo manual en mantenimiento y administración del sistema. Los scripts desarrollados una sola vez pueden reutilizarse múltiples veces y adaptarse fácilmente, lo que ahorra tiempo y recursos.

Documentar y versionar los cambios

Todos los cambios en scripts de automatización deben documentarse y versionarse. Esto garantiza trazabilidad, permite reversiones y respalda la colaboración en equipo.

Componentes clave

1. Selección del lenguaje de programación

La elección del lenguaje de programación depende de la tarea y la plataforma. Python es adecuado para lógica compleja y soluciones multiplataforma, Bash para tareas del sistema en Linux, PowerShell para entornos Windows.

2. Herramientas de scripting dependientes de la plataforma

Cada plataforma ofrece herramientas de automatización específicas: Bash en Linux, PowerShell en Windows, Cron para programación de tareas en Linux, Task Scheduler en Windows. Estas herramientas permiten la integración directa con el sistema.

3. Definir objetivos de automatización

Antes de la implementación deben definirse objetivos claros: ¿qué tareas deben automatizarse? ¿Qué ganancias de eficiencia se esperan? ¿Qué requisitos de seguridad aplican?

4. Ejecución condicional y bucles

La automatización a menudo requiere lógica condicional (if/else) y bucles (for/while) para procesos recurrentes. Estas estructuras de control permiten automatización flexible y adaptativa.

5. Manejo de errores y logging

La automatización robusta necesita manejo exhaustivo de errores (try/catch, códigos de salida) y logging detallado. Esto asegura la estabilidad y trazabilidad de los procesos automatizados.

6. Tareas programadas por tiempo (Cron, Task Scheduler)

Cron en Linux y Task Scheduler en Windows permiten la ejecución de scripts según una programación temporal. Esto es ideal para tareas de mantenimiento periódicas como backups o limpieza.

7. Integración en sistemas existentes

Los scripts de automatización deben integrarse en sistemas existentes. Esto incluye APIs, bases de datos, sistemas de archivos y otros componentes de la infraestructura de TI.

8. Control de versiones de los scripts

Todos los scripts de automatización deben estar bajo control de versiones (Git). Esto permite el seguimiento de cambios, la colaboración y reversiones seguras.

9. Verificaciones de seguridad y conceptos de derechos

La automatización requiere conceptos de seguridad estrictos: verificación de permisos, gestión de roles, manejo seguro de contraseñas y logging de auditoría para operaciones sensibles.

10. Monitoreo de procesos automatizados

Los procesos automatizados deben supervisarse: verificación de éxito, notificación de errores, monitoreo de rendimiento y alertas ante desviaciones.

Ejemplo práctico simple (Backup en Python)

import shutil

shutil.copy('/home/user/data.db', '/backup/data.db')

Explicación: El script copia automáticamente un archivo a un directorio de backup y puede ejecutarse regularmente mediante un cronjob.

¿Por qué shutil en lugar de comandos copy normales?

shutil significa “shell utilities” (no shadow utility) y es un módulo estándar de Python para operaciones de archivos de alto nivel. Ofrece varias ventajas sobre comandos copy simples:

Ventajas de shutil.copy():

• Multiplataforma: Funciona idénticamente en Windows, Linux y macOS
• Manejo de errores: Genera excepciones limpias en caso de problemas (permisos, rutas inexistentes)
• Metadatos: Preserva permisos de archivos y metadatos (con shutil.copy2())
• Seguridad: Verifica automáticamente problemas de sobrescritura

Alternativas y por qué son peores:

# ❌ Malo: Específico del SO
import os
os.system('cp /home/user/data.db /backup/data.db')  # Solo Linux
os.system('copy C:\\data.db C:\\backup\\data.db')   # Solo Windows

# ❌ Malo: Sin manejo de errores
f = open('/home/user/data.db', 'rb')
data = f.read()
f.close()
f = open('/backup/data.db', 'wb')
f.write(data)
f.close()

# ✅ Bien: shutil.copy()
import shutil
shutil.copy('/home/user/data.db', '/backup/data.db')

shutil.copy() vs shutil.copy2():

• copy(): Copia contenido + permisos básicos
• copy2(): Copia contenido + todos los metadatos (timestamps, permisos) - generalmente la mejor opción para backups

Ventajas e inconvenientes

Ventajas

• Evitar errores mediante procesos automatizados
• Ahorro de tiempo en tareas recurrentes
• Trazabilidad a través de logs
• Portabilidad multiplataforma con lenguajes de scripting

Inconvenientes

• Los errores en el script pueden causar daños significativos
• Requiere pruebas y logging adecuado
• Comportamiento diferente según el entorno

Preguntas típicas de examen (con respuesta breve)

1. ¿Cuándo Bash en lugar de Python? Para tareas simples del sistema bajo Linux (archivos, procesos, pipes).
2. ¿Tareas típicas de automatización? Copias de seguridad, creación de usuarios, archivado de logs, deployment.
3. ¿Cómo hacer que la automatización sea segura? Logging, verificación de permisos, code reviews, evaluar exit status.
4. ¿Qué es un cronjob? Una tarea programada a tiempo en Linux.
5. ¿Ventajas de PowerShell frente a Bash? Basado en objetos e integrado profundamente en Windows y APIs.

Respuesta de desarrollo

El tema vincula la elección del lenguaje, la dependencia de la plataforma y el objetivo de automatización. En los exámenes suele ser relevante leer un script, extenderlo e implementar correctamente el manejo de errores y logging.

Estrategia de aprendizaje para este tema

1. Comprensión inicial: automatiza una copia de seguridad pequeña localmente

Solución:

# Script de copia de seguridad en Bash
#!/bin/bash
SOURCE="/home/user/documents"
TARGET="/backup/documents_$(date +%Y%m%d)"
mkdir -p "$TARGET"
cp -r "$SOURCE"/* "$TARGET/"
echo "Backup completado: $TARGET" >> /var/log/backup.log

Configurar cron: 0 2 * * * /ruta/al/backup.sh

2. Método de profundización: esboza un proceso de automatización (creación de usuarios + permisos)

Solución:

# Creación de usuarios en Python con verificación de permisos
import subprocess
import logging

logging.basicConfig(filename='user_management.log', level=logging.INFO)

def create_user(username, groups):
    try:
        # Crear usuario
        subprocess.run(['useradd', '-m', username], check=True)
        
        # Asignar grupos
        for group in groups:
            subprocess.run(['usermod', '-aG', group, username], check=True)
        
        logging.info(f"Usuario {username} creado con grupos: {groups}")
        print(f"Usuario {username} creado exitosamente")
        
    except subprocess.CalledProcessError as e:
        logging.error(f"Error al crear usuario {username}: {e}")
        print(f"Error: {e}")

# Uso
create_user("newuser", ["sudo", "developers"])

3. Entrenamiento enfocado en examen: analizar y complementar scripts

Solución:

• Análisis: verifica el script para manejo de errores, logging y seguridad
• Complemento: evalúa exit codes, valida parámetros, mejora mensajes de error
• Ejemplo de mejora:

import sys
import os

def validate_path(path):
    if not os.path.exists(path):
        print(f"Error: la ruta {path} no existe")
        sys.exit(1)
    return True

# Uso
if len(sys.argv) < 2:
    print("Uso: script.py <ruta>")
    sys.exit(1)

validate_path(sys.argv[1])
# Resto del script...

4. Evitar errores: implementa logging y verificación de permisos

Solución:

import logging
import os
import sys
from pathlib import Path

# Configurar logging
logging.basicConfig(
    level=logging.INFO,
    format='%(asctime)s - %(levelname)s - %(message)s',
    handlers=[
        logging.FileHandler('/var/log/automation.log'),
        logging.StreamHandler()
    ]
)

def check_permissions(required_uid=0):
    """Verifica si el script se ejecuta con los permisos correctos"""
    if os.geteuid() != required_uid:
        logging.error(f"El script debe ejecutarse como UID {required_uid}")
        sys.exit(1)

def safe_operation(operation_func):
    """Decorador para operaciones seguras"""
    def wrapper(*args, **kwargs):
        try:
            result = operation_func(*args, **kwargs)
            logging.info(f"Operación {operation_func.__name__} completada")
            return result
        except Exception as e:
            logging.error(f"Error en {operation_func.__name__}: {e}")
            raise
    return wrapper

@safe_operation
def critical_operation():
    # Operación crítica aquí
    pass

Análisis del tema

• Núcleo técnico: Shell, Python, scheduling, logging
• Desafíos de implementación: diferencias entre plataformas, prevención de errores
• Implicaciones de seguridad: asignación de permisos, abuso de scripts no seguros
• Obligaciones de documentación: control de versiones, registro de cambios
• Evaluación económica: menor esfuerzo manual gracias a la repetibilidad

Productos de software actuales

Plataformas de automatización

• Ansible: gestión de configuración y automatización (código abierto)
• Jenkins: pipelines CI/CD con soporte extensivo de plugins
• GitHub Actions: CI/CD nativo en la nube directamente en el repositorio
• GitLab CI/CD: solución de pipeline integrada con Auto DevOps

Ansible: Up and Running

Herramientas de scripting

• Visual Studio Code: editor moderno con soporte para Python/PowerShell
• PyCharm: IDE de Python con debugging y funciones de testing
• Windows Terminal: entorno terminal moderno para PowerShell
• WSL (Windows Subsystem for Linux): entorno Linux bajo Windows

Monitoreo y logging

• Prometheus: recopilación de métricas y alertas
• Grafana: visualización de datos de monitoreo
• ELK Stack: Elasticsearch, Logstash, Kibana para análisis de logs
• Nagios: sistema de monitoreo clásico

Automatización en la nube

• Azure Automation: Runbooks de PowerShell/Python en Azure
• AWS Lambda: funciones serverless para automatización
• Google Cloud Functions: automatización basada en eventos
• Terraform: infraestructura como código para múltiples nubes

Preguntas y respuestas

Pregunta 1: ¿Qué lenguaje para qué tarea de automatización?

Respuesta: Bash para tareas simples del sistema en Linux, PowerShell para integración con Windows, Python para lógica compleja y soluciones multiplataforma.

Pregunta 2: ¿Cómo asegurar los scripts de automatización?

Respuesta: control de versiones con Git, code reviews, manejo completo de excepciones, logging, verificación de permisos y auditorías de seguridad regulares.

Pregunta 3: ¿Cuáles son los errores típicos en la automatización?

Respuesta: manejo de errores ausente, logging insuficiente, hardcoding de rutas, falta de pruebas, ignorar exit codes.

Pregunta 4: ¿Cómo probar scripts de automatización?

Respuesta: unit tests para funciones, pruebas de integración para flujos, entornos staging para pruebas de producción, monitoreo en operación.

Pregunta 5: ¿Qué alternativas a cron existen?

Respuesta: systemd timers (alternativa más moderna), trabajos programados en Jenkins, Kubernetes CronJobs, schedulers específicos de nube (AWS EventBridge).

Información adicional

1. http://linuxcommand.org/lc3_learning_the_shell.php
2. https://learn.microsoft.com/de-de/powershell/
3. https://automatetheboringstuff.com/
4. https://docs.github.com/en/actions
5. https://crontab.guru/

Recomendación de libros

Amplía tu conocimiento sobre automatización con Python con este manual práctico orientado a principiantes y usuarios avanzados. Este libro ofrece soluciones concretas para tareas de automatización reales y te ayuda a desarrollar scripts eficientes.

Routineaufgaben mit Python automatisieren: Praktische Programmierlösungen für Einsteiger*innen

Este libro es ideal para:

• Principiantes que quieren aprender Python para automatización
• Administradores de sistemas que necesitan automatizar tareas recurrentes
• Desarrolladores que quieren optimizar sus flujos de trabajo diarios
• Cualquiera que busque soluciones prácticas en Python para problemas cotidianos

Colección de scripts pequeños para automatizaciones en Windows, Linux y Mac

Aquí hay 15 scripts Python prácticos para tareas de automatización típicas en diferentes plataformas.

Scripts Linux (5)

1. Reducir y optimizar imágenes

from PIL import Image
import os
from pathlib import Path

def resize_images(folder_path, max_size=1920, quality=85):
    """Reduce todas las imágenes en una carpeta"""
    for file in Path(folder_path).glob('*'):
        if file.suffix.lower() in ['.jpg', '.jpeg', '.png']:
            with Image.open(file) as img:
                img.thumbnail((max_size, max_size))
                img.save(file, quality=quality, optimize=True)
                print(f"Optimizado: {file.name}")

# Uso: resize_images('/home/user/Imágenes')

2. Encontrar archivos duplicados en una carpeta

import hashlib
from pathlib import Path
from collections import defaultdict

def find_duplicates(folder_path):
    """Encuentra archivos duplicados según hash MD5"""
    hashes = defaultdict(list)
    
    for file in Path(folder_path).rglob('*'):
        if file.is_file():
            file_hash = hashlib.md5(file.read_bytes()).hexdigest()
            hashes[file_hash].append(file)
    
    duplicates = {h: files for h, files in hashes.items() if len(files) > 1}
    
    for hash_val, files in duplicates.items():
        print(f"Duplicados encontrados ({len(files)} archivos):")
        for file in files:
            print(f"  - {file}")

# Uso: find_duplicates('/home/user/Descargas')

3. Limpiar archivos de log automáticamente

import os
from pathlib import Path
from datetime import datetime, timedelta

def clean_old_logs(log_folder, days=30):
    """Elimina archivos de log más antiguos que X días"""
    cutoff_date = datetime.now() - timedelta(days=days)
    
    for file in Path(log_folder).glob('*.log'):
        if datetime.fromtimestamp(file.stat().st_mtime) < cutoff_date:
            file.unlink()
            print(f"Eliminado: {file.name}")

# Uso: clean_old_logs('/var/log/myapp', days=30)

4. Fusionar archivos PDF

from PyPDF2 import PdfMerger
from pathlib import Path

def merge_pdfs(folder_path, output_name='combined.pdf'):
    """Combina todos los PDFs en una carpeta"""
    merger = PdfMerger()
    
    for file in sorted(Path(folder_path).glob('*.pdf')):
        merger.append(file)
        print(f"Añadido: {file.name}")
    
    merger.write(output_name)
    merger.close()
    print(f"Creado: {output_name}")

# Uso: merge_pdfs('/home/user/Documentos/PDFs')

5. Monitoreo del sistema y alertas

import psutil
import smtplib
from email.mime.text import MIMEText

def check_disk_usage(threshold=90):
    """Verifica uso de disco y envía alerta si se supera el umbral"""
    for partition in psutil.disk_partitions():
        usage = psutil.disk_usage(partition.mountpoint)
        percent = usage.percent
        
        if percent > threshold:
            print(f"ALERTA: {partition.mountpoint} está {percent}% lleno")
            # Aquí se podría enviar un correo
            # send_alert_email(partition.mountpoint, percent)

# Uso: check_disk_usage(threshold=90)

Scripts Windows (5)

1. Organizar el escritorio

import shutil
from pathlib import Path
from datetime import datetime

def clean_downloads():
    """Organiza archivos en la carpeta Descargas"""
    downloads = Path.home() / 'Downloads'
    
    # Crear carpetas
    folders = {
        'Imágenes': ['.jpg', '.jpeg', '.png', '.gif'],
        'Documentos': ['.pdf', '.doc', '.docx', '.txt'],
        'Música': ['.mp3', '.wav', '.flac'],
        'Videos': ['.mp4', '.avi', '.mkv'],
        'Archivos': ['.zip', '.rar', '.7z']
    }
    
    for folder, extensions in folders.items():
        target = downloads / folder
        target.mkdir(exist_ok=True)
        
        for file in downloads.glob('*'):
            if file.suffix.lower() in extensions and file.is_file():
                shutil.move(str(file), str(target / file.name))
                print(f"Movido: {file.name} -> {folder}")

# Uso: clean_downloads()

2. Eliminar archivos temporales

import shutil
import os
from pathlib import Path

def clean_temp_files():
    """Elimina archivos temporales y caché"""
    temp_folders = [
        Path(os.environ.get('TEMP', '')),
        Path(os.environ.get('TMP', '')),
        Path.home() / 'AppData' / 'Local' / 'Temp',
    ]
    
    for folder in temp_folders:
        if folder.exists():
            try:
                for file in folder.glob('*'):
                    if file.is_file():
                        file.unlink()
                print(f"Limpiado: {folder}")
            except PermissionError:
                print(f"Acceso denegado: {folder}")

# Uso: clean_temp_files()

3. Escanear redes WiFi

import subprocess

def scan_wifi_networks():
    """Muestra las redes WiFi disponibles"""
    try:
        result = subprocess.run(
            ['netsh', 'wlan', 'show', 'networks'],
            capture_output=True,
            text=True
        )
        
        networks = []
        for line in result.stdout.split('\n'):
            if 'SSID' in line:
                ssid = line.split(':')[1].strip()
                networks.append(ssid)
        
        print("Redes disponibles:")
        for network in networks:
            print(f"  - {network}")
            
    except Exception as e:
        print(f"Error: {e}")

# Uso: scan_wifi_networks()

4. Gestor del portapapeles

import pyperclip
import time
from datetime import datetime

def clipboard_history(max_entries=10):
    """Guarda el historial del portapapeles"""
    history = []
    last_content = ""
    
    print("Monitoreo del portapapeles iniciado (Ctrl+C para detener)")
    
    try:
        while True:
            current = pyperclip.paste()
            
            if current != last_content and current:
                timestamp = datetime.now().strftime("%H:%M:%S")
                history.insert(0, f"[{timestamp}] {current[:50]}...")
                
                if len(history) > max_entries:
                    history.pop()
                
                print(f"\nNueva entrada: {current[:30]}...")
                for entry in history:
                    print(f"  {entry}")
                
                last_content = current
            
            time.sleep(2)
            
    except KeyboardInterrupt:
        print("\nMonitoreo del portapapeles detenido")

# Uso: clipboard_history()

5. Captura de pantalla con tecla de acceso rápido

import pyautogui
import keyboard
from datetime import datetime
from pathlib import Path

def screenshot_on_hotkey(hotkey='f9'):
    """Captura una pantalla al presionar una tecla"""
    screenshots = Path.home() / 'Desktop' / 'Screenshots'
    screenshots.mkdir(exist_ok=True)
    
    def take_screenshot():
        timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
        filename = screenshots / f"screenshot_{timestamp}.png"
        pyautogui.screenshot(str(filename))
        print(f"Captura guardada: {filename}")
    
    keyboard.add_hotkey(hotkey, take_screenshot)
    print(f"Presiona {hotkey.upper()} para capturar pantalla")
    keyboard.wait()  # Espera presión de tecla

# Uso: screenshot_on_hotkey('f9')

Scripts para Mac (5)

1. Automatizar la organización del Finder

import shutil
from pathlib import Path

def organize_downloads():
    """Organiza la carpeta de Descargas en Mac"""
    downloads = Path.home() / 'Downloads'
    
    file_types = {
        'Bilder': ['.jpg', '.jpeg', '.png', '.heic', '.gif'],
        'Dokumente': ['.pdf', '.pages', '.doc', '.docx', '.txt'],
        'Musik': ['.mp3', '.m4a', '.wav', '.aiff'],
        'Videos': ['.mp4', '.mov', '.avi', '.mkv'],
        'Archiv': ['.zip', '.rar', '.7z', '.dmg']
    }
    
    for folder, extensions in file_types.items():
        target = downloads / folder
        target.mkdir(exist_ok=True)
        
        for file in downloads.glob('*'):
            if file.suffix.lower() in extensions and file.is_file():
                shutil.move(str(file), str(target / file.name))
                print(f"Organisiert: {file.name}")

# Nutzung: organize_downloads()

2. Enviar notificaciones en macOS

import subprocess

def send_notification(title, message):
    """Envía una notificación del sistema en macOS"""
    cmd = f"""
    osascript -e 'display notification "{message}" with title "{title}"'
    """
    subprocess.run(cmd, shell=True)

# Nutzung: send_notification("Backup fertig", "Dein Backup wurde erfolgreich abgeschlossen")

3. Controlar Spotify

import subprocess

def spotify_control(action):
    """Controla Spotify a través de AppleScript"""
    commands = {
        'play': 'tell application "Spotify" to play',
        'pause': 'tell application "Spotify" to pause',
        'next': 'tell application "Spotify" to next track',
        'previous': 'tell application "Spotify" to previous track'
    }
    
    if action in commands:
        cmd = f'osascript -e \'tell application "Spotify" to {commands[action]}\''
        subprocess.run(cmd, shell=True)
        print(f"Spotify: {action}")
    else:
        print(f"Unbekannter Befehl: {action}")

# Nutzung: spotify_control('play')

4. Controlar el brillo de la pantalla

import subprocess

def set_brightness(level):
    """Establece el brillo de la pantalla (0-100)"""
    if 0 <= level <= 100:
        cmd = f"""
        osascript -e 'tell application "System Events" to set brightness of every display to {level/100}'
        """
        subprocess.run(cmd, shell=True)
        print(f"Helligkeit auf {level}% gesetzt")
    else:
        print("Helligkeit muss zwischen 0 und 100 liegen")

# Nutzung: set_brightness(50)

5. Extraer texto de PDF

import PyPDF2
from pathlib import Path

def extract_pdf_text(pdf_path):
    """Extrae el texto de un archivo PDF"""
    text = ""
    
    with open(pdf_path, 'rb') as file:
        reader = PyPDF2.PdfReader(file)
        
        for page in reader.pages:
            text += page.extract_text() + "\n"
    
    return text

def save_pdf_text_to_file(pdf_path, output_path):
    """Guarda el texto extraído en un archivo"""
    text = extract_pdf_text(pdf_path)
    
    with open(output_path, 'w', encoding='utf-8') as file:
        file.write(text)
    
    print(f"Text extrahiert: {output_path}")

# Nutzung: save_pdf_text_to_file('/Users/user/Dokument.pdf', '/Users/user/Dokument.txt')

Notas sobre el uso

1. Instala los paquetes necesarios:

   pip install pillow pyperclip pyautogui pypdf2 psutil

2. Ten en cuenta los permisos: algunos scripts requieren derechos de administrador.

3. Adapta las rutas: modifica las rutas según tu entorno local.

4. Prueba primero: siempre ejecuta los scripts con datos de prueba antes.

5. Haz copias de seguridad: antes de ejecutar operaciones de eliminación, crea backups.

Colección de scripts Bash y PowerShell

Mac incluye Bash por defecto (o zsh en versiones recientes de macOS). Aquí hay 15 scripts adicionales en los lenguajes shell nativos.

Scripts Bash para Linux (5)

1. Backup automático con rotación

#!/bin/bash
# backup.sh - Automatisches Backup mit Rotation

SOURCE_DIR="/home/user/documents"
BACKUP_DIR="/backup"
MAX_BACKUPS=7
DATE=$(date +%Y%m%d_%H%M%S)
BACKUP_NAME="backup_$DATE.tar.gz"

# Backup erstellen
tar -czf "$BACKUP_DIR/$BACKUP_NAME" "$SOURCE_DIR"
echo "Backup erstellt: $BACKUP_NAME"

# Alte Backups löschen (Rotation)
cd "$BACKUP_DIR"
ls -t backup_*.tar.gz | tail -n +$((MAX_BACKUPS + 1)) | xargs -r rm
echo "Alte Backups bereinigt (max $MAX_BACKUPS behalten)"

2. Verificación del estado del servidor

#!/bin/bash
# server_check.sh - Prüft Server-Status und Services

echo "=== Server Status Check ==="
echo "Zeit: $(date)"
echo ""

# CPU-Last
echo "CPU-Last:"
top -bn1 | grep "Cpu(s)" | sed "s/.*, *\([0-9.]*\)%id.*/\1/" | awk '{print 100 - $1"%"}'

# Speicher
echo ""
echo "Speicher:"
free -h

# Festplatte
echo ""
echo "Festplatte:"
df -h

# Wichtige Services prüfen
echo ""
echo "Services:"
for service in nginx mysql docker; do
    if systemctl is-active --quiet "$service"; then
        echo "✓ $service läuft"
    else
        echo "✗ $service läuft nicht"
    fi
done

3. Analizar archivos de registro

#!/bin/bash
# log_analyzer.sh - Analysiert Log-Dateien auf Fehler

LOG_FILE="/var/log/syslog"
ERROR_COUNT=$(grep -i "error" "$LOG_FILE" | wc -l)
WARNING_COUNT=$(grep -i "warning" "$LOG_FILE" | wc -l)

echo "=== Log-Analyse für $LOG_FILE ==="
echo "Fehler: $ERROR_COUNT"
echo "Warnungen: $WARNING_COUNT"
echo ""

# Letzte 10 Fehler anzeigen
if [ $ERROR_COUNT -gt 0 ]; then
    echo "Letzte 10 Fehler:"
    grep -i "error" "$LOG_FILE" | tail -n 10
fi

4. Buscar archivos por tamaño

#!/bin/bash
# find_large_files.sh - Findet große Dateien

DIRECTORY="/home/user"
MIN_SIZE="100M"  # Mindestgröße

echo "=== Dateien größer als $MIN_SIZE in $DIRECTORY ==="
find "$DIRECTORY" -type f -size +"$MIN_SIZE" -exec ls -lh {} \; | awk '{print $5, $9}'

5. Monitoreo de procesos

#!/bin/bash
# process_monitor.sh - Überwacht spezifische Prozesse

PROCESSES=("nginx" "mysql" "redis" "docker")

echo "=== Prozess-Monitoring ==="
for process in "${PROCESSES[@]}"; do
    if pgrep -x "$process" > /dev/null; then
        PID=$(pgrep -x "$process")
        MEMORY=$(ps -p "$PID" -o %mem --no-headers)
        CPU=$(ps -p "$PID" -o %cpu --no-headers)
        echo "✓ $process (PID: $PID, RAM: $MEMORY%, CPU: $CPU%)"
    else
        echo "✗ $process läuft nicht"
    fi
done

Scripts PowerShell para Windows (5)

1. Recopilar información del sistema

# system_info.ps1 - Sammelt detaillierte System-Informationen

Write-Host "=== System-Informationen ===" -ForegroundColor Cyan

# System-Info
Write-Host "`nSystem:" -ForegroundColor Yellow
Get-CimInstance Win32_ComputerSystem | Select-Object Manufacturer, Model, TotalPhysicalMemory

# Betriebssystem
Write-Host "`nBetriebssystem:" -ForegroundColor Yellow
Get-CimInstance Win32_OperatingSystem | Select-Object Caption, Version, BuildNumber

# CPU
Write-Host "`nCPU:" -ForegroundColor Yellow
Get-CimInstance Win32_Processor | Select-Object Name, NumberOfCores, MaxClockSpeed

# Festplatten
Write-Host "`nFestplatten:" -ForegroundColor Yellow
Get-CimInstance Win32_LogicalDisk | Select-Object DeviceID, Size, FreeSpace | Format-Table

2. Listar cuentas de usuario

# user_accounts.ps1 - Listet alle User-Accounts auf

Write-Host "=== User-Accounts ===" -ForegroundColor Cyan

Get-LocalUser | Select-Object Name, Enabled, LastLogon | Format-Table

# Admin-Accounts
Write-Host "`nAdmin-Accounts:" -ForegroundColor Yellow
Get-LocalGroupMember -Group "Administrators" | Select-Object Name, PrincipalSource

3. Verificar actualizaciones de Windows

# check_updates.ps1 - Prüft auf Windows-Updates

Write-Host "=== Windows-Updates prüfen ===" -ForegroundColor Cyan

try {
    $UpdateSession = New-Object -ComObject Microsoft.Update.Session
    $UpdateSearcher = $UpdateSession.CreateUpdateSearcher()
    $Updates = $UpdateSearcher.Search("IsInstalled=0").Updates

    if ($Updates.Count -eq 0) {
        Write-Host "Keine ausstehenden Updates" -ForegroundColor Green
    } else {
        Write-Host "$($Updates.Count) ausstehende Updates:" -ForegroundColor Yellow
        foreach ($Update in $Updates) {
            Write-Host "  - $($Update.Title)"
        }
    }
} catch {
    Write-Host "Fehler beim Prüfen der Updates: $_" -ForegroundColor Red
}

4. Monitorear conexiones de red

# network_monitor.ps1 - Überwacht aktive Netzwerk-Verbindungen

Write-Host "=== Aktive Netzwerk-Verbindungen ===" -ForegroundColor Cyan

Get-NetTCPConnection | Where-Object {$_.State -eq "Established"} | 
    Select-Object LocalAddress, LocalPort, RemoteAddress, RemotePort, State | 
    Format-Table

# Offene Ports
Write-Host "`nOffene Ports:" -ForegroundColor Yellow
Get-NetTCPConnection | Where-Object {$_.State -eq "Listening"} | 
    Select-Object LocalAddress, LocalPort, OwningProcess | 
    Format-Table

5. Analizar registros de eventos

# event_log.ps1 - Analysiert Windows Event-Logs

Write-Host "=== Event-Log Analyse (letzte 24h) ===" -ForegroundColor Cyan

$StartDate = (Get-Date).AddDays(-1)

# Fehler im System-Log
Write-Host "`nSystem-Fehler:" -ForegroundColor Yellow
Get-EventLog -LogName System -EntryType Error -After $StartDate | 
    Select-Object TimeGenerated, Source, Message | 
    Format-Table -Wrap

# Warnungen im Application-Log
Write-Host "`nApplication-Warnungen:" -ForegroundColor Yellow
Get-EventLog -LogName Application -EntryType Warning -After $StartDate | 
    Select-Object TimeGenerated, Source, Message | 
    Format-Table -Wrap

Scripts Bash/zsh para Mac (5)

1. Información del sistema Mac

#!/bin/bash
# mac_info.sh - Muestra información del sistema Mac

echo "=== Información del sistema Mac ==="
echo ""

# Hardware
echo "Hardware:"
system_profiler SPHardwareDataType | grep -E "Model Name|Processor|Memory"

# Versión de macOS
echo ""
echo "Versión de macOS:"
sw_vers

# Disco duro
echo ""
echo "Disco duro:"
df -h

# Estado de la batería (portátiles)
echo ""
echo "Estado de la batería:"
system_profiler SPPowerDataType | grep -E "Charge|Capacity"

2. Actualizar paquetes de Homebrew

#!/bin/bash
# brew_update.sh - Actualiza paquetes de Homebrew

echo "=== Actualización de Homebrew ==="
echo ""

# Actualizar Homebrew
echo "Actualizando Homebrew..."
brew update

# Actualizar paquetes
echo ""
echo "Actualizando paquetes..."
brew upgrade

# Limpieza
echo ""
echo "Limpiando..."
brew cleanup

echo "¡Actualización completada!"

3. Automatizar la organización del Finder

#!/bin/bash
# finder_organize.sh - Organiza la carpeta de descargas

DOWNLOADS="$HOME/Downloads"

echo "=== Organizando descargas ==="

# Crear carpetas
mkdir -p "$DOWNLOADS"/{Imágenes,Documentos,Música,Vídeos,Archivos}

# Mover archivos
for file in "$DOWNLOADS"/*; do
    if [ -f "$file" ]; then
        case "${file,,}" in
            *.jpg|*.jpeg|*.png|*.gif|*.heic)
                mv "$file" "$DOWNLOADS/Imágenes/"
                echo "Imágenes: $(basename "$file")"
                ;;
            *.pdf|*.doc|*.docx|*.txt|*.pages)
                mv "$file" "$DOWNLOADS/Documentos/"
                echo "Documentos: $(basename "$file")"
                ;;
            *.mp3|*.m4a|*.wav|*.aiff)
                mv "$file" "$DOWNLOADS/Música/"
                echo "Música: $(basename "$file")"
                ;;
            *.mp4|*.mov|*.avi|*.mkv)
                mv "$file" "$DOWNLOADS/Vídeos/"
                echo "Vídeos: $(basename "$file")"
                ;;
            *.zip|*.rar|*.7z|*.dmg)
                mv "$file" "$DOWNLOADS/Archivos/"
                echo "Archivos: $(basename "$file")"
                ;;
        esac
    fi
done

echo "¡Organización completada!"

4. Gestionar claves SSH

#!/bin/bash
# ssh_keys.sh - Gestiona claves SSH

SSH_DIR="$HOME/.ssh"

echo "=== Gestión de claves SSH ==="
echo ""

# Verificar directorio SSH
if [ ! -d "$SSH_DIR" ]; then
    echo "Creando directorio SSH..."
    mkdir -p "$SSH_DIR"
    chmod 700 "$SSH_DIR"
fi

# Listar claves existentes
echo "Claves existentes:"
ls -la "$SSH_DIR"/*.pub 2>/dev/null || echo "No se encontraron claves públicas"

# Crear nueva clave (opcional)
read -p "¿Crear nueva clave SSH? (s/n): " create_key
if [ "$create_key" = "s" ]; then
    read -p "Nombre de la clave (p. ej. github): " key_name
    ssh-keygen -t ed25519 -f "$SSH_DIR/$key_name" -C "$key_name"
    echo "Clave creada: $SSH_DIR/$key_name"
fi

5. Copia de seguridad y restauración del Dock

#!/bin/bash
# dock_backup.sh - Realiza copia de seguridad y restaura la configuración del Dock

DOCK_PLIST="$HOME/Library/Preferences/com.apple.dock.plist"
BACKUP_DIR="$HOME/DockBackups"

echo "=== Copia de seguridad del Dock ==="

# Crear copia de seguridad
backup_dock() {
    mkdir -p "$BACKUP_DIR"
    DATE=$(date +%Y%m%d_%H%M%S)
    cp "$DOCK_PLIST" "$BACKUP_DIR/dock_backup_$DATE.plist"
    echo "Dock guardado: dock_backup_$DATE.plist"
}

# Restaurar Dock
restore_dock() {
    echo "Copias de seguridad disponibles:"
    ls -lt "$BACKUP_DIR"/*.plist
    read -p "Ingrese archivo de copia de seguridad: " backup_file
    cp "$backup_file" "$DOCK_PLIST"
    killall Dock
    echo "Dock restaurado y reiniciado"
}

# Menú
echo "1: Guardar Dock"
echo "2: Restaurar Dock"
read -p "Seleccione: " choice

case $choice in
    1) backup_dock ;;
    2) restore_dock ;;
    *) echo "Selección inválida" ;;
esac

Notas sobre el uso de scripts shell

Scripts Bash (Linux/Mac):

1. Hacer ejecutable: chmod +x script.sh
2. Ejecutar: ./script.sh o bash script.sh
3. Verificar ruta: which bash (generalmente /bin/bash)

Scripts PowerShell (Windows):

1. Verificar política de ejecución: Get-ExecutionPolicy
2. Cambiar si es necesario: Set-ExecutionPolicy RemoteSigned
3. Ejecutar: .\script.ps1 o powershell -File script.ps1

Específico de Mac:

• Shell por defecto es zsh (desde macOS Catalina)
• Bash sigue disponible: /bin/bash
• Para integración con AppleScript: usar osascript

Buenas prácticas para scripts de automatización

1. Estándares de código y convenciones de nombres

Python:

# ✅ Buenas convenciones de nombres
def backup_database(source_path: str, backup_dir: str) -> bool:
    """Crea copia de seguridad de la base de datos"""
    pass

# ❌ Malas convenciones de nombres
def db(s, d):
    pass

Bash:

# ✅ Buenas convenciones de nombres
backup_database() {
    local source_path="$1"
    local backup_dir="$2"
}

# ❌ Malas convenciones de nombres
bd() {
    s=$1
    d=$2
}

PowerShell:

# ✅ Buenas convenciones de nombres (PascalCase, Verb-Noun)
function Backup-Database {
    param(
        [string]$SourcePath,
        [string]$BackupDir
    )
}

# ❌ Malas convenciones de nombres
function bd {
    param($s, $d)
}

2. Documentación y comentarios

Docstrings y comentarios:

def process_files(directory: str, pattern: str) -> list:
    """
    Procesa todos los archivos en un directorio que coinciden con un patrón.

    Args:
        directory: Ruta del directorio
        pattern: Patrón de archivo (p. ej. '*.log')

    Returns:
        Lista de archivos procesados

    Raises:
        FileNotFoundError: Si el directorio no existe
    """
    if not os.path.exists(directory):
        raise FileNotFoundError(f"Directorio no encontrado: {directory}")
    # ... resto de la función

Comentarios en Bash:

#!/bin/bash
# backup.sh - Copia de seguridad automática con rotación
# Creado: 2026-07-14
# Autor: Tu nombre
# Versión: 1.0

# Configuración
SOURCE_DIR="/home/user/documents"  # Directorio de origen
BACKUP_DIR="/backup"               # Directorio de copia de seguridad
MAX_BACKUPS=7                      # Número máximo de copias

3. Modularización y reutilización

Funciones en lugar de monolitos:

# ✅ Modularizado
def validate_path(path: str) -> bool:
    """Verifica si la ruta existe"""
    return os.path.exists(path)

def copy_file(source: str, target: str) -> bool:
    """Copia archivo con manejo de errores"""
    try:
        shutil.copy(source, target)
        return True
    except Exception as e:
        log_error(f"Error de copia: {e}")
        return False

def backup_files(files: list, backup_dir: str) -> bool:
    """Función principal para copia de seguridad"""
    if not validate_path(backup_dir):
        return False

    for file in files:
        if not copy_file(file, backup_dir):
            return False

    return True

4. Separar configuración del código

Usar archivos de configuración:

# config.yaml
backup:
  source_dir: "/home/user/documents"
  backup_dir: "/backup"
  max_backups: 7
  compression: "gzip"

# script.py
import yaml

def load_config(config_file: str) -> dict:
    """Carga configuración desde archivo YAML"""
    with open(config_file, 'r') as f:
        return yaml.safe_load(f)

config = load_config('config.yaml')
backup_config = config['backup']

5. Logging y Monitoreo

Logging estructurado:

import logging
from datetime import datetime

# Logging konfigurieren
logging.basicConfig(
    level=logging.INFO,
    format='%(asctime)s - %(name)s - %(levelname)s - %(message)s',
    handlers=[
        logging.FileHandler('automation.log'),
        logging.StreamHandler()
    ]
)

logger = logging.getLogger(__name__)

def process_data(data: str):
    """Verarbeitet Daten mit Logging"""
    logger.info(f"Verarbeitung gestartet: {len(data)} Zeichen")
    try:
        result = transform_data(data)
        logger.info(f"Verarbeitung erfolgreich")
        return result
    except Exception as e:
        logger.error(f"Verarbeitung fehlgeschlagen: {e}")
        raise

6. Manejo de errores y recuperación

Gestión robusta de errores:

import time
from functools import wraps

def retry(max_attempts: int = 3, delay: int = 1):
    """Decorator für Retry-Mechanismus"""
    def decorator(func):
        @wraps(func)
        def wrapper(*args, **kwargs):
            for attempt in range(max_attempts):
                try:
                    return func(*args, **kwargs)
                except Exception as e:
                    if attempt == max_attempts - 1:
                        raise
                    logger.warning(f"Versuch {attempt + 1} fehlgeschlagen: {e}")
                    time.sleep(delay)
        return wrapper
    return decorator

@retry(max_attempts=3, delay=2)
def download_file(url: str, target: str):
    """Lädt Datei mit Retry herunter"""
    # Download-Logik
    pass

7. Testing y validación

Tests simples:

def test_validate_path():
    """Test für validate_path Funktion"""
    # Test mit existierendem Pfad
    assert validate_path('/tmp') == True

    # Test mit nicht-existierendem Pfad
    assert validate_path('/nicht/existent') == False

    print("✓ validate_path Tests bestanden")

if __name__ == '__main__':
    test_validate_path()

8. Optimización de rendimiento

Procesamiento eficiente:

# ❌ Ineffizient (für große Dateien)
def process_large_file_slow(filepath: str):
    with open(filepath, 'r') as f:
        content = f.read()  # Lädt alles in RAM
        return process_content(content)

# ✅ Effizient (streaming)
def process_large_file_fast(filepath: str):
    with open(filepath, 'r') as f:
        for line in f:  # Zeilenweise Verarbeitung
            yield process_line(line)

9. Seguridad en scripts

Sin hardcoding de secretos:

# ❌ Schlecht - Passworte im Code
password = "geheim123"

# ✅ Gut - Environment Variables
import os
password = os.getenv('DB_PASSWORD')

# ✅ Noch besser - Secrets Manager
from keyring import get_password
password = get_password('system', 'username')

10. Versionado y despliegue

Semantic Versioning:

# Versionierung im Skript
VERSION="1.2.3"  # MAJOR.MINOR.PATCH

# Changelog
# 1.2.3 - Bugfix für Backup-Rotation
# 1.2.0 - Neue Logging-Funktion
# 1.0.0 - Erste Version

Seguridad en scripts de automatización

1. Gestión de secretos

Variables de entorno:

# .env Datei (nicht in Git committen!)
DB_PASSWORD=geheim123
API_KEY=sk-1234567890

# script.py
import os
from dotenv import load_dotenv

load_dotenv()  # Lädt .env Datei

db_password = os.getenv('DB_PASSWORD')
api_key = os.getenv('API_KEY')

Secrets Manager:

# AWS Secrets Manager
import boto3

def get_secret(secret_name: str) -> str:
    """Holt Secret aus AWS Secrets Manager"""
    client = boto3.client('secretsmanager')
    response = client.get_secret_value(SecretId=secret_name)
    return response['SecretString']

# Azure Key Vault
from azure.keyvault.secrets import SecretClient
from azure.identity import DefaultAzureCredential

def get_azure_secret(vault_url: str, secret_name: str) -> str:
    """Holt Secret aus Azure Key Vault"""
    credential = DefaultAzureCredential()
    client = SecretClient(vault_url=vault_url, credential=credential)
    return client.get_secret(secret_name).value

2. Validación de entrada

Validar entradas de usuario:

import re
from pathlib import Path

def validate_filename(filename: str) -> bool:
    """Validiert Dateinamen auf Sicherheit"""
    # Erlaubt nur alphanumerische Zeichen, Unterstrich, Bindestrich
    pattern = r'^[a-zA-Z0-9_-]+\.[a-zA-Z0-9]{3,4}$'
    return bool(re.match(pattern, filename))

def validate_path(path: str) -> bool:
    """Validiert Pfad auf Path Traversal Angriffe"""
    resolved_path = Path(path).resolve()
    # Prüft ob Pfad außerhalb des erlaubten Bereichs liegt
    return not any(part in ['..', '~'] for part in resolved_path.parts)

# Nutzung
filename = input("Dateiname: ")
if not validate_filename(filename):
    raise ValueError("Ungültiger Dateiname")

3. Permisos y derechos

Principio del mínimo privilegio:

#!/bin/bash
# Skript mit minimalen Rechten ausführen

# Prüfen ob Skript als root läuft
if [ "$EUID" -eq 0 ]; then
    echo "Dieses Skript sollte nicht als root ausgeführt werden"
    exit 1
fi

# Dateien mit sicheren Permissions erstellen
umask 077  # Nur Owner hat Rechte
touch sensitive_file.txt

Permisos en Python:

import os
import stat

def set_secure_permissions(filepath: str):
    """Setzt sichere Datei-Permissions (nur Owner)"""
    os.chmod(filepath, stat.S_IRUSR | stat.S_IWUSR)  # 0600

def set_executable_permissions(filepath: str):
    """Setzt ausführbare Permissions (Owner rwx)"""
    os.chmod(filepath, stat.S_IRWXU)  # 0700

4. Prevención de SQL Injection

Consultas parametrizadas:

# ❌ UNSICHER - SQL Injection möglich
query = f"SELECT * FROM users WHERE name = '{username}'"
cursor.execute(query)

# ✅ SICHER - Parameterized Query
query = "SELECT * FROM users WHERE name = %s"
cursor.execute(query, (username,))

5. Prevención de Command Injection

Ejecución segura de shell:

import subprocess

# ❌ UNSICHER - Command Injection möglich
user_input = "user; rm -rf /"
os.system(f"echo {user_input}")

# ✅ SICHER - subprocess mit Liste
subprocess.run(['echo', user_input], check=True)

# ✅ Noch sicherer - shlex.quote
import shlex
safe_input = shlex.quote(user_input)
subprocess.run(f"echo {safe_input}", shell=True, check=True)

6. Operaciones de archivo seguras

Escritura segura de archivos:

import os
import tempfile

def safe_file_write(filepath: str, content: str):
    """Sicheres Schreiben mit Atomic Write"""
    # Temporäre Datei im gleichen Verzeichnis
    dir_path = os.path.dirname(filepath)
    with tempfile.NamedTemporaryFile(
        mode='w',
        dir=dir_path,
        delete=False
    ) as tmp_file:
        tmp_file.write(content)
        tmp_path = tmp_file.name
    
    # Atomic rename
    os.rename(tmp_path, filepath)

def safe_file_delete(filepath: str):
    """Sicheres Löschen mit Bestätigung"""
    if not os.path.exists(filepath):
        return False
    
    # Bestätigung einholen
    response = input(f"Soll {filepath} wirklich gelöscht werden? (j/n): ")
    if response.lower() == 'j':
        os.remove(filepath)
        return True
    return False

7. Logging de datos sensibles

Sin secretos en logs:

import logging
from typing import Any

class SecureFormatter(logging.Formatter):
    """Formatter der sensitive Daten maskiert"""
    SENSITIVE_KEYS = ['password', 'token', 'secret', 'key']
    
    def format(self, record: logging.LogRecord) -> str:
        msg = super().format(record)
        for key in self.SENSITIVE_KEYS:
            msg = msg.replace(f'{key}=', f'{key}=***')
        return msg

# Nutzung
logger = logging.getLogger(__name__)
handler = logging.StreamHandler()
handler.setFormatter(SecureFormatter())
logger.addHandler(handler)

logger.info("Login mit password=geheim123")
# Ausgabe: Login mit password=***

8. HTTPS y TLS

Conexiones de red seguras:

import requests
import urllib3

# SSL-Warnungen aktivieren (nicht in Production!)
urllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarning)

# ✅ SICHER - HTTPS mit Zertifikatsprüfung
response = requests.get('https://api.example.com', verify=True)

# ❌ UNSICHER - SSL-Verifikation deaktiviert
response = requests.get('https://api.example.com', verify=False)

# ✅ SICHER - Eigenes CA-Zertifikat
response = requests.get(
    'https://api.example.com',
    verify='/path/to/ca-bundle.crt'
)

9. Seguridad de dependencias

Escaneo de vulnerabilidades:

# Python Dependencies scannen
pip install safety
safety check

# npm Dependencies scannen
npm audit

# Alle Dependencies aktualisieren
pip install --upgrade pip
pip list --outdated

10. Auditorías de seguridad

Verificaciones de seguridad periódicas:

import subprocess
import json

def run_security_audit():
    """Ejecuta controles básicos de seguridad"""
    checks = {
        'python_version': check_python_version,
        'dependencies': check_dependencies,
        'file_permissions': check_file_permissions,
        'hardcoded_secrets': check_hardcoded_secrets
    }
    
    results = {}
    for check_name, check_func in checks.items():
        results[check_name] = check_func()
    
    return results

def check_hardcoded_secrets():
    """Busca secretos hardcodeados en el código"""
    import re
    secret_patterns = [
        r'password\s*=\s*["\'][^"\']+["\']',
        r'api_key\s*=\s*["\'][^"\']+["\']',
        r'secret\s*=\s*["\'][^"\']+["\']'
    ]
    
    findings = []
    for root, dirs, files in os.walk('.'):
        for file in files:
            if file.endswith('.py'):
                filepath = os.path.join(root, file)
                with open(filepath, 'r') as f:
                    content = f.read()
                    for pattern in secret_patterns:
                        if re.search(pattern, content):
                            findings.append(filepath)
    
    return findings

Pruebas de scripts de automatización

1. Unit Testing para scripts Python

Tests unitarios simples:

# test_backup.py
import unittest
import tempfile
import os
from backup_script import backup_files, validate_path

class TestBackupFunctions(unittest.TestCase):
    def setUp(self):
        """Configuración inicial para las pruebas"""
        self.temp_dir = tempfile.mkdtemp()
        self.test_file = os.path.join(self.temp_dir, 'test.txt')
        with open(self.test_file, 'w') as f:
            f.write('Test content')
    
    def tearDown(self):
        """Limpieza después de las pruebas"""
        import shutil
        shutil.rmtree(self.temp_dir)
    
    def test_validate_path_existing(self):
        """Prueba con una ruta existente"""
        self.assertTrue(validate_path(self.temp_dir))
    
    def test_validate_path_non_existing(self):
        """Prueba con una ruta inexistente"""
        self.assertFalse(validate_path('/nicht/existent'))
    
    def test_backup_files_success(self):
        """Prueba de backup exitoso"""
        backup_dir = os.path.join(self.temp_dir, 'backup')
        os.makedirs(backup_dir)
        
        result = backup_files([self.test_file], backup_dir)
        self.assertTrue(result)
        self.assertTrue(os.path.exists(os.path.join(backup_dir, 'test.txt')))

if __name__ == '__main__':
    unittest.main()

2. Mocking para dependencias externas

Mocking de operaciones del sistema de archivos:

from unittest.mock import patch, MagicMock
import unittest

class TestBackupWithMocking(unittest.TestCase):
    @patch('backup_script.shutil.copy')
    def test_backup_with_mock(self, mock_copy):
        """Prueba con la función shutil.copy simulada"""
        mock_copy.return_value = True
        
        result = backup_files(['file1.txt'], '/backup')
        
        self.assertTrue(result)
        mock_copy.assert_called_once()
    
    @patch('backup_script.os.path.exists')
    def test_validate_path_mock(self, mock_exists):
        """Prueba con os.path.exists simulada"""
        mock_exists.return_value = True
        
        result = validate_path('/any/path')
        
        self.assertTrue(result)
        mock_exists.assert_called_once_with('/any/path')

3. Integration Testing

Pruebas end-to-end:

# test_integration.py
import unittest
import tempfile
import os
import subprocess

class TestIntegration(unittest.TestCase):
    def test_script_execution(self):
        """Prueba de ejecución completa del script"""
        script_path = 'backup.sh'
        test_dir = tempfile.mkdtemp()
        
        # Ejecutar script
        result = subprocess.run(
            ['bash', script_path, test_dir],
            capture_output=True,
            text=True
        )
        
        # Verificar resultado
        self.assertEqual(result.returncode, 0)
        self.assertIn('Backup erstellt', result.stdout)
        
        # Limpieza
        import shutil
        shutil.rmtree(test_dir)

4. Pruebas de scripts Bash

Framework de Unit Testing para Bash:

#!/bin/bash
# test_backup.sh - Tests für backup.sh

# Test-Funktionen
test_validate_path() {
    source backup.sh
    
    # Test mit existierendem Pfad
    if validate_path "/tmp"; then
        echo "✓ validate_path: existierender Pfad"
    else
        echo "✗ validate_path: existierender Pfad fehlgeschlagen"
        exit 1
    fi
    
    # Test mit nicht-existierendem Pfad
    if ! validate_path "/nicht/existent"; then
        echo "✓ validate_path: nicht-existierender Pfad"
    else
        echo "✗ validate_path: nicht-existierender Pfad fehlgeschlagen"
        exit 1
    fi
}

test_backup_function() {
    source backup.sh
    
    local test_dir=$(mktemp -d)
    local test_file="$test_dir/test.txt"
    echo "Test content" > "$test_file"
    
    # Backup-Funktion testen
    backup_files "$test_file" "$test_dir/backup"
    
    if [ -f "$test_dir/backup/test.txt" ]; then
        echo "✓ backup_files: Datei erfolgreich kopiert"
    else
        echo "✗ backup_files: Datei nicht kopiert"
        exit 1
    fi
    
    # Cleanup
    rm -rf "$test_dir"
}

# Alle Tests ausführen
echo "=== Bash-Tests ==="
test_validate_path
test_backup_function
echo "Alle Tests bestanden!"

5. Pruebas con PowerShell

Framework Pester para PowerShell:

# test_backup.ps1 - Pester Tests für backup.ps1

Describe "Backup-Funktionen" {
    BeforeAll {
        . .\backup.ps1
        $testDir = Join-Path $env:TEMP "backup_test"
        New-Item -Path $testDir -ItemType Directory -Force | Out-Null
    }
    
    AfterAll {
        Remove-Item -Path $testDir -Recurse -Force -ErrorAction SilentlyContinue
    }
    
    Context "Validate-Path" {
        It "sollte existierenden Pfad zurückgeben" {
            $result = Validate-Path $testDir
            $result | Should -Be $true
        }
        
        It "sollte nicht-existierenden Pfad ablehnen" {
            $result = Validate-Path "C:\nicht\existent"
            $result | Should -Be $false
        }
    }
    
    Context "Backup-Files" {
        It "sollte Dateien erfolgreich kopieren" {
            $testFile = Join-Path $testDir "test.txt"
            "Test content" | Out-File -FilePath $testFile
            
            $backupDir = Join-Path $testDir "backup"
            $result = Backup-Files $testFile $backupDir
            
            $result | Should -Be $true
            (Test-Path (Join-Path $backupDir "test.txt")) | Should -Be $true
        }
    }
}

6. Test-Driven Development (TDD)

Flujo de TDD:

# 1. Escribir el test (fallará)
def test_calculate_backup_size():
    """Test para el cálculo del tamaño del backup"""
    result = calculate_backup_size('/tmp')
    assert result > 0  # Fallará, la función aún no existe

# 2. Implementar la función (para que el test pase)
def calculate_backup_size(directory: str) -> int:
    """Calcula el tamaño de un directorio"""
    total_size = 0
    for dirpath, dirnames, filenames in os.walk(directory):
        for filename in filenames:
            filepath = os.path.join(dirpath, filename)
            total_size += os.path.getsize(filepath)
    return total_size

# 3. Ejecutar el test (debería pasar ahora)
# 4. Refactorizar (mientras los tests sigan pasando)

7. Continuous Testing

Ejecutar tests automáticamente ante cambios:

# watchmedo para tests automáticos
pip install watchdog watchmedo

# Ejecutar tests cuando cambian los archivos
watchmedo shell-command \
    --patterns="*.py" \
    --recursive \
    --command='python -m pytest test_*.py -v'

8. Test-Coverage

Análisis de cobertura:

# Instalar pytest-cov
pip install pytest-cov

# Ejecutar tests con cobertura
pytest --cov=backup_script --cov-report=html

# Abrir el reporte de cobertura
open htmlcov/index.html

9. Property-Based Testing

Hypothesis para Python:

from hypothesis import given, strategies as st
import unittest

class TestBackupProperties(unittest.TestCase):
    @given(st.lists(st.text()))
    def test_backup_preserves_content(self, content_list):
        """Property test: el backup preserva el contenido"""
        # Test con diferentes entradas
        for content in content_list:
            with tempfile.NamedTemporaryFile(mode='w', delete=False) as f:
                f.write(content)
                temp_file = f.name
            
            # Crear backup
            backup_file = temp_file + '.backup'
            shutil.copy(temp_file, backup_file)
            
            # Verificar contenido
            with open(backup_file, 'r') as f:
                backup_content = f.read()
            
            self.assertEqual(content, backup_content)
            
            # Limpiar
            os.unlink(temp_file)
            os.unlink(backup_file)

10. Load-Testing para scripts

Tests de rendimiento:

import time
import statistics

def test_backup_performance():
    """Test del rendimiento del backup"""
    test_times = []
    
    for i in range(10):
        start_time = time.time()
        backup_files(['large_file.txt'], '/backup')
        end_time = time.time()
        
        test_times.append(end_time - start_time)
    
    avg_time = statistics.mean(test_times)
    max_time = max(test_times)
    
    print(f"Tiempo promedio: {avg_time:.2f}s")
    print(f"Tiempo máximo: {max_time:.2f}s")
    
    # Assertion de rendimiento
    assert avg_time < 5.0, "Backup muy lento"
    assert max_time < 10.0, "Tiempo máximo del backup excedido"

Control de versiones para scripts de automatización

1. Fundamentos de Git para scripts

Inicialización y primer commit:

# Inicializar repositorio Git
git init

# Crear .gitignore
cat > .gitignore << EOF
# Python
__pycache__/
*.py[cod]
*$py.class
*.so
.Python
env/
venv/
.venv/

# Secrets
.env
*.key
*.pem
secrets/

# Logs
*.log

# IDE
.vscode/
.idea/
*.swp
*.swo

# OS
.DS_Store
Thumbs.db
EOF

# Añadir scripts
git add backup.sh backup.py config.yaml
git commit -m "Initial commit: Backup scripts added"

2. Estrategias de branching

Git Flow para scripts:

# Branch main (producción)
git checkout -b main

# Branch develop (desarrollo)
git checkout -b develop

# Feature branch para nueva funcionalidad
git checkout -b feature/add-logging

# Hacer commit de los cambios
git add backup.py
git commit -m "Add logging functionality"

# Mergear de vuelta a develop
git checkout develop
git merge feature/add-logging

# Eliminar el feature branch
git branch -d feature/add-logging

3. Mensajes de commit convencionales

Conventional Commits:

# Formato: <type>(<scope>): <description>

# Ejemplos:
git commit -m "feat(backup): Add compression support"
git commit -m "fix(logging): Fix timestamp format"
git commit -m "docs(readme): Update installation instructions"
git commit -m "style(python): Apply PEP 8 formatting"
git commit -m "refactor(cleanup): Simplify error handling"
git commit -m "test(backup): Add unit tests for backup function"
git commit -m "chore(deps): Update dependencies"

4. Code reviews y Pull Requests

Flujo de Pull Request:

# Crear feature branch
git checkout -b feature/backup-rotation

# Hacer cambios y commit
git add backup.sh
git commit -m "feat(backup): Add automatic backup rotation"

# Push al remoto
git push origin feature/backup-rotation

# Crear Pull Request (GitHub/GitLab/Bitbucket)
# Los miembros del equipo revisan el código
# Después de la aprobación, mergear

5. Git hooks para asegurar calidad

Pre-commit Hook:

#!/bin/bash
# .git/hooks/pre-commit

# Formatear código Python
autopep8 --in-place --aggressive *.py

# Analizar código Python
pylint *.py

# Ejecutar tests
python -m pytest test_*.py

# Si todo es exitoso, exit 0
exit 0

Pre-push Hook:

#!/bin/bash
# .git/hooks/pre-push

# Ejecutar todos los tests
python -m pytest

# Security check
safety check

# Si todo es exitoso, exit 0
exit 0

6. Versionado con tags

Semantic Versioning:

# Major version (cambios incompatibles)
git tag -a v2.0.0 -m "Major release: Breaking changes"
git push origin v2.0.0

# Minor version (nuevas features, compatible hacia atrás)
git tag -a v1.2.0 -m "Minor release: New features"
git push origin v1.2.0

# Patch version (bugfixes, compatible hacia atrás)
git tag -a v1.1.1 -m "Patch release: Bug fixes"
git push origin v1.1.1

7. Mantener un changelog

Changelog automático:

# Con conventional-changelog
npm install -g conventional-changelog-cli

# Generar changelog
conventional-changelog -p angular -i CHANGELOG.md -s

# Estructura de CHANGELOG.md:
# ## [1.2.0] - 2026-07-14
# ### Added
# - Backup rotation añadido
# - Funcionalidad de logging implementada
#
# ### Fixed
# - Error en validación de rutas corregido
#
# ## [1.1.0] - 2026-06-01
# ### Added
# - Primera versión del script de backup

8. Configuración de Git

Configuración de Git para scripts:

# Configurar usuario
git config user.name "Tu Nombre"
git config user.email "tu@email.com"

# Line endings (importante para multiplataforma)
git config core.autocrlf input  # Linux/Mac
git config core.autocrlf true   # Windows

# Atributos de archivo
git config core.filemode false  # Ignorar permisos

# Configurar herramienta de diff
git config diff.tool vimdiff
git config merge.tool vimdiff

9. Git workflows para equipos

Forking Workflow:

# 1. Hacer fork del repositorio (GitHub/GitLab)
# 2. Clonar el fork
git clone https://github.com/tu-usuario/repo.git

# 3. Añadir remote upstream
git remote add upstream https://github.com/original/repo.git

# 4. Crear feature branch
git checkout -b feature/nueva-funcion

# 5. Hacer commit de los cambios
git add script.py
git commit -m "feat: nueva función añadida"

# 6. Push a tu fork
git push origin feature/nueva-funcion

# 7. Crear Pull Request
# 8. Después de la revisión, mergear

# 9. Obtener cambios de upstream
git fetch upstream
git checkout main
git merge upstream/main

10. Git como estrategia de respaldo

Git como sistema de respaldo:

#!/bin/bash
# git_backup.sh - Respaldo automático con Git

REPO_DIR="/path/to/repo"
BACKUP_REMOTE="git@github.com:user/backup-repo.git"

cd "$REPO_DIR"

# Agregar cambios
git add .

# Commit con timestamp
git commit -m "Auto-backup: $(date '+%Y-%m-%d %H:%M:%S')"

# Enviar al repositorio remoto
git push "$BACKUP_REMOTE" main

echo "Respaldo enviado exitosamente a Git"

Integración CI/CD para scripts de automatización

1. GitHub Actions para scripts Python

Archivo de workflow:

# .github/workflows/automation.yml
name: Automation Scripts CI/CD

on:
  push:
    branches: [ main, develop ]
  pull_request:
    branches: [ main ]

jobs:
  test:
    runs-on: ubuntu-latest
    
    steps:
    - uses: actions/checkout@v3
    
    - name: Set up Python
      uses: actions/setup-python@v4
      with:
        python-version: '3.11'
    
    - name: Install dependencies
      run: |
        python -m pip install --upgrade pip
        pip install -r requirements.txt
        pip install pytest pytest-cov pylint safety
    
    - name: Lint with pylint
      run: pylint *.py
    
    - name: Security check
      run: safety check
    
    - name: Run tests
      run: pytest --cov=. --cov-report=xml
    
    - name: Upload coverage
      uses: codecov/codecov-action@v3
      with:
        file: ./coverage.xml

  deploy:
    needs: test
    runs-on: ubuntu-latest
    if: github.ref == 'refs/heads/main'
    
    steps:
    - uses: actions/checkout@v3
    
    - name: Deploy to production
      run: |
        echo "Deploying automation scripts to production"
        # Deployment-Logik hier

2. GitLab CI para scripts Bash

Archivo GitLab CI:

# .gitlab-ci.yml
stages:
  - test
  - deploy

variables:
  SCRIPT_DIR: "/opt/automation"

test_scripts:
  stage: test
  image: alpine:latest
  
  before_script:
    - apk add --no-cache bash shellcheck
  
  script:
    - shellcheck *.sh
    - bash -n backup.sh
    - bash test_backup.sh
  
  artifacts:
    reports:
      junit: test-results.xml

deploy_production:
  stage: deploy
  image: alpine:latest
  only:
    - main
  
  script:
    - apk add --no-cache rsync openssh
    - mkdir -p ~/.ssh
    - echo "$SSH_PRIVATE_KEY" > ~/.ssh/id_rsa
    - chmod 600 ~/.ssh/id_rsa
    - rsync -avz --delete *.sh user@server:$SCRIPT_DIR/
    - ssh user@server "chmod +x $SCRIPT_DIR/*.sh"

3. Jenkins Pipeline

Jenkinsfile:

// Jenkinsfile
pipeline {
    agent any
    
    environment {
        PYTHON_VERSION = '3.11'
        DEPLOY_SERVER = 'user@server'
    }
    
    stages {
        stage('Checkout') {
            steps {
                checkout scm
            }
        }
        
        stage('Lint') {
            steps {
                sh 'pylint *.py || true'
                sh 'shellcheck *.sh || true'
            }
        }
        
        stage('Test') {
            steps {
                sh 'python -m pytest test_*.py --cov=. --cov-report=html'
            }
        }
        
        stage('Security Scan') {
            steps {
                sh 'safety check'
            }
        }
        
        stage('Deploy') {
            when {
                branch 'main'
            }
            steps {
                sh '''
                    rsync -avz --delete *.sh ${DEPLOY_SERVER}:/opt/automation/
                    ssh ${DEPLOY_SERVER} "chmod +x /opt/automation/*.sh"
                '''
            }
        }
    }
    
    post {
        always {
            archiveArtifacts artifacts: 'htmlcov/**', allowEmptyArchive: true
            publishHTML target: [
                reportDir: 'htmlcov',
                reportFiles: 'index.html',
                reportName: 'Coverage Report'
            ]
        }
    }
}

4. Azure DevOps Pipeline

azure-pipelines.yml:

# azure-pipelines.yml
trigger:
  branches:
    include:
    - main
    - develop

pool:
  vmImage: 'ubuntu-latest'

variables:
  pythonVersion: '3.11'

stages:
- stage: Test
  jobs:
  - job: TestScripts
    steps:
    - task: UsePythonVersion@0
      inputs:
        versionSpec: '$(pythonVersion)'
    
    - script: |
        python -m pip install --upgrade pip
        pip install pytest pytest-cov pylint safety
      displayName: 'Install dependencies'
    
    - script: |
        pylint *.py
      displayName: 'Lint Python'
    
    - script: |
        safety check
      displayName: 'Security check'
    
    - script: |
        pytest --cov=. --cov-report=xml
      displayName: 'Run tests'
    
    - task: PublishCodeCoverageResults@1
      inputs:
        codeCoverageTool: 'Cobertura'
        summaryFileLocation: '$(System.DefaultWorkingDirectory)/coverage.xml'

- stage: Deploy
  dependsOn: Test
  condition: and(succeeded(), eq(variables['Build.SourceBranch'], 'refs/heads/main'))
  jobs:
  - deployment: DeployScripts
    environment: 'Production'
    strategy:
      runOnce:
        deploy:
          steps:
          - script: |
              rsync -avz --delete *.sh user@server:/opt/automation/
            displayName: 'Deploy scripts'

5. CI/CD basado en Docker

Dockerfile para pruebas:

# Dockerfile.test
FROM python:3.11-slim

# Dependencias del sistema
RUN apt-get update && apt-get install -y \
    shellcheck \
    bash \
    && rm -rf /var/lib/apt/lists/*

# Dependencias de Python
COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt
RUN pip install --no-cache-dir pytest pytest-cov pylint safety

# Copiar scripts
COPY *.py *.sh /app/
WORKDIR /app

# Comando de pruebas
CMD ["pytest", "--cov=.", "--cov-report=html"]

Docker Compose para CI:

# docker-compose.test.yml
version: '3.8'

services:
  test:
    build:
      context: .
      dockerfile: Dockerfile.test
    volumes:
      - .:/app
      - ./htmlcov:/app/htmlcov
    command: pytest --cov=. --cov-report=html

6. Gestión de entornos

Deployment en múltiples entornos:

# .github/workflows/deploy.yml
name: Multi-Environment Deploy

on:
  push:
    branches: [ main ]

jobs:
  deploy_staging:
    runs-on: ubuntu-latest
    environment: staging
    
    steps:
    - uses: actions/checkout@v3
    
    - name: Deploy to Staging
      run: |
        rsync -avz --delete *.sh $STAGING_SERVER:/opt/automation/
      env:
        STAGING_SERVER: ${{ secrets.STAGING_SERVER }}

  deploy_production:
    runs-on: ubuntu-latest
    environment: production
    needs: deploy_staging
    
    steps:
    - uses: actions/checkout@v3
    
    - name: Deploy to Production
      run: |
        rsync -avz --delete *.sh $PROD_SERVER:/opt/automation/
      env:
        PROD_SERVER: ${{ secrets.PROD_SERVER }}

7. Estrategias de reversión

Reversión automática:

# .github/workflows/rollback.yml
name: Rollback Deployment

on:
  workflow_dispatch:
    inputs:
      version:
        description: 'Version to rollback to'
        required: true

jobs:
  rollback:
    runs-on: ubuntu-latest
    
    steps:
    - uses: actions/checkout@v3
      with:
        ref: ${{ github.event.inputs.version }}
    
    - name: Deploy previous version
      run: |
        rsync -avz --delete *.sh $PROD_SERVER:/opt/automation/
      env:
        PROD_SERVER: ${{ secrets.PROD_SERVER }}
    
    - name: Notify team
      run: |
        echo "Rollback to version ${{ github.event.inputs.version }} completed"

8. Monitoreo en CI/CD

Validaciones de salud en el pipeline:

# .github/workflows/monitoring.yml
name: Pipeline Monitoring

on:
  schedule:
    - cron: '0 * * * *'  # Cada hora

jobs:
  health_check:
    runs-on: ubuntu-latest
    
    steps:
    - uses: actions/checkout@v3
    
    - name: Check script health
      run: |
        # Verificar que los scripts se están ejecutando
        ssh $PROD_SERVER "ps aux | grep backup.sh"
        
        # Verificar que se escriben los logs
        ssh $PROD_SERVER "tail -n 10 /var/log/automation.log"
      env:
        PROD_SERVER: ${{ secrets.PROD_SERVER }}
    
    - name: Alert on failure
      if: failure()
      run: |
        # Enviar alerta (Slack, Email, etc.)
        curl -X POST $SLACK_WEBHOOK -d '{"text":"Automation health check failed"}'
      env:
        SLACK_WEBHOOK: ${{ secrets.SLACK_WEBHOOK }}

9. Pruebas automatizadas en el pipeline

Matriz de pruebas para diferentes versiones de Python:

# .github/workflows/test-matrix.yml
name: Test Matrix

on: [push, pull_request]

jobs:
  test:
    runs-on: ubuntu-latest
    strategy:
      matrix:
        python-version: ['3.9', '3.10', '3.11']
    
    steps:
    - uses: actions/checkout@v3
    
    - name: Set up Python ${{ matrix.python-version }}
      uses: actions/setup-python@v4
      with:
        python-version: ${{ matrix.python-version }}
    
    - name: Install dependencies
      run: |
        python -m pip install --upgrade pip
        pip install -r requirements.txt
        pip install pytest
    
    - name: Run tests
      run: pytest test_*.py

10. Gestión de secretos en CI/CD

Usar secretos de GitHub:

# .github/workflows/secure-deploy.yml
name: Secure Deployment

on:
  push:
    branches: [ main ]

jobs:
  deploy:
    runs-on: ubuntu-latest
    
    steps:
    - uses: actions/checkout@v3
    
    - name: Setup SSH key
      run: |
        mkdir -p ~/.ssh
        echo "${{ secrets.SSH_PRIVATE_KEY }}" > ~/.ssh/id_rsa
        chmod 600 ~/.ssh/id_rsa
        ssh-keyscan -H ${{ secrets.DEPLOY_SERVER }} >> ~/.ssh/known_hosts
    
    - name: Deploy with secrets
      run: |
        # Cargar secretos desde el entorno
        export DB_PASSWORD="${{ secrets.DB_PASSWORD }}"
        export API_KEY="${{ secrets.API_KEY }}"
        
        # Ejecutar el script
        rsync -avz --delete *.sh ${{ secrets.DEPLOY_SERVER }}:/opt/automation/
        ssh ${{ secrets.DEPLOY_SERVER }} "cd /opt/automation && ./deploy.sh"
      env:
        DEPLOY_SERVER: ${{ secrets.DEPLOY_SERVER }}

Containerización de scripts de automatización

1. Dockerfile para scripts Python

Dockerfile simple:

# Dockerfile
FROM python:3.11-slim

# Directorio de trabajo
WORKDIR /app

# Copiar dependencias
COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt

# Copiar scripts
COPY backup.py .
COPY config.yaml .

# Script como punto de entrada
ENTRYPOINT ["python", "backup.py"]
CMD ["--config", "config.yaml"]

2. Dockerfile para scripts Bash

Bash en el contenedor:

# Dockerfile.bash
FROM alpine:latest

# Instalar Bash y herramientas
RUN apk add --no-cache bash rsync coreutils

# Copiar scripts
COPY backup.sh /usr/local/bin/
RUN chmod +x /usr/local/bin/backup.sh

# Copiar configuración
COPY config.yaml /etc/backup/

# Directorio de trabajo
WORKDIR /data

ENTRYPOINT ["/usr/local/bin/backup.sh"]

3. Dockerfile multi-etapa

Build multi-etapa optimizado:

# Dockerfile.multi-stage
# Etapa de compilación
FROM python:3.11-slim AS builder

WORKDIR /app
COPY requirements.txt .
RUN pip install --user --no-cache-dir -r requirements.txt

# Etapa de ejecución
FROM python:3.11-slim

WORKDIR /app
COPY --from=builder /root/.local /root/.local
COPY backup.py .
COPY config.yaml .

ENV PATH=/root/.local/bin:$PATH

ENTRYPOINT ["python", "backup.py"]

4. Docker Compose para el entorno completo

docker-compose.yml:

version: '3.8'

services:
  backup:
    build:
      context: .
      dockerfile: Dockerfile
    volumes:
      - ./data:/data
      - ./config:/config
      - ./logs:/logs
    environment:
      - BACKUP_DIR=/data/backups
      - SOURCE_DIR=/data/source
      - LOG_LEVEL=INFO
    restart: unless-stopped

  monitoring:
    image: prom/prometheus
    ports:
      - "9090:9090"
    volumes:
      - ./prometheus.yml:/etc/prometheus/prometheus.yml

  grafana:
    image: grafana/grafana
    ports:
      - "3000:3000"
    volumes:
      - grafana-data:/var/lib/grafana

volumes:
  grafana-data:

5. Deployment en Kubernetes

YAML de Deployment:

# k8s/deployment.yaml
apiVersion: apps/v1
kind: Deployment
metadata:
  name: automation-scripts
spec:
  replicas: 2
  selector:
    matchLabels:
      app: automation
  template:
    metadata:
      labels:
        app: automation
    spec:
      containers:
      - name: backup
        image: your-registry/backup:latest
        env:
        - name: BACKUP_DIR
          value: "/data/backups"
        - name: SOURCE_DIR
          valueFrom:
            configMapKeyRef:
              name: backup-config
              key: source_dir
        volumeMounts:
        - name: data
          mountPath: /data
        - name: config
          mountPath: /config
      volumes:
      - name: data
        persistentVolumeClaim:
          claimName: backup-pvc
      - name: config
        configMap:
          name: backup-config

6. Health Checks en Docker

Health Check en el Dockerfile:

# Dockerfile con Health Check
FROM python:3.11-slim

WORKDIR /app
COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt
COPY backup.py health_check.py .

# Script de Health Check
HEALTHCHECK --interval=30s --timeout=10s --start-period=5s --retries=3 \
    CMD python health_check.py || exit 1

ENTRYPOINT ["python", "backup.py"]

Script de Health Check:

# health_check.py
import os
import sys

def check_health():
    """Verifica que el script de backup está en buen estado"""
    # Verificar que existe la configuración
    if not os.path.exists('config.yaml'):
        print("Config file missing")
        return False
    
    # Verificar que el directorio de backup es escribible
    backup_dir = os.getenv('BACKUP_DIR', '/data/backups')
    if not os.access(backup_dir, os.W_OK):
        print("Backup directory not writable")
        return False
    
    print("Health check passed")
    return True

if __name__ == '__main__':
    sys.exit(0 if check_health() else 1)

7. Optimización de contenedores

Buenas prácticas para imágenes pequeñas:

# Dockerfile optimizado
FROM python:3.11-alpine AS builder

# Solo dependencias necesarias
COPY requirements.txt .
RUN pip install --no-cache-dir --user --no-deps -r requirements.txt

FROM python:3.11-alpine

# Usuario sin permisos de root
RUN addgroup -g 1000 backup && \
    adduser -D -u 1000 -G backup backup

WORKDIR /app
COPY --from=builder /root/.local /root/.local
COPY --chown=backup:backup backup.py .

USER backup
ENV PATH=/root/.local/bin:$PATH

ENTRYPOINT ["python", "backup.py"]

8. Seguridad en contenedores

Escaneo de seguridad y hardening:

# Dockerfile reforzado en seguridad
FROM python:3.11-slim

# Actualizar seguridad
RUN apt-get update && \
    apt-get upgrade -y && \
    apt-get install -y --no-install-recommends \
    ca-certificates && \
    rm -rf /var/lib/apt/lists/*

# Usuario sin permisos de root
RUN groupadd -r backup && useradd -r -g backup backup

WORKDIR /app
COPY --chown=backup:backup requirements.txt .
RUN pip install --no-cache-dir --user -r requirements.txt

COPY --chown=backup:backup backup.py .

USER backup
ENTRYPOINT ["python", "backup.py"]

Escanear con Trivy:

# Escanear la imagen
trivy image your-registry/backup:latest

# Integrar en CI/CD
trivy image --exit-code 1 --severity HIGH,CRITICAL your-registry/backup:latest

9. Container-Logging

Logging estructurado:

# backup.py con Container-Logging
import logging
import json
import sys

class JSONFormatter(logging.Formatter):
    """JSON Formatter para logs de contenedores"""
    def format(self, record):
        log_data = {
            'timestamp': self.formatTime(record),
            'level': record.levelname,
            'message': record.getMessage(),
            'logger': record.name
        }
        if hasattr(record, 'extra_data'):
            log_data.update(record.extra_data)
        return json.dumps(log_data)

# Configurar logging
handler = logging.StreamHandler(sys.stdout)
handler.setFormatter(JSONFormatter())
logger = logging.getLogger(__name__)
logger.addHandler(handler)
logger.setLevel(logging.INFO)

10. Container-Orquestación

Servicio Docker Swarm:

# docker-stack.yml
version: '3.8'

services:
  backup:
    image: your-registry/backup:latest
    deploy:
      replicas: 3
      update_config:
        parallelism: 1
        delay: 10s
      restart_policy:
        condition: on-failure
        delay: 5s
        max_attempts: 3
    environment:
      - BACKUP_DIR=/data/backups
    volumes:
      - backup-data:/data
    networks:
      - automation-network

networks:
  automation-network:
    driver: overlay

volumes:
  backup-data:

Monitoreo y Alertas para Scripts de Automatización

1. Exportación de Métricas Prometheus

Métricas Python con Prometheus:

# metrics_exporter.py
from prometheus_client import start_http_server, Counter, Histogram, Gauge
import time

# Definir métricas
backup_counter = Counter('backup_total', 'Total number of backups', ['status'])
backup_duration = Histogram('backup_duration_seconds', 'Backup duration')
backup_size = Gauge('backup_size_bytes', 'Backup size in bytes')

def backup_with_metrics(source: str, target: str):
    """Backup con métricas Prometheus"""
    start_time = time.time()
    
    try:
        # Realizar backup
        size = perform_backup(source, target)
        
        # Actualizar métricas
        backup_counter.labels(status='success').inc()
        backup_duration.observe(time.time() - start_time)
        backup_size.set(size)
        
    except Exception as e:
        backup_counter.labels(status='failed').inc()
        raise

# Iniciar servidor de métricas
if __name__ == '__main__':
    start_http_server(8000)
    print("Metrics server started on port 8000")

2. Dashboard de Grafana

Configuración del dashboard de Grafana:

{
  "dashboard": {
    "title": "Automation Scripts Monitoring",
    "panels": [
      {
        "title": "Backup Success Rate",
        "targets": [
          {
            "expr": "rate(backup_total{status='success'}[5m])"
          }
        ]
      },
      {
        "title": "Backup Duration",
        "targets": [
          {
            "expr": "histogram_quantile(0.95, rate(backup_duration_seconds_bucket[5m]))"
          }
        ]
      },
      {
        "title": "Backup Size",
        "targets": [
          {
            "expr": "backup_size_bytes"
          }
        ]
      }
    ]
  }
}

3. Análisis de Logs con ELK Stack

Configuración de Filebeat:

# filebeat.yml
filebeat.inputs:
- type: log
  enabled: true
  paths:
    - /var/log/automation/*.log
  fields:
    app: automation
  fields_under_root: true

output.elasticsearch:
  hosts: ["localhost:9200"]
  index: "automation-logs-%{+yyyy.MM.dd}"

setup.kibana:
  host: "localhost:5601"

4. Alertas con Alertmanager

Configuración de Alertmanager:

# alertmanager.yml
global:
  resolve_timeout: 5m

route:
  group_by: ['alertname']
  group_wait: 10s
  group_interval: 10s
  repeat_interval: 1h
  receiver: 'web.hook'

receivers:
- name: 'web.hook'
  webhook_configs:
  - url: 'http://localhost:5001/alerts'

- name: 'slack'
  slack_configs:
  - api_url: 'YOUR_SLACK_WEBHOOK_URL'
    channel: '#automation-alerts'

Reglas de alerta de Prometheus:

# alerts.yml
groups:
- name: automation_alerts
  rules:
  - alert: BackupFailed
    expr: rate(backup_total{status='failed'}[5m]) > 0
    for: 1m
    labels:
      severity: critical
    annotations:
      summary: "Backup failed"
      description: "Backup has been failing for the last 5 minutes"

  - alert: BackupTooSlow
    expr: histogram_quantile(0.95, rate(backup_duration_seconds_bucket[5m])) > 300
    for: 5m
    labels:
      severity: warning
    annotations:
      summary: "Backup too slow"
      description: "Backup duration exceeds 5 minutes"

5. Endpoints de Health-Check

HTTP Health-Check:

# health_check.py
from flask import Flask, jsonify
import os
import psutil

app = Flask(__name__)

@app.route('/health')
def health_check():
    """Endpoint de Health-Check"""
    health_status = {
        'status': 'healthy',
        'checks': {}
    }
    
    # Verificar espacio en disco
    disk_usage = psutil.disk_usage('/')
    health_status['checks']['disk'] = {
        'status': 'ok' if disk_usage.percent < 90 else 'warning',
        'usage_percent': disk_usage.percent
    }
    
    # Verificar memoria
    memory = psutil.virtual_memory()
    health_status['checks']['memory'] = {
        'status': 'ok' if memory.percent < 90 else 'warning',
        'usage_percent': memory.percent
    }
    
    # Verificar archivo de configuración
    config_exists = os.path.exists('config.yaml')
    health_status['checks']['config'] = {
        'status': 'ok' if config_exists else 'error',
        'exists': config_exists
    }
    
    # Estado general
    if any(check['status'] == 'error' for check in health_status['checks'].values()):
        health_status['status'] = 'error'
    elif any(check['status'] == 'warning' for check in health_status['checks'].values()):
        health_status['status'] = 'warning'
    
    return jsonify(health_status)

if __name__ == '__main__':
    app.run(host='0.0.0.0', port=8080)

6. Integración con Slack

Enviar alertas a Slack:

# slack_alerts.py
import requests
import json

def send_slack_alert(webhook_url: str, message: str, level: str = 'info'):
    """Envía una alerta a Slack"""
    colors = {
        'info': '#36a64f',
        'warning': '#ff9900',
        'error': '#ff0000'
    }
    
    payload = {
        'attachments': [{
            'color': colors.get(level, '#36a64f'),
            'title': f'Automation Alert: {level.upper()}',
            'text': message,
            'footer': 'Automation Scripts',
            'ts': int(time.time())
        }]
    }
    
    response = requests.post(webhook_url, json=payload)
    return response.status_code == 200

# Uso
send_slack_alert(
    webhook_url='YOUR_SLACK_WEBHOOK',
    message='Backup failed for database',
    level='error'
)

7. Alertas por Email

Alertas por email con Python:

# email_alerts.py
import smtplib
from email.mime.text import MIMEText
from email.mime.multipart import MIMEMultipart

def send_email_alert(
    subject: str,
    message: str,
    to_email: str,
    from_email: str,
    smtp_server: str,
    smtp_port: int = 587
):
    """Envía una alerta por email"""
    msg = MIMEMultipart()
    msg['From'] = from_email
    msg['To'] = to_email
    msg['Subject'] = subject
    
    msg.attach(MIMEText(message, 'plain'))
    
    with smtplib.SMTP(smtp_server, smtp_port) as server:
        server.starttls()
        server.login(from_email, 'YOUR_PASSWORD')
        server.send_message(msg)

# Uso
send_email_alert(
    subject='Automation Alert: Backup Failed',
    message='The backup script failed with error: ...',
    to_email='admin@example.com',
    from_email='automation@example.com',
    smtp_server='smtp.example.com'
)

8. Monitoreo de sistemas con psutil

Supervisión de recursos del sistema:

# system_monitor.py
import psutil
import time

def monitor_system(interval: int = 60):
    """Supervisa los recursos del sistema"""
    while True:
        # CPU
        cpu_percent = psutil.cpu_percent(interval=1)
        
        # Memory
        memory = psutil.virtual_memory()
        
        # Disk
        disk = psutil.disk_usage('/')
        
        # Network
        network = psutil.net_io_counters()
        
        # Salida
        print(f"CPU: {cpu_percent}%")
        print(f"Memory: {memory.percent}%")
        print(f"Disk: {disk.percent}%")
        print(f"Network Sent: {network.bytes_sent} bytes")
        print(f"Network Recv: {network.bytes_recv} bytes")
        print("-" * 40)
        
        # Alertas ante valores elevados
        if cpu_percent > 90:
            send_alert(f"High CPU usage: {cpu_percent}%")
        if memory.percent > 90:
            send_alert(f"High memory usage: {memory.percent}%")
        if disk.percent > 90:
            send_alert(f"High disk usage: {disk.percent}%")
        
        time.sleep(interval)

9. Rotación y gestión de logs

Configuración de logrotate:

# /etc/logrotate.d/automation
/var/log/automation/*.log {
    daily
    rotate 14
    compress
    delaycompress
    missingok
    notifempty
    create 0640 automation automation
    sharedscripts
    postrotate
        systemctl reload automation-service > /dev/null 2>&1 || true
    endscript
}

10. Monitoreo de disponibilidad

Verificación de uptime con Python:

# uptime_monitor.py
import requests
import time
from datetime import datetime

def check_uptime(url: str, interval: int = 60):
    """Supervisa la disponibilidad de un servicio"""
    downtime_start = None
    total_downtime = 0
    
    while True:
        try:
            response = requests.get(url, timeout=10)
            if response.status_code == 200:
                if downtime_start:
                    downtime_duration = time.time() - downtime_start
                    total_downtime += downtime_duration
                    send_alert(f"Service is back up after {downtime_duration:.2f}s downtime")
                    downtime_start = None
                print(f"{datetime.now()}: Service is UP")
            else:
                if not downtime_start:
                    downtime_start = time.time()
                    send_alert(f"Service returned status {response.status_code}")
                print(f"{datetime.now()}: Service returned {response.status_code}")
        except requests.RequestException as e:
            if not downtime_start:
                downtime_start = time.time()
                send_alert(f"Service is down: {str(e)}")
            print(f"{datetime.now()}: Service is DOWN - {str(e)}")
        
        time.sleep(interval)

Manejo avanzado de errores en scripts de automatización

1. Manejo estructurado de excepciones

Clases de excepciones personalizadas:

# exceptions.py
class AutomationError(Exception):
    """Excepción base para scripts de automatización"""
    pass

class BackupError(AutomationError):
    """Errores específicos de backup"""
    pass

class ConfigError(AutomationError):
    """Errores de configuración"""
    pass

class NetworkError(AutomationError):
    """Errores de red"""
    pass

class ValidationError(AutomationError):
    """Errores de validación"""
    pass

2. Mecanismos de reintentos con backoff exponencial

Reintentos inteligentes:

# retry_handler.py
import time
import random
from functools import wraps
from typing import Callable, Type

def retry_with_backoff(
    max_retries: int = 3,
    initial_delay: float = 1.0,
    max_delay: float = 60.0,
    exponential_base: float = 2,
    jitter: bool = True,
    exceptions: tuple = (Exception,)
):
    """Decorador con backoff exponencial y jitter"""
    def decorator(func: Callable) -> Callable:
        @wraps(func)
        def wrapper(*args, **kwargs):
            delay = initial_delay
            
            for attempt in range(max_retries):
                try:
                    return func(*args, **kwargs)
                except exceptions as e:
                    if attempt == max_retries - 1:
                        raise
                    
                    # Agregar jitter para evitar thundering herd
                    if jitter:
                        delay = delay * (1 + random.uniform(-0.1, 0.1))
                    
                    print(f"Attempt {attempt + 1} failed: {e}. Retrying in {delay:.2f}s...")
                    time.sleep(min(delay, max_delay))
                    delay *= exponential_base
            
        return wrapper
    return decorator

# Uso
@retry_with_backoff(max_retries=5, initial_delay=2.0, exceptions=(ConnectionError, TimeoutError))
def download_file(url: str, target: str):
    """Descarga un archivo con reintentos"""
    # Lógica de descarga
    pass

3. Patrón Circuit Breaker

Circuit Breaker para servicios externos:

# circuit_breaker.py
from enum import Enum
import time

class CircuitState(Enum):
    CLOSED = "closed"  # Operación normal
    OPEN = "open"      # Circuito abierto, solicitudes fallan inmediatamente
    HALF_OPEN = "half_open"  # Prueba de recuperación del servicio

class CircuitBreaker:
    def __init__(
        self,
        failure_threshold: int = 5,
        recovery_timeout: float = 60.0,
        expected_exception: Exception = Exception
    ):
        self.failure_threshold = failure_threshold
        self.recovery_timeout = recovery_timeout
        self.expected_exception = expected_exception
        self.failure_count = 0
        self.state = CircuitState.CLOSED
        self.last_failure_time = None
    
    def call(self, func: Callable, *args, **kwargs):
        """Ejecuta una función con protección de circuit breaker"""
        if self.state == CircuitState.OPEN:
            if self._should_attempt_reset():
                self.state = CircuitState.HALF_OPEN
            else:
                raise Exception("Circuit breaker is OPEN")
        
        try:
            result = func(*args, **kwargs)
            self._on_success()
            return result
        except self.expected_exception as e:
            self._on_failure()
            raise
    
    def _should_attempt_reset(self) -> bool:
        """Verifica si el circuito debe reiniciarse"""
        return (
            self.last_failure_time and
            time.time() - self.last_failure_time >= self.recovery_timeout
        )
    
    def _on_success(self):
        """Tras una ejecución exitosa"""
        self.failure_count = 0
        if self.state == CircuitState.HALF_OPEN:
            self.state = CircuitState.CLOSED
    
    def _on_failure(self):
        """Tras una ejecución fallida"""
        self.failure_count += 1
        self.last_failure_time = time.time()
        
        if self.failure_count >= self.failure_threshold:
            self.state = CircuitState.OPEN

# Uso
circuit_breaker = CircuitBreaker(failure_threshold=3, recovery_timeout=30)

def call_external_api():
    """Llamada a API externa con circuit breaker"""
    return circuit_breaker.call(requests.get, "https://api.example.com")

4. Dead Letter Queue para tareas fallidas

Implementación de Dead Letter Queue:

# dead_letter_queue.py
import json
import os
from datetime import datetime
from typing import Any, Dict

class DeadLetterQueue:
    def __init__(self, queue_file: str = "dead_letter_queue.json"):
        self.queue_file = queue_file
        self._load_queue()
    
    def _load_queue(self):
        """Carga la queue desde archivo"""
        if os.path.exists(self.queue_file):
            with open(self.queue_file, 'r') as f:
                self.queue = json.load(f)
        else:
            self.queue = []
    
    def _save_queue(self):
        """Guarda la queue en archivo"""
        with open(self.queue_file, 'w') as f:
            json.dump(self.queue, f, indent=2)
    
    def add(self, task: Dict[str, Any], error: Exception):
        """Añade una tarea fallida a la queue"""
        dead_letter = {
            'task': task,
            'error': str(error),
            'error_type': type(error).__name__,
            'timestamp': datetime.now().isoformat(),
            'retry_count': 0
        }
        self.queue.append(dead_letter)
        self._save_queue()
    
    def get_next(self) -> Dict[str, Any]:
        """Obtiene la siguiente tarea de la queue"""
        if not self.queue:
            return None
        
        return self.queue.pop(0)
    
    def retry(self, max_retries: int = 3):
        """Reintenta ejecutar las tareas"""
        remaining_tasks = []
        
        for dead_letter in self.queue:
            if dead_letter['retry_count'] < max_retries:
                try:
                    # Ejecuta la tarea nuevamente
                    execute_task(dead_letter['task'])
                    print(f"Task {dead_letter['task']['id']} reintentada exitosamente")
                except Exception as e:
                    dead_letter['retry_count'] += 1
                    dead_letter['last_error'] = str(e)
                    remaining_tasks.append(dead_letter)
            else:
                print(f"Task {dead_letter['task']['id']} alcanzó el máximo de reintentos")
        
        self.queue = remaining_tasks
        self._save_queue()

# Uso
dlq = DeadLetterQueue()

try:
    execute_task(task)
except Exception as e:
    dlq.add(task, e)

5. Graceful Degradation

Estrategias de fallback:

# graceful_degradation.py
from typing import Optional, Callable

class FallbackChain:
    def __init__(self):
        self.fallbacks = []
    
    def add_fallback(self, func: Callable, priority: int = 0):
        """Añade una función de fallback"""
        self.fallbacks.append((priority, func))
        self.fallbacks.sort(key=lambda x: x[0], reverse=True)
    
    def execute(self, *args, **kwargs) -> Optional[Any]:
        """Ejecuta funciones con fallback automático"""
        for priority, func in self.fallbacks:
            try:
                result = func(*args, **kwargs)
                if result is not None:
                    return result
            except Exception as e:
                print(f"Fallback {func.__name__} falló: {e}")
                continue
        
        return None

# Uso
fallback_chain = FallbackChain()

# Método primario
def primary_backup(source: str, target: str):
    """Método de backup primario (Cloud)"""
    return upload_to_cloud(source, target)

# Fallback 1
def secondary_backup(source: str, target: str):
    """Método de backup secundario (FTP)"""
    return upload_to_ftp(source, target)

# Fallback 2
def tertiary_backup(source: str, target: str):
    """Método de backup terciario (Local)"""
    return copy_locally(source, target)

fallback_chain.add_fallback(primary_backup, priority=3)
fallback_chain.add_fallback(secondary_backup, priority=2)
fallback_chain.add_fallback(tertiary_backup, priority=1)

result = fallback_chain.execute(source, target)

6. Logging comprensivo para errores

Error logging detallado:

# error_logger.py
import logging
import traceback
import sys
from datetime import datetime
from typing import Optional

class ErrorLogger:
    def __init__(self, log_file: str = "automation_errors.log"):
        self.logger = logging.getLogger('error_logger')
        self.logger.setLevel(logging.ERROR)
        
        # File Handler
        file_handler = logging.FileHandler(log_file)
        file_handler.setLevel(logging.ERROR)
        
        # Formatter
        formatter = logging.Formatter(
            '%(asctime)s - %(name)s - %(levelname)s - %(message)s'
        )
        file_handler.setFormatter(formatter)
        
        self.logger.addHandler(file_handler)
    
    def log_error(
        self,
        error: Exception,
        context: Optional[dict] = None,
        critical: bool = False
    ):
        """Registra un error con contexto"""
        error_info = {
            'error_type': type(error).__name__,
            'error_message': str(error),
            'traceback': traceback.format_exc(),
            'context': context or {},
            'timestamp': datetime.now().isoformat()
        }
        
        log_message = f"Error: {error_info}"
        if critical:
            self.logger.critical(log_message)
        else:
            self.logger.error(log_message)
        
        return error_info
    
    def log_exception(self, func: Callable):
        """Decorador para logging automático de excepciones"""
        def wrapper(*args, **kwargs):
            try:
                return func(*args, **kwargs)
            except Exception as e:
                self.log_error(e, context={'function': func.__name__})
                raise
        return wrapper

# Uso
error_logger = ErrorLogger()

@error_logger.log_exception
def risky_operation():
    """Operación riesgosa con logging automático"""
    pass

7. Estrategias de recuperación de errores

Recuperación automática:

# recovery.py
from typing import Callable, Any

class RecoveryManager:
    def __init__(self):
        self.recovery_strategies = {}
    
    def register_recovery(self, error_type: type, recovery_func: Callable):
        """Registra una estrategia de recuperación para un tipo de error"""
        self.recovery_strategies[error_type] = recovery_func
    
    def execute_with_recovery(self, func: Callable, *args, **kwargs) -> Any:
        """Ejecuta una función con recuperación automática"""
        try:
            return func(*args, **kwargs)
        except Exception as e:
            error_type = type(e)
            
            if error_type in self.recovery_strategies:
                print(f"Intentando recuperación para {error_type.__name__}")
                recovery_func = self.recovery_strategies[error_type]
                return recovery_func(e, *args, **kwargs)
            else:
                raise

# Uso
recovery_manager = RecoveryManager()

def recover_from_connection_error(error: Exception, *args, **kwargs):
    """Recuperación de errores de conexión"""
    print("Reconectando...")
    time.sleep(5)
    return func(*args, **kwargs)

def recover_from_disk_full(error: Exception, *args, **kwargs):
    """Recuperación de disco lleno"""
    print("Limpiando archivos antiguos...")
    cleanup_old_files()
    return func(*args, **kwargs)

recovery_manager.register_recovery(ConnectionError, recover_from_connection_error)
recovery_manager.register_recovery(OSError, recover_from_disk_full)

8. Gestión de Timeouts

Timeout para operaciones de larga duración:

# timeout_handler.py
import signal
from contextlib import contextmanager
import time

class TimeoutError(Exception):
    """Custom Timeout Exception"""
    pass

@contextmanager
def timeout_handler(seconds: int):
    """Context Manager para timeout"""
    def signal_handler(signum, frame):
        raise TimeoutError(f"Operation timed out after {seconds} seconds")
    
    # Registrar signal handler
    old_handler = signal.signal(signal.SIGALRM, signal_handler)
    signal.alarm(seconds)
    
    try:
        yield
    finally:
        # Restaurar signal handler
        signal.alarm(0)
        signal.signal(signal.SIGALRM, old_handler)

# Uso
try:
    with timeout_handler(30):
        long_running_operation()
except TimeoutError:
    print("Operation timed out, using fallback")
    fallback_operation()

9. Agregación de Errores

Recopilar múltiples errores:

# error_aggregator.py
from typing import List, Dict, Any

class ErrorAggregator:
    def __init__(self):
        self.errors: List[Dict[str, Any]] = []
    
    def add_error(self, error: Exception, context: Dict[str, Any] = None):
        """Añade un error a la colección"""
        self.errors.append({
            'error': str(error),
            'type': type(error).__name__,
            'context': context or {},
            'timestamp': time.time()
        })
    
    def has_errors(self) -> bool:
        """Verifica si hay errores registrados"""
        return len(self.errors) > 0
    
    def get_summary(self) -> Dict[str, Any]:
        """Retorna un resumen de los errores"""
        error_types = {}
        for error in self.errors:
            error_type = error['type']
            error_types[error_type] = error_types.get(error_type, 0) + 1
        
        return {
            'total_errors': len(self.errors),
            'error_types': error_types,
            'errors': self.errors
        }
    
    def clear(self):
        """Borra todos los errores"""
        self.errors = []

# Uso
aggregator = ErrorAggregator()

for item in items:
    try:
        process_item(item)
    except Exception as e:
        aggregator.add_error(e, context={'item': item})

if aggregator.has_errors():
    print(f"Processing completed with {len(aggregator.errors)} errors")
    print(aggregator.get_summary())

10. Sistema de Notificación de Errores

Notificaciones ante errores críticos:

# error_notifier.py
from typing import List, Callable

class ErrorNotifier:
    def __init__(self):
        self.notifiers: List[Callable] = []
    
    def add_notifier(self, notifier: Callable):
        """Añade una función de notificación"""
        self.notifiers.append(notifier)
    
    def notify(self, error: Exception, context: dict = None):
        """Envía notificación a todos los notifiers registrados"""
        error_info = {
            'error': str(error),
            'type': type(error).__name__,
            'context': context or {},
            'timestamp': time.time()
        }
        
        for notifier in self.notifiers:
            try:
                notifier(error_info)
            except Exception as e:
                print(f"Notifier failed: {e}")

# Ejemplo de notifiers
def slack_notifier(error_info: dict):
    """Notificación por Slack"""
    send_slack_message(f"Error: {error_info['error']}")

def email_notifier(error_info: dict):
    """Notificación por email"""
    send_email(f"Error: {error_info['error']}", error_info)

# Uso
notifier = ErrorNotifier()
notifier.add_notifier(slack_notifier)
notifier.add_notifier(email_notifier)

try:
    critical_operation()
except Exception as e:
    notifier.notify(e, context={'operation': 'critical_operation'})