pywhatsweb
Advanced tools
Biblioteca Python para WhatsApp Web corporativo com sidecar Python integrado (padrão) ou Node.js (opcional)
PyWhatsWeb é uma biblioteca Python moderna para integração corporativa com WhatsApp Web. Diferente de soluções baseadas em Selenium, ela utiliza uma arquitetura sidecar Python integrado (padrão) ou sidecar Node.js (opcional) para operação headless e eventos em tempo real.
✅ Multi-sessão: Gerencia múltiplas sessões WhatsApp simultaneamente
✅ Eventos em tempo real: WebSocket para QR, mensagens, conexão, etc.
✅ Storage pluggable: FileSystem ou Django ORM (opcional)
✅ Sistema Kanban: Status NEW/ACTIVE/DONE para conversas
✅ API REST: Comunicação HTTP com sidecar
✅ Sem navegador: Operação completamente headless
✅ Multi-framework: Funciona em Django, Flask, FastAPI, scripts Python
✅ Reconexão automática: WebSocket com backoff exponencial
✅ Idempotência: Sistema de deduplicação e rastreamento
✅ Health & Métricas: Monitoramento completo do sistema
✅ Segurança: API key, CORS, rate limiting, logs seguros
✅ Exemplo Django completo: Dashboard funcional com WebSocket
✅ Autenticação WebSocket obrigatória: Token obrigatório + multi-tenant
✅ Verificação de compatibilidade: Fail-fast no boot (Python + qrcode[pil])
✅ Anti-loop: Sistema de origem de mensagens (inbound/outbound)
✅ Normalização E.164: Validação e formatação de números de telefone
✅ Catálogo de erros padronizados: 50+ códigos com HTTP status e ações
✅ Métricas Prometheus: Endpoint /metrics com histogramas e percentis
✅ Docker + Orquestração: Dockerfile otimizado + docker-compose completo
✅ Runbooks de operação: Procedimentos para cenários críticos em produção
✅ Python 3.13: Suporte completo e compatibilidade total
✅ Dataclasses corrigidas: Ordem de campos compatível com Python 3.13+
✅ Verificação automática: Compatibilidade detectada no import
✅ Configurações centralizadas: Todas as configurações em um local
✅ Auto-start do sidecar: Inicia automaticamente sem intervenção manual
✅ Cleanup automático: Gerencia ciclo de vida completo do sidecar
✅ Compatibilidade Windows: Suporte nativo para Windows e Linux/Mac
✅ Sidecar Python integrado: Opção de usar Python puro sem Node.js (PADRÃO)
✅ 100% Python nativo: Sem dependências externas por padrão
qrcode[pil] integradasidecar_type="python" ou sidecar_type="nodejs"TypeError: non-default argument follows default argumentMediaMessage da classe Message__post_init__ funcionando corretamenteqrcode[pil]1. INICIALIZAÇÃO
├── Python cria WhatsWebManager
├── Manager conecta com sidecar Node.js
└── Sidecar inicia Puppeteer + whatsapp-web.js
2. CRIAÇÃO DE SESSÃO
├── Manager.create_session() gera ID único
├── Sidecar cria instância WhatsApp
└── WebSocket estabelece conexão em tempo real
3. AUTENTICAÇÃO
├── Sidecar gera QR Code
├── WebSocket envia evento 'qr' para Python
├── Usuário escaneia QR no celular
└── Evento 'ready' confirma autenticação
4. OPERAÇÃO
├── Eventos 'message' chegam via WebSocket
├── Python processa com handlers @session.on()
├── Envio via session.send_text() → HTTP API
└── Sidecar envia via whatsapp-web.js
DISCONNECTED → CONNECTING → CONNECTED → READY → [OPERATION] → DISCONNECTED
↑ ↓
└─────────────── RECONEXÃO AUTOMÁTICA ←───────────────────────────┘
┌─────────────────────────────────────────────────────────────────┐
│ SEU APP PYTHON │
│ ┌─────────────────┐ ┌─────────────────┐ ┌─────────────┐ │
│ │ WhatsWebManager │ │ Session │ │ Storage │ │
│ │ (Core) │ │ (Individual) │ │ (Pluggable) │ │
│ └─────────────────┘ └─────────────────┘ └─────────────┘ │
└─────────────────────────────────────────────────────────────────┘
│
│ HTTP + WebSocket
▼
┌─────────────────────────────────────────────────────────────────┐
│ SIDECAR NODE.JS │
│ ┌─────────────────┐ ┌─────────────────┐ ┌─────────────┐ │
│ │ Express API │ │ WebSocket WS │ │whatsapp-web.│ │
│ │ (REST) │ │ (Events) │ │ js │ │
│ └─────────────────┘ └─────────────────┘ └─────────────┘ │
└─────────────────────────────────────────────────────────────────┘
pywhatsweb-lib/
├── 📂 pywhatsweb/ # 🐍 CORE PYTHON (biblioteca principal)
│ ├── __init__.py # 🚪 Ponto de entrada e imports
│ ├── manager.py # 🎮 Gerenciador de sessões
│ ├── session.py # 💬 Sessão WhatsApp individual
│ ├── enums.py # 📋 Enumerações e constantes
│ ├── models.py # 🗃️ Modelos de dados
│ ├── exceptions.py # 🚨 Exceções customizadas
│ └── storage/ # 💾 Sistema de persistência
│ ├── base.py # 🔌 Interface base
│ ├── filesystem.py # 📁 Storage em arquivo (padrão)
│ └── django.py # �� Storage Django ORM (opcional)
├── 📂 sidecar/ # 🟢 SIDECAR NODE.JS (servidor)
│ ├── src/server.js # 🖥️ Servidor principal
│ ├── package.json # 📦 Dependências Node.js
│ └── Dockerfile # 🐳 Container Docker
├── 📂 examples/ # 💡 EXEMPLOS DE USO
│ ├── basic_usage.py # 🚀 Uso básico
│ └── django_complete.py # 🎯 Exemplo Django completo
├── 📂 tests/ # 🧪 TESTES AUTOMATIZADOS
├── 📂 whatsapp_data/ # 💾 DADOS SALVOS (gerado automaticamente)
├── 📄 README.md # 📖 Documentação para usuários
├── 📄 README-for-devs.md # 🔧 Este arquivo (guia técnico)
├── 📄 requirements.txt # 🐍 Dependências Python
└── 📄 docker-compose.yml # 🐳 Orquestração Docker
pywhatsweb/utils.py - UTILITÁRIOS E VALIDAÇÕESfrom pywhatsweb.utils import normalize_phone_number, validate_phone_number
# Normalizar número para formato E.164
phone = normalize_phone_number("11999999999") # +5511999999999
is_valid, normalized = validate_phone_number("11999999999")
# Informações detalhadas do número
from pywhatsweb.utils import get_phone_info, format_phone_for_display
info = get_phone_info("+5511999999999")
# {
# 'number': '+5511999999999',
# 'country_code': 55,
# 'region': 'BR',
# 'carrier': 'Vivo',
# 'timezone': 'America/Sao_Paulo'
# }
# Formatação para exibição
display = format_phone_for_display("+5511999999999", "NATIONAL") # (11) 99999-9999
from pywhatsweb.utils import generate_session_id, validate_session_id
# Gerar ID único para sessão
session_id = generate_session_id("session", "tenant_123") # session_tenant_123_1703000000_abc12345
# Validar formato
is_valid = validate_session_id("session_123") # True/False
from pywhatsweb.utils import get_current_timezone, format_timestamp
current_tz = get_current_timezone() # 'America/Sao_Paulo'
formatted = format_timestamp(datetime.now(), 'America/Sao_Paulo') # '2024-12-19 10:30:00 BRT'
pywhatsweb/errors.py - CATÁLOGO DE ERROS PADRONIZADOSfrom pywhatsweb.errors import ErrorCode, ErrorSeverity, get_error_info, create_error_response
# Códigos de erro disponíveis
ErrorCode.E_SESSION_NOT_FOUND # 404 - Sessão não encontrada
ErrorCode.E_AUTHENTICATION_FAILED # 401 - Falha na autenticação
ErrorCode.E_MEDIA_TOO_LARGE # 400 - Arquivo muito grande
ErrorCode.E_WHATSAPP_2FA_REQUIRED # 400 - 2FA necessário
ErrorCode.E_WHATSAPP_BANNED # 403 - WhatsApp banido
# Severidades
ErrorSeverity.LOW # Aviso, não afeta funcionalidade
ErrorSeverity.MEDIUM # Erro recuperável
ErrorSeverity.HIGH # Erro crítico, afeta funcionalidade
ErrorSeverity.CRITICAL # Erro fatal, sistema inoperante
# Obter informações do erro
error_info = get_error_info(ErrorCode.E_SESSION_NOT_FOUND)
# {
# 'error_code': 'SESSION_NOT_FOUND',
# 'http_status': 404,
# 'message': 'Sessão não encontrada',
# 'severity': 'medium',
# 'retryable': False,
# 'action': 'Verificar se o session_id está correto...'
# }
# Criar resposta de erro padronizada
error_response = create_error_response(ErrorCode.E_SESSION_NOT_FOUND, {'session_id': '123'})
from pywhatsweb.errors import get_retryable_errors, get_errors_by_severity
# Erros que podem ser tentados novamente
retryable = get_retryable_errors()
# Erros por severidade
critical = get_errors_by_severity(ErrorSeverity.CRITICAL)
high = get_errors_by_severity(ErrorSeverity.HIGH)
pywhatsweb/ - CORE DA BIBLIOTECA__init__.py - Ponto de entrada principal# Importa todas as classes principais
from pywhatsweb import WhatsWebManager, Session, FileSystemStore
# Exemplo de uso básico
manager = WhatsWebManager()
session = manager.create_session("minha_sessao")
Funções expostas:
WhatsWebManager: Gerenciador principalSession: Sessão individualBaseStore, FileSystemStore, DjangoORMStore: StorageSessionStatus, MessageType, KanbanStatus: Enumsmanager.py - Gerenciador de sessõesclass WhatsWebManager:
def __init__(self, sidecar_host="localhost", sidecar_port=3000,
api_key="secret", storage=None):
# Configuração do sidecar e storage
def create_session(self, session_id: str, **kwargs) -> Session:
# Cria nova sessão WhatsApp
def get_session(self, session_id: str) -> Optional[Session]:
# Recupera sessão existente
def list_sessions(self) -> List[str]:
# Lista todas as sessões ativas
def remove_session(self, session_id: str) -> bool:
# Remove sessão específica
Métodos principais:
create_session(): Cria nova sessãoget_session(): Recupera sessão existentelist_sessions(): Lista todas as sessõesget_active_sessions(): Sessões conectadasclose_all_sessions(): Para todas as sessõescleanup(): Limpeza completasession.py - Sessão WhatsApp individualclass Session:
def __init__(self, session_id: str, manager, **kwargs):
# Inicializa sessão com ID único
def start(self) -> bool:
# Inicia conexão com sidecar
def stop(self) -> bool:
# Para a sessão
def send_text(self, to: str, text: str) -> str:
# Envia mensagem de texto
def send_media(self, to: str, media_path: str, caption: str = "") -> str:
# Envia arquivo de mídia
def on(self, event: str, handler: Callable):
# Registra handler para eventos
Eventos disponíveis:
qr: QR Code para autenticaçãoauthenticated: Autenticação bem-sucedidaready: WhatsApp pronto para usomessage: Nova mensagem recebidadisconnected: DesconexãoFuncionalidades avançadas:
Exemplo de uso:
@session.on("qr")
def on_qr(data):
print(f"QR Code: {data['qr']}")
@session.on("message")
def on_message(data):
print(f"Nova mensagem: {data['body']}")
session.start() # Inicia e aguarda QR
enums.py - Enumerações e contratos do sistemaclass SessionStatus(Enum):
DISCONNECTED = "disconnected" # Desconectado
CONNECTING = "connecting" # Conectando
CONNECTED = "connected" # Conectado
READY = "ready" # Pronto para uso
ERROR = "error" # Erro
class MessageType(Enum):
TEXT = "text" # Texto
IMAGE = "image" # Imagem
AUDIO = "audio" # Áudio
VIDEO = "video" # Vídeo
DOCUMENT = "document" # Documento
class KanbanStatus(Enum):
NEW = "new" # Aguardando
ACTIVE = "active" # Em atendimento
DONE = "done" # Concluídos
class MessageStatus(Enum):
PENDING = "pending" # Enfileirada
SENT = "sent" # Enviada
DELIVERED = "delivered" # Entregue
READ = "read" # Lida
FAILED = "failed" # Falhou
Contratos de eventos padronizados:
@dataclass
class EventPayload:
event: EventType # Tipo do evento
session_id: str # ID da sessão
timestamp: datetime # Timestamp do evento
trace_id: str # ID único para rastreamento
data: Dict[str, Any] # Dados específicos do evento
# Fábricas de eventos
create_qr_event(session_id, qr_data, expires_in=60)
create_message_event(session_id, message_data)
create_ready_event(session_id, phone_number, device_info)
create_error_event(session_id, error_code, error_message)
Constantes do sistema:
# Configurações de mídia
MAX_MEDIA_SIZE = 16 * 1024 * 1024 # 16MB (limite WhatsApp)
SUPPORTED_IMAGE_TYPES = ['image/jpeg', 'image/png', 'image/gif', 'image/webp']
# Timeouts
CONNECTION_TIMEOUT = 30 # Timeout de conexão (segundos)
MESSAGE_TIMEOUT = 60 # Timeout de envio de mensagem (segundos)
QR_EXPIRATION = 60 # Expiração do QR (segundos)
# Rate limits
MAX_MESSAGES_PER_MINUTE = 30 # Máximo de mensagens por minuto
MAX_SESSIONS_PER_IP = 10 # Máximo de sessões por IP
MAX_REQUESTS_PER_MINUTE = 100 # Máximo de requisições por minuto
exceptions.py - Exceções customizadasclass WhatsAppError(Exception): # Erro base
class SessionError(Exception): # Erro de sessão
class ConnectionError(Exception): # Erro de conexão
class MessageError(Exception): # Erro de mensagem
class StorageError(Exception): # Erro de storage
class AuthenticationError(Exception): # Erro de autenticação
class WebSocketError(Exception): # Erro de WebSocket
class APIError(Exception): # Erro de API
models.py - Modelos de dados com idempotência@dataclass
class Contact:
phone: str # Número do telefone
name: Optional[str] # Nome do contato
is_group: bool = False # É grupo?
is_business: bool = False # É business?
profile_picture: Optional[str] # Foto do perfil
status: Optional[str] # Status do contato
last_seen: Optional[datetime] # Última vez visto
created_at: datetime # Data de criação
updated_at: datetime # Data de atualização
@dataclass
class Message:
id: str # ID único da mensagem
content: str # Conteúdo da mensagem
sender: Contact # Remetente
recipient: Contact # Destinatário
message_type: MessageType # Tipo da mensagem
timestamp: datetime # Timestamp da mensagem
status: MessageStatus # Status de entrega
metadata: Dict[str, Any] # Metadados adicionais
# Campos de idempotência e rastreamento
trace_id: str # ID único para rastreamento
producer_id: Optional[str] # ID do produtor (para deduplicação)
correlation_id: Optional[str] # ID de correlação
sequence_number: Optional[int] # Número de sequência
# Campos de auditoria
created_at: datetime # Data de criação
updated_at: datetime # Data de atualização
@dataclass
class Chat:
chat_id: str # ID do chat
status: KanbanStatus # Status Kanban
owner_id: Optional[str] # Atendente responsável
last_message_at: Optional[datetime] # Última mensagem
created_at: datetime # Data de criação
updated_at: datetime # Data de atualização
# Campos de rastreamento
trace_id: str # ID único para rastreamento
correlation_id: Optional[str] # ID de correlação
# Campos de auditoria
assigned_at: Optional[datetime] # Data de atribuição
assigned_by: Optional[str] # Quem atribuiu
completed_at: Optional[datetime] # Data de conclusão
completed_by: Optional[str] # Quem concluiu
Métodos de idempotência:
# Gerar chave de idempotência única
message.get_idempotency_key() # producer_id:sequence_number ou trace_id:timestamp
# Marcar status de entrega
message.mark_as_sent() # PENDING → SENT
message.mark_as_delivered() # SENT → DELIVERED
message.mark_as_read() # DELIVERED → READ
message.mark_as_failed(error) # Qualquer → FAILED
# Gerenciar status Kanban
chat.assign_to(owner_id, assigned_by) # NEW → ACTIVE
chat.mark_as_done(completed_by) # ACTIVE → DONE
chat.reopen(reopened_by) # DONE → NEW
pywhatsweb/storage/ - SISTEMA DE PERSISTÊNCIAbase.py - Interface baseclass BaseStore(ABC):
@abstractmethod
def save_message(self, message: Message):
# Salva mensagem
@abstractmethod
def get_chat_messages(self, chat_id: str, limit: int = 50) -> List[Message]:
# Recupera mensagens do chat
@abstractmethod
def save_contact(self, contact: Contact):
# Salva contato
@abstractmethod
def get_contact(self, phone: str) -> Optional[Contact]:
# Recupera contato
@abstractmethod
def save_chat(self, chat: Chat):
# Salva chat
@abstractmethod
def get_chat(self, chat_id: str) -> Optional[Chat]:
# Recupera chat
filesystem.py - Storage em arquivoclass FileSystemStore(BaseStore):
def __init__(self, base_dir: str = "./whatsapp_data"):
# Armazena dados em arquivos JSON
def save_message(self, message: Message):
# Salva em JSON
def get_chat_messages(self, chat_id: str, limit: int = 50) -> List[Message]:
# Lê de JSON
Estrutura de arquivos:
whatsapp_data/
├── messages/
│ ├── chat_5511999999999.json
│ └── chat_5511888888888.json
├── contacts/
│ └── contacts.json
├── groups/
│ └── groups.json
└── chats/
└── chats.json
django.py - Storage Django ORM (OPCIONAL)class DjangoORMStore(BaseStore):
def __init__(self):
# Verifica se Django está disponível
def set_models(self, session_model, message_model, contact_model,
group_model, chat_model, event_model):
# Usuário deve configurar os models Django
IMPORTANTE: Django é OPCIONAL! A biblioteca funciona perfeitamente sem Django.
# 1. Instalar biblioteca
pip install pywhatsweb
# 2. Testar (sidecar inicia automaticamente)
python examples/basic_usage.py
# 1. Clonar e entrar
git clone <repo> && cd pywhatsweb-lib
# 2. Instalar Python
pip install -r requirements.txt
# 3. Instalar Node.js e iniciar sidecar
cd sidecar && npm install && npm start
# 4. Em outro terminal, testar
python examples/basic_usage.py
from pywhatsweb import WhatsWebManager
# Criar manager com sidecar Python (padrão)
manager = WhatsWebManager(
api_key="sua-chave",
sidecar_type="python" # Opcional, é o padrão
)
# Criar sessão
session = manager.create_session("teste")
# Handler para QR
@session.on("qr")
def on_qr(data):
print(f"QR: {data['qr']}") # Escanear no celular
# Handler para mensagens
@session.on("message")
def on_message(data):
print(f"Msg: {data['body']} de {data['from']}")
# Iniciar e aguardar QR
session.start()
from pywhatsweb import WhatsWebManager
# Criar manager com sidecar Node.js
manager = WhatsWebManager(
api_key="sua-chave",
sidecar_type="nodejs" # Especificar Node.js
)
# Criar sessão
session = manager.create_session("teste")
# Handler para QR
@session.on("qr")
def on_qr(data):
print(f"QR: {data['qr']}") # Escanear no celular
# Handler para mensagens
@session.on("message")
def on_message(data):
print(f"Msg: {data['body']} de {data['from']}")
# Iniciar e aguardar QR
session.start()
sidecar/ - SIDECAR NODE.JSpackage.json - Dependências Node.js{
"dependencies": {
"whatsapp-web.js": "^1.23.0", // Biblioteca WhatsApp
"express": "^4.18.2", // API REST
"ws": "^8.14.2", // WebSocket
"qrcode": "^1.5.3", // Geração de QR
"helmet": "^7.1.0", // Segurança
"express-rate-limit": "^7.1.5", // Rate limiting
"morgan": "^1.10.0", // Logging
"cors": "^2.8.5" // CORS
}
}
src/server.js - Servidor principal com verificação de compatibilidadeVerificação de compatibilidade no boot (FAIL-FAST):
// Verifica automaticamente no startup
const REQUIRED_NODE_VERSION = '18.0.0';
const REQUIRED_WHATSAPP_WEB_JS_VERSION = '1.23.0';
const REQUIRED_PUPPETEER_VERSION = '21.0.0';
function checkCompatibility() {
// Verificar Node.js
const nodeVersion = process.version;
if (!semver.gte(nodeVersion, REQUIRED_NODE_VERSION)) {
console.error(`❌ Node.js ${REQUIRED_NODE_VERSION}+ é obrigatório. Atual: ${nodeVersion}`);
process.exit(1);
}
// Verificar whatsapp-web.js
const whatsappVersion = require('whatsapp-web.js/package.json').version;
if (!semver.gte(whatsappVersion, REQUIRED_WHATSAPP_WEB_JS_VERSION)) {
console.error(`❌ whatsapp-web.js ${REQUIRED_WHATSAPP_WEB_JS_VERSION}+ é obrigatório. Atual: ${whatsappVersion}`);
process.exit(1);
}
console.log('✅ Compatibilidade verificada com sucesso');
}
// Executar verificação no startup
checkCompatibility();
| Componente | Versão Mínima | Versão Recomendada | Verificação | Status |
|---|---|---|---|---|
| Node.js | 18.0.0 | 18.17.0+ (LTS) | node --version | ✅ Verificado |
| whatsapp-web.js | 1.23.0 | 1.23.0+ | npm list whatsapp-web.js | ✅ Verificado |
| Python | 3.8 | 3.9+ (3.13+ recomendado) | python --version | ✅ Verificado |
| Sidecar | v0.5.1 | v0.5.1+ | Health check /health | ✅ Verificado |
| Puppeteer | 21.0.0 | 21.0.0+ | npm list puppeteer | ✅ Verificado |
# pywhatsweb/__init__.py - Verificação de compatibilidade aprimorada
import sys
def check_python_compatibility():
"""Verifica compatibilidade com versão do Python"""
if sys.version_info < (3, 8):
raise RuntimeError(
"PyWhatsWeb requer Python 3.8 ou superior. "
f"Versão atual: {sys.version}"
)
# Verificar versões recomendadas
if sys.version_info >= (3, 13):
print("✅ Python 3.13+ detectado - Suporte completo!")
elif sys.version_info >= (3, 11):
print("✅ Python 3.11+ detectado - Suporte completo!")
elif sys.version_info >= (3, 8):
print("✅ Python 3.8+ detectado - Suporte básico!")
# Verificar dependências opcionais
try:
import django
if django.VERSION >= (3, 2):
print("✅ Django 3.2+ detectado - Suporte completo!")
else:
print("⚠️ Django < 3.2 detectado - Funcionalidades limitadas")
except ImportError:
print("ℹ️ Django não instalado - Use 'pip install pywhatsweb[django]'")
try:
import websockets
if websockets.version_tuple >= (10, 0):
print("✅ websockets 10.0+ detectado - Suporte completo!")
else:
print("⚠️ websockets < 10.0 detectado - Funcionalidades limitadas")
except ImportError:
print("ℹ️ websockets não instalado - Funcionalidades limitadas")
# Executar verificação de compatibilidade
check_python_compatibility()
# Testar compatibilidade completa
pww test compatibility
# Verificar versões instaladas
pww version --all
# Teste de conectividade
pww test connectivity --sidecar --websocket --storage
Resultado da verificação (v0.5.1):
🔍 Verificação de Compatibilidade PyWhatsWeb v0.5.1
==================================================
✅ Node.js: v18.17.0 (requerido: 18.0.0+)
✅ whatsapp-web.js: v1.23.0 (requerido: 1.23.0+)
✅ Python: v3.13.0 (requerido: 3.8+, recomendado: 3.13+)
✅ Sidecar: v0.5.1 (requerido: v0.5.1+)
✅ Puppeteer: v21.0.0 (requerido: 21.0.0+)
🌐 Conectividade:
✅ Sidecar HTTP: http://localhost:3000
✅ Sidecar WebSocket: ws://localhost:3001
✅ Storage: FileSystem (./whatsapp_data)
🎯 Status: COMPATÍVEL ✅
🎯 Python 3.13: SUPORTE COMPLETO ✅
🎯 Websockets 15.0+: SUPORTE COMPLETO ✅
🎯 Dataclasses: SUPORTE COMPLETO ✅
Autenticação WebSocket obrigatória:
// Token obrigatório na conexão
const wss = new WebSocket.Server({
port: WS_PORT,
verifyClient: (info) => {
const token = info.req.headers['authorization'] ||
new URL(info.req.url, 'http://localhost').searchParams.get('token');
if (!token || token !== `Bearer ${API_KEY}`) {
return false; // Conexão rejeitada
}
// Suporte a multi-tenant
const tenantId = info.req.headers['x-tenant-id'] ||
new URL(info.req.url, 'http://localhost').searchParams.get('tenant_id');
if (tenantId) {
info.req.tenantId = tenantId;
}
return true;
}
});
Sistema anti-loop:
client.on('message', (message) => {
// Marcar origem da mensagem para evitar loops
const messageOrigin = 'inbound'; // Mensagem recebida do WhatsApp
const isFromSelf = message.fromMe;
broadcastEvent('message', {
// ... outros campos
origin: messageOrigin, // 'inbound' ou 'outbound'
fromSelf: isFromSelf, // true se enviada pelo próprio usuário
correlationId: `msg-${Date.now()}-${Math.random().toString(36).substr(2, 9)}`
});
});
// API REST endpoints
app.post('/api/session/:id/start', startSession);
app.post('/api/session/:id/stop', stopSession);
app.get('/api/session/:id/status', getSessionStatus);
app.post('/api/session/:id/send-message', sendMessage);
// Health check e métricas
app.get('/health', healthCheck);
app.get('/metrics', getMetrics);
// WebSocket events
ws.on('message', (data) => {
// Broadcast: qr, authenticated, ready, message, disconnected
});
Endpoints disponíveis:
GET /health - Health check (sem autenticação)GET /metrics - Métricas Prometheus (requer API key)GET /metrics.json - Métricas JSON (requer API key)POST /api/session/:id/start - Inicia sessãoPOST /api/session/:id/stop - Para sessãoGET /api/session/:id/status - Status da sessãoPOST /api/session/:id/send-message - Envia mensagemEventos WebSocket:
qr: QR Code para autenticaçãoauthenticated: Autenticação bem-sucedidaready: WhatsApp prontomessage: Nova mensagemdisconnected: DesconexãoFuncionalidades de segurança:
sidecar/ - DOCKER E ORQUESTRAÇÃODockerfile - Container otimizado# Base Node.js 18 Alpine
FROM node:18-alpine
# Dependências do Puppeteer
RUN apk add --no-cache chromium nss freetype freetype-dev harfbuzz ca-certificates ttf-freefont
# Variáveis de ambiente para Puppeteer
ENV PUPPETEER_SKIP_CHROMIUM_DOWNLOAD=true \
PUPPETEER_EXECUTABLE_PATH=/usr/bin/chromium-browser \
PUPPETEER_ARGS="--no-sandbox --disable-setuid-sandbox --disable-dev-shm-usage --disable-accelerated-2d-canvas --no-first-run --no-zygote --disable-gpu"
# Usuário não-root para segurança
USER node
# Health check automático
HEALTHCHECK --interval=30s --timeout=3s --start-period=5s --retries=3 \
CMD node -e "require('http').get('http://localhost:3000/health', (res) => { process.exit(res.statusCode === 200 ? 0 : 1) })"
docker-compose.yml - Orquestração completaservices:
# Sidecar principal
pywhatsweb-sidecar:
build: ./sidecar
ports: ["3000:3000", "3001:3001"]
environment:
- API_KEY=${WHATSAPP_API_KEY}
- ALLOWED_ORIGINS=${ALLOWED_ORIGINS}
healthcheck:
test: ["CMD", "node", "-e", "require('http').get('http://localhost:3000/health', (res) => { process.exit(res.statusCode === 200 ? 0 : 1) })"]
interval: 30s
timeout: 10s
retries: 3
# Prometheus para métricas
prometheus:
image: prom/prometheus:latest
ports: ["9090:9090"]
volumes: ["./prometheus:/etc/prometheus"]
# Grafana para visualização
grafana:
image: grafana/grafana:latest
ports: ["3002:3000"]
environment:
- GF_SECURITY_ADMIN_PASSWORD=${GRAFANA_PASSWORD}
examples/ - EXEMPLOS DE USORUNBOOKS.md - OPERAÇÃO EM PRODUÇÃO# Backup de sessões
curl -X POST -H "Authorization: Bearer ${API_KEY}" \
http://localhost:3000/api/sessions/stop-all
# Atualização do sidecar
docker-compose build pywhatsweb-sidecar
docker-compose up -d pywhatsweb-sidecar
# Limpeza de dados antigos
find ./whatsapp_data -name "*.json" -mtime +30 -delete
# Health check automático
curl -f http://localhost:3000/health || echo "SIDECAR DOWN"
# Verificar sessões ativas
ACTIVE_SESSIONS=$(curl -s -H "Authorization: Bearer ${API_KEY}" \
http://localhost:3000/metrics | jq -r '.sessions.active')
if [ "$ACTIVE_SESSIONS" -eq 0 ]; then
echo "ALERTA: Nenhuma sessão ativa!"
fi
basic_usage.py - Uso básicofrom pywhatsweb import WhatsWebManager, FileSystemStore
# Criar manager
manager = WhatsWebManager(
sidecar_host="localhost",
sidecar_port=3000,
api_key="sua-api-key"
)
# Criar sessão
session = manager.create_session("sessao_teste")
# Configurar eventos
@session.on("qr")
def on_qr(data):
print(f"QR Code: {data['qr']}")
@session.on("message")
def on_message(data):
print(f"Nova mensagem: {data['body']}")
# Iniciar sessão
session.start()
# Enviar mensagem
session.send_text("5511999999999", "Olá!")
django_complete_example.py - Exemplo Django completo# views.py - Views Django para WhatsApp
from django.views.generic import TemplateView
from django.http import JsonResponse
from pywhatsweb import WhatsWebManager, DjangoORMStore
class WhatsAppDashboardView(TemplateView):
template_name = 'whatsapp/dashboard.html'
def get_context_data(self, **kwargs):
# Inicializar manager e configurar models Django
manager = WhatsWebManager(
sidecar_host=settings.WHATSAPP_SIDECAR_HOST,
sidecar_port=settings.WHATSAPP_SIDECAR_PORT,
api_key=settings.WHATSAPP_API_KEY,
storage=DjangoORMStore()
)
# Configurar models Django
manager.storage.set_models(
session_model=WhatsAppSession,
message_model=WhatsAppMessage,
contact_model=WhatsAppContact
)
# Buscar sessões ativas
active_sessions = manager.get_active_sessions()
context = super().get_context_data(**kwargs)
context['active_sessions'] = active_sessions
context['manager'] = manager
return context
# API para enviar mensagens
def send_message_api(request):
if request.method == 'POST':
session_id = request.POST.get('session_id')
to = request.POST.get('to')
text = request.POST.get('text')
manager = WhatsWebManager()
session = manager.get_session(session_id)
try:
message_id = session.send_text(to, text)
return JsonResponse({'success': True, 'message_id': message_id})
except Exception as e:
return JsonResponse({'success': False, 'error': str(e)})
<!-- whatsapp/dashboard.html -->
{% extends 'base.html' %}
{% block content %}
<div class="whatsapp-dashboard">
<h1>Dashboard WhatsApp</h1>
<!-- Lista de sessões -->
<div class="sessions-list">
{% for session in active_sessions %}
<div class="session-card" data-session-id="{{ session.session_id }}">
<h3>Sessão: {{ session.session_id }}</h3>
<div class="session-status">
Status: <span class="status-{{ session.status }}">{{ session.status }}</span>
</div>
<!-- Container do QR Code -->
<div class="qr-container" id="qr-{{ session.session_id }}">
<img id="qr-code-{{ session.session_id }}" src="" alt="QR Code WhatsApp" style="display: none;">
<div id="qr-status-{{ session.session_id }}">Aguardando QR Code...</div>
</div>
<!-- Controles da sessão -->
<div class="session-controls">
<button onclick="startSession('{{ session.session_id }}')">Iniciar</button>
<button onclick="stopSession('{{ session.session_id }}')">Parar</button>
<button onclick="showQR('{{ session.session_id }}')">Mostrar QR</button>
</div>
<!-- Envio de mensagens -->
<div class="message-form">
<input type="text" id="to-{{ session.session_id }}" placeholder="Número (ex: 5511999999999)">
<input type="text" id="text-{{ session.session_id }}" placeholder="Mensagem">
<button onclick="sendMessage('{{ session.session_id }}')">Enviar</button>
</div>
</div>
{% endfor %}
</div>
</div>
<script>
// WebSocket para receber eventos WhatsApp
const ws = new WebSocket('ws://localhost:8000/ws/whatsapp/');
ws.onmessage = function(e) {
const data = JSON.parse(e.data);
const sessionId = data.session_id;
if (data.type === 'qr') {
// Mostrar QR Code
const qrImg = document.getElementById(`qr-code-${sessionId}`);
const qrStatus = document.getElementById(`qr-status-${sessionId}`);
qrImg.src = data.qr;
qrImg.style.display = 'block';
qrStatus.textContent = 'Escaneie o QR Code';
} else if (data.type === 'ready') {
// WhatsApp conectado
const qrImg = document.getElementById(`qr-code-${sessionId}`);
const qrStatus = document.getElementById(`qr-status-${sessionId}`);
qrImg.style.display = 'none';
qrStatus.textContent = 'WhatsApp conectado!';
} else if (data.type === 'message') {
// Nova mensagem recebida
console.log('Nova mensagem:', data);
// Aqui você pode atualizar a interface
}
};
// Funções para controlar sessões
function startSession(sessionId) {
fetch(`/whatsapp/session/${sessionId}/start/`, {method: 'POST'})
.then(response => response.json())
.then(data => console.log('Sessão iniciada:', data));
}
function stopSession(sessionId) {
fetch(`/whatsapp/session/${sessionId}/stop/`, {method: 'POST'})
.then(response => response.json())
.then(data => console.log('Sessão parada:', data));
}
function sendMessage(sessionId) {
const to = document.getElementById(`to-${sessionId}`).value;
const text = document.getElementById(`text-${sessionId}`).value;
fetch('/whatsapp/send-message/', {
method: 'POST',
headers: {'Content-Type': 'application/x-www-form-urlencoded'},
body: `session_id=${sessionId}&to=${to}&text=${text}`
})
.then(response => response.json())
.then(data => {
if (data.success) {
console.log('Mensagem enviada:', data.message_id);
} else {
console.error('Erro ao enviar:', data.error);
}
});
}
</script>
{% endblock %}
# consumers.py - Consumer WebSocket para Django Channels
import json
from channels.generic.websocket import AsyncWebsocketConsumer
from pywhatsweb import WhatsWebManager
class WhatsAppWebSocketConsumer(AsyncWebsocketConsumer):
async def connect(self):
await self.accept()
print("Cliente WebSocket conectado")
# Conectar ao WebSocket da sessão WhatsApp
self.whatsapp_manager = WhatsWebManager()
async def disconnect(self, close_code):
print(f"Cliente WebSocket desconectado: {close_code}")
async def receive(self, text_data):
# Processar mensagens do cliente
data = json.loads(text_data)
if data['action'] == 'send_message':
session_id = data['session_id']
to = data['to']
text = data['text']
try:
session = self.whatsapp_manager.get_session(session_id)
message_id = await session.send_text(to, text)
# Confirmar envio para o cliente
await self.send(text_data=json.dumps({
'type': 'message_sent',
'message_id': message_id,
'to': to
}))
except Exception as e:
await self.send(text_data=json.dumps({
'type': 'error',
'error': str(e)
}))
async def whatsapp_event(self, event):
# Enviar eventos WhatsApp para o cliente
await self.send(text_data=json.dumps({
'type': event['event_type'],
'session_id': event['session_id'],
'data': event['data']
}))
Funcionalidades do exemplo Django:
tests/ - TESTES AUTOMATIZADOStest_imports.py - Teste de importsdef test_imports():
# Testa se todas as classes podem ser importadas
from pywhatsweb import WhatsWebManager, Session
from pywhatsweb import FileSystemStore, DjangoORMStore
def test_instanciacao():
# Testa se as classes podem ser instanciadas
manager = WhatsWebManager()
storage = FileSystemStore("./test_data")
# Clonar repositório
git clone <repo>
cd pywhatsweb-lib
# Instalar dependências Python
pip install -r requirements.txt
# Instalar dependências Node.js
cd sidecar
npm install
cd sidecar
cp env.example .env
# Editar .env com suas configurações
npm start
from pywhatsweb import WhatsWebManager, FileSystemStore
manager = WhatsWebManager(
sidecar_host="localhost", # Host do sidecar
sidecar_port=3000, # Porta HTTP
sidecar_ws_port=3001, # Porta WebSocket
api_key="sua-api-key", # Chave de API
storage=FileSystemStore("./whatsapp_data") # Storage
)
# Criar sessão
session = manager.create_session("sessao_123")
# Configurar eventos
@session.on("qr")
def on_qr(data):
# Renderizar QR no template Django/Flask
qr_data_url = data['qr']
# <img src="{{ qr_data_url }}">
@session.on("ready")
def on_ready(data):
print("WhatsApp está pronto!")
@session.on("message")
def on_message(data):
# Processar mensagem recebida
sender = data['from']
content = data['body']
# Auto-resposta
if "oi" in content.lower():
session.send_text(sender, "Oi! Como posso ajudar?")
# Iniciar sessão
session.start()
# Criar múltiplas sessões com tenant
sessao1 = manager.create_session("sessao_1", tenant_id="empresa_a")
sessao2 = manager.create_session("sessao_2", tenant_id="empresa_b")
# Cada sessão é independente e isolada por tenant
sessao1.start()
sessao2.start()
# Listar sessões ativas por tenant
active_sessions = manager.get_active_sessions(tenant_id="empresa_a")
print(f"Sessões ativas empresa A: {active_sessions}")
# Criar múltiplas sessões
sessao1 = manager.create_session("sessao_1", phone_number="5511999999999")
sessao2 = manager.create_session("sessao_2", phone_number="5511888888888")
# Cada sessão é independente
sessao1.start()
sessao2.start()
# Listar sessões ativas
active_sessions = manager.get_active_sessions()
print(f"Sessões ativas: {active_sessions}")
# Texto
message_id = session.send_text("5511999999999", "Olá! Como vai?")
# Mídia
message_id = session.send_media(
to="5511999999999",
media_path="./imagem.jpg",
caption="Veja esta imagem!"
)
# Verificar status
if session.is_active():
print("Sessão está ativa e pronta")
# FileSystem (padrão) - armazena em JSON
from pywhatsweb import FileSystemStore
storage = FileSystemStore("./whatsapp_data")
# Django ORM (opcional) - integração com banco de dados
from pywhatsweb import DjangoORMStore
storage = DjangoORMStore()
# Redis (futuro) - para alta performance
# from pywhatsweb import RedisStore
# storage = RedisStore(redis_url="redis://localhost:6379")
# PostgreSQL (futuro) - para dados estruturados
# from pywhatsweb import PostgreSQLStore
# storage = PostgreSQLStore(connection_string="postgresql://...")
# models.py - Models Django para WhatsApp
from django.db import models
from pywhatsweb import KanbanStatus, MessageType
class WhatsAppSession(models.Model):
session_id = models.CharField(max_length=100, unique=True)
tenant_id = models.CharField(max_length=50, null=True)
status = models.CharField(max_length=20, choices=KanbanStatus.choices())
phone_number = models.CharField(max_length=20, null=True)
created_at = models.DateTimeField(auto_now_add=True)
updated_at = models.DateTimeField(auto_now=True)
class WhatsAppMessage(models.Model):
message_id = models.CharField(max_length=100, unique=True)
session = models.ForeignKey(WhatsAppSession, on_delete=models.CASCADE)
content = models.TextField()
sender = models.CharField(max_length=20)
recipient = models.CharField(max_length=20)
message_type = models.CharField(max_length=20, choices=MessageType.choices())
status = models.CharField(max_length=20)
trace_id = models.CharField(max_length=100)
created_at = models.DateTimeField(auto_now_add=True)
# Configurar storage
storage.set_models(
session_model=WhatsAppSession,
message_model=WhatsAppMessage,
contact_model=WhatsAppContact,
chat_model=WhatsAppChat
)
# Configurações de mídia
MEDIA_CONFIG = {
'max_size': 16 * 1024 * 1024, # 16MB (limite WhatsApp)
'allowed_types': ['image/jpeg', 'image/png', 'image/gif', 'image/webp'],
'timeout': 300, # 5 minutos para upload
'retry_count': 3,
'storage_backend': 'local' # 'local', 's3', 'minio'
}
# Envio com validação automática
try:
message_id = session.send_media(
to="5511999999999",
media_path="./imagem.jpg",
caption="Veja esta imagem!",
validate_media=True, # Validação automática
compress_if_needed=True # Compressão se necessário
)
except MediaTooLargeError:
print("Arquivo muito grande, comprimindo...")
compressed_path = compress_image("./imagem.jpg")
message_id = session.send_media(to="5511999999999", media_path=compressed_path)
# Configurar storage de mídia
manager = WhatsWebManager(
media_storage={
'backend': 's3', # 'local', 's3', 'minio'
'bucket': 'whatsapp-media',
'region': 'us-east-1',
'retention_days': 30, # Retenção automática
'encryption': True, # Criptografia em repouso
'cdn_enabled': True # CDN para entrega rápida
}
)
# Upload para S3/MinIO
media_url = session.upload_media_to_storage(
file_path="./imagem.jpg",
mime_type="image/jpeg",
metadata={'session_id': 'sessao_123'}
)
# Configurar política LGPD
manager = WhatsWebManager(
lgpd_compliance={
'data_retention_days': 90, # Retenção de mensagens
'media_retention_days': 30, # Retenção de mídia
'log_redaction': True, # Redação de dados sensíveis
'media_encryption': True, # Criptografia de anexos
'audit_trail': True, # Rastreamento de ações
'right_to_forget': True, # Direito ao esquecimento
'data_export': True # Exportação de dados
}
)
# Excluir dados de um usuário
storage.delete_user_data(
phone_number="5511999999999",
reason="direito_ao_esquecimento",
audit_user="admin"
)
# Exportar dados para LGPD
user_data = storage.export_user_data("5511999999999")
# Retorna JSON com todas as mensagens, contatos, etc.
# Logs sem conteúdo de mensagem (apenas metadados)
# [2024-12-19 10:30:00] [corr:req_456] [tenant:empresa_a] Mensagem enviada: ID=msg_123, Para=5511999999999, Tipo=texto
# Criptografia AES-256 em repouso para mídia
# Chaves gerenciadas por KMS ou variáveis de ambiente
# Rastreamento completo de acesso e modificações
audit_log = storage.get_audit_log(
user_id="5511999999999",
start_date="2024-12-01",
end_date="2024-12-19"
)
from pywhatsweb import KanbanStatus
# Criar chat com status
chat = Chat(
chat_id="5511999999999",
status=KanbanStatus.NEW
)
# Atribuir para atendente
chat.assign_to("atendente_123", "sistema")
# Status muda para ACTIVE
# Marcar como concluído
chat.mark_as_done("atendente_123")
# Status muda para DONE
# Reabrir conversa
chat.reopen("sistema")
# Status volta para NEW
# Auto-NEW em mensagem nova
@session.on("message")
def on_message(data):
if data['origin'] == 'inbound':
chat = storage.get_or_create_chat(data['from'])
if chat.status == KanbanStatus.DONE:
# Cliente respondeu após conclusão, reabrir
chat.reopen("sistema", reason="cliente_respondeu")
storage.save_chat(chat)
# Auto-DONE após N horas de inatividade
def auto_close_inactive_chats():
from datetime import timedelta
active_chats = storage.get_chats_by_status(KanbanStatus.ACTIVE)
cutoff_time = datetime.now() - timedelta(hours=24)
for chat in active_chats:
if chat.last_message_at < cutoff_time:
chat.mark_as_done("sistema", reason="inatividade_24h")
storage.save_chat(chat)
from pywhatsweb.utils import normalize_phone_number, validate_phone_number
# Normalização automática para formato E.164
phone = normalize_phone_number("11999999999") # +5511999999999
phone = normalize_phone_number("(11) 99999-9999") # +5511999999999
phone = normalize_phone_number("+55 11 99999 9999") # +5511999999999
# Validação completa
is_valid, normalized, country_code, region = validate_phone_number("11999999999")
# (True, "+5511999999999", 55, "BR")
# Informações detalhadas do número
info = get_phone_info("+5511999999999")
# {
# 'number': '+5511999999999',
# 'country_code': 55,
# 'region': 'BR',
# 'carrier': 'Vivo',
# 'timezone': 'America/Sao_Paulo',
# 'is_valid': True,
# 'is_mobile': True
# }
# Formatação para diferentes regiões
display_national = format_phone_for_display("+5511999999999", "NATIONAL") # (11) 99999-9999
display_international = format_phone_for_display("+5511999999999", "INTERNATIONAL") # +55 11 99999-9999
display_e164 = format_phone_for_display("+5511999999999", "E164") # +5511999999999
# Usar como context manager
with WhatsWebManager() as manager:
session = manager.create_session("test")
session.start()
# ... usar sessão
# Cleanup automático ao sair
import logging
# Configurar logging
logging.basicConfig(level=logging.DEBUG)
# Logs automáticos em todas as operações
manager = WhatsWebManager()
# Logs: "PyWhatsWeb Manager inicializado", "Sessão criada", etc.
# A sessão reconecta automaticamente com backoff exponencial
# Tentativa 1: 2s, Tentativa 2: 4s, Tentativa 3: 8s, etc.
# Máximo 5 tentativas
@session.on("disconnected")
def on_disconnected(data):
print(f"Desconectado: {data['reason']}")
# Reconexão automática em andamento
@session.on("ready")
def on_ready(data):
print("Reconectado e pronto!")
# Ping/pong automático a cada 30 segundos
# Se falhar 3 vezes consecutivas, reconecta automaticamente
@session.on("heartbeat")
def on_heartbeat(data):
print(f"Heartbeat: {data['latency']}ms")
@session.on("heartbeat_failed")
def on_heartbeat_failed(data):
print("Heartbeat falhou, reconectando...")
# Gerar chave de idempotência única
message_id = session.send_text_with_idempotency(
to="5511999999999",
text="Mensagem importante",
idempotency_key="msg_123_v1" # Chave única
)
# Verificar se mensagem já foi enviada
if session.is_message_sent("msg_123_v1"):
print("Mensagem já enviada, ignorando...")
# Sistema anti-loop automático
@session.on("message")
def on_message(data):
# data['origin'] = 'inbound' (recebida) ou 'outbound' (enviada)
# data['fromSelf'] = True se enviada pelo próprio usuário
if data['origin'] == 'inbound' and not data['fromSelf']:
# Processar mensagem recebida
process_incoming_message(data)
else:
# Ignorar mensagens próprias ou de saída
print("Mensagem ignorada (loop prevention)")
# WhatsApp suporta alguns estados de entrega
@session.on("message_status")
def on_message_status(data):
if data['status'] == 'sent':
print("Mensagem enviada para servidor")
elif data['status'] == 'delivered':
print("Mensagem entregue ao destinatário")
elif data['status'] == 'read':
print("Mensagem lida pelo destinatário")
elif data['status'] == 'failed':
print(f"Mensagem falhou: {data['error']}")
# Configurações de entrega
MESSAGE_TIMEOUT = 60 # Timeout de envio (segundos)
DELIVERY_TIMEOUT = 300 # Timeout de entrega (segundos)
MAX_RETRIES = 3 # Máximo de tentativas
RETRY_DELAY = 5 # Delay entre tentativas (segundos)
@session.on("disconnected") def on_disconnected(data): print(f"Desconectado: {data['reason']}") # Reconexão automática em andamento
@session.on("ready") def on_ready(data): print("Reconectado e pronto!")
---
## 📊 **STATUS E MONITORAMENTO**
### **Informações da Sessão**
```python
# Status básico
status = session.get_status()
print(f"Status: {status.value}")
# Informações detalhadas
info = session.get_status_info()
print(f"Conectado desde: {info['created_at']}")
print(f"Última atividade: {info['last_activity']}")
print(f"Autenticado: {info['is_authenticated']}")
print(f"WebSocket conectado: {info['ws_connected']}")
print(f"Tentativas de reconexão: {info['reconnection_attempts']}")
# Listar todas as sessões
sessions = manager.list_sessions()
print(f"Sessões: {sessions}")
# Contar sessões
count = manager.get_session_count()
print(f"Total: {count}")
# Informações do sidecar
sidecar_info = manager.get_sidecar_info()
print(f"Sidecar: {sidecar_info['base_url']}")
# Health check básico
curl http://localhost:3000/health
# Métricas Prometheus (formato padrão)
curl -H "Authorization: Bearer sua-api-key" http://localhost:3000/metrics
# Métricas JSON (para dashboards customizados)
curl -H "Authorization: Bearer sua-api-key" http://localhost:3000/metrics.json
# Métricas-chave disponíveis
METRICS = {
# Contadores
'messages_in_total': 'Total de mensagens recebidas',
'messages_out_total': 'Total de mensagens enviadas',
'reconnections_total': 'Total de reconexões',
'errors_total': 'Total de erros por código',
# Gauges
'sessions_active': 'Sessões ativas no momento',
'sessions_total': 'Total de sessões criadas',
'websocket_connections': 'Conexões WebSocket ativas',
# Histogramas
'message_latency_seconds': 'Latência de envio (p50, p95, p99)',
'reconnection_delay_seconds': 'Delay de reconexão',
'media_upload_duration_seconds': 'Duração de upload de mídia'
}
# Exemplo de métricas Prometheus
# whatsapp_messages_in_total{tenant="empresa_a",session="sessao_123"} 150
# whatsapp_message_latency_seconds_bucket{le="0.1"} 45
# whatsapp_message_latency_seconds_bucket{le="0.5"} 120
# whatsapp_message_latency_seconds_bucket{le="1.0"} 150
# Health check completo do sistema
health_status = manager.get_health_status()
# {
# 'status': 'healthy',
# 'timestamp': '2024-12-19T10:30:00.000Z',
# 'uptime': 3600000,
# 'components': {
# 'sidecar': 'healthy',
# 'websocket': 'healthy',
# 'storage': 'healthy',
# 'sessions': 'healthy'
# },
# 'sessions': {
# 'active': 2,
# 'total': 5,
# 'healthy': 2,
# 'unhealthy': 0
# },
# 'metrics': {
# 'messagesIn': 150,
# 'messagesOut': 120,
# 'reconnections': 3,
# 'errors': 1
# }
# }
# Health check de sessão específica
session_health = session.get_health_status()
# {
# 'status': 'healthy',
# 'websocket_connected': True,
# 'last_heartbeat': '2024-12-19T10:29:45.000Z',
# 'reconnection_attempts': 0,
# 'message_queue_size': 0
# }
# Configurar alertas automáticos
manager.setup_alerts({
'max_reconnection_attempts': 5,
'max_message_latency_ms': 5000,
'min_active_sessions': 1,
'max_error_rate': 0.05 # 5% de erro
})
# Handler para alertas
@manager.on("alert")
def on_alert(alert_data):
if alert_data['type'] == 'high_latency':
print(f"ALERTA: Latência alta: {alert_data['latency']}ms")
# Notificar equipe, escalar, etc.
elif alert_data['type'] == 'session_down':
print(f"ALERTA: Sessão {alert_data['session_id']} caiu")
# Tentar reconexão automática
Resposta do health check:
{
"status": "healthy",
"timestamp": "2024-12-19T10:30:00.000Z",
"uptime": 3600000,
"sessions": {
"active": 2,
"total": 5
},
"metrics": {
"messagesIn": 150,
"messagesOut": 120,
"reconnections": 3,
"errors": 1
},
"version": "0.3.2"
}
try:
session.start()
except SessionError as e:
print(f"Erro de sessão: {e}")
except ConnectionError as e:
print(f"Erro de conexão: {e}")
except AuthenticationError as e:
print(f"Erro de autenticação: {e}")
except WebSocketError as e:
print(f"Erro de WebSocket: {e}")
if not session.is_connected():
print("Sessão não está conectada")
return
if not session.is_active():
print("Sessão não está ativa")
return
# Verificar reconexão
if session._reconnection_attempts > 0:
print(f"Tentativas de reconexão: {session._reconnection_attempts}")
# Sempre use uma API key forte
manager = WhatsWebManager(
api_key="sua-api-key-super-secreta-aqui"
)
// Autenticação no handshake de conexão
const wss = new WebSocket.Server({
port: WS_PORT,
verifyClient: (info) => {
const token = info.req.headers['authorization'] ||
new URL(info.req.url, 'http://localhost').searchParams.get('token');
if (!token || token !== `Bearer ${API_KEY}`) {
return false; // Conexão rejeitada imediatamente
}
// Validar tenant_id se fornecido
const tenantId = info.req.headers['x-tenant-id'];
if (tenantId && !isValidTenant(tenantId)) {
return false;
}
return true;
}
});
# Criar sessão com tenant específico
session = manager.create_session(
"sessao_123",
tenant_id="empresa_a",
correlation_id="req_456"
)
# Headers de correlação automáticos
# X-Correlation-Id: req_456
# X-Tenant-Id: empresa_a
# X-Session-Id: sessao_123
# Logs automáticos com correlation ID
# [2024-12-19 10:30:00] [corr:req_456] [tenant:empresa_a] Sessão criada: sessao_123
# [2024-12-19 10:30:05] [corr:req_456] [tenant:empresa_a] QR Code gerado
# [2024-12-19 10:30:30] [corr:req_456] [tenant:empresa_a] WhatsApp autenticado
# Sempre use uma API key forte
manager = WhatsWebManager(
api_key="sua-api-key-super-secreta-aqui"
)
# Restringir acesso por IP se necessário
manager = WhatsWebManager(
sidecar_host="127.0.0.1", # Apenas localhost
sidecar_port=3000
)
// Configuração no sidecar
const limiter = rateLimit({
windowMs: 15 * 60 * 1000, // 15 minutos
max: 100, // máximo 100 requisições por IP
message: 'Muitas requisições deste IP, tente novamente mais tarde.'
});
@session.on("message")
def auto_response(data):
content = data['body'].lower()
sender = data['from']
if "preço" in content:
session.send_text(sender, "Nossos preços: Produto A - R$ 50, Produto B - R$ 100")
elif "contato" in content:
session.send_text(sender, "Entre em contato: (11) 99999-9999")
else:
session.send_text(sender, "Como posso ajudar? Digite 'preço' ou 'contato'")
# Atribuir chat para atendente
def assign_chat_to_attendant(chat_id, attendant_id):
chat = storage.get_chat(chat_id)
chat.assign_to(attendant_id, "sistema")
storage.save_chat(chat)
# Notificar atendente
session.send_text(attendant_id, f"Novo chat atribuído: {chat_id}")
# Verificar conversas ativas
def monitor_active_chats():
active_chats = storage.get_chats_by_status(KanbanStatus.ACTIVE)
for chat in active_chats:
# Verificar tempo de atendimento
if chat.is_taking_too_long():
# Escalar para supervisor
escalate_chat(chat)
# Rastrear todas as ações
def audit_action(action, user_id, details):
event = SessionEvent(
session_id=session.session_id,
event_type=action,
data=details,
user_id=user_id,
ip_address=request.META.get('REMOTE_ADDR'),
user_agent=request.META.get('HTTP_USER_AGENT')
)
storage.save_event(event)
# Teste de imports
python test_imports.py
# Teste com pytest
pytest tests/
# Teste com tox
tox
# Instalar em modo desenvolvimento
pip install -e .
# Executar exemplos
python examples/basic_usage.py
python examples/django_complete_example.py
# Instalar CLI globalmente
pip install pywhatsweb[cli]
# Ou usar via Python
python -m pywhatsweb.cli session start --id teste
# Verificar versão
pww --version
# Gerenciar sessões
pww session start --id sessao_123 --tenant empresa_a
pww session status --id sessao_123
pww session stop --id sessao_123
pww session list --active
pww session list --all
# Mostrar QR Code
pww session qr --id sessao_123
# Enviar mensagem de teste
pww session send --id sessao_123 --to 5511999999999 --text "Teste CLI"
# Diagnóstico completo
pww diag --session sessao_123
pww diag --sidecar
pww diag --storage
# Ver logs em tempo real
pww logs --follow --session sessao_123
pww logs --level debug --tenant empresa_a
# Health check completo
pww health --sidecar --sessions --storage
pww health --format json
pww health --format prometheus
# Iniciar múltiplas sessões
pww session start --id atendente_1 --tenant empresa_a
pww session start --id atendente_2 --tenant empresa_a
pww session start --id suporte --tenant empresa_b
# Monitorar sessões ativas
pww session list --active --format table
# +-------------+-----------+--------+---------------------+
# | Session ID | Tenant | Status | Last Activity |
# +-------------+-----------+--------+---------------------+
# | atendente_1 | empresa_a | ready | 2024-12-19 10:30:00 |
# | atendente_2 | empresa_a | ready | 2024-12-19 10:29:45 |
# | suporte | empresa_b | qr | 2024-12-19 10:25:00 |
# +-------------+-----------+--------+---------------------+
# Diagnóstico de problemas
pww diag --session atendente_1 --verbose
# Sessão: atendente_1
# Status: ready
# WebSocket: connected
# Heartbeat: 45ms
# Última mensagem: 2 minutos atrás
# Reconexões: 0
# Erros: 0
cd sidecar
# Instalar dependências
npm install
# Executar em modo desenvolvimento
npm run dev
# Testar endpoints
curl http://localhost:3000/health
curl -H "Authorization: Bearer sua-api-key" http://localhost:3000/metrics
# .env do sidecar
PORT=3000
WS_PORT=3001
API_KEY=sua-api-key-super-secreta
ALLOWED_ORIGINS=http://localhost:3000,http://127.0.0.1:3000
PUPPETEER_ARGS=--no-sandbox --disable-setuid-sandbox
LOG_LEVEL=info
# ❌ Erro: "Node.js version must be 18.0.0 or higher"
# ✅ Solução: Atualizar Node.js
node --version # Deve ser ≥18.0.0
nvm install 18 # ou baixar do nodejs.org
# ❌ Erro: "Cannot find module 'whatsapp-web.js'"
# ✅ Solução: Reinstalar dependências
cd sidecar && rm -rf node_modules package-lock.json
npm install
# ❌ Problema: Evento 'qr' não dispara
# ✅ Solução: Verificar WebSocket
import logging
logging.basicConfig(level=logging.DEBUG)
# Verificar se está conectando
@session.on("connecting")
def on_connecting(data):
print("Conectando...")
@session.on("qr")
def on_qr(data):
print(f"QR: {data['qr']}") # Deve aparecer aqui
# ❌ Problema: "Session not ready"
# ✅ Solução: Aguardar evento 'ready'
@session.on("ready")
def on_ready(data):
print("WhatsApp pronto! Agora pode enviar")
session.send_text("5511999999999", "Teste")
# ❌ Problema: "Rate limit exceeded"
# ✅ Solução: Aguardar e tentar novamente
import time
time.sleep(60) # Aguardar 1 minuto
# ❌ Problema: Eventos param de chegar
# ✅ Solução: Reconexão automática
@session.on("disconnected")
def on_disconnected(data):
print(f"Desconectado: {data['reason']}")
# Reconexão automática em andamento
@session.on("ready")
def on_ready(data):
print("Reconectado!")
# 1. Verificar sidecar
curl http://localhost:3000/health
# 2. Verificar sessões ativas
curl -H "Authorization: Bearer ${API_KEY}" \
http://localhost:3000/metrics
# 3. Ver logs do sidecar
cd sidecar && npm run dev
# 4. Ver logs Python
python -c "import logging; logging.basicConfig(level=logging.DEBUG)"
# 1. Parar todas as sessões
curl -X POST -H "Authorization: Bearer ${API_KEY}" \
http://localhost:3000/api/sessions/stop-all
# 2. Reiniciar sidecar
cd sidecar && npm restart
# 3. Recriar sessão
session = manager.create_session("nova_sessao")
session.start()
# ❌ Erro: "Account banned" ou "Too many requests"
# ✅ Solução: Aguardar e usar número diferente
import time
# Aguardar 24h para ban
time.sleep(24 * 60 * 60)
# Ou usar número alternativo
session2 = manager.create_session("sessao_2", phone_number="5511888888888")
import logging
logging.basicConfig(level=logging.DEBUG)
# Logs detalhados de todas as operações
manager = WhatsWebManager()
# Health check
curl http://localhost:3000/health
# Ver logs do sidecar
cd sidecar
npm run dev
A PyWhatsWeb v0.5.1 é uma biblioteca completa e funcional que oferece:
✅ Arquitetura moderna com sidecar Node.js
✅ Multi-sessão para múltiplos atendentes
✅ Eventos em tempo real via WebSocket
✅ Storage pluggable (FileSystem ou Django)
✅ Sistema Kanban para gestão de conversas
✅ API Python limpa e fácil de usar
✅ Sem dependências de navegador/Selenium
✅ Funciona em qualquer projeto Python
✅ Reconexão automática com backoff exponencial
✅ Sistema de idempotência para deduplicação
✅ Health check e métricas completas
✅ Segurança robusta com rate limiting e CORS
✅ Exemplo Django completo com dashboard funcional
Para começar agora:
cd sidecar && npm install && npm startA biblioteca está pronta para uso em produção com todas as funcionalidades essenciais implementadas! 🚀
# 1. MANAGER - Gerenciador principal
manager = WhatsWebManager(api_key="sua-chave")
manager.create_session("id") # Criar sessão
manager.list_sessions() # Listar todas
manager.get_active_sessions() # Sessões ativas
# 2. SESSION - Sessão individual
session = manager.get_session("id")
session.start() # Iniciar conexão
session.stop() # Parar sessão
session.send_text(to, text) # Enviar texto
session.send_media(to, path) # Enviar arquivo
# 3. EVENTS - Sistema de eventos
@session.on("qr", handler) # QR Code
@session.on("message", handler) # Mensagem recebida
@session.on("ready", handler) # WhatsApp pronto
@session.on("disconnected", handler) # Desconectado
# FileSystem (padrão)
storage = FileSystemStore("./data")
# Django ORM (opcional)
storage = DjangoORMStore()
storage.set_models(WhatsAppSession, WhatsAppMessage)
# Operações básicas
storage.save_message(message)
storage.get_chat_messages(chat_id)
storage.save_contact(contact)
from pywhatsweb.utils import *
# Números de telefone
normalize_phone_number("11999999999") # +5511999999999
validate_phone_number("11999999999") # (True, "+5511999999999")
# IDs e timestamps
generate_session_id("session", "tenant") # session_tenant_1703000000_abc123
format_timestamp(datetime.now(), "America/Sao_Paulo")
from pywhatsweb.errors import *
# Códigos de erro padronizados
ErrorCode.E_SESSION_NOT_FOUND # 404
ErrorCode.E_AUTHENTICATION_FAILED # 401
ErrorCode.E_MEDIA_TOO_LARGE # 400
# Criar resposta de erro
error_response = create_error_response(ErrorCode.E_SESSION_NOT_FOUND)
TypeError: non-default argument follows default argument em dataclassesMediaMessage e Chat__default_config____python_support__, __dependencies__, etc.AttributeError: module 'websockets' has no attribute 'version_tuple'version_tuple__version__ quando version_tuple não está disponívelTypeError: non-default argument follows default argumentMediaMessage da classe Message__post_init__ funcionando corretamenteFAQs
Biblioteca Python para WhatsApp Web corporativo com sidecar Python integrado (padrão) ou Node.js (opcional)
We found that pywhatsweb demonstrated a healthy version release cadence and project activity because the last version was released less than a year ago. It has 1 open source maintainer collaborating on the project.
Did you know?
Socket for GitHub automatically highlights issues in each pull request and monitors the health of all your open source dependencies. Discover the contents of your packages and block harmful activity before you install or update your dependencies.
Research
A supply chain attack on Axios introduced a malicious dependency, plain-crypto-js@4.2.1, published minutes earlier and absent from the project’s GitHub releases.
Research
Malicious versions of the Telnyx Python SDK on PyPI delivered credential-stealing malware via a multi-stage supply chain attack.
Security News
TeamPCP is partnering with ransomware group Vect to turn open source supply chain attacks on tools like Trivy and LiteLLM into large-scale ransomware operations.