#!/usr/bin/env python3 """ generate_commercial_report_v2.py =============================== Gerador de relatórios comerciais Hermes Insights — Método 7P. Uso: python3 generate_commercial_report_v2.py [agente_id] Exemplos: python3 generate_commercial_report_v2.py # todos os agentes python3 generate_commercial_report_v2.py 9 # só Vinicius Sales (id=9) python3 generate_commercial_report_v2.py 1,9 # Anderson e Vinicius Dependências (pip install): requests, google-generativeai Variáveis de ambiente: GEMINI_API_KEY — chave da API Gemini (obrigatório) CHATWOOT_TOKEN — token da API do Chatwoot (opcional, usa o do Anderson por padrão) CHATWOOT_URL — URL base do Chatwoot (padrão: https://chatwoot.rochasalesseguros.com.br) ACCOUNT_ID — ID da conta (padrão: 1) Arquitetura: 1. Busca todos os agentes do Chatwoot via API REST 2. Para cada agente, busca conversas atribuídas 3. Para cada conversa, busca todas as mensagens 4. Envia transcrição ao Gemini com prompt 7P 5. Gemini retorna scores e análise por P 6. Monta relatório HTML com visual dark/gold 7. Salva em /var/www/hermes-insights/relatorio_{agente}.html """ import os, sys, time, json, re, requests import urllib3 urllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarning) from datetime import datetime from urllib.parse import urljoin # ─── CONFIG ────────────────────────────────────────────────────────────────── CHATWOOT_URL = os.getenv("CHATWOOT_URL", "https://chatwoot.rochasalesseguros.com.br") CHATWOOT_TOKEN = os.getenv("CHATWOOT_TOKEN", "2AfB8VhS79FTmxG1dL8774RJ") # Anderson (admin) ACCOUNT_ID = os.getenv("ACCOUNT_ID", "1") GEMINI_API_KEY = os.getenv("GEMINI_API_KEY", "") OUTPUT_DIR = "/var/www/hermes-insights" HEADERS = { "api_access_token": CHATWOOT_TOKEN, "Content-Type": "application/json", } LOCK_FILE = "/tmp/gemini_7p.lock" MAX_CONV_PER_AGENT = 50 # máximo de conversas por agente (mais recentes) MAX_MSG_PER_CONV = 200 # máximo de mensagens por conversa GEMINI_MODEL = "gemini-2.0-flash" REQUEST_TIMEOUT = 120 # segundos MAX_BACKOFF = 480 # segundos INITIAL_BACKOFF = 30 # segundos # ─── PROMPT 7P ─────────────────────────────────────────────────────────────── PROMPT_7P = """Você é um ANALISTA COMERCIAL EXPERT do método 7P de vendas de seguros. Avalie a conversa abaixo e retorne APENAS um JSON válido (sem markdown, sem código, sem explanations). O método 7P: - P1 APRESENTAÇÃO: O vendedor apresenta seu trabalho e enfatiza que não vende a todo custo. - P2 BREVE HISTÓRIA: Criar conexão genuína com o cliente (perguntas pessoais, empatia). - P3 ENTENDER A DOR: Deixar o cliente falar e perguntar: "se não resolver agora, qual o impacto?". - P4 COMBINADO: Combinar com o cliente que ao final ele diz sim ou não — sem pressão. - P5 APRESENTAÇÃO DA SOLUÇÃO: Mostrar benefícios das operadoras antes dos valores. Deixar cliente falar após ver preço. - P6 FECHAMENTO: Perguntar opinião, cobrar combinado, tratar objeções, criar urgência (validade 2 dias). - P7 VENDA EXTRA: Levantar seguros existentes, upsell, pedir indicações. Retorne JSON com esta estrutura exata: { "scores": { "P1_Apresentacao": {"nota": 0-10, "evidencia": "trecho da conversa", "comentario": "breve análise"}, "P2_BreveHistoria": {"nota": 0-10, "evidencia": "trecho da conversa", "comentario": "breve análise"}, "P3_EntenderDor": {"nota": 0-10, "evidencia": "trecho da conversa", "comentario": "breve análise"}, "P4_Combinado": {"nota": 0-10, "evidencia": "trecho da conversa", "comentario": "breve análise"}, "P5_Solucao": {"nota": 0-10, "evidencia": "trecho da conversa", "comentario": "breve análise"}, "P6_Fechamento": {"nota": 0-10, "evidencia": "trecho da conversa", "comentario": "breve análise"}, "P7_VendaExtra": {"nota": 0-10, "evidencia": "trecho da conversa", "comentario": "breve análise"} }, "score_geral": 0-10, "classificacao": "Elite|Bom|Regular|Fraco|Crítico", "pontos_fortes": ["item1", "item2"], "pontos_melhoria": ["item1", "item2"], "objetivos_nao_tratados": ["item1", "item2"], "proximos_passos": ["item1", "item2"], "resumo_venda": "Resumo de uma linha do comportamento do vendedor nesta conversa", "nivel_engajamento": "Alto|Médio|Baixo" (Baseado em mensagens por hora, follow-ups, objeções tratadas) } IMPORTANTE: - nota é de 0 a 10 (0=não identificado, 10=exemplar) - evidencia deve ser um trecho REAL da conversa (máx 150 caracteres) - Se o P não foi identificado, coloque nota 0 e evidência "" - Responda APENAS com JSON válido. Nada mais. CONVERSA: {conversacao_text} """ # ─── UTILS ─────────────────────────────────────────────────────────────────── def acquire_lock(): """Bloqueia execução para evitar konkurrent Gemini calls.""" if os.path.exists(LOCK_FILE): # Lê PID e verifica se ainda roda try: with open(LOCK_FILE) as f: pid = int(f.read().strip()) os.kill(pid, 0) # raises OSError se morto print(f"[LOCK] Processo {pid} ainda roda. Abortando.") return False except (ValueError, OSError): print("[LOCK] Lock obsoleto, removendo.") os.remove(LOCK_FILE) with open(LOCK_FILE, "w") as f: f.write(str(os.getpid())) return True def release_lock(): try: os.remove(LOCK_FILE) except FileNotFoundError: pass def chatwoot_get(endpoint, params=None): """GET na API do Chatwoot com retry.""" url = urljoin(CHATWOOT_URL + "/", endpoint.lstrip("/")) for attempt in range(5): try: r = requests.get(url, headers=HEADERS, params=params, timeout=REQUEST_TIMEOUT, verify=False) if r.status_code == 429: backoff = min(INITIAL_BACKOFF * (2 ** attempt), MAX_BACKOFF) print(f"[429] Rate limit. Aguardando {backoff}s...") time.sleep(backoff) continue if r.status_code == 401: print(f"[401] Token inválido ou sem permissão para {endpoint}") return None r.raise_for_status() return r.json() except requests.exceptions.RequestException as e: backoff = min(INITIAL_BACKOFF * (2 ** attempt), MAX_BACKOFF) print(f"[ERROR] {e}. Retry {attempt+1}/5. Backoff {backoff}s") time.sleep(backoff) return None def fetch_messages(conversation_id): """Busca todas as mensagens de uma conversa (paginação).""" all_msgs = [] page = 1 while True: data = chatwoot_get( f"accounts/{ACCOUNT_ID}/conversations/{conversation_id}/messages", params={"per_page": 200, "page": page} ) if not data: break msgs = data.get("data", []) all_msgs.extend(msgs) if len(msgs) < 200: break page += 1 time.sleep(0.3) return all_msgs def build_transcript(messages, agent_name): """Converte lista de mensagens em texto legível.""" lines = [] for m in sorted(messages, key=lambda x: x.get("created_at", "")): sender = m.get("sender_type", "") content = m.get("content", "") or m.get("processed_message_content", "") or "" created = m.get("created_at", "")[:16] msg_type = m.get("message_type", -1) if not content.strip(): continue # message_type: 0=incoming (contact), 1=outgoing (agent), 2=activity if msg_type == 2: lines.append(f"[{created}] *** {content}") elif sender == "User" or msg_type == 1: lines.append(f"[{created}] VENDEDOR: {content[:500]}") elif sender == "Contact": lines.append(f"[{created}] CLIENTE: {content[:500]}") else: lines.append(f"[{created}] {content[:500]}") return "\n".join(lines[-MAX_MSG_PER_CONV:]) def call_gemini(transcript, agent_name, conversation_id): """Envia transcrição ao Gemini com backoff exponencial.""" if not GEMINI_API_KEY: print("[GEMINI] GEMINI_API_KEY não definida. Pulando análise.") return None url = f"https://generativelanguage.googleapis.com/v1beta/models/{GEMINI_MODEL}:generateContent?key={GEMINI_API_KEY}" prompt = PROMPT_7P.format(conversacao_text=transcript[:8000]) for attempt in range(8): try: backoff = min(INITIAL_BACKOFF * (2 ** attempt), MAX_BACKOFF) payload = { "contents": [{"parts": [{"text": prompt}]}], "generationConfig": { "temperature": 0.2, "maxOutputTokens": 2048, "topP": 0.8, "topK": 40 } } r = requests.post(url, json=payload, timeout=REQUEST_TIMEOUT) if r.status_code == 429: print(f"[GEMINI 429] Aguardando {backoff}s (attempt {attempt+1})...") time.sleep(backoff) continue if r.status_code == 503: print(f"[GEMINI 503] Serviço indisponível. Aguardando {backoff}s...") time.sleep(backoff) continue if r.status_code != 200: print(f"[GEMINI ERROR {r.status_code}] {r.text[:200]}") time.sleep(backoff) continue result = r.json() text = result["candidates"][0]["content"]["parts"][0]["text"] # Limpa markdown code blocks se houver text = re.sub(r"```json\s*", "", text.strip()) text = re.sub(r"```\s*$", "", text.strip()) return json.loads(text) except json.JSONDecodeError as e: print(f"[GEMINI JSON] Parse error: {e}. Tentativa {attempt+1}") time.sleep(backoff) except Exception as e: print(f"[GEMINI ERROR] {e}. Backoff {backoff}s") time.sleep(backoff) return None # ─── HTML TEMPLATE ──────────────────────────────────────────────────────────── def score_color(score): if score >= 9: return "#22c55e" if score >= 7: return "#c9a96e" if score >= 5: return "#f59e0b" return "#ef4444" def html_report(agent_name, agent_id, conversations_data, generated_at): total_score = sum(d["score_geral"] for d in conversations_data if d) / max(len(conversations_data), 1) nota_geral = round(total_score, 1) if nota_geral >= 9: badge_color = "linear-gradient(135deg,#22c55e,#16a34a)" elif nota_geral >= 7: badge_color = "linear-gradient(135deg,#c9a96e,#8B6914)" elif nota_geral >= 5: badge_color = "linear-gradient(135deg,#f59e0b,#d97706)" else: badge_color = "linear-gradient(135deg,#ef4444,#dc2626)" # Agregar pontuações por P p_keys = ["P1_Apresentacao","P2_BreveHistoria","P3_EntenderDor","P4_Combinado","P5_Solucao","P6_Fechamento","P7_VendaExtra"] p_labels = { "P1_Apresentacao": "1. Apresentação", "P2_BreveHistoria": "2. Conexão / Rapport", "P3_EntenderDor": "3. Entender a Dor", "P4_Combinado": "4. Combinado", "P5_Solucao": "5. Solução", "P6_Fechamento": "6. Fechamento", "P7_VendaExtra": "7. Venda Extra", } p_html = "" for pk in p_keys: vals = [d["scores"][pk]["nota"] for d in conversations_data if d and pk in d.get("scores",{})] avg = round(sum(vals)/max(len(vals),1), 1) if vals else 0 ev = next((d["scores"][pk]["evidencia"] for d in conversations_data if d and d["scores"].get(pk,{}).get("evidencia")), "") com = next((d["scores"][pk]["comentario"] for d in conversations_data if d and d["scores"].get(pk,{}).get("comentario")), "") color = score_color(avg) p_html += f"""
{p_labels[pk]}
{avg}
💬
"{ev}" {com}
""" # Pontos fortes pontos_fortes = [] for d in conversations_data: if d: pontos_fortes.extend(d.get("pontos_fortes", [])) pf_unique = list(dict.fromkeys(pontos_fortes))[:6] # Pontos melhoria pontos_melhoria = [] for d in conversations_data: if d: pontos_melhoria.extend(d.get("pontos_melhoria", [])) pm_unique = list(dict.fromkeys(pontos_melhoria))[:6] # Próximos passos proximos = [] for d in conversations_data: if d: proximos.extend(d.get("proximos_passos", [])) prox_unique = list(dict.fromkeys(proximos))[:6] # Conversas convs_html = "" for i, d in enumerate(conversations_data): if not d: continue cls = d.get("classificacao", "").lower() color = score_color(d.get("score_geral", 0)) resumo = d.get("resumo_venda", "") eng = d.get("nivel_engajamento", "") objs = "
".join(f"• {o}" for o in d.get("objetivos_nao_tratados",[])[:3]) convs_html += f"""
Conversa {i+1} {d.get("score_geral","?")} {eng}

{resumo}

{f'
Evidência: {d["scores"].get("P6_Fechamento",{}).get("evidencia","N/A")}
' if d.get("scores",{}).get("P6_Fechamento",{}).get("evidencia") else ""} {f'
Objeções não tratadas: {objs}
' if objs else ""}
""" html = f""" Relatório 7P — {agent_name}

📊 Relatório 7P — {agent_name}

{nota_geral}
{len(conversations_data)} conversas analisadas • Gerado em {generated_at}

🎯 Scores por Etapa — Método 7P

{p_html}

⭐ Pontos Fortes

{"".join(f'
{pf}
' for pf in pf_unique) or "

Nenhum ponto forte identificado

"}

📝 Pontos de Melhoria

{"".join(f'
{pm}
' for pm in pm_unique) or "

Nenhum ponto crítico identificado

"}

🔜 Próximos Passos

{"".join(f'
{px}
' for px in prox_unique) or "

Nenhum próximo passo definido

"}

💬 Análise por Conversa

{convs_html or "

Nenhuma conversa analisada

"}
Relatório gerado automaticamente por Hermes Insights 7P • Rocha Sales Seguros
""" return html # ─── MAIN ───────────────────────────────────────────────────────────────────── def get_agents(): """Busca lista de agentes via conversas — mais confiável que /users em self-hosted v3.""" agent_map = {} # id -> name for page in range(1, 6): data = chatwoot_get( f"accounts/{ACCOUNT_ID}/conversations", params={"per_page": 200, "page": page} ) if not data: break convs = data.get("data", {}).get("payload", []) if not convs: break for conv in convs: assignee = conv.get("meta", {}).get("assignee") if assignee: agent_map[assignee["id"]] = assignee.get("name", "Desconhecido") if len(convs) < 200: break time.sleep(0.3) if not agent_map: print("[AGENTS] Fallback: usando lista padrão.") return [{"id": 1, "name": "Anderson Ferreira Menezes"}] agents = [{"id": aid, "name": aname} for aid, aname in agent_map.items()] print(f"[AGENTS] Encontrados {len(agents)} agentes: {[a['name'] for a in agents]}") return agents def process_agent(agent_id, agent_name, target_ids=None): """Processa um único agente: busca conversas, analisa via Gemini, gera HTML.""" print(f"\n{'='*60}") print(f"PROCESSANDO: {agent_name} (ID: {agent_id})") print(f"{'='*60}") # Busca conversas do agente conversations = [] page = 1 while len(conversations) < MAX_CONV_PER_AGENT: data = chatwoot_get( f"accounts/{ACCOUNT_ID}/conversations", params={"per_page": 100, "page": page, "assignee_id": agent_id, "sort_by": "desc"} ) if not data: break convs = data.get("data", {}).get("payload", []) if not convs: break conversations.extend(convs) if len(convs) < 100: break page += 1 time.sleep(0.5) print(f" → {len(conversations)} conversas encontradas") # Filtra por target_ids se especificado if target_ids: conversations = [c for c in conversations if c.get("id") in target_ids] conversations = conversations[:MAX_CONV_PER_AGENT] print(f" → Processando {len(conversations)} conversas...") results = [] for i, conv in enumerate(conversations): conv_id = conv["id"] print(f" [{i+1}/{len(conversations)}] Conversa {conv_id}...", end=" ", flush=True) # Busca mensagens messages = fetch_messages(conv_id) if not messages: print("SEM MSGS") continue transcript = build_transcript(messages, agent_name) if len(transcript.strip()) < 50: print("TRANSCRIPT CURTO") continue # Análise Gemini result = call_gemini(transcript, agent_name, conv_id) if result: results.append(result) print(f"✅ {result.get('score_geral','?')}/10 [{result.get('classificacao','?')}]") else: print("❌ GEMINI FALHOU") time.sleep(2.5) # espaçamento entre chamadas Gemini if not results: print(f"\n⚠️ Nenhuma conversa analisada para {agent_name}. Pulando.") return # Gera HTML generated_at = datetime.now().strftime("%d/%m/%Y %H:%M") safe_name = agent_name.lower().replace(" ", "_").replace(".", "") filename = f"relatorio_{safe_name}.html" filepath = os.path.join(OUTPUT_DIR, filename) html = html_report(agent_name, agent_id, results, generated_at) with open(filepath, "w", encoding="utf-8") as f: f.write(html) avg = round(sum(r["score_geral"] for r in results) / len(results), 1) print(f"\n✅ {agent_name}: {len(results)}/{len(conversations)} conversas | Nota média: {avg}/10") print(f" Salvo em: {filepath}") def main(): if not acquire_lock(): print("Já existe uma instância rodando. Abortando.") sys.exit(0) try: if not GEMINI_API_KEY: print("⚠️ GEMINI_API_KEY não definida. Defina: export GEMINI_API_KEY='...'") print(" Prosseguindo apenas com coleta de dados (sem análise Gemini)...") target_ids = None if sys.argv[1:]: ids_str = sys.argv[1] # Support both comma and comma-space id_list = [int(x.strip()) for x in ids_str.split(",") if x.strip().isdigit() or x.strip().lstrip("-").isdigit()] if id_list: # First arg is agent ID(s) for aid in id_list: agents = get_agents() agent = next((a for a in agents if a["id"] == aid), None) if agent: process_agent(aid, agent["name"]) else: print(f"Agente ID {aid} não encontrado.") sys.exit(0) # Todos os agentes agents = get_agents() print(f"Agentes encontrados: {len(agents)}") for a in agents: print(f" - {a['id']}: {a['name']}") for agent in agents: process_agent(agent["id"], agent["name"]) time.sleep(3) # pausa entre agentes print(f"\n{'='*60}") print("✅ RELATÓRIOS GERADOS COM SUCESSO") print(f"{'='*60}") finally: release_lock() if __name__ == "__main__": import requests main()