Fontes Externas de Dados Abertos

Bases públicas de referência carregadas para o grafo da plataforma pelo Datta Extract: CVM (companhias abertas, fundos e processos sancionadores) e a base negativa pública — Portal da Transparência (CEIS/CNEP/CEPIM/PEP), Banco Central e a lista consolidada do Conselho de Segurança da ONU (ver Base Negativa Pública). São o insumo do vertical KYC/KYB: paradas de drill-down do beneficiário final, resolução de fundos e base negativa.

Como funciona

Cada fonte é um fluxo de dados do Designer de Pipeline — igual a qualquer integração desenhada por um analista na plataforma. Não existe serviço, conector ou classe Java dedicada: o fluxo inteiro (download, limpeza, junção ao grafo) é composto por componentes do designer e fica salvo como pipeline (origin=DESIGNER).

O caminho para chegar até ele é Extract › Pacotes — a galeria que inventaria todos os pipelines de extração da instalação. Cada fluxo aparece como um cartão (com origem, agendamento e última execução) e o clique o abre no editor onde ele é editável. Para os fluxos de fontes externas isso é o modo Avançado, porque o grafo deles não é a cadeia linear que o Assistente monta — a própria galeria diz isso no cartão. Não existe atalho direto pelo menu: sem a galeria, o editor abre em branco.

A execução é o caminho padrão de pipelines da plataforma: o pipeline-designer-service agenda pelo cron do fluxo e executa no motor da própria plataforma — o mesmo que roda qualquer outro pipeline desenhado na UI. Antes, a carga era submetida ao Spark; não é mais.

Os três fluxos da CVM (arquivos em scripts/pipelines/cvm/):

Anatomia comum dos fluxos:

1. Origem por URL (UrlSource) — baixa o CSV do portal de dados abertos da CVM com validação anti-SSRF a cada redirecionamento e rejeição de arquivo vazio.
2. Transformações do designer — Formula (CNPJ só dígitos, CNPJ básico, versão da carga), Filter (descarta linha sem chave), Rename, SelectColumns e Deduplicate. Nenhum tratamento em código.
3. Destino grafo (KnowledgeGraphTarget) — MERGE por chave no banco do contexto de referência (default FEBRABAN) e criação das arestas ao grafo da Receita. O nó da ponta Receita usa modo Match: registro sem correspondente fica sem aresta — legítimo, nunca inventado.

Quando a origem é uma API, e não um arquivo

Fonte que responde por API entra pelo nó API REST, sem nenhum código dedicado. O nó cobre os quatro modos de paginação (por número de página, por deslocamento, pelo caminho da URL e por cursor devolvido na própria resposta), resposta em JSON ou XML, chave de API em campo cifrado (em vez de escrita à mão no cabeçalho) e a leitura em duas etapas — o caso em que a consulta devolve um link para o arquivo e não os registros.

O destino de grafo, por sua vez, cobre a carga alimentada por várias origens que descrevem a mesma entidade: cascata de chaves candidatas (a linha funde pela primeira que tiver valor), campo vazio que preserva o valor já gravado, propriedade que acumula a procedência em lista sem duplicar, e propriedade gravada só na criação.

Cada campo, com exemplo, está em Campos avançados dos conectores.

Fluxos que mudam de alvo a cada execução

Um fluxo pode declarar parâmetros de execução — contexto de destino, termo de busca, competência, teto de registros —, cada um com valor padrão. É o que evita ter um pacote por combinação: o mesmo desenho serve a todos os alvos, e uma correção vale para todos de uma vez.

Como os valores chegam ao disparo, hoje:

Ou seja: para rodar com valor diferente do padrão sem editar o pacote, use o Executar agora da galeria (ou a API). O formulário vale só para aquela execução: o padrão gravado no pacote não muda, e portanto o que a execução agendada usa continua sendo o que está no pacote — para mudá-lo, edite o fluxo em Extract › Designer.

Os valores usados ficam registrados na auditoria do pipeline, então dois runs do mesmo fluxo com parâmetros diferentes não se confundem.

Registro dos fluxos (seed)

Automático, pelo instalador

O setup-datta.sh registra os fluxos na Fase 7B — Pacotes de extração de fontes externas, sem nenhum comando extra. A fase fala com o api-gateway pela ingress (o DNS interno do cluster não resolve do host do instalador) e descobre o banco do contexto de referência lendo GET /api/contextos — o que o usuário já cadastrou, nunca um valor inventado (Diretrizes do Projeto §9).

Se o contexto de referência ainda não existir, a fase não registra nada, avisa e segue — instalar a plataforma nunca depende de um contexto de negócio existir. O mesmo vale se a API estiver fora do ar ou a senha do Neo4j não estiver disponível (execução com --skip-deploy e sem .env): vira aviso no resumo final, nunca falha de instalação. Para pular a fase de propósito: ./setup-datta.sh --skip-seed-extract.

Manual

Rode depois de cadastrar o contexto, em uma das duas formas:

# 1) Pela ingress, com JWT de admin (de fora do cluster — igual ao instalador).
#    Os bancos POR ORIGEM são descobertos nos dataSources NEO4J do contexto
#    (role=cvm | base-negativa | bcb); informe as variáveis só para forçar outros.
DATTA_API_URL=http://<ip-do-gateway>:7070 \
DATTA_JWT=<accessToken> \
NEO4J_PASSWORD=<senha> \
./scripts/seed-extract-pipelines.sh

# 2) De dentro do cluster, com token S2S.
CVM_NEO4J_DB=<banco role cvm> \
BASE_NEGATIVA_NEO4J_DB=<banco role base-negativa> \
BCB_NEO4J_DB=<banco role bcb> \
NEO4J_PASSWORD=<senha> \
DATTA_INTERNAL_TOKEN=<token S2S> \
./scripts/seed-extract-pipelines.sh

O script é idempotente por nome: fluxo já cadastrado não é recriado, então edições feitas depois no designer nunca são sobrescritas — e rodar o instalador duas vezes não duplica pacote. No modo S2S, os três bancos por origem continuam obrigatórios e sem default, porque de dentro do cluster não há JWT para consultar os contextos.

Quando o pacote do repositório muda

Pular em silêncio o que já existe protege a edição feita no designer, mas custava o outro lado: correção de pacote no repositório nunca chegava a uma plataforma já instalada. Foi assim que a correção do pacote de sancionadores da CVM ficou parada — o seed dizia "já cadastrado" e o fluxo seguia rodando a definição velha, falhando toda madrugada.

Hoje o seed compara o desenho do repositório com o cadastrado e avisa quando divergem, sem aplicar:

! DIVERGENTE (não aplicado): CVM — Processos Sancionadores (dados abertos)

Para aplicar, quando a versão do repositório é a correta (fonte que mudou de formato, chave corrigida, coluna renomeada):

SEED_ATUALIZAR=true ./scripts/seed-extract-pipelines.sh

Isso sobrescreve o desenho cadastrado — edições feitas no designer nesses pacotes se perdem. O pipeline guarda a versão anterior no histórico (Extract › Designer › Versões), então dá para comparar antes e voltar depois.

A comparação ignora o que o servidor administra (id, versão, timestamps) e o que ele acrescenta ao gravar (config: {} nas arestas). Ignora também a senha, que a API devolve mascarada: troca de credencial no repositório não aparece como divergência, e se aplica rodando com SEED_ATUALIZAR=true.

O setup-datta.sh chama este script na FASE 7B — Pacotes de extração de fontes externas (setup-datta.sh, ver a chamada a scripts/seed-extract-pipelines.sh). Numa instalação nova, os fluxos são registrados sem comando extra. A fase pula com aviso — sem quebrar a instalação — quando o contexto de referência não existe, a API está fora do ar, a senha do Neo4j não está disponível (execução com --skip-deploy e sem .env) ou o script está ausente. Para pular de propósito: ./setup-datta.sh --skip-seed-extract.

Em cluster que já existia o seed continua manual: a fase só roda durante o setup-datta.sh, e uma instalação anterior nunca a executou. datta-install.sh e seed-demo-data.sh não chamam o seed.

Nome do contexto de referência: o seed procura o contexto por CONTEXTO_REFERENCIA (default FEBRABAN), que precisa ser o mesmo de datta.kyb.contexto-referencia — se divergirem, os pacotes ficam registrados apontando para bancos que o cruzamento KYB nunca consulta. O contexto declara um dataSource NEO4J por origem (role=cvm, base-negativa, bcb, cnpj) e o seed lê os bancos daí — inclusive CNPJ_NEO4J_DB, que voltou a ser usada pelos pacotes do CNPJ (a base da Receita fica no banco cnpj e o join com as demais origens é por chave cnpjBasico, nunca por aresta entre bancos). Se o nome cadastrado divergir do configurado, o seed lista os contextos existentes na mensagem, em vez de só dizer que não encontrou; se faltar um role, ele lista o que cadastrar.

Operação pela UI

O ponto de partida é Extract › Pacotes; buscar pelo nome do fluxo (ex.: "CVM") ou filtrar por origem/integração. A partir do cartão:

• Abrir — carrega o fluxo no ETL Designer com nós, posições e configuração originais; ali se ajustam transformações, URL da origem ou cron, e salva (o histórico de versões registra cada mudança). O salvamento reenvia a definição completa, então descrição, cron, política de execução e metadados não se perdem.
• Executar agora — dispara uma execução avulsa direto da galeria; o progresso por nó aparece no monitor de execuções (Ver histórico leva ao monitor já filtrado pelo fluxo). Fluxo que declara parâmetros de execução abre antes um formulário com eles, preenchido com o padrão do pacote: ajuste o que vale para esta execução e confirme. Fluxo sem parâmetros dispara direto, sem formulário. Campo obrigatório em branco é barrado ali mesmo, com a explicação no próprio campo; recusa do motor (opção inválida, valor fora do tipo) aparece dentro do formulário, sem perder o que você já preencheu.
• Pausar/Ativar — controla o agendamento sem apagar o fluxo.
• Duplicar — cria uma cópia (útil para a variante air-gapped abaixo). Atenção: a cópia vem com as credenciais mascaradas e precisa que elas sejam reinformadas no editor antes de executar.
• Arquivar / Excluir — tira de operação ou remove definitivamente (com confirmação dupla).
• Detalhes — painel lateral com origem, status, versão, autor, agendamento e o desenho do fluxo em canvas somente leitura, para conferir a anatomia sem abrir o editor.
• Testar — dentro do designer, o modo teste roda com amostra e valida o fluxo sem gravar no grafo.

Instalação sem internet (air-gapped)

Baixe o CSV da origem numa máquina conectada e disponibilize-o no storage da plataforma. Depois, em Extract › Pacotes, duplique o fluxo, abra a cópia no designer e troque o nó de origem de URL para Arquivo (FileSource) apontando para o CSV — mantendo delimitador ; e encoding ISO-8859-1. O restante do fluxo (transformações e destino) é idêntico. Pause o fluxo original com URL para não acumular falhas de agendamento.

Agendamento

O cron faz parte do próprio fluxo (editável no designer): companhias às 04h00, fundos às 04h10 e sancionadores às 04h20, fuso America/Sao_Paulo. Retentativa única com 5 minutos de espera; falha notifica pelo canal configurado na política de execução do pipeline.

Permissões

As mesmas dos pipelines em geral — não há permissão dedicada por fonte:

Atenção ao conceder: hoje o perfil Analista não tem nenhuma permissão PIPELINE_* — mesmo assim enxerga a galeria, porque GET /api/pipelines/summary aceita PIPELINE_VIEW ou DATTAX_EXECUTE — e o Usuário Avançado não tem PIPELINE_DELETE. Ver RBAC.

Dados da CVM no grafo

Destino: o banco Neo4j do contexto de referência (mesmo grafo da Receita — a junção é por CNPJ). Labels: CompanhiaAberta, FundoInvestimento, AdministradorFiduciario, GestorRecursos, ProcessoSancionador, todos com versaoCarga (data da execução).

Pontos de atenção:

• O sancionador não liga a :Empresa, e isso é da fonte. Desde a mudança de esquema da CVM (o antigo pas.csv saiu do ar em 08/2026), o dataset publica apenas o nome do acusado — não há CPF nem CNPJ. Sem documento não existe chave de junção determinística, então o fluxo carrega os acusados como nós e o casamento por nome (com score e revisão humana) fica no cruzamento KYB, como já acontece com a lista da ONU. Criar aresta de sanção por nome seria homonímia virando acusação.
• Dois CSVs no mesmo zip. O fluxo tem duas origens — uma por entrada do zip — e as liga por NUP, que é a chave do processo dentro do próprio dataset. Evita coincidência numérica.
• Recarga é MERGE por chave. Os cadastros da CVM são registros completos (não deltas); cada execução atualiza os nós existentes e cria os novos. Registro que sai do cadastro da origem permanece no grafo com versaoCarga antiga — critério de triagem pode usar a data para sinalizar desatualização.

Solução de problemas

Observabilidade

A execução herda a telemetria padrão de pipelines: runs e status por nó no monitor do Extract, spans do pipeline-designer-service nos índices OTel da plataforma, e auditoria de criação/edição/execução no trail do próprio pipeline (aba de auditoria do designer).