refactor: conforma schemas e tabelas à especificação do banco do MinC

[CaioMelo25]

Contexto

O MinC subiu um banco de homologação e existe uma especificação oficial que define os schemas, as tabelas e as chaves esperadas. O pipeline não estava em conformidade: escrevia em transferegov_fundo_a_fundo e bsc_pnab, não extraía metas, cobria no máximo 5 dos 11 programas e criava tabelas de planilha com nome derivado do nome da aba do Excel.

Isso não podia ser corrigido depois da primeira carga: ClientPostgresDB executa CREATE SCHEMA IF NOT EXISTS a cada insert, então rodar as DAGs como estavam criaria os schemas fora do padrão sozinho, no banco novo.

Schemas e tabelas

Antes	Agora
transferegov_fundo_a_fundo, bsc_pnab	transferegov, bbagil, relatorio_gestao
raw_programas, raw_planos_acao	programa_minc, plano_acao_minc
—	plano_acao_meta_minc, plano_acao_dado_bancario_minc
raw_bbagil_extrato_transacoes, raw_bbagil_subtransacoes	extrato_bbagil, subtransacao_bbagil
~15 tabelas de planilha, com nome vindo do nome da aba	6 tabelas fixas em relatorio_gestao

Os nomes passam a viver em plugins/schemas_minc.py. Antes eram literais repetidos em oito arquivos — foi assim que o código divergiu da especificação sem ninguém notar.

Extração

• Escopo: os 11 programas das 4 políticas. Estava [46, 47] numa DAG e [46, 47, 60, 61, 62] noutra, deixando LAB1 e PNAB Ciclo 2 de fora.
• api_plano_acao_meta_dag (nova): o endpoint /plano_acao_meta não era consultado por nenhuma DAG e a tabela de metas não existia.
• api_plano_acao_dado_bancario_dag (nova): grava todas as contas de cada plano com todas as colunas. Antes a informação era efeito colateral da DAG do BB Ágil, que guardava uma conta por plano e quatro campos; a escolha da conta virou regra de consumo dentro de extracao_bbagil_dag.
• Chaves de integração (id_programa, id_plano_acao, cod_ibge, id_agencia_conta, id_plano_acao_dado_bancario) propagadas até extrato, subtransação e planilhas.

Regra territorial

plugins/territorio_ibge.py deriva tipo_ente, cod_ibge, cod_ibge_uf e cod_ibge_municipio. A API do TransfereGov informa o código do município da capital quando o ente recebedor é um estado — sem essa regra, o plano estadual de São Paulo fica gravado como plano do município de São Paulo.

Planilhas dos relatórios de gestão

• A listagem passa a vir do Postgres (anexos_relatorios → relatorios_gestao → plano_acao_minc) em vez das keys do MinIO: varrer o bucket dá o arquivo, mas não diz de que plano de ação ele é.
• Cada linha carrega as colunas obrigatórias da especificação e é chaveada por hash_registro. insert_data_por_tabela inseria sem chave nenhuma e duplicava tudo a cada execução.
• Acima de 1200 colunas, coluna nova vai para payload_origem (JSON): consolidar milhares de layouts numa tabela só esbarraria no limite de 1600 colunas do Postgres.
• tabela_origem e aba_origem preservam de onde cada linha veio.

dbt

Os 19 modelos que liam os nomes antigos foram repontados. Onde houve renome, é troca de nome; onde 15 tabelas viraram 6, o modelo recupera a fatia por tabela_origem:

from {{ source('relatorio_gestao', 'planilha_dados_lpg') }}
where tabela_origem = 'lpg_dados_pessoa_fisica'

Alguns modelos encolheram: stg_agentes_pf/pj/coletivos faziam union all de 3-4 tabelas de layout idêntico e viraram uma varredura com where tabela_origem in (...). O substring(id_anexo from 'anexo_([0-9]+)') saiu de todos os modelos, porque id_anexo agora guarda o id do anexo e não mais o nome do arquivo.

Configuração necessária antes de rodar

Airflow Variable	Valor
transferegov_programas_ids	[7, 8, 9, 15, 46, 47, 60, 61, 62, 111, 112]
transferegov_politicas_publicas	[{"sigla": "LPG", "politica_publica": "...", "id_programas": [46, 47]}, ...]

A segunda existe porque a API não devolve sigla nem politica_publica, que a especificação exige em programa_minc. Sem ela a carga roda, mas essas duas colunas ficam nulas.

Ordem de execução: api_programas_dag → api_planos_acao_dag → (api_plano_acao_meta_dag, api_plano_acao_dado_bancario_dag, api_relatorios_gestao_dag → api_anexos_relatorios_dag → download_anexos_transferegov_dag → extracao_anexos_dag). extracao_bbagil_dag roda à parte e depende de plano_acao_dado_bancario_minc carregada.

Verificação

• 11/11 DAGs parseiam no container do Airflow.
• dbt run contra um Postgres com a estrutura nova: 32/32 modelos, 0 erros. As contagens por modelo batem com a fixture (as fatias sem linha da origem correspondente retornam 0).
• Smoke em Postgres real: nenhum ente estadual com cod_ibge de 7 dígitos, as duas contas do plano gravadas, seleção da conta ativa correta, carga de planilha idempotente em duas execuções, coluna excedente preservada em payload_origem e apenas os 3 schemas da especificação criados.
• 14 testes unitários (regra territorial e roteamento das planilhas). Ruff e mypy sem achado novo.

Falta a conferência numérica dos modelos dbt sobre dados reais — só é possível depois que a carga rodar em homologação.

Divergências assumidas

1. transferegov.politica_publica não foi criada: sigla e politica_publica são colunas de programa_minc, que é o que a especificação já exige nessa tabela.
2. Sem catálogo versionado (programas.json): o escopo é configuração de Airflow Variable.
3. linha_origem é a posição da linha dentro da subtabela extraída, não a linha física do Excel — quem identifica o registro é hash_registro.
4. bbagil.fato_bbagil segue declarada como source sem existir no banco; o modelo que a lê está enabled=false.

Fora do escopo

Anexos .ods e arquivos acima de 10 MB continuam sem virar linha; colunas técnicas restantes; índices; validações de qualidade automatizadas; tipagem (toda coluna nasce TEXT); FKs declaradas; mascaramento de dados pessoais; o filtro tipo_relatorio_gestao = 'FINAL' na ingestão de relatórios.

plugins/agencias_transferegov.py ficou sem nenhum importador depois que o BB Ágil passou a ler as contas do banco. Não foi removido aqui — merece PR próprio.