$v2ex，批量转账的 Python .方便佬哥们空投..

测试跑的 transaction：

https://solscan.io/tx/2bKiP2D47Z8WneNtn4CnbZM4QiTfmJTfzoXVabSPkf94rajSrMDwocASKty6ULYe1JEQZtexN4b5VQGBSya2kwsz

gas 是很低..用的公共节点.. 需要私钥在代码里..佬哥注意私钥安全.. 需要配置私钥、目标地址、目标金额就行。代币位数会自动处理。

pip install solana spl-token base58 

transaction.py

import os import time import base58 from solders.pubkey import Pubkey from solders.keypair import Keypair from solders.transaction import Transaction from solders.message import Message from solders.compute_budget import set_compute_unit_price from solana.rpc.api import Client from solana.rpc.types import TxOpts from solana.rpc.commitment import Confirmed from spl.token.client import Token from spl.token.constants import TOKEN_PROGRAM_ID from spl.token.instructions import get_associated_token_address, create_associated_token_account, transfer_checked, TransferCheckedParams # --- 1. 配置区域 --- # --- 请在这里修改您的转账信息 --- # 使用公共 RPC 节点 RPC_ENDPOINT = "https://api.mainnet-beta.solana.com" # 要转账的代币的 Mint Address TOKEN_MINT_ADDRESS = Pubkey.from_string("9raUVuzeWUk53co63M4WXLWPWE4Xc6Lpn7RS9dnkpump") PRIVATE_KEY_STRING= "PRIVATE_KEY_STRING" # 转账配置列表 TRANSFERS_COnFIG= [ {"address": "address", "amount": 2}, {"address": "address", "amount": 1}, {"address": "address", "amount": 2}, ] # 优先费 (Micro Lamports) PRIORITY_FEE_MICRO_LAMPORTS = 0 # --- 配置区域结束 --- def load_keypair_from_base58(private_key_b58: str) -> Keypair: try: keypair_bytes = base58.b58decode(private_key_b58) if len(keypair_bytes) != 64: raise ValueError(f"私钥解码后的长度应为 64 字节，实际为 {len(keypair_bytes)} 字节。") return Keypair.from_bytes(keypair_bytes) except Exception as e: print(f"错误：无法从 Base58 字符串加载密钥对。请检查.env 文件中的私钥。") raise e def main(): """主执行函数""" print("--- Solana SPL 代币批量转账脚本 (兼容新版 solana-py) ---") try: sender_keypair = load_keypair_from_base58(PRIVATE_KEY_STRING) client = Client(RPC_ENDPOINT) except Exception as e: print(f"初始化失败：{e}") return print(f"发送方地址: {sender_keypair.pubkey()}") print(f"代币地址: {TOKEN_MINT_ADDRESS}") print(f"共计 {len(TRANSFERS_CONFIG)} 笔转账任务") print("-" * 20) try: token_client = Token( cOnn=client, pubkey=TOKEN_MINT_ADDRESS, program_id=TOKEN_PROGRAM_ID, payer=sender_keypair ) mint_info = token_client.get_mint_info() decimals = mint_info.decimals print(f"代币小数位数: {decimals}") sender_ata = get_associated_token_address(sender_keypair.pubkey(), TOKEN_MINT_ADDRESS) print(f"发送方代币账户 (ATA): {sender_ata}") # --- 核心改动：新的交易构建流程 --- # 1. 将所有指令收集到一个列表中 instructiOns= [] if PRIORITY_FEE_MICRO_LAMPORTS > 0: instructions.append(set_compute_unit_price(PRIORITY_FEE_MICRO_LAMPORTS)) print(f"已添加优先费: {PRIORITY_FEE_MICRO_LAMPORTS} micro-lamports") print("正在为每个接收者生成转账指令...") for transfer_info in TRANSFERS_CONFIG: try: recipient_pubkey = Pubkey.from_string(transfer_info["address"]) amount = float(transfer_info["amount"]) amount_in_smallest_unit = int(amount * (10 ** decimals)) print(f" -> 任务: 向 {recipient_pubkey} 转账 {amount} 个代币") recipient_ata = get_associated_token_address(recipient_pubkey, TOKEN_MINT_ADDRESS) recipient_ata_info = client.get_account_info(recipient_ata).value if recipient_ata_info is None: print(f" 接收者 ATA ({recipient_ata}) 不存在，将添加 '创建 ATA' 指令。") instructions.append( create_associated_token_account( payer=sender_keypair.pubkey(), owner=recipient_pubkey, mint=TOKEN_MINT_ADDRESS ) ) instructions.append( transfer_checked( TransferCheckedParams( program_id=token_client.program_id, source=sender_ata, mint=TOKEN_MINT_ADDRESS, dest=recipient_ata, owner=sender_keypair.pubkey(), amount=amount_in_smallest_unit, decimals=decimals, signers=[] ) ) ) except Exception as e: print(f"处理转账任务 {transfer_info} 时出错: {e}") continue # 2. 获取最新区块哈希 print("正在获取最新区块哈希...") latest_blockhash = client.get_latest_blockhash(commitment=Confirmed).value.blockhash # 3. 使用指令、支付方和区块哈希创建 Message message = Message.new_with_blockhash( instructions, sender_keypair.pubkey(), latest_blockhash ) # 4. 使用 Message 和签名者创建 Transaction transaction = Transaction([sender_keypair], message, latest_blockhash) # --- 交易构建流程结束 --- print("正在发送交易到 Solana 网络...") opts = TxOpts(skip_preflight=False, preflight_commitment=Confirmed) # preflight 建议打开 tx_signature = client.send_transaction(transaction, opts=opts).value print("-" * 20) print(f" 交易已成功发送！") print(f" 交易签名: {tx_signature}") print(f" 在 Solscan 上查看: https://solscan.io/tx/{tx_signature}") print("正在等待交易确认...") client.confirm_transaction(tx_signature, commitment=Confirmed) print(" 交易已确认！批量转账完成。") except Exception as e: print(f"\n 执行过程中发生错误: {e}") if __name__ == "__main__": main()