Python Admin

Guia Completo de Automação de Tarefas em Python: subprocess, shutil e Scripting Real Já leu

Entendendo Automação de Tarefas em Python Automação de tarefas é um dos pilares da programação moderna. Como profissional, posso garantir que 70% do meu trabalho envolve eliminar tarefas repetitivas através de scripts bem estruturados. Python se destaca nessa área porque possui bibliotecas robustas que permitem interagir diretamente com o sistema operacional, executar programas externos e manipular arquivos de forma elegante. Neste artigo, vamos explorar três ferramentas essenciais: o módulo para execução de comandos do sistema, para manipulação de arquivos, e técnicas práticas de scripting que você usará em projetos reais. O objetivo não é apenas mostrar sintaxe, mas desenvolver a mentalidade de um profissional que escreve automações confiáveis e mantíveis. Subprocess: Executando Comandos do Sistema Conceitos Fundamentais O módulo permite que seu código Python execute comandos do terminal/prompt como se você estivesse digitando manualmente. A diferença crucial em relação aos módulos antigos ( ) é que você tem controle total sobre entrada, saída e códigos de erro. Isso significa capturar

Entendendo Automação de Tarefas em Python

Automação de tarefas é um dos pilares da programação moderna. Como profissional, posso garantir que 70% do meu trabalho envolve eliminar tarefas repetitivas através de scripts bem estruturados. Python se destaca nessa área porque possui bibliotecas robustas que permitem interagir diretamente com o sistema operacional, executar programas externos e manipular arquivos de forma elegante.

Neste artigo, vamos explorar três ferramentas essenciais: o módulo subprocess para execução de comandos do sistema, shutil para manipulação de arquivos, e técnicas práticas de scripting que você usará em projetos reais. O objetivo não é apenas mostrar sintaxe, mas desenvolver a mentalidade de um profissional que escreve automações confiáveis e mantíveis.

Subprocess: Executando Comandos do Sistema

Conceitos Fundamentais

O módulo subprocess permite que seu código Python execute comandos do terminal/prompt como se você estivesse digitando manualmente. A diferença crucial em relação aos módulos antigos (os.system()) é que você tem controle total sobre entrada, saída e códigos de erro. Isso significa capturar resultados, passar argumentos com segurança e tratar exceções adequadamente.

Existem dois cenários principais: você quer apenas executar um comando (sem capturar saída) ou precisa do resultado para processar depois. Para a maioria dos casos modernos, você usará subprocess.run() ou subprocess.Popen().

Usando subprocess.run() para Tarefas Simples

O subprocess.run() é a escolha padrão para comandos de execução única. Ele aguarda o término do processo e retorna um objeto com informações sobre a execução.

import subprocess

# Exemplo 1: Executar um comando simples
resultado = subprocess.run(['ls', '-la'], capture_output=True, text=True)
print("Saída:", resultado.stdout)
print("Código de retorno:", resultado.returncode)

# Exemplo 2: Tratando erros com check=True
try:
    subprocess.run(['mkdir', '/tmp/meu_diretorio'], check=True)
    print("Diretório criado com sucesso")
except subprocess.CalledProcessError as e:
    print(f"Erro ao criar diretório: {e}")

# Exemplo 3: Passando entrada para o comando
resultado = subprocess.run(
    ['grep', 'erro'],
    input='linha com erro\nlinha normal\n',
    capture_output=True,
    text=True
)
print("Resultados encontrados:", resultado.stdout)

Note que capture_output=True redireciona stdout e stderr, enquanto text=True converte bytes para string (python 3.7+). O check=True levanta uma exceção se o comando retornar código não-zero, o que é essencial para detecção de erros em scripts de automação.

Popen para Processos Contínuos

Quando você precisa interagir continuamente com um processo, ou precisa de mais controle granular, use Popen. Este é um cenário menos comum, mas fundamental em automações avançadas.

import subprocess
import time

# Executar um processo em background e monitorar
processo = subprocess.Popen(
    ['ping', '-c', '4', 'google.com'],
    stdout=subprocess.PIPE,
    stderr=subprocess.PIPE,
    text=True
)

# Ler output em tempo real
for linha in processo.stdout:
    print(f"Ping: {linha.strip()}")

codigo_retorno = processo.wait()
print(f"Processo finalizado com código: {codigo_retorno}")

# Exemplo 2: Timeout - importante para não travar
try:
    resultado = subprocess.run(
        ['sleep', '10'],
        timeout=2  # Vai cancelar após 2 segundos
    )
except subprocess.TimeoutExpired:
    print("Processo excedeu o timeout")

A lição crucial aqui: sempre use timeout em automações críticas. Processos que travarem podem derrubar sua automação inteira. Em produção, isso causa downtime.

Manipulação de Arquivos com Shutil

Por que não usar os.rename() e os.remove()?

O módulo shutil foi criado para operações de alto nível com arquivos. Enquanto os.remove() remove apenas um arquivo e os.rename() funciona apenas em casos simples, shutil oferece funções que funcionam consistentemente entre Windows e Linux, tratam permissões e funcionam com diretórios inteiros. Em automação real, essa compatibilidade é ouro.

Operações Essenciais: Copy, Move e Remove

import shutil
import os

# Exemplo 1: Copiar arquivo mantendo metadados
origem = '/tmp/arquivo_original.txt'
destino = '/tmp/backup/arquivo_original.txt'

# copy2 preserva timestamps e permissões (melhor que copy)
shutil.copy2(origem, destino)
print("Arquivo copiado com sucesso")

# Exemplo 2: Copiar árvore de diretórios inteira
shutil.copytree(
    '/tmp/projeto_antigo',
    '/tmp/projeto_backup',
    dirs_exist_ok=True  # Não falha se diretório existe
)

# Exemplo 3: Mover arquivo (renomear com segurança)
shutil.move('/tmp/arquivo.txt', '/home/usuario/arquivo.txt')

# Exemplo 4: Remover árvore de diretórios (CUIDADO!)
# Esta operação é irreversível
if os.path.exists('/tmp/pasta_temporaria'):
    shutil.rmtree('/tmp/pasta_temporaria')
    print("Pasta removida")

# Exemplo 5: Obter tamanho de um diretório
tamanho = shutil.disk_usage('/home/usuario')
print(f"Total: {tamanho.total / (1024**3):.2f} GB")
print(f"Livre: {tamanho.free / (1024**3):.2f} GB")

A grande vantagem do shutil.copytree() com dirs_exist_ok=True é que você pode rodar backups incrementais sem se preocupar com tratamento de exceções. Em comparação com os.rename(), o shutil.move() funciona até entre sistemas de arquivos diferentes (por exemplo, de /tmp para /home em Linux), algo que os.rename() não garante.

Compactação de Arquivos

Muitas automações reais precisam comprimir arquivos para backup ou transferência. O shutil integra isso elegantemente.

import shutil

# Exemplo 1: Criar arquivo tar.gz
shutil.make_archive(
    base_name='/tmp/backup_projeto',  # Sem extensão
    format='gztar',  # Cria .tar.gz
    root_dir='/home/usuario/projeto'
)

# Exemplo 2: Extrair arquivo
shutil.unpack_archive(
    '/tmp/backup_projeto.tar.gz',
    extract_dir='/tmp/restaurado'
)

# Exemplo 3: Listar formatos disponíveis
print(shutil.get_archive_formats())

Scripting Real: Integrando Tudo

Estrutura Profissional de um Script

Um script de automação profissional segue padrões claros. Não é apenas uma sequência de comandos; é código que pode falhar, ser debugado e mantido por outros. Vamos construir um script real de backup com logging, validações e tratamento de erro adequado.

#!/usr/bin/env python3
"""
Script de Backup Automatizado
Realiza backup de diretórios, compacta e limpa antigos
"""

import subprocess
import shutil
import os
import logging
from datetime import datetime, timedelta

# Configuração de logging
logging.basicConfig(
    level=logging.INFO,
    format='%(asctime)s - %(levelname)s - %(message)s',
    handlers=[
        logging.FileHandler('/var/log/backup.log'),
        logging.StreamHandler()  # Também imprime no console
    ]
)

logger = logging.getLogger(__name__)

class BackupAutomacao:
    def __init__(self, origem, destino, retencao_dias=30):
        self.origem = origem
        self.destino = destino
        self.retencao_dias = retencao_dias

    def validar_origem(self):
        """Verifica se origem existe"""
        if not os.path.exists(self.origem):
            logger.error(f"Origem não existe: {self.origem}")
            raise FileNotFoundError(f"Origem inválida: {self.origem}")
        logger.info(f"Origem validada: {self.origem}")

    def criar_destino(self):
        """Cria diretório de destino se não existir"""
        os.makedirs(self.destino, exist_ok=True)
        logger.info(f"Diretório de destino pronto: {self.destino}")

    def executar_backup(self):
        """Realiza o backup compactado"""
        timestamp = datetime.now().strftime('%Y%m%d_%H%M%S')
        nome_arquivo = f"backup_{timestamp}"
        caminho_completo = os.path.join(self.destino, nome_arquivo)

        try:
            logger.info(f"Iniciando backup: {self.origem}")
            shutil.make_archive(
                base_name=caminho_completo,
                format='gztar',
                root_dir=self.origem
            )
            tamanho = os.path.getsize(f"{caminho_completo}.tar.gz") / (1024**2)
            logger.info(f"Backup concluído: {caminho_completo}.tar.gz ({tamanho:.2f} MB)")
            return f"{caminho_completo}.tar.gz"

        except Exception as e:
            logger.error(f"Erro ao executar backup: {e}")
            raise

    def limpar_antigos(self):
        """Remove backups mais antigos que retencao_dias"""
        limite = datetime.now() - timedelta(days=self.retencao_dias)

        for arquivo in os.listdir(self.destino):
            caminho = os.path.join(self.destino, arquivo)

            # Obter tempo de modificação
            tempo_mod = datetime.fromtimestamp(os.path.getmtime(caminho))

            if tempo_mod < limite and arquivo.endswith('.tar.gz'):
                try:
                    os.remove(caminho)
                    logger.info(f"Backup antigo removido: {arquivo}")
                except Exception as e:
                    logger.warning(f"Não foi possível remover {arquivo}: {e}")

    def executar(self):
        """Executa pipeline completo"""
        try:
            self.validar_origem()
            self.criar_destino()
            self.executar_backup()
            self.limpar_antigos()
            logger.info("Pipeline de backup concluído com sucesso")
            return True
        except Exception as e:
            logger.critical(f"Falha no pipeline: {e}")
            return False

# Uso do script
if __name__ == '__main__':
    backup = BackupAutomacao(
        origem='/home/usuario/dados_importante',
        destino='/mnt/backup/diario',
        retencao_dias=30
    )

    sucesso = backup.executar()
    exit(0 if sucesso else 1)

Este script demonstra padrões profissionais: logging estruturado para debugging, classes para organização, validações antes de operações críticas, tratamento de exceções específicas, e um retorno de código que pode ser usado em cron jobs ou pipelines CI/CD.

Script de Sincronização e Monitoramento

Outro cenário comum é sincronizar diretórios e monitorar mudanças. Vamos criar um script que usa subprocess para chamar rsync (ferramenta de sincronização do sistema) e valida o resultado.

#!/usr/bin/env python3
"""
Sincronizador com Validação
Usa rsync para sincronizar e valida integridade
"""

import subprocess
import hashlib
import os

def sincronizar_com_rsync(origem, destino, verbose=False):
    """Sincroniza com rsync preservando permissões e timestamps"""
    comando = [
        'rsync',
        '-avz',  # archive, verbose, compress
        '--delete',  # Remove arquivos no destino que não existem na origem
        origem,
        destino
    ]

    try:
        resultado = subprocess.run(
            comando,
            capture_output=True,
            text=True,
            check=True
        )

        if verbose:
            print("Output rsync:", resultado.stdout)

        return True, "Sincronização bem-sucedida"

    except subprocess.CalledProcessError as e:
        return False, f"Erro rsync: {e.stderr}"

def validar_integridade(arquivo_origem, arquivo_destino):
    """Compara hash SHA256 de dois arquivos"""
    def calcular_hash(caminho):
        hash_obj = hashlib.sha256()
        with open(caminho, 'rb') as f:
            for chunk in iter(lambda: f.read(4096), b''):
                hash_obj.update(chunk)
        return hash_obj.hexdigest()

    hash_origem = calcular_hash(arquivo_origem)
    hash_destino = calcular_hash(arquivo_destino)

    return hash_origem == hash_destino

def executar_sincronizacao(origem, destino):
    """Pipeline: sincronizar, validar e relatar"""

    # Sincronizar
    sucesso, mensagem = sincronizar_com_rsync(origem, destino, verbose=True)

    if not sucesso:
        print(f"FALHA: {mensagem}")
        return False

    print(f"SUCESSO: {mensagem}")

    # Validar alguns arquivos críticos
    arquivos_criticos = [f for f in os.listdir(destino) if f.endswith('.conf')]

    if arquivos_criticos:
        arquivo_teste = os.path.join(destino, arquivos_criticos[0])
        arquivo_origem_teste = os.path.join(origem, arquivos_criticos[0])

        if validar_integridade(arquivo_origem_teste, arquivo_teste):
            print("✓ Validação de integridade OK")
            return True
        else:
            print("✗ Falha na validação de integridade")
            return False

    return True

# Uso
if __name__ == '__main__':
    resultado = executar_sincronizacao(
        '/home/usuario/dados/',
        '/backup/sincronizado/'
    )
    exit(0 if resultado else 1)

Padrões Avançados e Boas Práticas

Integração com Cron e Systemd

Scripts de automação real rodam em background, geralmente por agendadores do sistema. A integração adequada faz a diferença entre um script que funciona e um que falha silenciosamente.

#!/usr/bin/env python3
"""
Script preparado para rodar via cron ou systemd
"""

import sys
import logging
from pathlib import Path

# Garantir que logging funciona mesmo com cron
log_dir = Path('/var/log/meus_scripts')
log_dir.mkdir(exist_ok=True)

logging.basicConfig(
    level=logging.INFO,
    format='%(asctime)s - %(name)s - %(levelname)s - %(message)s',
    filename=str(log_dir / 'automacao.log')
)

logger = logging.getLogger(__name__)

try:
    # Seu código de automação aqui
    logger.info("Script iniciado")

    # Se tudo funcionar
    logger.info("Script finalizado com sucesso")
    sys.exit(0)

except Exception as e:
    logger.exception(f"Erro crítico: {e}")
    sys.exit(1)

Para rodar via cron, adicione à crontab:

0 2 * * * /usr/bin/python3 /home/usuario/automacao.py

Para systemd, crie /etc/systemd/system/automacao.service:

[Unit]
Description=Automação de Tarefas
After=network.target

[Service]
ExecStart=/usr/bin/python3 /home/usuario/automacao.py
User=usuario
StandardOutput=journal
StandardError=journal

[Install]
WantedBy=multi-user.target

Tratamento de Dependências Externas

Nem sempre rsync, git ou outras ferramentas estão instaladas. Um profissional verifica isso antes.

import subprocess
import shutil

def verificar_comando_disponivel(comando):
    """Verifica se um comando existe no PATH"""
    return shutil.which(comando) is not None

def instalar_se_necessario(comando, package_manager='apt'):
    """Tenta instalar dependência se não existir"""

    if verificar_comando_disponivel(comando):
        return True

    logger.warning(f"Comando {comando} não encontrado, tentando instalar")

    try:
        subprocess.run(
            [package_manager, 'install', '-y', comando],
            check=True,
            capture_output=True
        )
        logger.info(f"{comando} instalado com sucesso")
        return True
    except subprocess.CalledProcessError:
        logger.error(f"Falha ao instalar {comando}")
        return False

# No início do seu script
if not verificar_comando_disponivel('rsync'):
    if not instalar_se_necessario('rsync'):
        raise RuntimeError("rsync é necessário para este script")

Conclusão

Você aprendeu três lições fundamentais que separam profissionais de iniciantes: Primeiro, o subprocess não é apenas para rodar comandos — é sobre capturar resultados, tratar erros com check=True e timeout, e entender que falhas silenciosas são piores que exceções. Segundo, o shutil resolve problemas reais de forma multiplataforma, desde cópias com metadados até compactação, coisas que os simples não garante funcionarem em todos os sistemas. Terceiro, scripting real é sobre logging estruturado, validações antes de operações irreversíveis, e code que outras pessoas conseguem manter seis meses depois.

A mensagem final: automação não é conveniência, é confiabilidade. Um script que falha silenciosamente é pior que nenhum script. Use logging, retorne códigos de saída, valide entradas, e sempre tenha uma estratégia de rollback para operações críticas.

Referências


Artigos relacionados