""" TORQ SDR Bruno — webhook Flask porta 8920. Fluxo: 1. GHL manda POST /webhook (mensagem inbound WhatsApp) 2. Responde 200 imediatamente 3. Idempotencia via message_id (dedup em memoria) 4. Debounce: acumula msgs do mesmo contato durante DEBOUNCE_SECONDS 5. Busca historico da conversa via GHL API + contexto do contato 6. Chama Bruno (Claude Sonnet 4.6 com prompt caching) 7. Parseia resposta -> texto + meta (campos, stage, notify) 8. Envia texto pro lead via GHL (com typing delay + blocos [[BREAK]]) 9. Aplica meta: atualiza custom fields + move opportunity + notifica Renato Log estruturado em /opt/mia/logs/sdr_bruno.log """ from __future__ import annotations import json import logging import os import sys import threading import time from collections import OrderedDict from logging.handlers import RotatingFileHandler from pathlib import Path from typing import Any from flask import Flask, jsonify, request BASE_DIR = Path(__file__).resolve().parent # --------------------------------------------------------------------------- # Logging: stdout + /opt/mia/logs/sdr_bruno.log # --------------------------------------------------------------------------- LOG_FILE = Path("/opt/mia/logs/sdr_bruno.log") LOG_FILE.parent.mkdir(parents=True, exist_ok=True) fmt = logging.Formatter("%(asctime)s [%(levelname)s] %(name)s: %(message)s") file_handler = RotatingFileHandler(LOG_FILE, maxBytes=10 * 1024 * 1024, backupCount=5, encoding="utf-8") file_handler.setFormatter(fmt) stream_handler = logging.StreamHandler(sys.stdout) stream_handler.setFormatter(fmt) logging.basicConfig(level=logging.INFO, handlers=[file_handler, stream_handler]) log = logging.getLogger("torq_sdr") # --------------------------------------------------------------------------- # Env loading # --------------------------------------------------------------------------- try: from dotenv import load_dotenv load_dotenv("/opt/mia/.env") load_dotenv("/opt/mia-bot/.env") load_dotenv(BASE_DIR / "torq_bruno.env") except ImportError: for envfile in ("/opt/mia/.env", "/opt/mia-bot/.env", str(BASE_DIR / "torq_bruno.env")): if not os.path.exists(envfile): continue with open(envfile, encoding="utf-8") as f: for raw in f: line = raw.strip() if not line or line.startswith("#") or "=" not in line: continue k, v = line.split("=", 1) os.environ.setdefault(k.strip(), v.strip()) # Config LINKIA_TOKEN = os.getenv("LINKIA_TOKEN", "").strip() LINKIA_LOCATION_ID = os.getenv("LINKIA_LOCATION_ID", "").strip() PIPELINE_ID = os.getenv("PIPELINE_ID", "").strip() STAGE_MAP = { "novo_lead": os.getenv("STAGE_NOVO_LEAD", ""), "qualificado": os.getenv("STAGE_QUALIFICADO", ""), "test_drive_agendado": os.getenv("STAGE_TEST_DRIVE", ""), "proposta": os.getenv("STAGE_PROPOSTA", ""), "perdido": os.getenv("STAGE_PERDIDO", ""), "ganhamos": os.getenv("STAGE_GANHAMOS", ""), "perdemos": os.getenv("STAGE_PERDEMOS", ""), } DEBOUNCE_SECONDS = float(os.getenv("SDR_DEBOUNCE_SECONDS", "6.0")) TYPING_DELAY_SECONDS = float(os.getenv("SDR_TYPING_DELAY_SECONDS", "3.0")) BLOCK_DELAY_SECONDS = float(os.getenv("SDR_BLOCK_DELAY_SECONDS", "4.0")) NOTIFY_RENATO = os.getenv("NOTIFY_RENATO", "true").lower() == "true" if not LINKIA_TOKEN: log.error("LINKIA_TOKEN ausente — webhook nao vai conseguir responder") # Carrega map de custom fields CF_MAP_PATH = BASE_DIR / "custom_fields.json" CUSTOM_FIELD_ID_MAP: dict[str, str] = {} if CF_MAP_PATH.exists(): try: CUSTOM_FIELD_ID_MAP = json.loads(CF_MAP_PATH.read_text(encoding="utf-8")) log.info("carregado custom_fields.json com %d entradas", len(CUSTOM_FIELD_ID_MAP)) except Exception as e: log.warning("nao consegui carregar custom_fields.json: %s", e) else: log.warning("custom_fields.json nao existe — rode setup_custom_fields.py") # Import dos modulos internos (depois do load_dotenv) from bruno_agent import gerar_resposta # noqa: E402 from ghl_client import ( # noqa: E402 send_whatsapp_message, send_via_dinastia, add_contact_note, get_conversation_messages, get_contact, update_contact_custom_fields, find_opportunity_for_contact, create_opportunity, move_opportunity_stage, fetch_last_inbound_message, ) # Dinastia (opcional) — envio DIRETO pro WhatsApp em tempo real DINASTIA_TOKEN = os.getenv("DINASTIA_TOKEN", "").strip() DINASTIA_BASE = os.getenv("DINASTIA_BASE", "https://dinastiapi.agentesclimb.us").strip() app = Flask(__name__) # --------------------------------------------------------------------------- # Idempotencia (dedup message_id) # --------------------------------------------------------------------------- _seen_msg_ids: OrderedDict[str, float] = OrderedDict() _seen_lock = threading.Lock() DEDUP_TTL_S = 3600 DEDUP_MAX = 5000 def _seen_already(msg_id: str) -> bool: if not msg_id: return False now = time.time() with _seen_lock: # limpa antigos while _seen_msg_ids and (now - next(iter(_seen_msg_ids.values())) > DEDUP_TTL_S or len(_seen_msg_ids) > DEDUP_MAX): _seen_msg_ids.popitem(last=False) if msg_id in _seen_msg_ids: return True _seen_msg_ids[msg_id] = now return False # --------------------------------------------------------------------------- # Debounce buffer # --------------------------------------------------------------------------- _pending: dict[str, dict] = {} _pending_lock = threading.Lock() def _fire_contact(contact_id: str) -> None: """Processa msgs acumuladas do contato apos debounce.""" with _pending_lock: data = _pending.pop(contact_id, None) if not data: return messages_txt = data["messages"] conversation_id = data["conversation_id"] contact_meta = data["contact_meta"] mensagem = "\n".join(messages_txt) log.info("[processando] contact=%s msgs=%d preview=%r", contact_id, len(messages_txt), mensagem[:200]) # 1) Busca contexto do contato (custom fields ja preenchidos) custom_fields: dict[str, Any] = {} contato = get_contact(contact_id, LINKIA_TOKEN) if contato: # customFields vem como lista [{"id": "...", "value": "..."}] cf_list = contato.get("customFields", []) or [] # Reverse map: id -> field_key id_to_key = {v: k for k, v in CUSTOM_FIELD_ID_MAP.items()} for cf in cf_list: fid = cf.get("id") fkey = id_to_key.get(fid) if fkey: custom_fields[fkey] = cf.get("value") or cf.get("fieldValue") # 2) Busca historico GHL (fonte da verdade) historico_ghl: list[dict[str, str]] = [] if conversation_id: ghl_msgs = get_conversation_messages(conversation_id, LINKIA_TOKEN, limit=30) # normaliza pro formato Claude: user (lead=inbound), assistant (Bruno=outbound) for m in reversed(ghl_msgs): # GHL vem mais recente primeiro body = (m.get("body") or "").strip() if not body: continue direction = (m.get("direction") or "").lower() role = "user" if direction == "inbound" else "assistant" historico_ghl.append({"role": role, "content": body}) # 3) Chama Bruno try: texto, meta = gerar_resposta( contact_id=contact_id, mensagem_lead=mensagem, contact_meta=contact_meta, custom_fields=custom_fields, historico_ghl=historico_ghl or None, ) except Exception as e: log.exception("erro chamando Bruno: %s", e) return if not texto: log.warning("Bruno retornou texto vazio contact=%s", contact_id) return log.info("[bruno] contact=%s texto=%r meta=%s", contact_id, texto[:200], meta) # 4) Envia msg pro lead (blocos [[BREAK]] com typing delay) blocos = [b.strip() for b in texto.split("[[BREAK]]") if b.strip()] if not blocos: blocos = [texto] time.sleep(TYPING_DELAY_SECONDS) phone_para_dinastia = contact_meta.get("phone") or "" # ENVIO EXCLUSIVAMENTE PELA DINASTIA (WhatsApp nao-oficial de teste). # O envio via GHL /conversations/messages foi DESATIVADO em 2026-08-10 porque # dispara pelo canal WhatsApp API OFICIAL da location Climb Digital, o que # gera custo real por conversa e pode atingir numeros de producao. # Se um dia a location for exclusiva de teste, reativar o send_whatsapp_message. for i, bloco in enumerate(blocos): if i > 0: time.sleep(BLOCK_DELAY_SECONDS) if DINASTIA_TOKEN and phone_para_dinastia: ok_d, status_d, _ = send_via_dinastia( phone=phone_para_dinastia, message=bloco, dinastia_token=DINASTIA_TOKEN, dinastia_base=DINASTIA_BASE, ) log.info("[enviado_dinastia] bloco %d/%d ok=%s status=%s phone=%s", i + 1, len(blocos), ok_d, status_d, phone_para_dinastia) else: log.warning("[nao_enviado] DINASTIA_TOKEN ou phone ausente — msg nao enviada contact=%s", contact_id) # Registra a resposta como NOTA interna no CRM Linkia (nao dispara API oficial, # zero custo Meta). Assim o Renato ve os 2 lados da conversa no painel. nota_body = f"🤖 Bruno respondeu (via Dinastia):\n{bloco}" ok_n, status_n, _ = add_contact_note(contact_id, nota_body, LINKIA_TOKEN) log.info("[nota_crm] bloco %d/%d ok=%s status=%s contact=%s", i + 1, len(blocos), ok_n, status_n, contact_id) # 5) Aplica meta: campos + stage + notify _aplicar_meta(contact_id, meta, contact_meta) def _aplicar_meta(contact_id: str, meta: dict[str, Any], contact_meta: dict[str, Any]) -> None: if not meta: return # Atualiza custom fields campos = meta.get("campos") or {} if campos and CUSTOM_FIELD_ID_MAP: # Filtra so keys que existem no mapa valid = {k: v for k, v in campos.items() if k in CUSTOM_FIELD_ID_MAP} if valid: ok = update_contact_custom_fields( contact_id=contact_id, fields=valid, token=LINKIA_TOKEN, field_id_map=CUSTOM_FIELD_ID_MAP, ) log.info("[cf_update] contact=%s ok=%s fields=%s", contact_id, ok, list(valid.keys())) # Move stage stage_key = meta.get("stage") if stage_key and STAGE_MAP.get(stage_key): stage_id = STAGE_MAP[stage_key] opp = find_opportunity_for_contact( contact_id=contact_id, pipeline_id=PIPELINE_ID, token=LINKIA_TOKEN, location_id=LINKIA_LOCATION_ID, ) if not opp: # cria a opportunity se nao existir nome = contact_meta.get("first_name") or "Lead" opp = create_opportunity( contact_id=contact_id, pipeline_id=PIPELINE_ID, stage_id=STAGE_MAP.get("novo_lead") or stage_id, name=f"{nome} (SDR Bruno)", token=LINKIA_TOKEN, location_id=LINKIA_LOCATION_ID, ) log.info("[opp_created] contact=%s opp=%s", contact_id, opp.get("id") if opp else None) if opp and opp.get("id"): ok = move_opportunity_stage( opportunity_id=opp["id"], stage_id=stage_id, pipeline_id=PIPELINE_ID, token=LINKIA_TOKEN, ) log.info("[stage_move] contact=%s opp=%s stage=%s ok=%s", contact_id, opp["id"], stage_key, ok) # Notifica Renato via outbox (Mia bot) if meta.get("notificar_renato") and NOTIFY_RENATO: _notificar_renato(contact_id, contact_meta, meta, campos) def _notificar_renato(contact_id: str, contact_meta: dict[str, Any], meta: dict[str, Any], campos: dict[str, Any]) -> None: """Cria mensagem no outbox da Mia -> chega no Telegram do Renato.""" outbox = Path("/opt/mia-bot/outbox") if not outbox.exists(): log.warning("outbox nao existe — nao notifiquei Renato") return nome = contact_meta.get("first_name") or campos.get("sdr__nome_lead") or "sem nome" phone = contact_meta.get("phone") or "sem phone" cidade = campos.get("sdr__cidade", "") perfil = campos.get("sdr__perfil", "") modelo = campos.get("sdr__modelo_interesse_principal", "") stage = meta.get("stage", "") texto = ( f"[Bruno SDR TORQ] Lead pra voce, chefe:\n" f"{nome} ({phone})\n" f"cidade: {cidade} | perfil: {perfil} | modelo: {modelo}\n" f"stage: {stage}\n" f"contact_id: {contact_id}" ) try: fname = outbox / f"{int(time.time() * 1e9)}.json" fname.write_text(json.dumps({"text": texto}, ensure_ascii=False), encoding="utf-8") log.info("[notify_renato] enviado outbox=%s", fname.name) except Exception as e: log.exception("falha ao notificar Renato: %s", e) # --------------------------------------------------------------------------- # Extracao do payload GHL # --------------------------------------------------------------------------- def _get_first(d: dict, *keys: str, default: Any = None) -> Any: for k in keys: if "." in k: cur: Any = d ok = True for part in k.split("."): if isinstance(cur, dict) and part in cur: cur = cur[part] else: ok = False break if ok and cur not in (None, "", []): return cur elif k in d and d[k] not in (None, "", []): return d[k] return default def extrair_payload(payload: dict) -> dict: return { "message_id": _get_first(payload, "id", "messageId", "message.id"), "contact_id": _get_first(payload, "contactId", "contact_id", "contact.id"), "conversation_id": _get_first(payload, "conversationId", "conversation_id"), "mensagem": _get_first(payload, "body", "message", "messageBody", "text", "message.body"), "first_name": _get_first(payload, "firstName", "first_name", "full_name", "contact.firstName"), "last_name": _get_first(payload, "lastName", "last_name", "contact.lastName"), "phone": _get_first(payload, "phone", "contact.phone"), "direction": _get_first(payload, "direction", default="inbound"), "type": _get_first(payload, "type", "messageType", default="WhatsApp"), "location_id": _get_first(payload, "locationId", "location_id", "location.id"), } # --------------------------------------------------------------------------- # Rotas # --------------------------------------------------------------------------- @app.route("/health", methods=["GET"]) def health(): return jsonify({ "ok": True, "service": "torq_sdr_bruno", "port": int(os.getenv("PORT", "8920")), "linkia_token_ok": bool(LINKIA_TOKEN), "custom_fields_loaded": len(CUSTOM_FIELD_ID_MAP), "debounce_s": DEBOUNCE_SECONDS, "model": os.getenv("BRUNO_MODEL", "claude-sonnet-4-6"), }) @app.route("/webhook", methods=["POST"]) def webhook(): try: payload = request.get_json(force=True, silent=False) or {} except Exception as e: log.warning("payload nao-JSON: %s raw=%s", e, request.data[:400]) return jsonify({"ok": False, "error": "invalid_json"}), 400 log.info("[webhook] recebido: %s", json.dumps(payload, ensure_ascii=False)[:800]) dados = extrair_payload(payload) # Idempotencia if dados["message_id"] and _seen_already(str(dados["message_id"])): log.info("[dedup] msg %s ja processada — ignorando", dados["message_id"]) return jsonify({"ok": True, "skipped": "duplicate"}), 200 if not dados["contact_id"]: log.warning("sem contact_id: %s", json.dumps(payload, ensure_ascii=False)[:300]) return jsonify({"ok": False, "error": "missing_contact_id"}), 200 if str(dados["direction"]).lower() == "outbound": log.info("outbound — ignorando (evita loop)") return jsonify({"ok": True, "skipped": "outbound"}), 200 contact_id = str(dados["contact_id"]) conversation_id = str(dados["conversation_id"]) if dados["conversation_id"] else None # Se payload nao trouxe mensagem (ex: trigger 'Contato Criado' do GHL), # tenta puxar a ultima msg inbound do lead via GHL API. if not dados["mensagem"]: loc_id = dados.get("location_id") or LINKIA_LOCATION_ID if not loc_id: log.info("sem mensagem no payload e sem location_id — ignorando contact=%s", contact_id) return jsonify({"ok": True, "skipped": "no_message_no_location"}), 200 log.info("[fallback_api] payload sem msg — buscando ultima inbound contact=%s loc=%s", contact_id, loc_id) body, conv_id = fetch_last_inbound_message( contact_id=contact_id, location_id=loc_id, token=LINKIA_TOKEN, ) if not body: log.info("sem msg no payload nem na API — ignorando contact=%s", contact_id) return jsonify({"ok": True, "skipped": "no_message_api"}), 200 dados["mensagem"] = body if conv_id and not conversation_id: conversation_id = conv_id log.info("[fallback_api] contact=%s conv=%s recuperou msg preview=%r", contact_id, conversation_id, body[:120]) mensagem = str(dados["mensagem"]).strip() if not mensagem: log.info("mensagem vazia apos strip — ignorando contact=%s", contact_id) return jsonify({"ok": True, "skipped": "empty_message"}), 200 contact_meta = { "first_name": dados.get("first_name"), "last_name": dados.get("last_name"), "phone": dados.get("phone"), "contact_id": contact_id, "conversation_id": conversation_id, } with _pending_lock: if contact_id in _pending: existing = _pending[contact_id] if existing.get("timer"): existing["timer"].cancel() existing["messages"].append(mensagem) existing["conversation_id"] = conversation_id existing["contact_meta"] = contact_meta else: _pending[contact_id] = { "messages": [mensagem], "timer": None, "conversation_id": conversation_id, "contact_meta": contact_meta, } timer = threading.Timer(DEBOUNCE_SECONDS, _fire_contact, args=(contact_id,)) timer.daemon = True _pending[contact_id]["timer"] = timer timer.start() return jsonify({"ok": True, "queued": True, "debounce_s": DEBOUNCE_SECONDS}), 200 @app.route("/", methods=["GET"]) def index(): return jsonify({ "service": "TORQ SDR Bruno", "endpoints": { "health": "GET /health", "webhook": "POST /webhook", }, }) if __name__ == "__main__": port = int(os.getenv("PORT", "8920")) host = os.getenv("HOST", "0.0.0.0") log.info("subindo Flask em %s:%s (debounce=%.1fs model=%s)", host, port, DEBOUNCE_SECONDS, os.getenv("BRUNO_MODEL")) app.run(host=host, port=port, debug=False, threaded=True)