nia-etl-utils 0.5.1
Utilitários compartilhados para pipelines ETL do NIA/MPRJ
License: MIT
Requires Python: >=3.10
Keywords: etl, data-engineering, pipeline, mprj, nia

# Módulos Utilitários do NIA

## Visão Geral

Biblioteca Python centralizada contendo utilitários compartilhados para pipelines ETL do NIA/MPRJ. Consolida funções reutilizáveis para configuração de ambiente, notificações por email, conexões de banco de dados, logging padronizado e processamento de dados.

Desenvolvida para eliminar duplicação de código, padronizar boas práticas e facilitar manutenção em todos os projetos de engenharia de dados do NIA.

---

## Novidades da v0.5.1

- **Truncamento início + fim** — novo parâmetro `truncar_inicio_fim` nas funções de embedding permite truncar textos longos preservando início e fim do conteúdo
- **Exceção EmbeddingTruncadoError** — tratamento específico para erros em embeddings truncados com método início+fim

## Novidades da v0.5.0

- **Truncamento automático de tokens** — parâmetro `max_tokens` nas funções de embedding para truncar textos que excedem o limite do modelo (padrão: 8191 tokens)
- **Dependência tiktoken** — contagem precisa de tokens usando a biblioteca oficial da OpenAI

## Novidades da v0.4.1

- **Embedding em Batch** (`gerar_embedding_openai_batch`) — geração de embeddings para múltiplos textos em uma única chamada à API, mais eficiente que chamadas individuais

## Novidades da v0.4.0

- **Módulo Embedding** (`gerar_embedding_openai`) — geração de embeddings vetoriais via Azure OpenAI com retry automático
- **Exceções de Embedding** (`EmbeddingError`, `EmbeddingTimeoutError`) — tratamento granular de erros de embedding
- **Migração Oracle** — substituição de cx_Oracle por oracledb para compatibilidade com Python 3.13+

## Novidades da v0.2.2

- **Módulo OCR** (`executar_ocr`) — processamento de OCR via API IntelliDoc com suporte a PDF, imagens e detecção automática de formato
- **Exceções de OCR** (`OcrError`, `OcrSubmissaoError`, `OcrProcessamentoError`, `OcrTimeoutError`) — tratamento granular de erros de OCR

## Novidades da v0.2.0

- **Dataclasses de configuração** (`PostgresConfig`, `OracleConfig`, `SmtpConfig`, `LogConfig`) — configurações imutáveis e type-safe
- **Exceções customizadas hierárquicas** — tratamento de erros mais preciso e informativo
- **Dataclasses de resultado** (`Conexao`, `ResultadoExtracao`, `ResultadoLote`, `ResultadoEmail`)
- **Funções adicionais de env** — `obter_variavel_env_int`, `obter_variavel_env_bool`, `obter_variavel_env_lista`
- **Context managers** para conexões de banco — fechamento automático e seguro

---

## Estrutura do Projeto

```plaintext
.
├── src/
│   └── nia_etl_utils/
│       ├── __init__.py                  # Exporta funções principais
│       ├── config.py                    # Dataclasses de configuração
│       ├── exceptions.py                # Exceções customizadas
│       ├── results.py                   # Dataclasses de resultado
│       ├── env_config.py                # Gerenciamento de variáveis de ambiente
│       ├── email_smtp.py                # Envio de emails via SMTP
│       ├── database.py                  # Conexões PostgreSQL e Oracle
│       ├── logger_config.py             # Configuração de logging com Loguru
│       ├── processa_csv.py              # Processamento e exportação de CSV
│       ├── processa_csv_paralelo.py     # Processamento paralelo de CSV grandes
│       ├── limpeza_pastas.py            # Manipulação de arquivos e diretórios
│       ├── ocr.py                       # OCR via API IntelliDoc
│       └── embedding.py                 # Embeddings via Azure OpenAI
│
├── tests/                               # Testes unitários
├── .gitlab-ci.yml                       # Pipeline CI/CD
├── pyproject.toml                       # Configuração do pacote
└── README.md
```

---

## Módulos Disponíveis

### 1. Configuração de Ambiente (`env_config.py`)

Gerenciamento robusto de variáveis de ambiente com validação e tipagem.

```python
from nia_etl_utils import (
    obter_variavel_env,
    obter_variavel_env_int,
    obter_variavel_env_bool,
    obter_variavel_env_lista
)

# String obrigatória (falha com sys.exit(1) se não existir)
db_host = obter_variavel_env('DB_POSTGRESQL_HOST')

# String opcional com fallback
porta = obter_variavel_env('DB_PORT', default='5432')

# Inteiro
max_conexoes = obter_variavel_env_int('MAX_CONEXOES', default=10)

# Booleano (aceita: true/false, 1/0, yes/no, on/off)
debug = obter_variavel_env_bool('DEBUG_MODE', default=False)

# Lista (separada por vírgula)
destinatarios = obter_variavel_env_lista('EMAIL_DESTINATARIOS')
# ['email1@mprj.mp.br', 'email2@mprj.mp.br']
```

---

### 2. Dataclasses de Configuração (`config.py`)

Configurações imutáveis e type-safe com factory methods.

```python
from nia_etl_utils import PostgresConfig, OracleConfig, SmtpConfig, LogConfig

# Configuração explícita (recomendado para testes)
config = PostgresConfig(
    host="localhost",
    port="5432",
    database="teste",
    user="user",
    password="pass"
)

# Configuração via variáveis de ambiente (recomendado para produção)
config = PostgresConfig.from_env()           # usa DB_POSTGRESQL_*
config = PostgresConfig.from_env("_OPENGEO") # usa DB_POSTGRESQL_*_OPENGEO

# Connection string para SQLAlchemy
print(config.connection_string)
# postgresql+psycopg2://user:pass@localhost:5432/teste

# Oracle
oracle_config = OracleConfig.from_env()

# SMTP
smtp_config = SmtpConfig.from_env()

# Logging com padrões NIA
log_config = LogConfig.padrao_nia("meu_pipeline")
```

---

### 3. Exceções Customizadas (`exceptions.py`)

Hierarquia de exceções para tratamento preciso de erros.

```python
from nia_etl_utils import (
    # Base
    NiaEtlError,
    # Configuração
    ConfiguracaoError,
    VariavelAmbienteError,
    # Database
    DatabaseError,
    ConexaoError,
    # Arquivos
    ArquivoError,
    DiretorioError,
    EscritaArquivoError,
    LeituraArquivoError,
    # Extração
    ExtracaoError,
    ExtracaoVaziaError,
    ProcessamentoError,
    # Email
    EmailError,
    DestinatarioError,
    SmtpError,
    # Embedding
    EmbeddingError,
    EmbeddingTimeoutError,
    EmbeddingTruncadoError,
    # Validação
    ValidacaoError,
)
from nia_etl_utils import (
    # OCR
    OcrError,
    OcrSubmissaoError,
    OcrProcessamentoError,
    OcrTimeoutError,
)

# Uso em try/except
try:
    config = PostgresConfig.from_env("_INEXISTENTE")
except ConfiguracaoError as e:
    logger.error(f"Configuração inválida: {e}")
    logger.debug(f"Detalhes: {e.details}")

# Exceções incluem contexto
try:
    with conectar_postgresql(config) as conn:
        conn.cursor.execute("SELECT * FROM tabela")
except ConexaoError as e:
    # e.details contém informações adicionais
    print(e.details)  # {'host': 'localhost', 'database': 'teste', ...}
```

---

### 4. Dataclasses de Resultado (`results.py`)

Estruturas para retorno de operações.

```python
from nia_etl_utils import Conexao, ResultadoExtracao, ResultadoLote, ResultadoEmail

# Conexao - retornada por conectar_postgresql/oracle
with conectar_postgresql(config) as conn:
    conn.cursor.execute("SELECT 1")
    # conn.cursor e conn.connection disponíveis
    # Fechamento automático ao sair do context manager

# ResultadoExtracao - retornado por extrair_e_exportar_csv
resultado = extrair_e_exportar_csv(...)
print(resultado.sucesso)      # True/False
print(resultado.caminho)      # '/dados/arquivo.csv'
print(resultado.linhas)       # 1500
print(resultado.tempo_execucao)  # 2.34 (segundos)

# ResultadoLote - retornado por exportar_multiplos_csv
lote = exportar_multiplos_csv(...)
print(lote.total)             # 5
print(lote.sucessos)          # 4
print(lote.falhas)            # 1
print(lote.resultados)        # Lista de ResultadoExtracao
```

---

### 5. Conexões de Banco (`database.py`)

Conexões com context manager para fechamento automático.

#### PostgreSQL

```python
from nia_etl_utils import conectar_postgresql, PostgresConfig

# Com configuração explícita
config = PostgresConfig(
    host="localhost",
    port="5432",
    database="meu_banco",
    user="usuario",
    password="senha"
)

with conectar_postgresql(config) as conn:
    conn.cursor.execute("SELECT * FROM tabela")
    resultados = conn.cursor.fetchall()
# Conexão fechada automaticamente

# Com variáveis de ambiente
config = PostgresConfig.from_env()
with conectar_postgresql(config) as conn:
    conn.cursor.execute("SELECT 1")

# Wrappers de conveniência (mantidos para retrocompatibilidade)
from nia_etl_utils import conectar_postgresql_nia, conectar_postgresql_opengeo

with conectar_postgresql_nia() as conn:
    conn.cursor.execute("SELECT * FROM ouvidorias")

# Engine SQLAlchemy
from nia_etl_utils import obter_engine_postgresql
import pandas as pd

engine = obter_engine_postgresql(config)
df = pd.read_sql("SELECT * FROM tabela", engine)
```

#### Oracle

```python
from nia_etl_utils import conectar_oracle, OracleConfig

config = OracleConfig.from_env()
with conectar_oracle(config) as conn:
    conn.cursor.execute("SELECT * FROM tabela WHERE ROWNUM <= 10")
    resultados = conn.cursor.fetchall()
```

---

### 6. Email SMTP (`email_smtp.py`)

Envio de emails com suporte a anexos.

```python
from nia_etl_utils import enviar_email_smtp

# Uso padrão (destinatários da env var EMAIL_DESTINATARIOS)
enviar_email_smtp(
    corpo_do_email="Pipeline concluído com sucesso",
    assunto="[PROD] ETL Finalizado"
)

# Com destinatários específicos e anexo
enviar_email_smtp(
    destinatarios=["diretor@mprj.mp.br"],
    corpo_do_email="Relatório executivo anexo",
    assunto="Relatório Mensal",
    anexo="/tmp/relatorio.pdf"
)
```

---

### 7. Logging (`logger_config.py`)

Configuração padronizada do Loguru.

```python
from nia_etl_utils import configurar_logger_padrao_nia, configurar_logger
from loguru import logger

# Configuração rápida com padrões do NIA
caminho_log = configurar_logger_padrao_nia("ouvidorias_etl")
logger.info("Pipeline iniciado")

# Configuração customizada
caminho_log = configurar_logger(
    prefixo="meu_pipeline",
    data_extracao="2025_01_20",
    pasta_logs="/var/logs/nia",
    rotation="50 MB",
    retention="30 days",
    level="INFO"
)
```

---

### 8. Processamento CSV (`processa_csv.py`)

Exportação de DataFrames para CSV com nomenclatura padronizada.

```python
from nia_etl_utils import exportar_para_csv, extrair_e_exportar_csv
import pandas as pd

# Exportação simples
df = pd.DataFrame({"col1": [1, 2], "col2": [3, 4]})
caminho = exportar_para_csv(
    df=df,
    nome_arquivo="dados_clientes",
    data_extracao="2025_01_20",
    diretorio_base="/tmp/dados"
)

# Extração + Exportação
def extrair_dados():
    return pd.DataFrame({"dados": [1, 2, 3]})

resultado = extrair_e_exportar_csv(
    nome_extracao="dados_vendas",
    funcao_extracao=extrair_dados,
    data_extracao="2025_01_20",
    diretorio_base="/tmp/dados",
    falhar_se_vazio=True
)

# Múltiplas extrações em lote
from nia_etl_utils import exportar_multiplos_csv

extractions = [
    {"nome": "clientes", "funcao": extrair_clientes},
    {"nome": "vendas", "funcao": extrair_vendas}
]

lote = exportar_multiplos_csv(
    extractions=extractions,
    data_extracao="2025_01_20",
    diretorio_base="/tmp/dados"
)
```

---

### 9. Processamento Paralelo de CSV (`processa_csv_paralelo.py`)

Processa arquivos CSV grandes em paralelo.

```python
from nia_etl_utils import processar_csv_paralelo

def limpar_texto(texto):
    return texto.strip().upper()

processar_csv_paralelo(
    caminho_entrada="dados_brutos.csv",
    caminho_saida="dados_limpos.csv",
    colunas_para_tratar=["nome", "descricao"],
    funcao_transformacao=limpar_texto,
    remover_entrada=True
)
```

---

### 10. Manipulação de Arquivos (`limpeza_pastas.py`)

Utilitários para limpeza e criação de diretórios.

```python
from nia_etl_utils import limpar_pasta, remover_pasta_recursivamente, criar_pasta_se_nao_existir

limpar_pasta("/tmp/dados")
remover_pasta_recursivamente("/tmp/temporario")
criar_pasta_se_nao_existir("/dados/processados/2025/01")
```

---

### 11. OCR via API IntelliDoc (`ocr.py`)

Processamento de OCR (Reconhecimento Óptico de Caracteres) via API IntelliDoc do MPRJ.

A API processa documentos de forma assíncrona:
1. Submete documento → retorna `document_id`
2. Consulta status via polling → retorna resultado quando pronto

**Formatos suportados:** PDF, JPG, PNG, GIF, BMP, TIFF (detecção automática via magic bytes)

```python
from nia_etl_utils import executar_ocr, OcrError, OcrTimeoutError

# Uso básico com variável de ambiente
with open("documento.pdf", "rb") as f:
    resultado = executar_ocr(
        conteudo=f.read(),
        url_base="INTELLIDOC_URL",  # nome da env var
    )

print(resultado["full_text"])        # Texto extraído completo
print(resultado["overall_quality"])  # Qualidade do OCR (0-1)
print(resultado["total_pages"])      # Número de páginas

# Uso com URL direta e configurações customizadas
resultado = executar_ocr(
    conteudo=blob_bytes,
    url_base="http://google.com",
    timeout_polling=600,    # Timeout máximo em segundos (default: 300)
    max_tentativas=5,       # Tentativas de submissão (default: 3)
    intervalo_retry=10,     # Segundos entre retries (default: 5)
    intervalo_polling=2,    # Segundos entre consultas (default: 1)
)

# Suporta LOBs do Oracle diretamente
with conectar_oracle(config) as conn:
    conn.cursor.execute("SELECT blob_documento FROM documentos WHERE id = :id", {"id": 123})
    blob = conn.cursor.fetchone()[0]
    resultado = executar_ocr(blob)  # LOB é convertido automaticamente

# Acessando detalhes das páginas
for page in resultado["pages"]:
    print(f"Página {page['page_number']}: {page['extraction_method']}")

# Tratamento de erros
try:
    resultado = executar_ocr(conteudo, url_base="INTELLIDOC_URL")
except OcrTimeoutError as e:
    logger.error(f"Timeout aguardando OCR: {e}")
    logger.debug(f"Detalhes: {e.details}")  # {'document_id': '...', 'ultimo_status': 'PENDING'}
except OcrError as e:
    logger.error(f"Erro no OCR: {e}")
```

**Retorno da função `executar_ocr`:**

| Campo                | Tipo  | Descrição               |
| -------------------- | ----- | ----------------------- |
| `document_id`        | str   | ID único do documento   |
| `full_text`          | str   | Texto extraído completo |
| `mime_type`          | str   | Tipo MIME detectado     |
| `overall_quality`    | float | Qualidade geral (0-1)   |
| `total_pages`        | int   | Número de páginas       |
| `processing_time_ms` | int   | Tempo de processamento  |
| `pages`              | list  | Detalhes por página     |
| `metadata`           | dict  | Metadados adicionais    |

**Exceções:**

| Exceção                 | Quando ocorre                                                         |
| ----------------------- | --------------------------------------------------------------------- |
| `OcrSubmissaoError`     | Falha ao enviar documento (rede, timeout, resposta inválida)          |
| `OcrProcessamentoError` | API retornou FAILURE/REVOKED (documento corrompido, formato inválido) |
| `OcrTimeoutError`       | Tempo máximo de polling atingido                                      |
| `TypeError`             | Tipo de conteúdo não suportado                                        |

---

### 12. Embedding via Azure OpenAI (`embedding.py`)

Geração de embeddings vetoriais utilizando a API do Azure OpenAI.

#### Texto único

```python
from nia_etl_utils import (
    gerar_embedding_openai,
    obter_variavel_env,
    EmbeddingError,
    EmbeddingTimeoutError,
    EmbeddingTruncadoError,
)

# Uso básico
vetor = gerar_embedding_openai(
    endpoint=obter_variavel_env("AZURE_OPENAI_ENDPOINT"),
    api_key=obter_variavel_env("AZURE_OPENAI_API_KEY"),
    model_embedding=obter_variavel_env("AZURE_OPENAI_MODEL_EMBEDDING"),
    api_version=obter_variavel_env("AZURE_OPENAI_API_VERSION_EMBEDDING"),
    texto="Este é um texto de exemplo para embedding.",
)
print(f"Vetor com {len(vetor)} dimensões")  # 1536 para ada-002

# Com parâmetros de retry customizados
vetor = gerar_embedding_openai(
    endpoint=endpoint,
    api_key=api_key,
    model_embedding=model_embedding,
    api_version=api_version,
    texto="Outro texto",
    max_retries=5,           # Número máximo de tentativas (default: 3)
    intervalo_segundos=10,   # Intervalo entre retries (default: 5)
)

# Truncamento com início + fim (preserva contexto de início e fim do texto)
vetor = gerar_embedding_openai(
    endpoint=endpoint,
    api_key=api_key,
    model_embedding=model_embedding,
    api_version=api_version,
    texto="Texto muito longo...",
    truncar_inicio_fim=True,  # Usa metade início + [...TRUNCADO...] + metade fim
)

# Desabilitando truncamento automático de tokens
vetor = gerar_embedding_openai(
    endpoint=endpoint,
    api_key=api_key,
    model_embedding=model_embedding,
    api_version=api_version,
    texto="Texto muito longo...",
    max_tokens=None,  # Desabilita truncamento (API pode retornar erro se exceder limite)
)

# Tratamento de erros
try:
    vetor = gerar_embedding_openai(
        endpoint=endpoint,
        api_key=api_key,
        model_embedding=model_embedding,
        api_version=api_version,
        texto=texto,
        truncar_inicio_fim=True,
    )
except EmbeddingTruncadoError as e:
    logger.error(f"Erro no truncamento início+fim: {e}")
    logger.debug(f"Tokens originais: {e.tokens_originais}")
except EmbeddingTimeoutError as e:
    logger.error(f"Timeout após retries: {e}")
    logger.debug(f"Detalhes: {e.details}")  # {'tentativas': 3, 'model': '...'}
except EmbeddingError as e:
    logger.error(f"Erro no embedding: {e}")
```

#### Batch (múltiplos textos)

Mais eficiente que chamar `gerar_embedding_openai` múltiplas vezes, pois envia todos os textos em uma única requisição à API.

```python
from nia_etl_utils import gerar_embedding_openai_batch, obter_variavel_env

# Uso básico
textos = ["Primeiro texto", "Segundo texto", "Terceiro texto"]
vetores = gerar_embedding_openai_batch(
    endpoint=obter_variavel_env("AZURE_OPENAI_ENDPOINT"),
    api_key=obter_variavel_env("AZURE_OPENAI_API_KEY"),
    model_embedding=obter_variavel_env("AZURE_OPENAI_MODEL_EMBEDDING"),
    api_version=obter_variavel_env("AZURE_OPENAI_API_VERSION_EMBEDDING"),
    textos=textos,
)
print(f"{len(vetores)} vetores gerados")  # 3 vetores

# Textos vazios retornam lista vazia
textos = ["Texto válido", "", "Outro texto"]
vetores = gerar_embedding_openai_batch(...)
# vetores[0] -> [0.123, 0.456, ...]  # embedding do primeiro texto
# vetores[1] -> []                    # texto vazio retorna lista vazia
# vetores[2] -> [0.789, 0.012, ...]  # embedding do terceiro texto
```

**Parâmetros da função `gerar_embedding_openai`:**

| Parâmetro            | Tipo           | Descrição                                                                                                                        |
| -------------------- | -------------- | -------------------------------------------------------------------------------------------------------------------------------- |
| `endpoint`           | str            | URL do endpoint Azure OpenAI                                                                                                     |
| `api_key`            | str            | Chave de API do Azure OpenAI                                                                                                     |
| `model_embedding`    | str            | Nome do modelo (ex: `text-embedding-3-small`)                                                                                    |
| `api_version`        | str            | Versão da API (ex: `2024-02-01`)                                                                                                 |
| `texto`              | str            | Texto de entrada para gerar o embedding                                                                                          |
| `max_retries`        | int            | Tentativas em caso de timeout (default: 3)                                                                                       |
| `intervalo_segundos` | int            | Intervalo entre tentativas (default: 5)                                                                                          |
| `max_tokens`         | int ∣ None     | Limite máximo de tokens; textos maiores são truncados. Use `None` para desabilitar (default: 8191)                               |
| `truncar_inicio_fim` | bool           | Se `True`, trunca preservando início + fim com marcador `[...TRUNCADO...]`. Se `False`, trunca apenas do início (default: False) |

**Parâmetros da função `gerar_embedding_openai_batch`:**

| Parâmetro            | Tipo           | Descrição                                                                                                                        |
| -------------------- | -------------- | -------------------------------------------------------------------------------------------------------------------------------- |
| `endpoint`           | str            | URL do endpoint Azure OpenAI                                                                                                     |
| `api_key`            | str            | Chave de API do Azure OpenAI                                                                                                     |
| `model_embedding`    | str            | Nome do modelo (ex: `text-embedding-3-small`)                                                                                    |
| `api_version`        | str            | Versão da API (ex: `2024-02-01`)                                                                                                 |
| `textos`             | list[str]      | Lista de textos para gerar embeddings                                                                                            |
| `max_retries`        | int            | Tentativas em caso de timeout (default: 3)                                                                                       |
| `intervalo_segundos` | int            | Intervalo entre tentativas (default: 5)                                                                                          |
| `max_tokens`         | int ∣ None     | Limite máximo de tokens por texto; textos maiores são truncados. Use `None` para desabilitar (default: 8191)                     |
| `truncar_inicio_fim` | bool           | Se `True`, trunca preservando início + fim com marcador `[...TRUNCADO...]`. Se `False`, trunca apenas do início (default: False) |

**Retorno:**
- `gerar_embedding_openai`: `list[float]` — Vetor de embedding
- `gerar_embedding_openai_batch`: `list[list[float]]` — Lista de vetores (mesma ordem dos textos de entrada)

**Exceções:**

| Exceção                 | Quando ocorre                                                            |
| ----------------------- | ------------------------------------------------------------------------ |
| `ValueError`            | Texto ou parâmetros obrigatórios vazios/inválidos                        |
| `EmbeddingTimeoutError` | Todas as tentativas falharam por timeout                                 |
| `EmbeddingTruncadoError`| Erro ao processar embedding com truncamento início+fim                   |
| `EmbeddingError`        | Qualquer outro erro na geração (resposta inválida, API, etc)             |

---

## Instalação

### Via PyPI

```bash
pip install nia-etl-utils
```

### Via GitLab

```bash
pip install git+https://gitlab-dti.mprj.mp.br/nia/etl-nia/nia-etl-utils.git@v0.5.1
```

### Modo Desenvolvimento

```bash
git clone https://gitlab-dti.mprj.mp.br/nia/etl-nia/nia-etl-utils.git
cd nia-etl-utils
pip install -e ".[dev]"
```

---

## Configuração

### Variáveis de Ambiente

```env
# Email SMTP
MAIL_SMTP_SERVER=smtp.mprj.mp.br
MAIL_SMTP_PORT=587
MAIL_SENDER=etl@mprj.mp.br
EMAIL_DESTINATARIOS=equipe@mprj.mp.br,gestor@mprj.mp.br

# PostgreSQL - NIA
DB_POSTGRESQL_HOST=postgres-nia.mprj.mp.br
DB_POSTGRESQL_PORT=5432
DB_POSTGRESQL_DATABASE=nia_database
DB_POSTGRESQL_USER=usuario
DB_POSTGRESQL_PASSWORD=senha

# PostgreSQL - OpenGeo
DB_POSTGRESQL_HOST_OPENGEO=postgres-opengeo.mprj.mp.br
DB_POSTGRESQL_PORT_OPENGEO=5432
DB_POSTGRESQL_DATABASE_OPENGEO=opengeo_database
DB_POSTGRESQL_USER_OPENGEO=usuario
DB_POSTGRESQL_PASSWORD_OPENGEO=senha

# Oracle
DB_ORACLE_HOST=oracle.mprj.mp.br
DB_ORACLE_PORT=1521
DB_ORACLE_SERVICE_NAME=ORCL
DB_ORACLE_USER=usuario
DB_ORACLE_PASSWORD=senha

# OCR (IntelliDoc)
INTELLIDOC_URL=http://intellidoc.mprj.mp.br

# Azure OpenAI (Embedding)
AZURE_OPENAI_ENDPOINT=https://meu-recurso.openai.azure.com
AZURE_OPENAI_API_KEY=sua-chave-api
AZURE_OPENAI_MODEL_EMBEDDING=text-embedding-ada-002
AZURE_OPENAI_API_VERSION_EMBEDDING=2023-05-15
```

---

## Testes

```bash
# Todos os testes
pytest

# Com cobertura
pytest --cov=src/nia_etl_utils --cov-report=term-missing

# Ou usar o script helper
./run_tests.sh --coverage --verbose
```

---

## Exemplo de Uso Completo

```python
from nia_etl_utils import (
    configurar_logger_padrao_nia,
    PostgresConfig,
    conectar_postgresql,
    exportar_para_csv,
    ConexaoError,
)
from loguru import logger
import pandas as pd
from datetime import datetime

# 1. Configura logging
configurar_logger_padrao_nia("meu_pipeline")

# 2. Carrega configuração
config = PostgresConfig.from_env()

# 3. Conecta e extrai dados
try:
    with conectar_postgresql(config) as conn:
        logger.info("Extraindo dados...")
        conn.cursor.execute("SELECT * FROM tabela WHERE data >= CURRENT_DATE - 7")
        resultados = conn.cursor.fetchall()
        colunas = [desc[0] for desc in conn.cursor.description]
        df = pd.DataFrame(resultados, columns=colunas)
except ConexaoError as e:
    logger.error(f"Falha na conexão: {e}")
    raise

logger.info(f"Extração concluída: {len(df)} registros")

# 4. Exporta CSV
data_hoje = datetime.now().strftime("%Y_%m_%d")
caminho = exportar_para_csv(
    df=df,
    nome_arquivo="dados_extraidos",
    data_extracao=data_hoje,
    diretorio_base="/dados/processados"
)

logger.success(f"Pipeline concluído! Arquivo: {caminho}")
```

---

## Integração com Airflow

```python
from airflow.providers.cncf.kubernetes.operators.kubernetes_pod import KubernetesPodOperator

task = KubernetesPodOperator(
    task_id="meu_etl",
    name="meu-etl-pod",
    namespace="airflow-nia-stage",
    image="python:3.13.3",
    cmds=[
        "sh", "-c",
        "pip install nia-etl-utils && python src/extract.py"
    ],
    env_vars={
        "DB_POSTGRESQL_HOST": "...",
        "EMAIL_DESTINATARIOS": "equipe@mprj.mp.br"
    },
)
```

---

## Tecnologias Utilizadas

- Python 3.10+ (compatível até 3.13)
- Loguru (logging)
- python-dotenv (env vars)
- requests (HTTP/OCR)
- oracledb (Oracle - compatível com Python 3.13+)
- psycopg2 (PostgreSQL)
- SQLAlchemy (engines)
- pandas (processamento de dados)
- openai (Azure OpenAI - embeddings)
- tiktoken (contagem de tokens)
- pytest + pytest-cov (testes)
- ruff (linting)

---

## Versionamento

Este projeto usa [Semantic Versioning](https://semver.org/):

- **MAJOR**: Mudanças incompatíveis na API
- **MINOR**: Novas funcionalidades (retrocompatíveis)
- **PATCH**: Correções de bugs

**Versão atual:** `v0.5.1`

---

## CI/CD

Pipeline automatizado no GitLab:

- Testes unitários (pytest)
- Cobertura de código (>= 70%)
- Linting (ruff)
- Deploy automático no PyPI (em tags)

---

## Contribuição

Merge requests são bem-vindos. Sempre crie uma branch a partir de `main`.

### Checklist:

- [ ] Testes passam: `pytest`
- [ ] Cobertura >= 70%
- [ ] Lint OK: `ruff check src/ tests/`
- [ ] Commits semânticos
- [ ] Documentação atualizada

---

## Licença

Projeto de uso interno do MPRJ. Sem licença pública.

---

## Responsável Técnico

**Nícolas Galdino Esmael** | Engenheiro de Dados - NIA | MPRJ