Um script que só avisa não precisa de python-telegram-bot

Um pipeline de ETL termina de madrugada, ou falha no meio da rodada. Um deploy quebra. Um preço de mercado cruza um limite que alguém configurou. Em qualquer um desses casos você não precisa de conversa — precisa que uma mensagem chegue no celular, ponto.

A tentação comum é instalar python-telegram-bot só para isso: trazer o framework inteiro, com handler, Application e loop de long polling, para mandar uma linha de texto que ninguém vai responder. Se o bot só existe para mandar avisos e nunca precisa receber comando, sobra tudo isso. Um requests.post puro contra a API HTTP do Telegram resolve com uma dependência a menos — a troca é que esse bot nunca vai responder a um /start ou a pergunta nenhuma; ele só fala. O meu pipeline de dados de mercado, o Equity_Tracker, segue exatamente essa linha: requests==2.32.5 é a única dependência de transporte no requirements.txt, sem python-telegram-bot em lugar nenhum.

O padrão que resolve isso tem três peças, não mais: uma função de envio que respeita o limite de caracteres da API, uma convenção de formatação que sobrevive à prévia cortada da tela de bloqueio, e um ponto único no pipeline de onde a função é chamada. Nenhuma delas exige aprender a API de handlers de um framework de bot.

O que você precisa

• Um bot já criado no BotFather, com o token em mãos
• O chat ID de destino
• Python com requests instalado (pip install requests)
• Um pipeline, script ou job existente de onde disparar o alerta

Passo 1: Escrever a função de envio com chunking

O limite real do Telegram é 4096 caracteres por mensagem; a função abaixo usa 4000 como margem de segurança:

TG_API = "https://api.telegram.org/bot{token}/sendMessage"
TG_MSG_LIMIT = 4000  # Telegram hard limit is 4096; leave headroom

def send_alert(message: str) -> bool:
    token = os.environ.get("TELEGRAM_BOT_TOKEN")
    chat_id = os.environ.get("TELEGRAM_CHAT_ID")
    if not token or not chat_id:
        logger.warning("Telegram alert skipped: TELEGRAM_BOT_TOKEN / TELEGRAM_CHAT_ID not set")
        return False
    url = TG_API.format(token=token)
    sent_any = False
    for i in range(0, len(message), TG_MSG_LIMIT):
        chunk = message[i : i + TG_MSG_LIMIT]
        try:
            r = requests.post(url, data={"chat_id": chat_id, "text": chunk}, timeout=10)
            r.raise_for_status()
            sent_any = True
        except requests.RequestException as e:
            logger.error("Telegram send failed: %s", e)
            return False
    return sent_any

O for fatia a mensagem em blocos de 4000 caracteres — necessário para resumos longos de pipeline com várias etapas. Quando faltam variáveis de ambiente, a função retorna False em vez de quebrar, o que evita que uma rodada de CI ou de desenvolvimento falhe por falta de credencial. E o timeout=10 é a única proteção contra um envio que trava esperando resposta que não chega.

raise_for_status() transforma qualquer resposta HTTP de erro — token inválido, chat ID errado — numa exceção, que o except requests.RequestException captura junto com falha de conexão e timeout. Um chunk que falha interrompe o envio dos seguintes: a função prefere devolver False a mandar só metade de uma mensagem longa fora de ordem.

Passo 2: Colocar o emoji de status no primeiro caractere

A prévia de notificação na tela de bloqueio do celular corta a mensagem em cerca de 40 caracteres — é uma lição registrada no próprio Equity_Tracker. Se o status, sucesso ou falha, não estiver logo no início, ninguém vê nada útil sem desbloquear o telefone. O padrão que resolve isso é simples: ✅ para sucesso, 🔴 para falha, sempre como primeiro caractere da mensagem.

O Equity_Tracker também usa ícones por etapa dentro do corpo da mensagem — ✓, ⊘, ✗ — numa tabela de passos do pipeline. Não precisa reproduzir essa tabela para aproveitar a ideia central: o primeiro caractere é o que a tela de bloqueio mostra; o resto é para quem abre o aplicativo. Isso importa ainda mais quando a mensagem passa dos 4000 caracteres do Passo 1 e vira mais de um chunk — é o primeiro chunk que carrega o status, então é nele que o emoji precisa vir primeiro.

Passo 3: Chamar o alerta a partir do seu pipeline

if pipeline_name == "daily":
    try:
        send_alert(format_pipeline_summary(engine, pipeline_name, run_date, all_ok))
    except Exception as e:
        log.error("Telegram summary alert failed: %s", e)

O alerta dispara uma vez ao fim do pipeline, sucesso ou falha — não em intervalo fixo. É o próprio evento de término que decide o envio. O try/except externo garante que uma falha ao notificar não derruba o pipeline que já terminou: o trabalho real já foi feito, perder o aviso é ruim, mas não é motivo para o processo inteiro falhar por causa disso.

O if pipeline_name == "daily" restringe o aviso a essa rotina específica; outros pipelines do mesmo sistema podem rodar sem gerar alerta nenhum — escolha de quem mantém aquele pipeline, não um limite da função send_alert em si.

O que pode dar errado

• Mensagem maior que 4096 caracteres sem chunking: o Telegram rejeita o envio inteiro; sempre fatiar antes de mandar. Um resumo de pipeline com várias etapas passa desse tamanho com facilidade.
• Emoji de status no meio ou no fim da mensagem: a prévia da tela de bloqueio já cortou antes de chegar lá. O usuário só vê o resultado ao abrir o aplicativo — exatamente o atraso que o emoji na frente evita.
• TELEGRAM_BOT_TOKEN ou TELEGRAM_CHAT_ID ausentes sem checagem: eu trato isso com return False e um aviso de log, não um crash. Um ambiente de desenvolvimento sem essas variáveis continua rodando o pipeline normalmente, só sem mandar alerta.
• Sem timeout na chamada HTTP: uma instabilidade de rede trava o processo esperando resposta que não chega. Um pipeline que já terminou fica preso num alerta que nunca decide se falhou ou teve sucesso.

Próximos passos

Se o alerta precisar de nova tentativa automática, o meu script não implementa isso — falha uma vez e retorna False; é a extensão mais óbvia para quem for além deste tutorial. Um retry simples, com uma pequena espera entre tentativas, cobre a maior parte dos casos de instabilidade passageira de rede, sem precisar de fila nem serviço externo.

Se surgir a necessidade de receber comando — não só mandar aviso — essa função deixa de bastar. Migrar para python-telegram-bot é o próximo passo natural.

Reaproveitar o mesmo bot — mesmo token, mesmo chat ID — para vários pipelines diferentes é mais prático que criar um bot novo por projeto: eu reaproveito o token do bot de outro sistema meu, o SignalForge, para os alertas do Equity_Tracker.

Nem todo canal de aviso precisa de um framework. Precisa de uma chamada HTTP e uma disciplina de formatação.