""" GoHighLevel (Linkia) API client — Climb / TORQ. Wrappers: - send_whatsapp_message: envia msg WhatsApp de volta ao lead - get_conversation_messages: busca historico da conversa - get_contact: recupera contato + custom fields - update_contact_custom_fields: atualiza campos custom sdr__ - move_opportunity_stage: move opportunity entre stages do pipeline - list_custom_fields: lista custom fields da location - create_custom_field: cria custom field (usado no setup) """ from __future__ import annotations import logging from typing import Any import requests log = logging.getLogger("ghl_client") GHL_API_BASE = "https://services.leadconnectorhq.com" GHL_API_VERSION = "2021-07-28" def _headers(token: str) -> dict[str, str]: return { "Authorization": f"Bearer {token}", "Content-Type": "application/json", "Accept": "application/json", "Version": GHL_API_VERSION, } # --------------------------------------------------------------------------- # Mensagens # --------------------------------------------------------------------------- def send_via_dinastia( phone: str, message: str, dinastia_token: str, dinastia_base: str = "https://dinastiapi.agentesclimb.us", ) -> tuple[bool, int, str]: """Envia msg direto na Dinastia API — chega no WhatsApp do lead em tempo real. Complementa o send_whatsapp_message (que registra no CRM Linkia). Usar os dois em paralelo: Linkia pra histórico, Dinastia pra entrega. """ import re as _re fone = _re.sub(r"\D", "", phone or "") if not fone: return False, 0, "sem telefone" url = f"{dinastia_base.rstrip('/')}/chat/send/text" headers = {"token": dinastia_token, "Content-Type": "application/json"} body = {"Phone": fone, "Body": message, "NumberCheck": True} try: r = requests.post(url, headers=headers, json=body, timeout=20) ok = r.status_code in (200, 201) if ok: try: data = r.json() if isinstance(data, dict) and data.get("success") is False: log.warning("Dinastia send success=false phone=%s body=%s", fone, r.text[:200]) return False, r.status_code, r.text except Exception: pass else: log.warning("Dinastia send failed phone=%s status=%s body=%s", fone, r.status_code, r.text[:200]) return ok, r.status_code, r.text except requests.RequestException as e: log.exception("Dinastia send exception: %s", e) return False, 0, str(e) def send_whatsapp_message( contact_id: str, message: str, token: str, conversation_id: str | None = None, ) -> tuple[bool, int, str]: """Envia mensagem WhatsApp via GHL Conversations API.""" url = f"{GHL_API_BASE}/conversations/messages" payload: dict[str, Any] = { "type": "WhatsApp", "contactId": contact_id, "message": message, } if conversation_id: payload["conversationId"] = conversation_id try: r = requests.post(url, json=payload, headers=_headers(token), timeout=30) ok = r.ok if not ok: log.warning("GHL send failed contact=%s status=%s body=%s", contact_id, r.status_code, r.text[:400]) return ok, r.status_code, r.text except requests.RequestException as e: log.exception("GHL send exception: %s", e) return False, 0, str(e) def get_conversation_messages( conversation_id: str, token: str, limit: int = 30, types: str | None = None, ) -> list[dict[str, Any]]: """Busca historico de mensagens de uma conversa (mais recentes primeiro). `types` opcional: filtro tipo GHL (ex: 'TYPE_SMS,TYPE_WHATSAPP'). """ if not conversation_id: return [] url = f"{GHL_API_BASE}/conversations/{conversation_id}/messages" params: dict[str, Any] = {"limit": limit} if types: params["type"] = types try: r = requests.get(url, headers=_headers(token), params=params, timeout=20) if not r.ok: log.warning("GHL get_conversation_messages failed status=%s body=%s", r.status_code, r.text[:400]) return [] data = r.json() or {} # A API retorna {"messages": {"messages": [...], "lastMessageId": ...}} em algumas # verificacoes, ou {"messages": [...]} direto. Trata ambos. msgs = data.get("messages") if isinstance(msgs, dict): msgs = msgs.get("messages", []) return msgs or [] except requests.RequestException as e: log.exception("GHL get_conversation_messages exception: %s", e) return [] def search_conversations( contact_id: str, location_id: str, token: str, limit: int = 1, ) -> list[dict[str, Any]]: """Procura conversas de um contato (mais recente primeiro). Endpoint: GET /conversations/search?locationId=X&contactId=Y&limit=N Retorna lista de {id, contactId, lastMessageBody, ...}. """ if not contact_id or not location_id: return [] url = f"{GHL_API_BASE}/conversations/search" params = { "locationId": location_id, "contactId": contact_id, "limit": limit, } try: r = requests.get(url, headers=_headers(token), params=params, timeout=15) if not r.ok: log.warning("GHL search_conversations failed status=%s body=%s", r.status_code, r.text[:400]) return [] data = r.json() or {} convs = data.get("conversations") or data.get("data") or [] return convs or [] except requests.RequestException as e: log.exception("GHL search_conversations exception: %s", e) return [] def fetch_last_inbound_message( contact_id: str, location_id: str, token: str, max_attempts: int = 5, backoff_s: float = 3.0, ) -> tuple[str | None, str | None]: """Busca a ultima mensagem INBOUND do contato via GHL, com retry. Retorna (body, conversation_id) — ambos podem ser None se nao houver. Fluxo: search_conversations -> pega conv mais recente -> lista mensagens filtrando WhatsApp/SMS -> pega a primeira inbound (mais recente). Race condition tratada: o GHL leva 3-15s pra indexar conversa recém-criada pelo endpoint /conversations/search. Se nao achar, faz retry com backoff. """ import time conv_id: str | None = None for attempt in range(1, max_attempts + 1): convs = search_conversations(contact_id, location_id, token, limit=1) if not convs: if attempt < max_attempts: log.info("[fetch_last_inbound] contact=%s tentativa %d/%d sem conversa GHL, aguardando %.1fs", contact_id, attempt, max_attempts, backoff_s) time.sleep(backoff_s) continue log.info("[fetch_last_inbound] contact=%s sem conversa GHL apos %d tentativas", contact_id, max_attempts) return None, None conv = convs[0] conv_id = conv.get("id") or conv.get("_id") if not conv_id: log.info("[fetch_last_inbound] contact=%s conv sem id", contact_id) return None, None msgs = get_conversation_messages( conv_id, token, limit=10, types="TYPE_SMS,TYPE_WHATSAPP", ) for m in msgs: if (m.get("direction") or "").lower() != "inbound": continue body = (m.get("body") or "").strip() if body: return body, conv_id # conv existe mas sem msg inbound util ainda — retry if attempt < max_attempts: log.info("[fetch_last_inbound] contact=%s conv=%s sem msg inbound util, aguardando %.1fs (tentativa %d/%d)", contact_id, conv_id, backoff_s, attempt, max_attempts) time.sleep(backoff_s) continue log.info("[fetch_last_inbound] contact=%s conv=%s sem msg inbound util apos %d tentativas", contact_id, conv_id, max_attempts) return None, conv_id # --------------------------------------------------------------------------- # Contact + custom fields # --------------------------------------------------------------------------- def get_contact(contact_id: str, token: str) -> dict[str, Any] | None: url = f"{GHL_API_BASE}/contacts/{contact_id}" try: r = requests.get(url, headers=_headers(token), timeout=15) if not r.ok: log.warning("GHL get_contact failed status=%s body=%s", r.status_code, r.text[:400]) return None data = r.json() or {} return data.get("contact") or data except requests.RequestException as e: log.exception("GHL get_contact exception: %s", e) return None def update_contact_custom_fields( contact_id: str, fields: dict[str, Any], token: str, field_id_map: dict[str, str] | None = None, ) -> bool: """ Atualiza custom fields de um contato. `fields` = {"sdr__nome_lead": "Marcos", "sdr__cidade": "BH", ...} `field_id_map` = {"sdr__nome_lead": "aBc123XyZ", ...} (obrigatorio na v2 da API) """ if not fields: return True if not field_id_map: log.warning("update_contact_custom_fields sem field_id_map — nada feito") return False payload_fields = [] for key, val in fields.items(): fid = field_id_map.get(key) if not fid: log.warning("campo %s sem ID mapeado — ignorando", key) continue payload_fields.append({"id": fid, "field_value": val}) if not payload_fields: return False url = f"{GHL_API_BASE}/contacts/{contact_id}" payload = {"customFields": payload_fields} try: r = requests.put(url, json=payload, headers=_headers(token), timeout=20) if not r.ok: log.warning("GHL update_contact failed status=%s body=%s", r.status_code, r.text[:400]) return r.ok except requests.RequestException as e: log.exception("GHL update_contact exception: %s", e) return False # --------------------------------------------------------------------------- # Opportunities / pipeline # --------------------------------------------------------------------------- def find_opportunity_for_contact( contact_id: str, pipeline_id: str, token: str, location_id: str, ) -> dict[str, Any] | None: """Procura opportunity ativa do contato no pipeline.""" url = f"{GHL_API_BASE}/opportunities/search" params = { "location_id": location_id, "contact_id": contact_id, "pipeline_id": pipeline_id, "limit": 20, } try: r = requests.get(url, headers=_headers(token), params=params, timeout=15) if not r.ok: log.warning("GHL search opportunity failed status=%s body=%s", r.status_code, r.text[:400]) return None data = r.json() or {} opps = data.get("opportunities", []) return opps[0] if opps else None except requests.RequestException as e: log.exception("GHL search opportunity exception: %s", e) return None def create_opportunity( contact_id: str, pipeline_id: str, stage_id: str, name: str, token: str, location_id: str, ) -> dict[str, Any] | None: url = f"{GHL_API_BASE}/opportunities/" payload = { "pipelineId": pipeline_id, "locationId": location_id, "name": name, "pipelineStageId": stage_id, "status": "open", "contactId": contact_id, } try: r = requests.post(url, json=payload, headers=_headers(token), timeout=20) if not r.ok: log.warning("GHL create opportunity failed status=%s body=%s", r.status_code, r.text[:400]) return None return (r.json() or {}).get("opportunity") except requests.RequestException as e: log.exception("GHL create opportunity exception: %s", e) return None def move_opportunity_stage( opportunity_id: str, stage_id: str, pipeline_id: str, token: str, ) -> bool: url = f"{GHL_API_BASE}/opportunities/{opportunity_id}" payload = { "pipelineId": pipeline_id, "pipelineStageId": stage_id, } try: r = requests.put(url, json=payload, headers=_headers(token), timeout=15) if not r.ok: log.warning("GHL move opportunity failed status=%s body=%s", r.status_code, r.text[:400]) return r.ok except requests.RequestException as e: log.exception("GHL move opportunity exception: %s", e) return False # --------------------------------------------------------------------------- # Custom fields management (usado no setup one-shot) # --------------------------------------------------------------------------- def list_custom_fields(location_id: str, token: str) -> list[dict[str, Any]]: """Lista todos os custom fields da location.""" url = f"{GHL_API_BASE}/locations/{location_id}/customFields" try: r = requests.get(url, headers=_headers(token), timeout=15) if not r.ok: log.warning("GHL list_custom_fields failed status=%s body=%s", r.status_code, r.text[:400]) return [] return (r.json() or {}).get("customFields", []) except requests.RequestException as e: log.exception("GHL list_custom_fields exception: %s", e) return [] def create_custom_field( location_id: str, name: str, field_key: str, data_type: str, token: str, picklist_options: list[str] | None = None, ) -> dict[str, Any] | None: """ Cria custom field na location. data_type: TEXT, LARGE_TEXT, NUMERICAL, PHONE, MONETORY, CHECKBOX, DATE, SINGLE_OPTIONS (dropdown), MULTIPLE_OPTIONS, RADIO, FILE_UPLOAD """ url = f"{GHL_API_BASE}/locations/{location_id}/customFields" payload: dict[str, Any] = { "name": name, "dataType": data_type, "fieldKey": field_key, } if picklist_options: # GHL v2 espera options como array de strings payload["options"] = list(picklist_options) try: r = requests.post(url, json=payload, headers=_headers(token), timeout=20) if r.status_code == 400 and "already" in r.text.lower(): log.info("campo %s ja existe", field_key) return None if not r.ok: log.warning("GHL create_custom_field failed status=%s body=%s", r.status_code, r.text[:400]) return None return (r.json() or {}).get("customField") except requests.RequestException as e: log.exception("GHL create_custom_field exception: %s", e) return None def add_contact_note( contact_id: str, body: str, token: str, ) -> tuple[bool, int, str]: """Cria uma NOTA interna no contato (nao dispara canal WhatsApp/SMS). Usado pra registrar as respostas do Bruno no CRM sem acionar API oficial. Endpoint: POST /contacts/{contactId}/notes """ url = f"{GHL_API_BASE}/contacts/{contact_id}/notes" payload = {"body": body} try: r = requests.post(url, json=payload, headers=_headers(token), timeout=15) ok = r.ok if not ok: log.warning("GHL add_note failed contact=%s status=%s body=%s", contact_id, r.status_code, r.text[:300]) return ok, r.status_code, r.text except requests.RequestException as e: log.exception("GHL add_note exception: %s", e) return False, 0, str(e)