Scribd
Upload
5 views2 pages

Abb

The document outlines a Python script designed to scan cryptocurrency seeds for wallets with balances using asynchronous programming and multithreading. It includes functions for handling API calls, caching addresses, and logging statistics about the scanning process. The main function orchestrates the seed generation, scanning, and results logging, while managing shared resources safely.

Uploaded by

altcoinlx
Copyright
© © All Rights Reserved
We take content rights seriously. If you suspect this is your content, claim it here.
Available Formats
Download as TXT, PDF, TXT or read online on Scribd
5 views2 pages

Abb

The document outlines a Python script designed to scan cryptocurrency seeds for wallets with balances using asynchronous programming and multithreading. It includes functions for handling API calls, caching addresses, and logging statistics about the scanning process. The main function orchestrates the seed generation, scanning, and results logging, while managing shared resources safely.

Uploaded by

altcoinlx
Copyright
© © All Rights Reserved
We take content rights seriously. If you suspect this is your content, claim it here.
Available Formats
Download as TXT, PDF, TXT or read online on Scribd
You are on page 1/ 2

asfasfasfasfasfasfa

f"Type: {path_name}\nAddress: {address}\nBalance: {balance:.8f}


BTC\nPrivate: {priv}\nPath: {full_path}\n"
)

stats.seeds_scanned += 1
return "\n".join(result) if result else None

# ==================== WORKER ====================


loop = asyncio.new_event_loop()
lock = Lock()

def worker(seed: str, config: Config, cache: dict, stats: dict):


async def _run():
async with aiohttp.ClientSession() as session:
api = MultiAPI(config, session)
secure = SecureStorage(config.KEY_FILE)
addr_cache = AddressCache(config.CACHE_FILE, config.MAX_CACHE_SIZE,
secure)
addr_cache.cache.update(cache)
stat = Stats()
result = await scan_seed(seed, config, api, addr_cache, stat)
with lock:
for k in ['seeds_scanned', 'wallets_found', 'addresses_checked',
'api_calls']:
stats[k] += getattr(stat, k)
cache.update(addr_cache.cache)
return result
return asyncio.run(_run())

# ==================== MAIN ====================


def main():
start_time = time.time()
config = Config()
setup_logging(config)
validator = Bip39MnemonicValidator(Bip39Languages.ENGLISH)
sg = SeedGenerator(config)
secure = SecureStorage(config.KEY_FILE)
shared_cache = {}
shared_stats = {'seeds_scanned': 0, 'wallets_found': 0, 'addresses_checked': 0,
'api_calls': 0}
seeds = []

if Path(config.LEAK_FILE).exists():
with open(config.LEAK_FILE, "r", encoding="utf-8") as f:
seeds = [line.strip() for line in f if validator.IsValid(line.strip())]

batch = 0
with ThreadPoolExecutor(max_workers=config.THREADS) as pool:
while batch < config.TOTAL_BATCHES:
if not seeds:
seeds = [sg.generate() for _ in range(config.SEED_BATCH_SIZE)]
futures = [pool.submit(worker, s, config, shared_cache, shared_stats)
for s in seeds]
results = [f.result() for f in futures]
found = [r for r in results if r]
if found:
with open(config.RESULT_FILE, "a", encoding="utf-8") as f:
f.write("\n".join(found) + "\n")
elapsed = time.time() - start_time
speed = shared_stats['seeds_scanned'] / elapsed if elapsed > 0 else 0
logging.info("="*60)
logging.info(f"Seeds scanned :
{shared_stats['seeds_scanned']}")
logging.info(f"Addresses checked :
{shared_stats['addresses_checked']}")
logging.info(f"Wallets with balance :
{shared_stats['wallets_found']}")
logging.info(f"Elapsed time : {elapsed:.2f} seconds")
logging.info(f"Speed : {speed:.2f} seeds/sec")
logging.info("="*60)
seeds = []
batch += 1

AddressCache(config.CACHE_FILE, config.MAX_CACHE_SIZE, secure).save()

if __name__ == "__main__":
asyncio.set_event_loop(loop)
main()

You might also like