O oráculo e o camelo

Tive que carregar uma base de dados com o conteúdo de dois DVDs cheios de informações. Os dados estavam separados em arquivos texto que por sua vez estavam dentro vários níveis de pastas.

A primeira tentativa foi usar um script em Perl. A solução funcionou, mas não foi rápida o suficiente para o curto espaço de tempo disponível (como sempre, o prazo era exíguo). Rápido mesmo foi usar o SQL Loader.

Mesmo assim, vou descrever a solução em Perl. Ela não é tão rápida, mas ela é mais flexível.

Em primeiro lugar, usei o módulo Find para buscar os arquivos. O módulo é dos básicos, então sempre está disponível.

Uma linha basta para disparar uma busca em uma hierarquia de pastas:

find(\&process_file, $ARGV[0]);

Essa linha dispara a busca com o valor do primeiro argumento dado ao script. Esse argumento deve conter o caminho de uma pasta. Para cada arquivo que o find() encontrar, ele invocará a função process_file().

Antes de apresentar a process_file(), vou mostrar a estrutura de dados que faz o relacionamento entre os nomes dos arquivos, seus formatos e os respectivos inserts.

my $tables={
  'PRODUTOS' => {
    format => 'A4A20A10',
    insert => q{insert into produtos (codigo, descricao, valor) 
                values (?,?,?/100)}
  },
  'PESSOAS' => {
    format => 'A11A60A8',
    insert => q{insert into pessoas (cpf, nome, data_nascimento) 
                values (?,?,to_date(?, 'DDMMYYYY'))}
  }
};

A referência $table aponta para um hash; cada chave do hash representa uma tabela cujo nome coincide com o nome do arquivo; associado a cada nome de tabela está um formato e um insert. Uso referências como poderia usar hashes diretamente.

A função process_file(), então, determina se existe um mapeamento para cada arquivo encontrado e invoca a função load() quando existir.

sub process_file {
  my $filename=$File::Find::name;
  my $name=$_;
  
  if($name=~/(\w+)\.txt/) {
    if(exists $tables->{$1}) {
      load($filename, $1);
    }
  }
}

Os nomes dos arquivos têm os mesmos nomes das tabelas e o sufixo "txt". Usando uma expressão regular, determino se o arquivo tem o sufixo desejado e uso a primeira parte do nome ($1) para encontrar o mapeamento.

Para extrair os dados dos arquivos, uso a função unpack(). Esta função recebe um string que descreve uma linha e a linha propriamente dita. Usando a descrição, ela divide a linha e produz um array com os campos. Então, para percorrer um arquivo texto com 3 colunas de 10 caracteres cada, bastaria fazer o seguinte:

while(<FILE>) {
  my @cols=unpack("A10A10A10");
}

O segundo parâmetro é implícito (<FILE> coloca a linha lida em $_ e unpack() o usa como parâmetro implícito).

Para gravar os dados, uso o DBI, que é o módulo de acesso a bases de dados. O exemplo abaixo mostra como abrir uma conexão, preparar um comando e executá-lo.

$db = DBI->connect( "dbi:Oracle:host=HOST;sid=SID", "SCHEMA", "PASSSWD" )  
  || die( $DBI::errstr . "\n" );
$st=$db->prepare('INSERT INTO PRODUTOS (CODIGO, NOME, VALOR) VALUES(?,?,?)');
$st->execute('0123', 'BOLA', 49.95);

Juntando o unpack() com o DBI, o resultado é:

sub load {
  my ($filename, $table)=@_;
  my $format=$tables->{$table}->{format};
  my $insert=$tables->{$table}->{insert};

  open(my $fh, "<$filename");
  my $st=$db->prepare($insert);
  while(<$fh>) {
    my @row=unpack($format, $_);
    $st->execute(@row);
  }
  $db->commit();
  close($fh);
}

A função unpack() produz um array e o método execute() recebe um array. Não podia ser mais fácil. Executo um commit() por arquivo, mas poderia muito bem executar um único commit no fim de tudo.

E, finalmente, juntando todas as partes tem-se:

#!/usr/bin/perl
use DBI;
use File::Find;

my $db = DBI->connect( "dbi:Oracle:host=HOST;sid=SID", "SCHEMA", "PASSSWD" )  
  || die( $DBI::errstr . "\n" );
$db->{AutoCommit}    = 0;
$db->{RaiseError}    = 1;

my $tables={
  'PRODUTOS' => {
    format => 'A4A20A10',
    insert => q{insert into produtos (codigo, descricao, valor) 
                values (?,?,?/100)}
  },
  'PESSOAS' => {
    format => 'A11A60A8',
    insert => q{insert into pessoas (cpf, nome, data_nascimento) 
                values (?,?,to_date(?, 'DDMMYYYY'))}
  }
};

sub load {
  my ($filename, $table)=@_;
  my $format=$tables->{$table}->{format};
  my $insert=$tables->{$table}->{insert};

  open(my $fh, "<$filename");
  my $st=$db->prepare($insert);
  while(<$fh>) {
    my @row=unpack($format, $_);
    $st->execute(@row);
  }
  $db->commit();
  close($fh);
}

sub process_file {
  my $filename=$File::Find::name;
  my $name=$_;
  
  if($name=~/(\w+)\.txt/) {
    if(exists $tables->{$1}) {
      load($filename, $1);
    }
  }
}

find(\&process_file, $ARGV[0]);