Bot Python modular para monitorar promoções em canais Telegram, gerar links de afiliado com IA (OpenAI) e republicar automaticamente.
| Funcionalidade | Descrição |
|---|---|
| Monitoramento em tempo real | Escuta novas mensagens em um canal Telegram configurável |
| Extração inteligente de links | Regex robusta que ignora pontuação final, suporta subdomínios, encurtadores oficiais (meli.la, amzn.to, shp.ee, ali.sale, etc.) |
| Resolução de encurtadores | Segue redirects HTTP automaticamente e gera afiliado da URL final |
| Validação por domínio | Filtra apenas URLs permitidas (Amazon, Shopee, Mercado Livre, Magalu, AliExpress) |
| Geração de links de afiliado | Converte URLs em links de afiliado com parâmetros personalizados por domínio |
| Classificação por IA | OpenAI GPT-4o decide se a promoção é relevante, com circuit breaker pra resiliência |
| Mensagens sem link também | Cupons expirados e avisos sem URL são classificados e repostados se relevantes |
| Limpeza de mensagem | Remove linhas promocionais ("Conheça", "Veja mais", etc.) e URLs não afiliadas |
| Prevenção de duplicidade | SQLite com WAL mode + tabela posted_links (URL) + processed_messages (msg_id) |
| Substituição de links | Troca a URL original pelo link de afiliado diretamente no texto |
| Notificações ao admin | Avisos de rejeição, erros ou indisponibilidade da IA via DM |
| Logging robusto | Console + arquivo rotativo (10 MB, 5 backups) |
| Reconexão automática | Retry inteligente com tratamento de FloodWaitError, SlowModeWaitError, etc. |
| Graceful shutdown | Captura SIGTERM/SIGINT, faz cleanup ordenado de DB e clientes |
| Health check | HTTP /healthz (liveness) e /readyz (readiness) na porta 8080 + heartbeat a cada 5min |
| Docker prod-ready | Multi-stage, non-root, tini como PID 1, HEALTHCHECK integrado |
| Segurança SSRF | Bloqueia IPs privados e link-local antes de resolver encurtadores |
| StringSession | Suporte opcional pra armazenar sessão em env var (sem arquivo .session) |
| Cache de encurtadores | TTL 1h evita resolver a mesma URL múltiplas vezes |
Comando /test |
Admin pode testar posting no canal destino via DM do bot |
├── app.py # Entry point: wiring + signal handlers + handlers
├── pyproject.toml # ruff, mypy, pytest config
├── requirements.txt
├── Dockerfile # Multi-stage, non-root, tini, healthcheck
├── docker-compose.yml
├── .env.example # Template de configuração
├── .dockerignore
├── config/
│ └── settings.py # Config via pydantic-settings + validadores + AFFILIATE_CONFIG
├── core/
│ ├── logger.py # Logging console + RotatingFileHandler
│ ├── circuit_breaker.py # Circuit breaker para OpenAI
│ └── telegram_retry.py # Classificador de erros Telegram + retry com jitter
├── domain/
│ ├── link_model.py # Dataclass Link
│ ├── link_parser.py # Extração/validação + cache de encurtadores + afiliado
│ └── url_safety.py # Proteção SSRF (bloqueia IPs privados)
├── services/
│ ├── ai_classifier.py # Classificação por OpenAI + modo simulate + circuit breaker
│ ├── database.py # SQLite WAL mode + posted_links + processed_messages
│ ├── health_server.py # /healthz + /readyz na porta 8080
│ ├── notifier.py # Notificações ao administrador
│ ├── telegram_listener.py # Escuta o canal de origem (user session)
│ └── telegram_sender.py # Publica no canal de destino (bot session)
├── tests/ # 43 testes pytest-asyncio
│ ├── conftest.py
│ ├── test_link_parser.py # 24 testes — extração, validação, afiliados, encurtadores
│ ├── test_database.py # 9 testes — CRUD + idempotência por msg_id
│ ├── test_ai_classifier.py # 5 testes — classificação e fallbacks
│ └── test_url_safety.py # 7 testes — SSRF protection
└── docs/ # Documentação completa (SDD + ADRs + runbook + roadmap)
├── architecture/SDD.md
├── adr/
└── operations/
┌─────────────┐ escuta ┌──────────────────┐
│ Canal │◄───────────────│ TelegramListener │
│ de Origem │ │ (user session) │
└─────────────┘ └────────┬─────────┘
│ msg recebida
▼
┌──────────────────┐
│ clean_message │
│ remove promo + │
│ URLs não afil. │
└────────┬─────────┘
│
▼
┌──────────────────┐
│ LinkParser │
│ extrai + resolve │
│ encurtadores │
└────────┬─────────┘
│ links válidos
▼
┌──────────────────┐
│ DB.is_posted? │
└────┬─────────┬───┘
sim │ │ não
▼ ▼
┌────────┐ ┌──────────────────┐
│ skip │ │ AIClassifier │
└────────┘ │ (circuit breaker)│
└────────┬─────────┘
┌────────┬───────┴───────┐
│ │ │
"sim" "não" "erro"
│ │ │
▼ ▼ ▼
┌──────────────┐ ┌──────────────────┐
│TelegramSender│ │ Notifier │
│publica c/ │ │ notifica admin │
│afiliado │ └──────────────────┘
└──────┬───────┘
▼
┌──────────────┐
│DB │
│posted_links │
│+ msg_id │
└──────────────┘
2 sessões Telegram:
TelegramListener(user session): lê canal de origemTelegramSender+Notifier(bot session): posta no destino e notifica admin (compartilham o mesmoTelegramClient)
git clone https://github.com/herikk04/PromoMind.git
cd PromoMind
python -m venv .venv
# Windows
.venv\Scripts\activate
# macOS/Linux
source .venv/bin/activate
pip install -r requirements.txtCrie um .env na raiz (use .env.example como template):
# === Telegram (obrigatório) ===
TELEGRAM_API_ID=seu_api_id
TELEGRAM_API_HASH=seu_api_hash_32_hex
TELEGRAM_BOT_TOKEN=seu_token_botfather
TELEGRAM_SOURCE_CHANNEL_ID=-1001234567890
TELEGRAM_TARGET_CHANNEL_ID=-1001234567891
ADMIN_USER_ID=123456789
# === Opcional: StringSession (recomendado em produção) ===
# Gere com: python -c "import asyncio; from telethon import TelegramClient; from telethon.sessions import StringSession; asyncio.run((lambda: ...)())"
TELEGRAM_LISTENER_SESSION=
# === Opcional: Health server ===
HEALTH_SERVER_HOST=0.0.0.0
HEALTH_SERVER_PORT=8080
# === OpenAI (modo simulate se vazio) ===
AI_MODE=simulate # ou 'real'
OPENAI_API_KEY=sk-sua_chave| Variável | Onde obter |
|---|---|
TELEGRAM_API_ID / TELEGRAM_API_HASH |
https://my.telegram.org |
TELEGRAM_BOT_TOKEN |
@BotFather no Telegram |
TELEGRAM_SOURCE_CHANNEL_ID |
Encaminhe uma mensagem do canal para @RawDataBot |
TELEGRAM_TARGET_CHANNEL_ID |
ID do seu canal. Bot precisa ser admin com permissão de postar |
ADMIN_USER_ID |
Mande msg para @RawDataBot e copie from > id |
TELEGRAM_LISTENER_SESSION |
StringSession gerada (evita arquivo .session em disco) |
OPENAI_API_KEY |
https://platform.openai.com/api-keys |
Os IDs de afiliado padrão estão em config/settings.py. Para customizar via .env:
AFFILIATE_ID_AMAZON=seu-id-20
AFFILIATE_ID_SHOPEE=seu_id_shopee
AFFILIATE_ID_ML=seu_id_mercadolivre
AFFILIATE_ID_MAGALU=seu_id_magalu
AFFILIATE_ID_ALIEXPRESS=seu_id_aliexpresspython app.pyNa primeira execução o Telethon pede seu telefone + código de verificação (gera listener_session.session se TELEGRAM_LISTENER_SESSION estiver vazio).
Abra conversa privada com o bot no Telegram e digite /test. Ele tenta postar uma mensagem de teste no canal de destino e responde com sucesso/erro. Restrito ao admin.
curl http://localhost:8080/healthz # sempre 200
curl http://localhost:8080/readyz # 200 se Telegram conectado; 503 se caiuDefault é AI_MODE=simulate — não chama a OpenAI e publica tudo. Perfeito pra testar sem custo. Pra usar classificação real:
AI_MODE=real
OPENAI_API_KEY=sk-sua_chaveCircuit breaker abre após 5 falhas consecutivas da OpenAI e bloqueia chamadas por 30s — evita gastar tempo/$$ quando a API está fora.
docker compose up -d
docker compose logs -f
docker compose downImagem usa:
- Multi-stage build (menor)
- User
promomindnon-root tinicomo PID 1 (forward signals)HEALTHCHECKchamando/healthzSTOPSIGNAL SIGTERM(graceful shutdown)
| Resultado da IA | Ação |
|---|---|
"sim" |
Publica no destino com link de afiliado |
"não" |
Notifica admin com promoção rejeitada |
"erro" (IA indisponível) |
Notifica admin; não publica (fail-closed) |
pytest # roda os 43 testes
pytest --cov=. # com coberturaruff check . # lint
ruff format . # formatação
mypy . # type check| Documento | Descrição |
|---|---|
| SDD | Software Design Document (arc42) — arquitetura, diagramas Mermaid, decisões |
| ADRs | 7 Architecture Decision Records documentando escolhas-chave |
| Runbook | Guia operacional — setup, troubleshooting, manutenção |
| Roadmap | 25 melhorias futuras priorizadas (P0/P1/P2) — todas as P0 + P1 principais já aplicadas |
Fase 1 Bugs críticos 13 corrigidos (nomenclatura, async, imports)
Fase 2 Funcionalidade 6 melhorias (regex, replace, Link model, media)
Fase 3 Qualidade 7 itens (testes, Docker, mypy, ruff, health check)
Fase 4 Limpeza 1 melhoria (pydantic-settings)
Fase 5 Modo simulado AI_MODE, .env.example, teste sem OpenAI
Fase 6 Pipeline robusta 1 msg/link; encurtadores; limpeza; /test command
Fase 7 SDD + ADRs docs/ com Software Design Document e 7 ADRs
Fase 8 Robustez SDD SSRF, StringSession, retry Telegram, circuit breaker,
cache encurtadores, WAL mode, health server, graceful
shutdown, Dockerfile multi-stage, idempotência por msg_id
MIT