-
Notifications
You must be signed in to change notification settings - Fork 0
/
wallet_balance.py
84 lines (73 loc) · 3.36 KB
/
wallet_balance.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
import requests
import concurrent.futures
from UserAgenter import UserAgent # Supondo que UserAgent seja uma classe real
base_urls = [
"https://mempool.jednadvacet.org/api/address/",
"https://mempool.blocktrainer.de/api/address/",
"https://mempool.space/api/address/",
"https://blockstream.info/api/address/",
"https://btcscan.org/api/address/"
]
agent = UserAgent()
def get_wallet_balance(url):
try:
headers = {'User-Agent': agent.RandomAgent()}
response = requests.get(url, headers=headers)
response.raise_for_status()
data = response.json()
if "address" in data and "chain_stats" in data:
address = data["address"]
funded_txo_sum = data["chain_stats"].get("funded_txo_sum", 0)
spent_txo_sum = data["chain_stats"].get("spent_txo_sum", 0)
balance = (funded_txo_sum - spent_txo_sum) / 100_000_000 # Convertendo para BTC
return address, balance
else:
print(f"Formato de resposta inesperado para URL: {url}")
return None
except requests.exceptions.HTTPError as errh:
print(f"Erro HTTP ao consultar {url}: {errh}")
except requests.exceptions.ConnectionError as errc:
print(f"Erro de conexão ao consultar {url}: {errc}")
except requests.exceptions.Timeout as errt:
print(f"Timeout ao consultar {url}: {errt}")
except requests.exceptions.RequestException as err:
print(f"Erro ao consultar {url}: {err}")
except ValueError as ve:
print(f"Erro ao processar resposta JSON para URL {url}: {ve}")
except Exception as e:
print(f"Erro inesperado ao consultar {url}: {e}")
return None
def get_wallet_addresses(source):
if source == 'M':
user_input = input("Insira os endereços das wallets (separados por vírgulas, espaços ou linhas):\n")
addresses = [addr.strip() for addr in user_input.replace('\n', ',').replace(' ', ',').split(',') if addr.strip()]
elif source == 'A':
filename = input("Digite o nome do arquivo TXT contendo os endereços das wallets (um por linha):\n")
try:
with open(filename, 'r') as file:
addresses = [line.strip() for line in file.readlines() if line.strip()]
except FileNotFoundError:
print(f"Arquivo {filename} não encontrado.")
addresses = []
else:
print("Opção inválida. Por favor, escolha M para manual ou A para importar de arquivo.")
addresses = []
return addresses
def main():
addresses = get_wallet_addresses(source='M') # Exemplo usando entrada manual, pode ser adaptado para arquivo (A)
if not addresses:
return
urls = [f"{base_urls[i % len(base_urls)]}{address}" for i, address in enumerate(addresses)]
with concurrent.futures.ThreadPoolExecutor(max_workers=6) as executor:
future_to_url = {executor.submit(get_wallet_balance, url): url for url in urls}
for future in concurrent.futures.as_completed(future_to_url):
url = future_to_url[future]
try:
result = future.result()
if result:
address, balance = result
print(f"Address: {address}, Balance: {balance:.8f} BTC")
except Exception as e:
print(f"Erro ao processar a URL {url}: {e}")
if __name__ == "__main__":
main()