🚀 [v0.4.0-beta.0] Rastreabilidade Total & SSoT Hardening

.github/workflows/tests.yml

@@ -34,7 +34,8 @@ jobs:
       uses: codecov/codecov-action@v3
       with:
         file: ./coverage.xml
-        fail_ci_if_error: true
+        fail_ci_if_error: false
+        token: ${{ secrets.CODECOV_TOKEN }} # Opcional: Recomendado para evitar rate limits
 
   lint:
     runs-on: ubuntu-latest

autosinapi/config.py

@@ -28,6 +28,8 @@ class Config:
         # --- Constantes do ETL Pipeline ---
         "REFERENCE_FILE_KEYWORD": "Referência",
         "MAINTENANCE_FILE_KEYWORD": "Manuten",
+        "FAMILIES_FILE_KEYWORD": "familias",
+        "LABOR_FILE_KEYWORD": "mao_de_obra",
         "MAINTENANCE_DEACTIVATION_KEYWORD": "%DESATIVAÇÃO%",
 
         "TEMP_CSV_DIR": "csv_temp",
@@ -40,6 +42,7 @@ class Config:
         "STATUS_SUCCESS": "SUCESSO",
         "STATUS_SUCCESS_NO_DATA": "SUCESSO (SEM DADOS)",
         "STATUS_FAILURE": "FALHA",
+        "VERSION": "1.2.0",
 
         # --- Constantes do Pre-Processor ---
         "SHEETS_TO_CONVERT": ['CSD', 'CCD', 'CSE'],
@@ -67,6 +70,7 @@ class Config:
             "TIPO_ITEM": "TIPO_ITEM", "CODIGO_COMPOSICAO": "CODIGO_DA_COMPOSICAO",
             "CODIGO_ITEM": "CODIGO_DO_ITEM", "COEFICIENTE": "COEFICIENTE",
             "DESCRICAO_ITEM": "DESCRICAO", "UNIDADE_ITEM": "UNIDADE",
+            "GRUPO_COMPOSICAO": "GRUPO",
         },
 
         "HEADER_SEARCH_LIMIT": 20,
@@ -78,7 +82,8 @@ class Config:
         "UNPIVOT_VALUE_PRECO": "preco_mediano",
         "UNPIVOT_VALUE_CUSTO": "custo_total",
         "FINAL_CATALOG_COLUMNS": {
-            "CODIGO": "codigo", "DESCRICAO": "descricao", "UNIDADE": "unidade"
+            "CODIGO": "codigo", "DESCRICAO": "descricao", "UNIDADE": "unidade",
+            "CLASSIFICACAO": "classificacao", "GRUPO": "grupo"
         },
 
         # --- Constantes do Database ---
@@ -89,6 +94,10 @@ class Config:
         "DB_TABLE_COMPOSICAO_SUBCOMPOSICOES": "composicao_subcomposicoes",
         "DB_TABLE_PRECOS_INSUMOS": "precos_insumos_mensal",
         "DB_TABLE_CUSTOS_COMPOSICOES": "custos_composicoes_mensal",
+        "DB_TABLE_INSUMOS_FAMILIAS": "insumos_familias",
+        "DB_TABLE_COEFICIENTES_FAMILIA": "coeficientes_familia_mensal",
+        "DB_TABLE_COMPOSICOES_MIX_MO": "composicoes_mix_mao_de_obra",
+        "DB_TABLE_AUDIT_LOG": "sinapi_audit_log",
         "ITEM_TYPE_INSUMO": "INSUMO",
         "ITEM_TYPE_COMPOSICAO": "COMPOSICAO",
         "DB_DIALECT": "postgresql",
@@ -105,7 +114,7 @@ def __init__(
     ):
         """
         Inicializa e valida todas as configurações do AutoSINAPI.
-
+        
         Args:
             db_config: Dicionário com as configurações do banco de dados.
             sinapi_config: Dicionário com os parâmetros da extração SINAPI.
@@ -117,10 +126,10 @@ def __init__(
         self._validate_sinapi_config(sinapi_config)
         self.db_config = db_config
         self.sinapi_config = sinapi_config
-
+        
         # Valida e define o modo de operação
         self.mode = self._validate_mode(mode)
-
+        
         # --- Expõe as configurações como atributos de alto nível ---
         self.DOWNLOAD_DIR = "./downloads"
         self.YEAR = sinapi_config["year"]
@@ -132,19 +141,27 @@ def __init__(
         self.DB_NAME = db_config["database"]
         self.DB_USER = db_config["user"]
         self.DB_PASSWORD = db_config["password"]
-
+        
         # --- Carrega as constantes (customizadas ou padrão) ---
         # Isso permite que o usuário personalize nomes de tabelas, arquivos, etc.
         constants = self.DEFAULT_CONSTANTS.copy()
         if custom_constants:
             constants.update(custom_constants)
 
+        # Sandbox mode: prefix table names
+        self._sandbox_prefix = ""
+        if self.mode == "sandbox":
+            self._sandbox_prefix = "sandbox_"
+
         for key, value in constants.items():
+            # Add sandbox prefix to table names
+            if key.startswith("DB_TABLE_"):
+                value = f"{self._sandbox_prefix}{value}"
             setattr(self, key, value)
 
     def _validate_mode(self, mode: str) -> str:
-        if mode not in ("server", "local"):
-            raise ConfigurationError(f"Modo inválido: {mode}. Use 'server' ou 'local'")
+        if mode not in ("server", "local", "sandbox"):
+            raise ConfigurationError(f"Modo inválido: {mode}. Use 'server', 'local' ou 'sandbox'")
         return mode
 
     def _validate_db_config(self, config: Dict[str, Any]) -> Dict[str, Any]:

autosinapi/core/pre_processor.py

@@ -38,10 +38,11 @@
       serão consumidos posteriormente pela classe `Processor`.
 """
 
-import pandas as pd
-import os
 import logging
 from pathlib import Path
+from typing import List
+
+import pandas as pd
 
 from autosinapi.config import Config
 from autosinapi.exceptions import ProcessingError
@@ -50,7 +51,7 @@
 
 def convert_excel_sheets_to_csv(
     xlsx_full_path: Path,
-    sheets_to_convert: list[str],
+    sheets_to_convert: List[str],
     output_dir: Path,
     config: Config
 ):

autosinapi/core/processor.py

@@ -289,10 +289,11 @@ def process_composicao_itens(self, xlsx_path: str) -> Dict[str, pd.DataFrame]:
                     cols["CODIGO_COMPOSICAO"]: "codigo",
                     cols["DESCRICAO_ITEM"]: "descricao",
                     cols["UNIDADE_ITEM"]: "unidade",
+                    cols["GRUPO_COMPOSICAO"]: "grupo",
                 }
             )
             parent_composicoes_df = parent_composicoes_df[
-                ["codigo", "descricao", "unidade"]
+                ["codigo", "descricao", "unidade", "grupo"]
             ].drop_duplicates(subset=["codigo"])
 
             child_item_details = subitens[
@@ -335,10 +336,17 @@ def _process_precos_sheet(
 
             catalogo_df = pd.DataFrame()
             if "CODIGO" in df.columns and "DESCRICAO" in df.columns:
-                catalogo_df = df[["CODIGO", "DESCRICAO", "UNIDADE"]].copy()
+                cols_catalogo = ["CODIGO", "DESCRICAO", "UNIDADE"]
+                if "CLASSIFICACAO" in df.columns:
+                    cols_catalogo.append("CLASSIFICACAO")
+                catalogo_df = df[cols_catalogo].copy()
                 self.logger.debug(f"Extraídos {len(catalogo_df)} registros de catálogo da aba {sheet_name}.")
 
-            long_df = self._unpivot_data(df, ["CODIGO"], self.config.UNPIVOT_VALUE_PRECO)
+            id_vars = ["CODIGO"]
+            if "ORIGEM_DE_PRECO" in df.columns:
+                id_vars.append("ORIGEM_DE_PRECO")
+
+            long_df = self._unpivot_data(df, id_vars, self.config.UNPIVOT_VALUE_PRECO)
             self.logger.debug(f"Extraídos {len(long_df)} registros de preços da aba {sheet_name}.")
             return long_df, catalogo_df
         except Exception as e:
@@ -389,7 +397,10 @@ def clean_level0(val):
 
             catalogo_df = pd.DataFrame()
             if "CODIGO" in df.columns and "DESCRICAO" in df.columns:
-                catalogo_df = df[["CODIGO", "DESCRICAO", "UNIDADE"]].copy()
+                cols_catalogo = ["CODIGO", "DESCRICAO", "UNIDADE"]
+                if "GRUPO" in df.columns:
+                    cols_catalogo.append("GRUPO")
+                catalogo_df = df[cols_catalogo].copy()
 
             cost_cols = {
                 col.split("_")[0]: col
@@ -414,20 +425,35 @@ def _aggregate_final_dataframes(
         self, all_dfs: Dict, temp_insumos: List, temp_composicoes: List
     ) -> Dict:
         self.logger.info("Agregando e finalizando DataFrames...")
+
         if temp_insumos:
-            all_insumos = pd.concat(
-                temp_insumos, ignore_index=True
-            ).drop_duplicates(subset=["CODIGO"])
+            all_insumos = pd.concat(temp_insumos, ignore_index=True)
+
+            # Priorizar linhas com CLASSIFICACAO preenchida
+            if "CLASSIFICACAO" in all_insumos.columns:
+                # Cria coluna temporária: 1 se tem valor, 0 se é nulo/vazio
+                all_insumos["_has_class"] = all_insumos["CLASSIFICACAO"].notnull() & (all_insumos["CLASSIFICACAO"] != "")
+                all_insumos.sort_values(by=["CODIGO", "_has_class"], ascending=[True, False], inplace=True)
+                all_insumos.drop(columns=["_has_class"], inplace=True)
+
+            all_insumos.drop_duplicates(subset=["CODIGO"], keep="first", inplace=True)
             all_dfs["insumos"] = all_insumos.rename(
                 columns=self.config.FINAL_CATALOG_COLUMNS
             )
             self.logger.info(
                 f"Catálogo de insumos finalizado com {len(all_insumos)} registros únicos."
             )
+
         if temp_composicoes:
-            all_composicoes = pd.concat(
-                temp_composicoes, ignore_index=True
-            ).drop_duplicates(subset=["CODIGO"])
+            all_composicoes = pd.concat(temp_composicoes, ignore_index=True)
+
+            # Priorizar linhas com GRUPO preenchido
+            if "GRUPO" in all_composicoes.columns:
+                all_composicoes["_has_group"] = all_composicoes["GRUPO"].notnull() & (all_composicoes["GRUPO"] != "")
+                all_composicoes.sort_values(by=["CODIGO", "_has_group"], ascending=[True, False], inplace=True)
+                all_composicoes.drop(columns=["_has_group"], inplace=True)
+
+            all_composicoes.drop_duplicates(subset=["CODIGO"], keep="first", inplace=True)
             all_dfs["composicoes"] = all_composicoes.rename(
                 columns=self.config.FINAL_CATALOG_COLUMNS
             )
@@ -489,7 +515,7 @@ def process_catalogo_e_precos(self, xlsx_path: str) -> Dict[str, pd.DataFrame]:
                         if process_type == "precos"
                         else ("custos_composicoes_mensal", "composicao_codigo")
                     )
-                    long_df.rename(columns={"CODIGO": code}, inplace=True)
+                    long_df.rename(columns={"CODIGO": code, "ORIGEM_DE_PRECO": "origem_preco"}, inplace=True)
                     all_dfs.setdefault(table, []).append(long_df)
                     self.logger.info(f"Dados da aba '{sheet_name}' adicionados à chave '{table}'.")
 
@@ -499,4 +525,52 @@ def process_catalogo_e_precos(self, xlsx_path: str) -> Dict[str, pd.DataFrame]:
                     exc_info=True,
                 )
 
-        return self._aggregate_final_dataframes(all_dfs, temp_insumos, temp_composicoes)
+        return self._aggregate_final_dataframes(all_dfs, temp_insumos, temp_composicoes)
+
+    def process_familias_e_coeficientes(self, xlsx_path: str) -> Dict[str, pd.DataFrame]:
+        self.logger.info(f"Processando famílias e coeficientes: {xlsx_path}")
+        try:
+            df = pd.read_excel(xlsx_path, sheet_name=0, header=4)
+            df = self._normalize_cols(df)
+
+            # 1. Extração de Famílias
+            familias_df = df[["CODIGO_DA_FAMILIA", "CODIGO_DO_INSUMO", "CATEGORIA"]].copy()
+            familias_df.rename(columns={
+                "CODIGO_DA_FAMILIA": "codigo_familia",
+                "CODIGO_DO_INSUMO": "insumo_codigo",
+                "CATEGORIA": "categoria"
+            }, inplace=True)
+            familias_df["insumo_codigo"] = pd.to_numeric(familias_df["insumo_codigo"], errors="coerce").astype("Int64")
+            familias_df.dropna(subset=["insumo_codigo"], inplace=True)
+
+            # 2. Extração de Coeficientes (Unpivot UFs)
+            coef_df = self._unpivot_data(df, ["CODIGO_DO_INSUMO"], "coeficiente")
+            coef_df.rename(columns={"CODIGO_DO_INSUMO": "insumo_codigo"}, inplace=True)
+            coef_df["insumo_codigo"] = pd.to_numeric(coef_df["insumo_codigo"], errors="coerce").astype("Int64")
+            coef_df.dropna(subset=["insumo_codigo"], inplace=True)
+
+            return {
+                "insumos_familias": familias_df,
+                "coeficientes_familia_mensal": coef_df
+            }
+        except Exception as e:
+            self.logger.error(f"Erro ao processar famílias e coeficientes: {e}", exc_info=True)
+            return {}
+
+    def process_mao_de_obra(self, xlsx_path: str) -> pd.DataFrame:
+        self.logger.info(f"Processando porcentagem de mão de obra: {xlsx_path}")
+        try:
+            # Lemos a aba 'SEM Desoneração' por padrão para SSOT base
+            df = pd.read_excel(xlsx_path, sheet_name=0, header=4)
+            df = self._normalize_cols(df)
+
+            # Unpivot UFs para obter a porcentagem de MO
+            long_df = self._unpivot_data(df, ["CODIGO_DA_COMPOSICAO"], "porcentagem_mo")
+            long_df.rename(columns={"CODIGO_DA_COMPOSICAO": "composicao_codigo"}, inplace=True)
+            long_df["composicao_codigo"] = pd.to_numeric(long_df["composicao_codigo"], errors="coerce").astype("Int64")
+            long_df.dropna(subset=["composicao_codigo"], inplace=True)
+
+            return long_df
+        except Exception as e:
+            self.logger.error(f"Erro ao processar mix de mão de obra: {e}", exc_info=True)
+            return pd.DataFrame()