Update importar_excel.py

importar_excel.py

@@ -1,301 +1,433 @@
-
-import streamlit as st
-import pandas as pd
-from io import BytesIO
-from datetime import datetime, date
-
-from banco import SessionLocal
-from models import Equipamento
-from utils_auditoria import registrar_log
-
-
-# =====================================================
-# FUNÇÕES AUXILIARES
-# =====================================================
-def to_date(value):
-    """
-    Converte pandas.Timestamp ou datetime para datetime.date
-    Necessário para compatibilidade com SQLite
-    """
-    if value is None or pd.isna(value):
-        return None
-    if isinstance(value, pd.Timestamp):
-        return value.date()
-    if isinstance(value, datetime):
-        return value.date()
-    if isinstance(value, date):
-        return value
-    return None
-
-
-def safe_value(value):
-    """
-    Retorna 0 se o valor for vazio/NaN, senão retorna o próprio valor.
-    Usado para campos obrigatórios.
-    """
-    if value is None or pd.isna(value):
-        return 0
-    return value
-
-
-# =====================================================
-# MÓDULO PRINCIPAL
-# =====================================================
-def main():
-    st.title("📥 Importação de Dados via Excel")
-
-    st.markdown(
-        """
-        Este módulo permite:
-        - 📄 Baixar um **modelo Excel padrão**
-        - ✍️ Preencher os dados offline
-        - 🔍 Validar antes da gravação
-        - 💾 Importar os registros para o banco
-        """
-    )
-
-    # =====================================================
-    # 1️⃣ GERAR MODELO EXCEL
-    # =====================================================
-    st.subheader("📄 Baixar modelo Excel")
-
-    colunas = [
-        "fpso1", "fpso", "data_coleta", "especialista", "conferente", "osm", "modal",
-        "quant_equip", "mrob", "linhas_osm", "linhas_mrob", "linhas_erros",
-        "erro_storekeeper", "erro_operacao", "erro_especialista", "erro_outros",
-        "inclusao_exclusao", "po", "part_number", "material", "solicitante", "motivo",
-        "requisitante", "nota_fiscal", "impacto", "dimensao", "observacoes", "dia_inclusao",
-    ]
-
-    modelo_df = pd.DataFrame(columns=colunas)
-
-    buffer = BytesIO()
-    with pd.ExcelWriter(buffer, engine="openpyxl") as writer:
-        modelo_df.to_excel(writer, index=False, sheet_name="MODELO")
-
-    buffer.seek(0)
-
-    st.download_button(
-        label="⬇️ Baixar modelo Excel",
-        data=buffer,
-        file_name="modelo_importacao_load.xlsx",
-        mime="application/vnd.openxmlformats-officedocument.spreadsheetml.sheet",
-    )
-
-    st.divider()
-
-    # =====================================================
-    # 2️⃣ UPLOAD DO ARQUIVO
-    # =====================================================
-    st.subheader("📤 Importar arquivo preenchido")
-
-    arquivo = st.file_uploader(
-        "Selecione o arquivo Excel",
-        type=["xlsx"]
-    )
-
-    if not arquivo:
-        st.info("📌 Faça o upload de um arquivo para continuar.")
-        return
-
-    try:
-        df = pd.read_excel(arquivo)
-    except Exception as e:
-        st.error(f"❌ Erro ao ler o arquivo: {e}")
-        return
-
-    # Garante um identificador estável para cada linha durante as edições
-    if "_row_id" not in df.columns:
-        df["_row_id"] = range(len(df))
-
-    # Salva o DF bruto na sessão para persistência entre reruns
-    st.session_state["df_raw"] = df.copy()
-
-    st.success("✅ Arquivo carregado com sucesso!")
-
-    # =====================================================
-    # 3️⃣ PRÉVIA DOS DADOS
-    # =====================================================
-    st.subheader("🔍 Prévia dos dados")
-    st.dataframe(df, use_container_width=True)
-
-    # =====================================================
-    # 4️⃣ VERIFICAÇÃO DE DUPLICIDADE (com seleção de linhas a excluir)
-    # =====================================================
-    st.subheader("🧪 Verificação de duplicidade")
-
-    st.caption("Escolha as colunas que definem a duplicidade. Em seguida, marque o checkbox **_excluir** nas linhas que **não** deseja importar.")
-
-    # Sugerimos um conjunto padrão de chaves, mas só usamos as que existem no arquivo
-    sugestao_chaves = [c for c in ["fpso", "osm", "po", "part_number", "nota_fiscal", "data_coleta"] if c in df.columns]
-
-    chaves = st.multiselect(
-        "📌 Colunas para verificação de duplicidade:",
-        options=list(df.columns),
-        default=sugestao_chaves if len(sugestao_chaves) > 0 else []
-    )
-
-    # Prepara um DF de trabalho com colunas auxiliares
-    work_df = df.copy()
-    work_df["_duplicado"] = False
-    work_df["_excluir"] = False  # será editado pelo usuário
-
-    if len(chaves) == 0:
-        st.info("Selecione ao menos **uma** coluna para verificar duplicidade.")
-        # Mesmo sem duplicidade definida, permitimos marcar exclusões manuais, se quiser:
-        with st.expander("🔧 (Opcional) Excluir linhas manualmente mesmo sem duplicidade"):
-            manual_view = work_df.set_index("_row_id")[
-                [c for c in work_df.columns if c not in ["_duplicado"]] + ["_excluir"]
-            ]
-            edited_manual = st.data_editor(
-                manual_view,
-                use_container_width=True,
-                num_rows="fixed",
-                column_config={
-                    "_excluir": st.column_config.CheckboxColumn("Excluir da importação")
-                }
-            )
-            # Aplica exclusões manuais
-            work_df = work_df.set_index("_row_id")
-            work_df["_excluir"] = edited_manual["_excluir"].reindex(work_df.index).fillna(False).astype(bool)
-            work_df = work_df.reset_index()
-
-    else:
-        # Marca duplicadas com base nas chaves selecionadas
-        mask_dup_any = work_df.duplicated(subset=chaves, keep=False)
-        work_df["_duplicado"] = mask_dup_any
-
-        # Sugerimos exclusão automática das ocorrências não-primárias (o usuário pode alterar)
-        mask_dup_not_first = work_df.duplicated(subset=chaves, keep="first")
-        work_df.loc[mask_dup_not_first, "_excluir"] = True
-
-        if mask_dup_any.any():
-            st.warning("⚠️ Foram encontradas linhas duplicadas com base nas chaves selecionadas:")
-            # Mostrar somente as duplicadas para facilitar a decisão
-            cols_para_mostrar = chaves + [c for c in ["_duplicado", "_excluir"] if c not in chaves]
-            # Evita colunas repetidas mantendo ordem
-            seen = set()
-            cols_para_mostrar = [c for c in cols_para_mostrar if not (c in seen or seen.add(c))]
-
-            dup_view = work_df.loc[mask_dup_any].set_index("_row_id")[cols_para_mostrar]
-            edited_dup = st.data_editor(
-                dup_view,
-                use_container_width=True,
-                num_rows="fixed",
-                column_config={
-                    "_excluir": st.column_config.CheckboxColumn("Excluir da importação"),
-                    "_duplicado": st.column_config.CheckboxColumn("Duplicado", disabled=True)
-                }
-            )
-
-            # Mescla de volta as escolhas do usuário
-            work_df = work_df.set_index("_row_id")
-            if "_excluir" in edited_dup.columns:
-                work_df.loc[edited_dup.index, "_excluir"] = (
-                    edited_dup["_excluir"].reindex(work_df.index).fillna(work_df["_excluir"]).astype(bool)
-                )
-            work_df = work_df.reset_index()
-
-            st.info(
-                f"📊 Totais — Linhas: {len(work_df)} | Duplicadas: {mask_dup_any.sum()} | "
-                f"Marcadas para excluir: {int(work_df['_excluir'].sum())}"
-            )
-        else:
-            st.success("✅ Nenhuma duplicidade encontrada com as chaves selecionadas.")
-
-    st.divider()
-
-    # =====================================================
-    # 4.1️⃣ Prévia final do que será importado + download
-    # =====================================================
-    df_para_importar = work_df[~work_df["_excluir"]].drop(columns=["_duplicado", "_excluir"], errors="ignore")
-    st.session_state["df_para_importar"] = df_para_importar.copy()
-
-    st.subheader("🧾 Prévia do que será importado")
-    st.caption("A prévia abaixo desconsidera as linhas marcadas como **_excluir**.")
-    st.dataframe(df_para_importar.drop(columns=["_row_id"], errors="ignore"), use_container_width=True)
-
-    # Download da prévia para conferência
-    buf_prev = BytesIO()
-    with pd.ExcelWriter(buf_prev, engine="openpyxl") as writer:
-        df_para_importar.drop(columns=["_row_id"], errors="ignore").to_excel(writer, index=False, sheet_name="A_IMPORTAR")
-    buf_prev.seek(0)
-    st.download_button(
-        "⬇️ Baixar prévia (Excel)",
-        data=buf_prev,
-        file_name="previa_a_importar.xlsx",
-        mime="application/vnd.openxmlformats-officedocument.spreadsheetml.sheet",
-        help="Baixe a prévia do conjunto que será gravado."
-    )
-
-    # =====================================================
-    # 5️⃣ GRAVAÇÃO NO BANCO (usa o DataFrame filtrado)
-    # =====================================================
-    st.subheader("💾 Gravar dados no banco")
-
-    col1, col2 = st.columns(2)
-
-    if col1.button("💾 Salvar registros importados"):
-        df_import = st.session_state.get("df_para_importar", df)
-
-        if df_import.empty:
-            st.error("Não há registros para importar. Revise as exclusões.")
-            return
-
-        with SessionLocal() as db:
-            try:
-                for _, row in df_import.iterrows():
-                    registro = Equipamento(
-                        fpso1=safe_value(row.get("fpso1")),
-                        fpso=safe_value(row.get("fpso")),
-                        data_coleta=to_date(row.get("data_coleta")),
-                        especialista=safe_value(row.get("especialista")),
-                        conferente=safe_value(row.get("conferente")),
-                        osm=safe_value(row.get("osm")),
-                        modal=safe_value(row.get("modal")),
-                        quant_equip=int(row["quant_equip"]) if not pd.isna(row.get("quant_equip")) else 0,
-                        mrob=safe_value(row.get("mrob")),
-                        linhas_osm=int(row["linhas_osm"]) if not pd.isna(row.get("linhas_osm")) else 0,
-                        linhas_mrob=int(row["linhas_mrob"]) if not pd.isna(row.get("linhas_mrob")) else 0,
-                        linhas_erros=int(row["linhas_erros"]) if not pd.isna(row.get("linhas_erros")) else 0,
-                        erro_storekeeper=safe_value(row.get("erro_storekeeper")),
-                        erro_operacao=safe_value(row.get("erro_operacao")),
-                        erro_especialista=safe_value(row.get("erro_especialista")),
-                        erro_outros=safe_value(row.get("erro_outros")),
-                        inclusao_exclusao=safe_value(row.get("inclusao_exclusao")),
-                        po=safe_value(row.get("po")),
-                        part_number=safe_value(row.get("part_number")),
-                        material=safe_value(row.get("material")),
-                        solicitante=safe_value(row.get("solicitante")),
-                        motivo=safe_value(row.get("motivo")),
-                        requisitante=safe_value(row.get("requisitante")),
-                        nota_fiscal=safe_value(row.get("nota_fiscal")),
-                        impacto=safe_value(row.get("impacto")),
-                        dimensao=safe_value(row.get("dimensao")),
-                        observacoes=safe_value(row.get("observacoes")),
-                        dia_inclusao=safe_value(row.get("dia_inclusao")),
-                    )
-                    db.add(registro)
-
-                db.commit()
-
-                registrar_log(
-                    usuario=st.session_state.get("usuario"),
-                    acao=f"IMPORTAÇÃO EXCEL ({len(df_import)} registros) - com filtro de duplicidade",
-                    tabela="equipamentos",
-                    registro_id=None
-                )
-
-                st.success(f"🎉 Importação concluída com sucesso! {len(df_import)} registros gravados.")
-
-            except Exception as e:
-                db.rollback()
-                st.error(f"❌ Erro ao gravar no banco: {e}")
-
-    if col2.button("❌ Cancelar importação"):
-        st.warning("Importação cancelada pelo usuário.")
-
-
-if __name__ == "__main__":
-    main()
-
+# -*- coding: utf-8 -*-
+"""
+importar_excel.py
+Importa planilhas Excel para a tabela Equipamento, com:
+- Geração de modelo Excel
+- Upload e pré-visualização
+- Verificação/edição de duplicidades
+- Conversões seguras de tipos (datas/números/strings)
+- Gravação transacional + auditoria resiliente
+Compatível com Linux (Spaces) e local/Windows.
+"""
+
+import streamlit as st
+import pandas as pd
+from io import BytesIO
+from datetime import datetime, date
+
+from banco import SessionLocal
+from models import Equipamento
+from utils_auditoria import registrar_log
+
+# =========================================
+# Configuração – nomes de colunas esperadas
+# =========================================
+COLUNAS_ESPERADAS = [
+    "fpso1", "fpso", "data_coleta", "especialista", "conferente", "osm", "modal",
+    "quant_equip", "mrob", "linhas_osm", "linhas_mrob", "linhas_erros",
+    "erro_storekeeper", "erro_operacao", "erro_especialista", "erro_outros",
+    "inclusao_exclusao", "po", "part_number", "material", "solicitante", "motivo",
+    "requisitante", "nota_fiscal", "impacto", "dimensao", "observacoes", "dia_inclusao",
+]
+
+# Colunas que tratamos como números inteiros na gravação
+COLS_INTEIRAS = {
+    "quant_equip", "linhas_osm", "linhas_mrob", "linhas_erros"
+}
+
+# =====================================================
+# FUNÇÕES AUXILIARES
+# =====================================================
+def _normalize_cols_case_insensitive(df: pd.DataFrame) -> pd.DataFrame:
+    """
+    Mapeia colunas do arquivo (case/espaços) para os nomes esperados em COLUNAS_ESPERADAS.
+    Ex.: " FPSo " -> "fpso"; "DATA_COLETA" -> "data_coleta".
+    Colunas não reconhecidas permanecem como estão.
+    """
+    if df is None or df.empty:
+        return df
+
+    # prioriza mapeamento por lower(strip)
+    existing = list(df.columns)
+    lower_map = {c.strip().lower(): c for c in existing}
+    new_names = {}
+    for wanted in COLUNAS_ESPERADAS:
+        key = wanted.lower()
+        if key in lower_map:
+            new_names[lower_map[key]] = wanted  # renomeia p/ o nome "oficial" esperado
+
+    # aplica renomeação
+    try:
+        df = df.rename(columns=new_names)
+    except Exception:
+        pass
+    return df
+
+
+def to_date(value):
+    """
+    Converte para date:
+    - pandas.Timestamp/datetime/date -> date
+    - string em formatos 'YYYY-MM-DD' ou 'DD/MM/YYYY'
+    - retorna None se não puder converter
+    Necessário para compatibilidade com SQLite e outros.
+    """
+    if value is None or (isinstance(value, float) and pd.isna(value)) or pd.isna(value):
+        return None
+
+    if isinstance(value, pd.Timestamp):
+        return value.date()
+    if isinstance(value, datetime):
+        return value.date()
+    if isinstance(value, date):
+        return value
+
+    if isinstance(value, str):
+        s = value.strip()
+        if not s:
+            return None
+        # tenta ISO
+        try:
+            return datetime.strptime(s, "%Y-%m-%d").date()
+        except Exception:
+            pass
+        # tenta BR
+        try:
+            return datetime.strptime(s, "%d/%m/%Y").date()
+        except Exception:
+            pass
+    return None
+
+
+def safe_value(value):
+    """
+    Retorna 0 se o valor for vazio/NaN; senão o próprio valor.
+    Usado para campos "livres" que não podem ir nulos.
+    """
+    if value is None or (isinstance(value, float) and pd.isna(value)) or pd.isna(value):
+        return 0
+    return value
+
+
+def safe_int(value) -> int:
+    """
+    Converte para int com fallback (None/NaN/erro -> 0).
+    """
+    try:
+        if value is None or pd.isna(value):
+            return 0
+        # strings vazias
+        if isinstance(value, str) and value.strip() == "":
+            return 0
+        return int(float(value))
+    except Exception:
+        return 0
+
+
+def safe_str(value) -> str:
+    """
+    Converte para string segura (None/NaN -> "").
+    """
+    if value is None or (isinstance(value, float) and pd.isna(value)) or pd.isna(value):
+        return ""
+    return str(value)
+
+
+def _df_template() -> pd.DataFrame:
+    """
+    Retorna um DataFrame vazio com as colunas esperadas (para gerar o modelo).
+    """
+    return pd.DataFrame(columns=COLUNAS_ESPERADAS)
+
+
+def _download_modelo_excel() -> BytesIO:
+    """
+    Gera um arquivo Excel de modelo (aba MODELO) e retorna buffer.
+    """
+    df = _df_template()
+    buf = BytesIO()
+    with pd.ExcelWriter(buf, engine="openpyxl") as writer:
+        df.to_excel(writer, index=False, sheet_name="MODELO")
+    buf.seek(0)
+    return buf
+
+
+def _validate_required_cols(df: pd.DataFrame) -> list:
+    """
+    Retorna a lista de colunas faltantes em relação a COLUNAS_ESPERADAS.
+    Apenas informa; a importação pode prosseguir mesmo faltando algumas
+    (desde que sua tabela/ETL aceite nulos).
+    """
+    missing = [c for c in COLUNAS_ESPERADAS if c not in df.columns]
+    return missing
+
+
+# =====================================================
+# MÓDULO PRINCIPAL
+# =====================================================
+def main():
+    st.title("📥 Importação de Dados via Excel")
+
+    st.markdown(
+        """
+        Este módulo permite:
+        - 📄 Baixar um **modelo Excel padrão**
+        - ✍️ Preencher os dados offline
+        - 🔍 Validar antes da gravação
+        - 💾 Importar os registros para o banco
+        """
+    )
+
+    # =====================================================
+    # 1️⃣ GERAR MODELO EXCEL
+    # =====================================================
+    st.subheader("📄 Baixar modelo Excel")
+
+    buffer = _download_modelo_excel()
+    st.download_button(
+        label="⬇️ Baixar modelo Excel",
+        data=buffer,
+        file_name="modelo_importacao_load.xlsx",
+        mime="application/vnd.openxmlformats-officedocument.spreadsheetml.sheet",
+    )
+
+    st.divider()
+
+    # =====================================================
+    # 2️⃣ UPLOAD DO ARQUIVO
+    # =====================================================
+    st.subheader("📤 Importar arquivo preenchido")
+
+    arquivo = st.file_uploader(
+        "Selecione o arquivo Excel (.xlsx)",
+        type=["xlsx"]
+    )
+
+    if not arquivo:
+        st.info("📌 Faça o upload de um arquivo para continuar.")
+        return
+
+    # Leitura com engine explícito para maior compatibilidade
+    try:
+        df = pd.read_excel(arquivo, engine="openpyxl")
+    except Exception as e:
+        st.error(f"❌ Erro ao ler o arquivo: {e}")
+        return
+
+    if df is None or df.empty:
+        st.error("❌ O arquivo não possui dados (planilha vazia).")
+        return
+
+    # Normaliza colunas (case/espaços) para nomes esperados
+    df = _normalize_cols_case_insensitive(df)
+
+    # Garante um identificador estável para cada linha durante as edições
+    if "_row_id" not in df.columns:
+        df["_row_id"] = range(len(df))
+
+    # Salva o DF bruto na sessão para persistência entre reruns
+    st.session_state["df_raw"] = df.copy()
+
+    # Diagnóstico de colunas
+    faltantes = _validate_required_cols(df)
+    if faltantes:
+        st.warning(
+            "⚠️ Algumas colunas esperadas **não** foram encontradas no arquivo ("
+            + ", ".join(faltantes) + "). "
+            "Se esses campos forem obrigatórios em seu banco, a importação pode falhar."
+        )
+
+    st.success("✅ Arquivo carregado com sucesso!")
+
+    # =====================================================
+    # 3️⃣ PRÉVIA DOS DADOS
+    # =====================================================
+    st.subheader("🔍 Prévia dos dados (arquivo lido)")
+    st.dataframe(df, use_container_width=True)
+
+    # =====================================================
+    # 4️⃣ VERIFICAÇÃO DE DUPLICIDADE (com seleção de linhas a excluir)
+    # =====================================================
+    st.subheader("🧪 Verificação de duplicidade")
+
+    st.caption(
+        "Escolha as colunas que definem a duplicidade. "
+        "Em seguida, marque o checkbox **_excluir** nas linhas que **não** deseja importar."
+    )
+
+    # Sugerimos um conjunto padrão de chaves, mas só usamos as que existem no arquivo
+    sugestao_chaves = [c for c in ["fpso", "osm", "po", "part_number", "nota_fiscal", "data_coleta"] if c in df.columns]
+
+    chaves = st.multiselect(
+        "📌 Colunas para verificação de duplicidade:",
+        options=list(df.columns),
+        default=sugestao_chaves if len(sugestao_chaves) > 0 else []
+    )
+
+    # Prepara um DF de trabalho com colunas auxiliares
+    work_df = df.copy()
+    work_df["_duplicado"] = False
+    work_df["_excluir"] = False  # será editado pelo usuário
+
+    if len(chaves) == 0:
+        st.info("Selecione ao menos **uma** coluna para verificar duplicidade.")
+        # Mesmo sem duplicidade definida, permitimos marcar exclusões manuais:
+        with st.expander("🔧 (Opcional) Excluir linhas manualmente mesmo sem duplicidade"):
+            manual_view = work_df.set_index("_row_id")[
+                [c for c in work_df.columns if c not in ["_duplicado"]] + ["_excluir"]
+            ]
+            edited_manual = st.data_editor(
+                manual_view,
+                use_container_width=True,
+                num_rows="fixed",
+                column_config={
+                    "_excluir": st.column_config.CheckboxColumn("Excluir da importação")
+                }
+            )
+            # Aplica exclusões manuais
+            work_df = work_df.set_index("_row_id")
+            work_df["_excluir"] = edited_manual["_excluir"].reindex(work_df.index).fillna(False).astype(bool)
+            work_df = work_df.reset_index()
+
+    else:
+        # Marca duplicadas com base nas chaves selecionadas
+        mask_dup_any = work_df.duplicated(subset=chaves, keep=False)
+        work_df["_duplicado"] = mask_dup_any
+
+        # Sugerimos exclusão automática das ocorrências não-primárias (o usuário pode alterar)
+        mask_dup_not_first = work_df.duplicated(subset=chaves, keep="first")
+        work_df.loc[mask_dup_not_first, "_excluir"] = True
+
+        if mask_dup_any.any():
+            st.warning("⚠️ Foram encontradas linhas duplicadas com base nas chaves selecionadas:")
+            # Mostrar somente as duplicadas para facilitar a decisão
+            cols_para_mostrar = chaves + [c for c in ["_duplicado", "_excluir"] if c not in chaves]
+            # Evita colunas repetidas mantendo ordem
+            seen = set()
+            cols_para_mostrar = [c for c in cols_para_mostrar if not (c in seen or seen.add(c))]
+
+            dup_view = work_df.loc[mask_dup_any].set_index("_row_id")[cols_para_mostrar]
+            edited_dup = st.data_editor(
+                dup_view,
+                use_container_width=True,
+                num_rows="fixed",
+                column_config={
+                    "_excluir": st.column_config.CheckboxColumn("Excluir da importação"),
+                    "_duplicado": st.column_config.CheckboxColumn("Duplicado", disabled=True)
+                }
+            )
+
+            # Mescla de volta as escolhas do usuário
+            work_df = work_df.set_index("_row_id")
+            if "_excluir" in edited_dup.columns:
+                work_df.loc[edited_dup.index, "_excluir"] = (
+                    edited_dup["_excluir"].reindex(work_df.index).fillna(work_df["_excluir"]).astype(bool)
+                )
+            work_df = work_df.reset_index()
+
+            st.info(
+                f"📊 Totais — Linhas: {len(work_df)} | Duplicadas: {mask_dup_any.sum()} | "
+                f"Marcadas para excluir: {int(work_df['_excluir'].sum())}"
+            )
+        else:
+            st.success("✅ Nenhuma duplicidade encontrada com as chaves selecionadas.")
+
+    st.divider()
+
+    # =====================================================
+    # 4.1️⃣ Prévia final do que será importado + download
+    # =====================================================
+    df_para_importar = work_df[~work_df["_excluir"]].drop(columns=["_duplicado", "_excluir"], errors="ignore")
+    st.session_state["df_para_importar"] = df_para_importar.copy()
+
+    st.subheader("🧾 Prévia do que será importado")
+    st.caption("A prévia abaixo desconsidera as linhas marcadas como **_excluir**.")
+    st.dataframe(df_para_importar.drop(columns=["_row_id"], errors="ignore"), use_container_width=True)
+
+    # Download da prévia para conferência
+    buf_prev = BytesIO()
+    with pd.ExcelWriter(buf_prev, engine="openpyxl") as writer:
+        df_para_importar.drop(columns=["_row_id"], errors="ignore").to_excel(writer, index=False, sheet_name="A_IMPORTAR")
+    buf_prev.seek(0)
+    st.download_button(
+        "⬇️ Baixar prévia (Excel)",
+        data=buf_prev,
+        file_name="previa_a_importar.xlsx",
+        mime="application/vnd.openxmlformats-officedocument.spreadsheetml.sheet",
+        help="Baixe a prévia do conjunto que será gravado."
+    )
+
+    # =====================================================
+    # 5️⃣ GRAVAÇÃO NO BANCO (usa o DataFrame filtrado)
+    # =====================================================
+    st.subheader("💾 Gravar dados no banco")
+
+    col1, col2 = st.columns(2)
+
+    if col1.button("💾 Salvar registros importados"):
+        df_import = st.session_state.get("df_para_importar", df)
+
+        if df_import.empty:
+            st.error("Não há registros para importar. Revise as exclusões.")
+            return
+
+        with SessionLocal() as db:
+            try:
+                for _, row in df_import.iterrows():
+                    registro = Equipamento(
+                        fpso1=safe_str(row.get("fpso1")),
+                        fpso=safe_str(row.get("fpso")),
+                        data_coleta=to_date(row.get("data_coleta")),
+                        especialista=safe_str(row.get("especialista")),
+                        conferente=safe_str(row.get("conferente")),
+                        osm=safe_str(row.get("osm")),
+                        modal=safe_str(row.get("modal")),
+                        quant_equip=safe_int(row.get("quant_equip")),
+                        mrob=safe_str(row.get("mrob")),
+                        linhas_osm=safe_int(row.get("linhas_osm")),
+                        linhas_mrob=safe_int(row.get("linhas_mrob")),
+                        linhas_erros=safe_int(row.get("linhas_erros")),
+                        erro_storekeeper=safe_str(row.get("erro_storekeeper")),
+                        erro_operacao=safe_str(row.get("erro_operacao")),
+                        erro_especialista=safe_str(row.get("erro_especialista")),
+                        erro_outros=safe_str(row.get("erro_outros")),
+                        inclusao_exclusao=safe_str(row.get("inclusao_exclusao")),
+                        po=safe_str(row.get("po")),
+                        part_number=safe_str(row.get("part_number")),
+                        material=safe_str(row.get("material")),
+                        solicitante=safe_str(row.get("solicitante")),
+                        motivo=safe_str(row.get("motivo")),
+                        requisitante=safe_str(row.get("requisitante")),
+                        nota_fiscal=safe_str(row.get("nota_fiscal")),
+                        impacto=safe_str(row.get("impacto")),
+                        dimensao=safe_str(row.get("dimensao")),
+                        observacoes=safe_str(row.get("observacoes")),
+                        dia_inclusao=safe_str(row.get("dia_inclusao")),
+                    )
+                    db.add(registro)
+
+                db.commit()
+
+                try:
+                    registrar_log(
+                        usuario=st.session_state.get("usuario"),
+                        acao=f"IMPORTAÇÃO EXCEL ({len(df_import)} registros) - com filtro de duplicidade",
+                        tabela="equipamentos",
+                        registro_id=None
+                    )
+                except Exception:
+                    # Não quebra o fluxo se auditoria falhar
+                    pass
+
+                st.success(f"🎉 Importação concluída com sucesso! {len(df_import)} registros gravados.")
+
+            except Exception as e:
+                db.rollback()
+                st.error(f"❌ Erro ao gravar no banco: {e}")
+
+    if col2.button("❌ Cancelar importação"):
+        st.warning("Importação cancelada pelo usuário.")
+
+
+if __name__ == "__main__":
+    main()