Todo administrador de sistemas chega a um ponto em que os scripts Bash se tornam incontrolaveis. Um alerta de uso de disco que comecou como um script de cinco linhas agora abrange trezentas linhas, lida mal com casos extremos e falha silenciosamente em uma distribuicao diferente. Python preenche a lacuna entre os comandos rapidos de shell e as ferramentas completas de gerenciamento de configuracao como Ansible. Ele oferece tratamento adequado de erros, estruturas de dados ricas, milhares de bibliotecas e codigo que e legivel meses depois de ter sido escrito.

Este artigo percorre as habilidades essenciais de Python que um administrador de sistemas Linux precisa: operacoes com arquivos, gerenciamento de subprocessos, automacao SSH, integracao com APIs REST, analise de logs, monitoramento de sistemas e notificacoes. Cada secao inclui codigo pronto para producao que voce pode adaptar ao seu ambiente.


Pre-requisitos

Antes de comecar, certifique-se de ter o seguinte:

  • Python 3.10+ — verifique com python3 --version
  • pip — o instalador de pacotes Python (geralmente incluido com Python)
  • venv — o modulo integrado de ambientes virtuais
  • Um servidor Linux (Ubuntu 22.04/24.04, Debian 12 ou RHEL 9 recomendados)
  • Familiaridade basica com a linha de comando Linux e Bash
  • Um par de chaves SSH configurado para acesso a servidores remotos (veja Hardening SSH)
# Verificar a instalacao do 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 um Ambiente Python

Nunca instale pacotes Python globalmente em um servidor de producao. Use ambientes virtuais para isolar as dependencias de cada projeto.

Criando um ambiente virtual

# Criar um diretorio de projeto
mkdir -p ~/sysadmin-scripts
cd ~/sysadmin-scripts

# Criar um ambiente virtual
python3 -m venv venv

# Ativa-lo
source venv/bin/activate

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

Instalando bibliotecas essenciais

# Instalar todas as bibliotecas que usaremos neste artigo
pip install paramiko requests psutil python-dotenv jinja2

# Congelar dependencias para reprodutibilidade
pip freeze > requirements.txt

Estrutura do projeto

Um projeto de scripts de administracao de sistemas bem organizado se parece com isto:

sysadmin-scripts/
├── venv/
├── requirements.txt
├── .env                  # Segredos (chaves de API, senhas)
├── config.yaml           # Listas de servidores, limites
├── scripts/
│   ├── disk_monitor.py
│   ├── log_parser.py
│   ├── remote_exec.py
│   └── backup_rotation.py
└── logs/
    └── automation.log

Operacoes com Arquivos e Diretorios

O modulo pathlib do Python (parte da biblioteca padrao desde Python 3.4) fornece uma interface orientada a objetos para caminhos do sistema de arquivos. Combinado com shutil, ele lida com praticamente qualquer operacao de arquivo que um administrador de sistemas precise.

Trabalhando com pathlib

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

# Criar caminhos (funciona em qualquer SO)
log_dir = Path("/var/log/myapp")
backup_dir = Path("/backup/logs")

# Criar diretorios (parents=True funciona como mkdir -p)
backup_dir.mkdir(parents=True, exist_ok=True)

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

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

Script de rotacao automatica de logs

#!/usr/bin/env python3
"""Rotacionar e comprimir arquivos de log mais antigos 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 o arquivo
        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"Arquivado: {log_file.name} -> {gz_path.name}")

Script de backup com diretorios com marca temporal

#!/usr/bin/env python3
"""Criar backups com marca temporal e remover backups mais antigos que 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

# Criar backup com marca temporal
timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
backup_path = BACKUP_ROOT / timestamp

shutil.copytree(SOURCE, backup_path)
print(f"Backup criado: {backup_path}")

# Limpar backups antigos
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"Backup antigo removido: {old_backup.name}")

Executando Comandos do Sistema com subprocess

O modulo subprocess permite executar comandos de shell a partir do Python com controle total sobre entrada, saida e tratamento de erros. Sempre prefira subprocess.run() sobre o mais antigo os.system().

Execucao basica de comandos

import subprocess

# Executar um comando e capturar a saida
result = subprocess.run(
    ["df", "-h", "/"],
    capture_output=True,
    text=True,
    timeout=30
)

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

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

Executando multiplos comandos com seguranca

import subprocess
import sys

def run_cmd(cmd, description=""):
    """Executar um comando e tratar erros."""
    try:
        result = subprocess.run(
            cmd,
            capture_output=True,
            text=True,
            timeout=60,
            check=True  # Lanca CalledProcessError em codigo de saida nao zero
        )
        print(f"[OK] {description}")
        return result.stdout
    except subprocess.CalledProcessError as e:
        print(f"[FALHA] {description}: {e.stderr.strip()}")
        sys.exit(1)
    except subprocess.TimeoutExpired:
        print(f"[TIMEOUT] {description}: comando excedeu 60s")
        sys.exit(1)

# Exemplo: sequencia de atualizacao do sistema
run_cmd(["sudo", "apt", "update"], "Atualizar listas de pacotes")
run_cmd(["sudo", "apt", "upgrade", "-y"], "Atualizar pacotes")
run_cmd(["sudo", "apt", "autoremove", "-y"], "Remover pacotes nao utilizados")

Analisando a saida de comandos

import subprocess

def get_listening_ports():
    """Retornar uma lista de portas com servicos escutando."""
    result = subprocess.run(
        ["ss", "-tlnp"],
        capture_output=True,
        text=True,
        check=True
    )
    ports = []
    for line in result.stdout.strip().split("\n")[1:]:  # Pular cabecalho
        parts = line.split()
        if len(parts) >= 4:
            addr = parts[3]
            port = addr.rsplit(":", 1)[-1]
            ports.append(port)
    return sorted(set(ports))

print("Portas escutando:", get_listening_ports())

Automacao SSH com Paramiko

Paramiko e a biblioteca Python padrao para conexoes SSH2. Permite executar comandos em servidores remotos, transferir arquivos via SFTP e gerenciar chaves SSH — tudo a partir do Python.

Instalando Paramiko

pip install paramiko

Executando comandos remotos

import paramiko

def ssh_exec(hostname, username, command, key_path="~/.ssh/id_ed25519"):
    """Executar um comando em um 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()

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

Executando comandos em multiplos 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 o tempo de atividade em um unico 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"ERRO: {e}"
    finally:
        client.close()

# Executar em paralelo com 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 arquivos via SFTP

import paramiko
from pathlib import Path

def upload_file(hostname, username, local_path, remote_path):
    """Enviar um arquivo para um 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"Enviado {local_path} -> {hostname}:{remote_path}")
    finally:
        sftp.close()
        transport.close()

# Enviar um arquivo de configuracao
upload_file(
    "web01.example.com",
    "deploy",
    Path("./configs/nginx.conf"),
    "/tmp/nginx.conf"
)

Trabalhando com APIs REST

A biblioteca requests torna as chamadas HTTP simples. Administradores de sistemas usam APIs para interagir com plataformas de monitoramento, provedores de nuvem, servicos DNS e ferramentas internas.

Requisicoes GET — obtendo dados

import requests

# Verificar o status do servidor a partir de uma API de monitoramento
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"Erro da API: {response.status_code} {response.text}")

Requisicoes POST — enviando dados

import requests
import json

# Criar um 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 criado: {result['result']['id']}")
else:
    print(f"Erro: {result.get('errors')}")

Cliente API robusto com tentativas

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

def create_api_session(base_url, api_key, retries=3):
    """Criar uma sessao de requests com logica de tentativas."""
    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())

Analise e Processamento de Logs

Administradores de sistemas dedicam uma quantidade significativa de tempo a analise de arquivos de log. Python facilita a analise, filtragem e agregacao de dados de log.

Analisando syslog com expressoes regulares

#!/usr/bin/env python3
"""Analisar syslog e reportar tentativas falhadas de login 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} tentativas")

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

Analisando logs estruturados (JSON)

#!/usr/bin/env python3
"""Analisar logs de aplicacao em 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("Distribuicao de codigos de status:")
for code, count in sorted(status_codes.items()):
    print(f"  {code}: {count}")

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

Acompanhamento de logs em tempo real

#!/usr/bin/env python3
"""Acompanhar um arquivo de log e alertar sobre padroes de erro."""

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):
    """Produzir novas linhas conforme sao adicionadas a um arquivo."""
    with open(filepath) as f:
        f.seek(0, 2)  # Ir para o final do arquivo
        while True:
            line = f.readline()
            if line:
                yield line.strip()
            else:
                time.sleep(interval)

print(f"Acompanhando {LOG_FILE}... (Ctrl+C para parar)")
for line in tail_file(LOG_FILE):
    for pattern in ALERT_PATTERNS:
        if pattern in line:
            print(f"[ALERTA] {line}")
            # Aqui voce poderia enviar uma notificacao
            break

Scripts de Monitoramento do Sistema

A biblioteca psutil fornece acesso multiplataforma a metricas do sistema — CPU, memoria, disco, rede e informacoes de processos. E a base para construir ferramentas de monitoramento personalizadas.

Instalando psutil

pip install psutil

Verificacao completa da saude do sistema

#!/usr/bin/env python3
"""Script completo de verificacao da saude do sistema."""

import psutil
import socket
from datetime import datetime

def bytes_to_human(n):
    """Converter bytes para formato legivel por 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():
    """Executar todas as verificacoes do sistema e 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

    # Rede
    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"=== Relatorio de Saude do Sistema: {hostname} ===")
print(f"Marca temporal: {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 = "AVISO" if disk["used_percent"] > 85 else "OK"
    print(f"Disco {disk['mount']}: {disk['used_percent']}% "
          f"de {disk['total']} [{status}]")

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

Alertas de uso de disco com limites

#!/usr/bin/env python3
"""Monitorar o uso de disco e alertar quando os limites forem excedidos."""

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"(limite: {alert['threshold']}%, "
              f"livre: {alert['free_gb']:.1f} GB)")
else:
    print("Todas as particoes de disco dentro dos limites normais.")

Monitoramento de processos

#!/usr/bin/env python3
"""Monitorar processos especificos e alertar se pararem de executar."""

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"  [EXECUTANDO] {proc_name}")
    else:
        print(f"  [PARADO] {proc_name} -- ACAO NECESSARIA")

Enviando Notificacoes

A automacao so e util se voce souber quando algo da errado. Python pode enviar alertas por email, Slack ou qualquer servico baseado em webhooks.

Enviando notificacoes por email

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 uma notificacao por email."""
    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)

# Exemplo de uso
send_email(
    subject="[ALERTA] Uso de disco acima de 90% no web01",
    body="A particao /var esta em 92% de uso. Espaco livre: 3.2 GB.",
    to_addr="admin@example.com",
    from_addr="alertas@example.com",
    smtp_server="smtp.example.com",
    username="alertas@example.com",
    password="senha-de-aplicacao"
)

Notificacoes por webhook do Slack

import requests
import json

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

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

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

# Exemplo de uso
send_slack_alert(
    webhook_url="https://hooks.slack.com/services/T00/B00/XXXX",
    message="O uso de disco em /var excedeu 90% no web01.example.com",
    severity="critical"
)

Combinando monitoramento com notificacoes

#!/usr/bin/env python3
"""Monitorar recursos do sistema e enviar alertas quando limites forem excedidos."""

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 para o Slack."""
    payload = {
        "text": f"*[{severity.upper()}]* `{HOSTNAME}` - {message}\n"
                f"_Marca temporal: {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 em {cpu}% (limite: {CPU_THRESHOLD}%)", "critical")

# Verificar memoria
mem = psutil.virtual_memory()
if mem.percent > MEMORY_THRESHOLD:
    alert(f"Uso de memoria em {mem.percent}% (limite: {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} em {usage.percent}% "
                  f"(limite: {DISK_THRESHOLD}%)", "warning")
    except PermissionError:
        continue

Agendando Scripts Python

Usando cron

A forma mais direta de agendar scripts Python no Linux e com cron.

# Editar seu crontab
crontab -e

# Executar monitor de disco a 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

# Executar verificacao completa de saude a 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

# Executar analise de logs diariamente as 6h
0 6 * * * /home/deploy/sysadmin-scripts/venv/bin/python /home/deploy/sysadmin-scripts/scripts/log_parser.py >> /var/log/log_analysis.log 2>&1

Sempre use o caminho completo para o binario Python dentro do seu ambiente virtual. Isso garante que o script use a versao correta do Python e tenha acesso aos pacotes instalados.

Usando timers do systemd

Para agendamento mais avancado com integracao de registro de logs e gerenciamento de dependencias, use timers do 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 o timer
sudo systemctl daemon-reload
sudo systemctl enable health-check.timer
sudo systemctl start health-check.timer

# Verificar o status do timer
systemctl list-timers --all | grep health

Tabela de Referencia de Bibliotecas Uteis

BibliotecaPropositoInstalacaoDocumentacao
pathlibOperacoes de caminhos de arquivos/diretoriosIntegradadocs.python.org/3/library/pathlib
subprocessExecutar comandos do sistemaIntegradadocs.python.org/3/library/subprocess
shutilOperacoes de arquivos de alto nivel (copiar, mover, arquivar)Integradadocs.python.org/3/library/shutil
osInterface de baixo nivel com o SO, variaveis de ambienteIntegradadocs.python.org/3/library/os
paramikoProtocolo SSH2, execucao remota de comandos, SFTPpip install paramikoparamiko.org
requestsCliente HTTP para APIs RESTpip install requestsrequests.readthedocs.io
psutilMonitoramento do sistema (CPU, memoria, disco, rede)pip install psutilpsutil.readthedocs.io
python-dotenvCarregar variaveis de ambiente de arquivos .envpip install python-dotenvpypi.org/project/python-dotenv
jinja2Motor de templates para geracao de arquivos de configuracaopip install jinja2jinja.palletsprojects.com
pyyamlAnalisar e escrever arquivos de configuracao YAMLpip install pyyamlpyyaml.org
fabricAutomacao SSH de alto nivel (construido sobre Paramiko)pip install fabricfabfile.org
scheduleAgendamento de tarefas amigavel para humanospip install scheduleschedule.readthedocs.io
loggingRegistro estruturado de aplicacoesIntegradadocs.python.org/3/library/logging

Solucao de Problemas

ModuleNotFoundError ao executar via cron

O cron usa um ambiente minimo. Se seu script falhar com ModuleNotFoundError, voce provavelmente nao esta usando o caminho completo para o binario Python do ambiente virtual.

# Errado -- usa o Python do sistema, que nao tem seus pacotes
* * * * * python3 /home/deploy/scripts/monitor.py

# Correto -- usa o Python do venv
* * * * * /home/deploy/sysadmin-scripts/venv/bin/python /home/deploy/scripts/monitor.py

Conexao recusada no Paramiko

Se o Paramiko lancar ConnectionRefusedError, verifique se o SSH esta executando no destino e se a porta esta correta.

# Especificar uma porta personalizada se seu servidor usa uma porta SSH nao padrao
client.connect(hostname="web01.example.com", port=2222, username="deploy",
               key_filename="/home/deploy/.ssh/id_ed25519")

Permissao negada em operacoes de arquivos

Quando scripts executam como um usuario nao root, podem nao ter acesso a arquivos em /var/log ou /etc. Execute o script com sudo ou ajuste as permissoes de arquivos.

# Conceder acesso de leitura ao auth.log para o usuario deploy
sudo usermod -aG adm deploy

# Ou use ACLs para controle detalhado
sudo setfacl -m u:deploy:r /var/log/auth.log

psutil retorna 0.0% de CPU na primeira chamada

psutil.cpu_percent() retorna 0.0 na primeira chamada porque precisa de duas medicoes para calcular o uso. Sempre passe um intervalo ou chame-o duas vezes.

# Correto: passar intervalo para uma medicao bloqueante
cpu = psutil.cpu_percent(interval=1)

# Ou chama-lo duas vezes com um atraso
psutil.cpu_percent()
import time; time.sleep(1)
cpu = psutil.cpu_percent()

Erros de timeout ou SSL no requests

Ao conectar-se a APIs atras de proxies corporativos ou com certificados autoassinados, voce pode encontrar erros de SSL.

# Desativar verificacao SSL (somente desenvolvimento -- nunca em producao)
response = requests.get("https://internal-api.local", verify=False)

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

Resumo

Python e o canivete suico da administracao de sistemas. Neste guia voce aprendeu como:

  • Configurar ambientes Python isolados com venv para projetos de administracao de sistemas
  • Gerenciar arquivos e diretorios programaticamente com pathlib e shutil
  • Executar e analisar comandos do sistema com subprocess
  • Automatizar conexoes SSH e transferencias de arquivos com paramiko
  • Interagir com APIs REST usando requests com logica de tentativas adequada
  • Analisar e processar arquivos de log com expressoes regulares e processamento JSON
  • Construir scripts de monitoramento usando psutil para metricas de CPU, memoria, disco e rede
  • Enviar notificacoes por email e webhooks do Slack
  • Agendar scripts de forma confiavel com cron e timers do systemd

Comece pequeno: escolha uma tarefa repetitiva que voce realiza semanalmente (rotacao de logs, verificacoes de saude, verificacao de backups) e automatize-a com Python. Conforme sua confianca cresce, encadeie esses scripts em um kit de ferramentas de automacao completo.

Para guias relacionados sobre como proteger os servidores onde esses scripts serao executados, consulte a Lista de Verificacao de Seguranca de Servidores Linux e Hardening SSH: 12 Passos para Proteger seu Servidor Linux.