refactor: migra extração BB Ágil de JSON local para Postgres raw

[CaioMelo25]

Resumo

Migra a extração financeira do BB Gestão Ágil (BSC/SERPRO) de checkpoint em arquivo JSON local para tabelas raw + controle no Postgres, e conclui a extração histórica completa (677 mil combinações plano de ação × período).

Motivação original: os JSONs continham CPF/CNPJ/conta bancária de beneficiário não anonimizados, salvos só no disco de quem roda a DAG (sem backup nem controle de acesso), e o pipeline recalculava o fato inteiro do zero a cada execução varrendo o diretório — apagar um JSON corrompia silenciosamente o resultado final (upsert sobrescreve, não soma).

O que mudou

Arquitetura (extracao_bbagil_dag.py)

• 4 tasks: extrair_agencias_transferegov → persistir_agencias_contas_transferegov → extrair_extrato_bbagil → extrair_subtransacoes_bbagil. As tasks antigas de consolidação/fato em pandas foram removidas — esta DAG agora só extrai e deposita dado bruto no Postgres.
• Tabelas novas no schema bsc_pnab: raw_bbagil_extrato_transacoes, raw_bbagil_subtransacoes, controle_extracao_bbagil_extrato, controle_extracao_bbagil_subtransacoes (a tabela de controle substitui o Path.exists() como checkpoint — necessária porque ela registra tentativas por plano de ação × período, enquanto a tabela raw só tem linha quando existe transação; sem ela não dá pra distinguir "período vazio confirmado" de "ainda não verificado").
• execucao_assincrona_bsc.executar_lote ganhou tamanho_lote/ao_concluir_lote opcionais: persiste a cada 2.000 itens processados em vez de só no final — evita perder o trabalho inteiro numa interrupção no meio de uma execução de muitas horas.
• agencias_transferegov.get_contas_agencias_programas ganhou cache de ids_plano_acao_conhecidos: evita rebuscar agência/conta de planos de ação já persistidos em execuções anteriores (caiu de ~18 mil chamadas HTTP repetidas por execução pra só as novas).
• max_active_runs=1: DagRuns concorrentes autenticavam independentemente no SCA com o mesmo client_id/secret, e o SCA invalida o token anterior ao emitir um novo — isso derrubava o token de execuções paralelas e causava 401/403 sem relação com throttle de requisições.
• retries=100/retry_delay=10min: o BSC aplica bloqueio temporário (401/403) após uso sustentado (~20-40min de chamadas contínuas), independente do throttle configurado — é padrão esperado, não falha real, e o checkpoint em lote torna cada retry barato.
• Coluna ente renomeada para id_plano_acao nas 4 tabelas do schema bsc_pnab: o nome antigo sugeria ser o município/estado real, mas sempre guardou o ID do plano de ação — um mesmo ente pode ter vários planos de ação (cada um com conta bancária segregada própria), então o nome confundia a granularidade real da chave.

Removido

• extracao_beneficiarios_dag.py: dependia do fato_bbagil.parquet que não existe mais desde esta migração, e nenhum model/DAG consumia o output das suas 5 tasks (CPF list, CNPJ, BPC, CadÚnico, relação trabalhista) — ficava sem produtor e sem consumidor.
• Camada de transformação dbt (bronze/silver/gold) que havia sido desenhada e validada durante a migração foi revertida — prioridade definida foi completar a extração raw primeiro; fica para uma próxima PR.
• dags/data_ingest/transferegov_fundo_a_fundo/sql/ (DDL de referência que fugia do padrão do repo — as tabelas já são criadas em runtime pelo ClientPostgresDB).

Corrigido de quebra

• Conflito de merge não resolvido em dbt/minc/models/sources.yml.

Resultado da extração (dados reais, schema bsc_pnab)

• 413.858 linhas de transação em raw_bbagil_extrato_transacoes (de 677.207 combinações testadas; a diferença retornou "sem lançamentos", registrado como fato de negócio via controle_extracao_bbagil_extrato, não erro)
• 11.694 entes distintos, 9 programas (PNAB Municípios/Estados, Lei Paulo Gustavo Municípios/Estados, Lei Aldir Blanc), período Jan/2023–Jul/2026
• Volume financeiro equilibrado: R$ 12,19 bi em créditos / R$ 12,19 bi em débitos — consistente com extrato bancário
• 89.034 linhas de subtransação em raw_bbagil_subtransacoes
• 0 linhas com erro de dado

Não incluído nesta PR (próximos passos)

• Camada de transformação dbt (bronze/silver/gold) sobre as tabelas raw
• Migração de extracao_beneficiarios_bbagil_dag pro mesmo padrão (precisa antes recriar o fato_bbagil via dbt)
• Subir os dados extraídos pro ambiente/banco do MinC (ainda local, aguardando VPN de acesso ao ambiente definitivo)

Test plan

• DAG rodada de ponta a ponta em produção local, extração completa das 677.207 combinações + 89.034 subtransações, dag_run.state=success
• Rename de coluna validado com run de teste pós-migração: SELECT, JOIN com raw_planos_acao_dado_bancario e INSERT/upsert confirmados funcionando com id_plano_acao
• Persistência em lote testada com mocks (abort no meio preserva lotes já persistidos)
• Cache de agência/conta testado com mocks
• max_active_runs=1 confirmado no Postgres (SELECT max_active_runs FROM dag) e sem mais 401/403 por concorrência entre DagRuns