Monitor do Grupo WhatsApp com resumos de IA e relatórios de e-mail agendados

Crie uma plataforma completa de monitoramento do WhatsApp usando a Evolution API, FastAPI, PostgreSQL, Claude AI e Resend – nenhuma API oficial do WhatsApp Business é necessária.

Por que monitorar os grupos do WhatsApp?

O WhatsApp é o canal de comunicação de fato para equipes em toda a Europa, América Latina, África e Ásia. Para empresas que executam operações distribuídas – call centers, equipes de campo, configurações de vários escritórios – informações críticas fluem através de grupos do WhatsApp todos os dias:

• As decisões operacionais são tomadas em chats em grupo, não em e-mail.
• As transferências de deslocamento acontecem através de notas de voz e mensagens rápidas.
• Escaladas de clientes chegam como capturas de tela encaminhadas.
• Problemas de pessoal surgem primeiro em grupos de equipe, horas antes de atingirem o sistema de emissão de bilhetes.

O problema: o WhatsApp não tem pesquisa integrada entre grupos, nenhuma análise, nenhuma exportação, nenhuma maneira de a gerência ver o que aconteceu ontem sem percorrer centenas de mensagens em um telefone.

O que este sistema resolve:

Problema	Solução
“O que a equipe discutiu ontem?”	Resumos diários gerados por IA, entregues por e-mail
“Quais grupos ficaram em silêncio?”	Alertas de atividade sinalizam grupos de mensagens zero
“Quem são os participantes mais ativos?”	Estatísticas por grupo com os principais remetentes
“Encontre essa mensagem do mês passado”	Pesquisa de texto completo em todos os grupos
“Preciso de um relatório semanal para o conselho”	E-mails digeridos agendados com estatísticas e destaques
“Precisamos manter registros para conformidade”	Todas as mensagens armazenadas no PostgreSQL com timestamps

Custo: Essencialmente gratuito. A Evolution API é de código aberto. O PostgreSQL é gratuito. Reenviar dá-lhe 3.000 e-mails / mês no nível gratuito. O único componente pago é a Claude API para resumos, que custa cerca de US $ 0,02-0,05 por grupo por dia usando Haiku.

---

2. 2. Visão geral da arquitetura

                                    ┌─────────────────────────┐
                                    │      Your Phone         │
                                    │    (WhatsApp linked)    │
                                    └────────┬────────────────┘
                                             │ Baileys Protocol
                                             │ (WebSocket)
┌─────────────────────────────────────────────┼──────────────────────────────┐
│  Docker Network (monitoring)                │                              │
│                                             │                              │
│  ┌──────────────────────┐                   │                              │
│  │   Redis 7 Alpine     │◄──cache──┐        │                              │
│  │   (session store)    │          │        │                              │
│  └──────────────────────┘          │        │                              │
│                                    │        ▼                              │
│  ┌──────────────────────┐    ┌─────┴────────────────┐                      │
│  │  PostgreSQL 16       │◄───│   Evolution API      │                      │
│  │  (evolution DB)      │    │   v2.x (Baileys)     │                      │
│  │  (whatsapp_monitor)  │    │   Port 8080          │                      │
│  └──────────┬───────────┘    └─────────┬────────────┘                      │
│             │                          │ Webhook POST                      │
│             │                          │ /webhook/evolution                │
│             │                          ▼                                   │
│             │              ┌───────────────────────┐                       │
│             │              │   FastAPI Service      │                       │
│             │◄─── asyncpg ─│   (wa-monitor)        │──► Claude API         │
│             │              │   Port 8086            │    (Haiku/Sonnet)     │
│             │              │   systemd managed      │                       │
│             │              └───────────┬────────────┘                       │
│             │                          │                                   │
│             │                  ┌───────┼────────┐                          │
│             │                  │       │        │                          │
│             │                  ▼       ▼        ▼                          │
│             │            /metrics  /health  /reports                       │
│             │                  │                 │                          │
│  ┌──────────┴───┐              │                 │                          │
│  │  Prometheus   │◄── scrape ──┘                 │                          │
│  └──────┬───────┘                                │                          │
│         │                                        ▼                          │
│  ┌──────┴───────┐                    ┌──────────────────┐                  │
│  │   Grafana     │                    │  Resend API      │                  │
│  │   Dashboards  │                    │  (email delivery)│                  │
│  └──────────────┘                    └──────────────────┘                  │
│                                                                            │
│  ┌──────────────────────────────────────────────────────┐                  │
│  │   PHP Dashboard (Apache, port 8082)                   │                  │
│  │   whatsapp.php ── Group monitor, messages, AI chat    │                  │
│  │   email-reports.php ── Report CRUD, SMTP config       │                  │
│  │   whatsapp-data.php ── AJAX proxy to FastAPI          │                  │
│  │   whatsapp-ask.php ── AI analysis proxy               │                  │
│  └──────────────────────────────────────────────────────┘                  │
└────────────────────────────────────────────────────────────────────────────┘

Resumo do fluxo de dados:

1. Seu telefone fica conectado ao WhatsApp através da integração Baileys da Evolution API (reimplementação de código aberto do protocolo WhatsApp Web).
2. Cada mensagem de grupo aciona um webhook POST para o serviço FastAPI.
3. O FastAPI analisa a mensagem, armazena-a em uma tabela particionada do PostgreSQL e, opcionalmente, baixa mídia.
4. Um trabalho cron noturno gera resumos de IA para cada grupo usando Claude Haiku.
5. A cada 15 minutos, o agendador de relatórios verifica os devidos relatórios de e-mail e os envia via Resend.
6. O painel PHP fornece uma interface da Web para grupos de navegação, pesquisa de mensagens e gerenciamento de relatórios.

---

3. Pré-requisitos

Requisitos do servidor:

• Servidor Linux (Ubuntu 22.04/24.04 recomendado), 2+ núcleos de CPU, 4+ GB de RAM
• Docker e Docker Compose instalados
• Python 3.11+ com pip e venv
• Um número de telefone dedicado ao monitoramento do WhatsApp (este telefone permanece “conectado”)

Contas necessárias:

• Chave de API antrópica — para resumos de Claude AI (console.anthropic.com)
• Resend account — para entrega de e-mail (resend.com), nível gratuito: 3.000 e-mails/mês

Instale dependências do sistema:

# Docker
curl -fsSL https://get.docker.com | sh

# Python 3.12 + venv
apt update && apt install -y python3.12 python3.12-venv python3-pip ffmpeg

# Optional: for image thumbnail generation
pip3 install Pillow

---

4. 4. Evolution API: Configuração do Docker

A Evolution API é a ponte entre o WhatsApp e sua aplicação. Ele usa a biblioteca Baileys (um cliente WhatsApp Web com engenharia reversa) para manter uma conexão persistente – sem aprovação oficial da API, sem verificação do Facebook Business, sem taxas mensais.

Docker Compõe a Configuração

Crie o seu docker-compose.yml. Os principais serviços são Evolution API, Redis (para cache de sessão) e PostgreSQL (compartilhado com seus outros serviços):

version: "3.8"

services:
  # PostgreSQL — shared database for Evolution API + your monitor
  postgres:
    image: postgres:16-alpine
    container_name: postgres
    restart: unless-stopped
    environment:
      POSTGRES_USER: postgres
      POSTGRES_PASSWORD: ${POSTGRES_PASSWORD}
    volumes:
      - postgres_data:/var/lib/postgresql/data
    healthcheck:
      test: ["CMD-SHELL", "pg_isready -U postgres"]
      interval: 10s
      timeout: 5s
      retries: 5
    networks:
      - monitoring

  # Evolution API — WhatsApp connection via Baileys
  evolution-api:
    image: evoapicloud/evolution-api:v2.3.7
    container_name: evolution-api
    restart: unless-stopped
    ports:
      - "8080:8080"
    environment:
      # Server
      - SERVER_URL=http://YOUR_SERVER_IP:8080
      - AUTHENTICATION_API_KEY=${EVOLUTION_API_KEY}
      # Database persistence
      - DATABASE_ENABLED=true
      - DATABASE_PROVIDER=postgresql
      - DATABASE_CONNECTION_URI=postgresql://postgres:${POSTGRES_PASSWORD}@postgres:5432/evolution
      - DATABASE_SAVE_DATA_INSTANCE=true
      - DATABASE_SAVE_DATA_NEW_MESSAGE=true
      - DATABASE_SAVE_DATA_CONTACTS=true
      - DATABASE_SAVE_DATA_CHATS=true
      # Redis caching
      - CACHE_REDIS_ENABLED=true
      - CACHE_REDIS_URI=redis://redis:6379/0
      # Logging
      - LOG_LEVEL=WARN
      # Global webhook — sends all events to your FastAPI service
      - WEBHOOK_GLOBAL_ENABLED=true
      - WEBHOOK_GLOBAL_URL=http://172.18.0.1:8086/webhook/evolution
      - WEBHOOK_GLOBAL_WEBHOOK_BY_EVENTS=false
      - WEBHOOK_EVENTS_MESSAGES_UPSERT=true
      - WEBHOOK_EVENTS_GROUPS_UPSERT=true
      - WEBHOOK_EVENTS_GROUP_PARTICIPANTS_UPDATE=true
      - WEBHOOK_EVENTS_CONNECTION_UPDATE=true
      - WEBHOOK_EVENTS_QRCODE_UPDATED=true
      # Disable unused integrations
      - TYPEBOT_ENABLED=false
      - CHATWOOT_ENABLED=false
      - OPENAI_ENABLED=false
      - DIFY_ENABLED=false
      - S3_ENABLED=false
    volumes:
      - evolution_instances:/evolution/instances
    depends_on:
      postgres:
        condition: service_healthy
      redis:
        condition: service_healthy
    networks:
      - monitoring

  # Redis — session and cache store for Evolution API
  redis:
    image: redis:7-alpine
    container_name: redis
    restart: unless-stopped
    command: redis-server --maxmemory 256mb --maxmemory-policy allkeys-lru
    volumes:
      - redis_data:/data
    healthcheck:
      test: ["CMD", "redis-cli", "ping"]
      interval: 10s
      timeout: 3s
      retries: 5
    networks:
      - monitoring

volumes:
  postgres_data:
  evolution_instances:
  redis_data:

networks:
  monitoring:
    driver: bridge

Arquivo de ambiente

Criar a .envarquivo ao lado do seu arquivo de composição:

POSTGRES_PASSWORD=your_secure_pg_password_here
EVOLUTION_API_KEY=your_evo_api_key_here

Começando a pilha

docker compose up -d
docker compose logs -f evolution-api  # Watch for startup

Criando uma instância do WhatsApp e vinculando seu telefone

Uma vez que a Evolution API esteja em execução, crie uma instância e digitalize o código QR:

# Create instance
curl -s -X POST http://localhost:8080/instance/create \
  -H "apikey: YOUR_EVOLUTION_API_KEY" \
  -H "Content-Type: application/json" \
  -d '{
    "instanceName": "wa-monitor",
    "integration": "WHATSAPP-BAILEYS",
    "qrcode": true,
    "webhook": {
      "url": "http://172.18.0.1:8086/webhook/evolution",
      "byEvents": false,
      "base64": true,
      "events": [
        "MESSAGES_UPSERT",
        "GROUPS_UPSERT",
        "GROUP_PARTICIPANTS_UPDATE",
        "CONNECTION_UPDATE",
        "QRCODE_UPDATED"
      ]
    }
  }'

# Get QR code (returns base64-encoded image)
curl -s http://localhost:8080/instance/connect/wa-monitor \
  -H "apikey: YOUR_EVOLUTION_API_KEY" | jq -r '.base64'

Abra o WhatsApp no seu telefone, vá para Dispositivos vinculados e digitalize o código QR. A conexão persiste em reinicializações porque a Evolution API armazena a sessão no PostgreSQL.

Importante: O URL do webhook usa 172.18.0.1porque a Evolution API é executada dentro do Docker, mas precisa acessar seu serviço FastAPI em execução no host. Este é o Gateway da ponte Docker IP. Ajuste se sua rede Docker usar uma sub-rede diferente.

Verificando a conexão

curl -s http://localhost:8080/instance/connectionState/wa-monitor \
  -H "apikey: YOUR_EVOLUTION_API_KEY"

# Expected: {"instance":{"instanceName":"wa-monitor","state":"open"}}

---

5. Esquema PostgreSQL com Particionamento de Tabela

O volume de mensagens de grupos ativos do WhatsApp pode crescer rapidamente. Um único grupo de 50 pessoas pode produzir mais de 500 mensagens por dia. Multiplique por 10-20 grupos, adicione mídia e você está olhando para milhões de linhas dentro de meses.

O particionamento de tabela resolve isso dividindo wa_messagesem pedaços mensais. Cada mês recebe sua própria tabela física, o que significa:

• Consultas que filtram por data apenas digitalizam a partição relevante
• Dados antigos podem ser descartados simplesmente desconectando uma partição (instante, tempo de inatividade zero)
• A manutenção de vácuo e índice operam em tabelas menores
• Os backups podem ter como alvo meses específicos

Projeto do Esquema

-- Run inside your PostgreSQL container
CREATE DATABASE whatsapp_monitor;
\c whatsapp_monitor

-- ── Groups ──
CREATE TABLE wa_groups (
    jid TEXT PRIMARY KEY,           -- WhatsApp JID: [email protected]
    name TEXT NOT NULL DEFAULT '',
    description TEXT DEFAULT '',
    participant_count INTEGER DEFAULT 0,
    created_at TIMESTAMPTZ DEFAULT NOW(),
    updated_at TIMESTAMPTZ DEFAULT NOW(),
    is_monitored BOOLEAN DEFAULT true
);

-- ── Participants ──
CREATE TABLE wa_participants (
    id SERIAL PRIMARY KEY,
    group_jid TEXT NOT NULL REFERENCES wa_groups(jid),
    phone TEXT NOT NULL,
    push_name TEXT DEFAULT '',
    is_admin BOOLEAN DEFAULT false,
    first_seen TIMESTAMPTZ DEFAULT NOW(),
    last_seen TIMESTAMPTZ DEFAULT NOW(),
    UNIQUE(group_jid, phone)
);

-- ── Messages (partitioned by month) ──
CREATE TABLE wa_messages (
    id TEXT NOT NULL,                -- WhatsApp message ID
    group_jid TEXT NOT NULL,         -- Group JID
    sender_phone TEXT NOT NULL,      -- Sender phone number
    sender_name TEXT DEFAULT '',     -- Push name (display name)
    message_type TEXT NOT NULL DEFAULT 'conversation',
    content TEXT DEFAULT '',
    -- Full-text search vector, auto-generated from content
    content_search TSVECTOR GENERATED ALWAYS AS (
        to_tsvector('simple', coalesce(content, ''))
    ) STORED,
    media_type TEXT,                 -- image, audio, video, document
    media_path TEXT,                 -- Local file path after download
    media_thumb_path TEXT,           -- Thumbnail path
    media_mime TEXT,                 -- MIME type
    media_size INTEGER,             -- File size in bytes
    media_duration INTEGER,         -- Audio/video duration in seconds
    reply_to_id TEXT,               -- Quoted message ID
    timestamp TIMESTAMPTZ NOT NULL, -- Original WhatsApp timestamp
    raw_payload JSONB,              -- Full webhook payload for debugging
    created_at TIMESTAMPTZ DEFAULT NOW(),
    PRIMARY KEY (id, timestamp)     -- Composite PK required for partitioning
) PARTITION BY RANGE (timestamp);

-- Create partitions for current + next 3 months
-- (Replace 2026 with your current year)
CREATE TABLE wa_messages_2026_03 PARTITION OF wa_messages
    FOR VALUES FROM ('2026-03-01') TO ('2026-04-01');
CREATE TABLE wa_messages_2026_04 PARTITION OF wa_messages
    FOR VALUES FROM ('2026-04-01') TO ('2026-05-01');
CREATE TABLE wa_messages_2026_05 PARTITION OF wa_messages
    FOR VALUES FROM ('2026-05-01') TO ('2026-06-01');
CREATE TABLE wa_messages_2026_06 PARTITION OF wa_messages
    FOR VALUES FROM ('2026-06-01') TO ('2026-07-01');

-- Performance indexes
CREATE INDEX idx_wa_msg_group_ts ON wa_messages (group_jid, timestamp DESC);
CREATE INDEX idx_wa_msg_sender ON wa_messages (sender_phone, timestamp DESC);
CREATE INDEX idx_wa_msg_search ON wa_messages USING GIN (content_search);

-- ── AI Summary Cache ──
CREATE TABLE wa_daily_summaries (
    id SERIAL PRIMARY KEY,
    group_jid TEXT NOT NULL REFERENCES wa_groups(jid),
    summary_date DATE NOT NULL,
    message_count INTEGER DEFAULT 0,
    summary TEXT NOT NULL,
    model TEXT DEFAULT 'haiku',
    created_at TIMESTAMPTZ DEFAULT NOW(),
    UNIQUE(group_jid, summary_date)
);

-- ── Email Configuration (singleton row) ──
CREATE TABLE wa_email_config (
    id INTEGER PRIMARY KEY DEFAULT 1,
    smtp_host TEXT DEFAULT '',      -- 'resend' for Resend API mode
    smtp_port INTEGER DEFAULT 587,
    smtp_user TEXT DEFAULT '',
    smtp_pass TEXT DEFAULT '',      -- API key for Resend mode
    smtp_tls BOOLEAN DEFAULT true,
    from_email TEXT DEFAULT '',
    from_name TEXT DEFAULT 'WhatsApp Monitor',
    updated_at TIMESTAMPTZ DEFAULT NOW()
);

-- ── Scheduled Reports ──
CREATE TABLE wa_scheduled_reports (
    id SERIAL PRIMARY KEY,
    name TEXT NOT NULL,
    report_type TEXT NOT NULL DEFAULT 'daily_summary',
    schedule TEXT NOT NULL DEFAULT 'daily',      -- daily, weekly, monthly
    schedule_time TEXT DEFAULT '08:00',           -- HH:MM (UTC)
    schedule_day INTEGER DEFAULT 1,              -- Day of week (1=Mon) or month
    recipients TEXT NOT NULL DEFAULT '',          -- Comma-separated emails
    group_filter TEXT DEFAULT '',                 -- Comma-separated JIDs, empty = all
    include_stats BOOLEAN DEFAULT true,
    is_active BOOLEAN DEFAULT true,
    last_sent_at TIMESTAMPTZ,
    last_status TEXT DEFAULT '',
    created_at TIMESTAMPTZ DEFAULT NOW(),
    updated_at TIMESTAMPTZ DEFAULT NOW()
);

-- ── Webhook Debug Log ──
CREATE TABLE wa_webhook_log (
    id BIGSERIAL PRIMARY KEY,
    event_type TEXT NOT NULL,
    instance TEXT,
    payload JSONB NOT NULL,
    processed BOOLEAN DEFAULT false,
    created_at TIMESTAMPTZ DEFAULT NOW()
);
CREATE INDEX idx_wa_webhook_unprocessed
    ON wa_webhook_log (processed) WHERE NOT processed;

Por que essa estratégia de particionamento funciona

O particionamento de intervalo por mês é a escolha certa aqui porque:

1. As consultas são quase sempre limitadas pelo tempo. “Mostre-me as mensagens de ontem” ou “resuma a semana passada” apenas toque em 1-2 partições.
2. A retenção de dados é simples. Para soltar mensagens com mais de 12 meses: ALTER TABLE wa_messages DETACH PARTITION wa_messages_2025_01; DROP TABLE wa_messages_2025_01;
3. A coluna TSVECTOR gerada (content_search) é herdado por cada partição, dando-lhe pesquisa de texto completo com índices GIN que são partição-local (menor, mais rápido).
4. A chave primária composta (id, timestamp)é exigido pelo PostgreSQL — a chave de partição deve fazer parte de qualquer restrição única. Isso é bom porque os IDs de mensagem do WhatsApp são globalmente únicos dentro de uma janela de tempo razoável.

Auto-Criando Partições

O serviço cria partições na inicialização e via cron. Aqui está a lógica:

async def _ensure_partitions(conn):
    """Create partitions for current month + next 3 months."""
    now = datetime.utcnow()
    for i in range(4):
        m = now.month + i
        y = now.year + (m - 1) // 12
        m = ((m - 1) % 12) + 1
        nm = m + 1
        ny = y + (nm - 1) // 12
        nm = ((nm - 1) % 12) + 1
        part_name = f"wa_messages_{y}_{m:02d}"
        start = f"{y}-{m:02d}-01"
        end = f"{ny}-{nm:02d}-01"
        await conn.execute(f"""
            CREATE TABLE IF NOT EXISTS {part_name}
            PARTITION OF wa_messages
            FOR VALUES FROM ('{start}') TO ('{end}')
        """)

Isso é executado em todas as startups de serviços e no dia 25 de cada mês via cron, garantindo que a partição do próximo mês sempre exista antes de ser necessária.

---

6. Receptor de Webhook FastAPI

O receptor webhook é o núcleo do sistema. Ele recebe solicitações POST da Evolution API para cada evento (mensagens, atualizações de grupo, alterações de participantes, estado de conexão) e processa-as de forma assíncrona.

Estrutura de Serviço

#!/usr/bin/env python3
"""
WhatsApp Group Monitor -- FastAPI Service
Port 8086. Receives Evolution API webhooks, stores to PostgreSQL.
"""

import os
import re
import json
import time
import logging
import asyncio
from datetime import datetime, date, timedelta
from pathlib import Path
from typing import Optional

import asyncpg
import httpx
from fastapi import FastAPI, HTTPException, Query, BackgroundTasks, Request
from fastapi.responses import JSONResponse, FileResponse

# ── Config ──
PG_DSN = "postgresql://postgres:YOUR_PG_PASSWORD@POSTGRES_HOST:5432/whatsapp_monitor"
EVOLUTION_URL = "http://localhost:8080"
EVOLUTION_API_KEY = os.environ.get("EVOLUTION_API_KEY", "YOUR_EVO_KEY")
ANTHROPIC_KEY_FILE = "/opt/wa-monitor/.api_key"
MEDIA_DIR = Path("/opt/wa-monitor/media")
EVOLUTION_INSTANCE = "wa-monitor"
MEDIA_DIR.mkdir(parents=True, exist_ok=True)

logging.basicConfig(level=logging.INFO,
                    format="%(asctime)s %(levelname)s %(message)s")
log = logging.getLogger("wa-monitor")

app = FastAPI(title="WhatsApp Monitor", version="1.0")
pool: Optional[asyncpg.Pool] = None
_startup_time = time.time()

Conexão Pool e Startup

Usando asyncpgpara acesso async PostgreSQL com um pool de conexão:

@app.on_event("startup")
async def startup():
    global pool
    pool = await asyncpg.create_pool(PG_DSN, min_size=2, max_size=10)
    log.info("PostgreSQL pool created")
    await _ensure_schema()
    # Sync groups from Evolution API on startup
    asyncio.create_task(_sync_groups_from_evolution())
    # Start background watchdog
    asyncio.create_task(_watchdog_loop())


@app.on_event("shutdown")
async def shutdown():
    if pool:
        await pool.close()

O ponto final do Webhook

A decisão crítica do projeto: aceitar rápido, processar mais tarde. O ponto de extremidade do webhook registra a carga útil bruta e processa a mensagem em uma tarefa em segundo plano. Isso garante que a Evolution API nunca espere pela sua resposta:

@app.post("/webhook/evolution")
async def webhook_evolution(request: Request,
                            background_tasks: BackgroundTasks):
    """Receive Evolution API webhooks. Store raw, process in background."""
    try:
        body = await request.json()
    except Exception:
        raise HTTPException(400, "Invalid JSON")

    event = body.get("event", "unknown")
    instance = body.get("instance", "")

    # Log raw webhook for debugging
    async with pool.acquire() as conn:
        await conn.execute(
            """INSERT INTO wa_webhook_log (event_type, instance, payload)
               VALUES ($1, $2, $3)""",
            event, instance, json.dumps(body)
        )

    # Process in background -- return 200 immediately
    background_tasks.add_task(_process_webhook, event, instance, body)

    return {"status": "ok"}

Roteamento Webhook

Diferentes eventos obtêm diferentes manipuladores:

async def _process_webhook(event: str, instance: str, body: dict):
    """Route webhook events to appropriate handlers."""
    try:
        data = body.get("data", {})

        if event == "messages.upsert":
            await _process_message(instance, data)
        elif event == "groups.upsert":
            await _process_group_upsert(data)
        elif event == "group-participants.update":
            await _process_participant_update(data)
        elif event == "connection.update":
            state = data.get("state", "")
            log.info(f"Connection update: {instance} -> {state}")
        elif event == "qrcode.updated":
            log.info(f"QR code updated for {instance}")

    except Exception as e:
        log.error(f"Webhook processing error ({event}): {e}",
                  exc_info=True)

---

7. Pipeline de Processamento de Mensagens

Análise de tipos de mensagens do WhatsApp

As mensagens do WhatsApp vêm em muitos sabores. A carga útil do Evolution API webhook aninha o conteúdo de forma diferente para cada tipo. Veja como lidar com todos eles:

async def _process_message(instance: str, data: dict):
    """Parse an incoming message and store it."""
    key = data.get("key", {})
    msg_id = key.get("id", "")
    remote_jid = key.get("remoteJid", "")

    # Only process group messages (JIDs ending in @g.us)
    if not remote_jid.endswith("@g.us"):
        return

    # Extract sender
    participant = (key.get("participant", "")
                   or data.get("participant", ""))
    sender_phone = participant.split("@")[0] if participant else ""
    sender_name = data.get("pushName", "")

    # Extract message content based on type
    message = data.get("message", {})
    msg_type = "conversation"
    content = ""
    media_type = None
    media_mime = None

    if "conversation" in message:
        content = message["conversation"]
        msg_type = "conversation"
    elif "extendedTextMessage" in message:
        content = message["extendedTextMessage"].get("text", "")
        msg_type = "extendedText"
    elif "imageMessage" in message:
        msg_type = "image"
        media_type = "image"
        content = message["imageMessage"].get("caption", "")
        media_mime = message["imageMessage"].get("mimetype", "image/jpeg")
    elif "audioMessage" in message:
        msg_type = "audio"
        media_type = "audio"
        media_mime = message["audioMessage"].get("mimetype", "audio/ogg")
    elif "videoMessage" in message:
        msg_type = "video"
        media_type = "video"
        content = message["videoMessage"].get("caption", "")
        media_mime = message["videoMessage"].get("mimetype", "video/mp4")
    elif "documentMessage" in message:
        msg_type = "document"
        media_type = "document"
        fname = message["documentMessage"].get("fileName", "")
        content = f"[Document: {fname}]" if fname else "[Document]"
        media_mime = message["documentMessage"].get("mimetype", "")
    elif "stickerMessage" in message:
        msg_type = "sticker"
        content = "[Sticker]"
    elif "reactionMessage" in message:
        msg_type = "reaction"
        content = message["reactionMessage"].get("text", "")
    elif "contactMessage" in message:
        msg_type = "contact"
        display = message["contactMessage"].get("displayName", "")
        content = f"[Contact: {display}]"
    elif "locationMessage" in message:
        msg_type = "location"
        lat = message["locationMessage"].get("degreesLatitude", "")
        lon = message["locationMessage"].get("degreesLongitude", "")
        content = f"[Location: {lat}, {lon}]"
    else:
        msg_type = list(message.keys())[0] if message else "unknown"

    # Parse timestamp
    ts_epoch = data.get("messageTimestamp", 0)
    if isinstance(ts_epoch, str):
        ts_epoch = int(ts_epoch)
    ts = (datetime.utcfromtimestamp(ts_epoch)
          if ts_epoch else datetime.utcnow())

    # Extract reply context
    reply_to = None
    for msg_key in ("extendedTextMessage", "imageMessage", "videoMessage"):
        ctx = message.get(msg_key, {}).get("contextInfo", {})
        if ctx:
            reply_to = ctx.get("stanzaId")
            break

    # Ensure group exists in our database
    await _ensure_group(remote_jid)

    # Insert message (idempotent via ON CONFLICT)
    async with pool.acquire() as conn:
        await conn.execute("""
            INSERT INTO wa_messages
                (id, group_jid, sender_phone, sender_name, message_type,
                 content, media_type, media_mime, reply_to_id,
                 timestamp, raw_payload)
            VALUES ($1,$2,$3,$4,$5,$6,$7,$8,$9,$10,$11)
            ON CONFLICT (id, timestamp) DO NOTHING
        """, msg_id, remote_jid, sender_phone, sender_name, msg_type,
             content, media_type, media_mime, reply_to, ts,
             json.dumps(data))

    # Update participant tracking
    if sender_phone:
        await _upsert_participant(remote_jid, sender_phone, sender_name)

    # Download media in background
    if media_type in ("image", "audio", "video", "document"):
        asyncio.create_task(_download_and_process_media(
            instance, data, msg_id, remote_jid,
            media_type, media_mime, ts
        ))

    log.info(f"Stored: {msg_id} in {remote_jid} "
             f"from {sender_phone} ({msg_type})")

Rastreamento de Grupo e Participante

async def _ensure_group(jid: str, name: str = ""):
    """Insert group if not exists."""
    async with pool.acquire() as conn:
        await conn.execute("""
            INSERT INTO wa_groups (jid, name)
            VALUES ($1, $2)
            ON CONFLICT (jid) DO NOTHING
        """, jid, name)


async def _upsert_participant(group_jid: str, phone: str,
                               push_name: str = ""):
    """Track participant activity."""
    async with pool.acquire() as conn:
        await conn.execute("""
            INSERT INTO wa_participants
                (group_jid, phone, push_name, last_seen)
            VALUES ($1, $2, $3, NOW())
            ON CONFLICT (group_jid, phone) DO UPDATE SET
                push_name = CASE
                    WHEN EXCLUDED.push_name != ''
                    THEN EXCLUDED.push_name
                    ELSE wa_participants.push_name
                END,
                last_seen = NOW()
        """, group_jid, phone, push_name)

Download e processamento de mídia

Os arquivos de mídia são baixados da Evolution API, armazenados localmente em diretórios baseados em data e processados opcionalmente (thumbnais para imagens, extração de duração para áudio / vídeo):

async def _download_and_process_media(instance, data, msg_id,
                                       group_jid, media_type,
                                       media_mime, ts):
    """Download media via Evolution API, generate thumbnails."""
    try:
        key = data.get("key", {})
        message = data.get("message", {})

        # Date-based directory structure
        date_dir = MEDIA_DIR / group_jid / ts.strftime("%Y-%m-%d")
        date_dir.mkdir(parents=True, exist_ok=True)

        # Map MIME types to file extensions
        ext_map = {
            "image/jpeg": ".jpg", "image/png": ".png",
            "image/webp": ".webp", "audio/ogg": ".ogg",
            "audio/ogg; codecs=opus": ".ogg", "audio/mp4": ".m4a",
            "video/mp4": ".mp4", "application/pdf": ".pdf",
        }
        ext = ext_map.get(media_mime, ".bin")
        media_path = date_dir / f"{msg_id}{ext}"

        # Download via Evolution API's base64 endpoint
        async with httpx.AsyncClient(timeout=60) as client:
            resp = await client.post(
                f"{EVOLUTION_URL}/chat/getBase64FromMediaMessage"
                f"/{instance}",
                json={"message": {"key": key, "message": message}},
                headers={"apikey": EVOLUTION_API_KEY}
            )
            if resp.status_code != 200:
                log.warning(f"Media download failed: {msg_id}")
                return

            import base64
            b64_data = resp.json().get("base64", "")
            if not b64_data:
                return
            raw = base64.b64decode(b64_data)
            media_path.write_bytes(raw)

        # Update database with file path and size
        async with pool.acquire() as conn:
            await conn.execute("""
                UPDATE wa_messages
                SET media_path = $1, media_size = $2
                WHERE id = $3 AND timestamp = $4
            """, str(media_path), len(raw), msg_id, ts)

        log.info(f"Media saved: {msg_id} ({len(raw)} bytes)")

    except Exception as e:
        log.error(f"Media processing failed: {msg_id}: {e}")

---

8. Resumos diários movidos a IA

É aqui que o sistema se torna genuinamente valioso. Em vez de ler 300 mensagens do bate-papo da equipe de ontem, você recebe um resumo de 4 parágrafos destacando decisões, itens de ação e discussões-chave.

Integração com Claude API

def _get_api_key() -> str:
    """Read Anthropic API key from file."""
    key_file = Path(ANTHROPIC_KEY_FILE)
    if key_file.exists():
        key = key_file.read_text().strip()
        if key:
            return key
    raise RuntimeError("Anthropic API key not configured")


def _call_claude(messages: list, model: str = "claude-haiku-4-5-20251001",
                 max_tokens: int = 2000, system: str = "") -> str:
    """Call Claude API and return the response text."""
    import requests
    api_key = _get_api_key()
    body = {
        "model": model,
        "max_tokens": max_tokens,
        "messages": messages,
    }
    if system:
        body["system"] = system

    resp = requests.post(
        "https://api.anthropic.com/v1/messages",
        headers={
            "x-api-key": api_key,
            "content-type": "application/json",
            "anthropic-version": "2023-06-01",
        },
        json=body,
        timeout=120,
    )
    if resp.status_code != 200:
        err = resp.json().get("error", {}).get("message",
                                                resp.text[:300])
        raise RuntimeError(f"API error ({resp.status_code}): {err}")
    return resp.json()["content"][0]["text"].strip()

O Ponto Final de Geração de Resumo

@app.post("/summarize/{jid}")
async def summarize_group(jid: str, target_date: str = ""):
    """Generate or retrieve a daily summary for a group."""
    if not target_date:
        target_date = (date.today() - timedelta(days=1)).isoformat()

    summary_date = date.fromisoformat(target_date)

    # Check cache first
    async with pool.acquire() as conn:
        existing = await conn.fetchrow(
            """SELECT summary, message_count, model
               FROM wa_daily_summaries
               WHERE group_jid = $1 AND summary_date = $2""",
            jid, summary_date
        )
        if existing:
            return {
                "summary": existing["summary"],
                "message_count": existing["message_count"],
                "cached": True,
            }

        # Fetch messages for the target date
        messages = await conn.fetch("""
            SELECT sender_name, sender_phone, content,
                   message_type, timestamp
            FROM wa_messages
            WHERE group_jid = $1 AND timestamp::date = $2
            ORDER BY timestamp ASC
        """, jid, summary_date)

        if not messages:
            return {"summary": "No messages on this date.",
                    "message_count": 0, "cached": False}

        group = await conn.fetchrow(
            "SELECT name FROM wa_groups WHERE jid = $1", jid)
        group_name = group["name"] if group else jid

    # Build the prompt
    msg_text = []
    for m in messages:
        name = m["sender_name"] or m["sender_phone"]
        ts = m["timestamp"].strftime("%H:%M")
        content = m["content"] or f"[{m['message_type']}]"
        msg_text.append(f"[{ts}] {name}: {content}")

    prompt = f"""Summarize the following WhatsApp group messages \
from "{group_name}" on {target_date}.

Group by topic/thread. Highlight any decisions, action items, \
or important announcements.
Be concise (2-4 paragraphs max). If messages are in a \
non-English language, summarize in English.

Messages:
{chr(10).join(msg_text)}"""

    summary = _call_claude(
        [{"role": "user", "content": prompt}],
        model="claude-haiku-4-5-20251001",
        max_tokens=1000
    )

    # Cache the summary
    async with pool.acquire() as conn:
        await conn.execute("""
            INSERT INTO wa_daily_summaries
                (group_jid, summary_date, message_count, summary, model)
            VALUES ($1, $2, $3, $4, 'haiku')
            ON CONFLICT (group_jid, summary_date) DO UPDATE SET
                summary = EXCLUDED.summary,
                message_count = EXCLUDED.message_count
        """, jid, summary_date, len(messages), summary)

    return {"summary": summary, "message_count": len(messages),
            "cached": False}

Análise de IA Interativa

Além dos resumos diários, o sistema suporta perguntas de forma livre sobre conversas em grupo. Isso usa o Claude Sonnet para uma análise mais profunda:

WA_SYSTEM_PROMPT = """You are a WhatsApp group analysis assistant. \
You have access to WhatsApp group messages from staff groups.

Your job: Summarize discussions, identify key topics, flag important \
items (decisions, deadlines, complaints, requests), and answer \
questions about group conversations.

When analyzing messages:
- Group by topic/thread when summarizing
- Highlight action items and decisions
- Note who said what when relevant
- Flag urgent or important messages
- Be concise but thorough
- Use sender names when available, phone numbers when not
- Present information in a clear, structured format
- If messages are in a non-English language, provide analysis in English

Format your response in markdown."""


@app.post("/ask")
async def ask_ai(payload: dict):
    """AI analysis endpoint -- answers questions about messages."""
    question = payload.get("question", "").strip()
    group_jid = payload.get("group_jid", "")

    # Gather context: recent messages from relevant group(s)
    context_parts = []
    async with pool.acquire() as conn:
        if group_jid:
            messages = await conn.fetch("""
                SELECT sender_name, sender_phone, content,
                       message_type, timestamp
                FROM wa_messages
                WHERE group_jid = $1
                  AND timestamp >= CURRENT_DATE - INTERVAL '7 days'
                ORDER BY timestamp ASC LIMIT 500
            """, group_jid)
            for m in messages:
                name = m["sender_name"] or m["sender_phone"]
                ts = m["timestamp"].strftime("%Y-%m-%d %H:%M")
                content = m["content"] or f"[{m['message_type']}]"
                context_parts.append(f"[{ts}] {name}: {content}")

    context = "\n".join(context_parts)
    user_msg = (f"Here is the WhatsApp message data:\n\n{context}"
                f"\n\n---\n\nQuestion: {question}")

    answer = _call_claude(
        [{"role": "user", "content": user_msg}],
        model="claude-sonnet-4-6",
        max_tokens=3000,
        system=WA_SYSTEM_PROMPT
    )
    return {"answer": answer, "model": "claude-sonnet-4.6"}

Dicas de engenharia de prompt de IA

Obter bons resumos requer um cuidado cuidadoso. Aqui está o que funciona:

1. Sempre inclua timestamps. Sem eles, a IA não pode distinguir as discussões da manhã dos acompanhamentos da tarde.
2. Incluir nomes de remetentes. “Marco disse que precisamos contratar mais 3 agentes” é muito mais útil do que “Alguém disse que precisamos contratar”.
3. Solicitar agrupamento de tópicos. Sem essa instrução, a IA tende a produzir uma releitura cronológica em vez de um resumo temático.
4. Especifique a linguagem de saída. Se seus grupos usarem italiano, albanês ou espanhol, solicite explicitamente resumos em inglês.
5. Limite o orçamento simbólico. Haiku com max_tokens=1000produz resumos apertados. Ir mais alto tende a adicionar preenchimento.
6. Use Haiku para resumos em massa, Sonnet para análise interativa. Haiku é 10-20x mais barato e rápido o suficiente para trabalhos noturnos cron. O raciocínio mais profundo do Sonnet vale o custo para o “o que aconteceu” em tempo real? perguntas.

---

9. Relatórios de e-mail com Resend

Por que resender?

Característica	Reenviar	Brevo	Categoria: Mailjet
Nível livre	3.000/mês	300/dia	200/dia
Tempo de configuração	2 minutos	10 minutos	10 minutos
Complexidade da API	1 endpoint	Médio	Médio
KYC/verificação	Apenas e-mail	Email + telefone	E-mail
Domínio personalizado necessário?	Não (domínio de teste disponível)	Sim para a produção	Sim para a produção

O nível gratuito de Resend é perfeito para monitorar os relatórios. A API é um único ponto de extremidade POST. Nenhum SDK necessário – simples urllibobras.

E-mail Envio Implementação

O serviço suporta tanto a API REST do Resend quanto o SMTP tradicional, selecionado por configuração smtp_hostpara "resend":

def _send_email_sync(config: dict, to_list: list,
                     subject: str, html_body: str) -> str:
    """Send email. Returns empty string on success, error on failure."""
    import json as _json
    from urllib.request import Request, urlopen
    from urllib.error import HTTPError

    # ── Resend API mode ──
    if config.get("smtp_host", "").lower() == "resend":
        api_key = config.get("smtp_pass", "")
        if not api_key:
            return "Resend API key not configured"

        from_str = (f"{config.get('from_name', 'WhatsApp Monitor')} "
                    f"<{config.get('from_email', '[email protected]')}>")

        # Generate plain-text fallback by stripping HTML
        plain = re.sub(r'<[^>]+>', '', html_body)
        plain = re.sub(r'\s+', ' ', plain).strip()

        payload = _json.dumps({
            "from": from_str,
            "to": to_list,
            "subject": subject,
            "html": html_body,
            "text": plain,
        }).encode()

        req = Request("https://api.resend.com/emails",
                      data=payload, method="POST")
        req.add_header("Authorization", f"Bearer {api_key}")
        req.add_header("Content-Type", "application/json")

        try:
            resp = urlopen(req, timeout=30)
            data = _json.loads(resp.read())
            return "" if data.get("id") else f"Resend error: {data}"
        except HTTPError as e:
            body = e.read().decode()
            return f"Resend HTTP {e.code}: {body}"
        except Exception as e:
            return f"Resend error: {e}"

    # ── Traditional SMTP mode ──
    import smtplib
    from email.mime.multipart import MIMEMultipart
    from email.mime.text import MIMEText

    if not config.get("smtp_host"):
        return "Email not configured"

    msg = MIMEMultipart("alternative")
    msg["Subject"] = subject
    msg["From"] = (f"{config.get('from_name')} "
                   f"<{config.get('from_email')}>")
    msg["To"] = ", ".join(to_list)

    plain = re.sub(r'<[^>]+>', '', html_body)
    msg.attach(MIMEText(plain, "plain", "utf-8"))
    msg.attach(MIMEText(html_body, "html", "utf-8"))

    try:
        if config.get("smtp_tls", True):
            server = smtplib.SMTP(config["smtp_host"],
                                  config.get("smtp_port", 587),
                                  timeout=30)
            server.starttls()
        else:
            server = smtplib.SMTP(config["smtp_host"],
                                  config.get("smtp_port", 25),
                                  timeout=30)

        if config.get("smtp_user") and config.get("smtp_pass"):
            server.login(config["smtp_user"], config["smtp_pass"])

        server.sendmail(msg["From"], to_list, msg.as_string())
        server.quit()
        return ""
    except Exception as e:
        return str(e)

Modelo de e-mail HTML

Modelo de e-mail temático escuro profissional que renderiza bem entre os clientes:

def _build_email_html(title: str, sections: list) -> str:
    """Build HTML email. sections = [(heading, content_html), ...]"""
    sections_html = ""
    for heading, content in sections:
        sections_html += f'''
        <tr><td style="padding:0 24px 20px;">
            <h2 style="margin:0 0 8px;font-size:14px;color:#25D366;
                       font-family:'Courier New',monospace;
                       border-bottom:1px solid #2a2a2e;
                       padding-bottom:6px;">{heading}</h2>
            <div style="font-size:13px;line-height:1.7;
                        color:#d4d4d8;">{content}</div>
        </td></tr>'''

    return f'''<!DOCTYPE html><html><head>
    <meta charset="utf-8"></head>
    <body style="margin:0;padding:0;background:#0a0a0b;
                 font-family:-apple-system,sans-serif;">
    <table width="100%" cellpadding="0" cellspacing="0"
           style="background:#0a0a0b;padding:20px 0;">
    <tr><td align="center">
    <table width="600" cellpadding="0" cellspacing="0"
           style="background:#111113;border:1px solid #27272a;
                  border-radius:8px;overflow:hidden;">
        <tr><td style="background:#18181b;padding:16px 24px;
                       border-bottom:1px solid #27272a;">
            <span style="font-family:'Courier New',monospace;
                         font-size:15px;font-weight:bold;
                         color:#e4e4e7;">
                <span style="color:#25D366;">&#9679;</span> {title}
            </span>
        </td></tr>
        {sections_html}
        <tr><td style="padding:16px 24px;
                       border-top:1px solid #27272a;
                       text-align:center;">
            <span style="font-size:10px;color:#52525b;
                         font-family:'Courier New',monospace;">
                WhatsApp Monitor -- Automated Report
            </span>
        </td></tr>
    </table>
    </td></tr></table>
    </body></html>'''

Tipos de Relatório

O sistema suporta três tipos de relatórios integrados:

1. 1. Resumo Diário (daily_summary)

• Busca resumos gerados por IA (de wa_daily_summaries) para cada grupo
• Recua às visualizações de mensagens brutas se não houver resumo
• Mostra a contagem de mensagens por grupo

2. 2. Digest Semanal (weekly_digest)

• Agrega 7 dias de dados: mensagens totais, participantes únicos, remetentes de topo
• Inclui destaques de resumos diários
• Bom para relatórios de gestão

3. Alerta de Atividades (activity_alert)

• Listas de grupos ativos vs. inativos
• Bandeiras grupos com zero mensagens hoje
• Útil para gestores comunitários que precisam identificar grupos desengajados

Motor de geração de relatórios

async def _send_single_report(report: dict) -> dict:
    """Generate and send a single report."""
    config = await _get_email_config()
    if not config.get("smtp_host"):
        return {"status": "error", "detail": "SMTP not configured"}

    recipients = [r.strip()
                  for r in report["recipients"].split(",")
                  if r.strip()]
    report_type = report["report_type"]

    sections = []
    async with pool.acquire() as conn:
        groups = await conn.fetch(
            "SELECT jid, name FROM wa_groups "
            "WHERE is_monitored = true")

        if report_type == "daily_summary":
            yesterday = (date.today() - timedelta(days=1)).isoformat()
            for g in groups:
                summary = await conn.fetchrow(
                    """SELECT summary, message_count
                       FROM wa_daily_summaries
                       WHERE group_jid = $1
                         AND summary_date = $2""",
                    g["jid"], date.fromisoformat(yesterday))

                if summary and summary["message_count"] > 0:
                    content = summary["summary"].replace("\n", "<br>")
                    content = (f'<p style="color:#a1a1aa;">'
                               f'{summary["message_count"]} messages'
                               f'</p>' + content)
                    sections.append((g["name"] or g["jid"], content))

            subject = f"WhatsApp Daily Summary -- {yesterday}"
            title = f"Daily Summary -- {yesterday}"

        # ... similar logic for weekly_digest and activity_alert

    html = _build_email_html(title, sections)

    err = await asyncio.get_event_loop().run_in_executor(
        None, _send_email_sync, config, recipients, subject, html)

    # Track send status
    async with pool.acquire() as conn:
        status = "sent" if not err else f"error: {err}"
        await conn.execute(
            """UPDATE wa_scheduled_reports
               SET last_sent_at = NOW(), last_status = $2
               WHERE id = $1""",
            report["id"], status)

    if err:
        return {"status": "error", "detail": err}
    return {"status": "ok", "sections": len(sections)}

Relatório CRUD API

API REST completa para gerenciamento de relatórios no painel:

GET    /reports                 -- List all scheduled reports
POST   /reports                 -- Create a new report
PUT    /reports/{id}            -- Update a report
DELETE /reports/{id}            -- Delete a report
POST   /reports/{id}/send       -- Send a report immediately
POST   /cron/send-reports       -- Check and send all due reports

---

10. Interface do painel (PHP)

O painel fornece quatro páginas PHP que trabalham em conjunto:

whatsapp.php — Monitor do Grupo

A interface principal para visualização de grupos e mensagens. Características:

• Lista de grupos com estatísticas ao vivo (mensagens hoje, esta semana, última mensagem)
• Visualizador de mensagens com paginação, filtragem de remetente, seleção de intervalo de datas
• Pesquisa de texto completo em todos os grupos (alimentada pelo PostgreSQL tsvector)
• Painel de bate-papo de IA – faça perguntas sobre as conversas de qualquer grupo
• Status da conexão – mostra se o WhatsApp está conectado, com exibição de código QR para reconexão

email-reports.php — Gestão de Relatórios

Uma interface CRUD autônoma para relatórios de e-mail. Características:

• SMTP/Resend painel de configuração — formulário dobrável para configurações do provedor de e-mail
• Cartões de relatório – cada relatório mostra o cronograma, os destinatários, o status do último envio e o interruptor de alternância
• Formulário modal — crie/edite relatórios com tipo, horário, dia, destinatários, filtro de grupo
• Enviar agora botão — acione manualmente qualquer relatório
• Teste e-mail — verificar configuração SMTP

whatsapp-data.php — AJAX ProxyTradução

Um proxy PHP fino que encaminha solicitações do navegador para o serviço FastAPI. Isso resolve dois problemas:

1. CORS – o navegador não pode ligar diretamente localhost:8086a partir de uma página servida no porto 8082.
2. Autenticação – o proxy impõe acesso somente para administradores usando o sistema de autenticação PHP existente.

// Simplified flow
$endpoint = $_GET['endpoint'];  // e.g., "/groups" or "/reports/5"
$url = "http://127.0.0.1:8086{$endpoint}";
$response = curl_exec($url);    // Forward request
echo $response;                  // Return response

O proxy lista prefixos de endpoint específicos para impedir o acesso não autorizado a APIs internas.

whatsapp-ask.php — Análise de IA Proxy

Proxy dedicado para o /askendpoint, encaminhando perguntas do usuário para Claude através do serviço FastAPI. Suporta conversas de várias voltas passando o histórico completo de conversa.

---

11. Ponto Final de Métricas Prometheus

O /metricsendpoint expõe métricas compatíveis com Prometheus para monitoramento de Grafana:

@app.get("/metrics")
async def prometheus_metrics():
    """Prometheus-compatible metrics for Grafana."""
    lines = []

    lines.append("# HELP wa_monitor_up Service status")
    lines.append("# TYPE wa_monitor_up gauge")
    wa_up = 1 if _watchdog_stats.get("last_state") == "open" else 0
    lines.append(f"wa_monitor_up {wa_up}")

    lines.append("# HELP wa_monitor_uptime_seconds Uptime")
    lines.append("# TYPE wa_monitor_uptime_seconds gauge")
    lines.append(f"wa_monitor_uptime_seconds "
                 f"{int(time.time() - _startup_time)}")

    lines.append("# HELP wa_monitor_reconnects_total Reconnects")
    lines.append("# TYPE wa_monitor_reconnects_total counter")
    lines.append(f'wa_monitor_reconnects_total '
                 f'{_watchdog_stats["reconnects"]}')

    async with pool.acquire() as conn:
        today = await conn.fetchval(
            "SELECT COUNT(*) FROM wa_messages "
            "WHERE timestamp >= CURRENT_DATE")
        total = await conn.fetchval(
            "SELECT COUNT(*) FROM wa_messages")
        groups = await conn.fetchval(
            "SELECT COUNT(*) FROM wa_groups "
            "WHERE is_monitored = true")

    lines.append(f"wa_monitor_messages_today {today}")
    lines.append(f"wa_monitor_messages_total {total}")
    lines.append(f"wa_monitor_groups {groups}")

    from fastapi.responses import PlainTextResponse
    return PlainTextResponse(
        "\n".join(lines) + "\n",
        media_type="text/plain; version=0.0.4")

Prometheus scrape config:

# In prometheus.yml
scrape_configs:
  - job_name: 'wa-monitor'
    scrape_interval: 30s
    static_configs:
      - targets: ['172.18.0.1:8086']

Painéis úteis de Grafana:

• wa_monitor_up— alertar se o WhatsApp se desconectar
• rate(wa_monitor_messages_today)— velocidade da mensagem
• wa_monitor_reconnects_total— estabilidade da conexão
• wa_monitor_groups— contagem de grupos ao longo do tempo

---

12. Cron Empregos e Automação

Quatro trabalhos cron mantêm o sistema funcionando sem intervenção humana:

# ── Daily AI Summaries (2 AM) ──
# Generates summaries for all groups for yesterday
0 2 * * * curl -s -X POST http://localhost:8086/cron/daily-summaries \
    > /dev/null 2>&1

# ── Partition Creation (25th of each month, 3 AM) ──
# Creates PostgreSQL partitions for the next 3+ months
0 3 25 * * curl -s -X POST http://localhost:8086/cron/create-partitions \
    > /dev/null 2>&1

# ── Media Cleanup (Sundays at 4 AM) ──
# Deletes downloaded media files older than 90 days
0 4 * * 0 find /opt/wa-monitor/media -type f -mtime +90 -delete \
    2>/dev/null; \
    find /opt/wa-monitor/media -type d -empty -delete 2>/dev/null

# ── Report Scheduler (every 15 minutes) ──
# Checks for due reports and sends them
*/15 * * * * curl -s -X POST http://localhost:8086/cron/send-reports \
    > /dev/null 2>&1

Como funciona o agendador de relatórios

O /cron/send-reportsendpoint roda a cada 15 minutos. Para cada relatório ativo:

1. Analisar o schedule_time(por exemplo, “08:00”) e verifique se a hora atual está dentro de uma janela de 15 minutos após o horário agendado.
2. Verifique o schedule tipo:
   • daily— enviar todos os dias
   • weekly— enviar apenas sobre a correspondência schedule_day(1=Segunda-feira a 7=Domingo)
   • monthly— enviar apenas quando o dia atual do mês corresponder schedule_day
3. Verifica last_sent_at— pular se já enviado hoje (impede duplicatas de sobreposição de cron runs).
4. Gerar o conteúdo do relatório, enviar o e-mail e atualizar last_sent_ate last_status.

@app.post("/cron/send-reports")
async def cron_send_reports():
    """Check and send any due reports."""
    now = datetime.utcnow()
    results = []

    async with pool.acquire() as conn:
        reports = await conn.fetch(
            "SELECT * FROM wa_scheduled_reports "
            "WHERE is_active = true")

    for r in reports:
        report = dict(r)
        last_sent = report.get("last_sent_at")

        # Already sent today? Skip.
        if last_sent and last_sent.date() == now.date():
            continue

        # Parse schedule time
        sched_h, sched_m = map(int,
            report.get("schedule_time", "08:00").split(":"))

        # Check 15-minute window
        now_min = now.hour * 60 + now.minute
        sched_min = sched_h * 60 + sched_m
        if now_min < sched_min or now_min > sched_min + 15:
            continue

        # Check schedule type
        should_send = False
        if report["schedule"] == "daily":
            should_send = True
        elif report["schedule"] == "weekly":
            should_send = (now.isoweekday() == report["schedule_day"])
        elif report["schedule"] == "monthly":
            should_send = (now.day == report["schedule_day"])

        if should_send:
            result = await _send_single_report(report)
            results.append({"report": report["name"], **result})

    return {"sent": results}

---

13. Conexão Watchdog & Auto-Reconnect

As conexões do WhatsApp via Baileys não são permanentes. A bateria do telefone morre, o Wi-Fi cai, o WhatsApp envia atualizações de protocolo. O cão de guarda é executado a cada 60 segundos e lida com tudo isso automaticamente:

_watchdog_stats = {
    "checks": 0,
    "reconnects": 0,
    "last_check": None,
    "last_state": "unknown"
}


async def _watchdog_loop():
    """Background watchdog: auto-reconnect, sync groups, cleanup."""
    await asyncio.sleep(10)  # Let startup finish
    log.info("Watchdog started")

    while True:
        try:
            _watchdog_stats["checks"] += 1
            _watchdog_stats["last_check"] = (
                datetime.utcnow().isoformat())

            # 1. Check WhatsApp connection
            async with httpx.AsyncClient(timeout=10) as client:
                resp = await client.get(
                    f"{EVOLUTION_URL}/instance/connectionState"
                    f"/{EVOLUTION_INSTANCE}",
                    headers={"apikey": EVOLUTION_API_KEY})

                if resp.status_code == 200:
                    state = (resp.json().get("instance", {})
                             .get("state", "unknown"))
                    _watchdog_stats["last_state"] = state

                    if state in ("close", "connecting"):
                        # Disconnected -- attempt reconnect
                        log.warning(f"WhatsApp disconnected "
                                    f"(state={state}), reconnecting...")
                        await client.get(
                            f"{EVOLUTION_URL}/instance/connect"
                            f"/{EVOLUTION_INSTANCE}",
                            headers={"apikey": EVOLUTION_API_KEY})
                        _watchdog_stats["reconnects"] += 1

                elif resp.status_code == 404:
                    # Instance was deleted -- recreate it
                    log.warning("Instance missing, recreating...")
                    await client.post(
                        f"{EVOLUTION_URL}/instance/create",
                        headers={
                            "apikey": EVOLUTION_API_KEY,
                            "Content-Type": "application/json"},
                        json={
                            "instanceName": EVOLUTION_INSTANCE,
                            "integration": "WHATSAPP-BAILEYS",
                            "qrcode": True,
                            "webhook": {
                                "url": ("http://172.18.0.1:8086"
                                        "/webhook/evolution"),
                                "byEvents": False,
                                "base64": True,
                                "events": [
                                    "MESSAGES_UPSERT",
                                    "GROUPS_UPSERT",
                                    "GROUP_PARTICIPANTS_UPDATE",
                                    "CONNECTION_UPDATE",
                                    "QRCODE_UPDATED"
                                ]
                            }
                        })
                    _watchdog_stats["reconnects"] += 1

            # 2. Periodic group sync (every 6 hours)
            # Refreshes group names and participant counts
            # from Evolution API

            # 3. Webhook log cleanup (30-day retention)
            async with pool.acquire() as conn:
                await conn.execute(
                    "DELETE FROM wa_webhook_log "
                    "WHERE created_at < NOW() - INTERVAL '30 days'")

        except asyncio.CancelledError:
            return
        except Exception as e:
            log.error(f"Watchdog error: {e}")

        await asyncio.sleep(60)

---

14. Dicas de Produção & Lições Aprendidas

Manipulação de inundações de Webhook

Quando você se conecta pela primeira vez a uma conta do WhatsApp que vem acumulando mensagens offline, a Evolution API pode despejar centenas de mensagens históricas como webhooks em rápida sucessão. Isso pode sobrecarregar seu banco de dados se você não tiver cuidado.

Mitigações:

1. Processamento de tarefas em segundo plano. Nunca processe webhooks no manipulador de solicitações. Retorno 200 OKimediatamente e processar de forma assíncrona.
2. ON CONFLICT DO NOTHING. Cada INSERT usa esse padrão, tornando a ingestão de mensagens totalmente idempotente. Webhooks duplicados são silenciosamente ignorados.
3. Dimensionamento do pool de conexão. min_size=2, max_size=10lida com rajadas sem esgotar as conexões PostgreSQL.
4. Taxa-limite downloads de mídia. O download de mídia é a parte mais lenta. Uso asyncio.create_task()assim, ele é executado em segundo plano e não bloqueia o processamento de mensagens.

Manutenção de Partição

• Sempre crie partições antes do tempo. Se uma mensagem chegar por um mês sem uma partição, o PostgreSQL levanta um erro e a mensagem é perdida. O código de inicialização cria 4 meses à frente.
• Execute a criação de partição no dia 25, não no 1o. Isso lhe dá 5 dias de buffer no caso de o cron falhar.
• Para soltar dados antigos, desprenda a partição primeiro e, em seguida, solte:

-- Remove January 2025 data
ALTER TABLE wa_messages DETACH PARTITION wa_messages_2025_01;
DROP TABLE wa_messages_2025_01;

Armazenamento de Mídia

• Use diretórios baseados em data (/media/{group_jid}/2026-03-12/) para manter o sistema de arquivos gerenciável.
• Defina uma política de retenção. O trabalho cron exclui mídia com mais de 90 dias. As mensagens de texto permanecem para sempre no PostgreSQL; apenas os arquivos binários são limpos.
• Monitore o uso do disco. Um único grupo ocupado que envia fotos pode consumir 500MB + por mês. Imagens média de 100-300KB cada; notas de voz 50-200KB; vídeos 2-10MB.

Reenviar Limites De Nível Gratuitos

• 3.000 e-mails por mês, 100 por dia.
• O [email protected]o remetente de teste funciona sem verificação de domínio, mas pode chegar em spam para alguns provedores.
• Para produção: Adicione seu próprio domínio em Resend (a verificação de DNS leva 5 minutos) e, em seguida, atualize o from_emailno painel.
• Cada e-mail de relatório conta como 1 envio por destinatário. Um relatório diário para 3 pessoas = 3 envios = 90/mês.

Considerações de segurança

1. Armazene a chave da API Antrópica em um arquivo (/opt/wa-monitor/.api_key), não em variáveis de ambiente. Os arquivos são menos propensos a vazar em registros do Docker ou listagens de processos.
2. O ponto final do webhook não tem autenticação. Isso é intencional porque a Evolution API não suporta assinatura de webhook. Mitigar por porta de ligação 8086 para localhost (--host 127.0.0.1) se a Evolution API estiver na mesma máquina.
3. O proxy PHP reforça o auth de administrador. Todas as solicitações de painel para o serviço FastAPI passam por whatsapp-data.php, que verifica as permissões de administrador.
4. As senhas SMTP são mascaradas na API. O GET /settings/emailendpoint retorna "********"em vez da senha real. As atualizações preservam a senha existente se o valor mascarado for enviado de volta.

Evolução API Estabilidade

• Fixar a versão em docker-compose (v2.3.7, não latest). Baileys é um protocolo com engenharia inversa; atualizações podem quebrar as coisas.
• Espere desconexão. WhatsApp ocasionalmente força a re-autenticação. O cão de guarda lida com isso, mas você pode precisar digitalizar novamente um código QR a cada poucas semanas.
• Não utilize o número de monitorização para mensagens pessoais. A Evolution API expõe todas as mensagens a webhooks, incluindo chats privados. Use um número de telefone dedicado.
• Redis maxmemorypolítica de allkeys-lruimpede a exaustão da memória. Definido para 256MB para uma implantação típica.

Desempenho em escala

Para referência, esta arquitetura lida confortavelmente:

Métrica	Capacidade
Grupos monitorados	50+
Mensagens por dia	10.000+
Processamento de webhook simultâneo	Limitado por pool de assíncopg (10 conexões)
Velocidade de pesquisa de texto completo	<100ms para a maioria das consultas (índice GIN)
Geração de resumo de IA	~ 5 segundos por grupo por dia (Haiku)
Geração de relatório de e-mail	<10 segundos incluindo o fallback da IA

---

15. Referência Completa de Arquivos

Arquivos de serviço

Arquivo	Propósito
/opt/wa-monitor/service.py	Aplicação FastAPI principal (receptor de webhook, endpoints de API, IA, e-mail, cão de guarda)
/opt/wa-monitor/schema.sql	Esquema PostgreSQL (execute uma vez para a configuração inicial)
/opt/wa-monitor/.api_key	Chave de API antrópica (texto simples, chmod 600)
/opt/wa-monitor/media/	Arquivos de mídia baixados (data-organizado)
/opt/wa-monitor/venv/	Ambiente virtual Python
/opt/wa-monitor/service.log	Logs de aplicação (stdout + stderr do systemd)

Arquivos do Dashboard

Arquivo	Propósito
whatsapp.php	Monitor de grupo IU — procure grupos, mensagens, chat de IA
whatsapp-data.php	AJAX proxy – encaminha solicitações de navegador para FastAPI
whatsapp-ask.php	Proxy de análise de IA – encaminha perguntas para Claude
email-reports.php	Interface do usuário do gerenciamento de relatórios — Configuração SMTP, relatório CRUD

Arquivos de configuração

Arquivo	Propósito
docker-compose.yml	Docker stack (PostgreSQL, Evolution API, Redis) (em inglês)
.env	Variáveis de ambiente (senhas, chaves de API)
/etc/systemd/system/wa-monitor.service	Unidade sistematada para serviço FastAPI

Unidade Systemd

[Unit]
Description=WhatsApp Monitor Service (FastAPI)
After=network.target docker.service

[Service]
Type=simple
User=root
WorkingDirectory=/opt/wa-monitor
ExecStart=/opt/wa-monitor/venv/bin/python3 -m uvicorn \
    service:app --host 0.0.0.0 --port 8086 --workers 1
Restart=always
RestartSec=5
StandardOutput=append:/opt/wa-monitor/service.log
StandardError=append:/opt/wa-monitor/service.log
Environment=PYTHONUNBUFFERED=1

[Install]
WantedBy=multi-user.target

Python Dependências

# requirements.txt
fastapi>=0.104.0
uvicorn>=0.24.0
asyncpg>=0.29.0
httpx>=0.25.0
requests>=2.31.0
Pillow>=10.0.0           # Image thumbnails
faster-whisper>=0.9.0    # Audio transcription (optional)

Referência de Endpoints de API

Método	Ponto final	Propósito
POSTAGEM	/webhook/evolution	Receber Webhooks da Evolution API
OBTER	/health	Verificação de saúde com status de DB, WhatsApp e cão de guarda
OBTER	/metrics	Métricas do Prometheus
OBTER	/groups	Liste todos os grupos monitorados com estatísticas
OBTER	/groups/{jid}/messages	Mensagens paginadas com filtros
OBTER	/groups/{jid}/stats	Principais remetentes, atividade horária/diária
POSTAGEM	/groups/sync	Força a sincronização de grupo da Evolution API
OBTER	/search?q=term	Pesquisa de texto completo em todos os grupos
OBTER	/media/{path}	Servir arquivos de mídia baixados
POSTAGEM	/ask	Resposta de perguntas com inteligência artificial
POSTAGEM	/summarize/{jid}	Gerar/recuperar resumo diário
OBTER	/settings/email	Obter configuração de e-mail
POSTAGEM	/settings/email	Salvar configuração de e-mail
POSTAGEM	/settings/email/test	Enviar e-mail de teste
OBTER	/reports	Lista relatórios agendados
POSTAGEM	/reports	Criar um relatório
COLOCAR	/reports/{id}	Atualizar um relatório
EXCLUIR	/reports/{id}	Eliminar um relatório
POSTAGEM	/reports/{id}/send	Envie um relatório imediatamente
OBTER	/connection/status	Estado de conexão do WhatsApp
OBTER	/connection/qrcode	Obter código QR para vinculação
POSTAGEM	/connection/create	Criar/reconectar instância
POSTAGEM	/cron/daily-summaries	Gerar todos os resumos diários
POSTAGEM	/cron/create-partitions	Crie as próximas partições
POSTAGEM	/cron/send-reports	Verificar e enviar relatórios de vencimento

---

Lista de verificação de início rápido

1. [ ] Instalar o Docker, Python 3.11+, ffmpeg
2. [ ] Criar docker-compose.ymlcom PostgreSQL, Evolution API, Redis
3. [ ] Criar .envcom POSTGRES_PASSWORDe EVOLUTION_API_KEY
4. [ ] docker compose up -d
5. [ ] Criar o whatsapp_monitorbanco de dados e executar schema.sql
6. [ ] Configurar o Python venv e instalar dependências
7. [ ] Salve sua chave de API Antrópica para /opt/wa-monitor/.api_key
8. [ ] Implantar service.pye instalar a unidade systemd
9. [ ] systemctl enable --now wa-monitor
10. [ ] Criar a instância da API Evolution e digitalizar o código QR
11. [ ] Configurar a chave da API de Reenvio no painel (shost SMTP = resend)
12. [ ] Crie seu primeiro relatório agendado
13. [ ] Adicionar cron empregos para resumos diários, criação de partições, limpeza de mídia, envio de relatórios
14. [ ] Adicionar Prometheus alvo de raspão para /metrics
15. [ ] Aguarde que as mensagens fluam e aproveite seu e-mail de resumo da manhã

---

Construído com Evolution API, FastAPI, PostgreSQL, Claude AI e Resend. Nenhuma API oficial do WhatsApp Business é necessária.