Scripting y Automatización del Sistema con Python

Scripting y Automatización del Sistema con Python

Python es el lenguaje de scripting por excelencia en sistemas Unix/Linux y Windows. Sus módulos de biblioteca estándar cubren desde ejecutar comandos del sistema hasta construir CLIs completas y monitorizar el sistema de archivos.

subprocess — ejecutar comandos externos

import subprocess

# Forma recomendada: subprocess.run (bloquea hasta que termina)
resultado = subprocess.run(
    ['ls', '-la', '/tmp'],   # lista — nunca shell=True con argumentos externos
    capture_output=True,     # captura stdout y stderr
    text=True,               # decodifica bytes a str
    timeout=10,              # máximo 10 segundos
    check=True,              # lanza CalledProcessError si returncode != 0
)
print(resultado.stdout)
print("Código de salida:", resultado.returncode)

# Capturar stderr por separado
resultado2 = subprocess.run(
    ['python3', '-c', 'import sys; sys.stderr.write("error!\\n")'],
    capture_output=True,
    text=True,
)
print("stderr:", resultado2.stderr)

# Pasar datos por stdin
resultado3 = subprocess.run(
    ['sort'],
    input='banana\nmanzana\napple\n',
    capture_output=True,
    text=True,
)
print(resultado3.stdout)

Procesos de larga duración con Popen

import subprocess
import sys

def ejecutar_con_streaming(comando: list[str]) -> int:
    """Ejecuta un comando y muestra su salida línea a línea en tiempo real."""
    with subprocess.Popen(
        comando,
        stdout=subprocess.PIPE,
        stderr=subprocess.STDOUT,   # mezclar stderr con stdout
        text=True,
        bufsize=1,                  # line-buffered
    ) as proc:
        for linea in proc.stdout:
            print(linea, end='', flush=True)
    return proc.returncode

codigo = ejecutar_con_streaming(['ping', '-c', '4', '8.8.8.8'])
print(f"Salió con código: {codigo}")

Pipelines — encadenar comandos

# Equivalente a: ps aux | grep python | wc -l
ps = subprocess.Popen(['ps', 'aux'], stdout=subprocess.PIPE)
grep = subprocess.Popen(
    ['grep', 'python'],
    stdin=ps.stdout,
    stdout=subprocess.PIPE,
)
ps.stdout.close()  # permite que ps reciba SIGPIPE si grep termina antes
wc = subprocess.run(['wc', '-l'], stdin=grep.stdout, capture_output=True, text=True)
grep.stdout.close()
ps.wait()
grep.wait()
print(f"Procesos Python activos: {wc.stdout.strip()}")

os y pathlib — sistema de archivos

import os
import pathlib

# pathlib es la API moderna (recomendada sobre os.path)
p = pathlib.Path('/home/usuario/proyecto')

# Información
p.exists()          # ¿existe?
p.is_dir()          # ¿es directorio?
p.is_file()         # ¿es archivo?
p.stat().st_size    # tamaño en bytes
p.stat().st_mtime   # timestamp de modificación

# Navegar
p.parent            # /home/usuario
p.name              # 'proyecto'
p.stem              # 'proyecto' (sin extensión)
p.suffix            # '' (sin extensión) o '.txt'

# Construir rutas (portable en todos los SO)
config = p / 'config' / 'settings.json'

# Crear directorios
(p / 'logs').mkdir(parents=True, exist_ok=True)

# Listar archivos
for archivo in p.glob('**/*.py'):   # recursivo
    print(archivo)

# Solo archivos Python modificados hoy
import time
hoy = time.time() - 86400
recientes = [f for f in p.glob('**/*.py') if f.stat().st_mtime > hoy]

shutil — operaciones de alto nivel

import shutil
import pathlib

src  = pathlib.Path('carpeta_origen')
dest = pathlib.Path('carpeta_destino')

# Copiar árbol de directorios completo
shutil.copytree(src, dest, dirs_exist_ok=True)

# Copiar un archivo
shutil.copy2(src / 'archivo.txt', dest / 'archivo.txt')  # preserva metadatos

# Mover archivo o directorio
shutil.move(str(src / 'viejo.txt'), str(dest / 'nuevo.txt'))

# Eliminar árbol de directorios
shutil.rmtree(dest, ignore_errors=True)

# Crear archivo ZIP
shutil.make_archive('backup_2024', 'zip', root_dir='proyecto/', base_dir='.')

# Descomprimir
shutil.unpack_archive('backup_2024.zip', extract_dir='restaurado/')

# Espacio en disco
total, usado, libre = shutil.disk_usage('/')
print(f"Libre: {libre / 1024**3:.1f} GB")

# Encontrar ejecutable en PATH
python_path = shutil.which('python3')
print(f"Python en: {python_path}")

argparse — crear CLIs robustas

import argparse
import sys
import pathlib

def crear_parser() -> argparse.ArgumentParser:
    parser = argparse.ArgumentParser(
        prog='conversor',
        description='Convierte archivos entre formatos.',
        formatter_class=argparse.RawDescriptionHelpFormatter,
        epilog='Ejemplos:\n  conversor imagen.jpg output.png\n  conversor -v -q 85 foto.jpg out.webp',
    )

    # Argumentos posicionales
    parser.add_argument('entrada',  type=pathlib.Path, help='Archivo de entrada')
    parser.add_argument('salida',   type=pathlib.Path, help='Archivo de salida')

    # Opciones
    parser.add_argument('-q', '--calidad', type=int, default=85,
                        choices=range(1, 101), metavar='1-100',
                        help='Calidad de compresión (default: 85)')
    parser.add_argument('-v', '--verbose', action='store_true',
                        help='Activar salida detallada')
    parser.add_argument('--formato', choices=['jpg', 'png', 'webp', 'avif'],
                        help='Forzar formato de salida')
    parser.add_argument('--resize', nargs=2, type=int, metavar=('ANCHO', 'ALTO'),
                        help='Redimensionar a ANCHO×ALTO píxeles')

    return parser

def main():
    parser = crear_parser()
    args = parser.parse_args()

    if not args.entrada.exists():
        parser.error(f"El archivo '{args.entrada}' no existe")

    if args.verbose:
        print(f"Convirtiendo {args.entrada} → {args.salida}")
        print(f"Calidad: {args.calidad}")
        if args.resize:
            print(f"Tamaño: {args.resize[0]}×{args.resize[1]}")

    # Lógica de conversión...
    print(f"Conversión completada: {args.salida}")

if __name__ == '__main__':
    main()

Subcomandos (git-style CLI)

import argparse

parser = argparse.ArgumentParser(prog='mi_herramienta')
subparsers = parser.add_subparsers(dest='comando', required=True)

# Subcomando 'convertir'
conv = subparsers.add_parser('convertir', help='Convertir archivos')
conv.add_argument('archivo')
conv.add_argument('--formato', required=True)

# Subcomando 'listar'
lista = subparsers.add_parser('listar', help='Listar conversiones pendientes')
lista.add_argument('--estado', choices=['todos', 'pendientes', 'completados'])

args = parser.parse_args()
print(f"Comando: {args.comando}")
if args.comando == 'convertir':
    print(f"Convertir {args.archivo} a {args.formato}")

Automatización de tareas con schedule

import schedule
import time

def limpiar_temporales():
    import pathlib, shutil
    tmp = pathlib.Path('/tmp/mi_app')
    if tmp.exists():
        shutil.rmtree(tmp)
        tmp.mkdir()
        print("Temporales limpiados")

def generar_reporte():
    print("Generando reporte diario...")

# Definir tareas
schedule.every(10).minutes.do(limpiar_temporales)
schedule.every().day.at("08:00").do(generar_reporte)
schedule.every().monday.at("09:00").do(generar_reporte)

# Bucle principal
while True:
    schedule.run_pending()
    time.sleep(1)

watchdog — monitorizar cambios en el sistema de archivos

from watchdog.observers import Observer    # pip install watchdog
from watchdog.events import FileSystemEventHandler
import time
import pathlib

class ManejadorConversiones(FileSystemEventHandler):
    def on_created(self, event):
        if event.is_directory:
            return
        ruta = pathlib.Path(event.src_path)
        if ruta.suffix in {'.jpg', '.png', '.tiff', '.bmp'}:
            print(f"Nueva imagen detectada: {ruta.name}")
            self.procesar_imagen(ruta)

    def on_modified(self, event):
        ruta = pathlib.Path(event.src_path)
        print(f"Archivo modificado: {ruta}")

    def procesar_imagen(self, ruta: pathlib.Path):
        # Aquí iría la lógica de conversión
        print(f"  → Procesando {ruta.name}...")

def monitorizar_carpeta(ruta: str, recursivo: bool = False):
    manejador = ManejadorConversiones()
    observador = Observer()
    observador.schedule(manejador, ruta, recursive=recursivo)
    observador.start()
    print(f"Monitorizando {ruta}... (Ctrl+C para salir)")
    try:
        while True:
            time.sleep(1)
    except KeyboardInterrupt:
        observador.stop()
    observador.join()

# monitorizar_carpeta('/ruta/a/entrada', recursivo=True)

Script de backup automatizado completo

#!/usr/bin/env python3
"""
backup.py — Script de backup incremental con rotación y logging.
Uso: python backup.py --origen /datos --destino /backups --dias 7
"""
import argparse
import logging
import pathlib
import shutil
import time
from datetime import datetime

logging.basicConfig(
    level=logging.INFO,
    format='%(asctime)s [%(levelname)s] %(message)s',
    handlers=[
        logging.FileHandler('backup.log'),
        logging.StreamHandler(),
    ]
)
log = logging.getLogger('backup')

def hacer_backup(origen: pathlib.Path, destino: pathlib.Path) -> pathlib.Path:
    timestamp = datetime.now().strftime('%Y%m%d_%H%M%S')
    nombre_backup = destino / f"backup_{timestamp}"
    log.info(f"Iniciando backup: {origen} → {nombre_backup}")
    t0 = time.perf_counter()
    shutil.copytree(origen, nombre_backup, dirs_exist_ok=False)
    elapsed = time.perf_counter() - t0
    log.info(f"Backup completado en {elapsed:.1f}s")
    return nombre_backup

def rotar_backups(destino: pathlib.Path, mantener_dias: int):
    limite = time.time() - mantener_dias * 86400
    eliminados = 0
    for d in sorted(destino.glob('backup_*')):
        if d.is_dir() and d.stat().st_mtime < limite:
            shutil.rmtree(d)
            log.info(f"Backup antiguo eliminado: {d.name}")
            eliminados += 1
    if eliminados:
        log.info(f"Rotación: {eliminados} backup(s) eliminado(s)")

def main():
    parser = argparse.ArgumentParser(description='Script de backup con rotación')
    parser.add_argument('--origen',  type=pathlib.Path, required=True)
    parser.add_argument('--destino', type=pathlib.Path, required=True)
    parser.add_argument('--dias',    type=int, default=7, help='Días de retención')
    args = parser.parse_args()

    args.destino.mkdir(parents=True, exist_ok=True)

    try:
        ruta_backup = hacer_backup(args.origen, args.destino)
        rotar_backups(args.destino, args.dias)
        log.info(f"Todo correcto. Último backup: {ruta_backup}")
    except Exception:
        log.exception("Error durante el backup")
        raise SystemExit(1)

if __name__ == '__main__':
    main()

Buenas prácticas

• Usar listas en subprocess.run(), nunca strings con shell=True — evita inyección de comandos si los argumentos vienen del exterior.
• pathlib.Path sobre os.path — más legible, orientada a objetos y portable entre SO.
• check=True en subprocess.run() para que los errores de comandos no pasen desapercibidos.
• timeout= siempre en llamadas a procesos externos — previene que el script se quede colgado indefinidamente.
• shutil.which('ffmpeg') al inicio del script para comprobar que las dependencias externas están instaladas antes de necesitarlas.
• if __name__ == '__main__': en todos los scripts para poder importarlos en tests sin que se ejecute la lógica principal.

Conversiones relacionadas

Conversiones frecuentes del catálogo:

• JPG a PNG
• MP4 a MP3
• PDF a DOCX
• PNG a JPG
• MP3 a WAV
• DOCX a PDF