Metadata-Version: 2.4
Name: dq-engine
Version: 0.7.6
Summary: Framework de validação e limpeza de dados com PySpark
Author-email: Bruna Cataldo <bruna.cataldo@autoglass.com.br>
Project-URL: Homepage, https://bitbucket.org/ced_engenharia/data_quality
Requires-Python: >=3.10
Description-Content-Type: text/markdown
Requires-Dist: pyspark<4.0.0,>=3.4.0
Provides-Extra: dev
Requires-Dist: pytest<9.0.0,>=8.0.0; extra == "dev"
Requires-Dist: pytest-mock<4.0.0,>=3.12.0; extra == "dev"
Provides-Extra: notebook
Requires-Dist: openpyxl<4.0.0,>=3.1.0; extra == "notebook"
Requires-Dist: pandas<3.0.0,>=2.0.0; extra == "notebook"
Provides-Extra: snowflake
Requires-Dist: snowflake-connector-python<4.0.0,>=3.6.0; extra == "snowflake"
Requires-Dist: pandas<3.0.0,>=2.0.0; extra == "snowflake"
Requires-Dist: pyyaml<7.0,>=6.0; extra == "snowflake"
Requires-Dist: python-dotenv<2.0.0,>=1.0.0; extra == "snowflake"
Requires-Dist: openpyxl<4.0.0,>=3.1.0; extra == "snowflake"

# dq-engine

Framework PySpark para validacao e tratamento de qualidade de dados.

## O que faz

- Resolve rule_set por nome de coluna
- Compila regras declarativas em expressoes Spark
- Gera score e status por coluna
- Aplica tratamento em lote (inclui colapso de espacos internos em texto DSC_)
- Executa pipeline antes/depois com comparativo
- Detecta avisos (informativos, nao alteram o score):
  - cardinalidade IDT_ fora do padrao;
  - similaridade categorica DSC_/TPO_ — mesma categoria com grafia diferente, por pontuacao/espacos (ex.: "S.A" vs "S/A");
  - nulos padronizados (`N/D`/`0`/`1800-01-01`) por coluna, marcando `[DROP?]` quando sao maioria.
- Registra em `dq_historico.xlsx` (abas Historico com `% Colunas Aprovadas`, Relatorio com `Causa`/`Acao Sugerida`, e Avisos)
- Suporta modo producao (valida o DataFrame ja tratado e marca situacao CORRIGIDO)

## Requisitos

- Python 3.10+
- PySpark 3.4+

## Instalacao

```bash
pip install -e .
```

## Uso rapido

```python
from pyspark.sql import SparkSession
from dq_engine import DataQualityEngine
from dq_engine.config.conventions import CONVENTIONS

spark = SparkSession.builder.getOrCreate()
engine = DataQualityEngine(spark=spark, conventions=CONVENTIONS)

df = spark.table("lakehouse.clientes")
result = engine.validate_table(df=df, table_name="CLIENTES", run_id="2026-07-02")

result.output_df.orderBy("column_name").show(truncate=False)
```

## Dois modos de uso

| Modo | Quando usar | O que faz |
|------|-------------|-----------|
| **Simulacao** (padrao) | Antes de levar o tratamento para o ETL | Valida o dado bruto, aplica o tratamento, valida de novo e compara ANTES x DEPOIS. Imprime o trecho de ETL (`treat_*`) para copiar e colar. |
| **Producao** (`producao=True`) | Depois que o ETL real ja aplicou o data clean | Valida **uma unica vez** o dado ja tratado (nao re-trata) e marca a situacao como CORRIGIDO, ou PENDENTE_FONTE se sobrar falha. |

### Simulacao

```python
df_tratado, result_before, result_after = engine.run_dq_pipeline(
    df=df,
    table_name="CLIENTES",
    output_dir="/mnt/dq_output",  # opcional: sem ele nada e gravado em arquivo
)
```

### Producao

```python
df_validado, result, _ = engine.run_dq_pipeline(
    df=df_ja_tratado,
    table_name="CLIENTES",
    output_dir="/mnt/dq_output",
    producao=True,
    ignore_columns_from_score=[],   # lista de colunas a tirar do score (veja abaixo)
)
```

### Ignorar colunas no score (`ignore_columns_from_score`)

Para uma particularidade ja conhecida e nao corrigivel agora (ex.: `IDT_X` que na verdade
deveria ser `TPO_X`), informe o nome exato da coluna:

```python
df_validado, result, _ = engine.run_dq_pipeline(
    df=df_ja_tratado,
    table_name="CLIENTES",
    output_dir="/mnt/dq_output",
    producao=True,
    ignore_columns_from_score=["IDT_SITUACAO", "IDT_SUBCATEGORIA"],
)
```

- Vale igual em **simulacao e producao**: a coluna **continua sendo validada**, mas fica fora do
  `quality_score`/indicador da tabela (aba `Historico`) e do diagnostico de falha residual
  (aba `Relatorio`). Nao e manipular score, e priorizar.
- Em simulacao, a coluna tambem fica **fora do trecho de ETL** e **nao e tratada** na simulacao
  (o resultado simulado bate com o que o ETL gerado vai fazer); aparece numa linha "fora do plano
  (lista de excecao)".
- O estado real da coluna e mostrado no console/`.txt` (subsecao "Colunas monitoradas") e
  **gravado na aba `Avisos` do Excel** (`Tipo Aviso = COLUNA_MONITORADA`) para ajuste posterior.
  No console, a coluna monitorada e **uma entrada so** (os avisos de regra, IDT_ e grafia dela
  entram resumidos: `avisos: idt_cardinality (5 valores)`); no Excel, cada aviso continua num
  registro proprio, para o painel contar por regra e categoria.
- `[]` (ou omitir) = todas as colunas entram no score.

Via `main.py`: `python main.py --table ... --producao --ignore-columns-producao IDT_X`
(flag repetivel), ou no `.env`: `DQ_PRODUCAO=true` e `DQ_IGNORE_COLUMNS_PRODUCAO=IDT_X,IDT_Y`.

## Como ler o output

O bloco AVISOS abre com o **RESUMO POR COLUNA**: uma linha por coluna com todos os avisos
dela, por exemplo:

```
  RESUMO POR COLUNA — 3 coluna(s) com aviso
  DSC_PLACA     placa padrao antigo (2); nulo fora do padrao (1)
  NUM_CHASSI    formato invalido (1); DV de chassi divergente (1)
  DSC_OUTRA     FALHA RESIDUAL score 0.0000 (upper)
```

Com `output_dir`, o console mostra **so o resumo**; o detalhe por regra (exemplos e acao)
fica no `.txt` salvo. Sem `output_dir`, o console mostra o resumo e o detalhe. O detalhe tem
dois blocos:

1. **PONTO DE ATENCAO — nao reduz o score.** Um bloco unico, em subsecoes:
   - `IDT_` fora do padrao de dominio (mais de 2 valores distintos alem de `N/D`: provavel
     erro de prefixo, deveria ser `TPO_`/`DSC_`);
   - regras WARNING (ex.: ano suspeito em datas, campos estritos como CPF/CNPJ/placa/chassi),
     **sempre com exemplos** dos valores que dispararam a regra. Colunas `DSC_` (campo
     aberto) nao validam `encoding_corruption`;
   - categorias duplicadas por grafia (`S.A` x `S/A`);
   - colunas monitoradas (fora do score).
2. **FALHA RESIDUAL — reduz o score.** Colunas que seguem com falha. Cada regra aparece
   separada, com o que significa, quantas linhas, exemplos **daquela regra** e a acao:
   - "o tratamento corrige" (nulo fora do padrao, placeholder, maiusculas...): se aparece em
     producao, a tabela nao foi reprocessada com o data_clean atual;
   - "o tratamento NAO corrige" (texto so com numero, encoding, dominio, formato): corrigir na
     fonte ou rever o prefixo da coluna.

Tudo isso vai tambem para o Excel (aba `Avisos`), com coluna, regra, categoria, quantidade e
mensagem (incluindo exemplos).

## Regras de tratamento de texto

Todo tratamento de string converte para **MAIUSCULAS**, remove acentos e trata nulos
(`N/D`), exceto:

- `COD_` — mantem o texto como veio (so `trim`): sem upper, preserva pontuacao, acentos e
  minusculas — codigos/chaves nao mudam. O nulo de `COD_` e **sempre `"0"`** (nunca `N/D`),
  inclusive em `COD_` com regra propria pelo nome (`COD_CHASSI`, `COD_UF`, `COD_CPF`...), que
  mantem a validacao/avisos do campo mas devolvem `"0"` no nulo;
- colunas com `PATH`, `LINK` ou `ANEXO` no nome (qualquer prefixo) e `DSC_` com `BUCKET`/`CAMINHO`
  (`treat_dsc_path`) — nao passam para maiusculas: preservam o caminho/link e usam minusculas.

Datas (`DAT_`/`DTH_`) **mantem o tipo**: coluna `date` continua `date` e coluna `timestamp`
continua `timestamp` (verificacao pelo schema). So coluna texto (ou inteiro, ex.: `20240101`) e
classificada pelas amostras: com hora diferente de `00:00:00` e `timestamp`, senao `date`. O
trecho de ETL sugere e explica a escolha:

```
      "DAT_NASCIMENTO": treat_date("DAT_NASCIMENTO", tipo="date"),
      "DAT_ABERTURA":   treat_datetime("DAT_ABERTURA"),
  # datas: schema date/timestamp mantém o tipo; texto pelas amostras (com hora = timestamp)
  #   DAT_ABERTURA: timestamp (schema)
  #   DAT_NASCIMENTO: date (schema)
```

Chamada direta sem `tipo`: `treat_date` e `treat_datetime` devolvem `timestamp` (compativel com
os ETLs em producao). Formatos aceitos: ISO, `dd/MM/yyyy`, `dd/MM/yyyy HH:mm[:ss]` e `yyyyMMdd`;
o parse so e tentado no formato que casa com o texto, entao funciona tambem com
`spark.sql.ansi.enabled=true`. Nulo/invalido vira `1800-01-01 00:00:00`, assim como datas com ano abaixo de
`suspicious_year_min` (default 1900, ex.: `0001-01-01`). Datas futuras sao mantidas e so geram
aviso (ano suspeito). Tabelas `REP_`/`REF_` (colunas `date`) nao tem tratamento especifico no
dq-engine.

`DSC_` (texto livre ou categorica, de baixa cardinalidade) e `STR_` usam a mesma funcao,
`treat_dsc`. DSC_ capturadas por nome (ex.: `DSC_PLACA`) mantem o tratamento especifico mesmo
com baixa cardinalidade.

- **Simbolos em qualquer posicao**: `treat_dsc` mantem letras, numeros, espaco e a pontuacao
  `. , ; : - / ( ) [ ] ' " ? ! % & + = @ _`; o resto (`* # | ~ ^ $ < >`, emojis...) e removido
  tambem no meio do texto (`"EVENTO** RELATO"` -> `EVENTO RELATO`).
- **Ruido nas bordas**: simbolos soltos no inicio/fim saem (`"SUZANO *"` -> `SUZANO`,
  `"SUZANO ."` -> `SUZANO`), mas `(`/`[` no inicio e `)`/`]`/`%` no fim ficam
  (`"... (GASOLINA / FLEX)"` mantem o `)` final).
- **Placeholders**: `X`, `TESTE`, sequencia de zeros (`0`, `000000`) e o mesmo digito repetido
  3+ vezes (`111`, `11111111111111111111111`) viram `N/D`. `1`, `22`, `208` continuam valor.
- **Ponto final**: `.` colado em letra/numero fica (abreviacao: `16V AUT.`); ponto solto
  (`SUZANO .`) ou repetido (`AUT..`) sai.
- **Localidade** (`CIDADE`, `MUNICIPIO`, `ESTADO`, `BAIRRO`, `PAIS`, `LOCALIDADE`, `DISTRITO`,
  `REGIAO` no nome — via `treat_dsc` ou `treat_string`): limpeza estrita, so letras, numeros,
  espaco, hifen e apostrofo (`UBERABA.` -> `UBERABA`, `Embu-Guaçu` -> `EMBU-GUACU`,
  `Santa Bárbara D'Oeste` -> `SANTA BARBARA D'OESTE`). UF usa `treat_uf` (so sigla).
  Cidade com UF junto (`VILA VELHA, ES`, `UBERABA - MG`, `CAMPINAS (SP)`) gera aviso
  `cidade_com_uf` sugerindo separar em duas colunas (cidade e UF). `VAL_`/`QTD_`/`NUM_`/`COD_`
  com `CIDADE` no nome (ex.: `VAL_CUSTO_CIDADE_MENSAL`) seguem a regra do prefixo.
- Colunas com `MODELO` no nome nao usam a regra "texto so com numero" (`208`, `3008`, `500`
  sao modelos reais).

Funcoes `treat_*` do trecho de ETL sao sempre as mesmas desde a 0.7.2 — ajustes novos mudam o
comportamento dessas funcoes, sem criar nomes novos. Para os ajustes valerem nos ETLs que ja
estao em producao (que chamam a funcao gravada no trecho da epoca), cada `treat_*` aplica a
regra da familia pelo **nome da coluna**:

| Chamada no ETL | Coluna | Regra aplicada |
|---|---|---|
| `treat_string` / `treat_dsc` / `treat_dsc_path` | `COD_*` (ex.: `COD_APOLICE`, `COD_DOCUMENTO`, `COD_LINK`) | `treat_cod` (texto mantido, nulo `"0"`) |
| `treat_string` / `treat_dsc` | nome com `PATH`/`LINK` | `treat_dsc_path` |
| `treat_string` | `DSC_*` / `STR_*` | `treat_dsc` |
| `treat_float` | `NUM_*` / `QTD_*` | `treat_int` |

`COD_` com campo estrito (`COD_CHASSI`, `COD_UF`, `COD_CPF`...) continua com a validacao do campo,
mas o nulo e `"0"`.

## Tipos numericos

| Prefixo | Tipo | Tratamento |
|---------|------|------------|
| `VAL_`, `PERC_` | float | `treat_float` (`treat_string` em `VAL_`/`PERC_` tambem vira float) |
| `NUM_`, `QTD_` | int | `treat_int` (decimais sao truncados; coluna com casas decimais deve ser `VAL_`) |

Excecoes pelo nome continuam com a regra propria: `NUM_APOLICE`/`NUM_DOCUMENTO` (texto),
`NUM_TELEFONE`, `NUM_CHASSI`, `NUM_CPF...`/`NUM_CNPJ...`. Atencao: `int` vai ate
2.147.483.647 — colunas `NUM_` com mais digitos (protocolos, notas fiscais) estouram.

`QTD_`, `VAL_` e `PERC_` seguem sempre o prefixo, mesmo com `UF`/`CPF`/`PLACA` no nome
(`QTD_ORDEM_UF_MES` e int, nao UF). A sigla `UF` so casa como palavra inteira do nome (`COD_UF`,
`UF_ORIGEM`).

**Sentinelas numericos** (`9999`, `-999`, `-9999` — `numeric_sentinel_values` — e `999` so em
`VAL_`), comuns em `fill_na` de ETLs de score (`VAL_SCORE_* = 999`, `VAL_FRANQUIA = -9999.0`),
geram aviso `numeric_sentinel` (categoria `NULO`, nao afeta o score). O valor **nao** e trocado
no tratamento; o ajuste e no `fill_na` do ETL (nulo numerico padrao `0`).

**`999` e `998` sao valores reais**, nunca nulo: `COD_SEGURADORA 999` = loja propria / TODAS
CIAS, `998` = RLP/RPS, `NUM_SEQUENCIA 999` = loja propria. `treat_cod`/`treat_int` mantem o
valor e a validacao nao acusa nulo fora do padrao.

## Campos estritos: so aviso, valor invalido mantido

CPF, CNPJ, CEP, UF, telefone, placa e chassi (resolvidos pelo nome da coluna) **nao afetam o
score** — todas as regras sao `WARNING` e aparecem na aba `Avisos`. No tratamento:

- **nulo disfarcado** (`VAZIO`, `0`, `TESTE`, `000.000.000-00`, vazio) -> `N/D`;
- **valor valido** -> forma padronizada (sem mascara; telefone como `(XX) XXXXX-XXXX`);
- **valor invalido** -> **mantido como veio**, com aviso "valor mantido; corrigir na origem"
  (ex.: CNPJ incompleto continua incompleto).

Avisos especificos, alem de formato/nulo:

| Aviso | Quando |
|-------|--------|
| `cpf_check_digit` / `cnpj_check_digit` | digito verificador invalido ou digitos repetidos |
| `chassi_check_digit` | chassi com formato valido mas digito verificador (ISO 3779) divergente |
| `telefone_ddd` | DDD inexistente na Anatel |
| `placa_padrao_antigo` | placa `ABC1234` — **nao** e convertida para Mercosul |
| `domain` (UF) | nome por extenso ou codigo IBGE — **nao** e convertido para sigla |

No modo simulacao, problemas que o tratamento corrigiu (ex.: nulo disfarcado -> `N/D`) tambem
continuam na aba `Avisos`, com a indicacao "trocado(s) por N/D no tratamento".

### Colunas `_NORMATIZADO` (PKs: chassi, placa, CPF, CNPJ)

Para colunas de chassi, placa, CPF e CNPJ (`normatizado_rule_sets`), o trecho de ETL ganha um
bloco extra que cria `<COLUNA>_NORMATIZADO`. Nessa coluna, valor valido vira a forma padronizada
e valor invalido ou nulo vira `N/D` (`"0"` em `COD_`). A coluna original segue o tratamento
normal (invalido mantido + aviso):

```python
df = df.withColumns({
    "NUM_CHASSI_NORMATIZADO": treat_chassi("NUM_CHASSI", normatizado=True),
    "NUM_CPF_NORMATIZADO":    treat_cpf("NUM_CPF", normatizado=True),
})
```

| Valor em `NUM_CPF` | `NUM_CPF` (tratado) | `NUM_CPF_NORMATIZADO` |
|---|---|---|
| `529.982.247-25` | `52998224725` | `52998224725` |
| `123.456.789-00` (DV errado) | `123.456.789-00` | `N/D` |
| `000.000.000-00` | `N/D` | `N/D` |

`normatizado=True` e um parametro das mesmas `treat_*` (sem nome novo). `NUM_APOLICE` nao tem
validacao de formato (e texto) e por isso nao ganha coluna `_NORMATIZADO`.

## Onde os arquivos sao salvos

Quando `output_dir` e informado, o `DqOutputSaver` grava:

```
{output_dir}/
├── table_output/
│   └── {TABELA}_{TIMESTAMP}.txt        # relatorio detalhado de cada execucao
└── output_consolidado/
    └── dq_historico_v2.xlsx            # historico acumulado (nome via historico_filename)
```

Abas do Excel: `Historico` (indicador/score por execucao), `Relatorio` (diagnostico por coluna),
`Avisos` (todos os avisos acima) e `Distribuicao` (opcional, `include_distribution=True`).

No `main.py` a configuracao e:

| Item | Flag / `.env` | Default |
|------|---------------|---------|
| Pasta de saida | `--output-dir` / `DQ_OUTPUT_DIR` | `\KIRIBATI.autoglass.com.br\Departamento$\Célula de Dados e Implantação\Engenharia\Outputs_Data_Quality` |
| Nome do Excel | `--historico-filename` / `DQ_HISTORICO_FILENAME` | `dq_historico_v2.xlsx` |
| Tabelas ja processadas | `--processed-file` / `DQ_PROCESSED_FILE` | `state/processed_tables.yaml` |
| Tabelas com falha | `--failed-file` / `DQ_FAILED_FILE` | `state/failed_tables.yaml` |

Se o Excel estiver aberto, o save avisa e nao grava o registro: feche e rode de novo.

## Leitura via Snowflake (opcional)

O engine sempre recebe um DataFrame Spark comum — a origem dos dados e desacoplada. Para descobrir e ler tabelas do Snowflake por schema, instale o extra `snowflake`:

```bash
pip install -e ".[snowflake]"
```

```python
from dq_engine.sources.snowflake_source import SnowflakeReader, discover_tables

reader = SnowflakeReader()  # credenciais via variaveis de ambiente (.env)
table_names = discover_tables(reader, schema_like="%NOME_DO_SCHEMA%")
df = reader.read(spark, table_names[0])

result = engine.validate_table(df=df, table_name=table_names[0], run_id="2026-07-14")
```

O script `main.py` automatiza o fluxo (descobre tabelas do schema -> roda dq-engine -> salva consolidado/historico) e pode ser agendado no Sonata:

```bash
python main.py --schema-like "%NOME_DO_SCHEMA%"
```

O padrao LIKE tambem pode vir do `.env` (variavel `TABLE_SCHEMA`), dispensando `--schema-like` em toda chamada.

Tambem e possivel rodar uma lista fixa de tabelas em vez de (ou combinada com) o schema, via `--table` (repetivel) ou pela variavel `TABLE_LIST` no `.env` (nomes completos `DATABASE.SCHEMA.TABELA` separados por virgula):

```bash
python main.py --table TRUSTED.MAX_APOLICE.TABELA_1 --table TRUSTED.MAX_APOLICE.TABELA_2
```

Tambem e possivel apontar um YAML com lista simples de tabelas (como `to_process.yaml`) usando `--tables-yaml` ou a variavel `TABLE_LIST_FILE` no `.env`:

```bash
python main.py --tables-yaml to_process.yaml
```

Sem `--schema-like`/`TABLE_SCHEMA`, o main.py roda so as tabelas informadas (`--table`, `TABLE_LIST` ou `--tables-yaml`), sem descoberta via `INFORMATION_SCHEMA`; com os dois informados, a lista filtra o resultado da descoberta.

Tabelas processadas com sucesso ficam registradas em `processed_tables.yaml` (arquivo temporario, desacoplado do dq-engine) e sao puladas nas proximas execucoes; use `--rerun-processed` para forcar tudo de novo.

## API principal

- DataQualityEngine.validate_table
- DataQualityEngine.apply_treatment
- DataQualityEngine.run_dq_pipeline
- DataQualityEngine.get_column_treatment_contexts
- DataQualityResult.table_summary
- DataQualityResult.failed_columns

## Configuracao

Arquivo central de convencoes:

- src/dq_engine/config/conventions.py

## Documentacao

- docs/architecture.md
- docs/rules.md
