"""Aba Roadmap — visualizacao grid mensal com 5 sub-tabs. Estrutura: Time -> OKR -> Iniciativa -> Etapas, em grid mensal (6 meses pra tras + 6 pra frente, sliding window centrado em hoje). Fontes de dados: - Master Roadmap planilha (aba 'Risco e Seguranca') — grid mensal de iniciativas - OKRs sheet — metricas dos KRs por mes - Doc regulatorio (Plano R10-25) — iniciativas regulatorias com prazo - Jira SAF (boards 3542 epicos, 4848 iniciativas) — datas e sub-tasks Classificacao em 5 categorias: KR1.1 (FPR), KR1.2 (Deteccao), KR1.3 (Chargeback), Debitos Tecnicos, Regulatorios. """ import logging import re import unicodedata from dataclasses import dataclass, field from datetime import date from difflib import SequenceMatcher from typing import Optional import streamlit as st # Threshold de similaridade pra considerar match planilha<->jira FUZZY_MATCH_THRESHOLD = 0.70 def _norm_titulo(s: str) -> str: """Normaliza titulo pra comparacao fuzzy: lowercase, sem acento, sem prefixo numerico ('1.', '1.2 -', '#5'), sem pontuacao, espacos colapsados.""" if not s: return "" # Remove acento s = unicodedata.normalize("NFKD", s).encode("ascii", "ignore").decode() s = s.lower() # Remove prefixos comuns: "1.", "1.2", "AP.R10-25.3 -", "#5", "- " s = re.sub(r"^[\s#\-–—]*(ap[.\s]?r\d+[-.]\d+[.\d()\s]*[-–—]?\s*)?", "", s) s = re.sub(r"^\d+(\.\d+)*\s*[.\-–—:]?\s*", "", s) # Remove pontuacao s = re.sub(r"[^\w\s]", " ", s) # Colapsa espacos return re.sub(r"\s+", " ", s).strip() def _similaridade(a: str, b: str) -> float: a_norm = _norm_titulo(a) b_norm = _norm_titulo(b) if not a_norm or not b_norm: return 0.0 return SequenceMatcher(None, a_norm, b_norm).ratio() # --------------------------------------------------------------------------- # Constantes # --------------------------------------------------------------------------- MASTER_ROADMAP_SHEET_ID = "1mZi798rR3WWBnUyiZHhH6ISUkPYTXHnfF7roD09xbuE" MASTER_ROADMAP_TAB = "Risco e Segurança" # Doc com plano de acao regulatorio (R10-25) DOC_REGULATORIO_ID = "1HNxReXrXIxcFW5aW-g0bLtPFdAPEylLJFRrksSfiEXE" # Boards Jira a consultar (excluindo ATENDIMENTO) JIRA_BOARDS = [3542, 4848] JIRA_PROJECT = "SAF" # Team folder do squad — root pra busca de docs/pastas linkadas TEAM_FOLDER_ID = "182IVNPuvuwEXVOTB1tdkt47TL6BpxMc7" # Threshold pra match titulo<->arquivo Drive (scoring hibrido char+token) DRIVE_MATCH_THRESHOLD = 0.45 # Pastas internas do agente que NAO sao docs de iniciativa EXCLUDED_DRIVE_FOLDER_NAMES_LC = { "pm-assistant", "historico", "memoria", "memoria_usuarios", "templates", } # Stopwords PT/EN pra ignorar no token matching STOPWORDS_MATCH = { "de", "da", "do", "dos", "das", "para", "com", "sem", "em", "no", "na", "the", "for", "and", "of", "to", "in", "on", "at", "fase", "etapa", "novo", "nova", "old", "antigo", "v1", "v2", "v3", "draft", "rascunho", "final", "copy", "copia", "backup", "projeto", "iniciativa", "task", "epic", "epico", "story", "historia", } # Keywords pra classificar iniciativas — alinhado com squad_health.KR_KEYWORDS KR_KEYWORDS = { "KR1.1": [ "fpr", "falso positivo", "biometria", "unico", "facetec", "fric", "fricao", "fricção", "desbloqueio", "poc biometria", "idoneo", "idoneos", "idôneo", "impacto em clientes bons", ], "KR1.2": [ "detec", "fraud lab", "ds boletos", "sentinel", "trava", "antecip", "modelo ml", "modelo de ml", "modelo pix", "behavior", "blocklist", "device_id", "device id", "filas inteligentes", "retool bpo", "score", ], "KR1.3": [ "cbk", "chargeback", "lp ", "link de pagamento", "cybersource", "visa", "antifraude pagamento", "pagamentos", "3ds", ], } REGULATORIO_KEYWORDS = [ "regulator", "bacen", "auditor", "lgpd", "compliance", "judicial", "sisbajud", "csnu", "sigilo", "rufra", "ofício", "oficio", "judblocker", "jud blocker", "pld", "ap.r10", "ap r10", "r10-25", ] DEBITO_KEYWORDS = [ "migra", "migração", "migracao", "refator", "retool", "scaling", "infra", "manutencao", "manutenção", "manuten", "operacional", "deprec", "tombamento", "tombar", "ghost", ] # Mapeamento de cores por status STATUS_COLORS = { "done": "#86efac", # verde claro "em-andamento": "#7dd3fc", # azul claro "atrasado": "#fca5a5", # vermelho claro "planejado": "#e5e7eb", # cinza } PRIORIDADE_COLORS = { "alta": "#dc2626", "media": "#f59e0b", "baixa": "#10b981", } # Mapping mes en→pt — usado pra parsear cabecalhos da planilha MESES_PT = { "janeiro": 1, "fevereiro": 2, "março": 3, "marco": 3, "abril": 4, "maio": 5, "junho": 6, "julho": 7, "agosto": 8, "setembro": 9, "outubro": 10, "novembro": 11, "dezembro": 12, } MESES_PT_ABREV = ["Jan", "Fev", "Mar", "Abr", "Mai", "Jun", "Jul", "Ago", "Set", "Out", "Nov", "Dez"] # --------------------------------------------------------------------------- # Dataclass de Iniciativa # --------------------------------------------------------------------------- @dataclass class Iniciativa: """Representa uma iniciativa no roadmap.""" titulo: str fonte: str # 'planilha', 'jira', 'doc-regulatorio', 'pm-os' categoria: str = "?" # 'KR1.1', 'KR1.2', 'KR1.3', 'DEBITO', 'REGULATORIO', '?' data_inicio: Optional[date] = None # se conhecida (Jira/doc) data_fim: Optional[date] = None mes_planilha: Optional[tuple[int, int]] = None # (ano, mes) — quando vem da planilha status: str = "planejado" # 'done', 'em-andamento', 'atrasado', 'planejado' prioridade: Optional[str] = None # 'alta', 'media', 'baixa' link: Optional[str] = None descricao: str = "" raw: dict = field(default_factory=dict) # Fase 2: flags de mesclagem tambem_na_planilha: bool = False # True quando epico Jira tem match na planilha n_tasks_filhas: int = 0 # quantas tasks/historias filhas o epico tem (Jira) data_fim_derivada: bool = False # True se data_fim veio da maior duedate das tasks # Link Drive (pasta ou arquivo) descoberto por fuzzy match no titulo link_drive: Optional[dict] = None # {id, name, mimeType, url, _match_type, _score} # Fase 3b: matching com doc regulatorio (Plano R10-25) tambem_no_doc_regulatorio: bool = False # True quando item tem match em AP.R10-25.X doc_regulatorio_link: Optional[str] = None # link pro doc R10-25 doc_regulatorio_ap_code: Optional[str] = None # ex: 'AP.R10-25.3' # Sub-items (usado pra consolidar doc-regulatorio em 1 linha: 1 mini-barra # por AP.R10-25.X no timeline, posicionada pelo prazo do sub-item) sub_items: list = field(default_factory=list) # list[dict] # --------------------------------------------------------------------------- # Janela temporal — 6 meses pra tras + 6 pra frente # --------------------------------------------------------------------------- def _janela_meses() -> list[tuple[int, int]]: """Retorna lista de (ano, mes) cobrindo os ultimos 6 meses ate dezembro do ano corrente. Janela encolhe ao longo do ano: em jan tem 6+12=18 meses, em dez tem 6+1=7 meses. Iniciativas com data_fim antes do inicio sao filtradas em _calcular_span.""" hoje = date.today() janela: list[tuple[int, int]] = [] # Inicio: 6 meses atras (inclui o mes atual no meio) ano = hoje.year mes = hoje.month - 6 while mes <= 0: mes += 12 ano -= 1 # Avanca mes a mes ate dezembro do ano corrente while (ano, mes) <= (hoje.year, 12): janela.append((ano, mes)) mes += 1 if mes > 12: mes = 1 ano += 1 return janela def _label_mes(ano: int, mes: int) -> str: return f"{MESES_PT_ABREV[mes - 1]}/{str(ano)[2:]}" # --------------------------------------------------------------------------- # Classificacao em KR / Debito / Regulatorio # --------------------------------------------------------------------------- def _normalizar(s: str) -> str: return (s or "").lower().strip() def _classificar_categoria(titulo: str, descricao: str = "", fonte: str = "", raw: dict | None = None) -> str: """Decide a categoria da iniciativa. Prioridade: 1) Se fonte == 'doc-regulatorio' -> REGULATORIO 2) Se bate REGULATORIO_KEYWORDS -> REGULATORIO 3) Se bate DEBITO_KEYWORDS e nao bate KR -> DEBITO 4) Se bate KR_KEYWORDS -> KR correspondente 5) Default -> '?' (vai pra painel 'Sem categoria') """ if fonte == "doc-regulatorio": return "REGULATORIO" texto = _normalizar(f"{titulo} {descricao}") raw = raw or {} # 2) Regulatorio por keyword if any(kw in texto for kw in REGULATORIO_KEYWORDS): return "REGULATORIO" # 4) KR (verificamos primeiro pra dar preferencia se houver match forte) matches_kr = {} for kr, kws in KR_KEYWORDS.items(): n = sum(1 for kw in kws if kw in texto) if n > 0: matches_kr[kr] = n # 3) Debito tecnico (somente se nao houver KR claro) if not matches_kr and any(kw in texto for kw in DEBITO_KEYWORDS): return "DEBITO" if matches_kr: return max(matches_kr, key=lambda k: matches_kr[k]) # 5) Default return "?" # --------------------------------------------------------------------------- # Fetcher: Planilha Master Roadmap # --------------------------------------------------------------------------- def _build_sheets_service(): """Helper pra conectar no Sheets API com a Service Account.""" import base64 import json as _json import os from google.oauth2 import service_account as _sa from googleapiclient.discovery import build SCOPES = ["https://www.googleapis.com/auth/spreadsheets.readonly"] if os.path.exists("drive_key.json"): creds = _sa.Credentials.from_service_account_file("drive_key.json", scopes=SCOPES) elif os.environ.get("GOOGLE_SERVICE_ACCOUNT_JSON"): sa_raw = os.environ["GOOGLE_SERVICE_ACCOUNT_JSON"] if not sa_raw.strip().startswith("{"): sa_raw = base64.b64decode(sa_raw).decode("utf-8") creds = _sa.Credentials.from_service_account_info(_json.loads(sa_raw), scopes=SCOPES) else: return None return build("sheets", "v4", credentials=creds) def _fetch_planilha_roadmap() -> list[Iniciativa]: """Le a aba 'Risco e Seguranca' do Master Roadmap e converte cada celula em uma Iniciativa. A planilha eh um grid: linhas = categorias, colunas = meses. Tambem le os ranges MESCLADOS (`merges`) — quando uma iniciativa ocupa visualmente 2+ meses adjacentes, a API retorna o valor so na 1a coluna. O merge info permite detectar isso e criar 1 Iniciativa com data_inicio e data_fim cobrindo todo o range. """ svc = _build_sheets_service() if not svc: logging.warning("[Roadmap] Sheets API indisponivel") return [] try: result = svc.spreadsheets().values().get( spreadsheetId=MASTER_ROADMAP_SHEET_ID, range=f"'{MASTER_ROADMAP_TAB}'!A1:Z50", ).execute() except Exception as e: logging.warning(f"[Roadmap] Erro lendo Master Roadmap: {e}") return [] # Le os merges da aba (celulas mescladas visualmente) merges_por_origem: dict[tuple[int, int], tuple[int, int]] = {} try: meta = svc.spreadsheets().get( spreadsheetId=MASTER_ROADMAP_SHEET_ID, fields="sheets(properties(title),merges)", ).execute() for sh in meta.get("sheets", []): if sh.get("properties", {}).get("title") == MASTER_ROADMAP_TAB: for m in sh.get("merges", []) or []: r0 = m.get("startRowIndex", 0) c0 = m.get("startColumnIndex", 0) c1_excl = m.get("endColumnIndex", c0 + 1) # Mapa (linha, coluna_origem) -> (col_inicio, col_fim_inclusiva) merges_por_origem[(r0, c0)] = (c0, c1_excl - 1) if merges_por_origem: logging.info(f"[Roadmap] {len(merges_por_origem)} celulas mescladas detectadas") except Exception as e: logging.warning(f"[Roadmap] Erro lendo merges: {e}") rows = result.get("values", []) if not rows or len(rows) < 2: return [] # Linha 1 (idx=1 no array) tem os meses no header a partir da coluna 3 header_row = rows[1] if len(rows) > 1 else [] # Mapa: indice_coluna -> (ano, mes). Sem ano explicito, faz ass: # primeiro mes encontrado eh o "Agosto" — assumimos Ago/(ano_atual-1). # Mas como nao temos ano, vou inferir baseado em hoje: # se ha 17 colunas de mes, distribui ao redor do mes atual. col_to_mes = {} meses_header = [] for i, val in enumerate(header_row): m_idx = MESES_PT.get(_normalizar(val)) if m_idx: meses_header.append((i, m_idx)) # Inferir anos: assumimos que o ano atual eh ~o meio da lista hoje = date.today() if meses_header: # Encontra mes mais proximo do mes atual na lista idx_atual = None for k, (col, m) in enumerate(meses_header): if m == hoje.month: idx_atual = k break if idx_atual is None: # fallback: assume linear forward do primeiro como hoje idx_atual = 0 # Distribui anos ano_corrente = hoje.year for k, (col, m) in enumerate(meses_header): offset = k - idx_atual # ajustar ano baseado em diff em meses anos_off = 0 mes_running = hoje.month + offset while mes_running <= 0: mes_running += 12 anos_off -= 1 while mes_running > 12: mes_running -= 12 anos_off += 1 col_to_mes[col] = (ano_corrente + anos_off, m) # Agora pega cada celula com texto fora do header iniciativas: list[Iniciativa] = [] for r_idx, row in enumerate(rows): if r_idx <= 1: continue for c_idx, cell in enumerate(row): if not cell or not cell.strip(): continue if c_idx not in col_to_mes: continue # Pula celulas que sao apenas categoria (geralmente coluna 2) if c_idx <= 2: continue # Extrair titulo + descricao da celula (formato comum: "Titulo\nDescricao") partes = cell.split("\n", 1) titulo = partes[0].strip() descricao = partes[1].strip() if len(partes) > 1 else "" ano, mes = col_to_mes[c_idx] # Se esta celula eh inicio de um range MESCLADO, calcula data_fim # cobrindo todos os meses do merge. data_inicio_ini: Optional[date] = None data_fim_ini: Optional[date] = None ano_ref, mes_ref = ano, mes # pro status (usa o mes inicial) if (r_idx, c_idx) in merges_por_origem: col_start, col_end = merges_por_origem[(r_idx, c_idx)] if col_start in col_to_mes and col_end in col_to_mes: ano_ini, mes_ini = col_to_mes[col_start] ano_end, mes_end = col_to_mes[col_end] data_inicio_ini = date(ano_ini, mes_ini, 1) data_fim_ini = date(ano_end, mes_end, 28) ano_ref, mes_ref = ano_ini, mes_ini # Detectar status pelo checkmark status = "done" if "✅" in titulo or "✅" in descricao else "planejado" # Se o mes ja passou e nao tem check, considerar atrasado if status != "done": if (ano_ref, mes_ref) < (hoje.year, hoje.month): status = "atrasado" elif (ano_ref, mes_ref) == (hoje.year, hoje.month): status = "em-andamento" ini = Iniciativa( titulo=titulo.replace("✅", "").strip(), fonte="planilha", mes_planilha=(ano_ref, mes_ref), data_inicio=data_inicio_ini, data_fim=data_fim_ini, status=status, descricao=descricao, raw={"row": r_idx, "col": c_idx}, ) ini.categoria = _classificar_categoria(ini.titulo, ini.descricao, ini.fonte, ini.raw) iniciativas.append(ini) return iniciativas # --------------------------------------------------------------------------- # Fetcher: Doc Regulatorio (R10-25) # --------------------------------------------------------------------------- # Regex pra extrair items do doc _RE_ITEM = re.compile( r"^(\d+)\.\s*(AP\.R10[-.][\d.]+(?:\s*\([\d.]+\))?)\s*[-–]\s*(.+?)\s*$", re.MULTILINE, ) _RE_PRAZO = re.compile(r"Prazo:\s*(\d{1,2}/\d{1,2}(?:/\d{2,4})?(?:\s*[-–]\s*\d{1,2}/\d{1,2}(?:/\d{2,4})?)?)", re.I) _RE_PRIORIDADE = re.compile(r"(🔴|🟡|🟢)") def _parse_prazo(prazo_str: str) -> tuple[Optional[date], Optional[date]]: """Parseia 'DD/MM/YYYY' ou 'DD/MM - DD/MM/YYYY' ou 'DD/MM'. Retorna (data_inicio, data_fim). Se so 1 data, fim eh ela.""" if not prazo_str: return None, None hoje = date.today() # divide por - ou – partes = re.split(r"\s*[-–]\s*", prazo_str.strip()) def _parse_uma(s: str) -> Optional[date]: s = s.strip() m = re.match(r"(\d{1,2})/(\d{1,2})(?:/(\d{2,4}))?", s) if not m: return None d, mo, y = m.groups() try: dia = int(d) mes = int(mo) ano = int(y) if y else hoje.year if y and len(y) == 2: ano = 2000 + int(y) return date(ano, mes, dia) except (ValueError, TypeError): return None if len(partes) == 2: return _parse_uma(partes[0]), _parse_uma(partes[1]) elif len(partes) == 1: d1 = _parse_uma(partes[0]) return d1, d1 return None, None def _fetch_doc_regulatorio() -> list[Iniciativa]: """Le o Plano R10-25 e parseia as iniciativas regulatorias.""" try: from modules.drive_utils import _ler_conteudo_drive_arquivo, conectar_drive except Exception as e: logging.warning(f"[Roadmap] Erro importando drive_utils: {e}") return [] svc = conectar_drive() if not svc: return [] try: meta = svc.files().get(fileId=DOC_REGULATORIO_ID, fields="id,name,mimeType").execute() conteudo = _ler_conteudo_drive_arquivo(svc, DOC_REGULATORIO_ID, meta["mimeType"], meta["name"]) except Exception as e: logging.warning(f"[Roadmap] Erro lendo doc regulatorio: {e}") return [] if not conteudo: return [] iniciativas: list[Iniciativa] = [] # Dividir por separador "─────..." blocos = re.split(r"─{5,}", conteudo) for bloco in blocos: bloco = bloco.strip() if not bloco: continue # 1ª linha esperada: "N. AP.R10-25.X - Titulo" m = _RE_ITEM.search(bloco) if not m: continue _num, codigo, titulo = m.groups() # Prioridade m_prio = _RE_PRIORIDADE.search(bloco) prioridade = None if m_prio: emoji = m_prio.group(1) prioridade = {"🔴": "alta", "🟡": "media", "🟢": "baixa"}.get(emoji) # Prazo (pega o ultimo "Prazo:" que costuma ser o atualizado) prazos = _RE_PRAZO.findall(bloco) data_ini, data_fim = (None, None) if prazos: data_ini, data_fim = _parse_prazo(prazos[-1]) # Status: ⬜ Nao iniciado, em andamento, etc. status = "planejado" if "✅" in bloco or "concluído" in bloco.lower() or "concluido" in bloco.lower(): status = "done" elif data_fim and data_fim < date.today(): status = "atrasado" elif data_ini and data_ini <= date.today() <= (data_fim or date.today()): status = "em-andamento" # Descricao: pegar tudo apos a 1a linha, limitado linhas = bloco.split("\n") descricao = "\n".join(linhas[1:]).strip()[:600] ini = Iniciativa( titulo=f"{codigo} — {titulo.strip()}", fonte="doc-regulatorio", categoria="REGULATORIO", data_inicio=data_ini, data_fim=data_fim, status=status, prioridade=prioridade, descricao=descricao, link=f"https://docs.google.com/document/d/{DOC_REGULATORIO_ID}/edit", raw={"codigo": codigo}, ) # Se tem data, derivar mes_planilha pro grid if data_fim: ini.mes_planilha = (data_fim.year, data_fim.month) iniciativas.append(ini) return iniciativas # --------------------------------------------------------------------------- # Fetcher: Jira SAF # --------------------------------------------------------------------------- def _fetch_jira_epicos() -> list[Iniciativa]: """Le APENAS Epicos do Jira SAF (sem Iniciativas). Usa _build_type_map() do jira_utils pra detectar o nome EXATO do tipo 'Épico' (com acento, em PT-BR) no Jira da Cora — mesma tecnica que o agente ja usa pra criar issues. Sem isso, o JQL com `issuetype = Epico` nao casa o nome real `Épico` e retorna 0. """ try: from modules.jira_utils import _build_type_map, conectar_jira except Exception: return [] jira = conectar_jira() if not jira: logging.info("[Roadmap] Jira indisponivel — pulando") return [] # Descobre o nome real do tipo Epic no Jira da Cora try: type_map = _build_type_map(jira, JIRA_PROJECT) logging.info(f"[Roadmap] Type map SAF: {type_map}") except Exception as e: logging.warning(f"[Roadmap] Erro construindo type_map: {e}") type_map = {} epic_name = (type_map.get("epico") or {}).get("name", "Epic") # JQL: SOMENTE Epicos (Iniciativas excluidas a pedido do PM pra ficar # mais legivel) jql = f'project = {JIRA_PROJECT} AND issuetype = "{epic_name}" ORDER BY updated DESC' iniciativas: list[Iniciativa] = [] try: issues = jira.search_issues( jql, maxResults=200, fields="summary,description,status,labels,duedate,customfield_10015,components,issuetype", ) logging.info(f"[Roadmap] Jira retornou {len(issues)} epicos") except Exception as e: logging.warning(f"[Roadmap] Erro buscando Jira (JQL='{jql}'): {e}") return [] # Componentes a EXCLUIR (variacoes case) COMPS_EXCLUIDOS = {"atendimento"} # Status a EXCLUIR STATUS_EXCLUIDOS = {"cancelled", "cancelado", "cancelada", "canceled"} hoje = date.today() descartados_por_comp = 0 descartados_por_status = 0 aceitos_keys: list[str] = [] # pra busca de tasks filhas for iss in issues: try: f = iss.fields comp_names = [_normalizar(c.name) for c in (getattr(f, "components", []) or [])] if any(c in COMPS_EXCLUIDOS for c in comp_names): descartados_por_comp += 1 continue status_jira = f.status.name if _normalizar(status_jira) in STATUS_EXCLUIDOS: descartados_por_status += 1 continue titulo = f.summary descricao = (getattr(f, "description", "") or "")[:500] status_norm_lc = status_jira.lower() labels = list(getattr(f, "labels", []) or []) due = getattr(f, "duedate", None) start = getattr(f, "customfield_10015", None) data_ini = _str_para_data(start) data_fim = _str_para_data(due) # Normalizacao expandida pra cobrir nomes de status PT-BR/EN # ("Concluído", "Resolvido", "Encerrado", "Aprovado", "Fechado", etc.) _DONE_KEYWORDS = ("done", "concl", "fechad", "resolv", "encerr", "complet", "finaliz", "aprovad", "closed") _PROGRESS_KEYWORDS = ("progress", "andamento", "em dev", "doing", "iniciado", "started") if any(k in status_norm_lc for k in _DONE_KEYWORDS): status_norm = "done" elif any(k in status_norm_lc for k in _PROGRESS_KEYWORDS): status_norm = "em-andamento" elif data_fim and data_fim < hoje: status_norm = "atrasado" else: status_norm = "planejado" ini = Iniciativa( titulo=titulo, fonte="jira", data_inicio=data_ini, data_fim=data_fim, status=status_norm, descricao=descricao, link=f"https://bancocora.atlassian.net/browse/{iss.key}", raw={"key": iss.key, "labels": labels, "tipo": f.issuetype.name}, ) cat = _categoria_de_labels(labels) if cat: ini.categoria = cat else: ini.categoria = _classificar_categoria(titulo, descricao, "jira", {"labels": labels}) if data_fim: ini.mes_planilha = (data_fim.year, data_fim.month) iniciativas.append(ini) aceitos_keys.append(iss.key) except Exception as e: logging.warning(f"[Roadmap] Erro processando issue Jira: {e}") continue # Buscar tasks/historias filhas em batch pra derivar data fim quando epico # nao tem duedate. Decisao do usuario: usar max(duedate) das filhas como fim. if aceitos_keys: try: tasks_por_epico = _fetch_tasks_filhas(jira, aceitos_keys) n_derivados = 0 for ini in iniciativas: key = ini.raw.get("key") tasks_info = tasks_por_epico.get(key, {}) ini.n_tasks_filhas = tasks_info.get("count", 0) max_due = tasks_info.get("max_due") # Se o epico nao tem duedate proprio mas as filhas tem, deriva if not ini.data_fim and max_due: ini.data_fim = max_due ini.data_fim_derivada = True ini.mes_planilha = (max_due.year, max_due.month) # Re-avalia status atrasado se aplicavel if ini.status == "planejado" and max_due < hoje: ini.status = "atrasado" n_derivados += 1 logging.info(f"[Roadmap] {n_derivados} epicos receberam data fim derivada das tasks filhas") except Exception as e: logging.warning(f"[Roadmap] Erro buscando tasks filhas: {e}") logging.info( f"[Roadmap] Jira: {len(iniciativas)} aceitos · " f"{descartados_por_comp} por componente · " f"{descartados_por_status} por status" ) return iniciativas def _fetch_tasks_filhas(jira, epic_keys: list[str]) -> dict[str, dict]: """Busca em batch as tasks/historias filhas dos epicos via JQL `parent in (...)`. Retorna mapa: epic_key -> {'count': int, 'max_due': date | None}. Usa `parent` field (Jira Cloud moderno) ao inves de `Epic Link` legado — eh a sintaxe que o agente ja usa em jira_utils.criar_issues_do_json. Se a busca falhar (alguns Jiras antigos), tenta fallback com Epic Link. """ if not epic_keys: return {} # JQL com chunks pra evitar URL gigante (max ~50 keys por query) resultado: dict[str, dict] = {k: {"count": 0, "max_due": None} for k in epic_keys} CHUNK = 50 for i in range(0, len(epic_keys), CHUNK): chunk_keys = epic_keys[i:i + CHUNK] keys_str = ", ".join(chunk_keys) jql_modern = f"parent in ({keys_str})" try: filhas = jira.search_issues( jql_modern, maxResults=500, fields="duedate,parent,status", ) for f_iss in filhas: try: parent = getattr(f_iss.fields, "parent", None) if not parent: continue parent_key = parent.key if parent_key not in resultado: continue resultado[parent_key]["count"] += 1 due_str = getattr(f_iss.fields, "duedate", None) due = _str_para_data(due_str) if due: cur_max = resultado[parent_key]["max_due"] if not cur_max or due > cur_max: resultado[parent_key]["max_due"] = due except Exception as e: logging.warning(f"[Roadmap] Erro processando filha: {e}") continue except Exception as e: logging.warning(f"[Roadmap] JQL `parent in (...)` falhou (chunk {i}): {e}") continue total_com_filhas = sum(1 for v in resultado.values() if v["count"] > 0) total_com_due = sum(1 for v in resultado.values() if v["max_due"]) logging.info( f"[Roadmap] Tasks filhas: {total_com_filhas}/{len(epic_keys)} epicos tem filhas · " f"{total_com_due} tem max(duedate)" ) return resultado def _str_para_data(s: Optional[str]) -> Optional[date]: if not s: return None try: return date.fromisoformat(s[:10]) except (ValueError, TypeError): return None def _calc_span_indices(it: dict, janela: list[tuple[int, int]]) -> Optional[tuple[int, int]]: """Retorna (start_idx, end_idx) na janela ou None se nao deve aparecer. Se a iniciativa tem data_inicio E data_fim, a barra ocupa todo o range de meses entre elas (clipado a janela). Caso contrario usa mes_planilha como celula unica. """ di_str = it.get("data_inicio") df_str = it.get("data_fim") mp = it.get("mes_planilha") start_ym: Optional[tuple[int, int]] = None end_ym: Optional[tuple[int, int]] = None if di_str and df_str: d_ini = date.fromisoformat(di_str) if isinstance(di_str, str) else di_str d_fim = date.fromisoformat(df_str) if isinstance(df_str, str) else df_str # Garante ordem (caso de fim antes de inicio em dados sujos) if d_fim < d_ini: d_ini, d_fim = d_fim, d_ini start_ym = (d_ini.year, d_ini.month) end_ym = (d_fim.year, d_fim.month) elif mp: mp_t = (mp[0], mp[1]) if isinstance(mp, (list, tuple)) else None if mp_t: start_ym = end_ym = mp_t if start_ym is None or end_ym is None: return None # Totalmente fora da janela? if end_ym < janela[0] or start_ym > janela[-1]: return None # start_idx: primeiro mes da janela >= start_ym (ou 0 se ja passou) if start_ym <= janela[0]: start_idx = 0 else: start_idx = 0 for i, ym in enumerate(janela): if ym >= start_ym: start_idx = i break # end_idx: ultimo mes da janela <= end_ym (ou len-1 se passa do fim) if end_ym >= janela[-1]: end_idx = len(janela) - 1 else: end_idx = 0 for i, ym in enumerate(janela): if ym <= end_ym: end_idx = i return (start_idx, end_idx) def _categoria_de_labels(labels: list[str]) -> Optional[str]: """Detecta categoria a partir de labels do Jira.""" if not labels: return None lset = {_normalizar(label_) for label_ in labels} if any("kr1.1" in lset or label_ in lset for label_ in ["kr11", "kr1.1", "fpr"]): if "kr11" in lset or "kr1.1" in lset or "fpr" in lset: return "KR1.1" if "kr12" in lset or "kr1.2" in lset or "deteccao" in lset: return "KR1.2" if "kr13" in lset or "kr1.3" in lset or "chargeback" in lset or "cbk" in lset: return "KR1.3" if "debito-tecnico" in lset or "tech-debt" in lset or "debito_tecnico" in lset: return "DEBITO" if "regulatorio" in lset or "compliance" in lset or "regulatory" in lset: return "REGULATORIO" return None # --------------------------------------------------------------------------- # Fase 3a: WRITE-BACK (editar iniciativa via agente) # --------------------------------------------------------------------------- def _conectar_sheets_usuario(): """Sheets service usando OAuth do usuario logado (necessario pra escrita). A Service Account so tem leitura na planilha do squad.""" import os from google.oauth2.credentials import Credentials as OAuthCredentials from googleapiclient.discovery import build access_token = st.session_state.get("usuario_access_token") if not access_token: return None try: creds = OAuthCredentials( token=access_token, refresh_token=st.session_state.get("usuario_refresh_token"), token_uri="https://oauth2.googleapis.com/token", client_id=os.environ.get("GOOGLE_CLIENT_ID"), client_secret=os.environ.get("GOOGLE_CLIENT_SECRET"), ) return build("sheets", "v4", credentials=creds) except Exception as e: logging.warning(f"[Roadmap] Sheets OAuth falhou: {e}") return None def _atualizar_jira_epico( epic_key: str, *, titulo: Optional[str] = None, data_inicio: Optional[date] = None, data_fim: Optional[date] = None, categoria: Optional[str] = None, status_alvo: Optional[str] = None, ) -> dict: """Atualiza um epico no Jira. Retorna {ok: bool, mudancas: list[str], erro: str}. - titulo: muda summary - data_inicio: muda customfield_10015 (start date) - data_fim: muda duedate - categoria: adiciona/troca label kr1.1/kr1.2/kr1.3/debito/regulatorio - status_alvo: tenta transicionar pro status indicado """ try: from modules.jira_utils import conectar_jira except Exception as e: return {"ok": False, "mudancas": [], "erro": f"Jira indisponivel: {e}"} jira = conectar_jira() if not jira: return {"ok": False, "mudancas": [], "erro": "Jira indisponivel"} mudancas: list[str] = [] erros: list[str] = [] try: issue = jira.issue(epic_key) except Exception as e: return {"ok": False, "mudancas": [], "erro": f"Issue {epic_key} nao encontrada: {e}"} fields_update: dict = {} if titulo: fields_update["summary"] = titulo if data_inicio: fields_update["customfield_10015"] = data_inicio.isoformat() if data_fim: fields_update["duedate"] = data_fim.isoformat() if categoria and categoria != "?": # Substitui labels KR/debito/regulatorio existentes pela nova cur_labels = list(getattr(issue.fields, "labels", []) or []) cur_labels = [ label_ for label_ in cur_labels if not any(p in label_.lower() for p in ("kr1", "debito", "regulatorio")) ] label_nova = categoria.lower().replace(".", "") cur_labels.append(label_nova) fields_update["labels"] = cur_labels if fields_update: try: issue.update(fields=fields_update) mudancas.append(f"campos Jira: {list(fields_update.keys())}") except Exception as e: erros.append(f"falha ao atualizar campos: {e}") # Status: tenta transicionar if status_alvo: try: transitions = jira.transitions(issue) # Mapa expandido pra cobrir varios workflows PT-BR/EN do Cora Jira. # Match e bidirectional (transition name in nomes OR nome in transition). mapa_alvo = { "done": [ "done", "concluido", "concluído", "concluir", "fechado", "fechar", "resolvido", "resolver", "resolved", "encerrado", "encerrar", "completo", "completar", "finalizado", "finalizar", "aprovado", "aprovar", "closed", "close", ], "em-andamento": [ "in progress", "em andamento", "doing", "iniciar", "comecar", "começar", "desenvolver", "em dev", "started", "start", ], "atrasado": [], # nao eh status Jira, e derivado "planejado": [ "to do", "todo", "backlog", "open", "novo", "reabrir", "reopen", "revert", "reverter", ], } nomes_alvo = mapa_alvo.get(status_alvo, []) disponiveis = [t.get("name", "") for t in transitions] logging.info( f"[Roadmap] Transitions Jira {issue.key}: {disponiveis} | " f"alvo='{status_alvo}' procurando {nomes_alvo}" ) transition_id = None for t in transitions: t_nome = _normalizar(t.get("name", "")) if t_nome in nomes_alvo or any(n in t_nome for n in nomes_alvo): transition_id = t["id"] logging.info(f"[Roadmap] Match transition: '{t_nome}' -> id={transition_id}") break if transition_id: jira.transition_issue(issue, transition_id) mudancas.append(f"status -> {status_alvo}") elif status_alvo != "atrasado": erros.append( f"transition pra '{status_alvo}' nao encontrada. " f"Disponiveis: {disponiveis}" ) except Exception as e: erros.append(f"falha na transition: {e}") return { "ok": not erros, "mudancas": mudancas, "erro": "; ".join(erros) if erros else "", } def _get_planilha_meta() -> dict: """Re-deriva metadata da planilha (sheet_id numerico + col_to_mes + mes_to_col). Usado pelos writes pra localizar celulas. Cache curto pq raramente muda.""" svc = _build_sheets_service() if not svc: return {} meta_out: dict = {} try: meta = svc.spreadsheets().get( spreadsheetId=MASTER_ROADMAP_SHEET_ID, fields="sheets(properties)", ).execute() for sh in meta.get("sheets", []): if sh["properties"]["title"] == MASTER_ROADMAP_TAB: meta_out["sheet_id"] = sh["properties"]["sheetId"] break except Exception as e: logging.warning(f"[Roadmap] Erro pegando sheetId: {e}") return {} # Re-deriva col_to_mes (mesma logica de _fetch_planilha_roadmap) try: result = svc.spreadsheets().values().get( spreadsheetId=MASTER_ROADMAP_SHEET_ID, range=f"'{MASTER_ROADMAP_TAB}'!A1:Z2", ).execute() rows = result.get("values", []) if len(rows) < 2: return meta_out header_row = rows[1] meses_header = [] for i, val in enumerate(header_row): m_idx = MESES_PT.get(_normalizar(val)) if m_idx: meses_header.append((i, m_idx)) hoje = date.today() if meses_header: idx_atual = None for k, (col, m) in enumerate(meses_header): if m == hoje.month: idx_atual = k break if idx_atual is None: idx_atual = 0 ano_corrente = hoje.year col_to_mes = {} mes_to_col = {} for k, (col, m) in enumerate(meses_header): offset = k - idx_atual anos_off = 0 mes_running = hoje.month + offset while mes_running <= 0: mes_running += 12 anos_off -= 1 while mes_running > 12: mes_running -= 12 anos_off += 1 ym = (ano_corrente + anos_off, m) col_to_mes[col] = ym mes_to_col[ym] = col meta_out["col_to_mes"] = col_to_mes meta_out["mes_to_col"] = mes_to_col except Exception as e: logging.warning(f"[Roadmap] Erro derivando col_to_mes: {e}") return meta_out def _atualizar_planilha_iniciativa( row_idx: int, col_origem: int, *, titulo: Optional[str] = None, data_inicio: Optional[date] = None, data_fim: Optional[date] = None, ) -> dict: """Atualiza ou MOVE uma iniciativa na planilha. Operacoes (via batchUpdate): 1. Detecta merge atual em (row_idx, col_origem) — pode ser celula unica 2. Unmerge a celula atual se necessario 3. Limpa valor da celula antiga 4. Calcula nova col_inicio/col_fim a partir das datas (ou mantem origem) 5. Escreve titulo na nova col_inicio 6. Mescla [col_inicio, col_fim] se range > 1 Requer OAuth do usuario (Service Account so tem leitura).""" svc = _conectar_sheets_usuario() if not svc: return {"ok": False, "erro": "Sem OAuth do usuario; faca logout/login pra renovar token"} pm = _get_planilha_meta() sheet_id = pm.get("sheet_id") mes_to_col: dict = pm.get("mes_to_col", {}) if sheet_id is None or not mes_to_col: return {"ok": False, "erro": "Nao foi possivel ler metadata da planilha"} # Calcula nova col_inicio / col_fim if data_inicio and data_fim: ym_ini = (data_inicio.year, data_inicio.month) ym_fim = (data_fim.year, data_fim.month) col_inicio = mes_to_col.get(ym_ini, col_origem) col_fim = mes_to_col.get(ym_fim, col_inicio) if col_fim < col_inicio: col_inicio, col_fim = col_fim, col_inicio elif data_fim: ym = (data_fim.year, data_fim.month) col_inicio = col_fim = mes_to_col.get(ym, col_origem) else: col_inicio = col_fim = col_origem # Lista de requests pro batchUpdate requests_list: list[dict] = [] mudancas: list[str] = [] # 1) Tenta unmerge a celula antiga (pode nao estar mesclada, ai da erro mas # ignoramos). Limites generosos pra cobrir possivel range mesclado. requests_list.append({ "unmergeCells": { "range": { "sheetId": sheet_id, "startRowIndex": row_idx, "endRowIndex": row_idx + 1, "startColumnIndex": col_origem, "endColumnIndex": col_origem + 12, } } }) # 2) Limpa valores num range generoso ao redor da posicao antiga requests_list.append({ "updateCells": { "range": { "sheetId": sheet_id, "startRowIndex": row_idx, "endRowIndex": row_idx + 1, "startColumnIndex": col_origem, "endColumnIndex": col_origem + 12, }, "fields": "userEnteredValue", "rows": [{"values": [{"userEnteredValue": {"stringValue": ""}}] * 12}], } }) # 3) Escreve novo titulo na col_inicio if titulo: requests_list.append({ "updateCells": { "range": { "sheetId": sheet_id, "startRowIndex": row_idx, "endRowIndex": row_idx + 1, "startColumnIndex": col_inicio, "endColumnIndex": col_inicio + 1, }, "fields": "userEnteredValue", "rows": [{"values": [{"userEnteredValue": {"stringValue": titulo}}]}], } }) mudancas.append(f"titulo escrito em col {col_inicio}") # 4) Mescla [col_inicio, col_fim] se range > 1 if col_fim > col_inicio: requests_list.append({ "mergeCells": { "range": { "sheetId": sheet_id, "startRowIndex": row_idx, "endRowIndex": row_idx + 1, "startColumnIndex": col_inicio, "endColumnIndex": col_fim + 1, }, "mergeType": "MERGE_ALL", } }) mudancas.append(f"mesclado cols {col_inicio}-{col_fim}") try: # unmerge pode falhar se a celula nao tava mesclada — fazemos isolado # com try pra nao bloquear o resto try: svc.spreadsheets().batchUpdate( spreadsheetId=MASTER_ROADMAP_SHEET_ID, body={"requests": [requests_list[0]]}, ).execute() except Exception as e: logging.info(f"[Roadmap] Unmerge ignorado (provavel celula nao mesclada): {e}") # demais requests (limpeza + escrita + merge) em batch svc.spreadsheets().batchUpdate( spreadsheetId=MASTER_ROADMAP_SHEET_ID, body={"requests": requests_list[1:]}, ).execute() return {"ok": True, "mudancas": mudancas, "erro": ""} except Exception as e: return {"ok": False, "mudancas": mudancas, "erro": str(e)[:300]} # --------------------------------------------------------------------------- # Drive: indexa team folder pra linkar docs/pastas nas iniciativas # --------------------------------------------------------------------------- def _drive_url(file_id: str, mime_type: str) -> str: """Gera URL pra abrir item do Drive baseado no mimeType.""" if mime_type == "application/vnd.google-apps.folder": return f"https://drive.google.com/drive/folders/{file_id}" if mime_type == "application/vnd.google-apps.document": return f"https://docs.google.com/document/d/{file_id}/edit" if mime_type == "application/vnd.google-apps.spreadsheet": return f"https://docs.google.com/spreadsheets/d/{file_id}/edit" if mime_type == "application/vnd.google-apps.presentation": return f"https://docs.google.com/presentation/d/{file_id}/edit" return f"https://drive.google.com/file/d/{file_id}/view" def _tipo_label(mime: str) -> str: if mime == "application/vnd.google-apps.folder": return "Pasta" if mime == "application/vnd.google-apps.document": return "Doc" if mime == "application/vnd.google-apps.spreadsheet": return "Sheet" if mime == "application/vnd.google-apps.presentation": return "Slides" if mime == "application/pdf": return "PDF" return "Arquivo" @st.cache_data(ttl=7200, show_spinner=False) def _index_drive_team_folder() -> tuple[list[dict], list[dict]]: """BFS na team folder do squad ate profundidade 4. Retorna (folders, files). Cada item: {id, name, mimeType, url}. Cache 2h (refresh manual via botao). Pula pastas internas do agente (PM-Assistant, memoria, etc).""" try: from modules.drive_utils import conectar_drive except Exception as e: logging.warning(f"[Roadmap] Drive import falhou: {e}") return [], [] drive = conectar_drive() if not drive: return [], [] MAX_DEPTH = 6 folders_out: list[dict] = [] files_out: list[dict] = [] to_visit: list[tuple[str, int]] = [(TEAM_FOLDER_ID, 0)] visited: set[str] = set() while to_visit: folder_id, depth = to_visit.pop(0) if folder_id in visited or depth > MAX_DEPTH: continue visited.add(folder_id) try: page_token = None while True: resp = drive.files().list( q=f"'{folder_id}' in parents and trashed = false", fields="nextPageToken,files(id,name,mimeType)", pageSize=200, pageToken=page_token, supportsAllDrives=True, includeItemsFromAllDrives=True, ).execute() for f in resp.get("files", []): name = f.get("name", "") mime = f.get("mimeType", "") if name.lower() in EXCLUDED_DRIVE_FOLDER_NAMES_LC: continue entry = {"id": f["id"], "name": name, "mimeType": mime, "url": _drive_url(f["id"], mime)} if mime == "application/vnd.google-apps.folder": folders_out.append(entry) to_visit.append((entry["id"], depth + 1)) else: files_out.append(entry) page_token = resp.get("nextPageToken") if not page_token: break except Exception as e: logging.warning(f"[Roadmap] Erro listando pasta {folder_id}: {e}") continue logging.info(f"[Roadmap] Drive index: {len(folders_out)} pastas · {len(files_out)} arquivos") return folders_out, files_out def _score_drive_match(titulo: str, nome: str) -> float: """Score hibrido pra matching iniciativa <-> arquivo/pasta Drive. Combina: - SequenceMatcher (similaridade de chars) — pega misspellings/typos - Token overlap (F1 de palavras significativas) — robusto a diferencas de tamanho (titulo longo 'Implementar Cybersource 3DS' vs pasta 'Cybersource') - Substring bonus — se titulo contem o nome (ou vice-versa) Retorna o MAIOR dos 3 sinais. Util pra pegar matches que o ratio puro perde. """ a = _norm_titulo(titulo) b = _norm_titulo(nome) if not a or not b: return 0.0 # 1) Char-level ratio base = SequenceMatcher(None, a, b).ratio() # 2) Substring (um contem o outro inteiro) if a in b or b in a: base = max(base, 0.80) # 3) Token overlap (F1) tokens_a = {t for t in a.split() if len(t) >= 3 and t not in STOPWORDS_MATCH} tokens_b = {t for t in b.split() if len(t) >= 3 and t not in STOPWORDS_MATCH} if tokens_a and tokens_b: overlap = len(tokens_a & tokens_b) if overlap > 0: recall = overlap / len(tokens_a) precision = overlap / len(tokens_b) f1 = 2 * recall * precision / (recall + precision) # Boost: se >= 2 tokens significativos batem, eh match forte token_score = f1 if overlap < 2 else min(f1 + 0.10, 0.95) base = max(base, token_score) return base def _find_drive_link( titulo: str, folders: list[dict], files: list[dict], threshold: float = DRIVE_MATCH_THRESHOLD, ) -> Optional[dict]: """Acha melhor link Drive pra uma iniciativa. PASTA tem prioridade; se nao houver pasta com match >= threshold, tenta ARQUIVO. None se nada.""" best_folder = None best_folder_score = 0.0 for fo in folders: s = _score_drive_match(titulo, fo["name"]) if s > best_folder_score: best_folder_score = s best_folder = fo if best_folder and best_folder_score >= threshold: return {**best_folder, "_match_type": "folder", "_score": best_folder_score} best_file = None best_file_score = 0.0 for fi in files: s = _score_drive_match(titulo, fi["name"]) if s > best_file_score: best_file_score = s best_file = fi if best_file and best_file_score >= threshold: return {**best_file, "_match_type": "file", "_score": best_file_score} return None # --------------------------------------------------------------------------- # Cache + Carregamento conjunto # --------------------------------------------------------------------------- @st.cache_data(ttl=7200, show_spinner=False) def _carregar_dados_cached() -> tuple[list[dict], dict]: """Carrega das 3 fontes, mescla planilha<->jira por fuzzy match, e retorna (lista_iniciativas, diagnostico). Cache 2h (refresh manual via botao). Diagnostico inclui contagem de matches.""" diag: dict = {} planilha_items: list[Iniciativa] = [] jira_items: list[Iniciativa] = [] doc_items: list[Iniciativa] = [] try: planilha_items = _fetch_planilha_roadmap() diag["planilha"] = (len(planilha_items), None) except Exception as e: diag["planilha"] = (0, str(e)[:200]) try: jira_items = _fetch_jira_epicos() diag["jira"] = (len(jira_items), None) except Exception as e: diag["jira"] = (0, str(e)[:200]) # Fase 2: mesclagem planilha<->jira. Jira vence quando bate. # Item da planilha mesclado some da lista final; epico Jira ganha flag. todas, n_matches = _mesclar_planilha_jira(planilha_items, jira_items) diag["matches"] = n_matches # Enriquecimento Drive: linka pasta (prioritario) ou doc por fuzzy match try: folders, files = _index_drive_team_folder() n_enriched = 0 for ini in todas: # Doc regulatorio ja tem link proprio (pro doc R10-25) if ini.fonte == "doc-regulatorio": continue link_info = _find_drive_link(ini.titulo, folders, files) if link_info: ini.link_drive = link_info n_enriched += 1 logging.info(f"[Roadmap] Drive enrichment: {n_enriched}/{len(todas)} com link") diag["drive"] = (n_enriched, None) except Exception as e: logging.warning(f"[Roadmap] Drive enrichment falhou: {e}") diag["drive"] = (0, str(e)[:200]) return [_ini_to_dict(i) for i in todas], diag def _mesclar_planilha_jira( planilha: list[Iniciativa], jira: list[Iniciativa] ) -> tuple[list[Iniciativa], int]: """Mescla iniciativas da planilha com epicos do Jira por fuzzy match de titulo. Quando similaridade >= FUZZY_MATCH_THRESHOLD, o epico Jira ganha `tambem_na_planilha=True` e o item da planilha eh removido (Jira vence, conforme decisao do usuario). Retorna (lista_final, n_matches). """ if not planilha or not jira: return planilha + jira, 0 matched_planilha_idx: set[int] = set() n_matches = 0 for j in jira: melhor_idx = -1 melhor_score = 0.0 for idx, p in enumerate(planilha): if idx in matched_planilha_idx: continue score = _similaridade(j.titulo, p.titulo) if score > melhor_score: melhor_score = score melhor_idx = idx if melhor_idx >= 0 and melhor_score >= FUZZY_MATCH_THRESHOLD: matched_planilha_idx.add(melhor_idx) p_match = planilha[melhor_idx] j.tambem_na_planilha = True n_matches += 1 # Track posicao na planilha pra write-back funcionar j.raw["planilha_row"] = p_match.raw.get("row") j.raw["planilha_col"] = p_match.raw.get("col") # Herda descricao se Jira nao tem if not j.descricao and p_match.descricao: j.descricao = p_match.descricao # CRITICO: herda dados da planilha quando o epico Jira nao tem # duedate proprio nem deriva das tasks. Sem isso, epicos 'in # progress' que estao na planilha sumiam do roadmap. if not j.mes_planilha and p_match.mes_planilha: j.mes_planilha = p_match.mes_planilha # Herda range completo se a celula era mesclada (spanning) if not j.data_inicio and p_match.data_inicio: j.data_inicio = p_match.data_inicio if not j.data_fim: if p_match.data_fim: j.data_fim = p_match.data_fim else: ano, mes = p_match.mes_planilha j.data_fim = date(ano, mes, 15) logging.info( f"[Roadmap] Herdou data da planilha pro Jira sem due: " f"'{j.titulo[:40]}' -> {p_match.mes_planilha}" f"{' (range mesclado)' if p_match.data_inicio else ''}" ) logging.info( f"[Roadmap] Match {melhor_score:.0%}: '{j.titulo[:40]}' " f"<-> '{p_match.titulo[:40]}'" ) # Planilha sem match continua na lista planilha_remanescente = [ p for idx, p in enumerate(planilha) if idx not in matched_planilha_idx ] logging.info( f"[Roadmap] Merge: {n_matches} matches · " f"{len(planilha_remanescente)} planilha solo · {len(jira)} jira (todos)" ) return planilha_remanescente + jira, n_matches def _ini_to_dict(i: Iniciativa) -> dict: return { "titulo": i.titulo, "fonte": i.fonte, "categoria": i.categoria, "data_inicio": i.data_inicio.isoformat() if i.data_inicio else None, "data_fim": i.data_fim.isoformat() if i.data_fim else None, "mes_planilha": i.mes_planilha, "status": i.status, "prioridade": i.prioridade, "link": i.link, "descricao": i.descricao, "raw": i.raw, "tambem_na_planilha": i.tambem_na_planilha, "n_tasks_filhas": i.n_tasks_filhas, "data_fim_derivada": i.data_fim_derivada, "link_drive": i.link_drive, "sub_items": i.sub_items, } # --------------------------------------------------------------------------- # Consolidacao doc-regulatorio: agrupa items AP.R10-25.X em uma linha unica # "Atendimento auditoria (Bacenjud) - Automatizar processos manuais", com # mini-barras na timeline (1 por sub-item, posicionada pelo prazo). # --------------------------------------------------------------------------- # Titulo padrao quando nao encontra o item pai na planilha/Jira _DOC_REG_PARENT_TITLE = ( "Atendimento auditoria (Bacenjud) - Automatizar processos manuais" ) # Keywords pra encontrar o item pai existente na planilha/Jira _DOC_REG_PARENT_KEYWORDS = ["bacenjud", "atendimento auditoria", "auditoria bacenjud"] def _consolidar_doc_regulatorio( items: list[Iniciativa], doc_items: list[Iniciativa] ) -> list[Iniciativa]: """Consolida items doc-regulatorio em uma unica linha do roadmap. Procura item pai existente em `items` por keyword no titulo (bacenjud, atendimento auditoria). Se acha: anexa doc_items como sub_items. Se nao acha: cria item virtual com titulo padrao. Cada sub_item vira um dict com {titulo, codigo, data_inicio, data_fim, status, prioridade, link} pra ser renderizado como mini-barra na timeline pelo _render_swim_lanes. """ if not doc_items: return items # Procura item pai existente por keyword parent_idx = -1 for idx, it in enumerate(items): titulo_lower = (it.titulo or "").lower() if any(kw in titulo_lower for kw in _DOC_REG_PARENT_KEYWORDS): parent_idx = idx break # Converte doc_items pra dicts de sub_item sub_items_dicts = [] for d in doc_items: sub_items_dicts.append({ "titulo": d.titulo, "codigo": d.raw.get("codigo", ""), "data_inicio": d.data_inicio.isoformat() if d.data_inicio else None, "data_fim": d.data_fim.isoformat() if d.data_fim else None, "status": d.status, "prioridade": d.prioridade, "link": d.link, "descricao": d.descricao, }) # Calcula range total. data_inicio prefere o menor data_inicio dos # sub-items; cai pro menor data_fim se nenhum sub-item tem data_inicio # (caso comum em doc-regulatorio onde so se conhece o prazo). datas_inicio = [d.data_inicio for d in doc_items if d.data_inicio] datas_fim = [d.data_fim for d in doc_items if d.data_fim] if datas_inicio: range_inicio = min(datas_inicio) elif datas_fim: range_inicio = min(datas_fim) else: range_inicio = None range_fim = max(datas_fim) if datas_fim else None if parent_idx >= 0: # Encontrou pai: anexa sub_items, expande range pra cobrir TODOS os # prazos (mini-bars precisam de cell visivel no span do pai). parent = items[parent_idx] parent.sub_items = sub_items_dicts if range_inicio and (not parent.data_inicio or range_inicio < parent.data_inicio): parent.data_inicio = range_inicio if range_fim and (not parent.data_fim or range_fim > parent.data_fim): parent.data_fim = range_fim # Garante doc_regulatorio_link pro CTA "Abrir documentacao" if not parent.doc_regulatorio_link and doc_items: parent.doc_regulatorio_link = doc_items[0].link parent.tambem_no_doc_regulatorio = True logging.info( f"[Roadmap] doc-reg consolidado em pai existente: " f"'{parent.titulo[:50]}' com {len(sub_items_dicts)} sub-items" ) return items # Nao encontrou pai: cria item virtual parent_link = doc_items[0].link if doc_items else None virtual = Iniciativa( titulo=_DOC_REG_PARENT_TITLE, fonte="doc-regulatorio", categoria="REGULATORIO", data_inicio=range_inicio, data_fim=range_fim, mes_planilha=(range_fim.year, range_fim.month) if range_fim else None, status="em-andamento", link=parent_link, descricao=f"Plano R10-25: {len(sub_items_dicts)} acoes regulatorias com prazos individuais", raw={"virtual_doc_reg": True}, sub_items=sub_items_dicts, doc_regulatorio_link=parent_link, tambem_no_doc_regulatorio=True, ) logging.info( f"[Roadmap] doc-reg consolidado em item virtual com " f"{len(sub_items_dicts)} sub-items (sem pai na planilha/Jira)" ) return items + [virtual] # --------------------------------------------------------------------------- # Render — UI # --------------------------------------------------------------------------- def render_roadmap_tab(): """Renderiza a aba Roadmap completa — single page com 5 swim lanes horizontais empilhadas em UM grid CSS unico (estilo Emily Bustamante).""" # CSS agressivo pra forcar full-width na aba Roadmap. O Streamlit por # default limita o conteudo em ~1200px mesmo com layout='wide'. Aqui # quebramos esse limite forcando max-width: 100% em todos os containers # ancestrais conhecidos e usando margin negativo via vw pra estourar. st.markdown( """ """, unsafe_allow_html=True, ) st.markdown("### Roadmap — Squad Risco & Segurança") col_btn, col_info = st.columns([1, 4]) with col_btn: if st.button("Atualizar dados", help="Recarrega das fontes (limpa cache)"): _carregar_dados_cached.clear() _index_drive_team_folder.clear() # Limpa overlay otimista (saves recentes ainda nao confirmados # pelo cache). Refresh do cache traz dados reais do Jira/MR. st.session_state.pop("rmap_edit_overlay", None) st.rerun() with col_info: st.caption(f"Atualizado: {date.today().strftime('%d/%m/%Y')}") with st.spinner("Carregando iniciativas..."): iniciativas, diag = _carregar_dados_cached() if not iniciativas: st.warning("Nenhuma iniciativa carregada.") return # Aplica overlay otimista: saves recentes que mexeram em memoria sem # invalidar cache. Permite UI instantanea ao salvar (sem refetch lento). overlay = st.session_state.get("rmap_edit_overlay", {}) if overlay: for it in iniciativas: eid = _build_edit_id(it) ov = overlay.get(eid) if not ov: continue for k, v in ov.items(): if v is not None: it[k] = v # Recomputa mes_planilha se data_fim mudou df_str = it.get("data_fim") if df_str: try: df = date.fromisoformat(df_str) it["mes_planilha"] = (df.year, df.month) except Exception: pass # Filtros sempre visiveis (sem expander) — aplicados ANTES de separar com_data/sem_data st.caption(f"**Filtros** · {len(iniciativas)} iniciativas no total") c1, c2, c3, c4 = st.columns([1.2, 1.2, 1.2, 2]) with c1: f_status = st.multiselect( "Status", options=["em-andamento", "atrasado", "planejado", "done"], format_func=_status_label, key="rmap_f_status", ) with c2: f_categoria = st.multiselect( "KR / Categoria", options=["KR1.1", "KR1.2", "KR1.3", "DEBITO", "REGULATORIO", "?"], key="rmap_f_categoria", ) with c3: f_fonte = st.multiselect( "Fonte", options=["jira", "planilha"], format_func=_fonte_label, key="rmap_f_fonte", ) with c4: f_busca = st.text_input( "Buscar no titulo ou descricao", placeholder="ex: biometria, fpr, cybersource", key="rmap_f_busca", ) iniciativas = _aplicar_filtros(iniciativas, f_status, f_categoria, f_fonte, f_busca) if not iniciativas: st.info("Nenhuma iniciativa bate com os filtros aplicados.") return # Separa por categoria + sem_categoria + backlog com_data = [i for i in iniciativas if i.get("mes_planilha")] sem_data = [i for i in iniciativas if not i.get("mes_planilha")] # Botao "Abrir em fullscreen" - escondido quando ja em fullscreen ja_em_fullscreen = st.query_params.get("view", "") == "roadmap-fullscreen" if not ja_em_fullscreen: import streamlit.components.v1 as components fs_url_full = _build_fullscreen_url() btn_html = f""" """ components.html(btn_html, height=70) # Renderiza o grid principal. _render_swim_lanes agora retorna o valor # do custom component (None ou {action, edit_id, ...}). grid_event = _render_swim_lanes(com_data) # Processa evento vindo do custom component (sem URL navigation). # Usa key de session_state pra evitar reprocessar o mesmo evento em # reruns subsequentes (component retorna o ultimo valor enquanto nao # houver novo). if grid_event and isinstance(grid_event, dict): evt_id = f"{grid_event.get('action')}|{grid_event.get('edit_id')}|" + \ f"{grid_event.get('titulo','')}|{grid_event.get('data_fim','')}|" + \ f"{grid_event.get('link_url','')}" if st.session_state.get("_rmap_last_grid_event_id") != evt_id: st.session_state["_rmap_last_grid_event_id"] = evt_id _processar_evento_componente(grid_event, iniciativas) st.rerun() # Editor entre grid e backlog (fallback @st.dialog via URL params) st.markdown("---") _render_editor_iniciativa(iniciativas) # Backlog (sem data) embaixo (UMA UNICA VEZ) st.markdown("---") _render_backlog(sem_data) def _processar_evento_componente(evt: dict, iniciativas: list[dict]) -> None: """Processa eventos vindos do custom component (sem URL navigation). Aplica edits/links direto no session_state overlay - rerun do Streamlit aplica overlay sobre cache. Sem refetch das APIs externas. """ action = evt.get("action") edit_id = evt.get("edit_id") or "" if not action or not edit_id: return matching = next( (i for i in iniciativas if _build_edit_id(i) == edit_id), None, ) if not matching: return if action == "save": titulo = evt.get("titulo") or matching.get("titulo", "") di_str = evt.get("data_inicio") or None df_str = evt.get("data_fim") or None categoria = evt.get("categoria") or "?" status_alvo = evt.get("status") or "planejado" try: data_inicio = date.fromisoformat(di_str) if di_str else None except Exception: data_inicio = None try: data_fim = date.fromisoformat(df_str) if df_str else None except Exception: data_fim = None ok = _executar_edicao( matching, titulo, data_inicio, data_fim, categoria, status_alvo, ) if ok: overlay = st.session_state.setdefault("rmap_edit_overlay", {}) overlay[edit_id] = { "titulo": titulo, "data_inicio": di_str, "data_fim": df_str, "categoria": categoria, "status": status_alvo, } st.toast("Iniciativa atualizada", icon="✅") return if action == "link_url": link_type = evt.get("link_type", "") link_url = evt.get("link_url", "") if not link_type or not link_url: return overlay = st.session_state.setdefault("rmap_edit_overlay", {}) cur = overlay.get(edit_id) or {} if link_type == "doc": cur["link_drive"] = { "id": "manual-link", "name": "Documentação (linkada manualmente)", "mimeType": "application/vnd.google-apps.document", "url": link_url, "_match_type": "file", "_score": 1.0, } elif link_type == "jira": cur["link"] = link_url if link_url.startswith("http") else ( f"https://bancocora.atlassian.net/browse/{link_url}" ) cur["_user_linked_jira"] = link_url elif link_type == "mr": cur["tambem_na_planilha"] = True cur["_user_linked_mr"] = link_url overlay[edit_id] = cur st.toast(f"Link {link_type} salvo", icon="✅") def _render_editor_iniciativa(iniciativas: list[dict]): """Dispatcher do editor de iniciativas. Dois caminhos: 1. ?action=save&edit=&titulo=...&data_inicio=... — processa save direto (modal HTML do iframe enviou o form via URL). Limpa params e rerun. 2. ?edit= sem action — fallback: abre o @st.dialog modal Streamlit (caminho antigo, ainda funciona se JS falhar). """ if not iniciativas: return action = st.query_params.get("action", "") edit_id_param = st.query_params.get("edit", "") if not edit_id_param: return matching = next( (i for i in iniciativas if _build_edit_id(i) == edit_id_param), None, ) if not matching: return if action == "link_url": # Linka URL externa (doc/jira/MR) sem editar nada na fonte. # Armazena no session_state overlay; iniciativa ganha link_drive # ou tambem_na_planilha via overlay no proximo render. link_type = st.query_params.get("link_type", "") link_url = st.query_params.get("link_url", "") if link_type and link_url: overlay = st.session_state.setdefault("rmap_edit_overlay", {}) cur = overlay.get(edit_id_param) or {} if link_type == "doc": # Simula um link_drive (mesmo shape que o fetch real) cur["link_drive"] = { "id": "manual-link", "name": "Documentação (linkada manualmente)", "mimeType": "application/vnd.google-apps.document", "url": link_url, "_match_type": "file", "_score": 1.0, } elif link_type == "jira": # User informou que existe um Jira correspondente cur["link"] = link_url if link_url.startswith("http") else ( f"https://bancocora.atlassian.net/browse/{link_url}" ) cur["_user_linked_jira"] = link_url elif link_type == "mr": # Apenas marca como tambem_na_planilha (link visual nao temos) cur["tambem_na_planilha"] = True cur["_user_linked_mr"] = link_url overlay[edit_id_param] = cur st.toast(f"Link {link_type} salvo (overlay local)", icon="✅") for k in ("action", "edit", "link_type", "link_url"): if k in st.query_params: del st.query_params[k] st.rerun() return if action == "save": # Save vindo do modal HTML do iframe via URL titulo = st.query_params.get("titulo", "") or matching.get("titulo", "") di_str = st.query_params.get("data_inicio", "") df_str = st.query_params.get("data_fim", "") categoria = st.query_params.get("categoria", "?") status_alvo = st.query_params.get("status", "planejado") try: data_inicio = date.fromisoformat(di_str) if di_str else None except Exception: data_inicio = None try: data_fim = date.fromisoformat(df_str) if df_str else None except Exception: data_fim = None ok = _executar_edicao( matching, titulo, data_inicio, data_fim, categoria, status_alvo, ) # Limpa params do save SEMPRE for k in ("action", "edit", "titulo", "data_inicio", "data_fim", "categoria", "status"): if k in st.query_params: del st.query_params[k] if ok: # Overlay otimista: aplica o novo valor em memoria sem invalidar # cache. UI ja mostra o estado novo no rerun (rapido, sem refetch # das APIs Jira/Sheets). Cache eventualmente refresca em 2h ou # via botao "Atualizar dados", que limpa o overlay junto. overlay = st.session_state.setdefault("rmap_edit_overlay", {}) overlay[edit_id_param] = { "titulo": titulo, "data_inicio": di_str or None, "data_fim": df_str or None, "categoria": categoria, "status": status_alvo, } st.toast("Iniciativa atualizada", icon="✅") st.rerun() return # Fallback: caminho antigo (@st.dialog) — usado se JS do modal falhar _editor_modal(matching) @st.dialog("Editar / mover iniciativa", width="large") def _editor_modal(ini: dict): """Modal pra editar titulo / datas / categoria / status. Escreve em Jira + planilha quando aplicavel. Abre via ?edit= na URL.""" fonte = ini.get("fonte", "") tem_match = bool(ini.get("tambem_na_planilha")) # Header info da iniciativa selecionada fonte_l = _fonte_label(fonte) match_tag = " · ↔ tambem no Master Roadmap" if tem_match else "" st.markdown(f"**{_escape_html(ini.get('titulo', ''))}**") st.caption(f"Fonte: {fonte_l}{match_tag}") if fonte == "jira" and not tem_match: st.warning( "Sem match na planilha. Edicoes atualizam apenas o Jira. " "Pra refletir tambem no Master Roadmap, adicione manualmente la.", icon="⚠️", ) elif fonte == "planilha": st.info( "Apenas na planilha. Edicoes atualizam apenas o Master Roadmap. " "Pra refletir tambem no Jira, crie o epico manualmente la.", icon="ℹ️", ) titulo_default = ini.get("titulo", "") di_str = ini.get("data_inicio") df_str = ini.get("data_fim") di_default = date.fromisoformat(di_str) if di_str else None df_default = date.fromisoformat(df_str) if df_str else None cat_default = ini.get("categoria", "?") status_default = ini.get("status", "planejado") cats = ["KR1.1", "KR1.2", "KR1.3", "DEBITO", "REGULATORIO", "?"] with st.form(f"edit_modal_{_build_edit_id(ini)}"): novo_titulo = st.text_input("Titulo", value=titulo_default) col_d1, col_d2 = st.columns(2) with col_d1: nova_di = st.date_input( "Data inicio", value=di_default, help="Jira: customfield_10015 (start). Planilha: coluna inicial.", ) with col_d2: nova_df = st.date_input( "Data fim", value=df_default, help="Jira: duedate. Planilha: coluna final do merge.", ) col_c1, col_c2 = st.columns(2) with col_c1: nova_cat = st.selectbox( "Categoria / KR", options=cats, index=cats.index(cat_default) if cat_default in cats else 5, help="Jira: atualiza labels. Planilha: mover de linha nao suportado.", ) with col_c2: opts_status = ["planejado", "em-andamento", "atrasado", "done"] novo_status = st.selectbox( "Status", options=opts_status, index=opts_status.index(status_default) if status_default in opts_status else 0, format_func=_status_label, help="Jira: tenta transition. 'Atrasado' eh derivado.", ) col_btn1, col_btn2 = st.columns(2) with col_btn1: sub = st.form_submit_button("Salvar alteracoes", type="primary", use_container_width=True) with col_btn2: cancelar = st.form_submit_button("Cancelar", use_container_width=True) if cancelar: if "edit" in st.query_params: del st.query_params["edit"] st.rerun() if sub: ok = _executar_edicao( ini, novo_titulo, nova_di, nova_df, nova_cat, novo_status, ) if ok: if "edit" in st.query_params: del st.query_params["edit"] _carregar_dados_cached.clear() st.rerun() def _executar_edicao(ini: dict, titulo: str, data_inicio: Optional[date], data_fim: Optional[date], categoria: str, status_alvo: str) -> bool: """Aplica edicao em Jira e/ou planilha. Toast em sucesso, st.error em falha (modal mantem aberto pra ler erro). Retorna True se tudo OK.""" fonte = ini.get("fonte", "") raw = ini.get("raw", {}) or {} epic_key = raw.get("key") pl_row = raw.get("row") if fonte == "planilha" else raw.get("planilha_row") pl_col = raw.get("col") if fonte == "planilha" else raw.get("planilha_col") # Detecta o que mudou (so envia o que mudou) mudou_titulo = titulo and titulo != ini.get("titulo") cur_di = date.fromisoformat(ini["data_inicio"]) if ini.get("data_inicio") else None cur_df = date.fromisoformat(ini["data_fim"]) if ini.get("data_fim") else None mudou_di = data_inicio and data_inicio != cur_di mudou_df = data_fim and data_fim != cur_df mudou_cat = categoria and categoria != ini.get("categoria") mudou_status = status_alvo and status_alvo != ini.get("status") if not any([mudou_titulo, mudou_di, mudou_df, mudou_cat, mudou_status]): st.info("Nenhuma mudanca detectada.") return False overall_ok = True # 1) Atualizar Jira (se aplicavel) if epic_key: with st.spinner(f"Atualizando Jira {epic_key}..."): r_jira = _atualizar_jira_epico( epic_key, titulo=titulo if mudou_titulo else None, data_inicio=data_inicio if mudou_di else None, data_fim=data_fim if mudou_df else None, categoria=categoria if mudou_cat else None, status_alvo=status_alvo if mudou_status else None, ) if r_jira["ok"]: st.toast(f"✓ Jira {epic_key} atualizado", icon="✅") else: st.error(f"✗ Jira {epic_key}: {r_jira['erro']}") overall_ok = False # 2) Atualizar planilha (se aplicavel) if pl_row is not None and pl_col is not None and (mudou_titulo or mudou_di or mudou_df): with st.spinner("Atualizando planilha..."): r_pl = _atualizar_planilha_iniciativa( pl_row, pl_col, titulo=titulo if mudou_titulo else ini.get("titulo"), data_inicio=data_inicio if mudou_di else cur_di, data_fim=data_fim if mudou_df else cur_df, ) if r_pl["ok"]: st.toast("✓ Master Roadmap atualizado", icon="✅") else: st.error(f"✗ Planilha: {r_pl['erro']}") overall_ok = False if mudou_cat and fonte == "planilha": st.warning( "Mudanca de categoria na planilha exige mover de linha — ainda nao " "suportado. A categoria nova so vai refletir na visualizacao.", icon="⚠️", ) return overall_ok # --------------------------------------------------------------------------- # Swim lanes — grid CSS unico com 5 sections empilhadas # --------------------------------------------------------------------------- # Cores por swim lane (borda esquerda + cor do section header) SWIM_LANE_CONFIG = [ ("KR1.1", "KR1.1 — FPR (Reduzir impacto em clientes bons)", "#FE3E6D"), ("KR1.2", "KR1.2 — Detecção (Aumentar % de fraude detectada)", "#2563eb"), ("KR1.3", "KR1.3 — Chargeback (Manter CBK de LP < 1%)", "#059669"), ("DEBITO", "Débitos Técnicos — Manutenção operacional, scaling, migrações", "#6b7280"), ("REGULATORIO", "Regulatórios — R10-25, BACEN, LGPD, judiciais", "#f59e0b"), ] def _render_swim_lanes(items_com_data: list[dict]): """Renderiza grid CSS com 5 swim lanes empilhadas — design rico inspirado no roadmap da Emily Bustamante. Inclui: header sticky com meses, linha vermelha vertical de TODAY, section headers coloridos por lane, badges de status e fonte, cards visuais com cores de status.""" janela = _janela_meses() hoje = date.today() n_meses = len(janela) col_label_w = 320 col_mes_w = 140 # Posicao da linha TODAY: % dentro da janela. # Calcula o offset do mes atual + offset dentro do mes (dia/30). today_idx = next((i for i, (a, m) in enumerate(janela) if (a, m) == (hoje.year, hoje.month)), None) today_offset_px = None if today_idx is not None: # Posicao no header de meses + posicao dentro do mes dia_pct = hoje.day / 30.0 today_offset_px = col_label_w + (today_idx + dia_pct) * col_mes_w css = f""" """ # Linha de header (meses) header_cells = ['
Iniciativa
'] for ano, mes in janela: cls = "rmap-mes" + (" atual" if (ano, mes) == (hoje.year, hoje.month) else "") header_cells.append(f'
{_label_mes(ano, mes)}
') # Map cor → soft pra usar nos CSS vars cor_soft_map = { "#FE3E6D": "var(--rmap-pink-soft)", "#2563eb": "var(--rmap-blue-soft)", "#059669": "var(--rmap-green-soft)", "#6b7280": "var(--rmap-gray-soft)", "#f59e0b": "var(--rmap-amber-soft)", } # 5 sections empilhadas section_html = [] for cat_key, cat_titulo, cor in SWIM_LANE_CONFIG: items_cat = [i for i in items_com_data if i["categoria"] == cat_key] # Section header partes = cat_titulo.split(" — ", 1) nome_lane = partes[0] subtitulo = partes[1] if len(partes) > 1 else "" soft = cor_soft_map.get(cor, "#fafafa") section_html.append( f'
' f'{_escape_html(nome_lane)}' f'{len(items_cat)}' f'{_escape_html(subtitulo)}' f'
' ) if not items_cat: section_html.append( '
Nenhuma iniciativa nesta categoria
' ) continue # Linhas de iniciativas (ordena por data) items_cat = sorted(items_cat, key=lambda i: i["mes_planilha"]) for it in items_cat: span = _calc_span_indices(it, janela) if span is None: continue start_idx, end_idx = span n_span = end_idx - start_idx + 1 titulo = _escape_html(it["titulo"])[:90] link = it.get("link") or "" fonte = it.get("fonte", "") descricao = _escape_html((it.get("descricao") or "")[:120]) status = it.get("status", "planejado") prioridade = it.get("prioridade") # Titulo sempre texto puro (nao clicavel). Acesso a documentacao # eh via CTA "Abrir documentacao" abaixo. titulo_html = titulo # Monta badges badges = [f'{_status_label(status)}'] badges.append( f'{_fonte_label(fonte)}' ) if prioridade: badges.append( f'{_prio_label(prioridade)}' ) if it.get("tambem_na_planilha"): badges.append( '↔ planilha' ) n_tasks = it.get("n_tasks_filhas") or 0 if n_tasks > 0: derived = " (data fim derivada)" if it.get("data_fim_derivada") else "" badges.append( f'{n_tasks}↳' ) # GAPS — clicaveis abrem modal INLINE pra colar URL do doc/jira/MR. # Sao apenas links (armazenados em session_state ate o usuario # registrar permanentemente no sistema correspondente). eid_for_link = _build_edit_id(it) link_url_for_save = _build_app_url(eid_for_link) if not it.get("link_drive") and fonte != "doc-regulatorio": badges.append( f'+ doc' ) if fonte == "planilha" and not it.get("_user_linked_jira"): badges.append( f'+ Jira' ) elif fonte == "jira" and not it.get("tambem_na_planilha"): badges.append( f'+ MR' ) # CTA "Abrir documentacao" — aparece quando ha link Drive (pasta ou # doc) OU quando a propria fonte eh um documento (doc-regulatorio). ld = it.get("link_drive") drive_cta = "" row_cls = "rmap-row-label" if ld: row_cls += " has-drive" nome_drive = _escape_html(ld.get("name", "")) cls_drive = ( "drive-folder" if ld.get("_match_type") == "folder" else "drive-file" ) drive_cta = ( f'' f'↗ Abrir documentação' ) elif fonte == "doc-regulatorio" and link: row_cls += " has-drive" drive_cta = ( f'' f'↗ Abrir documentação' ) # Botao Editar — abre modal HTML INSTANTANEAMENTE via JS # (sem reload). Data-attrs carregam os valores atuais do item; # JS popula o modal e mostra. Save submete via navegacao # window.parent (in-place reload). edit_btn = "" if fonte in ("jira", "planilha"): eid = _build_edit_id(it) edit_url = _build_app_url(eid) tem_match = "true" if it.get("tambem_na_planilha") else "false" data_inicio_attr = _escape_html(it.get("data_inicio") or "") data_fim_attr = _escape_html(it.get("data_fim") or "") cat_attr = _escape_html(it.get("categoria") or "?") status_attr = _escape_html(it.get("status") or "planejado") titulo_attr = _escape_html(it.get("titulo") or "") edit_btn = ( f'✎ Editar' ) # Label da iniciativa label_html = ( f'
' f'{titulo_html}' ) # Acoes (Editar + Drive) na mesma linha if edit_btn or drive_cta: label_html += f'
{edit_btn}{drive_cta}
' if descricao: label_html += f'{descricao}' label_html += f'
{"".join(badges)}
' section_html.append(label_html) # Celulas da timeline: vazias antes do start_idx. desc_attr = _escape_html((it.get("descricao") or "")[:300]) for ji in range(start_idx): ja, jm = janela[ji] cell_cls = "rmap-cell" + (" atual" if (ja, jm) == (hoje.year, hoje.month) else "") section_html.append(f'
') # Branch: item com sub_items (doc-regulatorio consolidado) -> mini-barras # por mes, posicionadas pelo prazo de cada sub-item. Outros items # mantem 1 barra spanning n_span colunas. sub_items = it.get("sub_items") or [] if sub_items: # Agrupa sub-items pelo mes do prazo (data_fim) sub_por_mes: dict[tuple[int, int], list[dict]] = {} for sub in sub_items: df_str = sub.get("data_fim") if not df_str: continue try: sub_df = date.fromisoformat(df_str) except Exception: continue sub_por_mes.setdefault((sub_df.year, sub_df.month), []).append(sub) # Renderiza 1 cell por mes do span; se tem sub-item naquele mes, # poe 1 ou mais mini-barras. Senao, cell vazia. for ji in range(start_idx, end_idx + 1): ja, jm = janela[ji] cell_cls = "rmap-cell" + (" atual" if (ja, jm) == (hoje.year, hoje.month) else "") subs_aqui = sub_por_mes.get((ja, jm), []) if subs_aqui: mini_bars = [] for sub in subs_aqui: sub_titulo = sub.get("titulo") or "" sub_codigo = sub.get("codigo") or "" sub_status = sub.get("status") or "planejado" sub_prio = sub.get("prioridade") or "" # Label curto: codigo abreviado (AP.3) + titulo label = sub_codigo.replace("AP.R10-25.", "AP.") if sub_codigo else "AP" # Tooltip rico tooltip_parts = [sub_titulo] if sub.get("data_fim"): tooltip_parts.append(f"Prazo: {sub['data_fim']}") if sub_prio: tooltip_parts.append(f"Prioridade: {sub_prio}") tooltip = _escape_html(" | ".join(tooltip_parts)) prio_cls = f" prio-{sub_prio}" if sub_prio else "" mini_bars.append( f'
{_escape_html(label)}
' ) section_html.append( f'
' f'{"".join(mini_bars)}
' ) else: section_html.append(f'
') else: bar = ( f'
' f'{titulo[:60 if n_span > 1 else 35]}' f'
{_status_label(status)}
' f'
' ) section_html.append( f'
{bar}
' ) for ji in range(end_idx + 1, len(janela)): ja, jm = janela[ji] cell_cls = "rmap-cell" + (" atual" if (ja, jm) == (hoje.year, hoje.month) else "") section_html.append(f'
') # Linha TODAY vertical (sobreposta no grid) today_line = "" if today_offset_px is not None: today_line = f'
' # Modal HTML de edicao: renderizado uma vez no iframe, populado # dinamicamente via JS ao clicar Editar. Abre INSTANTANEO (sem reload). # Save submete via window.parent.location.href (in-place reload). edit_modal_html = """

Editar / mover iniciativa

 """ # Modal SECUNDARIO: linkar doc/Jira/MR. User cola a URL ou key # (ex: 'https://docs.google.com/document/d/abc' ou 'SAF-1234') e o save # armazena no session_state overlay - mostra como CTA proximo aos cards. link_modal_html = """ """ html_grid = ( css + '
' + "".join(header_cells) + "".join(section_html) + today_line + "
" + edit_modal_html + link_modal_html ) # Cap altura pequena pra forcar scroll INTERNO do iframe (necessario pro # header sticky de meses funcionar — sem isso o iframe cresce pra caber # tudo, scroll vai pra pagina e sticky perde referencia). # Em fullscreen damos mais espaco vertical. is_fs = st.query_params.get("view", "") == "roadmap-fullscreen" altura = 800 if is_fs else 650 # Custom component (declared) - iframe fica MOUNTED entre reruns. # Recebe markup via postMessage; ao salvar/linkar, retorna o evento via # Streamlit.setComponentValue (sem URL navigation/reload). Retorna None # ou {action, edit_id, ...}. from modules.components.roadmap_grid import roadmap_grid return roadmap_grid( grid_markup=html_grid, height=altura, key="rmap_grid_main", default=None, ) def _build_full_html(items_com_data: list[dict], items_sem_data: list[dict]) -> str: """Constroi um documento HTML standalone completo (DOCTYPE + head + body) pra abrir em nova aba do browser em fullscreen. Inclui o mesmo grid e CSS que o _render_swim_lanes, mas em pagina propria sem limites de iframe.""" janela = _janela_meses() hoje = date.today() n_meses = len(janela) col_label_w = 320 col_mes_w = 160 # um pouco maior em fullscreen today_idx = next((i for i, (a, m) in enumerate(janela) if (a, m) == (hoje.year, hoje.month)), None) today_offset_px = None if today_idx is not None: dia_pct = hoje.day / 30.0 today_offset_px = col_label_w + (today_idx + dia_pct) * col_mes_w # CSS (reuso do mesmo padrao do _render_swim_lanes + estilos de body) css = f""" """ # Header (linha de meses) header_cells = ['
Iniciativa
'] for ano, mes in janela: cls = "rmap-mes" + (" atual" if (ano, mes) == (hoje.year, hoje.month) else "") header_cells.append(f'
{_label_mes(ano, mes)}
') cor_soft_map = { "#FE3E6D": "var(--rmap-pink-soft)", "#2563eb": "var(--rmap-blue-soft)", "#059669": "var(--rmap-green-soft)", "#6b7280": "var(--rmap-gray-soft)", "#f59e0b": "var(--rmap-amber-soft)", } section_html = [] for cat_key, cat_titulo, cor in SWIM_LANE_CONFIG: items_cat = [i for i in items_com_data if i["categoria"] == cat_key] partes = cat_titulo.split(" — ", 1) nome_lane = partes[0] subtitulo = partes[1] if len(partes) > 1 else "" soft = cor_soft_map.get(cor, "#fafafa") section_html.append( f'
' f'{_escape_html(nome_lane)}' f'{len(items_cat)}' f'{_escape_html(subtitulo)}
' ) if not items_cat: section_html.append('
Nenhuma iniciativa nesta categoria
') continue items_cat = sorted(items_cat, key=lambda i: i["mes_planilha"]) for it in items_cat: span = _calc_span_indices(it, janela) if span is None: continue start_idx, end_idx = span n_span = end_idx - start_idx + 1 titulo = _escape_html(it["titulo"])[:90] link = it.get("link") or "" fonte = it.get("fonte", "") descricao = _escape_html((it.get("descricao") or "")[:120]) status = it.get("status", "planejado") prioridade = it.get("prioridade") titulo_html = titulo # nunca clicavel; acesso via CTA "Abrir documentacao" badges = [f'{_status_label(status)}', f'{_fonte_label(fonte)}'] if prioridade: badges.append(f'{_prio_label(prioridade)}') if it.get("tambem_na_planilha"): badges.append('↔ planilha') n_tasks = it.get("n_tasks_filhas") or 0 if n_tasks > 0: derived = " (derivada)" if it.get("data_fim_derivada") else "" badges.append(f'{n_tasks}↳') if not it.get("link_drive") and fonte != "doc-regulatorio": badges.append('Sem doc') ld = it.get("link_drive") drive_cta = "" row_cls = "rmap-row-label" if ld: row_cls += " has-drive" nome_drive = _escape_html(ld.get("name", "")) cls_drive = "drive-folder" if ld.get("_match_type") == "folder" else "drive-file" drive_cta = f'↗ Abrir documentação' elif fonte == "doc-regulatorio" and link: row_cls += " has-drive" drive_cta = f'↗ Abrir documentação' label_html = f'
{titulo_html}' if drive_cta: label_html += drive_cta if descricao: label_html += f'{descricao}' label_html += f'
{"".join(badges)}
' section_html.append(label_html) for ji in range(start_idx): ja, jm = janela[ji] cell_cls = "rmap-cell" + (" atual" if (ja, jm) == (hoje.year, hoje.month) else "") section_html.append(f'
') bar = f'
{titulo[:60 if n_span > 1 else 35]}
{_status_label(status)}
' section_html.append( f'
{bar}
' ) for ji in range(end_idx + 1, len(janela)): ja, jm = janela[ji] cell_cls = "rmap-cell" + (" atual" if (ja, jm) == (hoje.year, hoje.month) else "") section_html.append(f'
') today_line = "" if today_offset_px is not None: today_line = f'
' # Backlog cards backlog_html = "" if items_sem_data: cards = [] for it in items_sem_data: link = it.get("link") or "" titulo = _escape_html(it["titulo"])[:80] desc = _escape_html((it.get("descricao") or "")[:100]) fonte = it.get("fonte", "") titulo_html = f'{titulo}' if link else titulo cards.append( f'
{titulo_html}
' f'
{desc}
' f'
_{fonte}_
' ) backlog_html = ( f'

Backlog — {len(items_sem_data)} sem mês

' f'
{"".join(cards)}
' ) return f""" Roadmap — Risco & Segurança {css}

Roadmap — Squad Risco & Segurança

Single view · 5 swim lanes · Sliding window 6m+6m · Atualizado: {hoje.strftime('%d/%m/%Y')}
{"".join(header_cells)}{"".join(section_html)}{today_line}
{backlog_html} """ def _status_label(status: str) -> str: return { "done": "ENTREGUE ✓", "em-andamento": "EM ANDAMENTO", "atrasado": "ATRASADO", "planejado": "PLANEJADO", }.get(status, status.upper()) def _fonte_label(fonte: str) -> str: return { "jira": "JIRA", "planilha": "ROADMAP", "doc-regulatorio": "R10-25", }.get(fonte, fonte.upper()) def _prio_label(prio: str) -> str: return { "alta": "CRÍTICA", "media": "ALTA", "baixa": "MÉDIA", }.get(prio, prio.upper()) # --------------------------------------------------------------------------- # Backlog (iniciativas sem mes definido) com botao "Priorizar" # --------------------------------------------------------------------------- def _render_backlog(items: list[dict]): """Renderiza o backlog de iniciativas sem mes definido como cards horizontais, cada um com um botao 'Priorizar' que abre form pra setar categoria + mes. Em Fase 1, save eh stub (mostra toast). Fase 2 grava de volta na fonte (Jira/planilha).""" if not items: st.info("Sem iniciativas no backlog — tudo tem mês definido.") return st.markdown(f"### Backlog — {len(items)} iniciativa(s) sem mês definido") st.caption( "Cards abaixo são iniciativas sem mês na fonte. Clica em " "**Priorizar →** pra mover pra uma swim lane com data definida." ) # Estado: id do item sendo priorizado (mostra form expandido) if "_roadmap_priorizando" not in st.session_state: st.session_state["_roadmap_priorizando"] = None # Grid de cards: 3 por linha cards_per_row = 3 for i in range(0, len(items), cards_per_row): cols = st.columns(cards_per_row) for col_idx, it in enumerate(items[i:i + cards_per_row]): with cols[col_idx]: _render_backlog_card(it, key_prefix=f"bk_{i + col_idx}") def _render_backlog_card(it: dict, key_prefix: str): """Card individual do backlog.""" titulo = it["titulo"] fonte = it.get("fonte", "") link = it.get("link") or "" descricao = (it.get("descricao") or "")[:100] categoria_atual = it.get("categoria", "?") titulo_curto = titulo[:60] + ("..." if len(titulo) > 60 else "") with st.container(border=True): st.markdown(f"**{titulo_curto}**") if descricao: st.caption(descricao) st.caption(f"_{fonte}_ · categoria atual: `{categoria_atual}`") if link: st.markdown(f"[Abrir fonte]({link})") # Botao priorizar priorizando_key = key_prefix + "_priorizando" if st.session_state.get(priorizando_key): _render_form_priorizar(it, key_prefix) else: if st.button("Priorizar →", key=key_prefix + "_btn"): st.session_state[priorizando_key] = True st.rerun() def _render_form_priorizar(it: dict, key_prefix: str): """Form pra priorizar item do backlog: escolhe categoria + mes alvo + status, escreve no Jira via _atualizar_jira_epico.""" fonte = it.get("fonte", "") epic_key = (it.get("raw") or {}).get("key") if fonte != "jira" or not epic_key: st.warning( f"Priorizacao automatica suportada apenas pra itens Jira. " f"Esta iniciativa eh fonte '{fonte}'. Edite manualmente na fonte.", icon="⚠️", ) if st.button("Fechar", key=key_prefix + "_fechar"): st.session_state[key_prefix + "_priorizando"] = False st.rerun() return swim_options = [c[0] for c in SWIM_LANE_CONFIG] swim_labels = [c[1].split(" — ")[0] for c in SWIM_LANE_CONFIG] janela = _janela_meses() mes_options = [_label_mes(a, m) for a, m in janela] nova_cat = st.selectbox( "Categoria / KR", options=swim_options, format_func=lambda c: swim_labels[swim_options.index(c)], key=key_prefix + "_cat", ) col_mi, col_mf = st.columns(2) with col_mi: mes_inicio_label = st.selectbox( "Mes inicio (data inicio)", options=mes_options, index=len(mes_options) // 2, # default: mes atual key=key_prefix + "_mes_ini", ) with col_mf: mes_fim_label = st.selectbox( "Mes fim (data fim)", options=mes_options, index=len(mes_options) // 2, # default: mes atual key=key_prefix + "_mes_fim", ) novo_status = st.selectbox( "Status", options=["planejado", "em-andamento", "done"], format_func=_status_label, key=key_prefix + "_status", ) col1, col2 = st.columns(2) with col1: if st.button("Salvar (escreve no Jira)", key=key_prefix + "_salvar", type="primary"): idx_ini = mes_options.index(mes_inicio_label) idx_fim = mes_options.index(mes_fim_label) # Se usuario inverteu, troca if idx_fim < idx_ini: idx_ini, idx_fim = idx_fim, idx_ini ano_i, mes_i = janela[idx_ini] ano_f, mes_f = janela[idx_fim] data_inicio_alvo = date(ano_i, mes_i, 1) data_fim_alvo = date(ano_f, mes_f, 28) with st.spinner(f"Atualizando Jira {epic_key}..."): r = _atualizar_jira_epico( epic_key, data_inicio=data_inicio_alvo, data_fim=data_fim_alvo, categoria=nova_cat, status_alvo=novo_status, ) if r["ok"]: st.success( f"✓ Jira {epic_key} priorizado: " f"{', '.join(r['mudancas']) or 'sem mudancas'}" ) _carregar_dados_cached.clear() st.session_state[key_prefix + "_priorizando"] = False st.toast( "Cache invalidado — proxima visualizacao mostra atualizado", icon="ℹ️", ) st.rerun() else: st.error(f"Falha: {r['erro']}") with col2: if st.button("Cancelar", key=key_prefix + "_cancelar"): st.session_state[key_prefix + "_priorizando"] = False st.rerun() def _escape_html(s: str) -> str: import html return html.escape(s or "") def _build_edit_id(it: dict) -> str: """ID curto e estavel pra identificar iniciativa via ?edit= na URL.""" fonte = it.get("fonte", "") raw = it.get("raw") or {} if fonte == "jira": key = raw.get("key", "") if key: return f"j-{key}" if fonte == "planilha": r, c = raw.get("row"), raw.get("col") if r is not None and c is not None: return f"p-{r}-{c}" import hashlib return "h-" + hashlib.md5(it.get("titulo", "").encode()).hexdigest()[:8] def _build_app_url(edit_id: str) -> str: """URL pra navegacao IN-PLACE do botao Editar. CRITICO: usa hf.space direto (NAO o wrapper huggingface.co). Razao: - O iframe Streamlit roda em hf.space. window.parent.location.href pra hf.space e' same-origin -> navega o iframe Streamlit in-place sem sair do wrapper HF. - Se usar wrapper URL (huggingface.co), o browser tenta carregar HF wrapper DENTRO do iframe -> HF retorna X-Frame-Options: DENY -> 'conexao recusada'. Propaga sid pra auto-login restaurar sessao.""" import os space_host = os.environ.get("SPACE_HOST", "").strip() if space_host: base = f"https://{space_host}" else: try: host = (st.context.headers.get("host") or "").strip() if st.context else "" except Exception: host = "" if not host: host = "localhost:8501" proto = "http" if host.startswith("localhost") or host.startswith("127.0.0.1") else "https" base = f"{proto}://{host}" params: list[str] = [] cur_sid = st.query_params.get("sid", "") if cur_sid: params.append(f"sid={cur_sid}") cur_view = st.query_params.get("view", "") if cur_view: params.append(f"view={cur_view}") if edit_id: params.append(f"edit={edit_id}") return f"{base}/?{'&'.join(params)}" if params else f"{base}/" def _build_fullscreen_url() -> str: """URL pra abrir o roadmap em fullscreen em NOVA ABA. Usa WRAPPER huggingface.co (nao hf.space direto) porque o usuario quer a experiencia HF Spaces completa na nova aba (header, sidebar do HF, etc). Como abre em nova aba (window.open _blank), nao tem problema com X-Frame-Options. Propaga sid pra auto-login funcionar.""" import os space_id = os.environ.get("SPACE_ID", "").strip() if space_id: base = f"https://huggingface.co/spaces/{space_id}" else: try: host = (st.context.headers.get("host") or "").strip() if st.context else "" except Exception: host = "" if not host: host = os.environ.get("SPACE_HOST") or "localhost:8501" proto = "http" if host.startswith("localhost") or host.startswith("127.0.0.1") else "https" base = f"{proto}://{host}" params: list[str] = [] cur_sid = st.query_params.get("sid", "") if cur_sid: params.append(f"sid={cur_sid}") params.append("view=roadmap-fullscreen") return f"{base}/?{'&'.join(params)}" def _aplicar_filtros( iniciativas: list[dict], status_sel: list[str], cat_sel: list[str], fonte_sel: list[str], busca: str, ) -> list[dict]: out = iniciativas if status_sel: out = [i for i in out if i.get("status") in status_sel] if cat_sel: out = [i for i in out if i.get("categoria") in cat_sel] if fonte_sel: out = [i for i in out if i.get("fonte") in fonte_sel] if busca and busca.strip(): b = busca.lower().strip() out = [i for i in out if b in (i.get("titulo") or "").lower() or b in (i.get("descricao") or "").lower()] return out