nia-etl-utils 0.6.0

Utilitários compartilhados para pipelines ETL do NIA/MPRJ

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.6.0

• Autenticação OCR — suporte a autenticação via API Key (header X-API-Key) na função executar_ocr
• Exceção OcrAutenticacaoError — tratamento específico para erros de autenticação (401/403)
• Parâmetro api_key — novo parâmetro opcional que aceita chave direta ou nome de variável de ambiente

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

.
├── 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.

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')
# ['[e-mail]', '[e-mail]']

---

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

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

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.

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,
    OcrAutenticacaoError,
    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.

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

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

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.

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=["[e-mail]"],
    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.

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.

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.

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.

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)

Autenticação: Via header X-API-Key (configurável via parâmetro ou variável de ambiente)

from nia_etl_utils import executar_ocr, OcrError, OcrAutenticacaoError, OcrTimeoutError

# Uso básico com variáveis de ambiente (INTELLIDOC_URL e INTELLIDOC_API_KEY)
with open("documento.pdf", "rb") as f:
    resultado = executar_ocr(conteudo=f.read())

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 e API key diretas
resultado = executar_ocr(
    conteudo=blob_bytes,
    url_base="http://intellidoc.mprj.mp.br",
    api_key="minha-chave-secreta",
    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)
)

# Uso sem autenticação (para APIs internas)
resultado = executar_ocr(conteudo=blob_bytes, api_key=None)

# 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)
except OcrAutenticacaoError as e:
    logger.error(f"Erro de autenticação: {e}")  # 401/403
    logger.debug(f"Detalhes: {e.details}")  # {'status': 401, 'api_key_fornecida': True}
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}")

Parâmetros da função executar_ocr:

Parâmetro	Tipo	Descrição
conteudo	bytes/object	Bytes do documento ou objeto com método read() (LOB Oracle)
url_base	str	URL da API ou nome da variável de ambiente (default: "INTELLIDOC_URL")
api_key	str ∣ None	API key ou nome da variável de ambiente. None desabilita autenticação (default: "INTELLIDOC_API_KEY")
timeout_polling	int	Tempo máximo em segundos para aguardar resultado (default: 300)
max_tentativas	int	Número de tentativas para submissão (default: 3)
intervalo_retry	int	Segundos entre tentativas de submissão (default: 5)
intervalo_polling	int	Segundos entre consultas de status (default: 1)

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
OcrAutenticacaoError	Falha na autenticação (401/403)
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

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.

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

pip install nia-etl-utils

Via GitLab

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

Modo Desenvolvimento

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

# Email SMTP
MAIL_SMTP_SERVER=smtp.mprj.mp.br
MAIL_SMTP_PORT=587
MAIL_SENDER=[e-mail]
EMAIL_DESTINATARIOS=[e-mail],[e-mail]

# 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
INTELLIDOC_API_KEY=sua-chave-api

# 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

# 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

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

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": "[e-mail]"
    },
)

---

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:

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

Versão atual: v0.6.0

---

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