Herramientas de Automatización del Sistema con Python

Este artículo presenta cinco scripts en Python diseñados para tareas comunes de administración de sistemas: organización de archivos por extensión, limpieza de directorios vacíos, finalización segura de procesos, monitoreo proactivo del espacio en disco y verificación del estado de servidores web.

1. Clasificación automática de archivos por extensión

Mueve o copia archivos desde un directorio raíz a subcarpetas nombraads según su extensión (por ejemplo, .pdf → carpeta pdf). El script crea automáticamente las carpetas destino si no existen y opera en modo copia por defecto (para evitar pérdida accidental de datos).

import os
import shutil
import sys

def extract_extension(file_path):
    _, ext = os.path.splitext(file_path)
    return ext.lstrip('.') if ext else 'sin_extension'

def organize_by_extension(root_dir, mode='copy'):
    for entry in os.listdir(root_dir):
        full_path = os.path.join(root_dir, entry)
        if os.path.isfile(full_path):
            ext = extract_extension(entry)
            target_dir = os.path.join(root_dir, ext)
            
            os.makedirs(target_dir, exist_ok=True)
            
            destination = os.path.join(target_dir, entry)
            if mode == 'move':
                shutil.move(full_path, destination)
            else:
                shutil.copy2(full_path, destination)

if len(sys.argv) > 1:
    organize_by_extension(sys.argv[1])
else:
    print("Uso: python organizar.py <ruta_del_directorio>")

2. Detección y registro de carpetas vacías

Recorre recursivamente un árbol de directorios desde abajo hacia arriba (topdown=False) para identificar carpetas sin contenido. Registra sus rutas absolutas en un archivo de texto y ofrece opción de eliminación segura mediante confirmación explícita.

import os
import sys

def log_empty_dirs(base_path, log_file):
    empty_paths = []
    
    for root, dirs, files in os.walk(base_path, topdown=False):
        for d in dirs:
            path = os.path.join(root, d)
            if not os.listdir(path):
                empty_paths.append(path)
    
    with open(log_file, 'w', encoding='utf-8') as f:
        for p in empty_paths:
            f.write(p + '\n')
    
    print(f"Se registraron {len(empty_paths)} directorios vacíos en {log_file}")

if len(sys.argv) == 3:
    log_empty_dirs(sys.argv[1], sys.argv[2])
else:
    print("Uso: python vacios.py <directorio_raiz> <archivo_log.txt>")

3. Finalización selectiva de procesos por nombre

Busca instancias activas de un proceso dado (ignorando mayúsculas/minúsculas), muestra información detallada (PID, nombre, usuario) y permite su terminación controlada. Incluye manejo robusto de excepciones como procesos ya finalizados o permisos insuficientes.

import psutil
import sys

def find_processes(name):
    matches = []
    for proc in psutil.process_iter(['pid', 'name', 'username']):
        try:
            if proc.info['name'].lower() == name.lower():
                matches.append(proc.info)
        except (psutil.NoSuchProcess, psutil.AccessDenied):
            continue
    return matches

def terminate_all(name):
    for proc in find_processes(name):
        try:
            p = psutil.Process(proc['pid'])
            p.terminate()
            print(f"✅ Terminado: {proc['name']} (PID {proc['pid']})")
        except psutil.NoSuchProcess:
            print(f"⚠️  Ya finalizado: PID {proc['pid']}")
        except Exception as e:
            print(f"❌ Error al terminar PID {proc['pid']}: {e}")

if __name__ == "__main__":
    if len(sys.argv) != 2:
        print("Uso: python matar.py <nombre_del_proceso>")
        sys.exit(1)
    
    target = sys.argv[1]
    found = find_processes(target)
    
    if not found:
        print(f"🔍 No se encontró ningún proceso llamado '{target}'.")
    else:
        print(f"📋 Procesos encontrados ({len(found)}):")
        for p in found:
            print(f"  PID {p['pid']} — {p['name']} — Usuario: {p['username']}")
        terminate_all(target)

4. Alerta automática por bajo espacio en disco

Evalúa el uso del sistema de archivos raíz y envía una notificación por correo electrónico cuando el espacio libre cae por debajo de un umbral configurable (en GB). Usa SMTP seguro con autenticación y soporta credenciales mediante variables de entorno para mayor seguridad.

import psutil
import smtplib
import os
from email.mime.text import MIMEText
from email.mime.multipart import MIMEMultipart

def send_alert(recipient, free_gb, threshold):
    subject = "⚠️ Alerta de almacenamiento: espacio crítico"
    body = f"""El sistema ha detectado que el espacio libre en '/' es de {free_gb:.1f} GB,
muy por debajo del umbral configurado de {threshold} GB.

Por favor, revise los archivos temporales, registros o cachés acumulados.
"""
    
    msg = MIMEMultipart()
    msg['Subject'] = subject
    msg['From'] = os.getenv('SMTP_USER', 'alert@server.local')
    msg['To'] = recipient
    msg.attach(MIMEText(body, 'plain'))

    try:
        server = smtplib.SMTP_SSL(
            os.getenv('SMTP_HOST', 'smtp.gmail.com'),
            int(os.getenv('SMTP_PORT', '465'))
        )
        server.login(
            os.getenv('SMTP_USER'),
            os.getenv('SMTP_PASS')
        )
        server.send_message(msg)
        server.quit()
        print("📧 Alerta enviada correctamente.")
    except Exception as e:
        print(f"❌ Fallo al enviar alerta: {e}")

def monitor_disk(threshold_gb=20, alert_to="admin@example.com"):
    usage = psutil.disk_usage('/')
    free_gb = usage.free / (1024**3)
    
    if free_gb < threshold_gb:
        send_alert(alert_to, free_gb, threshold_gb)
    else:
        print(f"🟢 Espacio disponible: {free_gb:.1f} GB (umbral: {threshold_gb} GB)")

# Ejecución directa con valores predeterminados
monitor_disk(threshold_gb=25, alert_to="admin@empresa.local")

5. Verificación de disponibilidad HTTP

Realiza solicitudes GET a una URL especificada y valida tanto la respuesta HTTP como la conectividad de red. Soporta tiempos de espera personalizables y reporta errores específicos (timeout, DNS falido, código de estado inesperado).

import requests
import sys

def health_check(url, timeout=10):
    try:
        response = requests.get(url, timeout=timeout, allow_redirects=True)
        status = response.status_code
        
        if 200 <= status < 400:
            print(f"🟢 {url} está operativo (HTTP {status})")
            return True
        elif status >= 400:
            print(f"🔴 {url} responde con error HTTP {status}")
            return False
    except requests.exceptions.Timeout:
        print(f"🔴 {url} agotó el tiempo de espera ({timeout}s)")
    except requests.exceptions.ConnectionError:
        print(f"🔴 No se puede conectar a {url} (error de red o DNS)")
    except requests.exceptions.RequestException as e:
        print(f"🔴 Error inesperado al verificar {url}: {e}")
    
    return False

if __name__ == "__main__":
    url = sys.argv[1] if len(sys.argv) > 1 else "https://httpbin.org/get"
    health_check(url)

Etiquetas: Python psutil smpt requests filesystem

Publicado el 9-5 23:34