Python para Administradores de Sistemas: Automatizando Tareas de Infraestructura

Todo administrador de sistemas llega a un punto en el que los scripts de Bash se vuelven inmanejables. Una alerta de uso de disco que comenzo como un script de cinco lineas ahora abarca trescientas lineas, maneja mal los casos limite y falla silenciosamente en una distribucion diferente. Python cierra la brecha entre los comandos rapidos de shell y las herramientas completas de gestion de configuracion como Ansible. Te ofrece manejo adecuado de errores, estructuras de datos ricas, miles de librerias y codigo que es legible meses despues de haberlo escrito.

Este articulo recorre las habilidades esenciales de Python que un administrador de sistemas Linux necesita: operaciones con archivos, gestion de subprocesos, automatizacion SSH, integracion con APIs REST, analisis de logs, monitoreo de sistemas y notificaciones. Cada seccion incluye codigo listo para produccion que puedes adaptar a tu entorno.

---

Requisitos Previos

Antes de comenzar, asegurate de tener lo siguiente:

• Python 3.10+ — verifica con python3 --version
• pip — el instalador de paquetes de Python (generalmente incluido con Python)
• venv — el modulo integrado de entornos virtuales
• Un servidor Linux (Ubuntu 22.04/24.04, Debian 12 o RHEL 9 recomendados)
• Familiaridad basica con la linea de comandos de Linux y Bash
• Un par de claves SSH configurado para acceso a servidores remotos (ver Hardening SSH)

# Verificar la instalacion de Python
python3 --version
# Python 3.12.3

# Verificar pip
pip3 --version
# pip 24.0 from /usr/lib/python3/dist-packages/pip (python 3.12)

---

Configurando un Entorno de Python

Nunca instales paquetes de Python de forma global en un servidor de produccion. Usa entornos virtuales para aislar las dependencias de cada proyecto.

Creando un entorno virtual

# Crear un directorio de proyecto
mkdir -p ~/sysadmin-scripts
cd ~/sysadmin-scripts

# Crear un entorno virtual
python3 -m venv venv

# Activarlo
source venv/bin/activate

# Tu prompt cambia para mostrar (venv)
(venv) user@server:~/sysadmin-scripts$

Instalando librerias esenciales

# Instalar todas las librerias que usaremos en este articulo
pip install paramiko requests psutil python-dotenv jinja2

# Congelar dependencias para reproducibilidad
pip freeze > requirements.txt

Estructura del proyecto

Un proyecto de scripts de administracion de sistemas bien organizado luce asi:

sysadmin-scripts/
├── venv/
├── requirements.txt
├── .env                  # Secretos (claves API, contrasenas)
├── config.yaml           # Listas de servidores, umbrales
├── scripts/
│   ├── disk_monitor.py
│   ├── log_parser.py
│   ├── remote_exec.py
│   └── backup_rotation.py
└── logs/
    └── automation.log

---

Operaciones con Archivos y Directorios

El modulo pathlib de Python (parte de la libreria estandar desde Python 3.4) proporciona una interfaz orientada a objetos para rutas del sistema de archivos. Combinado con shutil, maneja practicamente cualquier operacion de archivos que un administrador de sistemas necesite.

Trabajando con pathlib

from pathlib import Path
import shutil
from datetime import datetime, timedelta

# Crear rutas (funciona en cualquier SO)
log_dir = Path("/var/log/myapp")
backup_dir = Path("/backup/logs")

# Crear directorios (parents=True actua como mkdir -p)
backup_dir.mkdir(parents=True, exist_ok=True)

# Listar todos los archivos .log
for log_file in log_dir.glob("*.log"):
    print(f"{log_file.name}  {log_file.stat().st_size / 1024:.1f} KB")

# Buscar archivos recursivamente
for conf in Path("/etc").rglob("*.conf"):
    print(conf)

Script de rotacion automatica de logs

#!/usr/bin/env python3
"""Rotar y comprimir archivos de log mas antiguos que N dias."""

import gzip
import shutil
from pathlib import Path
from datetime import datetime, timedelta

LOG_DIR = Path("/var/log/myapp")
ARCHIVE_DIR = Path("/var/log/myapp/archive")
MAX_AGE_DAYS = 7

ARCHIVE_DIR.mkdir(parents=True, exist_ok=True)

cutoff = datetime.now() - timedelta(days=MAX_AGE_DAYS)

for log_file in LOG_DIR.glob("*.log"):
    mtime = datetime.fromtimestamp(log_file.stat().st_mtime)
    if mtime < cutoff:
        # Comprimir el archivo
        gz_path = ARCHIVE_DIR / f"{log_file.name}.gz"
        with open(log_file, "rb") as f_in:
            with gzip.open(gz_path, "wb") as f_out:
                shutil.copyfileobj(f_in, f_out)
        log_file.unlink()
        print(f"Archivado: {log_file.name} -> {gz_path.name}")

Script de respaldo con directorios con marca de tiempo

#!/usr/bin/env python3
"""Crear respaldos con marca de tiempo y eliminar respaldos mayores a 30 dias."""

import shutil
from pathlib import Path
from datetime import datetime, timedelta

SOURCE = Path("/etc/nginx")
BACKUP_ROOT = Path("/backup/nginx")
RETENTION_DAYS = 30

# Crear respaldo con marca de tiempo
timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
backup_path = BACKUP_ROOT / timestamp

shutil.copytree(SOURCE, backup_path)
print(f"Respaldo creado: {backup_path}")

# Limpiar respaldos antiguos
cutoff = datetime.now() - timedelta(days=RETENTION_DAYS)
for old_backup in BACKUP_ROOT.iterdir():
    if old_backup.is_dir():
        mtime = datetime.fromtimestamp(old_backup.stat().st_mtime)
        if mtime < cutoff:
            shutil.rmtree(old_backup)
            print(f"Respaldo antiguo eliminado: {old_backup.name}")

---

Ejecutando Comandos del Sistema con subprocess

El modulo subprocess te permite ejecutar comandos de shell desde Python con control total sobre la entrada, salida y manejo de errores. Siempre prefiere subprocess.run() sobre el mas antiguo os.system().

Ejecucion basica de comandos

import subprocess

# Ejecutar un comando y capturar la salida
result = subprocess.run(
    ["df", "-h", "/"],
    capture_output=True,
    text=True,
    timeout=30
)

print(result.stdout)
print(f"Codigo de retorno: {result.returncode}")

# Verificar errores
if result.returncode != 0:
    print(f"Error: {result.stderr}")

Ejecutando multiples comandos de forma segura

import subprocess
import sys

def run_cmd(cmd, description=""):
    """Ejecutar un comando y manejar errores."""
    try:
        result = subprocess.run(
            cmd,
            capture_output=True,
            text=True,
            timeout=60,
            check=True  # Lanza CalledProcessError en codigo de salida no cero
        )
        print(f"[OK] {description}")
        return result.stdout
    except subprocess.CalledProcessError as e:
        print(f"[FALLO] {description}: {e.stderr.strip()}")
        sys.exit(1)
    except subprocess.TimeoutExpired:
        print(f"[TIMEOUT] {description}: el comando excedio 60s")
        sys.exit(1)

# Ejemplo: secuencia de actualizacion del sistema
run_cmd(["sudo", "apt", "update"], "Actualizar listas de paquetes")
run_cmd(["sudo", "apt", "upgrade", "-y"], "Actualizar paquetes")
run_cmd(["sudo", "apt", "autoremove", "-y"], "Eliminar paquetes no utilizados")

Analizando la salida de comandos

import subprocess

def get_listening_ports():
    """Retornar una lista de puertos con servicios escuchando."""
    result = subprocess.run(
        ["ss", "-tlnp"],
        capture_output=True,
        text=True,
        check=True
    )
    ports = []
    for line in result.stdout.strip().split("\n")[1:]:  # Saltar encabezado
        parts = line.split()
        if len(parts) >= 4:
            addr = parts[3]
            port = addr.rsplit(":", 1)[-1]
            ports.append(port)
    return sorted(set(ports))

print("Puertos escuchando:", get_listening_ports())

---

Automatizacion SSH con Paramiko

Paramiko es la libreria estandar de Python para conexiones SSH2. Te permite ejecutar comandos en servidores remotos, transferir archivos por SFTP y gestionar claves SSH — todo desde Python.

Instalando Paramiko

pip install paramiko

Ejecutando comandos remotos

import paramiko

def ssh_exec(hostname, username, command, key_path="~/.ssh/id_ed25519"):
    """Ejecutar un comando en un servidor remoto via SSH."""
    client = paramiko.SSHClient()
    client.set_missing_host_key_policy(paramiko.AutoAddPolicy())

    try:
        client.connect(
            hostname=hostname,
            username=username,
            key_filename=str(Path(key_path).expanduser()),
            timeout=10
        )
        stdin, stdout, stderr = client.exec_command(command, timeout=30)
        exit_code = stdout.channel.recv_exit_status()

        return {
            "stdout": stdout.read().decode().strip(),
            "stderr": stderr.read().decode().strip(),
            "exit_code": exit_code
        }
    finally:
        client.close()

# Ejemplo de uso
result = ssh_exec("web01.example.com", "deploy", "uptime")
print(result["stdout"])

Ejecutando comandos en multiples servidores

import paramiko
from pathlib import Path
from concurrent.futures import ThreadPoolExecutor, as_completed

SERVERS = [
    {"host": "web01.example.com", "user": "deploy"},
    {"host": "web02.example.com", "user": "deploy"},
    {"host": "db01.example.com",  "user": "deploy"},
]

def check_uptime(server):
    """Verificar el tiempo de actividad en un solo servidor."""
    client = paramiko.SSHClient()
    client.set_missing_host_key_policy(paramiko.AutoAddPolicy())
    try:
        client.connect(
            hostname=server["host"],
            username=server["user"],
            key_filename=str(Path("~/.ssh/id_ed25519").expanduser()),
            timeout=10
        )
        _, stdout, _ = client.exec_command("uptime -p")
        return server["host"], stdout.read().decode().strip()
    except Exception as e:
        return server["host"], f"ERROR: {e}"
    finally:
        client.close()

# Ejecutar en paralelo con ThreadPoolExecutor
with ThreadPoolExecutor(max_workers=5) as executor:
    futures = {executor.submit(check_uptime, s): s for s in SERVERS}
    for future in as_completed(futures):
        host, uptime = future.result()
        print(f"{host}: {uptime}")

Transferencias de archivos por SFTP

import paramiko
from pathlib import Path

def upload_file(hostname, username, local_path, remote_path):
    """Subir un archivo a un servidor remoto via SFTP."""
    transport = paramiko.Transport((hostname, 22))
    key = paramiko.Ed25519Key.from_private_key_file(
        str(Path("~/.ssh/id_ed25519").expanduser())
    )
    transport.connect(username=username, pkey=key)
    sftp = paramiko.SFTPClient.from_transport(transport)

    try:
        sftp.put(str(local_path), str(remote_path))
        print(f"Subido {local_path} -> {hostname}:{remote_path}")
    finally:
        sftp.close()
        transport.close()

# Subir un archivo de configuracion
upload_file(
    "web01.example.com",
    "deploy",
    Path("./configs/nginx.conf"),
    "/tmp/nginx.conf"
)

---

Trabajando con APIs REST

La libreria requests hace que las llamadas HTTP sean sencillas. Los administradores de sistemas usan APIs para interactuar con plataformas de monitoreo, proveedores de nube, servicios DNS y herramientas internas.

Solicitudes GET — obteniendo datos

import requests

# Verificar el estado del servidor desde una API de monitoreo
response = requests.get(
    "https://api.example.com/v1/servers",
    headers={"Authorization": "Bearer YOUR_API_KEY"},
    timeout=10
)

if response.status_code == 200:
    servers = response.json()
    for server in servers:
        print(f"{server['name']}: {server['status']}")
else:
    print(f"Error de API: {response.status_code} {response.text}")

Solicitudes POST — enviando datos

import requests
import json

# Crear un registro DNS via API
payload = {
    "type": "A",
    "name": "web03.example.com",
    "content": "203.0.113.50",
    "ttl": 300
}

response = requests.post(
    "https://api.cloudflare.com/client/v4/zones/ZONE_ID/dns_records",
    headers={
        "Authorization": "Bearer YOUR_CF_TOKEN",
        "Content-Type": "application/json"
    },
    json=payload,
    timeout=15
)

result = response.json()
if result.get("success"):
    print(f"Registro DNS creado: {result['result']['id']}")
else:
    print(f"Error: {result.get('errors')}")

Cliente API robusto con reintentos

import requests
from requests.adapters import HTTPAdapter
from urllib3.util.retry import Retry

def create_api_session(base_url, api_key, retries=3):
    """Crear una sesion de requests con logica de reintentos."""
    session = requests.Session()
    session.headers.update({
        "Authorization": f"Bearer {api_key}",
        "Content-Type": "application/json"
    })

    retry_strategy = Retry(
        total=retries,
        backoff_factor=1,
        status_forcelist=[429, 500, 502, 503, 504]
    )
    adapter = HTTPAdapter(max_retries=retry_strategy)
    session.mount("https://", adapter)
    session.mount("http://", adapter)

    return session

# Uso
api = create_api_session("https://api.example.com", "YOUR_KEY")
response = api.get("https://api.example.com/v1/health", timeout=10)
print(response.json())

---

Analisis y Procesamiento de Logs

Los administradores de sistemas dedican una cantidad significativa de tiempo al analisis de archivos de log. Python facilita el analisis, filtrado y agregacion de datos de log.

Analizando syslog con expresiones regulares

#!/usr/bin/env python3
"""Analizar syslog y reportar intentos fallidos de inicio de sesion SSH."""

import re
from collections import Counter
from pathlib import Path

LOG_FILE = Path("/var/log/auth.log")
PATTERN = re.compile(
    r"(\w+\s+\d+\s+[\d:]+)\s+\S+\s+sshd\[\d+\]:\s+"
    r"Failed password for (?:invalid user )?(\S+) from (\S+)"
)

ip_counter = Counter()
user_counter = Counter()

with open(LOG_FILE) as f:
    for line in f:
        match = PATTERN.search(line)
        if match:
            timestamp, user, ip = match.groups()
            ip_counter[ip] += 1
            user_counter[user] += 1

print("Top 10 IPs atacantes:")
for ip, count in ip_counter.most_common(10):
    print(f"  {ip:20s} {count:5d} intentos")

print("\nTop 10 nombres de usuario objetivo:")
for user, count in user_counter.most_common(10):
    print(f"  {user:20s} {count:5d} intentos")

Analizando logs estructurados (JSON)

#!/usr/bin/env python3
"""Analizar logs de aplicacion en formato JSON."""

import json
from collections import Counter
from pathlib import Path
from datetime import datetime

LOG_FILE = Path("/var/log/myapp/access.json")

status_codes = Counter()
slow_requests = []

with open(LOG_FILE) as f:
    for line in f:
        try:
            entry = json.loads(line)
            status_codes[entry["status"]] += 1
            if entry.get("response_time", 0) > 2.0:
                slow_requests.append(entry)
        except json.JSONDecodeError:
            continue

print("Distribucion de codigos de estado:")
for code, count in sorted(status_codes.items()):
    print(f"  {code}: {count}")

print(f"\nSolicitudes lentas (>2s): {len(slow_requests)}")
for req in slow_requests[:5]:
    print(f"  {req['method']} {req['path']} - {req['response_time']:.2f}s")

Seguimiento de logs en tiempo real

#!/usr/bin/env python3
"""Seguir un archivo de log y alertar sobre patrones de error."""

import time
from pathlib import Path

LOG_FILE = Path("/var/log/myapp/error.log")
ALERT_PATTERNS = ["CRITICAL", "OOM", "segfault", "disk full"]

def tail_file(filepath, interval=1.0):
    """Producir nuevas lineas a medida que se agregan a un archivo."""
    with open(filepath) as f:
        f.seek(0, 2)  # Ir al final del archivo
        while True:
            line = f.readline()
            if line:
                yield line.strip()
            else:
                time.sleep(interval)

print(f"Siguiendo {LOG_FILE}... (Ctrl+C para detener)")
for line in tail_file(LOG_FILE):
    for pattern in ALERT_PATTERNS:
        if pattern in line:
            print(f"[ALERTA] {line}")
            # Aqui podrias enviar una notificacion
            break

---

Scripts de Monitoreo del Sistema

La libreria psutil proporciona acceso multiplataforma a metricas del sistema — CPU, memoria, disco, red e informacion de procesos. Es la base para construir herramientas de monitoreo personalizadas.

Instalando psutil

pip install psutil

Verificacion completa del estado del sistema

#!/usr/bin/env python3
"""Script completo de verificacion del estado del sistema."""

import psutil
import socket
from datetime import datetime

def bytes_to_human(n):
    """Convertir bytes a formato legible para humanos."""
    for unit in ["B", "KB", "MB", "GB", "TB"]:
        if n < 1024:
            return f"{n:.1f} {unit}"
        n /= 1024
    return f"{n:.1f} PB"

def check_system():
    """Ejecutar todas las verificaciones del sistema y retornar resultados."""
    results = {}
    hostname = socket.gethostname()

    # CPU
    cpu_percent = psutil.cpu_percent(interval=1)
    cpu_count = psutil.cpu_count()
    load_avg = psutil.getloadavg()
    results["cpu"] = {
        "usage_percent": cpu_percent,
        "cores": cpu_count,
        "load_avg_1m": load_avg[0],
        "load_avg_5m": load_avg[1],
        "load_avg_15m": load_avg[2],
    }

    # Memoria
    mem = psutil.virtual_memory()
    swap = psutil.swap_memory()
    results["memory"] = {
        "total": bytes_to_human(mem.total),
        "used": bytes_to_human(mem.used),
        "available": bytes_to_human(mem.available),
        "percent": mem.percent,
        "swap_percent": swap.percent,
    }

    # Disco
    results["disks"] = []
    for partition in psutil.disk_partitions():
        try:
            usage = psutil.disk_usage(partition.mountpoint)
            results["disks"].append({
                "mount": partition.mountpoint,
                "device": partition.device,
                "total": bytes_to_human(usage.total),
                "used_percent": usage.percent,
            })
        except PermissionError:
            continue

    # Red
    net = psutil.net_io_counters()
    results["network"] = {
        "bytes_sent": bytes_to_human(net.bytes_sent),
        "bytes_recv": bytes_to_human(net.bytes_recv),
    }

    return hostname, results

hostname, health = check_system()
print(f"=== Reporte de Estado del Sistema: {hostname} ===")
print(f"Marca de tiempo: {datetime.now().isoformat()}\n")

print(f"CPU: {health['cpu']['usage_percent']}% "
      f"({health['cpu']['cores']} nucleos, "
      f"carga: {health['cpu']['load_avg_1m']:.2f})")

print(f"Memoria: {health['memory']['percent']}% "
      f"({health['memory']['used']} / {health['memory']['total']}) "
      f"Swap: {health['memory']['swap_percent']}%")

for disk in health["disks"]:
    status = "ADVERTENCIA" if disk["used_percent"] > 85 else "OK"
    print(f"Disco {disk['mount']}: {disk['used_percent']}% "
          f"de {disk['total']} [{status}]")

print(f"Red: enviados={health['network']['bytes_sent']}, "
      f"recibidos={health['network']['bytes_recv']}")

Alertas de uso de disco con umbrales

#!/usr/bin/env python3
"""Monitorear el uso de disco y alertar cuando se excedan los umbrales."""

import psutil

THRESHOLDS = {
    "/": 85,
    "/var": 80,
    "/home": 90,
    "/tmp": 75,
}

alerts = []

for partition in psutil.disk_partitions():
    mount = partition.mountpoint
    if mount in THRESHOLDS:
        usage = psutil.disk_usage(mount)
        if usage.percent > THRESHOLDS[mount]:
            alerts.append({
                "mount": mount,
                "usage": usage.percent,
                "threshold": THRESHOLDS[mount],
                "free_gb": usage.free / (1024 ** 3),
            })

if alerts:
    print("ALERTAS DE USO DE DISCO:")
    for alert in alerts:
        print(f"  {alert['mount']}: {alert['usage']}% usado "
              f"(umbral: {alert['threshold']}%, "
              f"libre: {alert['free_gb']:.1f} GB)")
else:
    print("Todas las particiones de disco dentro de los umbrales normales.")

Monitoreo de procesos

#!/usr/bin/env python3
"""Monitorear procesos especificos y alertar si dejan de ejecutarse."""

import psutil

REQUIRED_PROCESSES = ["nginx", "postgresql", "redis-server", "sshd"]

running = {p.name() for p in psutil.process_iter(["name"])}

for proc_name in REQUIRED_PROCESSES:
    if proc_name in running:
        print(f"  [EJECUTANDO] {proc_name}")
    else:
        print(f"  [DETENIDO] {proc_name} -- ACCION REQUERIDA")

---

Enviando Notificaciones

La automatizacion solo es util si sabes cuando algo sale mal. Python puede enviar alertas por correo electronico, Slack o cualquier servicio basado en webhooks.

Enviando notificaciones por correo electronico

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

def send_email(subject, body, to_addr, from_addr, smtp_server, smtp_port=587,
               username=None, password=None):
    """Enviar una notificacion por correo electronico."""
    msg = MIMEMultipart()
    msg["From"] = from_addr
    msg["To"] = to_addr
    msg["Subject"] = subject
    msg.attach(MIMEText(body, "plain"))

    with smtplib.SMTP(smtp_server, smtp_port) as server:
        server.starttls()
        if username and password:
            server.login(username, password)
        server.send_message(msg)

# Ejemplo de uso
send_email(
    subject="[ALERTA] Uso de disco superior al 90% en web01",
    body="La particion /var esta al 92% de uso. Espacio libre: 3.2 GB.",
    to_addr="admin@example.com",
    from_addr="alertas@example.com",
    smtp_server="smtp.example.com",
    username="alertas@example.com",
    password="contrasena-de-aplicacion"
)

Notificaciones por webhook de Slack

import requests
import json

def send_slack_alert(webhook_url, message, severity="warning"):
    """Enviar una alerta a un canal de Slack via webhook."""
    colors = {
        "info": "#36a64f",
        "warning": "#ff9900",
        "critical": "#ff0000",
    }

    payload = {
        "attachments": [{
            "color": colors.get(severity, "#36a64f"),
            "title": f"Alerta del Servidor ({severity.upper()})",
            "text": message,
            "footer": "Automatizacion SysAdmin",
        }]
    }

    response = requests.post(
        webhook_url,
        json=payload,
        timeout=10
    )
    return response.status_code == 200

# Ejemplo de uso
send_slack_alert(
    webhook_url="https://hooks.slack.com/services/T00/B00/XXXX",
    message="El uso de disco en /var excedio el 90% en web01.example.com",
    severity="critical"
)

Combinando monitoreo con notificaciones

#!/usr/bin/env python3
"""Monitorear recursos del sistema y enviar alertas cuando se excedan los umbrales."""

import psutil
import requests
import socket
from datetime import datetime

SLACK_WEBHOOK = "https://hooks.slack.com/services/T00/B00/XXXX"
HOSTNAME = socket.gethostname()
CPU_THRESHOLD = 90
MEMORY_THRESHOLD = 85
DISK_THRESHOLD = 85

def alert(message, severity="warning"):
    """Enviar alerta a Slack."""
    payload = {
        "text": f"*[{severity.upper()}]* `{HOSTNAME}` - {message}\n"
                f"_Marca de tiempo: {datetime.now().isoformat()}_"
    }
    requests.post(SLACK_WEBHOOK, json=payload, timeout=10)

# Verificar CPU
cpu = psutil.cpu_percent(interval=2)
if cpu > CPU_THRESHOLD:
    alert(f"Uso de CPU al {cpu}% (umbral: {CPU_THRESHOLD}%)", "critical")

# Verificar memoria
mem = psutil.virtual_memory()
if mem.percent > MEMORY_THRESHOLD:
    alert(f"Uso de memoria al {mem.percent}% (umbral: {MEMORY_THRESHOLD}%)", "critical")

# Verificar discos
for part in psutil.disk_partitions():
    try:
        usage = psutil.disk_usage(part.mountpoint)
        if usage.percent > DISK_THRESHOLD:
            alert(f"Disco {part.mountpoint} al {usage.percent}% "
                  f"(umbral: {DISK_THRESHOLD}%)", "warning")
    except PermissionError:
        continue

---

Programando Scripts de Python

Usando cron

La forma mas directa de programar scripts de Python en Linux es con cron.

# Editar tu crontab
crontab -e

# Ejecutar monitor de disco cada 15 minutos
*/15 * * * * /home/deploy/sysadmin-scripts/venv/bin/python /home/deploy/sysadmin-scripts/scripts/disk_monitor.py >> /var/log/disk_monitor.log 2>&1

# Ejecutar verificacion completa de estado cada hora
0 * * * * /home/deploy/sysadmin-scripts/venv/bin/python /home/deploy/sysadmin-scripts/scripts/health_check.py >> /var/log/health_check.log 2>&1

# Ejecutar analisis de logs diariamente a las 6 AM
0 6 * * * /home/deploy/sysadmin-scripts/venv/bin/python /home/deploy/sysadmin-scripts/scripts/log_parser.py >> /var/log/log_analysis.log 2>&1

Siempre usa la ruta completa al binario de Python dentro de tu entorno virtual. Esto asegura que el script use la version correcta de Python y tenga acceso a los paquetes instalados.

Usando temporizadores systemd

Para programacion mas avanzada con integracion de registro de logs y gestion de dependencias, usa temporizadores systemd.

# /etc/systemd/system/health-check.service
[Unit]
Description=System Health Check
After=network.target

[Service]
Type=oneshot
User=deploy
ExecStart=/home/deploy/sysadmin-scripts/venv/bin/python /home/deploy/sysadmin-scripts/scripts/health_check.py
StandardOutput=journal
StandardError=journal

# /etc/systemd/system/health-check.timer
[Unit]
Description=Run health check every 15 minutes

[Timer]
OnBootSec=5min
OnUnitActiveSec=15min
Persistent=true

[Install]
WantedBy=timers.target

# Habilitar e iniciar el temporizador
sudo systemctl daemon-reload
sudo systemctl enable health-check.timer
sudo systemctl start health-check.timer

# Verificar el estado del temporizador
systemctl list-timers --all | grep health

---

Tabla de Referencia de Librerias Utiles

Libreria	Proposito	Instalacion	Documentacion
pathlib	Operaciones de rutas de archivos/directorios	Integrada	docs.python.org/3/library/pathlib
subprocess	Ejecutar comandos del sistema	Integrada	docs.python.org/3/library/subprocess
shutil	Operaciones de archivos de alto nivel (copiar, mover, archivar)	Integrada	docs.python.org/3/library/shutil
os	Interfaz de bajo nivel con el SO, variables de entorno	Integrada	docs.python.org/3/library/os
paramiko	Protocolo SSH2, ejecucion remota de comandos, SFTP	pip install paramiko	paramiko.org
requests	Cliente HTTP para APIs REST	pip install requests	requests.readthedocs.io
psutil	Monitoreo del sistema (CPU, memoria, disco, red)	pip install psutil	psutil.readthedocs.io
python-dotenv	Cargar variables de entorno desde archivos .env	pip install python-dotenv	pypi.org/project/python-dotenv
jinja2	Motor de plantillas para generacion de archivos de configuracion	pip install jinja2	jinja.palletsprojects.com
pyyaml	Analizar y escribir archivos de configuracion YAML	pip install pyyaml	pyyaml.org
fabric	Automatizacion SSH de alto nivel (construido sobre Paramiko)	pip install fabric	fabfile.org
schedule	Programacion de tareas amigable para humanos	pip install schedule	schedule.readthedocs.io
logging	Registro estructurado de aplicaciones	Integrada	docs.python.org/3/library/logging

---

Solucion de Problemas

ModuleNotFoundError al ejecutar desde cron

Cron usa un entorno minimo. Si tu script falla con ModuleNotFoundError, probablemente no estas usando la ruta completa al binario de Python del entorno virtual.

# Incorrecto -- usa el Python del sistema, que carece de tus paquetes
* * * * * python3 /home/deploy/scripts/monitor.py

# Correcto -- usa el Python del venv
* * * * * /home/deploy/sysadmin-scripts/venv/bin/python /home/deploy/scripts/monitor.py

Conexion rechazada en Paramiko

Si Paramiko lanza ConnectionRefusedError, verifica que SSH este ejecutandose en el destino y que el puerto sea correcto.

# Especificar un puerto personalizado si tu servidor usa un puerto SSH no estandar
client.connect(hostname="web01.example.com", port=2222, username="deploy",
               key_filename="/home/deploy/.ssh/id_ed25519")

Permiso denegado en operaciones de archivos

Cuando los scripts se ejecutan como un usuario no root, pueden no tener acceso a archivos bajo /var/log o /etc. Ejecuta el script con sudo o ajusta los permisos de archivos.

# Otorgar acceso de lectura a auth.log para el usuario deploy
sudo usermod -aG adm deploy

# O usa ACLs para control detallado
sudo setfacl -m u:deploy:r /var/log/auth.log

psutil retorna 0.0% de CPU en la primera llamada

psutil.cpu_percent() retorna 0.0 en la primera llamada porque necesita dos mediciones para calcular el uso. Siempre pasa un intervalo o llamalo dos veces.

# Correcto: pasar intervalo para una medicion bloqueante
cpu = psutil.cpu_percent(interval=1)

# O llamarlo dos veces con un retraso
psutil.cpu_percent()
import time; time.sleep(1)
cpu = psutil.cpu_percent()

Errores de timeout o SSL en requests

Al conectarte a APIs detras de proxies corporativos o con certificados autofirmados, puedes encontrar errores de SSL.

# Desactivar verificacion SSL (solo desarrollo -- nunca en produccion)
response = requests.get("https://internal-api.local", verify=False)

# O especificar un paquete CA personalizado
response = requests.get("https://api.example.com", verify="/etc/ssl/custom-ca.pem")

---

Resumen

Python es la navaja suiza de la administracion de sistemas. En esta guia aprendiste como:

• Configurar entornos de Python aislados con venv para proyectos de administracion de sistemas
• Gestionar archivos y directorios programaticamente con pathlib y shutil
• Ejecutar y analizar comandos del sistema con subprocess
• Automatizar conexiones SSH y transferencias de archivos con paramiko
• Interactuar con APIs REST usando requests con logica de reintentos adecuada
• Analizar y procesar archivos de log con expresiones regulares y procesamiento JSON
• Construir scripts de monitoreo usando psutil para metricas de CPU, memoria, disco y red
• Enviar notificaciones por correo electronico y webhooks de Slack
• Programar scripts de forma confiable con cron y temporizadores systemd

Comienza con algo pequeno: elige una tarea repetitiva que realizas semanalmente (rotacion de logs, verificaciones de estado, verificacion de respaldos) y automatizala con Python. A medida que crece tu confianza, encadena estos scripts en un kit de herramientas de automatizacion completo.

Para guias relacionadas sobre como asegurar los servidores donde se ejecutaran estos scripts, consulta la Lista de Verificacion de Seguridad de Servidores Linux y Hardening SSH: 12 Pasos para Asegurar tu Servidor Linux.