Como Processar 60+ milhões de CNPJs com Python: Arquitetura e Decisões Técnicas

# src/database/factory.py
ADAPTERS: Dict[str, Type[DatabaseAdapter]] = {
    "postgresql": PostgreSQLAdapter,
    # Preparado para: MySQL, BigQuery, SQLite
}

def create_database_adapter(config) -> DatabaseAdapter:
    backend = config.database_backend.value
    adapter_class = ADAPTERS.get(backend)
    if not adapter_class:
        raise ValueError(f"Unsupported database backend: {backend}")
    return adapter_class(config)

# src/download_strategies/parallel.py
class ParallelDownloadStrategy(DownloadStrategy):
    def download_files(self, directory: str, files: List[str]) -> Iterator[Path]:
        # Separa referências de dados
        reference_files, data_files = self._categorize_files(files)

        # Referências primeiro (sequencial por segurança)
        for csv in self._download_sequential(reference_files):
            yield csv

        # Dados em paralelo
        with ThreadPoolExecutor(max_workers=4) as executor:
            futures = {
                executor.submit(self.download_single_file, dir, f): f 
                for f in data_files
            }
            for future in as_completed(futures):
                yield future.result()

# src/downloader.py
def organize_files_by_dependencies(self, files: List[str]) -> Tuple[List[str], Dict]:
    # Tabelas de referência primeiro (sem FKs)
    REFERENCE_TABLES = {
        "Cnaes.zip", "Motivos.zip", "Municipios.zip", 
        "Naturezas.zip", "Paises.zip", "Qualificacoes.zip"
    }

    # Ordem respeitando dependências
    ORDERED_PATTERNS = [
        "Empresas",          # Depende de: naturezas_juridicas
        "Estabelecimentos",  # Depende de: empresas, municipios, motivos
        "Socios",           # Depende de: empresas
        "Simples",          # Depende de: empresas
    ]

    # Organiza arquivos por categoria
    reference_files = [f for f in files if f in REFERENCE_TABLES]
    data_files = {pattern: [] for pattern in ORDERED_PATTERNS}

    for filename in files:
        for pattern in ORDERED_PATTERNS:
            if filename.startswith(pattern):
                data_files[pattern].append(filename)
                break

    # Monta ordem final
    ordered_files = reference_files[:]
    for pattern in ORDERED_PATTERNS:
        ordered_files.extend(sorted(data_files[pattern]))

    return ordered_files, categorization_info

# src/config.py
def _detect_strategy(self) -> ProcessingStrategy:
    memory_gb = psutil.virtual_memory().total / (1024**3)

    if memory_gb < 8:
        return ProcessingStrategy.MEMORY_CONSTRAINED
    elif memory_gb < 32:
        return ProcessingStrategy.HIGH_MEMORY
    else:
        return ProcessingStrategy.DISTRIBUTED

@property
def optimal_chunk_size(self) -> int:
    memory_gb = psutil.virtual_memory().total / (1024**3)

    # Chunks adaptados à memória disponível
    if self.processing_strategy == ProcessingStrategy.MEMORY_CONSTRAINED:
        return 100_000 if memory_gb < 4 else 500_000
    elif self.processing_strategy == ProcessingStrategy.HIGH_MEMORY:
        return 2_000_000
    else:
        return 5_000_000

# src/processor.py
def _process_chunked(self, file_path: Path, db, table_name: str):
    chunk_size = self.config.optimal_chunk_size
    offset = 0

    while True:
        chunk_df = pl.read_csv(
            file_path,
            separator=";",
            encoding="utf8",
            has_header=False,
            skip_rows=offset,
            n_rows=chunk_size,
        )

        if len(chunk_df) == 0:
            break

        # Transformações: conversão de tipos, limpeza de datas
        chunk_df = self._apply_transformations(chunk_df, file_type)

        # Bulk insert otimizado
        db.bulk_upsert(chunk_df, table_name)

        offset += len(chunk_df)
        del chunk_df
        gc.collect()  # Força liberação de memória

# Datas inválidas: "00000000" → NULL
df = df.with_columns(
    pl.when(pl.col(date_col) == "0")
    .then(None)
    .otherwise(pl.col(date_col))
)

# Decimais com vírgula: "1234,56" → 1234.56
df = df.with_columns(
    pl.col(numeric_col).str.replace(",", ".").cast(pl.Float64)
)

# Encoding corrompido: detecção e conversão automática
def _convert_file_encoding_chunked(self, input_file: Path):
    chunk_size = self.config.encoding_chunk_size  # 50MB chunks
    with open(input_file, "r", encoding="ISO-8859-1") as infile:
        with open(output_file, "w", encoding="UTF-8") as outfile:
            while chunk := infile.read(chunk_size):
                outfile.write(chunk)

def retry_db_connection(max_retries=3, base_delay=1.0):
    def decorator(func):
        @wraps(func)
        def wrapper(*args, **kwargs):
            for attempt in range(max_retries + 1):
                try:
                    return func(*args, **kwargs)
                except (psycopg2.OperationalError, psycopg2.DatabaseError) as e:
                    if attempt == max_retries:
                        raise
                    delay = base_delay * (2**attempt)
                    time.sleep(delay)
        return wrapper
    return decorator