import io import logging import unicodedata import streamlit as st from modules.config import DRIVE_FOLDER_ID from modules.drive_utils import conectar_drive, conectar_drive_usuario, _ler_conteudo_drive_arquivo SQUAD_CONTEXT_FILES = [ "Risco e Seguranca - Planejamento 2026", "Estado da Cora - Jan 2026", ] SQUAD_CONTEXT_SHEETS = [ { "id": "1mZi798rR3WWBnUyiZHhH6ISUkPYTXHnfF7roD09xbuE", "nome": "Master Product Roadmap (Google Sheets)", "abas": ["Orientacoes Gerais", "Risco e Seguranca"], }, { "id": "1AQtmCSmSsX67rkbrQcc26Wc69Y1OlNBAQnjEzvMkxm4", "nome": "Metricas da Squad (Google Sheets)", "abas": ["OKRs Tech", "Health Metrics"], "usar_oauth": True, }, { "id": "1q4MLQwncd7iaxK8oXdNrNB-r10hu2N4QeBZ9fjgOG8k", "nome": "Plano de Acao R10-25 Cronograma Visao Risk (Google Sheets)", "abas": None, "usar_oauth": True, }, ] def _normalizar_texto(s: str) -> str: return unicodedata.normalize("NFKD", s).encode("ascii", "ignore").decode("ascii").lower() def _ler_google_sheet(sheet_id: str, abas: list = None, usar_oauth: bool = False) -> str: svc = None if usar_oauth: svc = conectar_drive_usuario() if svc: logging.info(f"[Contexto] Lendo sheet '{sheet_id}' via OAuth do usuario") if not svc: svc = conectar_drive() if not svc: return "" try: import pandas as pd conteudo = svc.files().export( fileId=sheet_id, mimeType="application/vnd.openxmlformats-officedocument.spreadsheetml.sheet" ).execute() xls = pd.ExcelFile(io.BytesIO(conteudo)) resultado = "" abas_para_ler = abas if abas else xls.sheet_names[:2] for nome_aba in abas_para_ler: aba_match = None nome_norm = _normalizar_texto(nome_aba) for sn in xls.sheet_names: if nome_norm in _normalizar_texto(sn): aba_match = sn break if not aba_match: continue df = pd.read_excel(xls, sheet_name=aba_match, header=None) df = df.dropna(how="all").dropna(axis=1, how="all") df = df.fillna("") # Detectar header: primeira linha com >= 3 celulas nao-vazias headers = [] header_row_idx = None for ridx in range(min(5, len(df))): row_vals = [str(v).strip() for v in df.iloc[ridx]] non_empty = [v for v in row_vals if v and v != "nan"] if len(non_empty) >= 3: headers = row_vals header_row_idx = ridx break # Gerar letras de coluna (A, B, C, ...) como fallback col_letters = [chr(65 + i) if i < 26 else f"A{chr(65 + i - 26)}" for i in range(len(df.columns))] if header_row_idx is not None and len(df) > header_row_idx + 1: texto_aba = "" # Incluir header como referencia para o Claude header_ref = " | ".join( f"Col{col_letters[i]}={h}" if h and h != "nan" else f"Col{col_letters[i]}" for i, h in enumerate(headers) ) texto_aba += f"COLUNAS: {header_ref}\n" for idx in range(header_row_idx + 1, len(df)): row = df.iloc[idx] valores = [str(v).strip() for v in row] pares = [] for col_idx, val in enumerate(valores): if val and val != "nan": col_nome = headers[col_idx] if col_idx < len(headers) and headers[col_idx] and headers[col_idx] != "nan" else "" col_ref = f"Col{col_letters[col_idx]}" label = col_nome or col_ref pares.append(f"[{label}] {val}") if pares: texto_aba += " | ".join(pares) + "\n" else: texto_aba = "" for row_idx, row in df.iterrows(): valores = [str(v).strip() for v in row] pares = [] for col_idx, val in enumerate(valores): if val and val != "nan": pares.append(f"[Col{col_letters[col_idx]}] {val}") if pares: texto_aba += " | ".join(pares) + "\n" if texto_aba.strip(): resultado += f"\n[Aba: {aba_match}]\n{texto_aba}\n" return resultado except Exception as e: logging.warning(f"[Contexto] Erro ao ler Google Sheet '{sheet_id}': {e}") return "" KEYWORDS_SHEET_0 = ["roadmap", "iniciativa", "prioridade", "planejado", "master product", "trimestre", "mes que vem", "este mes", "esse mes", "proximo mes", "q1", "q2", "q3", "q4", "timeline", "cronograma"] KEYWORDS_SHEET_1 = ["metrica", "metricas", "okr", "okrs", "health", "health metric", "indicador", "indicadores", "kpi", "kpis", "resultado chave", "resultado-chave", "key result", "tech metric", "disponibilidade", "latencia", "uptime", "sla", "slo", "erro rate", "error rate"] KEYWORDS_SHEET_2 = ["plano de acao", "r10", "r10-25", "cronograma", "entrega", "subacao", "auditoria", "csnu", "rufra", "dict", "judblock", "bacen", "regulatorio", "plano acao", "visao risk", "acao regulatoria"] KEYWORDS_FILE_0 = ["squad", "planejamento", "aposta", "risco", "seguranca", "fricao", "modelo", "biometria", "fraude", "trava"] KEYWORDS_FILE_1 = ["cora", "empresa", "estado", "companhia", "banco", "negocio", "mercado"] def detectar_skill_pm_os(user_input: str) -> str | None: """Proxy para deteccao de skill PM OS (importacao lazy para evitar circular).""" from modules.pm_os import detectar_skill_pm_os as _detectar return _detectar(user_input) def _detectar_fontes(user_input: str) -> dict: inp = _normalizar_texto(user_input) fontes = {"files": set(), "sheets": set()} for kw in KEYWORDS_SHEET_0: if kw in inp: fontes["sheets"].add(0) for kw in KEYWORDS_SHEET_1: if kw in inp: fontes["sheets"].add(1) for kw in KEYWORDS_SHEET_2: if kw in inp: fontes["sheets"].add(2) for kw in KEYWORDS_FILE_0: if kw in inp: fontes["files"].add(0) for kw in KEYWORDS_FILE_1: if kw in inp: fontes["files"].add(1) if not fontes["files"] and not fontes["sheets"]: fontes["files"].add(0) return fontes @st.cache_data(ttl=300, show_spinner=False) def _carregar_arquivo_drive(_cache_key: str, nome_arquivo: str) -> str: svc = conectar_drive() if not svc: return "" try: safe_name = nome_arquivo.replace("\\", "\\\\").replace("'", "\\'") results = svc.files().list( q=f"name contains '{safe_name}' and trashed=false", fields="files(id, name, mimeType)", supportsAllDrives=True, includeItemsFromAllDrives=True ).execute() arquivos = results.get("files", []) if arquivos: arq = arquivos[0] return _ler_conteudo_drive_arquivo(svc, arq["id"], arq["mimeType"], arq["name"]) except Exception as e: logging.warning(f"[Contexto] Erro ao carregar arquivo '{nome_arquivo}': {e}") return "" @st.cache_data(ttl=300, show_spinner=False) def _carregar_sheet_contexto(_cache_key: str, sheet_id: str, abas: tuple, usar_oauth: bool = False) -> str: return _ler_google_sheet(sheet_id, list(abas), usar_oauth=usar_oauth) def carregar_contexto_sob_demanda(user_input: str) -> tuple: fontes = _detectar_fontes(user_input) contexto = "" log = [] for idx in fontes["files"]: if idx < len(SQUAD_CONTEXT_FILES): nome = SQUAD_CONTEXT_FILES[idx] c = _carregar_arquivo_drive(DRIVE_FOLDER_ID or "none", nome) if c.strip(): contexto += f"\n--- {nome} ---\n{c}\n" log.append(f"OK: {nome} ({len(c)} chars)") else: log.append(f"VAZIO: {nome}") for idx in fontes["sheets"]: if idx < len(SQUAD_CONTEXT_SHEETS): sheet = SQUAD_CONTEXT_SHEETS[idx] abas_t = tuple(sheet.get("abas", [])) oauth = sheet.get("usar_oauth", False) cache_key = DRIVE_FOLDER_ID or "none" if oauth: cache_key += ":" + st.session_state.get("usuario_email", "anon") c = _carregar_sheet_contexto(cache_key, sheet["id"], abas_t, oauth) if c.strip(): contexto += f"\n--- {sheet['nome']} ---\n{c}\n" log.append(f"OK: {sheet['nome']} ({len(c)} chars)") else: log.append(f"VAZIO: {sheet['nome']}") return contexto, log