Nesta aula, vamos colocar em prática os conceitos aprendidos ao longo do curso para construir um projeto real de automação de tarefas. O objetivo é criar um script Python que automatize o processamento de arquivos de texto, execute tarefas em horários agendados, registre logs detalhados e seja robusto o suficiente para lidar com erros inesperados. Você aprenderá a combinar manipulação de arquivos, agendamento, logging e tratamento de exceções em um único projeto coeso.

A automação de tarefas é uma das aplicações mais úteis do Python no dia a dia. Seja para organizar arquivos, gerar relatórios, enviar e-mails ou fazer backup, dominar essas técnicas aumenta sua produtividade e a confiabilidade dos processos. Ao final desta aula, você terá um script funcional que pode ser adaptado para suas próprias necessidades.

Manipulação de Arquivos

A manipulação de arquivos é a base de muitas tarefas de automação. Em Python, podemos ler, escrever, renomear, mover e excluir arquivos usando funções nativas e o módulo os. No nosso projeto, vamos processar arquivos de texto de um diretório de entrada, extrair informações e gerar arquivos de saída.

Para garantir a robustez, é importante verificar a existência dos arquivos, usar caminhos absolutos e trabalhar com context managers (with) para gerenciar a abertura e fechamento automático. Vamos usar o módulo pathlib, que oferece uma API orientada a objetos para manipulação de caminhos, mais moderna e segura que os.path.

from pathlib import Path

def process_file(input_path: Path, output_dir: Path) -> None:
    """Processa um arquivo de texto e salva um resumo no diretório de saída."""
    if not input_path.exists():
        raise FileNotFoundError(f"Arquivo não encontrado: {input_path}")

    # Lê o conteúdo do arquivo
    with input_path.open('r', encoding='utf-8') as f:
        content = f.read()

    # Processa: conta palavras e linhas
    word_count = len(content.split())
    line_count = content.count('\n') + 1

    # Cria o arquivo de saída
    output_path = output_dir / f"{input_path.stem}_resumo.txt"
    with output_path.open('w', encoding='utf-8') as f:
        f.write(f"Arquivo: {input_path.name}\n")
        f.write(f"Palavras: {word_count}\n")
        f.write(f"Linhas: {line_count}\n")

    print(f"Processado: {input_path.name} - {word_count} palavras, {line_count} linhas")

Nesse exemplo, usamos pathlib.Path para representar caminhos e métodos como open(), exists() e name. A função process_file lê um arquivo, conta palavras e linhas e cria um resumo. Essa função será o núcleo do nosso projeto.

Agendamento (visão geral)

Agendar tarefas significa executar o script automaticamente em horários específicos. Em Python, temos a biblioteca schedule, que é simples e adequada para scripts que rodam continuamente. Para tarefas mais complexas, podemos usar o APScheduler ou o agendador do sistema operacional (cron no Linux, Task Scheduler no Windows). Nesta aula, usaremos schedule para executar a função de processamento a cada intervalo definido.

O agendamento é essencial para automações que precisam rodar sem intervenção humana, como backups noturnos ou relatórios diários. A biblioteca schedule permite agendar funções com sintaxe legível e é ideal para scripts pequenos e médios. Vamos agendar a verificação de novos arquivos em um diretório a cada 10 segundos para demonstrar.

import schedule
import time

def job():
    print("Executando tarefa agendada...")
    # Aqui chamamos a função de processamento

# Agendar a cada 10 segundos
schedule.every(10).seconds.do(job)

while True:
    schedule.run_pending()
    time.sleep(1)

No código acima, schedule.every(10).seconds.do(job) agenda a função job para rodar a cada 10 segundos. O loop infinito mantém o script em execução, verificando a cada segundo se há tarefas pendentes. Em um projeto real, você pode substituir job pela função que processa os arquivos. Lembre-se de que essa biblioteca não é persistente entre reinicializações; para agendamentos robustos, prefira o cron do sistema.

Logging

Logging é o registro de eventos que ocorrem durante a execução do programa. É fundamental para depuração e monitoramento de automações. O módulo logging da biblioteca padrão oferece uma estrutura flexível: podemos configurar níveis (DEBUG, INFO, WARNING, ERROR, CRITICAL), formatar mensagens e direcionar a saída para arquivos ou console.

No nosso projeto, vamos configurar um logger que escreve em um arquivo automacao.log e também imprime no console. Assim, temos um histórico das execuções e conseguimos identificar problemas. Vamos criar uma função de configuração que pode ser chamada no início do script.

import logging

def setup_logging(log_file: str = 'automacao.log') -> None:
    """Configura o logging para arquivo e console."""
    logger = logging.getLogger()
    logger.setLevel(logging.DEBUG)

    # Formato da mensagem
    formatter = logging.Formatter('%(asctime)s - %(levelname)s - %(message)s')

    # Handler para arquivo
    file_handler = logging.FileHandler(log_file, encoding='utf-8')
    file_handler.setLevel(logging.DEBUG)
    file_handler.setFormatter(formatter)
    logger.addHandler(file_handler)

    # Handler para console
    console_handler = logging.StreamHandler()
    console_handler.setLevel(logging.INFO)
    console_handler.setFormatter(formatter)
    logger.addHandler(console_handler)

Com essa configuração, podemos usar logging.info(), logging.error() etc. em todo o script. Por exemplo, na função process_file, podemos registrar o início e o fim do processamento. Isso nos ajuda a rastrear o que aconteceu e a diagnosticar falhas.

def process_file(input_path: Path, output_dir: Path) -> None:
    logging.info(f"Processando arquivo: {input_path}")
    try:
        # ... código de processamento ...
        logging.info(f"Arquivo processado com sucesso: {input_path.name}")
    except Exception as e:
        logging.error(f"Erro ao processar {input_path}: {e}")

Robustez

Robustez é a capacidade do programa de lidar com condições inesperadas sem quebrar. Isso inclui tratamento de exceções, validação de entradas, criação de diretórios automaticamente e uso de retry para operações que podem falhar temporariamente. Um script robusto é essencial para automações, pois muitas vezes roda sem supervisão.

Vamos aplicar várias técnicas de robustez no nosso projeto: verificar se os diretórios existem antes de usar, capturar exceções específicas (como FileNotFoundError e PermissionError) e usar try/except para evitar que o script pare em um erro isolado. Além disso, podemos adicionar um mecanismo de retry para arquivos que estão sendo usados por outro processo.

import time
from pathlib import Path

def ensure_directory(dir_path: Path) -> None:
    """Cria o diretório se não existir."""
    dir_path.mkdir(parents=True, exist_ok=True)

def process_file_with_retry(input_path: Path, output_dir: Path, max_retries: int = 3) -> bool:
    """Tenta processar um arquivo com retry em caso de erro."""
    for attempt in range(max_retries):
        try:
            process_file(input_path, output_dir)
            return True
        except PermissionError:
            logging.warning(f"Arquivo {input_path} em uso, tentativa {attempt+1}")
            time.sleep(2)
        except Exception as e:
            logging.error(f"Erro não recuperável: {e}")
            return False
    logging.error(f"Falha após {max_retries} tentativas")
    return False

Nesse exemplo, ensure_directory cria o diretório de saída se necessário, e process_file_with_retry tenta processar o arquivo várias vezes em caso de PermissionError. Isso é útil em ambientes onde arquivos podem estar bloqueados temporariamente.

Projeto Completo: Script de Automação

Agora vamos integrar tudo em um script completo. O script vai monitorar um diretório de entrada (por exemplo, entrada/), processar cada arquivo de texto encontrado, gerar um resumo em um diretório de saída (saida/) e mover o arquivo processado para um diretório de processados (processados/). Vamos usar agendamento para executar a verificação periodicamente e logging para registrar tudo.

import schedule
import time
import logging
from pathlib import Path

# Configuração
BASE_DIR = Path(__file__).parent
INPUT_DIR = BASE_DIR / "entrada"
OUTPUT_DIR = BASE_DIR / "saida"
PROCESSED_DIR = BASE_DIR / "processados"
LOG_FILE = BASE_DIR / "automacao.log"

# Configurar logging
logging.basicConfig(
    level=logging.INFO,
    format='%(asctime)s - %(levelname)s - %(message)s',
    handlers=[
        logging.FileHandler(LOG_FILE, encoding='utf-8'),
        logging.StreamHandler()
    ]
)

def ensure_directories():
    """Cria os diretórios necessários."""
    for directory in [INPUT_DIR, OUTPUT_DIR, PROCESSED_DIR]:
        directory.mkdir(parents=True, exist_ok=True)

def process_file(input_path: Path) -> None:
    """Processa um arquivo de texto e gera resumo."""
    try:
        logging.info(f"Processando: {input_path.name}")
        with input_path.open('r', encoding='utf-8') as f:
            content = f.read()
        word_count = len(content.split())
        line_count = content.count('\n') + 1

        output_path = OUTPUT_DIR / f"{input_path.stem}_resumo.txt"
        with output_path.open('w', encoding='utf-8') as f:
            f.write(f"Arquivo: {input_path.name}\n")
            f.write(f"Palavras: {word_count}\n")
            f.write(f"Linhas: {line_count}\n")

        # Move o arquivo para processados
        processed_path = PROCESSED_DIR / input_path.name
        input_path.rename(processed_path)
        logging.info(f"Processado e movido: {input_path.name}")
    except Exception as e:
        logging.error(f"Erro ao processar {input_path}: {e}")

def scan_and_process():
    """Verifica novos arquivos no diretório de entrada e processa."""
    ensure_directories()
    for file_path in INPUT_DIR.iterdir():
        if file_path.is_file() and file_path.suffix == '.txt':
            process_file(file_path)

def main():
    ensure_directories()
    logging.info("Iniciando automação...")
    # Executa imediatamente
    scan_and_process()
    # Agenda a cada 30 segundos
    schedule.every(30).seconds.do(scan_and_process)
    while True:
        schedule.run_pending()
        time.sleep(1)

if __name__ == "__main__":
    main()

Esse script é um exemplo funcional de automação. Ele cria os diretórios, processa arquivos .txt, gera resumos e move os arquivos. O agendamento a cada 30 segundos pode ser ajustado conforme a necessidade. A robustez é garantida com o uso de try/except e logging.

Boas Práticas e Observações Finais

Ao desenvolver automações, siga estas boas práticas: use caminhos relativos com Path(__file__).parent para que o script funcione em qualquer máquina; evite hardcoded paths; sempre use encoding explícito (utf-8) ao ler/escrever arquivos; configure logging para ter rastreabilidade; e trate exceções de forma específica para não mascarar erros. Além disso, teste o script em um ambiente controlado antes de colocá-lo em produção.

Outra dica importante: em automações de longa duração, monitore o consumo de memória e CPU. O uso de bibliotecas como schedule é leve, mas se você processar muitos arquivos, considere usar filas ou paralelismo. Por fim, documente seu código e mantenha os logs rotativos para não encher o disco.

Referências

Exercícios

  1. Escreva uma função que receba um diretório e retorne uma lista de todos os arquivos com extensão .txt, usando pathlib. Em seguida, teste com um diretório criado por você.
  2. Modifique o script de automação para que, em vez de mover o arquivo, ele o copie para o diretório de processados, mantendo o original no local.
  3. Adicione um mecanismo de retry com no máximo 3 tentativas para a função process_file em caso de PermissionError, com um intervalo de 2 segundos entre tentativas.
  4. Configure o logging para gravar mensagens de nível DEBUG em um arquivo separado e mensagens INFO no console. Use dois handlers diferentes.
  5. Implemente uma função que agende a execução da automação para rodar todos os dias às 8h da manhã usando a biblioteca schedule. Inclua o código necessário para manter o script em execução.

✓ Resposta: 
from pathlib import Path

def list_txt_files(directory: Path) -> list:
    return [f for f in directory.iterdir() if f.is_file() and f.suffix == '.txt']

# Exemplo de uso:
# dir = Path('meus_arquivos')
# arquivos = list_txt_files(dir)
# print(arquivos)

✓ Resposta: Para copiar em vez de mover, use shutil.copy2 em vez de rename. Exemplo: 
import shutil
from pathlib import Path

def process_file(input_path: Path) -> None:
    # ... processamento ...
    processed_path = PROCESSED_DIR / input_path.name
    shutil.copy2(input_path, processed_path)
    # Não remove o original

✓ Resposta: 
import time
from pathlib import Path

def process_file_with_retry(input_path: Path, output_dir: Path, max_retries: int = 3) -> bool:
    for attempt in range(max_retries):
        try:
            process_file(input_path, output_dir)
            return True
        except PermissionError:
            logging.warning(f"Arquivo em uso, tentativa {attempt+1}")
            time.sleep(2)
    logging.error("Falha após tentativas")
    return False

✓ Resposta: 
import logging

logger = logging.getLogger()
logger.setLevel(logging.DEBUG)

# Handler para DEBUG em arquivo
file_handler = logging.FileHandler('debug.log')
file_handler.setLevel(logging.DEBUG)
file_handler.setFormatter(logging.Formatter('%(asctime)s - %(levelname)s - %(message)s'))
logger.addHandler(file_handler)

# Handler para INFO no console
console_handler = logging.StreamHandler()
console_handler.setLevel(logging.INFO)
console_handler.setFormatter(logging.Formatter('%(levelname)s - %(message)s'))
logger.addHandler(console_handler)

✓ Resposta: 
import schedule
import time

def job():
    print("Executando automação diária")
    # chamar a função principal

# Agendar todos os dias às 8h
schedule.every().day.at("08:00").do(job)

while True:
    schedule.run_pending()
    time.sleep(1)