""" Follow-up automatico para o SDR Gi (Sweet Angels / Fesqua). Regras: - 1h sem resposta apos ultima msg outbound da Gi -> follow-up leve. - 24h sem resposta apos o follow-up de 1h -> segunda tentativa. - Nao envia se conversa foi encerrada (tag [[QUALIFICADO]] ou @sweetangelsgastronomia na ultima msg da Gi). - Somente em horario comercial (08h-20h America/Sao_Paulo). - Reset do estado a cada resposta nova do lead. - Maximo 2 follow-ups por conversa. State persistido em followup_state.json. """ from __future__ import annotations import json import logging import os import threading import time from datetime import datetime, timedelta from pathlib import Path from typing import Any, Callable from zoneinfo import ZoneInfo log = logging.getLogger("followup_manager") BASE_DIR = Path(__file__).resolve().parent STATE_PATH = BASE_DIR / "followup_state.json" TZ = ZoneInfo("America/Sao_Paulo") FOLLOWUP_1H_SECONDS = int(os.getenv("SDR_FOLLOWUP_1H_SECONDS", str(60 * 60))) FOLLOWUP_24H_SECONDS = int(os.getenv("SDR_FOLLOWUP_24H_SECONDS", str(24 * 60 * 60))) CHECK_INTERVAL_SECONDS = int(os.getenv("SDR_FOLLOWUP_CHECK_INTERVAL", "60")) BUSINESS_HOUR_START = int(os.getenv("SDR_FOLLOWUP_HOUR_START", "8")) BUSINESS_HOUR_END = int(os.getenv("SDR_FOLLOWUP_HOUR_END", "20")) class FollowupManager: def __init__( self, send_fn: Callable[[str, str, str | None], tuple[bool, int, str]], generate_followup_fn: Callable[[str, str], str], is_encerrada_fn: Callable[[str], bool], on_conversation_closed_fn: Callable[[str], None] | None = None, ): """ send_fn(contact_id, message, conversation_id) -> (ok, status, body) generate_followup_fn(contact_id, tipo) -> str, tipo in ("1h", "24h") is_encerrada_fn(contact_id) -> bool on_conversation_closed_fn(contact_id) -> None (chamada apos followup 24h) """ self.send_fn = send_fn self.generate_followup_fn = generate_followup_fn self.is_encerrada_fn = is_encerrada_fn self.on_conversation_closed_fn = on_conversation_closed_fn self._state: dict[str, dict[str, Any]] = {} self._lock = threading.Lock() self._thread: threading.Thread | None = None self._stop_event = threading.Event() self._load_state() # ------------------------------------------------------------------ # Persistencia # ------------------------------------------------------------------ def _load_state(self) -> None: if not STATE_PATH.exists(): self._state = {} return try: raw = json.loads(STATE_PATH.read_text(encoding="utf-8")) if isinstance(raw, dict): self._state = raw log.info("followup state carregado: %d contatos", len(self._state)) else: self._state = {} except (json.JSONDecodeError, OSError) as e: log.warning("state corrompido, recomecando: %s", e) self._state = {} def _save_state_locked(self) -> None: try: STATE_PATH.write_text( json.dumps(self._state, ensure_ascii=False, indent=2), encoding="utf-8", ) except OSError as e: log.warning("nao consegui salvar state: %s", e) # ------------------------------------------------------------------ # API publica # ------------------------------------------------------------------ def registrar_mensagem_enviada( self, contact_id: str, conversation_id: str | None = None, ) -> None: """Chamado apos a Gi enviar uma msg outbound pro lead.""" now = time.time() with self._lock: entry = self._state.get(contact_id, {}) entry["last_outbound"] = now # Nao reseta flags se o follow-up de 1h ja foi enviado. # O ciclo e: 1h follow-up → 24h encerramento. Nao recomeça. if not entry.get("followup_1h_sent"): entry["followup_1h_sent"] = False entry["followup_24h_sent"] = False entry["followup_1h_at"] = None if conversation_id: entry["conversation_id"] = conversation_id entry.setdefault("encerrado", False) self._state[contact_id] = entry self._save_state_locked() log.info("registrada msg outbound contact=%s", contact_id) def registrar_resposta_recebida(self, contact_id: str) -> None: """Chamado quando o lead responde (inbound).""" with self._lock: entry = self._state.get(contact_id) if not entry: return # Nao reseta flags se o follow-up de 1h ja foi enviado. if not entry.get("followup_1h_sent"): entry["followup_1h_sent"] = False entry["followup_24h_sent"] = False entry["followup_1h_at"] = None entry["last_outbound"] = None self._save_state_locked() log.info("lead respondeu contact=%s", contact_id) def marcar_encerrado(self, contact_id: str) -> None: with self._lock: entry = self._state.get(contact_id, {}) entry["encerrado"] = True self._state[contact_id] = entry self._save_state_locked() log.info("conversa marcada como encerrada contact=%s", contact_id) def set_conversation_id(self, contact_id: str, conversation_id: str | None) -> None: if not conversation_id: return with self._lock: entry = self._state.get(contact_id, {}) entry["conversation_id"] = conversation_id self._state[contact_id] = entry self._save_state_locked() # ------------------------------------------------------------------ # Loop background # ------------------------------------------------------------------ def start(self) -> None: if self._thread and self._thread.is_alive(): return self._stop_event.clear() t = threading.Thread(target=self._run_loop, name="followup-loop", daemon=True) self._thread = t t.start() log.info( "followup loop iniciado (check %ds, janela %02dh-%02dh %s)", CHECK_INTERVAL_SECONDS, BUSINESS_HOUR_START, BUSINESS_HOUR_END, TZ.key, ) def stop(self) -> None: self._stop_event.set() def _run_loop(self) -> None: while not self._stop_event.is_set(): try: self._tick() except Exception as e: log.exception("erro no tick do followup: %s", e) # dorme mas responde a stop self._stop_event.wait(CHECK_INTERVAL_SECONDS) # ------------------------------------------------------------------ # Tick / decisao # ------------------------------------------------------------------ def _is_business_hour(self, now_ts: float | None = None) -> bool: dt = datetime.fromtimestamp(now_ts or time.time(), TZ) return BUSINESS_HOUR_START <= dt.hour < BUSINESS_HOUR_END def _tick(self) -> None: now = time.time() # copia snapshot pra iterar sem segurar o lock durante I/O with self._lock: snapshot = list(self._state.items()) for contact_id, entry in snapshot: if entry.get("encerrado"): continue last_outbound = entry.get("last_outbound") if not last_outbound: continue # Verifica encerramento no historico (fonte da verdade) try: if self.is_encerrada_fn(contact_id): self.marcar_encerrado(contact_id) continue except Exception as e: log.warning("falha checando encerramento contact=%s: %s", contact_id, e) if not self._is_business_hour(now): continue elapsed = now - last_outbound # Follow-up 1h if ( not entry.get("followup_1h_sent") and elapsed >= FOLLOWUP_1H_SECONDS ): self._send_followup(contact_id, "1h") continue # Follow-up 24h: 24h apos o ultimo outbound (follow-up ou resposta) followup_1h_at = entry.get("followup_1h_at") last_outbound = entry.get("last_outbound") or 0 ref_24h = max(followup_1h_at or 0, last_outbound) if ( entry.get("followup_1h_sent") and not entry.get("followup_24h_sent") and ref_24h and (now - ref_24h) >= FOLLOWUP_24H_SECONDS ): self._send_followup(contact_id, "24h") continue def _send_followup(self, contact_id: str, tipo: str) -> None: with self._lock: entry = self._state.get(contact_id) if not entry: return conversation_id = entry.get("conversation_id") log.info("gerando followup %s contact=%s", tipo, contact_id) try: mensagem = self.generate_followup_fn(contact_id, tipo) except Exception as e: log.exception("falha gerando followup %s contact=%s: %s", tipo, contact_id, e) return if not mensagem or not mensagem.strip(): log.warning("followup %s vazio contact=%s", tipo, contact_id) return # Envia (pode ter [[BREAK]]) blocos = [b.strip() for b in mensagem.split("[[BREAK]]") if b.strip()] all_ok = True for i, bloco in enumerate(blocos): if i > 0: time.sleep(2) ok, status, _body = self.send_fn(contact_id, bloco, conversation_id) log.info( "followup %s bloco %d/%d contact=%s ok=%s status=%s preview=%r", tipo, i + 1, len(blocos), contact_id, ok, status, bloco[:80], ) if not ok: all_ok = False if not all_ok: log.warning("followup %s parcial/falhou contact=%s", tipo, contact_id) # Atualiza state now = time.time() with self._lock: entry = self._state.get(contact_id, {}) if tipo == "1h": entry["followup_1h_sent"] = True entry["followup_1h_at"] = now elif tipo == "24h": entry["followup_24h_sent"] = True entry["encerrado"] = True self._state[contact_id] = entry self._save_state_locked() if tipo == "24h" and self.on_conversation_closed_fn: try: self.on_conversation_closed_fn(contact_id) except Exception as e: log.exception("on_conversation_closed_fn falhou contact=%s: %s", contact_id, e)