# -*- coding: utf-8 -*- import streamlit as st import pandas as pd from io import BytesIO from banco import SessionLocal from models import Equipamento # Auto-refresh from streamlit_autorefresh import st_autorefresh from datetime import datetime, timedelta # SQL util from sqlalchemy import text # ====== Gráficos: Altair (preferência) + fallback Matplotlib ====== ALT_AVAILABLE = True try: import altair as alt try: alt.data_transformers.disable_max_rows() except Exception: pass except Exception: ALT_AVAILABLE = False import matplotlib matplotlib.use("Agg") import matplotlib.pyplot as plt # NumPy para cálculos numéricos robustos import numpy as np # =============================== # Fotos de Responsáveis — Helpers (DB) # =============================== def _ensure_foto_table(db) -> None: """Cria a tabela responsavel_foto se não existir (SQLite/PostgreSQL/MySQL).""" dialect = db.bind.dialect.name if dialect == "sqlite": sql = """ CREATE TABLE IF NOT EXISTS responsavel_foto ( id INTEGER PRIMARY KEY AUTOINCREMENT, tipo TEXT NOT NULL, -- 'especialista' | 'conferente' nome TEXT NOT NULL, imagem BLOB NOT NULL, -- bytes mimetype TEXT, updated_at TIMESTAMP NOT NULL DEFAULT CURRENT_TIMESTAMP, UNIQUE (tipo, nome) ) """ elif dialect in ("postgresql", "postgres"): sql = """ CREATE TABLE IF NOT EXISTS responsavel_foto ( id SERIAL PRIMARY KEY, tipo TEXT NOT NULL, -- 'especialista' | 'conferente' nome TEXT NOT NULL, imagem BYTEA NOT NULL, -- bytes mimetype TEXT, updated_at TIMESTAMPTZ NOT NULL DEFAULT NOW(), UNIQUE (tipo, nome) ) """ else: # mysql/mariadb sql = """ CREATE TABLE IF NOT EXISTS responsavel_foto ( id INT AUTO_INCREMENT PRIMARY KEY, tipo VARCHAR(32) NOT NULL, nome VARCHAR(255) NOT NULL, imagem LONGBLOB NOT NULL, mimetype VARCHAR(64), updated_at TIMESTAMP NOT NULL DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP, UNIQUE KEY uq_tipo_nome (tipo, nome) ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 """ db.execute(text(sql)) db.commit() def _get_foto(db, tipo: str, nome: str): """Retorna (bytes_imagem, mimetype, updated_at) ou (None, None, None).""" if not (tipo and nome): return None, None, None _ensure_foto_table(db) row = db.execute( text( "SELECT imagem, mimetype, updated_at " "FROM responsavel_foto WHERE tipo = :t AND nome = :n LIMIT 1" ), {"t": tipo, "n": nome}, ).fetchone() if row: return row[0], (row[1] or "image/jpeg"), row[2] return None, None, None def _set_foto(db, tipo: str, nome: str, content: bytes, mimetype: str) -> None: """Upsert simples por (tipo, nome).""" if not (tipo and nome and content): return _ensure_foto_table(db) upd = db.execute( text( "UPDATE responsavel_foto " "SET imagem=:img, mimetype=:mt, updated_at=CURRENT_TIMESTAMP " "WHERE tipo=:t AND nome=:n" ), {"img": content, "mt": mimetype or "image/jpeg", "t": tipo, "n": nome}, ) if upd.rowcount == 0: db.execute( text( "INSERT INTO responsavel_foto (tipo, nome, imagem, mimetype) " "VALUES (:t, :n, :img, :mt)" ), {"t": tipo, "n": nome, "img": content, "mt": mimetype or "image/jpeg"}, ) db.commit() def _del_foto(db, tipo: str, nome: str) -> None: if not (tipo and nome): return _ensure_foto_table(db) db.execute(text("DELETE FROM responsavel_foto WHERE tipo=:t AND nome=:n"), {"t": tipo, "n": nome}) db.commit() # =============================== # Estado # =============================== def limpar_estado_prod_esp(): """Remove do session_state qualquer dado do módulo Produtividade_Especialista.""" for key in list(st.session_state.keys()): if key.startswith("prod_esp_"): del st.session_state[key] # =============================== # UI – Gerenciar fotos de responsáveis # =============================== def _ui_fotos_responsaveis(df: pd.DataFrame): """Bloco para cadastrar/atualizar/remover fotos de Especialistas e Conferentes.""" st.subheader("📸 Fotos dos Responsáveis") especialistas = sorted([x for x in df["Especialista"].dropna().astype(str).unique() if x.strip()]) conferentes = sorted([x for x in df["Conferente"].dropna().astype(str).unique() if x.strip()]) tab_esp, tab_conf = st.tabs(["Especialista", "Conferente"]) # ---------- Especialista ---------- with tab_esp: col_e1, col_e2 = st.columns([1, 2]) with col_e1: nome_esp = st.selectbox("Especialista", options=["(selecione)"] + especialistas, index=0, key="prod_esp_foto_esp_sel") file_esp = st.file_uploader( "Carregar foto (PNG/JPG/JPEG/GIF/WEBP) — Especialista", type=["png", "jpg", "jpeg", "gif", "webp"], key="prod_esp_foto_esp_up" ) salvar_esp = st.button("💾 Salvar/Atualizar foto (Especialista)", key="prod_esp_foto_esp_salvar") remover_esp = st.button("🗑️ Remover foto (Especialista)", key="prod_esp_foto_esp_remover") with col_e2: db = SessionLocal() try: if nome_esp and nome_esp != "(selecione)": img_bytes, mt, updt = _get_foto(db, "especialista", nome_esp) if img_bytes: st.caption(f"Foto atual de **{nome_esp}** (atualizada em {updt})") st.image(img_bytes, caption=nome_esp, use_container_width=False, width=220) else: st.info("Nenhuma foto cadastrada para este Especialista.") finally: db.close() if salvar_esp: if not (nome_esp and nome_esp != "(selecione)"): st.warning("Selecione um Especialista.") elif not file_esp: st.warning("Escolha um arquivo de imagem para enviar.") else: content = file_esp.read() mt = file_esp.type or "image/jpeg" db = SessionLocal() try: _set_foto(db, "especialista", nome_esp, content, mt) st.success("Foto salva/atualizada com sucesso!") st.rerun() except Exception as e: db.rollback() st.error(f"Erro ao salvar foto: {e}") finally: db.close() if remover_esp: if not (nome_esp and nome_esp != "(selecione)"): st.warning("Selecione um Especialista.") else: db = SessionLocal() try: _del_foto(db, "especialista", nome_esp) st.info("Foto removida.") st.rerun() except Exception as e: db.rollback() st.error(f"Erro ao remover foto: {e}") finally: db.close() # ---------- Conferente ---------- with tab_conf: col_c1, col_c2 = st.columns([1, 2]) with col_c1: nome_conf = st.selectbox("Conferente", options=["(selecione)"] + conferentes, index=0, key="prod_esp_foto_conf_sel") file_conf = st.file_uploader( "Carregar foto (PNG/JPG/JPEG/GIF/WEBP) — Conferente", type=["png", "jpg", "jpeg", "gif", "webp"], key="prod_esp_foto_conf_up" ) salvar_conf = st.button("💾 Salvar/Atualizar foto (Conferente)", key="prod_esp_foto_conf_salvar") remover_conf = st.button("🗑️ Remover foto (Conferente)", key="prod_esp_foto_conf_remover") with col_c2: db = SessionLocal() try: if nome_conf and nome_conf != "(selecione)": img_bytes, mt, updt = _get_foto(db, "conferente", nome_conf) if img_bytes: st.caption(f"Foto atual de **{nome_conf}** (atualizada em {updt})") st.image(img_bytes, caption=nome_conf, use_container_width=False, width=220) else: st.info("Nenhuma foto cadastrada para este Conferente.") finally: db.close() if salvar_conf: if not (nome_conf and nome_conf != "(selecione)"): st.warning("Selecione um Conferente.") elif not file_conf: st.warning("Escolha um arquivo de imagem para enviar.") else: content = file_conf.read() mt = file_conf.type or "image/jpeg" db = SessionLocal() try: _set_foto(db, "conferente", nome_conf, content, mt) st.success("Foto salva/atualizada com sucesso!") st.rerun() except Exception as e: db.rollback() st.error(f"Erro ao salvar foto: {e}") finally: db.close() if remover_conf: if not (nome_conf and nome_conf != "(selecione)"): st.warning("Selecione um Conferente.") else: db = SessionLocal() try: _del_foto(db, "conferente", nome_conf) st.info("Foto removida.") st.rerun() except Exception as e: db.rollback() st.error(f"Erro ao remover foto: {e}") finally: db.close() # =============================== # Mini-gráfico mensal (% acertos) — Helpers # =============================== def _normalize_responsaveis(df: pd.DataFrame) -> pd.DataFrame: """Normaliza nomes (remove espaços/None) para evitar falhas de comparação.""" for col in ["Especialista", "Conferente"]: df[col] = df[col].astype(str).fillna("").str.strip() df[col] = df[col].replace({"None": ""}) return df def _month_labels_last_n(n: int) -> pd.DataFrame: """Retorna DataFrame com os últimos n meses e rótulos MES/AA, em ordem cronológica.""" base = pd.Timestamp(datetime.now().replace(day=1)) months = [base - pd.DateOffset(months=i) for i in range(n-1, -1, -1)] return pd.DataFrame({ "YM": [pd.Period(m, freq="M") for m in months], "mes": [m.strftime("%b/%y").upper() for m in months] }) def _serie_pct_mensal(df: pd.DataFrame, resp_col: str, nome: str, months: int = 6) -> pd.DataFrame: """ Série mensal (últimos 'months' meses) de % acertos (MROB) para um responsável. Retorna DataFrame com ['mes', 'pct', 'MROB', 'ERROS'] (meses sem dados => 0). Corrige dtype para evitar TypeError: Expected numeric dtype, got object instead. """ if not (nome and resp_col in df.columns and "Data Coleta (dt)" in df.columns): return pd.DataFrame(columns=["mes", "pct", "MROB", "ERROS"]) nome = str(nome).strip() d = df[df[resp_col].astype(str).str.strip() == nome].copy() d = d.dropna(subset=["Data Coleta (dt)"]) # Linha do tempo alvo (sempre haverá N meses) base = _month_labels_last_n(months) # Se não há dados, devolve zeros if d.empty: base["MROB"] = 0.0 base["ERROS"] = 0.0 base["pct"] = 0.0 return base[["mes", "pct", "MROB", "ERROS"]] d["YM"] = d["Data Coleta (dt)"].dt.to_period("M") g = ( d.groupby("YM", as_index=False) .agg(MROB=("Linhas MROB", "sum"), ERROS=("Linhas Erros MROB", "sum")) ) # Merge garante a linha do tempo completa — aqui o dtype pode virar 'object' m = base.merge(g, on="YM", how="left") # Coerção numérica robusta pós-merge (evita object -> round error) m["MROB"] = pd.to_numeric(m["MROB"], errors="coerce").fillna(0).astype("float64") m["ERROS"] = pd.to_numeric(m["ERROS"], errors="coerce").fillna(0).astype("float64") # % acertos (evita divisão por zero, resultado sempre float) m["pct"] = np.where( m["MROB"] > 0, ((m["MROB"] - m["ERROS"]) / m["MROB"]) * 100.0, 0.0 ) m["pct"] = pd.to_numeric(m["pct"], errors="coerce").fillna(0).astype("float64").round(2) # Seleciona e ordena colunas finais out = m[["mes", "pct", "MROB", "ERROS"]].copy() # Garantia de dtype correto (evita regressões futuras) out["MROB"] = out["MROB"].astype("float64") out["ERROS"] = out["ERROS"].astype("float64") out["pct"] = out["pct"].astype("float64") return out def _mini_grafico_pct_mensal(df_m: pd.DataFrame, meta: float, chart_type: str = "Linha", show_meta: bool = True, titulo: str = "% Acertos por mês"): """ Renderiza mini‑gráfico compacto (% acertos | 0–100) com fallback: 1) Altair (linha/barras + meta) 2) Matplotlib 3) Tabela """ if df_m.empty: st.caption("Sem dados mensais para o período/seleção atual.") return # 1) ALTair if ALT_AVAILABLE: try: base = alt.Chart(df_m).encode( x=alt.X("mes:N", title="Mês"), y=alt.Y("pct:Q", title="% Acertos", scale=alt.Scale(domain=[0, 100])), tooltip=[ alt.Tooltip("mes:N", title="Mês"), alt.Tooltip("pct:Q", title="% Acertos (%)"), alt.Tooltip("MROB:Q", title="MROB (Σ)"), alt.Tooltip("ERROS:Q", title="Erros MROB (Σ)") ] ) chart = base.mark_line(point=True, interpolate="monotone", color="#0d6efd") if chart_type == "Linha" \ else base.mark_bar(size=18, color="#0d6efd") final = chart.properties(width=260, height=150, title=titulo) if show_meta: meta_df = pd.DataFrame({"y": [meta]}) meta_rule = alt.Chart(meta_df).mark_rule(color="#16a34a", strokeDash=[6, 4]).encode(y="y:Q") final = final + meta_rule st.altair_chart(final, use_container_width=False) return except Exception as e: st.info(f"Render ALTair indisponível, usando fallback (detalhe: {e})") # 2) Matplotlib fallback try: fig, ax = plt.subplots(figsize=(3.2, 1.6), dpi=150) x = list(range(len(df_m["mes"]))) if chart_type == "Linha": ax.plot(x, df_m["pct"].values, marker="o", color="#0d6efd", linewidth=1.5) else: ax.bar(x, df_m["pct"].values, color="#0d6efd", width=0.6) if show_meta: ax.axhline(y=meta, color="#16a34a", linestyle="--", linewidth=1) ax.set_ylim(0, 100) ax.set_xticks(x) ax.set_xticklabels(df_m["mes"].tolist(), rotation=0, fontsize=7) ax.set_yticks([0, 20, 40, 60, 80, 100]) ax.set_title(titulo, fontsize=9) ax.grid(alpha=0.15, axis="y") plt.tight_layout() st.pyplot(fig, use_container_width=False) plt.close(fig) return except Exception as e: st.warning(f"Não foi possível renderizar o mini‑gráfico (fallback MPL): {e}") # 3) Último recurso st.caption("Exibindo dados da série por impossibilidade de gráfico:") st.dataframe(df_m, use_container_width=True) # =============================== # MAIN # =============================== def main(): # 🧹 LIMPA ESTADO AO ENTRAR if not st.session_state.get("_prod_esp_inicializado"): limpar_estado_prod_esp() st.session_state["_prod_esp_inicializado"] = True st.title("🏆 Produtividade por Especialista e Conferente") # 🔧 CONTROLES NA SIDEBAR with st.sidebar: st.markdown("### 🔄 Atualização automática") auto_on = st.checkbox("Ativar atualização automática", value=True, key="prod_esp_auto_on") auto_interval_s = st.slider("Intervalo (segundos)", min_value=10, max_value=300, value=30, step=5, key="prod_esp_auto_int") if "prod_esp_auto_int_effective" not in st.session_state: st.session_state["prod_esp_auto_int_effective"] = auto_interval_s if st.button("✅ Aplicar intervalo"): st.session_state["prod_esp_auto_int_effective"] = auto_interval_s st.success(f"Intervalo atualizado para {auto_interval_s}s") st.rerun() intervalo_efetivo = st.session_state.get("prod_esp_auto_int_effective", auto_interval_s) st.caption(f"⏲️ Intervalo atual: **{intervalo_efetivo}s**") st.markdown("---") st.markdown("### 🎯 Metas e Série") meta_pct_especialistas = st.number_input("Meta (% MROB/Geral) — Especialistas", min_value=0.0, max_value=100.0, value=98.8, step=0.5, key="prod_esp_meta_pct_esp") meta_pct_conferentes = st.number_input("Meta (% MROB/Geral) — Conferentes", min_value=0.0, max_value=100.0, value=98.8, step=0.5, key="prod_esp_meta_pct_conf") serie_meses = st.slider("Meses no mini‑gráfico", min_value=3, max_value=12, value=6, step=1, key="prod_esp_serie_meses") tipo_grafico = st.selectbox("Tipo do mini‑gráfico", ["Linha", "Barras"], index=0, key="prod_esp_tipo_grafico") linha_meta = st.checkbox("Mostrar linha de meta", value=True, key="prod_esp_show_meta") st.markdown("---") last_dt = st.session_state.get("prod_esp_last_update_dt") if last_dt: last_str = last_dt.strftime("%d/%m/%Y %H:%M:%S") st.caption(f"🕒 Última atualização: **{last_str}**") delta = datetime.now() - last_dt if delta < timedelta(minutes=1): ago_str = f"{delta.seconds}s" elif delta < timedelta(hours=1): mins = delta.seconds // 60 secs = delta.seconds % 60 ago_str = f"{mins}min {secs}s" else: hours = delta.seconds // 3600 mins = (delta.seconds % 3600) // 60 ago_str = f"{hours}h {mins}min" st.caption(f"⏱️ Atualizado há **{ago_str}**") if auto_on: try: nxt = (datetime.now() + timedelta(seconds=intervalo_efetivo)).strftime("%d/%m/%Y %H:%M:%S") st.caption(f"🔁 Próximo refresh: **{nxt}**") except Exception: pass else: st.caption("🕒 Última atualização: **—**") if auto_on: st_autorefresh(interval=intervalo_efetivo * 1000, limit=None, key="prod_esp_autorefresh") db = SessionLocal() try: registros = db.query(Equipamento).all() st.session_state["prod_esp_last_update_dt"] = datetime.now() if not registros: st.info("Nenhum registro encontrado.") return # ========== BASE DF ========== df = pd.DataFrame([{ "FPSO": getattr(r, "fpso", None), "Data Coleta": getattr(r, "data_coleta", None), "Modal": getattr(r, "modal", None), "Especialista": getattr(r, "especialista", None), "Conferente": getattr(r, "conferente", None), "Linhas OSM": getattr(r, "linhas_osm", 0), "Linhas MROB": getattr(r, "linhas_mrob", 0), "Linhas Erros MROB": getattr(r, "linhas_erros_mrob", None), "Linhas Erros (Genérico)": getattr(r, "linhas_erros", None), } for r in registros]) # Conversão robusta de datas df["Data Coleta (dt)"] = pd.to_datetime(df["Data Coleta"], errors="coerce", dayfirst=True) if df["Data Coleta (dt)"].isna().all(): # tenta novamente sem dayfirst df["Data Coleta (dt)"] = pd.to_datetime(df["Data Coleta"], errors="coerce", dayfirst=False) # Tipos numéricos for col in ["Linhas OSM", "Linhas MROB", "Linhas Erros MROB", "Linhas Erros (Genérico)"]: if col in df.columns: df[col] = pd.to_numeric(df[col], errors="coerce").fillna(0).astype("int64") # Fallback de erros MROB if ("Linhas Erros MROB" not in df.columns) or (df["Linhas Erros MROB"].sum() == 0 and df["Linhas Erros (Genérico)"].sum() > 0): df["Linhas Erros MROB"] = df.get("Linhas Erros (Genérico)", pd.Series([0] * len(df))) # Normaliza nomes df = _normalize_responsaveis(df) # ======== Fotos (cadastro/visualização) ======== _ui_fotos_responsaveis(df) # ========== FILTROS ========== st.subheader("🔎 Filtros") col1, col2, col3 = st.columns(3) with col1: filtro_fpso = st.multiselect("FPSO", sorted(df["FPSO"].dropna().unique()), key="prod_esp_fpso") with col2: filtro_modal = st.multiselect("Modal", sorted(df["Modal"].dropna().unique()), key="prod_esp_modal") with col3: periodo = st.date_input("Período de Coleta", value=None, key="prod_esp_periodo") df_filt = df.copy() if filtro_fpso: df_filt = df_filt[df_filt["FPSO"].isin(filtro_fpso)] if filtro_modal: df_filt = df_filt[df_filt["Modal"].isin(filtro_modal)] if isinstance(periodo, (list, tuple)) and len(periodo) == 2: data_inicio, data_fim = periodo if pd.notna(data_inicio): df_filt = df_filt[df_filt["Data Coleta (dt)"] >= pd.to_datetime(data_inicio)] if pd.notna(data_fim): df_filt = df_filt[df_filt["Data Coleta (dt)"] <= pd.to_datetime(data_fim) + pd.Timedelta(days=1) - pd.Timedelta(seconds=1)] # ======== Mapeamentos por responsável ======== fpsos_por_especialista = ( df_filt.groupby("Especialista", dropna=False)["FPSO"] .apply(lambda x: ", ".join(sorted(set(x.dropna())))) .to_dict() ) fpsos_por_conferente = ( df_filt.groupby("Conferente", dropna=False)["FPSO"] .apply(lambda x: ", ".join(sorted(set(x.dropna())))) .to_dict() ) # ======== Agregações ======== grp_esp = (df_filt.groupby("Especialista", dropna=False) .agg({"Linhas OSM":"sum","Linhas MROB":"sum","Linhas Erros MROB":"sum"}) .reset_index()) grp_esp["FPSO Responsável"] = grp_esp["Especialista"].map(lambda e: fpsos_por_especialista.get(e, "")) grp_esp["Especialista (FPSO)"] = grp_esp.apply( lambda r: f"{r['Especialista']} ({r['FPSO Responsável']})" if r["FPSO Responsável"] else str(r["Especialista"]), axis=1) grp_esp["Total de Erros (MROB - Erros MROB)"] = (grp_esp["Linhas MROB"] - grp_esp["Linhas Erros MROB"]).clip(lower=0) # ✅ Denominador numérico (float) para evitar dtype object denom_mrob_esp = pd.to_numeric(grp_esp["Linhas MROB"], errors="coerce").replace(0, np.nan).astype("float64") num_acertos_esp = pd.to_numeric(grp_esp["Linhas MROB"], errors="coerce") - pd.to_numeric(grp_esp["Linhas Erros MROB"], errors="coerce") num_erros_esp = pd.to_numeric(grp_esp["Linhas Erros MROB"], errors="coerce") grp_esp["% Acertos (MROB)"] = (num_acertos_esp / denom_mrob_esp * 100.0).round(2) grp_esp["% Erros (MROB)"] = (num_erros_esp / denom_mrob_esp * 100.0).round(2) grp_esp = grp_esp.sort_values(by="Linhas OSM", ascending=False) grp_esp = grp_esp[[ "Especialista (FPSO)","Especialista","FPSO Responsável", "Linhas OSM","Linhas MROB","Linhas Erros MROB", "Total de Erros (MROB - Erros MROB)","% Acertos (MROB)","% Erros (MROB)" ]] grp_conf = (df_filt.groupby("Conferente", dropna=False) .agg({"Linhas OSM":"sum","Linhas MROB":"sum","Linhas Erros MROB":"sum"}) .reset_index()) grp_conf["FPSO Responsável"] = grp_conf["Conferente"].map(lambda c: fpsos_por_conferente.get(c, "")) grp_conf["Conferente (FPSO)"] = grp_conf.apply( lambda r: f"{r['Conferente']} ({r['FPSO Responsável']})" if r["FPSO Responsável"] else str(r["Conferente"]), axis=1) grp_conf["Total de Erros (MROB - Erros MROB)"] = (grp_conf["Linhas MROB"] - grp_conf["Linhas Erros MROB"]).clip(lower=0) # ✅ Denominador numérico (float) para evitar dtype object denom_mrob_conf = pd.to_numeric(grp_conf["Linhas MROB"], errors="coerce").replace(0, np.nan).astype("float64") num_acertos_conf = pd.to_numeric(grp_conf["Linhas MROB"], errors="coerce") - pd.to_numeric(grp_conf["Linhas Erros MROB"], errors="coerce") num_erros_conf = pd.to_numeric(grp_conf["Linhas Erros MROB"], errors="coerce") grp_conf["% Acertos (MROB)"] = (num_acertos_conf / denom_mrob_conf * 100.0).round(2) grp_conf["% Erros (MROB)"] = (num_erros_conf / denom_mrob_conf * 100.0).round(2) grp_conf = grp_conf.sort_values(by="Linhas OSM", ascending=False) grp_conf = grp_conf[[ "Conferente (FPSO)","Conferente","FPSO Responsável", "Linhas OSM","Linhas MROB","Linhas Erros MROB", "Total de Erros (MROB - Erros MROB)","% Acertos (MROB)","% Erros (MROB)" ]] # ======== KPIs Gerais ======== st.subheader("📈 KPIs (dados filtrados) — Geral (Todos)") total_especialistas = grp_esp["Especialista"].nunique() total_conferentes = grp_conf["Conferente"].nunique() total_osm_geral = int(df_filt["Linhas OSM"].sum()) total_mrob_geral = int(df_filt["Linhas MROB"].sum()) total_erros_mrob_geral = int(df_filt["Linhas Erros MROB"].sum()) total_acertos_mrob_geral = (total_mrob_geral - total_erros_mrob_geral) pct_acertos_geral = round((total_acertos_mrob_geral / total_mrob_geral * 100), 2) if total_mrob_geral > 0 else 0.0 pct_erros_geral = round((total_erros_mrob_geral / total_mrob_geral * 100), 2) if total_mrob_geral > 0 else 0.0 k1,k2,k3,k4,k5 = st.columns(5) k1.metric("Especialistas", f"{total_especialistas}") k2.metric("Conferentes", f"{total_conferentes}") k3.metric("Linhas OSM (Σ)", f"{total_osm_geral:,}".replace(",", ".")) k4.metric("Linhas MROB (Σ)", f"{total_mrob_geral:,}".replace(",", ".")) color_geral = "#198754" if pct_acertos_geral >= meta_pct_especialistas else "#dc3545" k5.metric("% Acertos (MROB/Geral)", f"{pct_acertos_geral}%") # 🔧 HTML deve usar ..., não entidades <> st.markdown( f"Meta (Especialistas): {meta_pct_especialistas}% • " f"{'✅ Dentro da meta' if pct_acertos_geral >= meta_pct_especialistas else '⚠️ Abaixo da meta'}", unsafe_allow_html=True ) st.markdown(f"% Erros (MROB/Geral): {pct_erros_geral}%", unsafe_allow_html=True) st.divider() # ======== KPIs por Especialista (foto + mini‑gráfico) ======== st.subheader("🎯 KPIs por Especialista") especialistas_lista = ["(selecione)"] + list(grp_esp["Especialista"].astype(str).unique()) esp_sel = st.selectbox("Especialista:", especialistas_lista, index=0, key="prod_esp_kpi_esp") if esp_sel and esp_sel != "(selecione)": linha_esp = grp_esp[grp_esp["Especialista"] == esp_sel] if not linha_esp.empty: le_osm = int(linha_esp["Linhas OSM"].iloc[0]) le_mrob = int(linha_esp["Linhas MROB"].iloc[0]) le_err_mrob = int(linha_esp["Linhas Erros MROB"].iloc[0]) le_total_err = int(linha_esp["Total de Erros (MROB - Erros MROB)"].iloc[0]) le_pct_acertos = float(linha_esp["% Acertos (MROB)"].iloc[0]) if pd.notna(linha_esp["% Acertos (MROB)"].iloc[0]) else 0.0 le_pct_erros = float(linha_esp["% Erros (MROB)"].iloc[0]) if pd.notna(linha_esp["% Erros (MROB)"].iloc[0]) else 0.0 col_pic, col_chart, col_metrics = st.columns([1, 1.4, 3]) with col_pic: dbp = SessionLocal() try: img_b, mt, updt = _get_foto(dbp, "especialista", esp_sel) if img_b: st.image(img_b, caption=f"{esp_sel}", use_container_width=False, width=220) else: st.caption("Sem foto cadastrada.") finally: dbp.close() with col_chart: serie = _serie_pct_mensal(df_filt, "Especialista", esp_sel, months=serie_meses) _mini_grafico_pct_mensal(serie, meta=meta_pct_especialistas, chart_type=tipo_grafico, show_meta=linha_meta) with st.expander("🔧 Diagnóstico da série (Especialista)", expanded=False): st.dataframe(serie, use_container_width=True) with col_metrics: s1,s2,s3,s4,s5 = st.columns(5) s1.metric("Linhas OSM", f"{le_osm:,}".replace(",", ".")) s2.metric("Linhas MROB", f"{le_mrob:,}".replace(",", ".")) s3.metric("Erros MROB", f"{le_err_mrob:,}".replace(",", ".")) s4.metric("Total Erros (MROB−Erros)", f"{le_total_err:,}".replace(",", ".")) s5.metric("% Acertos (MROB)", f"{le_pct_acertos}%") # 🔧 HTML deve usar ..., não entidades <> st.markdown(f"% Erros (MROB): {le_pct_erros}%", unsafe_allow_html=True) st.divider() # ======== KPIs por Conferente (foto + mini‑gráfico) ======== st.subheader("🎯 KPIs por Conferente") conferentes_lista = ["(selecione)"] + list(grp_conf["Conferente"].astype(str).unique()) conf_sel = st.selectbox("Conferente:", conferentes_lista, index=0, key="prod_esp_kpi_conf") if conf_sel and conf_sel != "(selecione)": linha_conf = grp_conf[grp_conf["Conferente"] == conf_sel] if not linha_conf.empty: lc_osm = int(linha_conf["Linhas OSM"].iloc[0]) lc_mrob = int(linha_conf["Linhas MROB"].iloc[0]) lc_err_mrob = int(linha_conf["Linhas Erros MROB"].iloc[0]) lc_total_err = int(linha_conf["Total de Erros (MROB - Erros MROB)"].iloc[0]) lc_pct_acertos = float(linha_conf["% Acertos (MROB)"].iloc[0]) if pd.notna(linha_conf["% Acertos (MROB)"].iloc[0]) else 0.0 lc_pct_erros = float(linha_conf["% Erros (MROB)"].iloc[0]) if pd.notna(linha_conf["% Erros (MROB)"].iloc[0]) else 0.0 col_pic2, col_chart2, col_metrics2 = st.columns([1, 1.4, 3]) with col_pic2: dbp = SessionLocal() try: img_b, mt, updt = _get_foto(dbp, "conferente", conf_sel) if img_b: st.image(img_b, caption=f"{conf_sel}", use_container_width=False, width=220) else: st.caption("Sem foto cadastrada.") finally: dbp.close() with col_chart2: serie2 = _serie_pct_mensal(df_filt, "Conferente", conf_sel, months=serie_meses) _mini_grafico_pct_mensal(serie2, meta=meta_pct_conferentes, chart_type=tipo_grafico, show_meta=linha_meta) with st.expander("🔧 Diagnóstico da série (Conferente)", expanded=False): st.dataframe(serie2, use_container_width=True) with col_metrics2: d1,d2,d3,d4,d5 = st.columns(5) d1.metric("Linhas OSM", f"{lc_osm:,}".replace(",", ".")) d2.metric("Linhas MROB", f"{lc_mrob:,}".replace(",", ".")) d3.metric("Erros MROB", f"{lc_err_mrob:,}".replace(",", ".")) d4.metric("Total Erros (MROB−Erros)", f"{lc_total_err:,}".replace(",", ".")) d5.metric("% Acertos (MROB)", f"{lc_pct_acertos}%") # 🔧 HTML deve usar ..., não entidades <> st.markdown(f"% Erros (MROB): {lc_pct_erros}%", unsafe_allow_html=True) st.divider() # ======== Listas e Gráficos maiores ======== st.subheader("🧾 Lista por Especialista (com métricas)") st.dataframe(grp_esp, use_container_width=True) st.subheader("🧾 Lista por Conferente (com métricas)") st.dataframe(grp_conf, use_container_width=True) st.subheader("📊 Gráficos") try: st.caption("Linhas OSM por Especialista (FPSO)") st.bar_chart(data=grp_esp.set_index("Especialista (FPSO)")["Linhas OSM"]) st.caption("Linhas MROB por Especialista (FPSO)") st.bar_chart(data=grp_esp.set_index("Especialista (FPSO)")["Linhas MROB"]) st.caption("Linhas de Erros MROB por Especialista (FPSO)") st.bar_chart(data=grp_esp.set_index("Especialista (FPSO)")["Linhas Erros MROB"]) st.caption("Linhas OSM por Conferente (FPSO)") st.bar_chart(data=grp_conf.set_index("Conferente (FPSO)")["Linhas OSM"]) st.caption("Linhas MROB por Conferente (FPSO)") st.bar_chart(data=grp_conf.set_index("Conferente (FPSO)")["Linhas MROB"]) st.caption("Linhas de Erros MROB por Conferente (FPSO)") st.bar_chart(data=grp_conf.set_index("Conferente (FPSO)")["Linhas Erros MROB"]) except Exception as e: st.warning(f"Não foi possível renderizar alguns gráficos: {e}") st.divider() # ======== Exportação ======== st.subheader("⬇️ Exportar") buffer_esp = BytesIO() with pd.ExcelWriter(buffer_esp, engine="openpyxl") as writer: grp_esp.to_excel(writer, index=False, sheet_name="Prod_Especialista") buffer_esp.seek(0) st.download_button( label="⬇️ Exportar produtividade por Especialista (Excel)", data=buffer_esp, file_name="produtividade_especialista.xlsx", mime="application/vnd.openxmlformats-officedocument.spreadsheetml.sheet", key="prod_esp_export" ) buffer_conf = BytesIO() with pd.ExcelWriter(buffer_conf, engine="openpyxl") as writer: grp_conf.to_excel(writer, index=False, sheet_name="Prod_Conferente") buffer_conf.seek(0) st.download_button( label="⬇️ Exportar produtividade por Conferente (Excel)", data=buffer_conf, file_name="produtividade_conferente.xlsx", mime="application/vnd.openxmlformats-officedocument.spreadsheetml.sheet", key="prod_conf_export" ) finally: db.close()