Pythonでsymbolブロックチェーン上の複数のアドレスに一括でXYM送金。(23年2月) #Python

【概要】
symbolブロックチェーンにはアグリゲートトランザクションという複数のアドレス宛てのそれぞれのトランザクションを一括で送る機能がある。1件1件トランザクションを発生させて手数料を払わなくても、1回分の手数料で100件までのアドレス宛てに一斉に送信・送金が可能。
【環境・前提条件】
windows10
Python 3.10.9
symbol-sdk-python 3.0.3
【コード(testnet_aggregate_tx.py)】
それぞれのトランザクションを生成して、それらをまとめてアグリゲートトランザクションにしている感じ。必要なライブラリは、pip install <ライブラリ名>でインストール。

import sha3
import datetime
import http.client
from binascii import unhexlify
from symbolchain.CryptoTypes import PrivateKey
from symbolchain.symbol.KeyPair import KeyPair
from symbolchain.facade.SymbolFacade import SymbolFacade
from symbolchain.symbol.MerkleHashBuilder import MerkleHashBuilder
from symbolchain.CryptoTypes import Hash256
import requests
import json

#XYM送信に必要な項目を入力
node = "sym-test-04.opening-line.jp"
sender_private_key = "testnet private key"
fee = 0.5

# 宛先リスト["address", "amount", "message"]
toList = [
    ["TAFQVD**********3L7Q2BPVONS3FLM6VM5QKHQ", 1, "dogs" ],
    ["TA6USN**********UG5G7AEOMWVA4FYWO2EWM4I", 5, "cats"],
    ["TDTFCO**********353SR5VNMPXOBDEIVXYDXMQ", 7, "birds" ] 
    ]

#ノードからプロパティを取得
def get_node_properties():
        """
        network type, epock ajustment, currency mosaic id, generation hashを取得する
        """
        node_url = "https://" + node+ ":3001"
        node_url_properties = node_url + "/network/properties"
        response = requests.get(node_url_properties)
        if response.status_code != 200:
            raise Exception("status code is {}".format(response.status_code))
        contents = json.loads(response.text)
        network = str(contents["network"]["identifier"].replace("'", ""))
        epoch_adjustment = int(contents["network"]["epochAdjustment"].replace("s", ""))
        currency_mosaic_id = int(contents["chain"]["currencyMosaicId"].replace("'", ""), 16)
        generation_hash_seed = str(contents["network"]["generationHashSeed"].replace("'", ""))
        return network, epoch_adjustment,currency_mosaic_id, generation_hash_seed, node_url

#nodeをmainnetノードに変えればにプロパティも変わる
NETWORK_TYPE = get_node_properties()[0]
EPOCH_ADJUSTMENT = get_node_properties()[1]
MOSAICID = get_node_properties()[2]
NODE_URL = get_node_properties()[4]

#GENARATION_HASHは送金に利用しないが、確認用。
GENERATOIN_HASH = get_node_properties()[3]


#各プロパティ確認用
print("Network type => " + str(NETWORK_TYPE))
print("Epoch ajustment => " + str(EPOCH_ADJUSTMENT))
print("Mosaic ID => " + str(hex(MOSAICID)).upper())
print("Generation hash => " + str(GENERATOIN_HASH))
print("Using node url => " + str(NODE_URL))


# testnetで実施
facade = SymbolFacade(NETWORK_TYPE)

# 送信元の設定
Prikey     = PrivateKey(unhexlify(sender_private_key))
Keypair    = KeyPair(Prikey)
Pubkey     = Keypair.public_key
Address    = facade.network.public_key_to_address(Pubkey)
print("send from :", Address)

# 受信先の設定(toList[]の宛先1件ずつのトランザクション生成）
addressTx = []
for to in toList:
    address = SymbolFacade.Address(to[0])
    mosaic_size = to[1]
    msg = to[2]
    tx  = facade.transaction_factory.create_embedded({
        'type': 'transfer_transaction',
        'signer_public_key': Pubkey,
        'recipient_address': address,
        'mosaics': [{'mosaic_id':MOSAICID, 'amount':int(mosaic_size * 1000000)}],
        'message': bytes(1) + msg.encode('utf8')
    })
    addressTx.append(tx)
　  print("Recipient address => " + str(address) + "  /Amount => " + str(mosaic_size) + "  /Message => " + msg)

# マークルハッシュの作成
hash_builder = MerkleHashBuilder()
for tx in addressTx:
    hash_builder.update(Hash256(sha3.sha3_256(tx.serialize()).digest()))
merkle_hash = hash_builder.final()

# アグリゲートトランザクションの作成
deadline = (int((datetime.datetime.today() + datetime.timedelta(hours=2)).timestamp()) - EPOCH_ADJUSTMENT) * 1000
aggregate_tx = facade.transaction_factory.create({
    'type': 'aggregate_complete_transaction',
    'signer_public_key': Pubkey,
    'fee': int(fee * 1000000),
    'deadline': deadline,
    'transactions_hash': merkle_hash,
    'transactions': addressTx
})

#ハードフォークによりアグリゲートトランザクションはver.2を使用する。
aggregate_tx.version = 2

#署名＆ペイロード生成(v3.0.3から書き方変更)
signature = facade.sign_transaction(Keypair, aggregate_tx)
json_payload = facade.transaction_factory.attach_signature(aggregate_tx, signature)

#ネットワークへアナウンス
headers = {'Content-type': 'application/json'}
conn = http.client.HTTPConnection("sym-test-04.opening-line.jp",3000)
conn.request("PUT", "/transactions", json_payload, headers)
response = conn.getresponse()
print(response.status, response.reason)

# 確認
hash = facade.hash_transaction(aggregate_tx)
print(NODE_URL + "/transactionStatus/" + str(hash))

mosaicIDを変えれば自身で作成したモザイク（トークン）も送ることが出来ます。
testnetの場合、generationhashがライブラリ内のファイルにあるので、うまくいかないときは調べてみると良いかも。
参照元参考:
/Local/Programs/Python/Python310/Lib/site-packages/symbolchain/symbol/Network.py

【ターミナル出力】
無事送金出来ました。
メッセージも個々に変えることが出来ます。

Network type => testnet
Epoch ajustment => 1667250467
Mosaic ID => 0X72C0212E67A08BCE
Generation hash => 49D6E1CE276A85B70EAFE52349AACCA389302E7A9754BCF1221E79494FC665A4
Using node url => https://sym-test-04.opening-line.jp:3001
send from : TDT*****************BDEIVXYDXMQ
Recipient address => TAFQVD**********3L7Q2BPVONS3FLM6VM5QKHQ  /Amount => 1  /Message => dogs
Recipient address => TA6USN**********UG5G7AEOMWVA4FYWO2EWM4I  /Amount => 5  /Message => cats
Recipient address => TDTFCO**********353SR5VNMPXOBDEIVXYDXMQ  /Amount => 7  /Message => birds
202 Accepted
https://sym-test-04.opening-line.jp:3001/transactionStatus/6943A****************600245E2CAC87F5BDC45

【コード(mainnet_aggregate_tx.py)】
node・privatekey・toList[]内のaddressを変えればmainnet用になります。

node = "mainnet node"
private_key = "mainnet privatekey"
toList = [["mainnet_address", "amount", "message"]]

【引用リンク】

Symbolブロックチェーンで一人が複数人に送金するトランザクションをPythonで作成する
https://qiita.com/nem_takanobu/items/d148dee444a74737769b

最新SymbolSDKを使って複数人に個別枚数を一斉送信するプログラムを実装してみた。
https://note.com/kotopapa8/n/ne2c60623e5c5

ENTERキー1発　Pythonで複数アドレスに一斉送金【実行編】
https://nemlog.nem.social/blog/62080

agreegate version 2について
https://note.com/the_symbol_news/n/n945ea80f6390