Formulário de contato

Nome

E-mail *

Mensagem *

Este blog é um complemento do nosso canal no YouTube. Clique em @CanalQb para seguir e acompanhar nossos vídeos!
Imagem
CanalQbCanalQb

Importar Dados de Endereços Bitcoin para SQLite com Python

@CanalQb no YouTube

@CanalQb

Importar Dados de Endereços Bitcoin para SQLite com Python

Sempre crie uma frase de segurança única para jogos, testnets ou airdrops e evite usar sua carteira principal.


Importar Dados de Endereços Bitcoin para SQLite com Python

Este tutorial detalha como realizar o download, descompressão e importação de um grande arquivo contendo endereços de Bitcoin e seus respectivos saldos diretamente para um banco de dados SQLite. O objetivo é facilitar a análise local e o processamento de dados em grande escala, com otimização de desempenho e uso eficiente de memória.

📥 Fonte dos Dados

O arquivo utilizado neste processo é fornecido publicamente por Loyce.club, contendo uma lista de endereços de Bitcoin e seus saldos atualizados. O link direto para o arquivo é:

blockchair_bitcoin_addresses_and_balance_LATEST.tsv.gz

⚙️ Etapas Realizadas no Script

  1. Download do arquivo comprimido .gz contendo os dados TSV;
  2. Extração do conteúdo para um arquivo .tsv legível;
  3. Importação otimizada linha a linha para banco de dados SQLite;
  4. Criação de índice no campo address para facilitar consultas rápidas.

📌 Informações Técnicas

O script usa as bibliotecas:

  • requests – para baixar o arquivo com barra de progresso via tqdm;
  • gzip e shutil – para descompactar o arquivo .gz;
  • sqlite3 – banco de dados relacional leve embutido no Python;
  • gc – limpeza de memória durante grandes volumes de dados.

💻 Exemplo de Uso no Windows

O script detecta se está sendo executado no Windows e altera dinamicamente o título do console para: "Sistema SNAPSHOT @CANALQB - V 1.0", utilizando ctypes.windll.

📊 Otimização de Importação no SQLite

Para evitar lentidão na importação de milhões de linhas, algumas otimizações são aplicadas:

  • PRAGMA synchronous = OFF – desabilita sincronização segura para ganhar desempenho;
  • PRAGMA journal_mode = MEMORY – usa o journal em memória ao invés de disco;
  • INSERT OR IGNORE – evita duplicações sem erro.

🔁 Exemplo de Fluxo da Importação


with open(filename_tsv, 'r') as f:
    for line in f:
        if line.strip():
            address, balance = line.split('\t')
            cursor.execute('INSERT OR IGNORE INTO relatorio_btc (address, balance) VALUES (?, ?)', (address, float(balance)))

📈 Criação de Índice no Campo Address

Para consultas rápidas por endereço, o script cria um índice ao final:


cursor.execute('CREATE INDEX IF NOT EXISTS idx_address ON relatorio_btc(address);')

🛑 Aviso Importante

Este conteúdo tem caráter educativo e técnico. Nenhuma informação aqui deve ser utilizada para finalidades ilegais ou que violem os termos de uso das redes blockchain. O arquivo contém apenas dados públicos, e a manipulação deles localmente é permitida apenas para fins de análise.

🔗 Créditos e Links Oficiais

📦 Arquitetura do Projeto

Após a execução do script, os seguintes arquivos serão gerados:

  • blockchair_bitcoin_addresses_and_balance_ddmmaaaa.tsv.gz – arquivo comprimido original;
  • blockchair_bitcoin_addresses_and_balance_ddmmaaaa.tsv – versão descomprimida em texto;
  • relatorio_btc.db – banco de dados SQLite contendo todos os endereços e saldos.

✅ Considerações Finais

Este projeto é ideal para desenvolvedores que desejam trabalhar com grandes volumes de dados relacionados a criptomoedas localmente, sem depender de APIs de terceiros. Ele também pode ser facilmente adaptado para integração com interfaces web ou dashboards de análise.

Script Completo

import requests
from tqdm import tqdm
from datetime import datetime
import gzip
import shutil
import os
import sqlite3
import gc

import platform
import ctypes
if platform.system() == "Windows":
    ctypes.windll.kernel32.SetConsoleTitleW("Sistema SNAPSHOT @CANALQB - V 1.0")

# URL do arquivo
url = "http://addresses.loyce.club/blockchair_bitcoin_addresses_and_balance_LATEST.tsv.gz"

# Obtém a data de hoje no formato ddmmyyyy
datadehoje = datetime.today().strftime('%d%m%Y')

# Cria o nome para o arquivo com a data de hoje
filename_gz = f"blockchair_bitcoin_addresses_and_balance_{datadehoje}.tsv.gz"
filename_tsv = f"blockchair_bitcoin_addresses_and_balance_{datadehoje}.tsv"

# Verifica se o arquivo .gz já existe
if os.path.exists(filename_gz):
    resposta = input(f"O arquivo {filename_gz} já existe. Deseja baixar e sobrescrever? (s/n): ").strip().lower()
    if resposta != 's':
        print("Pulando o download e usando o arquivo existente.")
    else:
        # Fazendo o download do arquivo com barra de progresso
        print("Iniciando o download para sobrescrever o arquivo...")
        response = requests.get(url, stream=True)
        total_size = int(response.headers.get('content-length', 0))

        # Inicia a barra de progresso
        with open(filename_gz, 'wb') as file, tqdm(
            desc=filename_gz, total=total_size, unit='B', unit_scale=True
        ) as bar:
            for data in response.iter_content(chunk_size=1024):
                file.write(data)
                bar.update(len(data))

        print(f"Download concluído: {filename_gz}")

else:
    # Caso o arquivo .gz não exista, inicia o download normalmente
    print("Arquivo não encontrado. Iniciando o download...")
    response = requests.get(url, stream=True)
    total_size = int(response.headers.get('content-length', 0))

    # Inicia a barra de progresso
    with open(filename_gz, 'wb') as file, tqdm(
        desc=filename_gz, total=total_size, unit='B', unit_scale=True
    ) as bar:
        for data in response.iter_content(chunk_size=1024):
            file.write(data)
            bar.update(len(data))

    print(f"Download concluído: {filename_gz}")
 
if os.path.exists(filename_tsv):
    # Agora, descomprime o arquivo .gz para .tsv com barra de progresso
    with gzip.open(filename_gz, 'rb') as f_in:
        total_size = os.path.getsize(filename_gz)  # Tamanho total do arquivo .gz
        with open(filename_tsv, 'wb') as f_out:
            # Criação da barra de progresso para descompressão
            with tqdm(total=total_size, desc="Extraindo", unit='B', unit_scale=True) as bar:
                buffer_size = 1024 * 1024  # 1MB por vez para leitura
                while True:
                    buf = f_in.read(buffer_size)
                    if not buf:
                        break
                    f_out.write(buf)
                    bar.update(len(buf))  # Atualiza a barra de progresso com cada buffer lido

    print(f"Arquivo descomprimido para: {filename_tsv}")

# Excluir o arquivo .gz após a extração
# os.remove(filename_gz)
print(f"Arquivo {filename_gz} excluído.")

# Conectar ao banco de dados SQLite (cria o banco se não existir)
conn = sqlite3.connect('relatorio_btc.db')
cursor = conn.cursor()

# Desabilita o autocommit e usa transações em massa para melhorar a performance
cursor.execute('PRAGMA foreign_keys = OFF;')  # Desabilita restrições de chave estrangeira
cursor.execute('PRAGMA synchronous = OFF;')  # Desabilita o controle de sincronia para acelerar as inserções
cursor.execute('PRAGMA journal_mode = MEMORY;')  # Usa o journal em memória, que é mais rápido
cursor.execute('PRAGMA cache_size = 10000;')  # Ajusta o tamanho do cache, se necessário

# Criação da tabela relatorio_btc, caso não exista
cursor.execute('''CREATE TABLE IF NOT EXISTS relatorio_btc (
    address TEXT PRIMARY KEY,
    balance REAL
)''')

# Inicia a importação manual sem usar pandas
print("\nIniciando a importação dos dados para o banco de dados...")

# Lê o arquivo TSV linha por linha
total_lines = sum(1 for _ in open(filename_tsv))
processed_lines = 0

# Mensagem indicando que a importação começou e o total de linhas
print(f"Total de linhas a serem processadas: {total_lines}")

# Desabilita o autocommit e começa a transação
conn.isolation_level = None  # Desativa o autocommit
cursor.execute('BEGIN TRANSACTION;')

# Abre o arquivo TSV para leitura
with open(filename_tsv, 'r', encoding='utf-8') as f:
    # Lê linha por linha
    for line in f:
        if line.strip():  # Ignora linhas vazias
            address, balance = line.split('\t')  # Separa por tabulação
            try:
                balance = float(balance)  # Converte o balance para float
                cursor.execute('''INSERT OR IGNORE INTO relatorio_btc (address, balance)
                                  VALUES (?, ?)''', (address, balance))
                processed_lines += 1

                # Exibe o progresso da importação
                progress = (processed_lines / total_lines) * 100
                print(f"\r{progress:.2f}% já importado.", end='', flush=True)

            except Exception as e:
                print(f"Erro ao processar linha: {line.strip()}. Erro: {e}")
                continue

        # Limpeza de memória
        gc.collect()

    # Commit final após processar todas as linhas
    conn.commit()

# Criação do índice para otimizar as consultas
cursor.execute('CREATE INDEX IF NOT EXISTS idx_address ON relatorio_btc(address);')

# Finaliza a transação
conn.commit()

# Restaura as configurações originais do SQLite
cursor.execute('PRAGMA foreign_keys = ON;')
cursor.execute('PRAGMA synchronous = FULL;')

# Finaliza a conexão
conn.close()

print("\nImportação concluída com sucesso.")
Tags:

Comentários

Comente só assim vamos crescer juntos!

© CanalQb - Scripts e Criptos. All Rights Reserved