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

V2EX = way to explore

V2EX 是一个关于分享和探索的地方

现在注册

已注册用户请登录

V2EX 支持 Solana 登录

你可以在 V2EX 设置中绑定 Solana 地址，然后就可以用 Phantom / Glow 登录 V2EX。

Get V2EX Coin

9raUVuzeWUk53co63M4WXLWPWE4Xc6Lpn7RS9dnkpump

测试跑的 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()

Solana

Python

批量转账

7 条回复 • 2025-08-22 16:45:57 +08:00