Tabelas dados CNPJ em Python

Mesclando dados de duas tabelas (Simples/Estabelecimentos)

import duckdb

arquivo_estab = "C:/ProgramData/MySQL/MySQL Server 8.0/Uploads/estabelecimentos/K3241.K03200Y0.D51108.ESTABELE.utf8.csv"
arquivo_simples = "C:/ProgramData/MySQL/MySQL Server 8.0/Uploads/simples/F.K03200$W.SIMPLES.CSV.D60110"
saida = "C:/ProgramData/MySQL/MySQL Server 8.0/Uploads/resultado_final.csv"

con = duckdb.connect()

# Criar tabelas temporárias
con.execute(f"""
CREATE OR REPLACE TEMPORARY TABLE estab AS
SELECT
column00 AS cnpj,
column01 AS ordem,
column02 AS dv,
column05 AS situacao_cadastral,
column19 AS uf,
column20 AS municipio,
column28 AS situacao_especial
FROM read_csv_auto('{arquivo_estab}', delim=';', header=False)
""")

con.execute(f"""
CREATE OR REPLACE TEMPORARY TABLE simples AS
SELECT
column0 AS cnpj,
CAST(column1 AS VARCHAR) AS opcao_simples,
CAST(column2 AS VARCHAR) AS data_opcao_simples,
CAST(column3 AS VARCHAR) AS data_exclusao_simples,
CAST(column4 AS VARCHAR) AS opcao_mei,
CAST(column5 AS VARCHAR) AS data_opcao_mei,
CAST(column6 AS VARCHAR) AS data_exclusao_mei
FROM read_csv_auto('{arquivo_simples}', delim=';', header=False)
""")

# Fazer JOIN e exportar CSV
con.execute(f"""
COPY (
SELECT
e.cnpj AS CNPJ_BASICO,
e.ordem AS CNPJ_ORDEM,
e.dv AS CNPJ_DV,
e.situacao_cadastral AS SITUACAO_CADASTRAL,
e.uf AS UF,
e.municipio AS MUNICIPIO,
e.situacao_especial AS SITUACAO_ESPECIAL,
COALESCE(s.opcao_simples, '') AS OPCAO_SIMPLES,
COALESCE(s.data_opcao_simples, '') AS DATA_OPCAO_SIMPLES,
COALESCE(s.data_exclusao_simples, '') AS DATA_EXCLUSAO_SIMPLES,
COALESCE(s.opcao_mei, '') AS OPCAO_MEI,
COALESCE(s.data_opcao_mei, '') AS DATA_OPCAO_MEI,
COALESCE(s.data_exclusao_mei, '') AS DATA_EXCLUSAO_MEI
FROM estab e
LEFT JOIN simples s
ON e.cnpj = s.cnpj
) TO '{saida}' (HEADER, DELIMITER ',');
""")

print("✅ FINALIZADO - CSV gerado em:", saida)

em txt:

Produção de dados (inclusão de informações na tabela)

import duckdb
import os

# =========================
# CONFIG
# =========================

# você vai preencher depois com os 8 arquivos
arquivos_estab = [
"C:/caminho/arquivo1.csv",
"C:/caminho/arquivo2.csv",
# add os outros aqui
]

arquivo_simples = "C:/ProgramData/MySQL/MySQL Server 8.0/Uploads/simples/F.K03200$W.SIMPLES.CSV.D60110"
saida = "C:/ProgramData/MySQL/MySQL Server 8.0/Uploads/resultado_final.csv"

# =========================

con = duckdb.connect()

# =========================
# 1. CARREGA SIMPLES UMA VEZ
# =========================
print("Carregando SIMPLES...")

con.execute(f"""
CREATE OR REPLACE TEMPORARY TABLE simples AS
SELECT
column0 AS cnpj,
CAST(column1 AS VARCHAR) AS opcao_simples,
CAST(column2 AS VARCHAR) AS data_opcao_simples,
CAST(column3 AS VARCHAR) AS data_exclusao_simples,
CAST(column4 AS VARCHAR) AS opcao_mei,
CAST(column5 AS VARCHAR) AS data_opcao_mei,
CAST(column6 AS VARCHAR) AS data_exclusao_mei
FROM read_csv_auto('{arquivo_simples}', delim=';', header=False)
""")

print("SIMPLES carregado.")

# =========================
# 2. PROCESSAR CADA ESTABELECIMENTO
# =========================

# remove arquivo final se existir (pra não duplicar)
if os.path.exists(saida):
os.remove(saida)

primeiro = True

for arquivo_estab in arquivos_estab:
print(f"Processando: {arquivo_estab}")

# cria tabela temporária para esse arquivo
con.execute(f"""
CREATE OR REPLACE TEMPORARY TABLE estab AS
SELECT
column00 AS cnpj,
column01 AS ordem,
column02 AS dv,
column05 AS situacao_cadastral,
column19 AS uf,
column20 AS municipio,
column28 AS situacao_especial
FROM read_csv_auto('{arquivo_estab}', delim=';', header=False)
""")

# se for o primeiro arquivo → cria com HEADER
if primeiro:
copy_mode = "HEADER"
primeiro = False
else:
copy_mode = "" # sem header pra não duplicar

# append no CSV final
con.execute(f"""
COPY (
SELECT
e.cnpj,
e.ordem,
e.dv,
e.situacao_cadastral,
e.uf,
e.municipio,
e.situacao_especial,
COALESCE(s.opcao_simples, ''),
COALESCE(s.data_opcao_simples, ''),
COALESCE(s.data_exclusao_simples, ''),
COALESCE(s.opcao_mei, ''),
COALESCE(s.data_opcao_mei, ''),
COALESCE(s.data_exclusao_mei, '')
FROM estab e
LEFT JOIN simples s
ON e.cnpj = s.cnpj
) TO '{saida}' (FORMAT CSV, {copy_mode}, DELIMITER ',', APPEND);
""")

print("\n✅ TODOS OS ARQUIVOS PROCESSADOS!")
print(f"Arquivo final: {saida}")

Importando para o Mysql:

ANTES DE RODAR (IMPORTANTE)

Execute isso pra garantir que o MySQL permite o arquivo:

SHOW VARIABLES LIKE 'secure_file_priv';
 
O arquivo CSV precisa estar exatamente nessa pasta.
 
DROP TABLE IF EXISTS resultado_final;

CREATE TABLE resultado_final (
    CNPJ_BASICO        VARCHAR(8),
    CNPJ_ORDEM         VARCHAR(4),
    CNPJ_DV            VARCHAR(2),
    SITUACAO_CADASTRAL VARCHAR(2),
    UF                 VARCHAR(2),
    MUNICIPIO          VARCHAR(10),
    SITUACAO_ESPECIAL  VARCHAR(50),

    OPCAO_SIMPLES          VARCHAR(1),
    DATA_OPCAO_SIMPLES     VARCHAR(8),
    DATA_EXCLUSAO_SIMPLES  VARCHAR(8),

    OPCAO_MEI              VARCHAR(1),
    DATA_OPCAO_MEI         VARCHAR(20),
    DATA_EXCLUSAO_MEI      VARCHAR(20)
)
ENGINE=InnoDB
DEFAULT CHARSET=utf8mb4;

-- 2. Importar CSV
LOAD DATA INFILE 'C:/ProgramData/MySQL/MySQL Server 8.0/Uploads/resultado_final.csv'
INTO TABLE resultado_final
CHARACTER SET utf8mb4
FIELDS TERMINATED BY ','
ENCLOSED BY '"'
LINES TERMINATED BY '\n'
IGNORE 1 LINES
(
    CNPJ_BASICO,
    CNPJ_ORDEM,
    CNPJ_DV,
    SITUACAO_CADASTRAL,
    UF,
    MUNICIPIO,
    SITUACAO_ESPECIAL,
    OPCAO_SIMPLES,
    DATA_OPCAO_SIMPLES,
    DATA_EXCLUSAO_SIMPLES,
    OPCAO_MEI,
    DATA_OPCAO_MEI,
    DATA_EXCLUSAO_MEI
);

-- 3. Índices (performance)
CREATE INDEX idx_cnpj_basico ON resultado_final (CNPJ_BASICO);
CREATE INDEX idx_municipio ON resultado_final (MUNICIPIO);
CREATE INDEX idx_uf ON resultado_final (UF);

KoahCont

Pesquisar este blog

Tabelas dados CNPJ em Python

Comentários

Postar um comentário