💬 Canal Telegram em 50 linhas

Validação do token (1 curl):

$ curl -s "https://api.telegram.org/bot${TG_TOKEN}/getMe" | jq

{
  "ok": true,
  "result": {
    "id": 7842913847,
    "is_bot": true,
    "first_name": "Intelecto Dev",
    "username": "intelecto_nm_bot",
    "can_join_groups": true,
    "can_read_all_group_messages": false,
    "supports_inline_queries": false
  }
}

Se ok: false, token errado ou revogado. Volte ao BotFather com /revoke e gere novo.

Anatomia de uma rodada de long-polling:

GET https://api.telegram.org/bot{TOKEN}/getUpdates
    ?offset={last_update_id + 1}
    &timeout=30
    &allowed_updates=["message","callback_query"]

# Resposta (vazia ou com batch):
{
  "ok": true,
  "result": [
    {
      "update_id": 819374821,
      "message": {
        "message_id": 47,
        "from": {"id": 12345, "first_name": "Nei"},
        "chat": {"id": 12345, "type": "private"},
        "date": 1715980800,
        "text": "oi agente"
      }
    }
  ]
}

channels/telegram.py

import asyncio, httpx, logging
from typing import Callable, Awaitable
from .base import Channel, IncomingMessage, OutgoingMessage

log = logging.getLogger(__name__)
API = "https://api.telegram.org/bot{token}/{method}"

class TelegramChannel(Channel):
    def __init__(self, token: str, allowed_user_ids: set[int] | None = None):
        self.token = token
        self.allowed = allowed_user_ids or set()
        self.client = httpx.AsyncClient(timeout=35.0)
        self._offset = 0
        self._running = False

    async def start(self, on_message: Callable[[IncomingMessage], Awaitable[None]]) -> None:
        self._running = True
        log.info("telegram.start", extra={"allowed": len(self.allowed)})
        while self._running:
            try:
                r = await self.client.get(API.format(token=self.token, method="getUpdates"),
                    params={"offset": self._offset, "timeout": 30,
                            "allowed_updates": '["message"]'})
                r.raise_for_status()
                for u in r.json().get("result", []):
                    self._offset = u["update_id"] + 1
                    msg = u.get("message") or {}
                    chat_id = msg.get("chat", {}).get("id")
                    text = msg.get("text") or ""
                    if self.allowed and chat_id not in self.allowed:
                        await self.send(OutgoingMessage(user_id=str(chat_id),
                            text="⛔ Bot privado. Solicite acesso ao admin."))
                        log.warning("telegram.denied", extra={"chat_id": chat_id})
                        continue
                    await on_message(IncomingMessage(
                        user_id=str(chat_id), text=text, timestamp=msg["date"],
                        metadata={"message_id": msg["message_id"], "voice": msg.get("voice")}))
            except httpx.HTTPError as e:
                log.error("telegram.poll_error", exc_info=e)
                await asyncio.sleep(3)  # backoff

    async def send(self, m: OutgoingMessage) -> str:
        for chunk in _chunks(m.text, 4000):
            r = await self.client.post(API.format(token=self.token, method="sendMessage"),
                json={"chat_id": int(m.user_id), "text": chunk,
                      "parse_mode": "MarkdownV2" if m.parse_mode == "markdown" else None,
                      "reply_to_message_id": int(m.reply_to) if m.reply_to else None})
            r.raise_for_status()
        return str(r.json()["result"]["message_id"])

    async def stop(self) -> None:
        self._running = False
        await self.client.aclose()

def _chunks(s: str, n: int):
    for i in range(0, len(s) or 1, n): yield s[i:i+n] or " "

Escaper canônico + fallback automático:

import re

_MDV2_SPECIAL = r'_*[]()~`>#+-=|{}.!'
_MDV2_RE = re.compile('([' + re.escape(_MDV2_SPECIAL) + '])')

def escape_mdv2(text: str) -> str:
    """Escapa todos os reservados FORA de blocos de código."""
    parts, out = text.split('```'), []
    for i, part in enumerate(parts):
        if i % 2 == 1:               # bloco de código: preserva intacto
            out.append('```' + part + '```')
        else:
            out.append(_MDV2_RE.sub(r'\\\1', part))
    return ''.join(out)

async def send_safe(ch, user_id, text):
    """Tenta MarkdownV2; se falhar, manda como plain."""
    try:
        await ch.send(OutgoingMessage(user_id=user_id,
            text=escape_mdv2(text), parse_mode="markdown"))
    except httpx.HTTPStatusError as e:
        if e.response.status_code == 400:
            log.warning("mdv2.fallback", extra={"err": e.response.text[:200]})
            await ch.send(OutgoingMessage(user_id=user_id,
                text=text, parse_mode="plain"))
        else:
            raise

.env + carregamento:

# .env
TG_TOKEN=7842913847:AAH...
ALLOWED_USER_IDS=12345,67890,11223      # vírgula, sem espaço
ADMIN_USER_ID=12345

# config.py
import os
ALLOWED = {int(x) for x in os.environ.get("ALLOWED_USER_IDS","").split(",") if x}
ADMIN   = int(os.environ.get("ADMIN_USER_ID", "0"))

Gate com onboarding amigável:

DENY_MSG = (
    "🔒 Este bot é privado.\n\n"
    "Seu ID: `{chat_id}`\n"
    "Envie esse número ao admin para liberar acesso.\n\n"
    "_Se isso foi um engano, pode ignorar — nenhuma mensagem foi processada._"
)

async def gate(msg: IncomingMessage, allowed: set[int], admin: int) -> bool:
    chat_id = int(msg.user_id)
    if chat_id in allowed:
        return True
    # log + alerta para admin (1x por user_id, com cooldown)
    log.warning("access_denied", extra={"chat_id": chat_id, "text": msg.text[:80]})
    await audit_save(event="access_denied", chat_id=chat_id, text=msg.text)
    if should_alert(chat_id):     # cooldown 1h por chat_id
        await ch.send(OutgoingMessage(user_id=str(admin),
            text=f"🚨 Tentativa de acesso: `{chat_id}` — \"{msg.text[:60]}\""))
    await ch.send(OutgoingMessage(user_id=str(chat_id),
        text=DENY_MSG.format(chat_id=chat_id)))
    return False

Download + transcrição:

from faster_whisper import WhisperModel
import httpx, tempfile, os, asyncio
from dataclasses import replace

# Carrega 1x no startup. small = 244MB, medium = 769MB
_model = WhisperModel("small", device="cpu", compute_type="int8")

async def download_voice(token: str, file_id: str) -> str:
    """Baixa OGG do Telegram, retorna path local."""
    async with httpx.AsyncClient(timeout=30) as c:
        meta = (await c.get(
            f"https://api.telegram.org/bot{token}/getFile",
            params={"file_id": file_id})).json()
        path = meta["result"]["file_path"]
        ogg = (await c.get(
            f"https://api.telegram.org/file/bot{token}/{path}")).content
    fd, fp = tempfile.mkstemp(suffix=".ogg")
    os.write(fd, ogg); os.close(fd)
    return fp

def transcribe(path: str, language: str = "pt") -> str:
    segments, _ = _model.transcribe(path, language=language, beam_size=1)
    return " ".join(s.text.strip() for s in segments).strip()

# Integração no handler:
async def on_message(msg: IncomingMessage):
    voice = msg.metadata.get("voice")
    if voice:
        path = await download_voice(TG_TOKEN, voice["file_id"])
        try:
            text = await asyncio.to_thread(transcribe, path)
            log.info("voice.transcribed", extra={"len": len(text), "dur": voice["duration"]})
            msg = replace(msg, text=text)
        finally:
            os.unlink(path)
    await agent.handle(msg)