"""Gera a planilha .xlsx do funil a partir dos dados recebidos pelo Webhook da Rubeus. Os dados sao enviados pela Rubeus (POST/JSON) e armazenados pelo webhook_server.py em RUBEUS_WEBHOOK_DIR. Este script le esses dados e escreve a planilha com o mesmo formato do modelo em tabela/Tabela-para-INSPIN-20-01-2026.xlsx (24 colunas). Uso: python gerar_planilha.py # gera a partir dos dados recebidos python gerar_planilha.py --limite 100 # limita a quantidade de registros python gerar_planilha.py --saida x.xlsx # define o arquivo de saida Parametros sao lidos do arquivo .env (veja .env.example). """ from __future__ import annotations import argparse import sys from datetime import date from pathlib import Path import openpyxl from config import BASE_DIR, Config from mapeamento import carregar_mapeamento, montar_linha from webhook_normalizador import montar_registros from webhook_store import WebhookStore TEMPLATE_PADRAO = BASE_DIR / "tabela" / "Tabela-para-INSPIN-20-01-2026.xlsx" def log(msg: str) -> None: print(msg, flush=True) # --------------------------------------------------------------------------- # # Ordenacao # --------------------------------------------------------------------------- # def _ordenar_por_contato(registros: list[dict]) -> list[dict]: """Agrupa as linhas por contato (todas as inscricoes do mesmo contato juntas).""" def chave(reg: dict): op = reg.get("oportunidade") or {} ct = reg.get("contato") or {} contato = (ct.get("nome") or op.get("pessoaNome") or "").strip().lower() contato_id = str(op.get("pessoa") or ct.get("id") or "") ps = str(op.get("processoSeletivoNome") or "") data = str(op.get("momento") or "") # contatos sem nome vao para o fim return (contato == "", contato, contato_id, ps, data) return sorted(registros, key=chave) # --------------------------------------------------------------------------- # # Planilha # --------------------------------------------------------------------------- # def ler_cabecalho_template(template: Path) -> list[str]: wb = openpyxl.load_workbook(template, read_only=True) ws = wb.active header = [c.value for c in next(ws.iter_rows(min_row=1, max_row=1))] while header and header[-1] is None: header.pop() wb.close() return header def escrever_planilha(saida: Path, header: list[str], linhas: list[list]) -> Path: saida.parent.mkdir(parents=True, exist_ok=True) wb = openpyxl.Workbook() ws = wb.active ws.title = "Planilha1" ws.append(header) for linha in linhas: ws.append(linha) try: wb.save(saida) return saida except PermissionError: # Arquivo bloqueado (provavelmente aberto no Excel). Tenta atualizar a # propria planilha aberta via automacao COM do Excel (Windows). if _atualizar_via_excel(saida, header, linhas): log(f"OK! Planilha aberta no Excel foi atualizada: {saida.name}") return saida # Sem Excel/pywin32 disponivel: salva com nome alternativo para nao perder o dado. from datetime import datetime alt = saida.with_name(f"{saida.stem}_{datetime.now():%H%M%S}{saida.suffix}") wb.save(alt) log( f"AVISO: '{saida.name}' estava em uso e nao foi possivel atualiza-lo via Excel. " f"Salvei como '{alt.name}'." ) return alt def _atualizar_via_excel(saida: Path, header: list[str], linhas: list[list]) -> bool: """Atualiza uma planilha aberta no Excel usando automacao COM (pywin32). Necessario apenas quando o arquivo esta bloqueado por estar aberto no Excel. Retorna True se conseguiu atualizar e salvar; False caso contrario (ex.: sem pywin32, sem Excel, ou rodando fora do Windows). """ try: import os import pythoncom import win32com.client as win32 except ImportError: return False alvo_caminho = os.path.normcase(os.path.abspath(str(saida))) coinicializado = False excel = None wb = None abriu_aqui = False try: try: pythoncom.CoInitialize() coinicializado = True except Exception: pass # Conecta a uma instancia do Excel ja em execucao; se nao houver, abre uma. try: excel = win32.GetActiveObject("Excel.Application") except Exception: try: excel = win32.Dispatch("Excel.Application") except Exception: return False # Procura a planilha ja aberta com esse caminho. for aberto in excel.Workbooks: try: if os.path.normcase(os.path.abspath(aberto.FullName)) == alvo_caminho: wb = aberto break except Exception: continue if wb is None: wb = excel.Workbooks.Open(str(saida)) abriu_aqui = True ws = wb.Worksheets(1) ws.Cells.Clear() dados = [header] + [list(linha) for linha in linhas] n_linhas = len(dados) n_colunas = len(header) # Matriz retangular, convertendo None -> "" (Excel nao aceita None). matriz = [ [("" if (i >= len(row) or row[i] is None) else row[i]) for i in range(n_colunas)] for row in dados ] rng = ws.Range(ws.Cells(1, 1), ws.Cells(n_linhas, n_colunas)) rng.NumberFormat = "@" # texto: preserva telefones/zeros a esquerda rng.Value = matriz wb.Save() if abriu_aqui: wb.Close(SaveChanges=True) return True except Exception: return False finally: if coinicializado: try: pythoncom.CoUninitialize() except Exception: pass def _montar_header(colunas: list[dict], template: Path) -> list[str]: """Monta o cabecalho final reaproveitando os nomes do template (24 colunas). Colunas extras no inicio do mapeamento (ex.: "Funil/Processo") sao prefixadas ao cabecalho do template, preservando os nomes originais (com acentos). """ header_template = ler_cabecalho_template(template) extras = max(0, len(colunas) - len(header_template)) header = [c["coluna"] for c in colunas[:extras]] + header_template if len(colunas) != len(header): log( f"AVISO: o mapeamento tem {len(colunas)} colunas e o cabecalho final " f"tem {len(header)}; verifique config/mapeamento_campos.json." ) return header def gerar_de_webhook(cfg: Config, store: WebhookStore, template: Path | None = None, saida_str: str | None = None, limite: int = 0) -> str: """Gera a planilha a partir dos dados recebidos pelo webhook (registros + contatos). Usada tanto pelo CLI (gerar_planilha.py) quanto pelo servidor (webhook_server). Retorna o caminho do arquivo .xlsx gerado. """ template = template or TEMPLATE_PADRAO colunas = carregar_mapeamento(cfg.caminho(cfg.mapeamento)) header = _montar_header(colunas, template) registros = montar_registros(store.carregar("registro"), store.carregar("contato")) if limite: registros = registros[:limite] registros = _ordenar_por_contato(registros) linhas = [montar_linha(reg, colunas) for reg in registros] saida_str = (saida_str or cfg.saida).replace("{data}", date.today().isoformat()) final = escrever_planilha(cfg.caminho(saida_str), header, linhas) log(f"OK! Planilha gerada com {len(linhas)} linha(s): {final}") return str(final) # --------------------------------------------------------------------------- # # Main # --------------------------------------------------------------------------- # def main(argv: list[str] | None = None) -> int: parser = argparse.ArgumentParser( description="Gera a planilha do funil a partir dos dados recebidos pelo Webhook da Rubeus." ) parser.add_argument("--limite", type=int, default=0, help="Limita a quantidade de registros (0 = todos).") parser.add_argument("--template", help="Caminho do .xlsx usado como modelo de cabecalho.") parser.add_argument("--saida", help="Caminho do .xlsx de saida.") args = parser.parse_args(argv) cfg = Config.carregar() template = Path(args.template) if args.template else TEMPLATE_PADRAO if not template.exists(): log(f"ERRO: template nao encontrado: {template}") return 1 store = WebhookStore(cfg.caminho(cfg.webhook_dir)) log(f"Lendo dados do webhook em: {cfg.caminho(cfg.webhook_dir)}") gerar_de_webhook(cfg, store, template=template, saida_str=args.saida, limite=args.limite) return 0 if __name__ == "__main__": sys.exit(main())